server/backtester.py

"""
Backtesting engine for the Quant Research Environment.

Ported from the RAETH Trading Eval verifier. Replays trade logs
against price series with transaction costs, computes Sharpe,
drawdown, and exposure metrics.
"""

import numpy as np
import pandas as pd

TRANSACTION_COST = 0.00009  # 0.9 bps
BARS_PER_DAY = 375
ANNUALIZATION_FACTOR = np.sqrt(252 * BARS_PER_DAY)
MAX_NET_EXPOSURE_RATIO = 0.80


def compute_exposure_ratio(
    nifty_pos: float, bn_pos: float, nifty_close: float, bn_close: float
) -> float:
    """Compute net exposure ratio at a given bar."""
    nifty_notional = nifty_pos * nifty_close
    bn_notional = bn_pos * bn_close
    gross = abs(nifty_notional) + abs(bn_notional)
    if gross == 0:
        return 0.0
    return abs(nifty_notional + bn_notional) / gross


def replay_trades_single(
    trades_df: pd.DataFrame, close_series: pd.Series
) -> dict:
    """
    Replay a single-instrument trade log against close prices.

    Parameters
    ----------
    trades_df : DataFrame with columns [bar, position].
    close_series : Series of close prices indexed 0..N-1.

    Returns
    -------
    dict with total_trades, total_pnl, max_drawdown, annualized_sharpe, bar_pnls.
    """
    n_bars = len(close_series)

    trade_map = {}
    for _, row in trades_df.iterrows():
        trade_map[int(row["bar"])] = row["position"]

    position = 0.0
    running_pnl = 0.0
    peak_pnl = 0.0
    max_drawdown = 0.0
    total_trades = 0
    bar_pnls = []

    for i in range(n_bars):
        bar_pnl = 0.0
        if i > 0:
            bar_pnl = position * (close_series.iloc[i] - close_series.iloc[i - 1])

        if i in trade_map:
            new_position = trade_map[i]
            if new_position != position:
                total_trades += 1
                bar_pnl -= TRANSACTION_COST * close_series.iloc[i] * abs(
                    new_position - position
                )
                position = new_position

        running_pnl += bar_pnl
        bar_pnls.append(bar_pnl)

        if running_pnl > peak_pnl:
            peak_pnl = running_pnl
        dd = peak_pnl - running_pnl
        if dd > max_drawdown:
            max_drawdown = dd

    bar_pnls_arr = np.array(bar_pnls)
    if len(bar_pnls_arr) > 1 and np.std(bar_pnls_arr) > 0:
        annualized_sharpe = (
            np.mean(bar_pnls_arr) / np.std(bar_pnls_arr) * ANNUALIZATION_FACTOR
        )
    elif len(bar_pnls_arr) > 1 and np.mean(bar_pnls_arr) > 0:
        annualized_sharpe = 99.0  # Zero variance with positive mean → very high Sharpe
    elif len(bar_pnls_arr) > 1 and np.mean(bar_pnls_arr) < 0:
        annualized_sharpe = -99.0  # Zero variance with negative mean → very low Sharpe
    else:
        annualized_sharpe = 0.0

    return {
        "total_trades": total_trades,
        "total_pnl": round(running_pnl, 2),
        "max_drawdown": round(max_drawdown, 2),
        "annualized_sharpe": round(annualized_sharpe, 4),
        "bar_pnls": bar_pnls,
    }


def replay_trades_multi(
    trades_df: pd.DataFrame,
    nifty_close: pd.Series,
    banknifty_close: pd.Series,
) -> dict:
    """
    Replay a multi-instrument (spread) trade log.

    Parameters
    ----------
    trades_df : DataFrame with columns [bar, nifty_position, banknifty_position].
    nifty_close, banknifty_close : Series of close prices indexed 0..N-1.

    Returns
    -------
    dict with total_trades_per_leg, total_spread_entries, total_pnl,
    max_drawdown, annualized_sharpe, bar_pnls, max_net_exposure_ratio,
    exposure_violations.
    """
    n_bars = min(len(nifty_close), len(banknifty_close))

    trade_map = {}
    for _, row in trades_df.iterrows():
        trade_map[int(row["bar"])] = (
            float(row["nifty_position"]),
            float(row["banknifty_position"]),
        )

    nifty_pos = 0.0
    bn_pos = 0.0
    running_pnl = 0.0
    peak_pnl = 0.0
    max_drawdown = 0.0
    total_trades_nifty = 0
    total_trades_bn = 0
    spread_entries = 0
    bar_pnls = []

    max_net_exposure = 0.0
    exposure_violations = []

    for i in range(n_bars):
        bar_pnl = 0.0
        if i > 0:
            bar_pnl = nifty_pos * (
                nifty_close.iloc[i] - nifty_close.iloc[i - 1]
            ) + bn_pos * (banknifty_close.iloc[i] - banknifty_close.iloc[i - 1])

        if i in trade_map:
            new_nifty_pos, new_bn_pos = trade_map[i]
            old_nifty_pos, old_bn_pos = nifty_pos, bn_pos

            if new_nifty_pos != nifty_pos:
                total_trades_nifty += 1
                bar_pnl -= TRANSACTION_COST * nifty_close.iloc[i] * abs(
                    new_nifty_pos - nifty_pos
                )
                nifty_pos = new_nifty_pos

            if new_bn_pos != bn_pos:
                total_trades_bn += 1
                bar_pnl -= TRANSACTION_COST * banknifty_close.iloc[i] * abs(
                    new_bn_pos - bn_pos
                )
                bn_pos = new_bn_pos

            both_were_flat = old_nifty_pos == 0 and old_bn_pos == 0
            any_now_active = nifty_pos != 0 or bn_pos != 0
            if both_were_flat and any_now_active:
                spread_entries += 1

        ratio = compute_exposure_ratio(
            nifty_pos, bn_pos, nifty_close.iloc[i], banknifty_close.iloc[i]
        )
        if ratio > max_net_exposure:
            max_net_exposure = ratio
        if ratio > MAX_NET_EXPOSURE_RATIO:
            exposure_violations.append({"bar": i, "ratio": round(ratio, 4)})

        running_pnl += bar_pnl
        bar_pnls.append(bar_pnl)

        if running_pnl > peak_pnl:
            peak_pnl = running_pnl
        dd = peak_pnl - running_pnl
        if dd > max_drawdown:
            max_drawdown = dd

    total_trades_per_leg = max(total_trades_nifty, total_trades_bn)

    bar_pnls_arr = np.array(bar_pnls)
    if len(bar_pnls_arr) > 1 and np.std(bar_pnls_arr) > 0:
        annualized_sharpe = (
            np.mean(bar_pnls_arr) / np.std(bar_pnls_arr) * ANNUALIZATION_FACTOR
        )
    elif len(bar_pnls_arr) > 1 and np.mean(bar_pnls_arr) > 0:
        annualized_sharpe = 99.0
    elif len(bar_pnls_arr) > 1 and np.mean(bar_pnls_arr) < 0:
        annualized_sharpe = -99.0
    else:
        annualized_sharpe = 0.0

    return {
        "total_trades_per_leg": total_trades_per_leg,
        "total_spread_entries": spread_entries,
        "total_pnl": round(running_pnl, 2),
        "max_drawdown": round(max_drawdown, 2),
        "annualized_sharpe": round(annualized_sharpe, 4),
        "bar_pnls": bar_pnls,
        "max_net_exposure_ratio": round(max_net_exposure, 4),
        "exposure_violations": exposure_violations,
    }