diff --git a/scripts/research/trackD_timing.py b/scripts/research/trackD_timing.py new file mode 100644 index 0000000..e39feba --- /dev/null +++ b/scripts/research/trackD_timing.py @@ -0,0 +1,155 @@ +"""TRACK D on DIFFERENT TIMEFRAMES — per-year PnL and per-year max drawdown. + +Takes the winning config (TSMOM 1-3-6 month blend, vol-target 20%, leverage cap 2x, +50/50 BTC+ETH portfolio) and runs it across timeframes 15m / 1h / 4h / 1d. + +Honesty preserved: same building blocks as trackD_trendport.py (positions shifted +1 bar, +fee 0.10% RT on turnover, vol-targeting on past-only realized vol). Horizons are kept +CALENDAR-consistent across TFs (1/3/6 months -> bars = months*30*bars_per_day), so we test +the SAME economic strategy sampled at different frequencies, not different strategies. + +4h/1d are RESAMPLED from the certified 1h feed (00:00 UTC boundaries). + +Run: uv run python scripts/research/trackD_timing.py +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from src.backtest.harness import load +from scripts.research.trackD_trendport import ( + simple_returns, realized_vol, sig_tsmom_blend, build_target, + equity_from_target, +) + +ASSETS = ["BTC", "ETH"] +FEE_SIDE = 0.0005 # 0.05%/side = 0.10% RT +TARGET_VOL = 0.20 +LEVERAGE = 2.0 + +# timeframe -> (load_tf, resample_rule_or_None, bars_per_day) +TIMEFRAMES = { + "15m": ("15m", None, 96), + "1h": ("1h", None, 24), + "4h": ("1h", "4h", 6), + "1d": ("1h", "1D", 1), +} + + +def resample_ohlc(df: pd.DataFrame, rule: str) -> pd.DataFrame: + g = df.copy() + idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True) + idx.name = "dt" + g.index = idx + out = g.resample(rule, label="left", closed="left").agg( + {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}) + out = out.dropna(subset=["open"]) + out["datetime"] = out.index + out["timestamp"] = (out.index.view("int64") // 1_000_000) + return out.reset_index(drop=True)[["timestamp", "open", "high", "low", "close", "volume", "datetime"]] + + +def get_df(tf_key: str, asset: str) -> pd.DataFrame: + load_tf, rule, _ = TIMEFRAMES[tf_key] + df = load(asset, load_tf) + if rule: + df = resample_ohlc(df, rule) + return df + + +def run_asset(df, bars_per_day, target_vol=TARGET_VOL, leverage=LEVERAGE, + long_only=False, fee_side=FEE_SIDE): + c = df["close"].values.astype(float) + r = simple_returns(c) + bpy = bars_per_day * 365.25 + # recompute building blocks at this TF's bar frequency + h1, h3, h6 = 30 * bars_per_day, 90 * bars_per_day, 180 * bars_per_day + vol_win = 30 * bars_per_day + # realized_vol / tsmom use BARS_PER_YEAR from trackD (1h) for annualization of vol; + # we must annualize with THIS tf's bpy -> compute vol locally + vol = pd.Series(r).rolling(vol_win, min_periods=vol_win // 2).std().values * np.sqrt(bpy) + direction = sig_tsmom_blend(c, horizons=(h1, h3, h6)) + tgt = build_target(direction, vol, target_vol, leverage, long_only) + equity, net = equity_from_target(tgt, r, fee_side) + return dict(net=net, ts=df["datetime"], equity=equity, bpy=bpy) + + +def portfolio_series(sleeves): + a = pd.Series(sleeves["BTC"]["net"], index=pd.to_datetime(sleeves["BTC"]["ts"].values)) + b = pd.Series(sleeves["ETH"]["net"], index=pd.to_datetime(sleeves["ETH"]["ts"].values)) + j = pd.concat([a.rename("a"), b.rename("b")], axis=1, join="inner").fillna(0.0) + combo = 0.5 * j["a"].values + 0.5 * j["b"].values + idx = pd.to_datetime(j.index) + equity = np.cumprod(1.0 + np.clip(combo, -0.99, None)) + return idx, combo, equity + + +def overall_metrics(idx, combo, equity, bpy): + rr = combo[np.isfinite(combo)] + sharpe = float(np.mean(rr) / np.std(rr) * np.sqrt(bpy)) if np.std(rr) > 0 else 0.0 + peak = np.maximum.accumulate(equity) + dd = float(np.max((peak - equity) / peak)) + span_days = (idx[-1] - idx[0]).total_seconds() / 86400 + years = span_days / 365.25 + total = equity[-1] / equity[0] + cagr = total ** (1 / years) - 1 if years > 0 and total > 0 else -1.0 + daily_2k = (2000 * total - 2000) / span_days if span_days > 0 else 0.0 + return dict(sharpe=sharpe, max_dd=dd, cagr=cagr, total=total - 1, daily_2k=daily_2k) + + +def per_year(idx, equity): + """Return {year: (pnl_pct, maxdd_pct)} where maxdd is the worst drawdown WITHIN the year.""" + eq = pd.Series(equity, index=idx) + out = {} + for y, g in eq.groupby(eq.index.year): + if len(g) < 2: + continue + pnl = g.iloc[-1] / g.iloc[0] - 1.0 + v = g.values + peak = np.maximum.accumulate(v) + ddy = float(np.max((peak - v) / peak)) + out[int(y)] = (float(pnl), ddy) + return out + + +ALL_YEARS = list(range(2018, 2027)) + + +def main(): + print("=" * 118) + print("# TRACK D WINNER ACROSS TIMEFRAMES — TSMOM 1-3-6m blend, vol-target 20%, lev 2x, 50/50 BTC+ETH") + print("# fee 0.10% RT on turnover, positions +1 bar (no look-ahead). 4h/1d resampled from certified 1h.") + print("=" * 118) + + for mode_long_only, mode_name in ((False, "LONG-SHORT"), (True, "LONG-FLAT")): + print("\n" + "#" * 118) + print(f"# MODE = {mode_name}") + print("#" * 118) + for tf_key in TIMEFRAMES: + bpd = TIMEFRAMES[tf_key][2] + sleeves = {a: run_asset(get_df(tf_key, a), bpd, long_only=mode_long_only) + for a in ASSETS} + idx, combo, equity = portfolio_series(sleeves) + ov = overall_metrics(idx, combo, equity, sleeves["BTC"]["bpy"]) + py = per_year(idx, equity) + + print(f"\n ── TF {tf_key:<3s} │ ret {ov['total']*100:>+8.0f}% CAGR {ov['cagr']*100:>+6.1f}% " + f"Sharpe {ov['sharpe']:>4.2f} maxDD {ov['max_dd']*100:>4.1f}% " + f"€/day(2k) {ov['daily_2k']:>+6.2f}") + # per-year PnL row + print(f" {'PnL %':<8s}" + "".join( + (" . " if y not in py else f"{py[y][0]*100:>+7.0f}") for y in ALL_YEARS)) + print(f" {'maxDD %':<8s}" + "".join( + (" . " if y not in py else f"{py[y][1]*100:>7.1f}") for y in ALL_YEARS)) + # year header for reference + print("\n " + "year ".ljust(8) + "".join(f"{y:>7d}" for y in ALL_YEARS)) + + +if __name__ == "__main__": + main()