@@ -1,701 +0,0 @@
""" 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 . xsec_worker import CrossSectionalWorker
from src . live . grid_worker import GridWorker
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 " , " xsec " )
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 ,
real_truth : bool = False ) :
""" 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). real_truth: il ledger `capital` si aggiorna col PnL
dei FILL REALI (sim ridotto a diagnostica) — inerte senza executor. """
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 ,
real_truth = real_truth ,
)
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 ,
)
if spec . kind == " xsec " :
pr = spec . params
return CrossSectionalWorker (
universe = pr [ " universe " ] , tf = pr . get ( " tf " , " 1h " ) ,
params = { " lb " : pr . get ( " lb " , 48 ) , " hold " : pr . get ( " hold " , 12 ) ,
" disp_min " : pr . get ( " disp_min " ) ,
" tranches " : pr . get ( " tranches " , 1 ) } ,
capital = alloc_capital , position_size = position_size , leverage = leverage ,
data_dir = data_dir ,
)
if spec . kind == " grid " :
# Price Ladder (griglia) — SIM/PAPER (shadow stage 1): nessun ordine reale.
return GridWorker (
sid = spec . sid , asset = spec . asset , params = spec . params , capital = alloc_capital ,
work_dir = Path ( data_dir ) / spec . sid , leverage = leverage ,
position_size = position_size , fee_side = 0.0005 ,
)
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 , real_truth = real_truth ,
)
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×
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 ( ) )
# i worker in posizione TRATTENGONO il loro capitale (deployato): vanno passati
# come `reserved` ad allocate, cosi' i flat si dividono solo il RESTO e l'equity
# totale e' conservata (fix doppio conteggio 2026-06-13, vedi ledger.allocate).
reserved = { sid : _worker_equity ( w ) for sid , w in workers . items ( )
if getattr ( getattr ( w , " worker " , w ) , " in_position " , False ) }
alloc = ledger . allocate ( weights , reserved = reserved )
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 " )
if spec . kind == " grid " :
return [ spec . asset ] , spec . params . get ( " tf " , spec . tf )
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 % d el 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 ] ) } )
# --- Gate feed CONGELATO (2026-06-15) -----------------------------------------
# Distinto da STALE_FEED (osservabilita') e FEED_BOOK_GAP (divergenza esecuzione):
# qui si AGISCE. Quando il feed di DECISIONE 1h di un asset e' CONGELATO (guasto
# testnet: ETH-PERPETUAL inverse fermo a 1661.95 per 36h+, prezzo MAI cambiato) gli
# sleeve CONCENTRATI che ne dipendono (single/ml/pairs) decidono entry/exit su un
# prezzo morto -> segnali spuri (z-score pairs estremi -3/-5, SH01 short a prezzo
# fermo) e perdite REALI (i fill avvengono al book vero ~1717, l'uscita sim al prezzo
# congelato: SH01_ETH ha realizzato -2.83 reali vs -0.09 sim su un solo close). Si
# SALTA il tick (entry E exit) finche' il feed non si sblocca: come un outage (i worker
# non valutano gli exit, protezione = disaster-SL on-book), auto-guarente.
#
# DISTINGUERE guasto da ILLIQUIDITA': SOL/LTC/ADA stampano molte barre flat (O=H=L=C,
# 50-95%) ma il prezzo SI muove ogni poche ore (run di close identiche CORTE, ~2-5). Il
# guasto e' il prezzo che non cambia MAI: run di close INVARIATE >= soglia (il freeze
# reale dura decine di barre). Un detector flat-bar ingenuo gaterebbe gli alt
# illiquidi-ma-VIVI -> si conta la run di close invariate, NON le barre flat.
#
# NB POST-FLAT: si gatea DURANTE il freeze (ultima barra completa ferma). La barra di
# RIPRESA e' non-flat -> la run si azzera -> il tick riprende SU di essa. NON e'
# l'entry-guard post-flat (BOCCIATA: la candela-gap e' l'overshoot che la fade fada con
# profitto, CLAUDE.md / 2026-06-05): quello salta la barra di ripresa, questo no.
#
# SOGLIA (misurata sul feed reale 2026-06-15): le due popolazioni sono ben separate ->
# MORTO ETH run 64 (1 val distinto/48h), BNB 64, DOGE 42 (feed pinnato a un valore)
# ILLIQUIDO LTC run 10, ADA 11, XRP 12, SOL 1 (5-31 val distinti/48h: SI muovono)
# A 24 (un giorno intero di prezzo immobile) il guasto multi-giorno e' preso con ampio
# margine senza gateare gli alt illiquidi-ma-VIVI (importante per PR_BTCLTC: BTC vivo +
# LTC solo illiquido NON deve sospendere il pair). ETH (run 64) gatea comunque subito.
_FREEZE_GATE_BARS = 24 # run di close INVARIATE sull'ultima barra completa = freeze
def _frozen_assets ( raw1h : dict [ str , pd . DataFrame ] , threshold : int ) - > set [ str ] :
""" Asset col feed di decisione 1h CONGELATO: ultima barra completa flat (O=H=L=C)
E close invariata da >= `threshold` barre complete consecutive. 0 = gate disattivo. """
from src . live . bars import last_settled_idx
frozen : set [ str ] = set ( )
if threshold < = 0 :
return frozen
for asset , df in raw1h . items ( ) :
o , h , l , c = ( df [ k ] . values for k in ( " open " , " high " , " low " , " close " ) )
i = len ( c ) + last_settled_idx ( df [ " timestamp " ] . values )
if i < 1 or not ( o [ i ] == h [ i ] == l [ i ] == c [ i ] ) :
continue # ultima barra completa non flat -> vivo
run = 1
while i - run > = 0 and c [ i - run ] == c [ i ] :
run + = 1
if run > = threshold :
frozen . add ( asset )
return frozen
def _feed_gated_sids ( live_specs , frozen : set [ str ] ) - > set [ str ] :
""" sid degli sleeve CONCENTRATI (single/ml/pairs) che dipendono da un asset col feed
congelato. I multi-asset (basket/rotation/tsmom/xsec, tutti PAPER) NON sono gateati:
diversificati su 8 asset, un singolo feed fermo non li compromette. """
if not frozen :
return set ( )
return { s . sid for s in live_specs
if s . kind in ( " single " , " ml " , " pairs " )
and any ( a in frozen for a in _spec_assets_tf ( s ) [ 0 ] ) }
_GAP_BPS_DEFAULT = 150.0 # |close feed - mark book| oltre cui il feed non e' affidabile
def _check_feed_book_gap ( client , raw1h , instruments , threshold_bps , alerted ) :
""" Osservabilita ' (2026-06-12): il feed candele e il book dove fillano gli ordini
REALI possono divergere — caso MR02_BTC: TP resting fillato a 60481 nella notte
col feed mai sceso sotto 63285 (-443 bps, scoperto solo al close sim); i wick
TP_PHANTOM sono il caso opposto (feed stampa, book non scambia). Confronta il
close della candela in corso col MARK dello strumento d ' ESECUZIONE (USDC):
oltre soglia -> alert FEED_BOOK_GAP, una notifica per episodio, recovery con
isteresi a soglia/2. Le decisioni restano sul feed (il sim e ' la verita ' che
guida): questo dice solo QUANDO i fill reali possono divergere dal sim. """
from src . live . telegram_notifier import notify_event
want = { a : inst for a , inst in instruments . items ( ) if a in raw1h }
if not want :
return
try :
out = client . get_ticker_batch ( list ( want . values ( ) ) )
marks = { t . get ( " instrument_name " ) : ( t . get ( " mark_price " ) or t . get ( " last_price " ) )
for t in out . get ( " tickers " , [ ] ) }
except Exception :
return # fail-open: solo osservabilita'
for asset , inst in want . items ( ) :
mark = marks . get ( inst )
feed = float ( raw1h [ asset ] [ " close " ] . iloc [ - 1 ] )
if not mark or not feed :
continue
gap_bps = abs ( feed / float ( mark ) - 1 ) * 10_000
if gap_bps > = threshold_bps and asset not in alerted :
alerted . add ( asset )
print ( f " [runner] FEED_BOOK_GAP { asset } : feed { feed } vs mark { mark } "
f " ( { gap_bps : .0f } bps) " )
notify_event ( " FEED_BOOK_GAP " , {
" asset " : asset , " feed_close " : feed , " mark_book " : float ( mark ) ,
" gap_bps " : round ( gap_bps , 1 ) ,
" nota " : " feed candele != book d ' esecuzione: i fill reali possono "
" divergere dal sim (TP fantasma / fill non visti dal feed) " } )
elif gap_bps < threshold_bps / 2 and asset in alerted :
alerted . discard ( asset )
notify_event ( " FEED_BOOK_GAP " , { " asset " : asset , " status " : " RIENTRATO " ,
" gap_bps " : round ( gap_bps , 1 ) } )
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 ]
# PAPER_EXTRA: sleeve paper definiti SOLO in config (NON in p.sleeves) -> NON entrano
# nel backtest canonico/regression-lock (all_sleeve_equities non sa costruirli). Nato
# per il Price Ladder (kind=grid, shadow stage 1 sim). Parsing DIFENSIVO: un errore qui
# non deve crashare il runner mainnet.
for _ex in ( _ov . get ( " paper_extra " ) or [ ] ) :
try :
paper_specs . append ( SleeveSpec (
kind = str ( _ex [ " kind " ] ) , name = str ( _ex . get ( " name " , _ex [ " sid " ] ) ) ,
sid = str ( _ex [ " sid " ] ) , asset = _ex . get ( " asset " ) ,
tf = str ( _ex . get ( " tf " , " 1h " ) ) , params = dict ( _ex . get ( " params " , { } ) ) ,
cluster = str ( _ex . get ( " cluster " , " " ) ) ,
) )
except Exception as e :
print ( f " [runner] WARN paper_extra ignorato ( { _ex } ): { e } " )
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
# REAL-TRUTH (2026-06-10): il capitale degli sleeve eseguiti si aggiorna col PnL
# dei fill reali (sim = diagnostica). Default False: va acceso esplicitamente in yml.
real_truth = bool ( _exec_cfg . get ( " real_truth " , False ) )
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 } "
f " real_truth= { real_truth } " )
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 ,
real_truth = real_truth )
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 + paper_specs : # live + PAPER (incl. paper_extra grid)
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 live_specs + paper_specs : # live + paper (incl. paper_extra grid)
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)
# guard feed-vs-book (2026-06-12): soglia bps in overrides.feed_book_gap_bps (0 = off)
gap_bps = float ( _ov . get ( " feed_book_gap_bps " , _GAP_BPS_DEFAULT ) )
gap_alerted : set [ str ] = set ( )
# gate feed CONGELATO (2026-06-15): salta gli sleeve concentrati su un asset col feed
# fermo. Soglia (barre 1h di close invariata) in overrides.feed_freeze_gate_bars (0 = off).
freeze_gate_bars = int ( _ov . get ( " feed_freeze_gate_bars " , _FREEZE_GATE_BARS ) )
spec_by_id = { s . sid : s for s in live_specs }
frozen_gated : set [ str ] = set ( ) # sleeve gateati ora (dedup alert + recovery)
# 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
worker_err_streak : dict [ str , int ] = { } # errori consecutivi per worker (isolamento tick)
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 )
if exec_enabled and gap_bps > 0 :
_check_feed_book_gap ( client , raw1h , exec_instr , gap_bps , gap_alerted )
# gate feed CONGELATO: sleeve concentrati su un asset col feed fermo -> tick saltato
frozen = _frozen_assets ( raw1h , freeze_gate_bars )
now_gated = _feed_gated_sids ( live_specs , frozen )
if now_gated != frozen_gated :
from src . live . telegram_notifier import notify_event
for sid in sorted ( now_gated - frozen_gated ) :
fa = sorted ( set ( _spec_assets_tf ( spec_by_id [ sid ] ) [ 0 ] ) & frozen )
print ( f " [runner] FEED_FROZEN_GATE { sid } : asset congelati { fa } -> tick sospeso " )
notify_event ( " FEED_FROZEN_GATE " , {
" sleeve " : sid , " asset_congelati " : " , " . join ( fa ) , " stato " : " GATED " ,
" nota " : " feed di decisione fermo: entry+exit sospesi finche ' non si "
" sblocca (disaster-SL on-book protegge le posizioni reali) " } )
for sid in sorted ( frozen_gated - now_gated ) :
print ( f " [runner] FEED_FROZEN_GATE { sid } : feed ripreso -> tick riattivato " )
notify_event ( " FEED_FROZEN_GATE " , { " sleeve " : sid , " stato " : " RIPRESO " } )
frozen_gated = now_gated
# 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 ) )
elif s . kind == " grid " :
# Price Ladder SIM/PAPER: ricomputa la griglia sul feed live (BTC 1h).
w . tick ( res [ s . asset ] )
else :
# single (fade/dip): StrategyWorker su feed live.
w . tick ( res [ s . asset ] )
# isolamento per-worker (audit 2026-06-11): un'eccezione in un tick NON
# deve saltare gli altri worker (= exit non valutati per tutti) ne'
# l'update dell'equity. Streak per-worker -> alert dopo 5 fail di fila.
def _tick_safe ( s , w ) :
try :
_tick ( s , w )
n_prev = worker_err_streak . pop ( s . sid , 0 )
if n_prev > = 5 :
# recovery dichiarato (code-review: alert one-shot senza
# ripresa = silenzio indistinguibile dal guasto)
from src . live . telegram_notifier import notify_event
notify_event ( " WORKER_ERROR_STREAK " ,
{ " worker " : s . sid , " status " : " RIPRESO " ,
" dopo_streak " : n_prev } )
except Exception as e :
n = worker_err_streak . get ( s . sid , 0 ) + 1
worker_err_streak [ s . sid ] = n
print ( f " [runner] errore worker { s . sid } : { e } (streak { n } ; gli altri proseguono) " )
# alert a 5 e poi ogni 50 poll (~1h): un worker rotto per
# giorni non deve sparire dopo il primo Telegram. Include
# se ha una posizione REALE aperta (exit non valutati!)
if n == 5 or n % 50 == 0 :
from src . live . telegram_notifier import notify_event
notify_event ( " WORKER_ERROR_STREAK " , {
" worker " : s . sid , " streak " : n , " errore " : str ( e ) [ : 200 ] ,
" real_in_position " : bool ( getattr ( w , " real_in_position " , False ) ) ,
" in_position " : bool ( getattr ( w , " in_position " , False ) ) ,
" nota " : " exit NON valutati per questo worker " } )
for s in live_specs :
if s . sid in now_gated :
continue # feed di decisione congelato -> entry+exit sospesi
_tick_safe ( s , workers [ s . sid ] )
# PAPER: ticcati per statistica, MAI nel ledger del portafoglio
for s in paper_specs :
_tick_safe ( 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 ( )
Adriano Dal Pastro