PythagorasGoal/scripts/research/trackA_trend.py

"""TRACK A — TREND / MOMENTUM research on certified BTC/ETH (Deribit mainnet).

Honest harness only (src.backtest.harness). Rules enforced:
  * Direction & entry price decided with data <= close[i]; fill at close[i].
  * Net of fees (0.10% RT baseline) + fee sweep + leverage stress.
  * IS / OOS split (65/35). Grid robustness across params AND both assets.

Run:  uv run python scripts/research/trackA_trend.py

This script is deliberately skeptical: it prints full grids so the reader can see
whether an "edge" is a single lucky cell or a robust neighborhood. The verdict at the
end is printed from the actual numbers, not asserted.
"""
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, backtest_signals, oos_split

ASSETS = ["BTC", "ETH"]
TFS = ["1h", "15m", "5m"]
FEE = 0.001


# ---------------------------------------------------------------------------
# Signal builders. Each returns a list[dict|None] of length len(df).
# All features use ONLY data up to and including close[i]. Entry fills at close[i].
# Position is approximated as a chained, non-overlapping hold of `hold` bars whose
# direction is recomputed at each (free) bar -> amortizes fee over `hold` bars while
# staying honest about responsiveness.
# ---------------------------------------------------------------------------
def sig_tsmom(df, lookback, hold, long_only=False):
    c = df["close"].values
    n = len(c)
    ent = [None] * n
    dirs = np.where(c[lookback:] > c[:-lookback], 1, -1)
    for k, d in enumerate(dirs):
        if long_only and d < 0:
            continue
        ent[lookback + k] = {"dir": int(d), "max_bars": hold}
    return ent


def _ema(x, span):
    return pd.Series(x).ewm(span=span, adjust=False).mean().values


def sig_ema_cross(df, fast, slow, hold, long_only=False):
    c = df["close"].values
    n = len(c)
    ef = _ema(c, fast)
    es = _ema(c, slow)
    ent = [None] * n
    for i in range(slow, n):
        d = 1 if ef[i] > es[i] else -1
        if long_only and d < 0:
            ent[i] = None
            continue
        ent[i] = {"dir": d, "max_bars": hold}
    return ent


def sig_donchian(df, lookback, hold, long_only=False):
    """Breakout: close[i] strictly above prior `lookback` highs -> long; below lows -> short.
    Detection AND entry both at close[i] (honest)."""
    c = df["close"].values
    h = df["high"].values
    l = df["low"].values
    n = len(c)
    ent = [None] * n
    # prior-window high/low EXCLUDING current bar (shift by 1) -> honest
    hh = pd.Series(h).rolling(lookback).max().shift(1).values
    ll = pd.Series(l).rolling(lookback).min().shift(1).values
    for i in range(lookback, n):
        if not np.isfinite(hh[i]):
            continue
        if c[i] > hh[i]:
            d = 1
        elif c[i] < ll[i]:
            d = -1
        else:
            continue
        if long_only and d < 0:
            continue
        ent[i] = {"dir": d, "max_bars": hold}
    return ent


def sig_vol_scaled_tsmom(df, lookback, hold, vol_win, z_gate):
    """Momentum gated by trend strength: only take a position when |past return| exceeds
    z_gate * rolling stdev of bar returns (regime gate). Honest: all <= close[i]."""
    c = df["close"].values
    n = len(c)
    logret = np.zeros(n)
    logret[1:] = np.diff(np.log(c))
    vol = pd.Series(logret).rolling(vol_win).std().values
    ent = [None] * n
    start = max(lookback, vol_win) + 1
    for i in range(start, n):
        r = np.log(c[i] / c[i - lookback])
        v = vol[i] * np.sqrt(lookback)
        if not np.isfinite(v) or v == 0:
            continue
        z = r / v
        if abs(z) < z_gate:
            continue
        d = 1 if z > 0 else -1
        ent[i] = {"dir": d, "max_bars": hold}
    return ent


# ---------------------------------------------------------------------------
# Evaluation helpers
# ---------------------------------------------------------------------------
def eval_is_oos(df, entries, asset, tf, fee=FEE, lev=1.0):
    cut = oos_split(df, 0.65)
    full = backtest_signals(df, entries, fee_rt=fee, leverage=lev, asset=asset, tf=tf)
    ent_is = [e if i < cut else None for i, e in enumerate(entries)]
    ent_oos = [e if i >= cut else None for i, e in enumerate(entries)]
    m_is = backtest_signals(df, ent_is, fee_rt=fee, leverage=lev, asset=asset, tf=tf)
    m_oos = backtest_signals(df, ent_oos, fee_rt=fee, leverage=lev, asset=asset, tf=tf)
    return full, m_is, m_oos


def buy_hold(df, cut=None):
    c = df["close"].values
    if cut is None:
        cut = oos_split(df, 0.65)
    return c[-1] / c[0] - 1, c[-1] / c[cut] - 1  # (full, oos)


