Data Transformation

Tools and patterns for transforming financial and economic data.

Balance of Payments Transformations

import pandas as pd
import numpy as np

def calculate_bop_components(df):
    """Calculate all BOP components from raw data."""

    # Trade balance
    df['trade_balance_goods'] = df['exports_goods'] - df['imports_goods']
    df['trade_balance_services'] = df['exports_services'] - df['imports_services']
    df['trade_balance_total'] = df['trade_balance_goods'] + df['trade_balance_services']

    # Income balance
    df['primary_income_net'] = df['primary_income_credit'] - df['primary_income_debit']

    # Transfers (secondary income)
    df['secondary_income_net'] = df['remittances_inflow'] + df['official_transfers']

    # Current account
    df['current_account'] = (
        df['trade_balance_total'] +
        df['primary_income_net'] +
        df['secondary_income_net']
    )

    return df

Percentage Calculations

def calculate_percentages(df):
    """Calculate key ratios and percentages."""

    total_exports = df['exports_goods'] + df['exports_services']
    total_imports = df['imports_goods'] + df['imports_services']

    # CA as percentage of exports
    df['ca_as_pct_exports'] = (df['current_account'] / total_exports) * 100

    # Trade balance as percentage of GDP (if GDP available)
    # df['tb_as_pct_gdp'] = (df['trade_balance_total'] / df['gdp']) * 100

    # Export/Import coverage ratio
    df['export_import_ratio'] = (total_exports / total_imports) * 100

    return df

Currency Conversion

def convert_currency(values, exchange_rate, direction='to_usd'):
    """
    Convert between currencies.

    Parameters:
    - values: Amount to convert
    - exchange_rate: Exchange rate (local currency per USD)
    - direction: 'to_usd' or 'from_usd'
    """
    if direction == 'to_usd':
        return values / exchange_rate
    else:
        return values * exchange_rate

Period Calculations

def calculate_period_string(df):
    """Generate period string like '2018Q1-2024Q4'."""
    first_date = df.index.min()
    last_date = df.index.max()

    first_q = f"{first_date.year}Q{first_date.quarter}"
    last_q = f"{last_date.year}Q{last_date.quarter}"

    return f"{first_q}-{last_q}"

Summary Statistics Generator

def generate_bop_summary(df):
    """Generate summary statistics for BOP analysis."""

    total_exports = df['exports_goods'] + df['exports_services']
    ca_pct = (df['current_account'] / total_exports) * 100

    # Determine trend from seasonal decomposition
    from statsmodels.tsa.seasonal import seasonal_decompose
    result = seasonal_decompose(df['current_account'], model='additive', period=4)
    trend = result.trend.dropna()
    trend_direction = "improving" if trend.iloc[-1] > trend.iloc[0] else "deteriorating"

    # Find seasonal peak
    seasonal_by_q = df.groupby(df.index.quarter)[df.columns[0]].count()
    # Use original seasonal component
    seasonal_avg = result.seasonal.groupby(result.seasonal.index.quarter).mean()
    peak_q = f"Q{seasonal_avg.idxmax()}"

    return {
        "period": calculate_period_string(df),
        "current_account_avg_mn_usd": round(df['current_account'].mean(), 1),
        "trade_balance_avg": round(df['trade_balance_total'].mean(), 1),
        "remittances_avg": round(df['remittances_inflow'].mean(), 1),
        "ca_as_pct_exports_avg": round(ca_pct.mean(), 1),
        "trend_direction": trend_direction,
        "seasonality_peak_quarter": peak_q
    }

Data Cleaning

def clean_bop_data(df):
    """Clean and prepare BOP data for analysis."""

    # Handle missing values
    df = df.fillna(method='ffill').fillna(method='bfill')

    # Ensure datetime index
    df.index = pd.to_datetime(df.index)

    # Sort by date
    df = df.sort_index()

    # Remove duplicates
    df = df[~df.index.duplicated(keep='first')]

    return df