import numpy as np
import pandas as pd
from scipy.stats import pearsonr
def calculate_currency_correlation_matrix(returns_dict):
    """
    Calculate correlation between currency pairs
    Returns_dict: {'EUR/USD': returns, 'GBP/USD': returns, ...}
    """
    pairs = list(returns_dict.keys())
    n_pairs = len(pairs)
    correlation_matrix = np.zeros((n_pairs, n_pairs))
for i, pair1 in enumerate(pairs):
        for j, pair2 in enumerate(pairs):
            if i == j:
                correlation_matrix[i, j] = 1.0
            else:
                corr, _ = pearsonr(returns_dict[pair1], returns_dict[pair2])
                correlation_matrix[i, j] = corr
return pd.DataFrame(
        correlation_matrix,
        index=pairs,
        columns=pairs
    )
Example: Monitor correlation to avoid redundant positions
correlations = calculate_currency_correlation_matrix({
    'EUR/USD': eur_usd_returns,
    'GBP/USD': gbp_usd_returns,
    'USD/JPY': usd_jpy_returns
})
print(correlations)
EUR/USD and GBP/USD have 0.92 correlation

Adding both is redundant; choose one

def adjust_backtest_for_overnight_gaps(prices, is_friday_close, is_monday_open):
    """
    Simulate overnight gaps in forex backtests
    Gaps are 1-2% for major pairs (EUR/USD, GBP/USD)
    """
    gap_adjusted_prices = prices.copy()
for i in range(len(prices)):
        if is_friday_close[i]:
            # Friday close
            next_monday_idx = i + 1
            while next_monday_idx < len(prices) and not is_monday_open[next_monday_idx]:
                next_monday_idx += 1
if next_monday_idx < len(prices):
                # Simulate 1-2% gap (mean = 0.8%, std = 0.4%)
                gap_size = np.random.normal(0.008, 0.004)
                gap_adjusted_prices[next_monday_idx] *= (1 + gap_size)
return gap_adjusted_prices
In backtest, check for Friday close bars
for i in range(len(prices)):
    day_of_week = get_day_of_week(dates[i])
    if day_of_week == 4:  # Friday
        # Apply weekend gap to Monday's price
        prices[i+2] *= np.random.normal(1.008, 0.004)

class ForexLeverageController:
    """Manage leverage and margin requirements"""
def __init__(
        self,
        account_size=10000,
        max_leverage=50,
        margin_requirement=0.02  # 2% for leverage 50:1
    ):
        self.account_size = account_size
        self.max_leverage = max_leverage
        self.margin_requirement = margin_requirement
        self.used_margin = 0
        self.open_positions = {}
def calculate_position_size(
        self,
        entry_price,
        stop_loss_price,
        pair='EUR/USD',
        use_leverage=1.0
    ):
        """
        Size position accounting for margin requirement
        """
        # 1 micro lot = 0.001 lots = 1000 units of base currency
        # For EUR/USD at 1.20: 1 micro lot = 1.20 USD notional value
micro_lot_value = entry_price * 1000  # Units of base currency
        risk_per_lot = abs(entry_price - stop_loss_price) * 1000
# Risk max 2% of account
        risk_amount = self.account_size * 0.02
        position_size_lots = risk_amount / (stop_loss_price - entry_price)
# Apply leverage cap
        notional_exposure = position_size_lots  micro_lot_value  1000
        max_exposure = self.account_size * self.max_leverage
if notional_exposure > max_exposure:
            # Reduce position for leverage constraint
            position_size_lots = max_exposure / (micro_lot_value * 1000)
# Calculate margin requirement
        margin_required = position_size_lots  micro_lot_value  self.margin_requirement
available_margin = self.account_size - self.used_margin
if margin_required > available_margin:
            # Insufficient margin for full position
            position_size_lots = available_margin / (micro_lot_value * self.margin_requirement)
return {
            'micro_lots': position_size_lots,
            'notional_value': position_size_lots  micro_lot_value  1000,
            'margin_required': position_size_lots  micro_lot_value  self.margin_requirement,
            'available_leverage': max_exposure / (position_size_lots  micro_lot_value  1000)
        }
def check_margin_call(self, unrealized_loss):
        """Check if drawdown triggers margin call"""
        available_equity = self.account_size - unrealized_loss
        total_notional = sum(pos['notional'] for pos in self.open_positions.values())
        total_margin_required = total_notional * self.margin_requirement
if available_equity < total_margin_required:
            return True  # Margin call triggered
return False

def model_forex_execution(
    price,
    bid_ask_spread_pips=1.2,
    pair='EUR/USD',
    is_buy=True,
    slippage_pips=0.5
):
    """
    Model realistic forex execution with spread and slippage
    Spread is 1-2 pips for major pairs (e.g., EUR/USD)
    Slippage adds 0-2 pips depending on liquidity and volatility
    """
    pip_value = 0.0001 if pair.endswith('JPY') else 0.0001
