fix(live): 4 hardening da code-review — REAL_DIVERGENCE, outage su feed vuoto, bars.py condiviso, DSL cancel retry

Review multi-agente 7-angoli su 8c4e1cd..HEAD + check trades live:

1. Alert Telegram REAL_DIVERGENCE quando |slippage sim/reale| >= 100bps a
   open/close. Causa scatenante: spike print testnet 10:37 (candela 10:00
   H=65618, O/C ~62400) -> 3 fade BTC short su close fantasma 65266.5, reale
   fillato a 62395 (-440bps), sim +2.26 mai esistiti — passato in silenzio.
2. FEED_OUTAGE anche su feed degradato SENZA eccezione (HTTP 200 + candles
   vuote: i worker saltavano il tick in silenzio, streak a 0). Helper unico
   _outage_tick; chiavi payload uniformate (minuti su start e RIPRESO).
3. src/live/bars.py: detection forming-bar unificata (bar_ms_of /
   last_bar_is_forming / last_settled_idx) — era copiata in 4 punti
   (strategy_worker, basket, pairs, _check_stale_feed hardcoded 1h).
   E' l'invariante di sicurezza EXIT-16: ora una sola implementazione testata.
4. DSL cancel hardening in _real_close: retry su errore transitorio + alert
   REAL_DSL_CANCEL_FAIL se lo stop resta forse orfano sul book (prima l'id
   veniva dimenticato in silenzio); order_not_found = probabile trigger in
   outage -> solo log (il close a valle esce gia' verified=False).

Refutato il finding top dei finder ("stop_market senza trigger"): cerbero-mcp
traduce price->trigger_price+mark, e in produzione 2 DSL armati + 1 ciclo
completo pulito (MR07_BTC).

Test: 83/83 (9 nuovi: bars helper + DSL/divergence con executor finto).
Smoke testnet 4 scenari verdi, conto flat, zero falsi allarmi.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/live/pairs_worker.py

@@ -16,7 +16,6 @@ Stato persistente (resume al restart) e log come StrategyWorker.
 from __future__ import annotations
 
 import json
-import time
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -179,13 +178,11 @@ class PairsWorker:
         m = df_a[["timestamp", "close"]].rename(columns={"close": "ca"}).merge(
             df_b[["timestamp", "close"]].rename(columns={"close": "cb"}), on="timestamp", how="inner"
         ).sort_values("timestamp").reset_index(drop=True)
-        # Scarta la barra IN FORMAZIONE (la riga -1 e' la candela in corso finche'
-        # non e' trascorsa la sua durata): entry ED exit valutati SOLO sul close di
+        # Scarta la barra IN FORMAZIONE: entry ED exit valutati SOLO sul close di
         # barra COMPLETA, come il backtest (pairs_research: close settled) —
-        # lezione EXIT-16 (strategy_worker.tick).
-        ts_arr = m["timestamp"].values.astype("int64")
-        bar_ms = int(np.median(np.diff(ts_arr[-50:]))) if len(ts_arr) > 1 else 0
-        if bar_ms and int(time.time() * 1000) < int(ts_arr[-1]) + bar_ms:
+        # lezione EXIT-16. Detection condivisa: src.live.bars.
+        from src.live.bars import last_bar_is_forming
+        if last_bar_is_forming(m["timestamp"].values):
             m = m.iloc[:-1]
         if len(m) < self.n + 2:
             return