Backtest Specification¶

# src/evaluation/backtester.py

class BacktestConfig:
    # Timing
    signal_delay_bars: int = 1        # Signal on bar N, execute on bar N+1
    execution_price: str = 'open'     # 'open', 'close', 'vwap', 'worst_case'

    # Position
    max_positions: int = 1            # Concurrent positions
    position_sizing: str = 'fixed'    # 'fixed', 'risk_pct', 'kelly'

    # Costs (applied per trade)
    apply_commission: bool = True
    apply_spread: bool = True
    apply_slippage: bool = True

    # Validation
    run_anti_bias_checks: bool = True
    require_oos_validation: bool = True

def get_spread_cost(symbol, volatility_regime='normal'):
    """
    Return spread in price units.

    Spreads widen during:
    - News events (2-5x)
    - Low liquidity periods (1.5-2x)
    - High volatility (1.5-3x)
    """
    base_spreads = {
        'EURUSD': 0.00010,  # 1.0 pip
        'GBPUSD': 0.00012,  # 1.2 pips
        'USDJPY': 0.012,    # 1.2 pips
        'USDCAD': 0.00015,  # 1.5 pips
        'BTCUSD': 5.0,      # $5
        'ETHUSD': 0.50,     # $0.50
        'NQ': 0.50,         # 0.5 points
        'GC': 0.30,         # $0.30
    }

    multipliers = {
        'normal': 1.0,
        'elevated': 1.5,
        'high': 2.5,
        'extreme': 5.0
    }

    return base_spreads.get(symbol, 0.0001) * multipliers[volatility_regime]

def calculate_slippage(symbol, order_type, volatility, position_size):
    """
    Calculate expected slippage.

    Factors:
    - Base slippage per instrument
    - Volatility multiplier (ATR-based)
    - Size impact (larger orders = more slippage)
    - Order type (market vs limit)
    """
    base_slippage = get_base_slippage(symbol)

    # Volatility adjustment
    vol_mult = max(1.0, volatility / get_avg_volatility(symbol))

    # Size impact (only for larger positions)
    size_mult = 1.0 + max(0, (position_size / get_avg_volume(symbol)) - 0.01)

    # Limit orders get 50% less slippage on average
    order_mult = 0.5 if order_type == 'limit' else 1.0

    return base_slippage * vol_mult * size_mult * order_mult

# Conservative assumption: No partial fills modeled
# All orders either fully filled or not filled

fill_rate_assumptions = {
    'market': 1.00,     # Always fills
    'limit': 0.80,      # 80% fill rate for limits at touch
    'stop': 0.95,       # 95% for stops (gaps can skip)
    'stop_limit': 0.70  # 70% for stop-limits
}

def test_no_lookahead(strategy, data):
    """
    Test that strategy doesn't use future data.

    Method:
    1. Run strategy on data[0:N]
    2. Run strategy on data[0:N+M] where M > 0
    3. Signals for bars 0:N must be identical

    If signals change, lookahead is present.
    """
    signals_original = strategy.generate_signals(data[:1000])
    signals_extended = strategy.generate_signals(data[:1500])

    # Compare signals for first 1000 bars
    assert signals_original.equals(signals_extended[:1000]), \
        "FAIL: Lookahead detected - signals changed with future data"

def test_overfitting(strategy, data, n_shuffles=100):
    """
    Compare strategy to randomized versions.

    Method:
    1. Run strategy, record Sharpe
    2. Shuffle signal timing randomly, run N times
    3. Strategy should beat 95% of random versions

    If not, edge may be spurious.
    """
    real_sharpe = strategy.backtest(data).sharpe_ratio

    random_sharpes = []
    for _ in range(n_shuffles):
        shuffled = shuffle_signals(strategy, data)
        random_sharpes.append(shuffled.backtest(data).sharpe_ratio)

    percentile = (np.array(random_sharpes) < real_sharpe).mean()
    assert percentile >= 0.95, \
        f"FAIL: Strategy only beats {percentile*100:.0f}% of random"

def walk_forward_validation(strategy, data, n_windows=6):
    """
    Rolling window validation.

    Parameters:
    - Training window: 12 months
    - Testing window: 2 months
    - Step: 2 months (non-overlapping test)
    - Total windows: 6

    Returns:
    - Per-window performance
    - Aggregate statistics
    - Consistency score (% of profitable windows)
    """
    window_results = []

    for i in range(n_windows):
        train_start = i * 2  # months
        train_end = train_start + 12
        test_start = train_end
        test_end = test_start + 2

        train_data = data[train_start:train_end]
        test_data = data[test_start:test_end]

        # Fit on train, evaluate on test
        strategy.fit(train_data)
        result = strategy.backtest(test_data)
        window_results.append(result)

    return aggregate_results(window_results)

class BacktestResult:
    # Returns
    total_return: float
    cagr: float
    sharpe_ratio: float
    sortino_ratio: float
    calmar_ratio: float

    # Risk
    max_drawdown: float
    avg_drawdown: float
    max_drawdown_duration: int  # bars

    # Trades
    total_trades: int
    win_rate: float
    avg_win: float
    avg_loss: float
    profit_factor: float
    expectancy: float

    # Validation
    is_sharpe: float
    oos_sharpe: float
    degradation_pct: float
    walk_forward_consistency: float
    anti_bias_passed: bool