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 <noreply@anthropic.com>

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] = {}

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(

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)

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.

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():