research: Track D winner across timeframes (15m-1d) + per-year PnL/DD
- trackD_timing.py: same TSMOM 1-3-6m blend config sampled at 15m/1h/4h/1d - robust plateau across all TFs; 4h marginally best (LF Sharpe 1.36, DD 13.8%) - per-year PnL and per-year max drawdown tables
This commit is contained in:
@@ -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()
|
||||
Reference in New Issue
Block a user