PythagorasGoal/src/strategies/trend_portfolio.py

"""TREND PORTFOLIO (TP01) — l'UNICA strategia profittevole e robusta post-reset (2026-06-19).

Vincitrice della ricerca su dati certificati BTC/ETH (Deribit mainnet). TSMOM multi-orizzonte
(1-3-6 mesi) vol-targeted, portafoglio 50/50 BTC+ETH. Validata onestamente (no look-ahead,
fee 0.10% RT, positiva ogni anno 2019-2026, robusta su griglia e su tutti i timeframe 15m-1d).

Config canonica deployabile (PORT LF4h):
    timeframe 4h, LONG-FLAT (niente short), vol-target 20%, leverage cap 2x.
    -> CAGR ~16.6%, Sharpe ~1.32, maxDD ~12.3% (backtest 2019-2026 su 50/50 BTC+ETH).

Perche' long-flat e 4h: gli short del trend rendono meno e aggiungono DD; il 4h e' il punto
dolce (meno rumore/fee del 15m, meno lag dell'1d). Vedi docs/diary/2026-06-19-research-synthesis.md
e scripts/research/trackD_*.py.

API (tutto causale, decide con dati <= close[i]):
    from src.strategies.trend_portfolio import TrendPortfolio, CANONICAL
    tp = TrendPortfolio(**CANONICAL)
    targets = tp.target_series(df_4h)        # array posizioni-bersaglio (frazione di equity, +/-)
    w = tp.current_target(df_4h)             # ultima posizione-bersaglio (per il live)
    res = tp.backtest_portfolio({'BTC': df_btc_4h, 'ETH': df_eth_4h})   # metriche onesta

NB: il vero "trade" e' un cambio di posizione; turnover basso (~37 ingressi/anno a 4h).
"""
from __future__ import annotations

from dataclasses import dataclass

import numpy as np
import pandas as pd

# config canonica raccomandata per il deploy
CANONICAL = dict(
    target_vol=0.20,
    leverage=2.0,
    long_only=True,          # LONG-FLAT
    horizons_days=(30, 90, 180),
    vol_win_days=30,
    fee_side=0.0005,         # 0.05%/lato = 0.10% RT (Deribit taker)
)

# variante headline long-short a 1h (riferimento storico, Sharpe ~1.0)
HEADLINE_LS_1H = dict(
    target_vol=0.20, leverage=2.0, long_only=False,
    horizons_days=(30, 90, 180), vol_win_days=30, fee_side=0.0005,
)

BARS_PER_DAY = {"5m": 288, "15m": 96, "1h": 24, "4h": 6, "1d": 1}


def simple_returns(c: np.ndarray) -> np.ndarray:
    r = np.zeros(len(c))
    r[1:] = c[1:] / c[:-1] - 1.0
    return r


