feat(portfolio): SleeveSpec/Portfolio/backtest con parità verso report_families

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-29 15:55:29 +02:00
parent 9ff469cb8e
commit 3f6b0ccf91
2 changed files with 103 additions and 0 deletions
@@ -0,0 +1,79 @@
+"""Portfolio: definizione (sleeve + schema pesi) con faccia di backtest.
+La faccia live è in runner.py."""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+import pandas as pd
+
+from src.portfolio import weighting as W
+from src.portfolio.sleeves import all_sleeve_equities, sleeve_returns_df
+from scripts.analysis.combine_portfolio import port_returns, metrics, yearly_returns, SPLIT
+
+
+@dataclass
+class SleeveSpec:
+    kind: str
+    name: str
+    sid: str
+    asset: str | None = None
+    a: str | None = None
+    b: str | None = None
+    tf: str = "1h"
+    params: dict = field(default_factory=dict)
+    cluster: str = ""
+
+
+@dataclass
+class PortfolioResult:
+    code: str
+    weights: dict
+    full: dict
+    oos: dict
+    yearly: dict
+    risk: dict
+
+
+@dataclass
+class Portfolio:
+    code: str
+    label: str
+    sleeves: list[SleeveSpec]
+    weighting: str = "equal"
+    weights: dict | None = None
+    caps: dict | None = None
+    total_capital: float = 1000.0
+    leverage: float = 3.0
+    rebalance: str = "1D"
+    vol_lookback: int = 90
+
+    @property
+    def sleeve_ids(self) -> list[str]:
+        return [s.sid for s in self.sleeves]
+
+    @property
+    def clusters(self) -> dict[str, str]:
+        return {s.sid: (s.cluster or s.sid) for s in self.sleeves}
+
+    def weight_vector(self, returns_df: pd.DataFrame | None = None) -> dict[str, float]:
+        return W.weight_vector(
+            self.weighting, self.sleeve_ids, returns_df,
+            weights=self.weights, caps=self.caps,
+            clusters=self.clusters, lookback=self.vol_lookback,
+        )
+
+    def backtest(self) -> PortfolioResult:
+        eq = all_sleeve_equities()
+        members = {sid: eq[sid] for sid in self.sleeve_ids}
+        dr = sleeve_returns_df(self.sleeve_ids)
+        w = self.weight_vector(dr)
+        port_dr = port_returns(members, w)
+        full, oos = metrics(port_dr), metrics(port_dr, lo=SPLIT)
+        import numpy as np
+        we = np.ones(len(self.sleeve_ids)) / len(self.sleeve_ids)
+        cov = dr.cov().values
+        pv = float(we @ cov @ we)
+        rc = we * (cov @ we)
+        risk = {sid: float(rc[k] / pv * 100) if pv > 0 else 0.0
+                for k, sid in enumerate(self.sleeve_ids)}
+        return PortfolioResult(self.code, w, full, oos, yearly_returns(port_dr), risk)