# Bid-ask spread
    bid = price - (bid_ask_spread_pips  0.5  pip_value)
    ask = price + (bid_ask_spread_pips  0.5  pip_value)
# Add slippage
    slippage = slippage_pips * pip_value
if is_buy:
        execution_price = ask + slippage
    else:
        execution_price = bid - slippage
return {
        'bid': bid,
        'ask': ask,
        'execution_price': execution_price,
        'cost_pips': (execution_price - price) / pip_value
    }
Example
EUR_price = 1.0950
execution = model_forex_execution(EUR_price, bid_ask_spread_pips=1.2, is_buy=True)
print(f"Bid: {execution['bid']:.4f}")
print(f"Ask: {execution['ask']:.4f}")
print(f"Execution: {execution['execution_price']:.4f}")
print(f"Cost: {execution['cost_pips']:.1f} pips")

class ForexRiskManagedBacktest:
    """Backtester optimized for forex trading"""
def __init__(
        self,
        prices,
        pairs,
        signals,
        initial_account=10000,
        max_leverage=50,
        max_daily_loss_pct=0.05,
        max_correlation=0.8
    ):
        self.prices = prices  # Dict of price series by pair
        self.pairs = pairs
        self.signals = signals
        self.account = initial_account
        self.max_leverage = max_leverage
        self.max_daily_loss_pct = max_daily_loss_pct
        self.max_correlation = max_correlation
self.leverage_controller = ForexLeverageController(
            account_size=initial_account,
            max_leverage=max_leverage
        )
self.trades = []
        self.equity_curve = [initial_account]
        self.daily_loss = 0
def calculate_portfolio_correlation_risk(self):
        """Ensure correlated positions don't create hidden leverage"""
        if len(self.leverage_controller.open_positions) < 2:
            return True
# Get recent returns for each open position
        position_returns = {}
        for pair, position in self.leverage_controller.open_positions.items():
            idx = position['index']
            pair_prices = self.prices[pair]
            recent_returns = np.diff(pair_prices[max(0, idx-20):idx]) / pair_prices[max(0, idx-21):idx-1]
            position_returns[pair] = recent_returns
# Check correlations
        pairs_list = list(position_returns.keys())
        for i in range(len(pairs_list)):
            for j in range(i + 1, len(pairs_list)):
                corr = np.corrcoef(position_returns[pairs_list[i]], position_returns[pairs_list[j]])[0, 1]
                if corr > self.max_correlation:
                    print(f"WARNING: {pairs_list[i]} and {pairs_list[j]} correlation {corr:.2f} > {self.max_correlation}")
                    return False
return True
def run(self):
        """Execute forex backtest with risk controls"""
        position = None
for pair in self.pairs:
            for i in range(1, len(self.signals[pair])):
                signal = self.signals[pair][i]
                price = self.prices[pair][i]
if signal == 0:
                    continue
# Check correlation before adding position
                if not self.calculate_portfolio_correlation_risk():
                    continue
# Calculate position size with leverage limits
                entry_price = price
                stop_loss = entry_price  0.995 if signal == 1 else entry_price  1.005
sizing = self.leverage_controller.calculate_position_size(
                    entry_price=entry_price,
                    stop_loss_price=stop_loss,
                    pair=pair,
                    use_leverage=1.0
                )
# Check margin call risk
                if self.leverage_controller.check_margin_call(0):
                    print(f"Insufficient margin for {pair}")
                    continue
# Apply spread and slippage
                execution = model_forex_execution(
                    price,
                    bid_ask_spread_pips=1.2,
                    pair=pair,
                    is_buy=(signal == 1)
                )
position = {
                    'pair': pair,
                    'signal': signal,
                    'entry': execution['execution_price'],
                    'stop': stop_loss,
                    'size_lots': sizing['micro_lots'],
                    'index': i
                }
return self.equity_curve, self.trades
def metrics(self):
        """Calculate forex-specific metrics"""
        if not self.trades:
            return {}
returns = np.array([t['return'] for t in self.trades])
return {
            'total_return': (self.account - 10000) / 10000,
            'sharpe_ratio': np.mean(returns) / np.std(returns) * np.sqrt(252) if np.std(returns) > 0 else 0,
            'max_drawdown': np.min(np.cumprod(1 + returns) / np.maximum.accumulate(np.cumprod(1 + returns)) - 1),
            'win_rate': (returns > 0).sum() / len(returns),
            'num_trades': len(self.trades),
            'avg_pips_per_trade': np.mean([abs(t['exit'] - t['entry']) / 0.0001 for t in self.trades])
        }