Adriano Dal Pastro
8e72d7ad02
Deribit testnet (test.deribit.com) va giu' periodicamente (502) e Cerbero lo rilancia -> il runner si bloccava senza dati. Aggiunto CerberoClient.get_historical_mainnet (Deribit MAINNET public, NO-AUTH, paginato sotto il cap ~5000 candele/chiamata) e fallback nel runner: try Cerbero -> on fail/empty usa mainnet. Prezzi REALI (meglio del testnet farlocco per il paper). Verificato durante l'outage: tutti gli 8 strumenti (BTC/ETH + alt _USDC) coperti su mainnet. Log una-tantum all'attivazione/disattivazione del fallback. Caveat: testnet e mainnet hanno prezzi diversi (~9%) -> al primo switch le posizioni aperte su prezzi testnet vanno resettate (transizione pulita). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
246 lines
11 KiB
Python
246 lines
11 KiB
Python
"""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 = {"1h": 90, "4h": 220, "1d": 440}
|
||
# 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
|
||
|
||
# stato del fallback dati: True quando Cerbero (testnet) è giù e usiamo Deribit MAINNET public
|
||
_MAINNET_FALLBACK = {"on": False}
|
||
|
||
|
||
def build_worker_for(spec: SleeveSpec, alloc_capital: float, leverage: float,
|
||
data_dir: Path = DATA_DIR, position_size: float = 0.15):
|
||
"""Costruisce il worker esecutore per uno sleeve con capitale = quota allocata."""
|
||
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,
|
||
)
|
||
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,
|
||
)
|
||
|
||
|
||
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()
|
||
|
||
|
||
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
|
||
|
||
|
||
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))
|
||
|
||
def _supported(s):
|
||
return s.kind in ("pairs",) + _MULTI_KINDS or s.name in _STRAT_MODULE
|
||
live_specs = [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}")
|
||
live_ids = [s.sid for s in live_specs]
|
||
clusters = {s.sid: (s.cluster or s.sid) for s in live_specs}
|
||
|
||
ledger = PortfolioLedger(p.code, total_capital=p.total_capital)
|
||
client = CerberoClient()
|
||
|
||
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 = {s.sid: build_worker_for(s, alloc[s.sid], p.leverage) for s in live_specs}
|
||
|
||
# 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)
|
||
|
||
inst_map = dict(INSTRUMENT_MAP)
|
||
last_day = ""
|
||
while True:
|
||
try:
|
||
# fetch 1h per asset al lookback massimo richiesto
|
||
raw1h: dict[str, pd.DataFrame] = {}
|
||
end = datetime.now(timezone.utc)
|
||
for asset, days in asset_days.items():
|
||
inst = inst_map.get(asset, f"{asset}-PERPETUAL")
|
||
start = end - timedelta(days=days)
|
||
candles = None
|
||
try:
|
||
candles = client.get_historical_v2(inst, start.strftime("%Y-%m-%d"),
|
||
end.strftime("%Y-%m-%d"), "1h")
|
||
except Exception:
|
||
candles = None
|
||
if not candles:
|
||
# FALLBACK: Cerbero (testnet) giù -> OHLCV reale da Deribit MAINNET public
|
||
candles = client.get_historical_mainnet(
|
||
inst, int(start.timestamp() * 1000), int(end.timestamp() * 1000), "60")
|
||
if candles and not _MAINNET_FALLBACK["on"]:
|
||
_MAINNET_FALLBACK["on"] = True
|
||
print("[runner] FALLBACK attivo: OHLCV da Deribit MAINNET (Cerbero/testnet non disponibile)")
|
||
elif _MAINNET_FALLBACK["on"]:
|
||
_MAINNET_FALLBACK["on"] = False
|
||
print("[runner] Cerbero tornato disponibile: fallback mainnet disattivato")
|
||
if candles:
|
||
df = pd.DataFrame(candles)
|
||
df["timestamp"] = df["timestamp"].astype("int64")
|
||
raw1h[asset] = df.sort_values("timestamp").reset_index(drop=True)
|
||
|
||
# tick di ogni worker col suo timeframe (resample dal 1h)
|
||
for s in live_specs:
|
||
w = workers[s.sid]
|
||
assets, tf = _spec_assets_tf(s)
|
||
if any(a not in raw1h for a in assets):
|
||
continue
|
||
res = {a: _resample(raw1h[a], tf) for a in assets}
|
||
if s.kind == "pairs":
|
||
w.tick(res[s.a], res[s.b])
|
||
elif s.kind in _MULTI_KINDS:
|
||
w.tick(res)
|
||
else:
|
||
# single (fade/dip) e ml (SH01): StrategyWorker. SH01 retraina dentro
|
||
# generate_signals (walk-forward) -> nessun training esterno.
|
||
w.tick(res[s.asset])
|
||
|
||
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()
|
||
except KeyboardInterrupt:
|
||
ledger.save()
|
||
print("shutdown")
|
||
break
|
||
except Exception as e:
|
||
print(f"[runner] errore: {e}")
|
||
time.sleep(poll)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run()
|