feat(portfolio): contenitore di strategie ESTENSIBILE — TP01 primo sleeve

src/portfolio/: Sleeve (serie rendimenti netti per-barra, causale/fee-aware) + StrategyPortfolio
(combina N sleeve per peso su griglia giornaliera comune, metriche FULL/HOLD-OUT/per-anno +
standalone per-sleeve, vs buy&hold). Registry sleeve attivi in sleeves.py: per ora SOLO TP01
(peso 100%); aggiungere = una riga (dopo validazione col gauntlet).

Report (run_portfolio.py): TP01 FULL Sh 1.30 / DD 14.3% / ~€1.52/g, HOLD-OUT 0.31 / +3.5%
(buy&hold -0.32 / -39%). Posizione corrente flat (difensivo). tests/test_portfolio.py (6 test).
CLAUDE.md aggiornato (struttura + comando + come aggiungere uno sleeve).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-06-19 19:17:18 +00:00
parent 12754c4908
commit ef52ad6a79
7 changed files with 343 additions and 3 deletions
+111
View File
@@ -0,0 +1,111 @@
"""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:
w = self.weights()
cols = {s.name: s.daily() for s in self.sleeves}
J = pd.concat(cols, axis=1, join="inner").dropna()
combo = sum(w[c] * J[c] for c in J.columns)
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}