fx_quant_engine/fx_quant_engine/backtest/engine.py

from __future__ import annotations

import numpy as np
import pandas as pd

from fx_quant_engine.schemas import BacktestResult


class BacktestEngine:
    def __init__(self, transaction_cost_bps: float = 1.5, slippage_bps: float = 1.0) -> None:
        self.tc = transaction_cost_bps / 10000.0
        self.slippage = slippage_bps / 10000.0

    def simulate_pair(self, prices: pd.Series, signal_strength: pd.Series) -> pd.Series:
        returns = prices.pct_change().fillna(0.0)
        position = signal_strength.shift(1).fillna(0.0)  # no lookahead
        traded = position.diff().abs().fillna(0.0)
        net = position * returns - traded * (self.tc + self.slippage)
        return net

    def simulate_portfolio(self, pair_returns: dict[str, pd.Series]) -> BacktestResult:
        if not pair_returns:
            return BacktestResult({}, 0.0, 0.0, 0.0, {})

        df = pd.DataFrame(pair_returns).fillna(0.0)
        port = df.mean(axis=1)
        equity = (1.0 + port).cumprod()
        dd = equity / equity.cummax() - 1.0

        pair_level = {k: float(v.mean() * 252.0) for k, v in pair_returns.items()}
        metrics = {
            "annualized_return": float(port.mean() * 252.0),
            "annualized_vol": float(port.std() * np.sqrt(252.0)),
            "sharpe": float((port.mean() / (port.std() + 1e-9)) * np.sqrt(252.0)),
        }

        return BacktestResult(
            pair_level_returns=pair_level,
            portfolio_return=float((equity.iloc[-1] - 1.0) if len(equity) else 0.0),
            max_drawdown=float(dd.min() if len(dd) else 0.0),
            turnover=float(df.diff().abs().sum().sum()),
            metrics=metrics,
        )

    def walk_forward(self, prices: pd.Series, signal_strength: pd.Series, train_window: int = 120, test_window: int = 20) -> pd.Series:
        out = []
        idx = prices.index
        i = train_window
        while i < len(idx):
            end = min(i + test_window, len(idx))
            segment = self.simulate_pair(prices.iloc[:end], signal_strength.iloc[:end])
            out.append(segment.iloc[i:end])
            i += test_window
        return pd.concat(out).sort_index() if out else pd.Series(dtype=float)