src/backtest_utils.py

"""Dual filter utilities — matches cleaned notebook implementation"""
import numpy as np

def dual_filter_backtest(preds, actuals, vols, ret30d, cost=0.0030, K=1.0,
                         vol_floor_pct=0.75, ret30d_thresh=0.05, extra_edge=0.0):
    """Exact dual filter logic from notebook"""
    vols = np.asarray(vols, dtype=float)
    ret30d = np.asarray(ret30d, dtype=float)
    finite_vols = vols[np.isfinite(vols)]
    vol_floor = np.nanpercentile(finite_vols, vol_floor_pct) if len(finite_vols) else np.nan
    
    trade_mask = (
        np.isfinite(vols) &
        np.isfinite(ret30d) &
        (np.abs(ret30d) > ret30d_thresh) &
        (vols > vol_floor)
    )
    
    preds = np.asarray(preds, dtype=float)
    actuals = np.asarray(actuals, dtype=float)
    valid = np.isfinite(preds) & np.isfinite(actuals) & trade_mask
    strong_signal = np.abs(preds) > (K * cost + extra_edge)
    pos = np.where(valid & strong_signal, np.sign(preds), 0.0)
    
    # Reuse notebook backtest result logic
    changes = np.abs(np.diff(pos, prepend=0.0))
    gross = pos * actuals
    net = gross - changes * cost
    eq = np.cumprod(1.0 + net) if len(net) else np.array([1.0])
    n_trades = int(np.sum(changes > 0))
    active = pos != 0
    
    sharpe = np.mean(net) / np.std(net) * np.sqrt(365) if np.std(net) > 0 else 0.0
    running_max = np.maximum.accumulate(eq)
    max_dd = np.min((eq - running_max) / running_max) if len(eq) > 0 else 0.0
    wr = np.mean(gross[active] > 0) if active.sum() else 0.0
    
    return {
        "sharpe": float(sharpe),
        "cum_return": float(eq[-1] - 1),
        "max_dd": float(max_dd),
        "wr": float(wr),
        "n_trades": n_trades,
        "pct_active": float(active.mean()),
        "eq": eq,
        "net": net,
        "pos": pos,
    }