PythagorasGoal/src/portfolio/portfolio.py

"""PORTAFOGLIO DI STRATEGIE — contenitore estensibile (v2.0.0).

Modello: ogni SLEEVE produce una serie di rendimenti netti per-barra (datetime-indexed, CAUSALE,
netto fee). Il portafoglio:
  1. porta ogni sleeve su una griglia GIORNALIERA comune (compounding intra-giorno) — così sleeve
     a TF diversi (1d, 1h, ...) si combinano in modo coerente;
  2. combina per PESO (rinormalizzato a 1) sui giorni comuni a tutti gli sleeve;
  3. = portafoglio equal-capital-by-weight ribilanciato di continuo (interpretazione del weighted-
     return combine). Equity = capitale · Π(1+combo).

AGGIUNGERE uno sleeve è una riga in src/portfolio/sleeves.py (vedi lì il template).
Metriche oneste: FULL + HOLD-OUT 2025-26 (bloccato) + per-anno, e standalone per-sleeve.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Callable
import numpy as np
import pandas as pd

DAYS_PER_YEAR = 365.25
HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC")


def to_daily(net: pd.Series) -> pd.Series:
    """Compound una serie di rendimenti netti per-barra a GIORNALIERA (griglia comune del portafoglio)."""
    s = net.dropna().sort_index()
    if not isinstance(s.index, pd.DatetimeIndex):
        s.index = pd.to_datetime(s.index, utc=True)
    if s.index.tz is None:
        s.index = s.index.tz_localize("UTC")
    return ((1.0 + s).resample("1D").prod() - 1.0).dropna()


@dataclass
class Sleeve:
    """Una strategia nel portafoglio. daily_fn() -> serie rendimenti netti per-barra (causale, netto fee).
    pos_fn() (opzionale) -> dict posizioni-bersaglio correnti, per introspezione live."""
    name: str
    weight: float
    daily_fn: Callable[[], pd.Series]
    pos_fn: Callable[[], dict] | None = None
    _cache: pd.Series | None = field(default=None, repr=False, compare=False)

    def daily(self) -> pd.Series:
        if self._cache is None:
            self._cache = to_daily(self.daily_fn())
        return self._cache


def metrics(daily: pd.Series) -> dict:
    r = np.asarray(daily.dropna().values, float)
    if len(r) < 2 or r.std() == 0:
        return dict(sharpe=0.0, cagr=0.0, maxdd=0.0, ret=0.0, n=int(len(r)))
    eq = np.cumprod(1.0 + r)
    pk = np.maximum.accumulate(eq)
    years = len(r) / DAYS_PER_YEAR
    return dict(sharpe=float(r.mean() / r.std() * np.sqrt(DAYS_PER_YEAR)),
                cagr=float(eq[-1] ** (1 / years) - 1) if years > 0 and eq[-1] > 0 else 0.0,
                maxdd=float(np.max((pk - eq) / pk)), ret=float(eq[-1] - 1), n=int(len(r)))


def yearly(daily: pd.Series) -> dict:
    out = {}
    for y, g in daily.groupby(daily.index.year):
        v = g.values
        eq = np.cumprod(1 + v); pk = np.maximum.accumulate(eq)
        out[int(y)] = dict(ret=float(eq[-1] - 1), dd=float(np.max((pk - eq) / pk)))
    return out


class StrategyPortfolio:
    def __init__(self, sleeves: list[Sleeve], capital: float = 2000.0):
        if not sleeves:
            raise ValueError("portafoglio vuoto: serve almeno uno sleeve")
        self.sleeves = sleeves
        self.capital = capital

    def weights(self) -> dict:
        tot = sum(s.weight for s in self.sleeves)
        if tot <= 0:
            raise ValueError("somma pesi non positiva")
        return {s.name: s.weight / tot for s in self.sleeves}

    def combined_daily(self, lo=None, hi=None) -> pd.Series:
        """Combina gli sleeve per peso. OUTER-join: sleeve con date d'inizio diverse
        (es. TP01 dal 2019, uno nuovo dal 2024) -> ogni giorno i pesi sono RINORMALIZZATI
        fra i soli sleeve con dato disponibile (uno sleeve "si attiva" quando parte la sua
        storia). Cosi' non si tronca il portafoglio alla finestra comune."""
        w = self.weights()
        cols = {s.name: s.daily() for s in self.sleeves}
        J = pd.concat(cols, axis=1, join="outer").sort_index()
        wv = np.array([w[c] for c in J.columns], float)
        active = J.notna().values * wv                       # peso solo dove c'e' dato
        rowsum = active.sum(axis=1, keepdims=True)
        wnorm = np.divide(active, rowsum, out=np.zeros_like(active), where=rowsum > 0)
        combo = pd.Series(np.nansum(np.nan_to_num(J.values) * wnorm, axis=1), index=J.index)
        combo = combo[J.notna().any(axis=1).values]          # togli i giorni senza alcun dato
        if lo is not None:
            combo = combo[combo.index >= lo]
        if hi is not None:
            combo = combo[combo.index < hi]
        return combo

    def backtest(self) -> dict:
        full = self.combined_daily()
        return dict(
            weights=self.weights(),
            full=metrics(full),
            holdout=metrics(self.combined_daily(lo=HOLDOUT)),
            yearly=yearly(full),
            per_sleeve={s.name: dict(weight=self.weights()[s.name],
                                     full=metrics(s.daily()),
                                     holdout=metrics(s.daily()[s.daily().index >= HOLDOUT]))
                        for s in self.sleeves},
            equity=self.capital * np.cumprod(1.0 + full.values),
            index=full.index,
        )

    def current_positions(self) -> dict:
        return {s.name: (s.pos_fn() if s.pos_fn else None) for s in self.sleeves}