PythagorasGoal/scripts/analysis/honest_trend.py

"""Strategia #3 candidata: time-series momentum / trend (TSMOM).
Posizione continua decisa a close[i] dai dati passati; fee SOLO sui cambi di
posizione (poche operazioni su TF alto = fee non letali). Niente look-ahead:
il rendimento del bar i->i+1 usa la direzione decisa a close[i].
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from scripts.analysis.honest_lab import ema, get_df, available_assets, FEE_RT  # noqa: E402

LEV = 3.0
POS = 0.15


def simulate_position(sig: np.ndarray, df: pd.DataFrame, fee_rt: float = FEE_RT,
                      lev: float = LEV, pos: float = POS) -> dict:
    """sig[i] in {-1,0,1} = direzione tenuta nel bar i->i+1, decisa a close[i].
    Fee one-way = fee_rt/2 su ogni unita' di variazione posizione."""
    c = df["close"].values
    n = len(c)
    ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    cap = peak = 1000.0
    max_dd = 0.0
    cur = 0.0
    flips = 0
    bars_in = 0
    yearly: dict[int, float] = {}
    for i in range(n - 1):
        s = sig[i]
        if not np.isfinite(s):
            s = 0.0
        if s != cur:
            cap -= cap * pos * (fee_rt / 2) * lev * abs(s - cur)
            flips += abs(s - cur) > 0
            cur = s
        pr = (c[i + 1] - c[i]) / c[i]
        bar_ret = pos * lev * pr * cur
        cap = max(cap * (1 + bar_ret), 10.0)
        peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak)
        if cur != 0:
            bars_in += 1
        yr = ts.iloc[i].year
        yearly[yr] = yearly.get(yr, 0.0) + bar_ret * 100
    return {
        "ret": (cap / 1000 - 1) * 100,
        "dd": max_dd * 100,
        "flips": flips,
        "exposure": bars_in / n * 100,
        "yearly": yearly,
        "pos_years": sum(1 for v in yearly.values() if v > 0),
        "n_years": len(yearly),
    }


def tsmom_signal(df, lookback=30, long_only=False):
    """+1 se close>close[-lookback], -1 (o 0 se long_only) altrimenti."""
    c = df["close"].values
    sig = np.zeros(len(c))
    for i in range(lookback, len(c)):
        up = c[i] > c[i - lookback]
        sig[i] = 1.0 if up else (0.0 if long_only else -1.0)
    return sig


def ema_dual_signal(df, fast=20, slow=100, long_only=False):
    """+1 se EMA_fast>EMA_slow."""
    c = df["close"].values
    ef, es = ema(c, fast), ema(c, slow)
    sig = np.where(ef > es, 1.0, 0.0 if long_only else -1.0)
    sig[:slow] = 0.0
    return sig


def oos(sig, df, frac=0.30):
    split = int(len(df) * (1 - frac))
    s2 = sig.copy(); s2[:split] = 0.0
    return simulate_position(s2, df)


def show(label, df, sig):
    full = simulate_position(sig, df)
    o = oos(sig, df)
    anni = f"{full['pos_years']}/{full['n_years']}"
    print(f"  {label:<26s}{full['flips']:>6d}{full['ret']:>+9.0f}{o['ret']:>+9.0f}"
          f"{full['dd']:>6.0f}{full['exposure']:>6.0f}{anni:>8s}")
    return full, o


if __name__ == "__main__":
    assets = available_assets()
    print(f"TSMOM / trend — fee {FEE_RT*100:.2f}% RT, leva3x pos15% | OOS30%")
    for tf in ["1d", "4h"]:
        print(f"\n  ###### TF {tf} ######")
        for a in assets:
            df = get_df(a, tf)
            print(f"\n  === {a} {tf} ===  {'Flip':>5s}{'FULL%':>8s}{'OOS%':>8s}{'DD%':>6s}{'Exp%':>6s}{'AnniP':>8s}")
            show("TSMOM lb30 long/short", df, tsmom_signal(df, 30))
            show("TSMOM lb30 long-only", df, tsmom_signal(df, 30, long_only=True))
            show("TSMOM lb90 long/short", df, tsmom_signal(df, 90))
            show("EMA 20/100 long/short", df, ema_dual_signal(df, 20, 100))
            show("EMA 20/100 long-only", df, ema_dual_signal(df, 20, 100, long_only=True))