Multi-Currency Accounting API Integration: Complete Developer Guide

Database Schema for Exchange Rates

-- Exchange rate master table with audit trail
CREATE TABLE exchange_rates (
    id SERIAL PRIMARY KEY,
    from_currency VARCHAR(3) NOT NULL,
    to_currency VARCHAR(3) NOT NULL,
    rate_date DATE NOT NULL,
    rate_type VARCHAR(20) DEFAULT 'daily', -- daily, month_end, average
    rate DECIMAL(18,6) NOT NULL,
    source VARCHAR(50) NOT NULL, -- API source attribution
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    created_by VARCHAR(100),

    -- Ensure uniqueness
    UNIQUE(from_currency, to_currency, rate_date, rate_type),

    -- Performance indexes
    INDEX idx_rate_lookup (from_currency, to_currency, rate_date),
    INDEX idx_rate_date (rate_date)
);

-- Transaction table with embedded rate for audit trail
CREATE TABLE financial_transactions (
    id SERIAL PRIMARY KEY,
    transaction_date DATE NOT NULL,
    transaction_currency VARCHAR(3) NOT NULL,
    functional_currency VARCHAR(3) NOT NULL,
    amount_transaction DECIMAL(18,2) NOT NULL,
    exchange_rate DECIMAL(18,6) NOT NULL,
    amount_functional DECIMAL(18,2) NOT NULL,
    rate_source VARCHAR(50) NOT NULL,

    -- Link to rate master for audit
    rate_id INTEGER REFERENCES exchange_rates(id),

    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

import requests
import logging
from datetime import datetime, timedelta
from decimal import Decimal

class ExchangeRateUpdater:
    def __init__(self, api_key, base_url):
        self.api_key = api_key
        self.base_url = base_url
        self.logger = logging.getLogger(__name__)

    def update_daily_rates(self, base_currency='USD'):
        """
        Fetches and stores latest exchange rates.
        Designed to run daily via cron job.
        """
        try:
            # Fetch latest rates from API
            response = requests.get(
                f'{self.base_url}/api/latest',
                params={'base': base_currency},
                headers={'Authorization': f'Bearer {self.api_key}'},
                timeout=10
            )
            response.raise_for_status()

            data = response.json()
            rate_date = datetime.strptime(data['date'], '%Y-%m-%d').date()

            # Store rates with audit trail
            for currency_code, rate_value in data['rates'].items():
                self.save_rate(
                    from_currency=base_currency,
                    to_currency=currency_code,
                    rate_date=rate_date,
                    rate=Decimal(str(rate_value)),
                    source=data.get('source', 'API'),
                    rate_type='daily'
                )

            self.logger.info(f"Updated {len(data['rates'])} rates for {rate_date}")
            return True

        except requests.exceptions.RequestException as e:
            self.logger.error(f"API request failed: {e}")
            self.send_alert(f"Exchange rate update failed: {e}")
            return False
        except Exception as e:
            self.logger.error(f"Unexpected error: {e}")
            return False

    def get_historical_rate(self, from_currency, to_currency, transaction_date):
        """
        Retrieves historical rate for a specific date.
        Checks local database first, fetches from API if not found.
        """
        # Try local database first
        rate = self.get_rate_from_db(from_currency, to_currency, transaction_date)

        if rate is None:
            # Fetch from API and store
            rate = self.fetch_and_store_historical_rate(
                from_currency, to_currency, transaction_date
            )

        return rate

    def fetch_and_store_historical_rate(self, from_currency, to_currency, date):
        """
        Fetches historical rate from API and stores for future use.
        """
        try:
            response = requests.get(
                f'{self.base_url}/api/historical',
                params={
                    'date': date.strftime('%Y-%m-%d'),
                    'base': from_currency
                },
                headers={'Authorization': f'Bearer {self.api_key}'},
                timeout=10
            )
            response.raise_for_status()

            data = response.json()
            rate_value = data['rates'].get(to_currency)

            if rate_value:
                self.save_rate(
                    from_currency=from_currency,
                    to_currency=to_currency,
                    rate_date=date,
                    rate=Decimal(str(rate_value)),
                    source=data.get('source', 'API'),
                    rate_type='historical'
                )
                return Decimal(str(rate_value))

            return None

        except Exception as e:
            self.logger.error(f"Failed to fetch historical rate: {e}")
            return None

    def save_rate(self, from_currency, to_currency, rate_date,
                  rate, source, rate_type='daily'):
        """
        Saves exchange rate to database with duplicate handling.
        """
        # Implementation depends on your database layer
        # This is pseudocode - adapt to your ORM/database
        pass

    def send_alert(self, message):
        """
        Sends alert notification when rate updates fail.
        Critical for period-end processing reliability.
        """
        # Implementation: email, Slack, PagerDuty, etc.
        pass

# Cron job setup (example crontab entry):
# 0 6 * * * /usr/bin/python3 /path/to/update_rates.py >> /var/log/rate_updates.log 2>&1

// Node.js example: Update QuickBooks exchange rates
const QuickBooks = require('node-quickbooks');
const axios = require('axios');

async function updateQuickBooksRates(qbo, currencyApiKey) {
    try {
        // Fetch latest rates from currency API
        const response = await axios.get('https://api.example.com/latest', {
            headers: { 'Authorization': `Bearer ${currencyApiKey}` },
            params: { 'base': 'USD' }
        });

        const rates = response.data.rates;
        const rateDate = response.data.date;

        // Update each currency in QuickBooks
        for (const [currency, rate] of Object.entries(rates)) {
            const exchangeRate = {
                SourceCurrencyCode: 'USD',
                TargetCurrencyCode: currency,
                Rate: rate,
                AsOfDate: rateDate
            };

            // Create or update exchange rate in QuickBooks
            await qbo.createExchangeRate(exchangeRate);
            console.log(`Updated ${currency}: ${rate}`);
        }

        return { success: true, updated: Object.keys(rates).length };

    } catch (error) {
        console.error('Failed to update QuickBooks rates:', error);
        return { success: false, error: error.message };
    }
}

// Schedule daily updates
// Run via cron or task scheduler
updateQuickBooksRates(qboInstance, process.env.CURRENCY_API_KEY);

# Python example: Xero currency rate integration
from xero_python.api_client import ApiClient
from xero_python.accounting import AccountingApi
from datetime import datetime
import requests

class XeroRateUpdater:
    def __init__(self, xero_tenant_id, api_client, currency_api_key):
        self.tenant_id = xero_tenant_id
        self.accounting_api = AccountingApi(api_client)
        self.currency_api_key = currency_api_key

    def update_rates_for_period_end(self, period_end_date):
        """
        Updates Xero with period-end rates for financial close.
        Useful for locking rates for month/quarter/year-end.
        """
        # Fetch rates from currency API
        response = requests.get(
            'https://api.example.com/historical',
            params={'date': period_end_date.strftime('%Y-%m-%d')},
            headers={'Authorization': f'Bearer {self.currency_api_key}'}
        )
        rates_data = response.json()

        # Get Xero currencies
        currencies = self.accounting_api.get_currencies(self.tenant_id)

        for currency in currencies.currencies:
            if currency.code != 'USD':  # Skip base currency
                rate = rates_data['rates'].get(currency.code)

                if rate:
                    # Create currency rate in Xero
                    currency_rate = {
                        'currency': currency.code,
                        'rate': rate,
                        'effective_date': period_end_date
                    }

                    self.accounting_api.create_currency_rate(
                        self.tenant_id,
                        currency_rate
                    )

                    print(f"Set {currency.code} rate to {rate} for {period_end_date}")

        return True

/**
 * @NApiVersion 2.1
 * @NScriptType ScheduledScript
 * @NModuleScope SameAccount
 *
 * NetSuite Scheduled Script: Daily Exchange Rate Update
 */
define(['N/record', 'N/https', 'N/runtime'],
    function(record, https, runtime) {

        function execute(context) {
            try {
                // Get API credentials from script parameters
                var script = runtime.getCurrentScript();
                var apiKey = script.getParameter({name: 'custscript_currency_api_key'});
                var apiUrl = script.getParameter({name: 'custscript_currency_api_url'});

                // Fetch latest rates
                var response = https.get({
                    url: apiUrl + '/api/latest?base=USD',
                    headers: {
                        'Authorization': 'Bearer ' + apiKey,
                        'Content-Type': 'application/json'
                    }
                });

                var ratesData = JSON.parse(response.body);
                var effectiveDate = new Date(ratesData.date);

                // Update each currency in NetSuite
                for (var currencyCode in ratesData.rates) {
                    updateCurrencyRate(
                        'USD',
                        currencyCode,
                        ratesData.rates[currencyCode],
                        effectiveDate
                    );
                }

                log.audit('Rate Update Complete',
                    'Updated ' + Object.keys(ratesData.rates).length + ' currencies');

            } catch (e) {
                log.error('Rate Update Failed', e.toString());
            }
        }

        function updateCurrencyRate(fromCurrency, toCurrency, rate, effectiveDate) {
            try {
                var currencyRateRec = record.create({
                    type: record.Type.CURRENCY_RATE
                });

                currencyRateRec.setValue({
                    fieldId: 'basecurrency',
                    value: fromCurrency
                });

                currencyRateRec.setValue({
                    fieldId: 'transactioncurrency',
                    value: toCurrency
                });

                currencyRateRec.setValue({
                    fieldId: 'exchangerate',
                    value: rate
                });

                currencyRateRec.setValue({
                    fieldId: 'effectivedate',
                    value: effectiveDate
                });

                var recordId = currencyRateRec.save();
                log.debug('Rate Updated', toCurrency + ': ' + rate);

            } catch (e) {
                log.error('Rate Update Failed for ' + toCurrency, e.toString());
            }
        }

        return {
            execute: execute
        };
    }
);

# Complete multi-currency transaction recording system
from decimal import Decimal
from datetime import datetime
from typing import Optional

class MultiCurrencyTransactionManager:
    def __init__(self, db_connection, rate_api):
        self.db = db_connection
        self.rate_api = rate_api

    def record_transaction(self,
                          transaction_date: datetime.date,
                          transaction_currency: str,
                          functional_currency: str,
                          amount: Decimal,
                          account_id: int,
                          description: str) -> dict:
        """
        Records a transaction with proper currency handling.
        Returns transaction details including applied exchange rate.
        """
        # Get exchange rate for transaction date
        exchange_rate = self.get_transaction_rate(
            transaction_date,
            transaction_currency,
            functional_currency
        )

        # Calculate functional currency amount
        functional_amount = amount * exchange_rate

        # Create transaction record
        transaction = {
            'date': transaction_date,
            'currency': transaction_currency,
            'amount': amount,
            'functional_currency': functional_currency,
            'functional_amount': functional_amount,
            'exchange_rate': exchange_rate,
            'account_id': account_id,
            'description': description,
            'created_at': datetime.now()
        }

        # Save to database
        transaction_id = self.db.insert('transactions', transaction)

        # Record in audit log
        self.log_rate_usage(
            transaction_id,
            exchange_rate,
            transaction_date,
            transaction_currency,
            functional_currency
        )

        return {
            'transaction_id': transaction_id,
            'rate_applied': float(exchange_rate),
            'functional_amount': float(functional_amount)
        }

    def get_transaction_rate(self,
                            date: datetime.date,
                            from_currency: str,
                            to_currency: str) -> Decimal:
        """
        Gets appropriate exchange rate for transaction date.
        Handles weekends/holidays by using previous business day rate.
        """
        # Check database cache first
        cached_rate = self.db.query_one(
            """
            SELECT rate FROM exchange_rates
            WHERE from_currency = ?
            AND to_currency = ?
            AND rate_date = ?
            """,
            (from_currency, to_currency, date)
        )

        if cached_rate:
            return Decimal(str(cached_rate['rate']))

        # Fetch from API if not cached
        rate = self.rate_api.get_historical_rate(
            date, from_currency, to_currency
        )

        if rate is None:
            # Fallback to previous business day
            rate = self.get_previous_business_day_rate(
                date, from_currency, to_currency
            )

        # Cache for future use
        self.cache_rate(date, from_currency, to_currency, rate)

        return rate

    def revalue_open_balances(self,
                              as_of_date: datetime.date,
                              functional_currency: str) -> list:
        """
        Performs currency revaluation for open balances.
        Calculates unrealized gains/losses.
        """
        revaluation_entries = []

        # Get all open foreign currency balances
        open_balances = self.db.query(
            """
            SELECT
                account_id,
                currency,
                SUM(amount) as balance
            FROM transactions
            WHERE currency != ?
            AND settled = 0
            GROUP BY account_id, currency
            HAVING SUM(amount) != 0
            """,
            (functional_currency,)
        )

        for balance in open_balances:
            # Get original functional amount
            original_func_amount = self.get_original_functional_amount(
                balance['account_id'],
                balance['currency']
            )

            # Get current rate
            current_rate = self.get_transaction_rate(
                as_of_date,
                balance['currency'],
                functional_currency
            )

            # Calculate revalued amount
            revalued_amount = balance['balance'] * current_rate

            # Calculate unrealized gain/loss
            unrealized_gl = revalued_amount - original_func_amount

            if unrealized_gl != 0:
                revaluation_entry = {
                    'date': as_of_date,
                    'account_id': balance['account_id'],
                    'currency': balance['currency'],
                    'balance': balance['balance'],
                    'original_functional': original_func_amount,
                    'revalued_functional': revalued_amount,
                    'unrealized_gain_loss': unrealized_gl,
                    'rate_used': current_rate
                }

                revaluation_entries.append(revaluation_entry)

        return revaluation_entries

from decimal import Decimal
from datetime import datetime
from typing import List, Dict

class CurrencyRevaluationEngine:
    def __init__(self, db, rate_api, functional_currency='USD'):
        self.db = db
        self.rate_api = rate_api
        self.functional_currency = functional_currency

    def run_period_end_revaluation(self, period_end_date: datetime.date) -> Dict:
        """
        Executes complete period-end currency revaluation process.
        Returns summary of revaluation entries created.
        """
        print(f"Starting revaluation for period ending {period_end_date}")

        # Step 1: Identify accounts requiring revaluation
        revaluation_accounts = self.identify_revaluation_accounts()

        # Step 2: Get period-end exchange rates
        period_end_rates = self.get_period_end_rates(period_end_date)

        # Step 3: Calculate revaluations
        revaluation_entries = []
        total_unrealized_gain = Decimal('0')
        total_unrealized_loss = Decimal('0')

        for account in revaluation_accounts:
            entry = self.calculate_account_revaluation(
                account,
                period_end_date,
                period_end_rates
            )

            if entry:
                revaluation_entries.append(entry)

                if entry['unrealized_amount'] > 0:
                    total_unrealized_gain += entry['unrealized_amount']
                else:
                    total_unrealized_loss += abs(entry['unrealized_amount'])

        # Step 4: Create journal entries
        journal_entry_id = self.create_revaluation_journal_entry(
            revaluation_entries,
            period_end_date
        )

        # Step 5: Generate revaluation report
        summary = {
            'period_end_date': period_end_date,
            'accounts_revalued': len(revaluation_entries),
            'total_unrealized_gain': float(total_unrealized_gain),
            'total_unrealized_loss': float(total_unrealized_loss),
            'net_impact': float(total_unrealized_gain - total_unrealized_loss),
            'journal_entry_id': journal_entry_id,
            'entries': revaluation_entries
        }

        return summary

    def identify_revaluation_accounts(self) -> List[Dict]:
        """
        Identifies all accounts with foreign currency balances
        that require revaluation.
        """
        accounts = self.db.query("""
            SELECT
                a.account_id,
                a.account_name,
                a.account_type,
                b.currency,
                SUM(b.amount_currency) as balance_currency,
                SUM(b.amount_functional) as balance_functional_original
            FROM accounts a
            JOIN balances b ON a.account_id = b.account_id
            WHERE b.currency != ?
            AND a.requires_revaluation = 1
            AND b.period_end IS NULL  -- Open balances only
            GROUP BY a.account_id, b.currency
            HAVING SUM(b.amount_currency) != 0
        """, (self.functional_currency,))

        return accounts

    def calculate_account_revaluation(self,
                                     account: Dict,
                                     period_end_date: datetime.date,
                                     period_end_rates: Dict) -> Dict:
        """
        Calculates unrealized gain/loss for a single account.
        """
        currency = account['currency']
        balance_fc = Decimal(str(account['balance_currency']))

        # Get period-end rate
        period_end_rate = period_end_rates.get(currency)

        if not period_end_rate:
            print(f"Warning: No rate found for {currency} on {period_end_date}")
            return None

        # Calculate revalued functional currency amount
        revalued_amount = balance_fc * period_end_rate

        # Original functional currency amount
        original_amount = Decimal(str(account['balance_functional_original']))

        # Unrealized gain/loss
        unrealized_amount = revalued_amount - original_amount

        return {
            'account_id': account['account_id'],
            'account_name': account['account_name'],
            'currency': currency,
            'balance_fc': float(balance_fc),
            'period_end_rate': float(period_end_rate),
            'original_functional': float(original_amount),
            'revalued_functional': float(revalued_amount),
            'unrealized_amount': float(unrealized_amount),
            'unrealized_type': 'gain' if unrealized_amount > 0 else 'loss'
        }

    def create_revaluation_journal_entry(self,
                                        revaluation_entries: List[Dict],
                                        period_end_date: datetime.date) -> int:
        """
        Creates journal entry for currency revaluation.

        Debit: Asset accounts with losses / Liability accounts with gains
        Credit: Asset accounts with gains / Liability accounts with losses
        Offset: Unrealized Gain/Loss account
        """
        journal_entry = {
            'entry_date': period_end_date,
            'entry_type': 'currency_revaluation',
            'description': f'Currency revaluation as of {period_end_date}',
            'created_at': datetime.now()
        }

        # Create journal entry header
        je_id = self.db.insert('journal_entries', journal_entry)

        # Create lines for each revaluation
        for entry in revaluation_entries:
            unrealized_amount = Decimal(str(entry['unrealized_amount']))

            if unrealized_amount != 0:
                # Line for the balance sheet account
                self.db.insert('journal_entry_lines', {
                    'journal_entry_id': je_id,
                    'account_id': entry['account_id'],
                    'debit': float(unrealized_amount) if unrealized_amount > 0 else 0,
                    'credit': float(abs(unrealized_amount)) if unrealized_amount < 0 else 0,
                    'currency': self.functional_currency,
                    'description': f"Revaluation adjustment for {entry['currency']}"
                })

                # Offset line to unrealized gain/loss account
                unrealized_gl_account = self.get_unrealized_gl_account()
                self.db.insert('journal_entry_lines', {
                    'journal_entry_id': je_id,
                    'account_id': unrealized_gl_account,
                    'debit': float(abs(unrealized_amount)) if unrealized_amount < 0 else 0,
                    'credit': float(unrealized_amount) if unrealized_amount > 0 else 0,
                    'currency': self.functional_currency,
                    'description': f"Unrealized {'gain' if unrealized_amount > 0 else 'loss'} on {entry['currency']}"
                })

        return je_id

    def get_period_end_rates(self, period_end_date: datetime.date) -> Dict[str, Decimal]:
        """
        Fetches all required exchange rates for period end.
        """
        # Get list of currencies used
        currencies = self.db.query("""
            SELECT DISTINCT currency
            FROM balances
            WHERE currency != ?
        """, (self.functional_currency,))

        rates = {}
        for curr in currencies:
            currency_code = curr['currency']
            rate = self.rate_api.get_historical_rate(
                period_end_date,
                currency_code,
                self.functional_currency
            )

            if rate:
                rates[currency_code] = rate

        return rates