From 612f2bfced9e2ec6e2d7495d656291135e53f8ad Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Fri, 12 Jun 2026 20:29:02 +0000 Subject: [PATCH] feat(live): reconcile resting + orphan single-leg + circuit-breaker venue-lock + FEED_BOOK_GAP Codice della tornata v1.1.27/28 (gia' in produzione, mai committato): - reconcile_account: estensione ordini RESTING (FILLED_UNBOOKED/MISSING/STALE, caso MR02_BTC: TP fillato di notte scoperto ore dopo) + expected_resting in books - strategy_worker: orphan_legs su REAL_CLOSE_PARTIAL anche single-leg, persistito - execution: circuit-breaker su venue-lock admin (stop ordini dopo errori ripetuti) - runner/hourly_report: alert FEED_BOOK_GAP + timestamp closed trades - cerbero_client: get_open_orders (merge all + trigger_all) Test: 12 nuovi, suite completa 126 passed. Co-Authored-By: Claude Fable 5 --- scripts/analysis/reconcile_account.py | 132 ++++++++++++++++++--- scripts/portfolios/hourly_report.py | 18 ++- src/live/books.py | 22 ++++ src/live/cerbero_client.py | 10 ++ src/live/execution.py | 65 +++++++++- src/live/strategy_worker.py | 23 +++- src/portfolio/runner.py | 49 ++++++++ tests/portfolio/test_feed_book_gap.py | 73 ++++++++++++ tests/portfolio/test_reconcile_resting.py | 96 +++++++++++++++ tests/portfolio/test_single_leg_orphan.py | 79 ++++++++++++ tests/portfolio/test_venue_lock_breaker.py | 86 ++++++++++++++ 11 files changed, 628 insertions(+), 25 deletions(-) create mode 100644 tests/portfolio/test_feed_book_gap.py create mode 100644 tests/portfolio/test_reconcile_resting.py create mode 100644 tests/portfolio/test_single_leg_orphan.py create mode 100644 tests/portfolio/test_venue_lock_breaker.py diff --git a/scripts/analysis/reconcile_account.py b/scripts/analysis/reconcile_account.py index f713822..5d34a18 100644 --- a/scripts/analysis/reconcile_account.py +++ b/scripts/analysis/reconcile_account.py @@ -29,7 +29,8 @@ sys.path.insert(0, str(PROJECT_ROOT)) from src.live.cerbero_client import CerberoClient from src.live.execution import contract_spec -from src.live.books import real_books, account_net # fonte UNICA dei libri (usata +from src.live.books import real_books, account_net, expected_resting, PAPER + # fonte UNICA dei libri (usata # anche dal guard del netting) RECHECK_SLEEP = 10 # anti-race: secondi fra i due passaggi @@ -53,15 +54,78 @@ def compute_drift(client: CerberoClient | None = None) -> list[dict]: return rows +def _book_orders(client: CerberoClient) -> dict[str, dict]: + """Ordini aperti sul conto, merge type='all' + 'trigger_all' per order_id + (Deribit puo' omettere i trigger untriggered da 'all').""" + orders: dict[str, dict] = {} + for typ in ("all", "trigger_all"): + for o in client.get_open_orders(currency="USDC", type=typ) or []: + oid = str(o.get("order_id") or "") + if oid: + orders[oid] = o + return orders + + +def _resting_filled(client: CerberoClient, instrument: str, order_id: str) -> float: + """Amount fillato di un ordine resting dal trade history (fonte autorevole).""" + try: + return sum(float(t.get("amount", 0) or 0) + for t in client.get_trade_history(limit=100, + instrument_name=instrument) + if str(t.get("order_id")) == str(order_id)) + except Exception: + return 0.0 + + +def compute_resting_drift(client: CerberoClient | None = None) -> list[dict]: + """Reconcile degli ordini RESTING (estensione 2026-06-12, dopo il caso MR02_BTC: + TP resting fillato sul book di notte + disaster-SL sparito, scoperti solo al + close sim ore dopo). Tre classi di anomalia: + + FILLED_UNBOOKED l'ordine atteso non e' in book e ha fill nel trade history + mentre il worker si crede ancora in posizione (il caso MR02) + MISSING l'ordine atteso non e' in book e non ha fill (cancellato da + altri/exchange; per il DSL triggered il fill ha un order_id + NUOVO -> qui appare MISSING e il drift posizioni completa) + STALE ordine in book con label di un nostro worker ma NON atteso + dai libri (worker flat/morto: fillerebbe a sorpresa) + """ + client = client or CerberoClient() + expected = expected_resting() + book = _book_orders(client) + rows: list[dict] = [] + for e in expected: + if e["order_id"] in book: + rows.append({**e, "status": "OK"}) + continue + filled = _resting_filled(client, e["instrument"], e["order_id"]) + rows.append({**e, "status": "FILLED_UNBOOKED" if filled > 0 else "MISSING", + "filled": filled}) + exp_ids = {e["order_id"] for e in expected} + workers = {p.name for p in PAPER.glob("*") if p.is_dir()} + for oid, o in book.items(): + if oid not in exp_ids and str(o.get("label") or "") in workers: + rows.append(dict(worker=o.get("label"), instrument=o.get("instrument"), + order_id=oid, kind=o.get("order_type"), + status="STALE")) + return rows + + def main(): - rows = compute_drift() + client = CerberoClient() + rows = compute_drift(client) + resting = compute_resting_drift(client) bad = [r for r in rows if not r["ok"]] - if bad: + bad_rest = [r for r in resting if r["status"] != "OK"] + if bad or bad_rest: # anti-race: un worker poteva essere a meta' open/close -> ricontrolla - print(f"drift su {len(bad)} strumenti: ricontrollo fra {RECHECK_SLEEP}s (anti-race)...") + print(f"anomalie (pos={len(bad)} resting={len(bad_rest)}): " + f"ricontrollo fra {RECHECK_SLEEP}s (anti-race)...") time.sleep(RECHECK_SLEEP) - rows = compute_drift() + rows = compute_drift(client) + resting = compute_resting_drift(client) bad = [r for r in rows if not r["ok"]] + bad_rest = [r for r in resting if r["status"] != "OK"] print(f"{'strumento':<22}{'libri':>10}{'orfani':>9}{'atteso':>10}{'conto':>10}{'drift':>10} esito") for r in rows: @@ -69,21 +133,53 @@ def main(): f"{r['real']:>10.4f}{r['drift']:>+10.4f} {'OK' if r['ok'] else '⚠️ DRIFT'}") if not rows: print("(nessuna posizione attesa ne' reale)") - print("\nESITO:", "OK — conto allineato ai libri" if not bad - else f"⚠️ DRIFT PERSISTENTE su {len(bad)} strumenti") - if bad and "--telegram" in sys.argv: + print(f"\n{'resting':<45}{'kind':>6} {'order_id':<22} stato") + for r in resting: + print(f"{r['worker']:<45}{r['kind']:>6} {r['order_id']:<22} " + f"{'OK' if r['status'] == 'OK' else '⚠️ ' + r['status']}") + if not resting: + print("(nessun ordine resting atteso ne' in book)") + + print("\nESITO:", "OK — conto allineato ai libri" if not (bad or bad_rest) + else f"⚠️ DRIFT PERSISTENTE (pos={len(bad)} resting={len(bad_rest)})") + + if (bad or bad_rest) and "--telegram" in sys.argv: from src.live.telegram_notifier import notify_event - notify_event("ACCOUNT_DRIFT", { - "strumenti": {r["inst"]: {"atteso": round(r["exp"], 5), - "conto": round(r["real"], 5), - "drift": round(r["drift"], 5)} for r in bad}, - "note": ("conto != libri worker oltre tolleranza (drift NON spiegato dagli " - "orfani registrati): verificare close cappati/gambe respinte — " - "vedi docs/diary/2026-06-11-system-audit.md")}) - print("[telegram] alert ACCOUNT_DRIFT inviato") - return bool(bad) + if bad: + notify_event("ACCOUNT_DRIFT", { + "strumenti": {r["inst"]: {"atteso": round(r["exp"], 5), + "conto": round(r["real"], 5), + "drift": round(r["drift"], 5)} for r in bad}, + "note": ("conto != libri worker oltre tolleranza (drift NON spiegato dagli " + "orfani registrati): verificare close cappati/gambe respinte — " + "vedi docs/diary/2026-06-11-system-audit.md")}) + print("[telegram] alert ACCOUNT_DRIFT inviato") + if bad_rest: + notify_event("RESTING_DRIFT", { + "ordini": [{k: r.get(k) for k in ("worker", "kind", "order_id", "status", + "filled")} for r in bad_rest], + "note": ("FILLED_UNBOOKED = resting fillato col worker ancora in posizione " + "(caso MR02_BTC 2026-06-12); MISSING = atteso ma non in book; " + "STALE = in book senza libro corrispondente")}) + print("[telegram] alert RESTING_DRIFT inviato") + return bool(bad or bad_rest) if __name__ == "__main__": - sys.exit(1 if main() else 0) + # Il run delle 11:40 del 2026-06-12 e' morto in silenzio su un 502 (Deribit/gateway + # giu') -> il guardiano che non suona e' indistinguibile dal "tutto ok". Su errore: + # alert RECONCILE_FAIL (con --telegram) + exit 2. + try: + sys.exit(1 if main() else 0) + except Exception as exc: + print(f"RECONCILE_FAIL: {exc}") + if "--telegram" in sys.argv: + try: + from src.live.telegram_notifier import notify_event + notify_event("RECONCILE_FAIL", { + "errore": str(exc)[:200], + "note": "reconciler NON eseguito: conto non verificato in quest'ora"}) + except Exception: + pass + sys.exit(2) diff --git a/scripts/portfolios/hourly_report.py b/scripts/portfolios/hourly_report.py index 65d8bc3..8cd7029 100644 --- a/scripts/portfolios/hourly_report.py +++ b/scripts/portfolios/hourly_report.py @@ -96,8 +96,14 @@ def lossguard_section() -> str: return "\n".join(L) +# Epoca v1.1.26 (deploy 2026-06-11 ~21:40 UTC): gate TP_PHANTOM attivo. I close +# precedenti includono il churn da TP fantasma dell'11-06 17:32-17:58 (~24 giri, +# win-rate inquinato) -> le accuratezze "pulite" si leggono da qui in poi. +EPOCH_V1126 = "2026-06-11T21:40:00" + + def collect(): - closed = [] # (sleeve, reason, net_return, pnl, win) + closed = [] # (sleeve, reason, net_return, pnl, win, ts) open_pos = [] # dict per posizione aperta realized = 0.0 for sp in sorted(glob.glob(str(PAPER / "*" / "status.json"))): @@ -119,7 +125,7 @@ def collect(): # resta il sim diagnostico: sui TP fantasma da spike testnet diceva # 26/0 mentre il reale era 11/15). Fallback nr>0 per eventi storici. closed.append((_short(wid), ev.get("reason", "?"), nr, pnl, - bool(ev.get("win", nr > 0)))) + bool(ev.get("win", nr > 0)), ev.get("ts", ""))) if "positions" in st or "weights" in st: continue # multi-asset (TR01/ROT02/TSM01): sezione dedicata if st.get("in_position"): @@ -187,7 +193,7 @@ def build_report() -> str: # breakdown per motivo by_reason = defaultdict(lambda: [0, 0, 0.0]) # reason -> [win, loss, pnl] - for _, reason, _, pnl, win in closed: + for _, reason, _, pnl, win, _ in closed: r = by_reason[reason] r[0 if win else 1] += 1 r[2] += pnl @@ -206,8 +212,12 @@ def build_report() -> str: if eq is not None: L.append(f"Equity €{eq:.2f} | Cap €{cap:.2f} | maxDD {dd:.3f}%") - # 1) CHIUSI + # 1) CHIUSI — totale storico + epoca corrente (post gate TP_PHANTOM): i + # numeri pre-v1.1.26 includono il churn fantasma e non misurano la strategia + cur = [c for c in closed if c[5] >= EPOCH_V1126] + cpos = sum(1 for c in cur if c[4]) L.append(f"\n✅ CHIUSI: {pos} positivi / {neg} negativi (netto fee)") + L.append(f" epoca v1.1.26+ (TP_PHANTOM attivo): {cpos}/{len(cur) - cpos}") rows = [f"{'motivo':<12}{'✅':>3}{'❌':>4}{'PnL€':>9}"] for reason, (w, l, pnl) in sorted(by_reason.items(), key=lambda x: x[1][2]): rows.append(f"{reason:<12}{w:>3}{l:>4}{pnl:>+9.2f}") diff --git a/src/live/books.py b/src/live/books.py index 652f29d..c6aa9e5 100644 --- a/src/live/books.py +++ b/src/live/books.py @@ -53,6 +53,28 @@ def real_books(exclude_worker: str | None = None) -> tuple[dict[str, float], dic return books, orphans +def expected_resting() -> list[dict]: + """Ordini resting ATTESI sul book dai libri dei worker single-leg in posizione + reale: TP limit reduce-only (real_tp_order_id) e disaster-SL stop_market + (real_dsl_order_id). I pairs non hanno resting. Ogni voce: + {worker, instrument, order_id, kind: 'tp'|'dsl'}.""" + out: list[dict] = [] + for sp in sorted(PAPER.glob("*/status.json")): + wid = sp.parent.name + try: + st = json.loads(sp.read_text()) + except Exception: + continue + if not st.get("real_in_position") or st.get("real_amount_a"): + continue + inst = _inst(wid.split("__")[1]) + for key, kind in (("real_tp_order_id", "tp"), ("real_dsl_order_id", "dsl")): + oid = st.get(key) + if oid: + out.append(dict(worker=wid, instrument=inst, order_id=str(oid), kind=kind)) + return out + + def account_net(client) -> dict[str, float]: """Posizioni reali per strumento dal conto (size USD / mark -> coin, firmato).""" out: dict[str, float] = {} diff --git a/src/live/cerbero_client.py b/src/live/cerbero_client.py index 82cf18b..f8b2463 100644 --- a/src/live/cerbero_client.py +++ b/src/live/cerbero_client.py @@ -80,6 +80,16 @@ class CerberoClient: def get_positions(self, currency: str = "ETH") -> list[dict]: return self._post("/mcp-deribit/tools/get_positions", {"currency": currency}) + def get_open_orders(self, currency: str = "USDC", type: str = "all") -> list[dict]: + """Ordini APERTI sul conto (limit resting + trigger non scattati). Ogni voce: + {order_id, instrument, direction, order_type, order_state, amount, + filled_amount, price, trigger_price, reduce_only, label}. NB Deribit puo' + omettere i trigger untriggered da type='all' -> per i bracket interrogare + anche type='trigger_all' e fare merge per order_id.""" + out = self._post("/mcp-deribit/tools/get_open_orders", + {"currency": currency, "type": type}) + return out if isinstance(out, list) else out.get("orders", []) + # --- Trading --- def place_order( diff --git a/src/live/execution.py b/src/live/execution.py index 53f1261..068400a 100644 --- a/src/live/execution.py +++ b/src/live/execution.py @@ -183,6 +183,20 @@ class ExecutionClient: # (poll-loop fermo = posizione reale senza valutazione exit). None = disattivo. # Configurato dal runner da overrides.execution.disaster_sl_pct. disaster_sl_pct: float | None = None + # Circuit-breaker venue-lock (2026-06-12): durante il lock admin del testnet + # (rollback conto, ~09:47) ogni place_order rispondeva 'locked_by_admin' ma i + # worker continuavano a tentare APERTURE (leg-fail pairs + unwind + fee + # sprecate sui leg parziali). Dopo lock_trip errori 'locked' consecutivi le + # aperture sono SOSPESE (Fill failed senza chiamata API -> i worker seguono il + # path REAL_OPEN_FAIL/sim_fallback gia' esistente); le CHIUSURE si tentano + # SEMPRE (path gia' sicuro: partial/orphan/netting). Riarmo: passato + # lock_cooldown_s la prossima apertura fa da probe — se passa il breaker si + # resetta (alert di rientro), se e' ancora locked riscatta subito. Stato in + # memoria: al restart il primo open rifiutato lo ri-arma. + lock_trip: int = 3 + lock_cooldown_s: float = 900.0 + _lock_streak: int = field(default=0, init=False, repr=False) + _lock_until: float = field(default=0.0, init=False, repr=False) # NB leva: su Deribit la leva per-strumento NON e' impostabile (private/set_leverage # risponde 400 Bad Request — verificato 2026-06-03 nei log Cerbero; il set_leverage # di Cerbero fallisce sempre, soppresso). Il campo "leverage: 50" in get_positions @@ -214,6 +228,46 @@ class ExecutionClient: pass return None + # --- circuit-breaker venue-lock --- + + def _notify_safe(self, event: str, data: dict): + try: + from src.live.telegram_notifier import notify_event + notify_event(event, data) + except Exception: + pass + + def lock_blocked(self) -> bool: + """True se le APERTURE sono sospese (breaker scattato e cooldown attivo).""" + return self._lock_streak >= self.lock_trip and time.monotonic() < self._lock_until + + def _lock_track(self, error: str): + """Conta gli errori 'locked' consecutivi; al trip sospende le aperture. + Ogni nuovo 'locked' (anche dalle chiusure) rinfresca il cooldown: finche' + il venue resta bloccato le aperture non riprendono. Gli errori di altra + natura NON toccano lo streak (un transitorio di rete non deve ne' + armare ne' disarmare il breaker).""" + if "locked" not in (error or "").lower(): + return + self._lock_streak += 1 + self._lock_until = time.monotonic() + self.lock_cooldown_s + if self._lock_streak == self.lock_trip: + print(f"[exec] VENUE_LOCK: {self._lock_streak} reject 'locked' consecutivi " + f"-> aperture sospese {self.lock_cooldown_s / 60:.0f}m (probe al termine)") + self._notify_safe("VENUE_LOCK", { + "reject_consecutivi": self._lock_streak, + "cooldown_min": round(self.lock_cooldown_s / 60), + "nota": "conto locked (admin/rollback testnet): aperture reali sospese, " + "chiusure sempre tentate, sim prosegue"}) + + def _lock_reset(self): + """Ordine accettato dal venue: se il breaker era scattato, dichiara il rientro.""" + if self._lock_streak >= self.lock_trip: + self._notify_safe("VENUE_LOCK", {"status": "RIENTRATO", + "dopo_reject": self._lock_streak}) + self._lock_streak = 0 + self._lock_until = 0.0 + # --- API --- def _mark_price(self, instrument: str) -> float | None: @@ -246,9 +300,11 @@ class ExecutionClient: resp = self.client.place_order(instrument, side, amount, order_type=order_type, price=price, label=label, reduce_only=reduce_only) if not isinstance(resp, dict) or resp.get("state") == "error" or "error" in resp: + self._lock_track(str(resp.get("error", "")) if isinstance(resp, dict) else "") return Fill(instrument, side, requested_notional, amount, None, 0.0, 0.0, None, "error", False, raw=resp if isinstance(resp, dict) else {}, notes=f"place_order error: {resp}") + self._lock_reset() order = resp.get("order", resp) or {} trades = resp.get("trades", []) or [] @@ -299,7 +355,14 @@ class ExecutionClient: def open(self, instrument: str, side: str, notional_usd: float, label: str | None = None) -> Fill: - """Apre la quota del worker (market, NON reduce_only).""" + """Apre la quota del worker (market, NON reduce_only). Con breaker + venue-lock attivo NON tocca l'API: Fill failed -> il chiamante segue il + path REAL_OPEN_FAIL/sim_fallback (per i pairs: entrambe le gambe + rifiutate localmente, nessun leg parziale da unwindare).""" + if self.lock_blocked(): + return Fill(instrument, side, notional_usd, 0.0, None, 0.0, 0.0, + None, "error", False, + notes="venue_lock_breaker: aperture sospese (conto locked)") amount = self.amount_for(instrument, notional_usd) return self._submit(instrument, side, amount, notional_usd, reduce_only=False, label=label) diff --git a/src/live/strategy_worker.py b/src/live/strategy_worker.py index 350e6bc..3e72f2b 100644 --- a/src/live/strategy_worker.py +++ b/src/live/strategy_worker.py @@ -71,6 +71,12 @@ class StrategyWorker: self.real_dsl_order_id = "" # STOP_MARKET disaster bracket on-book (persistito) self.real_trades = 0 self.real_first_notified = False # alert Telegram "esecuzione viva" una tantum + # Quote residue dei close FALLITI/cappati (2026-06-12, parità coi pairs): + # prima il REAL_CLOSE_PARTIAL single-leg NON registrava l'orfano e il + # reconciler vedeva drift NON spiegato (caso MR07 0.102 ETH nel lock + # testnet). Stessa semantica di PairsWorker.orphan_legs: posizioni che il + # conto ha ancora ma i libri hanno chiuso; le legge books.real_books. + self.orphan_legs: list[dict] = [] self._tp_phantom_ts = 0 # dedup log TP_PHANTOM per barra (non persistito) self._tp_phantom_notified = False # alert Telegram una tantum per processo self._tp_phantom_cache = (0, 0.0) # (bar_ts, monotonic): TTL del verdetto phantom @@ -144,6 +150,7 @@ class StrategyWorker: self.real_dsl_order_id = state.get("real_dsl_order_id", "") self.real_trades = state.get("real_trades", 0) self.real_first_notified = state.get("real_first_notified", False) + self.orphan_legs = state.get("orphan_legs", []) self._log("RESUME", {"capital": round(self.capital, 2), "total_trades": self.total_trades, @@ -178,6 +185,7 @@ class StrategyWorker: "real_dsl_order_id": self.real_dsl_order_id, "real_trades": self.real_trades, "real_first_notified": self.real_first_notified, + "orphan_legs": self.orphan_legs, "last_update": datetime.now(timezone.utc).isoformat(), } with open(self.status_path, "w") as f: @@ -412,10 +420,18 @@ class StrategyWorker: self._log("NET_CLOSE", {"note": fill.notes}) self._notify("NET_CLOSE", {"note": fill.notes}) if fill and market_amt < remainder - step / 2: + residual = round(remainder - market_amt, 6) + # registra l'orfano (come PairsWorker): il conto ha ancora questa quota + # ma il libro chiude -> il reconciler la conta come drift SPIEGATO + self.orphan_legs.append({ + "instrument": self.exec_instrument, "entry_side": self.real_side, + "amount": residual, + "ts": datetime.now(timezone.utc).isoformat(), "reason": reason}) data = {"requested": remainder, "filled": market_amt, - "residuo_orfano": round(remainder - market_amt, 6), + "residuo_orfano": residual, "note": ("close non completato (netting negato/leg fallito): " - "quota residua NON chiusa — verificare col reconciler")} + "quota residua registrata in orphan_legs — " + "verificare col reconciler")} self._log("REAL_CLOSE_PARTIAL", data) self._notify("REAL_CLOSE_PARTIAL", data) @@ -574,6 +590,9 @@ class StrategyWorker: self.tp = 0.0 self.sl = 0.0 self.max_bars = 0 + # persisti il booking della chiusura SUBITO (non solo al save del tick): + # un crash qui perderebbe capital/orphan_legs gia' contabilizzati + self._save_state() def tick(self, df: pd.DataFrame, df_1h: pd.DataFrame | None = None): """Chiamato ad ogni poll con DataFrame OHLCV aggiornato. diff --git a/src/portfolio/runner.py b/src/portfolio/runner.py index ea3a963..c3dc8cf 100644 --- a/src/portfolio/runner.py +++ b/src/portfolio/runner.py @@ -224,6 +224,49 @@ def _check_stale_feed(asset: str, df: pd.DataFrame, alerted: set[str]): "gap_pct": round(gap, 2), "prezzo": float(c[i])}) +_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.""" @@ -380,6 +423,9 @@ def run(config_path: str = "portfolios.yml"): 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() # 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. @@ -432,6 +478,9 @@ def run(config_path: str = "portfolios.yml"): 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) + # 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(): diff --git a/tests/portfolio/test_feed_book_gap.py b/tests/portfolio/test_feed_book_gap.py new file mode 100644 index 0000000..4fa2bb8 --- /dev/null +++ b/tests/portfolio/test_feed_book_gap.py @@ -0,0 +1,73 @@ +"""Guard FEED_BOOK_GAP (2026-06-12): il feed candele e il book d'esecuzione possono +divergere (MR02_BTC: fill reale a -443 bps dal sim; wick TP_PHANTOM il caso opposto). +Alert per episodio con isteresi a soglia/2, fail-open su errori di rete.""" +import sys +from pathlib import Path + +import pandas as pd + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from src.portfolio import runner + + +class _Client: + def __init__(self, marks): + self._marks = marks + self.calls = 0 + + def get_ticker_batch(self, instruments): + self.calls += 1 + return {"tickers": [{"instrument_name": i, "mark_price": self._marks[i]} + for i in instruments if i in self._marks]} + + +class _Boom: + def get_ticker_batch(self, instruments): + raise RuntimeError("rete giu'") + + +INSTR = {"BTC": "BTC_USDC-PERPETUAL"} + + +def _feed(close): + return {"BTC": pd.DataFrame({"close": [100.0, close]})} + + +def _events(monkeypatch): + sent = [] + monkeypatch.setattr("src.live.telegram_notifier.notify_event", + lambda ev, data: sent.append((ev, data))) + return sent + + +def test_gap_alerts_once_per_episode(monkeypatch): + sent = _events(monkeypatch) + cl = _Client({"BTC_USDC-PERPETUAL": 60481.0}) + alerted = set() + # feed a 63285 vs mark 60481 = il caso MR02 (~464 bps) -> alert, una volta sola + for _ in range(3): + runner._check_feed_book_gap(cl, _feed(63285.5), INSTR, 150.0, alerted) + assert len(sent) == 1 and sent[0][0] == "FEED_BOOK_GAP" + assert sent[0][1]["asset"] == "BTC" and sent[0][1]["gap_bps"] > 400 + + +def test_recovery_with_hysteresis(monkeypatch): + sent = _events(monkeypatch) + cl = _Client({"BTC_USDC-PERPETUAL": 60000.0}) + alerted = {"BTC"} + # gap fra soglia/2 e soglia: NESSUN recovery (isteresi) + runner._check_feed_book_gap(cl, _feed(60000 * 1.0090), INSTR, 150.0, alerted) + assert sent == [] and "BTC" in alerted + # gap sotto soglia/2 -> RIENTRATO + runner._check_feed_book_gap(cl, _feed(60010.0), INSTR, 150.0, alerted) + assert len(sent) == 1 and sent[0][1].get("status") == "RIENTRATO" + assert "BTC" not in alerted + + +def test_fail_open_and_no_instruments(monkeypatch): + sent = _events(monkeypatch) + runner._check_feed_book_gap(_Boom(), _feed(63285.5), INSTR, 150.0, set()) + cl = _Client({"BTC_USDC-PERPETUAL": 60481.0}) + runner._check_feed_book_gap(cl, {}, INSTR, 150.0, set()) # nessun feed + assert sent == [] and cl.calls == 0 diff --git a/tests/portfolio/test_reconcile_resting.py b/tests/portfolio/test_reconcile_resting.py new file mode 100644 index 0000000..6573d7f --- /dev/null +++ b/tests/portfolio/test_reconcile_resting.py @@ -0,0 +1,96 @@ +"""Reconcile degli ordini RESTING (2026-06-12): TP/DSL attesi dai libri vs ordini +in book + fill non bookati (caso MR02_BTC: TP resting fillato di notte e disaster-SL +sparito, scoperti solo al close sim ore dopo). Nessuna rete: client stubbato, +PAPER puntato su tmp_path.""" +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from src.live import books + + +class _Client: + def __init__(self, open_orders=None, trades=None): + self._oo = open_orders or [] + self._tr = trades or [] + + def get_open_orders(self, currency="USDC", type="all"): + # merge 'all'+'trigger_all' nel chiamante: qui si ritorna tutto due volte + return self._oo + + def get_trade_history(self, limit=100, instrument_name=None): + return [t for t in self._tr + if not instrument_name or t["instrument"] == instrument_name] + + +def _mk_worker(root: Path, wid: str, st: dict): + d = root / wid + d.mkdir(parents=True) + (d / "status.json").write_text(json.dumps(st)) + + +def _setup(tmp_path, monkeypatch, statuses: dict): + root = tmp_path / "portfolio_paper" + for wid, st in statuses.items(): + _mk_worker(root, wid, st) + monkeypatch.setattr(books, "PAPER", root) + import importlib + ra = importlib.import_module("scripts.analysis.reconcile_account") + monkeypatch.setattr(ra, "PAPER", root) + return ra + + +IN_POS = {"real_in_position": True, "real_tp_order_id": "TP-1", + "real_dsl_order_id": "DSL-1"} + + +def test_expected_resting_reads_single_leg(tmp_path, monkeypatch): + _setup(tmp_path, monkeypatch, { + "MR01_bollinger_fade__ETH__1h": IN_POS, + # pairs e worker flat: NESSUN resting atteso + "PR01_pairs_reversion__ETH_BTC__1h": {"real_in_position": True, + "real_amount_a": 0.1}, + "MR02_donchian_fade__BTC__1h": {"real_in_position": False, + "real_tp_order_id": "TP-9"}, + }) + exp = books.expected_resting() + assert {(e["order_id"], e["kind"]) for e in exp} == {("TP-1", "tp"), ("DSL-1", "dsl")} + assert all(e["instrument"] == "ETH_USDC-PERPETUAL" for e in exp) + + +def test_resting_ok_when_on_book(tmp_path, monkeypatch): + ra = _setup(tmp_path, monkeypatch, {"MR01_bollinger_fade__ETH__1h": IN_POS}) + cl = _Client(open_orders=[{"order_id": "TP-1", "label": "MR01_bollinger_fade__ETH__1h"}, + {"order_id": "DSL-1", "label": "MR01_bollinger_fade__ETH__1h"}]) + rows = ra.compute_resting_drift(cl) + assert [r["status"] for r in rows] == ["OK", "OK"] + + +def test_resting_filled_unbooked_vs_missing(tmp_path, monkeypatch): + # TP sparito dal book MA con fill nel trade history -> FILLED_UNBOOKED (caso MR02); + # DSL sparito senza fill (trigger genera order_id nuovo) -> MISSING + ra = _setup(tmp_path, monkeypatch, {"MR01_bollinger_fade__ETH__1h": IN_POS}) + cl = _Client(open_orders=[], + trades=[{"instrument": "ETH_USDC-PERPETUAL", "order_id": "TP-1", + "amount": 0.103, "price": 1640.6}]) + by_kind = {r["kind"]: r for r in ra.compute_resting_drift(cl)} + assert by_kind["tp"]["status"] == "FILLED_UNBOOKED" + assert abs(by_kind["tp"]["filled"] - 0.103) < 1e-9 + assert by_kind["dsl"]["status"] == "MISSING" + + +def test_resting_stale_order_from_flat_worker(tmp_path, monkeypatch): + # ordine in book con label di un NOSTRO worker flat -> STALE (fillerebbe a sorpresa); + # ordini di altri bot (label sconosciuta) ignorati + ra = _setup(tmp_path, monkeypatch, { + "MR01_bollinger_fade__ETH__1h": {"real_in_position": False}}) + cl = _Client(open_orders=[ + {"order_id": "OLD-7", "label": "MR01_bollinger_fade__ETH__1h", + "instrument": "ETH_USDC-PERPETUAL", "order_type": "limit"}, + {"order_id": "X-1", "label": "altro_bot", "instrument": "ETH_USDC-PERPETUAL"}, + ]) + rows = ra.compute_resting_drift(cl) + assert len(rows) == 1 + assert rows[0]["status"] == "STALE" and rows[0]["order_id"] == "OLD-7" diff --git a/tests/portfolio/test_single_leg_orphan.py b/tests/portfolio/test_single_leg_orphan.py new file mode 100644 index 0000000..4d58be8 --- /dev/null +++ b/tests/portfolio/test_single_leg_orphan.py @@ -0,0 +1,79 @@ +"""Orfani single-leg (2026-06-12): un close fallito/cappato registra la quota +residua in orphan_legs (parita' coi pairs) cosi' il reconciler la conta come +drift SPIEGATO — prima il REAL_CLOSE_PARTIAL di MR07 (0.102 ETH nel lock testnet) +lasciava drift non spiegato. Nessuna rete: executor finto.""" +import json +from types import SimpleNamespace + +from src.live.execution import Fill +from src.live.strategy_worker import StrategyWorker +from src.strategies.base import Signal +from src.live import books + + +class FailingCloseExec: + """Apre ok, chiude con fill ZERO (venue locked / netting negato).""" + verify_polls = 1 + verify_sleep = 0.0 + disaster_sl_pct = None + + def open(self, instrument, side, notional, label=None): + amt = round(notional / 100.0, 6) + return Fill(instrument, side, notional, amt, 100.0, 0.0, 0.05, + "oid-open", "filled", True, filled_amount=amt) + + def place_tp_limit(self, *a, **k): + return Fill("x", "sell", 0.0, 0.0, None, 0.0, 0.0, None, None, False) + + def cancel_order(self, oid): + return {"state": "cancelled"} + + def resting_fills(self, instrument, oid): + return 0.0, None, 0.0 + + def close_amount(self, instrument, side, amount, label=None): + return Fill(instrument, "sell", 0.0, amount, None, 0.0, 0.0, + None, "error", False, notes="place_order error: locked_by_admin", + filled_amount=0.0) + + +def _worker(tmp_path): + return StrategyWorker( + strategy=SimpleNamespace(name="MR07_return_reversal", fee_rt=0.001), + asset="ETH", tf="1h", capital=100.0, position_size=0.5, leverage=2.0, + data_dir=tmp_path, executor=FailingCloseExec(), + exec_instrument="ETH_USDC-PERPETUAL") + + +def test_failed_close_registers_orphan(tmp_path, monkeypatch): + monkeypatch.setattr("src.live.strategy_worker.notify_event", lambda *a, **k: None) + w = _worker(tmp_path) + w._open_position(Signal(idx=0, direction=1, entry_price=100.0, + metadata={"max_bars": 12}), 100.0) + amt = w.real_amount + assert w.real_in_position and amt > 0 + w._close_position(105.0, "time_limit") + assert not w.real_in_position + assert len(w.orphan_legs) == 1 + o = w.orphan_legs[0] + assert o["instrument"] == "ETH_USDC-PERPETUAL" + assert o["entry_side"] == "buy" and abs(o["amount"] - amt) < 1e-9 + + # persistito nello status (resume-safe) e visto da books.real_books + st = json.loads((tmp_path / w.worker_id / "status.json").read_text()) + assert st["orphan_legs"] == w.orphan_legs + monkeypatch.setattr(books, "PAPER", tmp_path) + _, orphans = books.real_books() + assert abs(orphans["ETH_USDC-PERPETUAL"] - amt) < 1e-9 # buy = firmato + + + +def test_clean_close_no_orphan(tmp_path, monkeypatch): + monkeypatch.setattr("src.live.strategy_worker.notify_event", lambda *a, **k: None) + w = _worker(tmp_path) + w.executor.close_amount = lambda instrument, side, amount, label=None: Fill( + instrument, "sell", 0.0, amount, 105.0, 0.0, 0.05, "oid-close", + "filled", True, filled_amount=amount) + w._open_position(Signal(idx=0, direction=1, entry_price=100.0, + metadata={"max_bars": 12}), 100.0) + w._close_position(105.0, "time_limit") + assert w.orphan_legs == [] diff --git a/tests/portfolio/test_venue_lock_breaker.py b/tests/portfolio/test_venue_lock_breaker.py new file mode 100644 index 0000000..183c8c5 --- /dev/null +++ b/tests/portfolio/test_venue_lock_breaker.py @@ -0,0 +1,86 @@ +"""Circuit-breaker venue-lock (2026-06-12): dopo lock_trip reject 'locked' +consecutivi le APERTURE sono sospese senza toccare l'API (i worker seguono il +path REAL_OPEN_FAIL/sim_fallback); le chiusure si tentano sempre; il probe a +fine cooldown riarma o resetta. Nessuna rete: CerberoClient stubbato.""" +import time + +from src.live.execution import ExecutionClient + + +class _Venue: + """Stub CerberoClient: locked finche' .locked=True, poi ordini ok.""" + def __init__(self): + self.locked = True + self.place_calls = 0 + + def place_order(self, instrument, side, amount, order_type="market", + price=None, label=None, reduce_only=False): + self.place_calls += 1 + if self.locked: + return {"state": "error", "error": "locked_by_admin"} + return {"order": {"order_id": "o1", "order_state": "filled", + "filled_amount": amount, "average_price": 100.0}, + "trades": [{"amount": amount, "price": 100.0, "fee": 0.0}]} + + def get_ticker(self, instrument): + return {"mark_price": 100.0} + + def get_trade_history(self, limit=50, instrument_name=None): + return [] + + +def _ec(monkeypatch, sent): + monkeypatch.setattr("src.live.telegram_notifier.notify_event", + lambda ev, data: sent.append((ev, data))) + venue = _Venue() + ec = ExecutionClient(client=venue) + return ec, venue + + +INST = "ETH_USDC-PERPETUAL" + + +def test_breaker_trips_and_blocks_opens(monkeypatch): + sent = [] + ec, venue = _ec(monkeypatch, sent) + for _ in range(ec.lock_trip): + f = ec.open(INST, "buy", 50.0) + assert not f.verified + assert venue.place_calls == ec.lock_trip and ec.lock_blocked() + assert [e for e, _ in sent] == ["VENUE_LOCK"] # un alert al trip + # aperture sospese: NESSUNA chiamata API, nota dedicata + f = ec.open(INST, "buy", 50.0) + assert venue.place_calls == ec.lock_trip + assert "venue_lock_breaker" in f.notes + assert sent == sent[:1] # niente spam + + +def test_closes_still_attempted_and_refresh_cooldown(monkeypatch): + sent = [] + ec, venue = _ec(monkeypatch, sent) + for _ in range(ec.lock_trip): + ec.open(INST, "buy", 50.0) + ec._net_close_allowance = lambda *a, **k: 0.0 # niente fallback netting + before = ec._lock_until + time.sleep(0.01) + f = ec.close_amount(INST, "buy", 0.5, label="w1") # la chiusura PASSA dall'API + assert venue.place_calls == ec.lock_trip + 1 + assert not f.verified + assert ec._lock_until > before # locked dal close -> cooldown esteso + + +def test_probe_rearms_or_resets(monkeypatch): + sent = [] + ec, venue = _ec(monkeypatch, sent) + for _ in range(ec.lock_trip): + ec.open(INST, "buy", 50.0) + # cooldown scaduto + venue ancora locked -> probe fallisce e riscatta + ec._lock_until = time.monotonic() - 1 + ec.open(INST, "buy", 50.0) + assert venue.place_calls == ec.lock_trip + 1 and ec.lock_blocked() + # venue sbloccato -> il probe passa, breaker resettato, alert di rientro + ec._lock_until = time.monotonic() - 1 + venue.locked = False + f = ec.open(INST, "buy", 50.0) + assert f.verified and not ec.lock_blocked() and ec._lock_streak == 0 + assert sent[-1][0] == "VENUE_LOCK" and sent[-1][1].get("status") == "RIENTRATO"