backtesting/framework/test_1_3_monkey_test.py

import sys, os
import numpy as np
import pandas as pd
import warnings; warnings.filterwarnings('ignore')

from backtesting.framework.config import STRATEGY_NAME, ACTIVE_STRATEGY_FN, ACTIVE_PARAMS, load_data

try:
    from v30_causal_engine import evaluate_slice, CAP
    from v36_engine import SECTOR_MAP
except ImportError:
    from backtesting.strategies.v30_engine import evaluate_slice, CAP
    from backtesting.strategies.v36_engine import SECTOR_MAP

def run_v68_monkey(dc, spy, vf, daily_ret, rebal_days=40, vol_target=0.18,
                 riskoff_haircut=0.50, sma_lookback=200, mom_long=175,
                 mom_short=21, txn_bps=20, consistency_window=63,
                 top_n=15, use_dd_stop=True, **kwargs):
    """
    EXACT copy of V68 Soft Tranche architecture, but replaces the Z-Score 
    momentum sorting with pure random selection.
    """
    sma = spy.rolling(sma_lookback).mean()

    nav = CAP
    paper_nav = CAP
    peak_paper_nav = CAP
    trough_paper_nav = CAP
    stop_active = False

    pick_tks = []
    current_weights = pd.Series(dtype=float)
    port_rets = []
    hist = []
    txn_frac = txn_bps / 10000.0
    days = 0

    spy_vals = spy.values
    sma_vals = sma.values

    for i in range(1, len(dc)):
        if len(port_rets) >= 21:
            w_window = port_rets[-60:] if len(port_rets) >= 60 else port_rets[-21:]
            vs = vol_target / (np.std(w_window)*np.sqrt(252)+1e-8)
        else: vs = 0.5

        sp, sm = spy_vals[i-1], sma_vals[i-1]
        if pd.isna(sm) or sp <= sm: vs *= riskoff_haircut
        vs = float(np.clip(vs, 0.05, 1.0))

        day_ret = 0.0
        if pick_tks:
            lr = daily_ret.iloc[i][[t for t in pick_tks if t in daily_ret.columns]].dropna()
            if not lr.empty:
                wt = current_weights.reindex(lr.index).fillna(0)
                if wt.sum() > 0: wt = wt / wt.sum()
                day_ret = (lr * wt).sum() * vs

        paper_nav *= (1 + day_ret)
        if not stop_active:
            nav *= (1 + day_ret)

        port_rets.append(day_ret)
        hist.append(nav)

        # Drawdown Stop Logic
        peak_paper_nav = max(peak_paper_nav, paper_nav)
        paper_dd = (paper_nav / peak_paper_nav) - 1.0

        if use_dd_stop:
            if not stop_active:
                if paper_dd <= -0.15:
                    stop_active = True
                    trough_paper_nav = paper_nav
                    nav -= nav * txn_frac 
            else:
                trough_paper_nav = min(trough_paper_nav, paper_nav)
                if paper_nav >= trough_paper_nav * 1.05:
                    stop_active = False
                    peak_paper_nav = paper_nav
                    nav -= nav * txn_frac 

        days += 1
        if days >= rebal_days:
            days = 0

            # Get universe of valid stocks at T-1
            valid_tks = [t for t in vf if pd.notna(dc[t].iloc[i-1])]
            
            if not valid_tks:
                continue

            # MONKEY: Randomly select Top N
            if len(valid_tks) >= top_n:
                new_picks = list(np.random.choice(valid_tks, top_n, replace=False))
            else:
                new_picks = list(np.random.choice(valid_tks, len(valid_tks), replace=False))

            # True Turnover Cost
            if pick_tks and new_picks:
                swaps = len(set(new_picks) - set(pick_tks))
                turnover_cost = (swaps / top_n) * txn_frac
                nav -= nav * turnover_cost
                paper_nav -= paper_nav * turnover_cost

            if new_picks:
                current_weights = pd.Series(1.0/len(new_picks), index=new_picks)

            pick_tks = new_picks

    return pd.Series(hist, index=dc.index[1:len(hist)+1])

def run_test_1_3():
    print("=" * 80)
    print(f" TEST 1.3: V68 ARCHITECTURE MONKEY TEST - {STRATEGY_NAME}")
    print("=" * 80)

    dc, spy, vf, daily_ret = load_data()

    # 1. Evaluate Active Strategy
    print("Evaluating Active V68 Strategy...")
    c_orig = ACTIVE_STRATEGY_FN(dc, spy, vf, daily_ret, **ACTIVE_PARAMS)
    if isinstance(c_orig, dict) and 'curve' in c_orig:
        c_orig = c_orig['curve']
    m_orig = evaluate_slice(c_orig, "2008-01-01", "2025-12-31")
    orig_sharpe = m_orig['sharpe']
    print(f"  Active Sharpe: {orig_sharpe:.4f}")

    # 2. Evaluate 50 Random Monkey Iterations (Inside V68 Framework)
    print("\nRunning 50 Random Monkey Iterations (Apples-to-Apples)...")
    np.random.seed(42) # Fixed seed for reproducibility
    rand_sharpes = []
    
    # Pre-calculate to speed up monkey tests (since we run 50 times)
    for step in range(50):
        c_rand = run_v68_monkey(dc, spy, vf, daily_ret, **ACTIVE_PARAMS)
        m_rand = evaluate_slice(c_rand, "2008-01-01", "2025-12-31")
        rand_sharpes.append(m_rand['sharpe'])
        if step % 10 == 0:
            print(".", end="", flush=True)
    print()

    r_med = np.median(rand_sharpes)
    r_max = np.max(rand_sharpes)

    print("-" * 80)
    print(f"  Random-Monkey Median: {r_med:.4f}")
    print(f"  Random-Monkey Max:    {r_max:.4f}")

    excess_monkey = orig_sharpe - r_med
    print(f"  Excess vs Monkey Median: {excess_monkey:+.4f}")

    if excess_monkey > 0.15:
        print("  VERDICT: PASS (Signal adds strong value beyond V68 risk management)")
    elif excess_monkey > 0.05:
        print("  VERDICT: WEAK PASS (Signal adds marginal value)")
    else:
        print("  VERDICT: FAIL (Signal has no mathematical edge over random selection inside V68)")

if __name__ == "__main__":
    run_test_1_3()