feat(exec): netting delle chiusure market — il residuo reduce-only cappato/respinto va in market puro

close_amount ora: (1) tenta il market reduce-only (sicurezza storica: un bug di
stato filla 0 invece di aprire posizioni); (2) il residuo cappato/respinto dal
netting di conto (worker in direzioni opposte sullo stesso strumento) viene
rieseguito in MARKET PURO con label '|net' — muove il conto esattamente del
delta del libro = netta contro le quote opposte. Niente piu' gambe pairs orfane
ne' close cappati per costruzione (close_pair passa da close_amount).

- _merge_close_fills: il chiamante riceve UN Fill (prezzo medio pesato sui fill,
  fee sommate, filled_amount totale, verified se copre il richiesto, notes
  'netting' quando il fallback scatta)
- worker single-leg + pairs: evento NET_CLOSE (log jsonl + Telegram) a ogni
  fallback — osservabilita' della frequenza dei conflitti di netting
- sicurezza persa sul residuo coperta dal reconciler orario (ACCOUNT_DRIFT);
  orphan_legs/REAL_CLOSE_PARTIAL restano come ultima difesa
- 4 test nuovi (full r-o senza fallback, cappato, respinto, doppio fail) -> 110 passed

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/live/execution.py

@@ -127,6 +127,27 @@ class Fill:
     filled_amount: float = 0.0
 
 
+def _merge_close_fills(ro: Fill, net: Fill, requested: float) -> Fill:
+    """Combina il reduce-only e il residuo netting in UN esito per il chiamante:
+    filled = somma, prezzo = media pesata sui fill, fee sommate. verified se i
+    contratti riscontrati coprono il richiesto (la stessa soglia del chiamante)."""
+    fa = ro.filled_amount if ro.verified else 0.0
+    fb = net.filled_amount if net.verified else 0.0
+    tot = fa + fb
+    parts = [(amt, f.fill_price) for amt, f in ((fa, ro), (fb, net))
+             if amt > 0 and f.fill_price]
+    px = (sum(a * p for a, p in parts) / sum(a for a, _ in parts)) if parts else None
+    step = contract_spec(ro.instrument).get("step", 0.001)
+    verified = tot >= requested - step / 2
+    return Fill(ro.instrument, ro.side, ro.requested_notional, requested, px,
+                ro.fee_coin + net.fee_coin, ro.fee_usd + net.fee_usd,
+                net.order_id or ro.order_id, net.order_state or ro.order_state,
+                verified, raw={"reduce_only": ro.raw, "net": net.raw},
+                notes=(f"netting: reduce-only {fa}/{requested}, residuo {fb} "
+                       f"market puro ({net.order_state})"),
+                filled_amount=tot)
+
+
 def _avg_fill_price(order: dict, trades: list[dict]) -> float | None:
     p = order.get("average_price")
     if p:
@@ -274,16 +295,33 @@ class ExecutionClient:
 
     def close_amount(self, instrument: str, entry_side: str, amount: float,
                      label: str | None = None) -> Fill:
-        """Chiude SOLO la quota del worker: market reduce_only di lato opposto,
-        stesso `amount` dell'apertura. Non usa close_position (flatterebbe anche
-        le quote degli altri worker sullo stesso strumento)."""
+        """Chiude SOLO la quota del worker. Non usa close_position (flatterebbe
+        anche le quote degli altri worker sullo stesso strumento).
+
+        NETTING (v1.1.25, 2026-06-11): prima tenta il market reduce-only (la
+        sicurezza storica: un bug di stato filla 0 invece di aprire posizioni).
+        Su un conto a NETTING pero' il reduce-only viene CAPPATO o RESPINTO quando
+        un altro worker e' in direzione OPPOSTA sullo stesso strumento (audit
+        2026-06-11: 3 gambe pairs orfane + 1 close cappato) → il RESIDUO viene
+        rieseguito in MARKET PURO: muove il conto esattamente del delta del libro,
+        cioe' netta contro le quote opposte. La sicurezza persa sul residuo e'
+        coperta dal reconciler orario (reconcile_account.py, alert ACCOUNT_DRIFT).
+        Il chiamante riceve UN Fill combinato (prezzo medio pesato, fee sommate);
+        notes contiene 'netting' quando il fallback e' scattato."""
         spec = contract_spec(instrument)
         # quantizza difensivamente: il chiamante puo' passare un residuo
         # (amount aperto − fill parziale del TP) con artefatti float
         amount = _quantize_step(amount, spec["step"], spec["min"]) if amount > 0 else 0.0
         opp = "sell" if entry_side == "buy" else "buy"
-        return self._submit(instrument, opp, amount, 0.0,
+        fill = self._submit(instrument, opp, amount, 0.0,
                             reduce_only=True, label=label)
+        residual = amount - fill.filled_amount
+        residual = _quantize_step(residual, spec["step"], spec["min"]) if residual > 0 else 0.0
+        if residual < spec["step"] / 2:
+            return fill
+        net = self._submit(instrument, opp, residual, 0.0, reduce_only=False,
+                           label=f"{label}|net" if label else "net")
+        return _merge_close_fills(fill, net, amount)
 
     def place_tp_limit(self, instrument: str, entry_side: str, amount: float,
                        tp_price: float, label: str | None = None) -> Fill: