"""PortfolioRunner: faccia live del portafoglio (capitale pool, sizing, ribilancio, ledger). Riusa i worker esistenti come esecutori e il data layer Cerbero v2. Worker per tipo di sleeve: single (fade/dip) -> StrategyWorker | ml (shape, SH01) -> StrategyWorker (WF interno) pairs -> PairsWorker (2 gambe) | basket (TR01) -> BasketTrendWorker rotation (ROT02) -> RotationWorker | tsmom (TSM01) -> TsmomWorker Feed: il runner fetcha candele 1h da Cerbero v2 e le RESAMPLA a 4h/1d (come get_df nel backtest) per i worker a cadenza piu' lenta. Il lookback per asset e' dimensionato sul worker piu' esigente (TSM01 usa 252 giorni).""" from __future__ import annotations from pathlib import Path import pandas as pd from src.portfolio.base import SleeveSpec, Portfolio from src.portfolio.ledger import PortfolioLedger from src.live.strategy_worker import StrategyWorker from src.live.pairs_worker import PairsWorker from src.live.basket_trend_worker import BasketTrendWorker from src.live.rotation_worker import RotationWorker from src.live.tsmom_worker import TsmomWorker from src.live.strategy_loader import load_strategy # Codice-breve sleeve -> nome modulo Strategy in scripts/strategies/ (worker single/ml) _STRAT_MODULE = { "MR01": "MR01_bollinger_fade", "MR02": "MR02_donchian_fade", "MR07": "MR07_return_reversal", "SH01": "SH01_shape_ml", "DIP01": "DIP01_dip_buy", } _MULTI_KINDS = ("basket", "rotation", "tsmom") DATA_DIR = Path("data/portfolio_paper") # giorni di storia da fetchare per timeframe (TSM01 1d usa 252 barre -> ~440 giorni col buffer) _LOOKBACK_DAYS = {"5m": 7, "15m": 14, "30m": 21, "1h": 90, "4h": 220, "1d": 440} # timeframe SUB-orari: si fetchano DIRETTI da Cerbero (non resamplabili dal 1h). _SUBHOURLY = {"5m", "15m", "30m"} # SH01 (ml) richiede >=4000 barre 1h (train_min di ml_wf_entries); 365g (~8760 barre) danno # margine ampio per il walk-forward. Difensivo: non dipende dal fetch 440g di TSM01/ROT02. _ML_LOOKBACK_DAYS = 365 def pos_for_spec(sid: str, global_ps: float, family_overrides: dict[str, float], sleeve_ps: float | None = None) -> float: """position_size effettivo di uno sleeve. Precedenza: override PER-SLEEVE (spec.params['position_size'], es. il 15m a 0.10) > override per-FAMIGLIA (weighting.family_of: PAIRS/FADE/...) > globale.""" from src.portfolio.weighting import family_of if sleeve_ps is not None: return float(sleeve_ps) return family_overrides.get(family_of(sid), global_ps) def build_worker_for(spec: SleeveSpec, alloc_capital: float, leverage: float, data_dir: Path = DATA_DIR, position_size: float = 0.15, executor=None, exec_instrument: str | None = None, pairs_executor=None, exec_instruments: dict | None = None): """Costruisce il worker esecutore per uno sleeve con capitale = quota allocata. executor/exec_instrument: per i fade single-leg, StrategyWorker affianca al fill sim un ordine REALE (shadow). pairs_executor/exec_instruments: idem per i PairsWorker (esecuzione reale a 2 gambe).""" if spec.kind == "pairs": return PairsWorker( asset_a=spec.a, asset_b=spec.b, tf=spec.tf, params=spec.params, capital=alloc_capital, position_size=position_size, leverage=leverage, fee_rt=0.001, name="PR01_pairs_reversion", data_dir=data_dir, executor=pairs_executor, exec_instruments=exec_instruments, ) if spec.kind == "basket": pr = spec.params return BasketTrendWorker( universe=pr["universe"], tf=pr.get("tf", "4h"), capital=alloc_capital, position_size=position_size, leverage=leverage, data_dir=data_dir, ) if spec.kind == "rotation": pr = spec.params return RotationWorker( universe=pr["universe"], top_k=pr.get("top_k", 3), gross=pr.get("gross", 0.45), tf=pr.get("tf", "1d"), capital=alloc_capital, data_dir=data_dir, ) if spec.kind == "tsmom": pr = spec.params return TsmomWorker( universe=pr["universe"], horizons=tuple(pr.get("horizons", (63, 126, 252))), thr=pr.get("thr", 1.0), gross=pr.get("gross", 0.30), tf=pr.get("tf", "1d"), capital=alloc_capital, data_dir=data_dir, ) module = _STRAT_MODULE.get(spec.name) if module is None: raise ValueError(f"sleeve live non supportato: {spec.name} (kind={spec.kind})") strategy = load_strategy(module) # SH01 (kind="ml") gira come StrategyWorker NORMALE: SH01_shape_ml.generate_signals fa il # walk-forward (retraining) internamente ad ogni tick ed emette metadata.max_bars=H -> gli # exit passano per StrategyWorker.tick (orizzonte H). NON usare il vecchio MLWorkerWrapper di # multi_runner: quello usa SignalEngine (famiglia squeeze SCARTATA), apre senza metadata ed # esce a hold_bars=3, ignorando del tutto SH01_shape_ml. Serve >=4000 barre 1h (train_min): # garantite da _ML_LOOKBACK_DAYS. return StrategyWorker( strategy=strategy, asset=spec.asset, tf=spec.tf, capital=alloc_capital, position_size=position_size, leverage=leverage, params=spec.params, data_dir=data_dir, executor=executor, exec_instrument=exec_instrument, ) def _worker_equity(w) -> float: inner = getattr(w, "worker", w) # smonta MLWorkerWrapper return float(getattr(inner, "capital", 0.0)) def rebalance_allocations(ledger: PortfolioLedger, workers: dict, weights: dict[str, float]): """Ribilancio: total_capital = Σ equity sleeve; riallinea il capitale-base di ogni worker a peso×total. I worker con posizione APERTA NON vengono ritoccati (la posizione mantiene il suo notional, come da approssimazione dichiarata): il nuovo capitale-base si applica alla prossima posizione, quando il worker è flat.""" ledger.total_capital = sum(_worker_equity(w) for w in workers.values()) alloc = ledger.allocate(weights) for sid, w in workers.items(): inner = getattr(w, "worker", w) if getattr(inner, "in_position", False): continue inner.capital = alloc.get(sid, inner.capital) ledger.save() _OHLCV = ["timestamp", "open", "high", "low", "close", "volume"] def _with_history(hist: pd.DataFrame | None, live: pd.DataFrame, warned: set | None = None, asset: str = "") -> pd.DataFrame: """Bootstrap storia per SH01 (punto-10, 2026-06-07): parquet locale PRIMA del feed live (dedup sul timestamp). La ri-validazione ha mostrato che l'edge SH01 richiede il training EXPANDING full-history: col solo lookback live (365g) la LogReg e' over-confident e la strategia NON e' robusta. Se c'e' un gap fra parquet e feed (parquet stantio oltre il lookback) si usa il SOLO feed con WARN: meglio il regime corto dichiarato che una serie con un buco.""" if hist is None or live is None or live.empty: return live lo = int(live["timestamp"].iloc[0]) h = hist[hist["timestamp"] < lo] if h.empty: return live if int(h["timestamp"].iloc[-1]) < lo - 2 * 3_600_000: # buco > 1 barra 1h if warned is not None and asset not in warned: warned.add(asset) print(f"[runner] WARN: gap fra parquet e feed live per {asset} " f"(parquet stantio? rilanciare download_all) — SH01 senza bootstrap") return live return pd.concat([h[_OHLCV], live[_OHLCV]], ignore_index=True) def _resample(df: pd.DataFrame, tf: str) -> pd.DataFrame: """Resampla candele 1h -> 4h/1d mantenendo timestamp ms reale (come get_df del backtest).""" if tf == "1h": return df rule = {"4h": "4h", "1d": "1D"}[tf] d = df.copy() d["dt"] = pd.to_datetime(d["timestamp"], unit="ms", utc=True) d = d.set_index("dt") agg = d.resample(rule).agg({"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}).dropna() epoch = pd.Timestamp("1970-01-01", tz="UTC") agg["timestamp"] = ((agg.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64") return agg.reset_index(drop=True) def _spec_assets_tf(spec: SleeveSpec): """(lista asset, tf) coinvolti da uno sleeve.""" if spec.kind == "pairs": return [spec.a, spec.b], spec.tf if spec.kind in _MULTI_KINDS: return list(spec.params["universe"]), spec.params.get("tf", "1d" if spec.kind != "basket" else "4h") return [spec.asset], spec.tf _STALE_BARS = 2 # barre 1h COMPLETE consecutive flat (O=H=L=C) -> feed fermo def _check_stale_feed(asset: str, df: pd.DataFrame, alerted: set[str]): """Osservabilita' (2026-06-05): alert Telegram quando il feed e' flat/fermo da >= _STALE_BARS barre 1h complete (i worker sono ciechi: il prossimo prezzo reale puo' gappare attraverso TP/SL, come ETH flat 13:00-14:50 -> gap 1655->1600) e al risveglio (con il gap % del primo prezzo reale). Una notifica per episodio. NB: SOLO osservabilita' — saltare gli ingressi post-flat PEGGIORA l'edge (testato: la candela-gap e' l'overshoot che la fade fada con profitto).""" from src.live.bars import last_settled_idx from src.live.telegram_notifier import notify_event if len(df) < _STALE_BARS + 2: return o, h, l, c = (df[k].values for k in ("open", "high", "low", "close")) # ultima barra COMPLETA (detection condivisa src.live.bars; prima hardcodava 1h) k = last_settled_idx(df["timestamp"].values) i = len(c) + k if i < 1: return flat = (o == h) & (h == l) & (l == c) n_flat = 0 while i - n_flat >= 0 and flat[i - n_flat]: n_flat += 1 if n_flat >= _STALE_BARS and asset not in alerted: alerted.add(asset) notify_event("STALE_FEED", {"asset": asset, "flat_bars_1h": n_flat, "ultimo_prezzo": float(c[i])}) elif n_flat == 0 and asset in alerted: alerted.discard(asset) gap = (c[i] / c[i - 1] - 1) * 100 if c[i - 1] else 0.0 notify_event("STALE_FEED", {"asset": asset, "status": "RIPRESO", "gap_pct": round(gap, 2), "prezzo": float(c[i])}) def run(config_path: str = "portfolios.yml"): """Loop live a portafoglio (tutti i tipi di sleeve). Data layer Cerbero v2 con resample; ribilancio a cambio giornata UTC.""" import time from datetime import datetime, timezone, timedelta import yaml from src.portfolio.base import load_active_portfolio from src.portfolio.sleeves import sleeve_returns_df from src.portfolio import weighting as W from src.live.cerbero_client import CerberoClient from src.live.multi_runner import INSTRUMENT_MAP p: Portfolio = load_active_portfolio(config_path) _ov = (yaml.safe_load(Path(config_path).read_text()) or {}).get("overrides", {}) poll = int(_ov.get("poll_seconds", 60)) # Frazione di capitale-sleeve impegnata per posizione (default 0.15 = canonico backtest). # Con leva 2x: notional = capital * position_size * 2. A 0.5 ogni sleeve in posizione # impegna il 100% della sua fetta (max impiego senza debito di margine); DD scala ~lineare. position_size = float(_ov.get("position_size", 0.15)) # Override PER-FAMIGLIA (improvement-sweep punto 8): la chiave e' la famiglia di # weighting.family_of (PAIRS/FADE/HONEST/SHAPE/TSM). Nato per i pairs: tutta la # validazione PR01 e' a pos 0.15 e la famiglia e' SENZA stop -> il pos globale 0.5 # la faceva girare a ~2.2x l'esposizione validata. Gate: pairspos_port06_impact.py. ps_family = {str(k).upper(): float(v) for k, v in (_ov.get("position_size_family") or {}).items()} def _supported(s): return s.kind in ("pairs",) + _MULTI_KINDS or s.name in _STRAT_MODULE supported = [s for s in p.sleeves if _supported(s)] skipped = [s.sid for s in p.sleeves if not _supported(s)] if skipped: print(f"[runner] sleeve saltati nel live (worker non disponibili): {skipped}") # SLEEVE "PAPER" (solo statistica, 2026-06-08): NON entrano nel pool/pesi/ledger del # portafoglio — i €total_capital si dividono SOLO tra gli sleeve reali. I paper girano # con un capitale nozionale fisso (la fetta equal che avrebbero avuto) per raccogliere # statistica in vista di future implementazioni reali. Default: TR01/ROT02/TSM01 # (multi-asset, esecuzione reale bloccata dal capitale). paper_codes = {str(c).upper() for c in (_ov.get("paper_sleeves") or [])} live_specs = [s for s in supported if s.name.upper() not in paper_codes] paper_specs = [s for s in supported if s.name.upper() in paper_codes] if paper_specs: print(f"[runner] sleeve PAPER (solo statistica, fuori dal pool): " f"{[s.sid for s in paper_specs]}") live_ids = [s.sid for s in live_specs] clusters = {s.sid: (s.cluster or s.sid) for s in live_specs} paper_notional = p.total_capital / max(len(supported), 1) # fetta equal nozionale ledger = PortfolioLedger(p.code, total_capital=p.total_capital) client = CerberoClient() # --- Esecuzione REALE (shadow) su Deribit testnet, solo sui fade abilitati --- # overrides.execution: {enabled, sleeves:[MR01,...], instruments:{BTC:..,ETH:..}} _exec_cfg = _ov.get("execution", {}) or {} exec_enabled = bool(_exec_cfg.get("enabled")) exec_sleeves = set(_exec_cfg.get("sleeves", [])) exec_instr = _exec_cfg.get("instruments", {}) or {} pairs_exec_enabled = bool(_exec_cfg.get("pairs_enabled")) # esecuzione reale 2 gambe executor = None pairs_executor = None if exec_enabled: from src.live.execution import ExecutionClient executor = ExecutionClient(client=client) # disaster-bracket on-book (~-30%): assicurazione outage sui fade reali executor.disaster_sl_pct = float(_exec_cfg.get("disaster_sl_pct", 0.30) or 0) or None print(f"[runner] ESECUZIONE REALE attiva (shadow) — sleeve={sorted(exec_sleeves)} " f"strumenti={exec_instr} disaster_sl={executor.disaster_sl_pct}") if pairs_exec_enabled: from src.live.execution import PairsExecutionClient pairs_executor = PairsExecutionClient(leg=executor) print(f"[runner] ESECUZIONE REALE PAIRS (2 gambe) attiva — strumenti={exec_instr}") def _exec_for(s): """(executor, exec_instrument) per uno sleeve single-leg ABILITATO. Kind: 'single' (fade/DIP01) e 'ml' (SH01). SH01 non ha TP/SL -> _place_real_tp ritorna subito e _real_close chiude tutto a market reduce-only (orizzonte): infrastruttura gia' presente. Il disaster-bracket on-book resta l'unica protezione di coda di SH01 durante un outage (esce a H=12 ben prima del -30%).""" if not exec_enabled or s.kind not in ("single", "ml") or s.name not in exec_sleeves: return None, None return executor, exec_instr.get(s.asset) def _pairs_exec_for(s): """(pairs_executor, {asset: instrument}) per uno sleeve pairs, se abilitato.""" if not pairs_exec_enabled or s.kind != "pairs": return None, None return pairs_executor, exec_instr dr = sleeve_returns_df(live_ids) weights = W.weight_vector(p.weighting, live_ids, dr, weights=p.weights, caps=p.caps, clusters=clusters, lookback=p.vol_lookback) alloc = ledger.allocate(weights) workers = {} for s in live_specs: ex, inst = _exec_for(s) pex, pinst = _pairs_exec_for(s) workers[s.sid] = build_worker_for(s, alloc[s.sid], p.leverage, position_size=pos_for_spec(s.sid, position_size, ps_family, s.params.get("position_size")), executor=ex, exec_instrument=inst, pairs_executor=pex, exec_instruments=pinst) if ps_family: print(f"[runner] position_size globale={position_size} override famiglia={ps_family}") # worker PAPER (solo statistica): capitale nozionale fisso, NESSUNA esecuzione reale, # NON nel ledger del portafoglio. Salvano in data/portfolio_paper_stats/. paper_dir = DATA_DIR.parent / "portfolio_paper_stats" paper_workers = {s.sid: build_worker_for(s, paper_notional, p.leverage, data_dir=paper_dir, position_size=pos_for_spec(s.sid, position_size, ps_family, s.params.get("position_size"))) for s in paper_specs} # bootstrap storia full per gli sleeve ML (SH01): parquet locale + feed live. # L'edge SH01 richiede train expanding full-history (sh01_trainwindow_validate); # il path live fitta solo l'ultimo blocco (last_block_only nei params SHAPE). ml_hist: dict[str, pd.DataFrame] = {} ml_gap_warned: set[str] = set() for a in sorted({s.asset for s in live_specs if s.kind == "ml"}): try: from src.data.downloader import load_data ml_hist[a] = load_data(a, "1h") last = pd.to_datetime(ml_hist[a]["timestamp"].iloc[-1], unit="ms", utc=True) print(f"[runner] bootstrap storia SH01 {a}: {len(ml_hist[a])} barre parquet " f"(fino a {last:%Y-%m-%d %H:%M})") except Exception as e: print(f"[runner] WARN bootstrap storia {a} fallito: {e} — SH01 col solo feed") # lookback (giorni) richiesto per ogni asset = max sui worker che lo usano asset_days: dict[str, int] = {} for s in live_specs: assets, tf = _spec_assets_tf(s) days = _LOOKBACK_DAYS.get(tf, 90) if s.kind == "ml": # SH01 ha bisogno di molta storia 1h days = max(days, _ML_LOOKBACK_DAYS) for a in assets: asset_days[a] = max(asset_days.get(a, 0), days) # timeframe SUB-orari (es. pairs 15m, flat-skip): non resamplabili dal 1h -> # fetch DIRETTO da Cerbero per (asset, tf). Inerte se nessuno sleeve e' sub-orario. subhourly_needs: dict[tuple[str, str], int] = {} for s in supported: # live + paper assets, tf = _spec_assets_tf(s) if tf in _SUBHOURLY: for a in assets: subhourly_needs[(a, tf)] = max(subhourly_needs.get((a, tf), 0), _LOOKBACK_DAYS.get(tf, 14)) if subhourly_needs: print(f"[runner] timeframe sub-orari (fetch diretto Cerbero): {sorted(subhourly_needs)}") inst_map = dict(INSTRUMENT_MAP) last_day = "" stale_alerted: set[str] = set() # asset con alert STALE_FEED attivo (dedup per episodio) # Osservabilita' outage (improvement-sweep 2026-06-06): il poll-loop intero e' in un # try/except → durante un outage i worker NON valutano gli exit. Alert Telegram dopo # _OUTAGE_POLLS poll falliti/DEGRADATI consecutivi + notifica di ripresa con durata. # "Degradato" include il caso HTTP-200-con-candles-vuote (code review 2026-06-07): # non solleva eccezione ma i worker dell'asset mancante saltano il tick in silenzio. _OUTAGE_POLLS = 5 fail_streak = 0 def _outage_tick(failed: bool, streak: int, detail: str = "") -> int: """Aggiorna lo streak e gestisce gli alert FEED_OUTAGE (start a soglia, una volta per episodio; RIPRESO al primo poll pulito). Ritorna il nuovo streak.""" from src.live.telegram_notifier import notify_event if failed: streak += 1 if streak == _OUTAGE_POLLS: real_open = sorted(sid for sid, wk in workers.items() if getattr(wk, "real_in_position", False)) notify_event("FEED_OUTAGE", { "poll_falliti": streak, "minuti": round(streak * poll / 60), "dettaglio": detail, "posizioni_reali_aperte": ", ".join(real_open) or "nessuna", "nota": "exit NON valutati durante l'outage; " "protezione = disaster-SL on-book sui fade reali"}) return streak if streak >= _OUTAGE_POLLS: notify_event("FEED_OUTAGE", {"status": "RIPRESO", "poll_falliti": streak, "minuti": round(streak * poll / 60)}) return 0 while True: try: # fetch 1h per asset al lookback massimo richiesto raw1h: dict[str, pd.DataFrame] = {} end = datetime.now(timezone.utc) # SOLO testnet (via Cerbero): il paper DEVE usare lo stesso venue dove gli ordini # verrebbero eseguiti (testnet). Mai sostituire con dati mainnet -> divergerebbe dal # comportamento reale (prezzi/liquidità testnet != mainnet). Durante un outage testnet # il runner si mette in pausa (corretto: senza il venue non si potrebbe eseguire). for asset, days in asset_days.items(): inst = inst_map.get(asset, f"{asset}-PERPETUAL") start = end - timedelta(days=days) candles = client.get_historical_v2(inst, start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d"), "1h") if candles: df = pd.DataFrame(candles) df["timestamp"] = df["timestamp"].astype("int64") raw1h[asset] = df.sort_values("timestamp").reset_index(drop=True) _check_stale_feed(asset, raw1h[asset], stale_alerted) # fetch DIRETTO dei timeframe sub-orari (15m...) per (asset, tf) raw_sub: dict[tuple[str, str], pd.DataFrame] = {} for (asset, tf), days in subhourly_needs.items(): inst = inst_map.get(asset, f"{asset}-PERPETUAL") start = end - timedelta(days=days) candles = client.get_historical_v2(inst, start.strftime("%Y-%m-%d"), end.strftime("%Y-%m-%d"), tf) if candles: df = pd.DataFrame(candles) df["timestamp"] = df["timestamp"].astype("int64") raw_sub[(asset, tf)] = df.sort_values("timestamp").reset_index(drop=True) def _series_for(a, tf): """Serie OHLC per (asset, tf): diretta se sub-oraria, altrimenti resample dal 1h.""" if tf in _SUBHOURLY: return raw_sub.get((a, tf)) return _resample(raw1h[a], tf) if a in raw1h else None # tick di ogni worker col suo timeframe def _tick(s, w): assets, tf = _spec_assets_tf(s) res = {a: _series_for(a, tf) for a in assets} if any(res[a] is None or len(res[a]) == 0 for a in assets): return if s.kind == "pairs": w.tick(res[s.a], res[s.b]) elif s.kind in _MULTI_KINDS: w.tick(res) elif s.kind == "ml": # SH01: storia full (parquet bootstrap + feed) -> il walk-forward # interno fitta solo l'ultimo blocco (last_block_only nei params). w.tick(_with_history(ml_hist.get(s.asset), res[s.asset], ml_gap_warned, s.asset)) else: # single (fade/dip): StrategyWorker su feed live. w.tick(res[s.asset]) for s in live_specs: _tick(s, workers[s.sid]) # PAPER: ticcati per statistica, MAI nel ledger del portafoglio for s in paper_specs: _tick(s, paper_workers[s.sid]) ledger.update_equity({sid: _worker_equity(wk) for sid, wk in workers.items()}) today = datetime.now(timezone.utc).strftime("%Y-%m-%d") if today != last_day and last_day: dr = sleeve_returns_df(live_ids) weights = W.weight_vector(p.weighting, live_ids, dr, weights=p.weights, caps=p.caps, clusters=clusters, lookback=p.vol_lookback) rebalance_allocations(ledger, workers, weights) last_day = today ledger.save() # feed degradato senza eccezione: asset richiesti ma senza candele missing = sorted(a for a in asset_days if a not in raw1h) if missing: print(f"[runner] feed incompleto: mancano {missing} (streak {fail_streak + 1})") fail_streak = _outage_tick(bool(missing), fail_streak, detail=f"feed senza candele per: {', '.join(missing)}") except KeyboardInterrupt: ledger.save() print("shutdown") break except Exception as e: print(f"[runner] errore: {e} (streak {fail_streak + 1})") fail_streak = _outage_tick(True, fail_streak, detail=f"eccezione: {e}") time.sleep(poll) if __name__ == "__main__": run()