diff --git a/src/portfolio/base.py b/src/portfolio/base.py new file mode 100644 index 0000000..892e887 --- /dev/null +++ b/src/portfolio/base.py @@ -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) diff --git a/tests/portfolio/test_backtest_parity.py b/tests/portfolio/test_backtest_parity.py new file mode 100644 index 0000000..3b7af9f --- /dev/null +++ b/tests/portfolio/test_backtest_parity.py @@ -0,0 +1,24 @@ +import pytest +from src.portfolio.base import Portfolio, SleeveSpec +from scripts.analysis.report_families import build_everything +from scripts.analysis.combine_portfolio import port_returns, metrics, SPLIT + + +def _master9_specs(): + fade = [SleeveSpec(kind="single", name=f"{c}", sid=f"{c}_{a}", asset=a, cluster=f"{a}-rev") + for a in ("BTC", "ETH") for c in ("MR01", "MR02", "MR07")] + honest = [SleeveSpec(kind="single", name="DIP01", sid="DIP01_BTC", asset="BTC", cluster="BTC-rev"), + SleeveSpec(kind="single", name="TR01", sid="TR01_basket", cluster="trend"), + SleeveSpec(kind="single", name="ROT02", sid="ROT02_rot", cluster="rotation")] + return fade + honest + + +def test_master9_backtest_matches_report(): + p = Portfolio(code="PORT03", label="Master", sleeves=_master9_specs(), weighting="equal") + res = p.backtest() + S, _, _, _ = build_everything() + dr_ref = port_returns(S) + ref_full, ref_oos = metrics(dr_ref), metrics(dr_ref, lo=SPLIT) + assert res.full["sharpe"] == pytest.approx(ref_full["sharpe"], abs=1e-6) + assert res.full["dd"] == pytest.approx(ref_full["dd"], abs=1e-6) + assert res.oos["sharpe"] == pytest.approx(ref_oos["sharpe"], abs=1e-6)