Spaces:
Running
Running
File size: 5,582 Bytes
1cd56b6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | 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() |