Adriano Dal Pastro
14522262e6
Reset del progetto su fondamenta verificate dopo la scoperta che l'intera libreria "validata OOS" era artefatto di feed contaminato (print fantasma del feed Cerbero TESTNET + storico Binance/USDT). - Storico ricostruito da Deribit MAINNET (ccxt pubblico, tokenless) e CERTIFICATO (certify_feed.py): BTC/ETH puliti su TUTTA la storia (mediana 2-6 bps vs Coinbase USD), integrita' OHLC + coerenza resample (maxΔ 0.00) + cross-venue OK. Alt esclusi (illiquidi/divergenti: LTC/DOGE 50-82% barre flat; XRP/BNB non certificabili). - Verdetto sul feed pulito: FADE / PAIRS / XS01 / TSM01 morti (ogni portafoglio Sharpe -2.3..-3.0, DD ~40%); solo SH01 e frammenti HONEST con segnale residuo, da ri-validare in isolamento. - Cleanup "restart pulito": strategie, stack live (src/live, src/portfolio, runner/executor, yml, docker), ~100 script ricerca/gate, waste/games/ portfolios, dati non certificati + cache e 60+ diari -> archiviati in Old/ (preservati, non cancellati). Diario consolidato in un unico documento. - Skeleton ricerca tenuto: Strategy ABC + indicatori + src/fractal + src/backtest/engine + load_data; tool dati certificati (rebuild_history, certify_feed, audit_feed, multi_source_check). - Universo dati ATTIVO: solo BTC/ETH (5m/15m/1h); guardrail fisico (load_data su alt -> FileNotFoundError). Esecuzione DISABILITATA, conto flat. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
"""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)
|
|
|
|
|
|
def load_active_portfolio(config_path) -> "Portfolio":
|
|
"""Carica il portafoglio attivo da portfolios.yml applicando gli override."""
|
|
import yaml
|
|
from pathlib import Path
|
|
from scripts.portfolios._defs import PORTFOLIOS
|
|
|
|
cfg = yaml.safe_load(Path(config_path).read_text())
|
|
p = PORTFOLIOS[cfg["active"]]
|
|
ov = cfg.get("overrides", {})
|
|
for k in ("total_capital", "weighting", "caps", "leverage", "rebalance", "vol_lookback"):
|
|
if k in ov and ov[k] is not None:
|
|
setattr(p, k, ov[k])
|
|
return p
|