def print_benchmarks():
    print("\n" + "=" * 110)
    print("# BUY & HOLD BENCHMARK (the bar any long/short trend edge must clear)")
    print("#   NOTE: OOS window is the LAST 35% = ~late-2023 -> 2026, a single (mostly bull) regime.")
    print("#   2018-2022 (bear+crash+bull+bear) is ENTIRELY in-sample. 'positive OOS' is weak evidence.")
    print("=" * 110)
    for tf in TFS:
        for asset in ASSETS:
            df = load(asset, tf)
            cut = oos_split(df, 0.65)
            bf, bo = buy_hold(df, cut)
            print(f"  {asset} {tf:>3s}  OOS starts {df['datetime'].iloc[cut].date()}  "
                  f"B&H full={bf*100:>+7.0f}%  B&H OOS={bo*100:>+7.0f}%")


def line(label, m):
    print(f"    {label:<30s} tr={m.n_trades:>6d} wr={m.win_rate:>4.1f}% "
          f"ret={m.net_return*100:>+8.0f}% CAGR={m.cagr*100:>+6.1f}% "
          f"Sh={m.sharpe:>5.2f} DD={m.max_dd*100:>4.1f}% mkt={m.time_in_market*100:>3.0f}% "
          f"€/d={m.daily_profit(2000):>+6.2f}")


# ---------------------------------------------------------------------------
# Experiments
# ---------------------------------------------------------------------------
def run_grid(name, builder, param_grid, builder_kwargs_fn, tfs=TFS, assets=ASSETS):
    """Generic grid runner. Prints OOS-focused table. Returns list of result dicts."""
    print("\n" + "=" * 110)
    print(f"# {name}")
    print("=" * 110)
    results = []
    for tf in tfs:
        for asset in assets:
            df = load(asset, tf)
            print(f"\n  -- {asset} {tf}  (n={len(df)}) --")
            for params in param_grid:
                ent = builder(df, **builder_kwargs_fn(params))
                full, m_is, m_oos = eval_is_oos(df, ent, asset, tf)
                tag = ",".join(f"{k}={v}" for k, v in params.items())
                line(f"{tag} [OOS]", m_oos)
                results.append(dict(name=name, asset=asset, tf=tf, params=params,
                                    full=full, is_=m_is, oos=m_oos))
    return results


def summarize_survivors(all_results):
    print("\n" + "#" * 110)
    print("# SURVIVOR SCREEN — positive OOS net return AND positive full-sample, Sharpe(OOS)>0")
    print("#" * 110)
    survivors = [r for r in all_results
                 if r["oos"].net_return > 0 and r["full"].net_return > 0
                 and r["oos"].sharpe > 0 and r["oos"].n_trades >= 20]
    if not survivors:
        print("  NONE. No config is net-positive OOS with positive full-sample and Sharpe>0.")
        return []
    survivors.sort(key=lambda r: r["oos"].sharpe, reverse=True)
    # precompute B&H OOS per (asset,tf)
    bh = {}
    for tf in TFS:
        for a in ASSETS:
            bh[(a, tf)] = buy_hold(load(a, tf))[1]
    print("  (BEATS B&H = OOS return exceeds buy&hold over same OOS window; otherwise it's just beta)")
    for r in survivors[:40]:
        tag = ",".join(f"{k}={v}" for k, v in r["params"].items())
        bho = bh[(r["asset"], r["tf"])]
        beat = "BEATS B&H" if r["oos"].net_return > bho else "<= B&H (beta)"
        print(f"  {r['name'][:18]:<18s} {r['asset']} {r['tf']:>3s} {tag:<28s} "
              f"OOS: ret={r['oos'].net_return*100:>+7.0f}% Sh={r['oos'].sharpe:>4.2f} "
              f"DD={r['oos'].max_dd*100:>4.0f}% €/d={r['oos'].daily_profit(2000):>+5.2f} | "
              f"B&H={bho*100:>+5.0f}% {beat}")
    return survivors