def realized_vol(r: np.ndarray, win: int, bars_per_year: float) -> np.ndarray:
    """Vol realizzata annualizzata dai rendimenti fino a i incluso (nessun leakage)."""
    return pd.Series(r).rolling(win, min_periods=win // 2).std().values * np.sqrt(bars_per_year)


def tsmom_blend(c: np.ndarray, horizons: tuple[int, ...]) -> np.ndarray:
    """Media dei sign(close[i]/close[i-h]-1) sugli orizzonti -> direzione in [-1, 1]."""
    n = len(c)
    acc = np.zeros(n)
    cnt = np.zeros(n)
    for h in horizons:
        s = np.full(n, np.nan)
        s[h:] = np.sign(c[h:] / c[:-h] - 1.0)
        valid = np.isfinite(s)
        acc[valid] += s[valid]
        cnt[valid] += 1
    out = np.zeros(n)
    nz = cnt > 0
    out[nz] = acc[nz] / cnt[nz]
    return out


@dataclass
class TrendPortfolio:
    target_vol: float = 0.20
    leverage: float = 2.0
    long_only: bool = True
    horizons_days: tuple[int, ...] = (30, 90, 180)
    vol_win_days: int = 30
    fee_side: float = 0.0005

    def _bpd(self, df: pd.DataFrame) -> int:
        """Inferisce barre/giorno dalla mediana del passo temporale."""
        dt = pd.to_datetime(df["datetime"]).diff().dt.total_seconds().median()
        return max(1, round(86400 / dt))

    def target_series(self, df: pd.DataFrame) -> np.ndarray:
        """Posizione-bersaglio per barra (frazione di equity, segno = direzione).
        target[i] usa SOLO dati <= close[i] -> va TENUTA durante la barra i+1."""
        c = df["close"].values.astype(float)
        bpd = self._bpd(df)
        bpy = bpd * 365.25
        r = simple_returns(c)
        vol = realized_vol(r, self.vol_win_days * bpd, bpy)
        horizons = tuple(d * bpd for d in self.horizons_days)
        direction = tsmom_blend(c, horizons)
        if self.long_only:
            direction = np.clip(direction, 0, None)
        scal = np.where((vol > 0) & np.isfinite(vol), self.target_vol / vol, 0.0)
        tgt = np.clip(direction * scal, -self.leverage, self.leverage)
        tgt[~np.isfinite(tgt)] = 0.0
        return tgt

    def current_target(self, df: pd.DataFrame) -> float:
        """Posizione-bersaglio decisa all'ultima barra CHIUSA (per il paper/live)."""
        return float(self.target_series(df)[-1])

    def net_returns(self, df: pd.DataFrame) -> tuple[np.ndarray, pd.Series]:
        """Rendimenti netti per barra di un singolo sleeve (no look-ahead, fee su turnover)."""
        c = df["close"].values.astype(float)
        r = simple_returns(c)
        tgt = self.target_series(df)
        pos_held = np.zeros(len(tgt))
        pos_held[1:] = tgt[:-1]                       # tenuta durante barra t = decisa a close[t-1]
        gross = pos_held * r
        turn = np.abs(np.diff(pos_held, prepend=0.0))
        net = gross - self.fee_side * turn
        net[0] = 0.0
        net = np.clip(net, -0.99, None)
        return net, pd.to_datetime(df["datetime"])

    def backtest_portfolio(self, dfs: dict[str, pd.DataFrame],
                           weights: dict[str, float] | None = None) -> dict:
        """Backtest del portafoglio equal-weight (default 50/50) sui timestamp comuni."""
        weights = weights or {a: 1.0 / len(dfs) for a in dfs}
        series = {}
        for a, df in dfs.items():
            net, ts = self.net_returns(df)
            series[a] = pd.Series(net, index=pd.to_datetime(ts.values))
        J = pd.concat(series, axis=1, join="inner").fillna(0.0)
        combo = sum(weights[a] * J[a].values for a in dfs)
        idx = J.index
        equity = np.cumprod(1.0 + np.clip(combo, -0.99, None))
        return _metrics(equity, combo, idx)


def _metrics(equity: np.ndarray, combo: np.ndarray, idx: pd.DatetimeIndex) -> dict:
    bpy = _bars_per_year(idx)
    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 if span_days > 0 else 1.0
    total = equity[-1] / equity[0]
    cagr = total ** (1 / years) - 1 if years > 0 and total > 0 else -1.0
    eq = pd.Series(equity, index=idx)
    yearly = {}
    for y, g in eq.groupby(eq.index.year):
        if len(g) > 1 and g.iloc[0] > 0:
            v = g.values
            pk = np.maximum.accumulate(v)
            yearly[int(y)] = dict(pnl=float(g.iloc[-1] / g.iloc[0] - 1),
                                  dd=float(np.max((pk - v) / pk)))
    return dict(sharpe=sharpe, max_dd=dd, cagr=cagr, total_return=total - 1,
                yearly=yearly, equity=equity, index=idx)


def _bars_per_year(idx: pd.DatetimeIndex) -> float:
    if len(idx) < 2:
        return 365.25
    dt = pd.Series(idx).diff().dt.total_seconds().median()
    return 86400 * 365.25 / dt if dt and dt > 0 else 365.25


def resample_4h(df_1h: pd.DataFrame) -> pd.DataFrame:
    """Resample 1h -> 4h (confini 00:00 UTC). Schema con 'datetime'."""
    g = df_1h.copy()
    idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True)
    idx.name = "dt"
    g.index = idx
    out = g.resample("4h", label="left", closed="left").agg(
        {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"})
    out = out.dropna(subset=["open"])
    out["datetime"] = out.index
    epoch = pd.Timestamp("1970-01-01", tz="UTC")
    out["timestamp"] = ((out.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64")
    return out.reset_index(drop=True)[["timestamp", "open", "high", "low", "close", "volume", "datetime"]]