137 lines
5.4 KiB
Python
137 lines
5.4 KiB
Python
"""
|
||
Evaluate interaction/ensemble strategies on 1/3/5/10y PIT windows with a
|
||
proper 500-day warmup preload, so 252d-warmup strategies are active from the
|
||
measurement start.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import warnings
|
||
|
||
import pandas as pd
|
||
|
||
import research.pit_backtest as pit
|
||
from research.alpha_factors import AlphaFactorStrategy
|
||
from research.interaction_alpha import (MultiplicativeFactorStrategy,
|
||
SubStrategyEnsemble, VotingFactorStrategy,
|
||
default_ensemble)
|
||
from strategies.factor_combo import FactorComboStrategy
|
||
from strategies.recovery_momentum import RecoveryMomentumStrategy
|
||
|
||
warnings.filterwarnings("ignore", category=FutureWarning)
|
||
warnings.filterwarnings("ignore", category=RuntimeWarning)
|
||
|
||
DATA_DIR = "data"
|
||
BENCHMARK = "SPY"
|
||
|
||
|
||
def load():
|
||
raw = pit.load_pit_prices()
|
||
masked = pit.pit_universe(raw)
|
||
masked[BENCHMARK] = raw[BENCHMARK]
|
||
return masked
|
||
|
||
|
||
def warmup_slice(df, years, warmup_days=500):
|
||
measurement_start = df.index[-1] - pd.DateOffset(years=years)
|
||
cutoff = max(df.index[0], measurement_start - pd.Timedelta(days=warmup_days * 1.5))
|
||
return df[df.index >= cutoff], measurement_start
|
||
|
||
|
||
def measure(eq, start, name=""):
|
||
eq = eq[eq.index >= start]
|
||
eq = eq / eq.iloc[0] * 10_000
|
||
return pit.summarize(eq, name=name)
|
||
|
||
|
||
def make_configs(mkt_ret):
|
||
pair = ["mom_12_1", "recovery_63"]
|
||
return {
|
||
"Mult(mom12×rec63) eq, tn=10":
|
||
lambda: MultiplicativeFactorStrategy(
|
||
factor_names=pair, top_n=10, rebal_freq=21, mkt_returns=mkt_ret),
|
||
"Mult(mom12×rec63) eq, tn=15":
|
||
lambda: MultiplicativeFactorStrategy(
|
||
factor_names=pair, top_n=15, rebal_freq=21, mkt_returns=mkt_ret),
|
||
"Mult(mom12×rec63) eq, rebal=10":
|
||
lambda: MultiplicativeFactorStrategy(
|
||
factor_names=pair, top_n=10, rebal_freq=10, mkt_returns=mkt_ret),
|
||
"Mult(mom12×rec63) sig^2, tn=15":
|
||
lambda: MultiplicativeFactorStrategy(
|
||
factor_names=pair, top_n=15, rebal_freq=21, mkt_returns=mkt_ret,
|
||
weighting="signal", signal_concentration=2.0),
|
||
"Mult(mom12×rec63) sig^4, tn=15":
|
||
lambda: MultiplicativeFactorStrategy(
|
||
factor_names=pair, top_n=15, rebal_freq=21, mkt_returns=mkt_ret,
|
||
weighting="signal", signal_concentration=4.0),
|
||
"Mult(mom12×rec63) disp-scale":
|
||
lambda: MultiplicativeFactorStrategy(
|
||
factor_names=pair, top_n=10, rebal_freq=21, mkt_returns=mkt_ret,
|
||
dispersion_scale=True),
|
||
"Mult(mom12×rec63) inv_vol":
|
||
lambda: MultiplicativeFactorStrategy(
|
||
factor_names=pair, top_n=10, rebal_freq=21, mkt_returns=mkt_ret,
|
||
weighting="inv_vol"),
|
||
"Ensemble3 (RM/upcap/mult)":
|
||
lambda: default_ensemble(mkt_ret),
|
||
"Recovery+Mom Top10":
|
||
lambda: RecoveryMomentumStrategy(top_n=10),
|
||
"fc_up_cap+mom_gap":
|
||
lambda: FactorComboStrategy("up_cap+mom_gap", rebal_freq=21, top_n=10),
|
||
}
|
||
|
||
|
||
def main():
|
||
print("Loading PIT data…")
|
||
masked = load()
|
||
tickers = [c for c in masked.columns if c != BENCHMARK]
|
||
mkt_ret = masked[BENCHMARK].pct_change(fill_method=None)
|
||
print(f" shape={masked.shape} range={masked.index[0].date()} → {masked.index[-1].date()}")
|
||
|
||
rows = []
|
||
for years in (10, 5, 3, 1):
|
||
sliced, start = warmup_slice(masked, years, warmup_days=500)
|
||
prices = sliced[tickers]
|
||
print(f"\n--- {years}y window "
|
||
f"(measure {start.date()} → {sliced.index[-1].date()}, "
|
||
f"warmup from {sliced.index[0].date()}) ---")
|
||
spy = sliced[BENCHMARK].dropna()
|
||
spy_eq = (spy / spy.iloc[0]) * 10_000
|
||
m = measure(spy_eq, start, "")
|
||
rows.append({"years": years, "strategy": "SPY buy-and-hold",
|
||
**{k: v for k, v in m.items() if k != "name"}})
|
||
configs = make_configs(mkt_ret)
|
||
for name, factory in configs.items():
|
||
strat = factory()
|
||
eq = pit.backtest(strategy=strat, prices=prices,
|
||
initial_capital=10_000, transaction_cost=0.001)
|
||
m = measure(eq, start, "")
|
||
rows.append({"years": years, "strategy": name,
|
||
**{k: v for k, v in m.items() if k != "name"}})
|
||
tail = [r for r in rows if r["years"] == years]
|
||
tail.sort(key=lambda r: r["Sharpe"], reverse=True)
|
||
for r in tail:
|
||
print(f" {r['strategy']:<34s} CAGR={r['CAGR']*100:>6.1f}% "
|
||
f"Sharpe={r['Sharpe']:>5.2f} Sortino={r['Sortino']:>5.2f} "
|
||
f"MaxDD={r['MaxDD']*100:>6.1f}% Calmar={r['Calmar']:>5.2f}")
|
||
|
||
df = pd.DataFrame(rows)
|
||
df.to_csv(os.path.join(DATA_DIR, "interaction_results.csv"), index=False)
|
||
|
||
print("\n=== Cross-window CAGR summary (sorted by 10y Sharpe) ===")
|
||
pv = df.pivot(index="strategy", columns="years", values="CAGR")
|
||
pv.columns = [f"CAGR_{y}y" for y in pv.columns]
|
||
sh10 = df[df["years"] == 10].set_index("strategy")["Sharpe"]
|
||
pv["Sharpe_10y"] = sh10
|
||
pv = pv.sort_values("Sharpe_10y", ascending=False)
|
||
print(pv.to_string(formatters={
|
||
"CAGR_10y": "{:.1%}".format, "CAGR_5y": "{:.1%}".format,
|
||
"CAGR_3y": "{:.1%}".format, "CAGR_1y": "{:.1%}".format,
|
||
"Sharpe_10y": "{:.2f}".format,
|
||
}))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|