feat(live): executor a 2 gambe per i pairs (PairsExecutionClient, shadow) — pronto, spento

PairsExecutionClient (compone ExecutionClient): open_pair (2 market long A/short B,
verifica per-gamba, LEG-RISK unwind se una sola filla), close_pair (2 reduce-only via
close_amount, MAI close_position), register_contract (fetch spec USDC). Spec LTC/ADA/SOL
aggiunti. PairsWorker: ledger reale shadow a 2 gambe resume-safe (_real_open_pair/
_real_close_pair, PnL per gamba dir A=+d/B=-d, doppio arrotondamento riportato). Runner:
pairs_executor gated su execution.pairs_enabled (false di default).

Validazione: test 92/92 (open/close, leg-risk unwind, resume) + smoke testnet
end-to-end (open 2 gambe verificate, close reduce-only, PnL reale -0.039 vs sim -0.036,
conto flat). Smoke ora aborta se ci sono posizioni di produzione + usa solo close_amount.

NB incidente testnet documentato (diario): pulizia manuale con close_position ha flattato
le quote shadow dei fade sul conto condiviso -> auto-riconciliazione al prossimo close.
Lezione: mai close_position su strumenti condivisi.

pairs_enabled resta FALSE: accendere con finestra a conto flat + osservazione.

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

src/portfolio/runner.py

@@ -49,16 +49,19 @@ def pos_for_spec(sid: str, global_ps: float, family_overrides: dict[str, float])
 
 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):
+                     executor=None, exec_instrument: str | None = None,
+                     pairs_executor=None, exec_instruments: dict | None = None):
     """Costruisce il worker esecutore per uno sleeve con capitale = quota allocata.
 
-    executor/exec_instrument: se valorizzati (solo per i fade single-leg abilitati),
-    lo StrategyWorker affianca al fill simulato un ordine REALE su Deribit (shadow)."""
+    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)."""
     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,
         )
     if spec.kind == "basket":
         pr = spec.params
@@ -245,7 +248,9 @@ def run(config_path: str = "portfolios.yml"):
     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
     executor = None
+    pairs_executor = None
     if exec_enabled:
         from src.live.execution import ExecutionClient
         executor = ExecutionClient(client=client)
@@ -253,6 +258,10 @@ def run(config_path: str = "portfolios.yml"):
         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}")
+        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, solo se fade single-leg abilitato."""
@@ -260,6 +269,12 @@ def run(config_path: str = "portfolios.yml"):
             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)
@@ -267,9 +282,11 @@ def run(config_path: str = "portfolios.yml"):
     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),
-                                          executor=ex, exec_instrument=inst)
+                                          executor=ex, exec_instrument=inst,
+                                          pairs_executor=pex, exec_instruments=pinst)
     if ps_family:
         print(f"[runner] position_size globale={position_size} override famiglia={ps_family}")