def robustness_report(survivors):
    """For top survivors, check fee sweep + leverage stress + cross-asset consistency."""
    if not survivors:
        return
    print("\n" + "#" * 110)
    print("# ROBUSTNESS: fee sweep (0.0005/0.001/0.0015/0.002) + leverage (1x/2x/3x) on top survivors")
    print("#" * 110)
    seen = set()
    for r in survivors[:8]:
        key = (r["name"], r["asset"], r["tf"], tuple(r["params"].items()))
        if key in seen:
            continue
        seen.add(key)
        df = load(r["asset"], r["tf"])
        # rebuild entries
        builder = BUILDERS[r["name"]]
        ent = builder(df, **KW_FN[r["name"]](r["params"]))
        tag = ",".join(f"{k}={v}" for k, v in r["params"].items())
        print(f"\n  {r['name']} {r['asset']} {r['tf']} {tag}")
        print("    fee sweep (OOS net return):")
        for fee in (0.0005, 0.001, 0.0015, 0.002):
            _, _, m_oos = eval_is_oos(df, ent, r["asset"], r["tf"], fee=fee)
            flag = "" if m_oos.net_return > 0 else "  <-- DIES"
            print(f"      fee={fee:.4f}: OOS ret={m_oos.net_return*100:>+8.0f}% Sh={m_oos.sharpe:>4.2f}{flag}")
        print("    leverage stress (OOS, fee=0.001):")
        for lev in (1.0, 2.0, 3.0):
            _, _, m_oos = eval_is_oos(df, ent, r["asset"], r["tf"], lev=lev)
            print(f"      {lev:.0f}x: OOS ret={m_oos.net_return*100:>+8.0f}% "
                  f"Sh={m_oos.sharpe:>4.2f} DD={m_oos.max_dd*100:>4.0f}% €/d={m_oos.daily_profit(2000):>+5.2f}")
        # yearly OOS
        _, _, m_oos = eval_is_oos(df, ent, r["asset"], r["tf"])
        print("    OOS yearly:")
        for y in sorted(m_oos.yearly):
            print(f"      {y}: {m_oos.yearly[y]*100:>+7.1f}%")


# registry so robustness_report can rebuild entries
BUILDERS = {
    "TSMOM": sig_tsmom,
    "TSMOM_LONG": sig_tsmom,
    "EMA_CROSS": sig_ema_cross,
    "DONCHIAN": sig_donchian,
    "VOLSCALED_TSMOM": sig_vol_scaled_tsmom,
}
KW_FN = {
    "TSMOM": lambda p: dict(lookback=p["N"], hold=p["H"]),
    "TSMOM_LONG": lambda p: dict(lookback=p["N"], hold=p["H"], long_only=True),
    "EMA_CROSS": lambda p: dict(fast=p["f"], slow=p["s"], hold=p["H"]),
    "DONCHIAN": lambda p: dict(lookback=p["N"], hold=p["H"]),
    "VOLSCALED_TSMOM": lambda p: dict(lookback=p["N"], hold=p["H"], vol_win=p["vw"], z_gate=p["z"]),
}


def main():
    pd.set_option("display.width", 200)
    print_benchmarks()
    all_results = []

    # ---- 1. TSMOM (long/short) ----
    tsmom_grid = [dict(N=n, H=h) for n in (10, 20, 50, 100, 200) for h in (6, 12, 24, 48)]
    all_results += run_grid("TSMOM", sig_tsmom, tsmom_grid,
                            KW_FN["TSMOM"])

    # ---- 2. TSMOM long-only (crypto has strong upward drift; honest to test) ----
    all_results += run_grid("TSMOM_LONG", lambda df, **k: sig_tsmom(df, long_only=True, **k),
                            [dict(N=n, H=h) for n in (20, 50, 100, 200) for h in (12, 24, 48)],
                            KW_FN["TSMOM"])

    # ---- 3. EMA crossover ----
    ema_grid = [dict(f=f, s=s, H=h)
                for (f, s) in ((10, 30), (20, 50), (20, 100), (50, 200))
                for h in (12, 24, 48)]
    all_results += run_grid("EMA_CROSS", sig_ema_cross, ema_grid, KW_FN["EMA_CROSS"])

    # ---- 4. Donchian breakout ----
    don_grid = [dict(N=n, H=h) for n in (20, 50, 100, 200) for h in (12, 24, 48)]
    all_results += run_grid("DONCHIAN", sig_donchian, don_grid, KW_FN["DONCHIAN"])

    # ---- 5. Vol-scaled / regime-gated TSMOM ----
    vs_grid = [dict(N=n, H=h, vw=vw, z=z)
               for n in (20, 50, 100) for h in (24, 48)
               for vw in (50, 100) for z in (0.5, 1.0)]
    all_results += run_grid("VOLSCALED_TSMOM", sig_vol_scaled_tsmom, vs_grid,
                            KW_FN["VOLSCALED_TSMOM"])

    # ---- survivor screen + robustness ----
    survivors = summarize_survivors(all_results)
    robustness_report(survivors)

    # ---- cross-asset robustness note ----
    print("\n" + "#" * 110)
    print("# CROSS-ASSET / CROSS-TF CONSISTENCY of survivors (a real edge holds on BOTH BTC & ETH)")
    print("#" * 110)
    from collections import defaultdict
    by_strat = defaultdict(list)
    for r in survivors:
        by_strat[(r["name"], r["tf"], tuple(r["params"].items()))].append(r["asset"])
    both = [(k, v) for k, v in by_strat.items() if set(v) >= {"BTC", "ETH"}]
    if not both:
        print("  No single (strategy, tf, params) cell is an OOS survivor on BOTH BTC and ETH.")
        print("  => any apparent edge is asset/regime-specific, not a robust trend edge.")
    else:
        for (name, tf, params), assets in both:
            print(f"  {name} {tf} {dict(params)} survives on: {assets}")

    print("\nDONE. Read the survivor screen + robustness above for the honest verdict.")


if __name__ == "__main__":
    main()