PINK DITAv2 Sprint 2-3: accounting parity + multi-leg groundwork

Sprint 2 (accounting + observability parity, PINK scope):
- Verified pink_clickhouse.py writes the 8 BLUE-legacy row families at
  matching schema and that capital authority in pink_direct.step() is
  solely kernel.account (no balance-poll overwrite in the hot loop).
- Report: prod/clean_arch/dita_v2/SPRINT2_ACCOUNTING_PARITY.md.

Sprint 3 offline groundwork (no exchange contact):
- Add _write_trade_exit_leg to pink_clickhouse.py: one BLUE-schema-faithful
  trade_exit_legs row per exit leg, with isolated (non-cumulative) per-leg
  deltas tracked via _leg_state (reset on ENTER). Closes the docstring gap.
- New offline suite test_pink_multi_exit_groundwork.py (3 passed):
  * Flaw 4 — two-leg exit closes once, realized accrues per leg, closed
    slot rejects further EXIT (no double-close).
  * Overshoot invariant — a final EXIT requesting more than the remaining
    size CLAMPS (size to 0, no oversell), retiring the Sprint 0 cumulative-
    ratio risk empirically.
  * trade_exit_legs delta + full BLUE column-set assertions.
- Persistence regression after edits: 10 passed.

BLUE untouched: no changes to dolphin.* / DOLPHIN_*_BLUE / nautilus_event_trader.py.
Live VST multi-leg run remains deferred pending explicit authorization.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

prod/clean_arch/persistence/pink_clickhouse.py

@@ -177,6 +177,11 @@ class PinkClickHousePersistence:
         )
         self._sink = sink or self._resolve_sink("pink")
         self._v7_sink = v7_sink or self._resolve_v7_sink("pink")
+        # Per-trade incremental leg state for trade_exit_legs row deltas.
+        # Keyed by trade_id; reset on ENTER. Tracks the cumulative realized PnL
+        # and remaining size observed at the previous leg so each leg row carries
+        # an isolated (non-cumulative) pnl_leg / exit_qty.
+        self._leg_state: dict[str, dict[str, Any]] = {}
 
     @staticmethod
     def _resolve_sink(strategy: str) -> Writer:
@@ -245,6 +250,15 @@ class PinkClickHousePersistence:
             return
 
         if decision.action == DecisionAction.ENTER:
+            # Reset per-trade leg deltas: a fresh position starts with zero
+            # realized PnL and the full initial size remaining.
+            self._leg_state[intent.trade_id] = {
+                "prev_realized": 0.0,
+                "prev_size": _safe_float(
+                    slot.get("initial_size", slot.get("size", 0.0)), 0.0
+                ) or _safe_float(intent.target_size, 0.0),
+                "prev_leg_id": "",
+            }
             self._write_trade_reconstruction(
                 snapshot, intent.trade_id,
                 event_type="ENTRY_FILLED",
@@ -264,6 +278,9 @@ class PinkClickHousePersistence:
             return
 
         partial = slot.get("closed", False) is False and slot.get("size", 0) > 0
+        # One trade_exit_legs row per exit leg (partial or final), BLUE-schema
+        # compatible so PINK multi-exit trades reconcile against the same table.
+        self._write_trade_exit_leg(snapshot, decision, intent, slot, outcome)
         self._write_trade_reconstruction(
             snapshot, intent.trade_id,
             event_type="PARTIAL_EXIT" if partial else "EXIT",
@@ -567,6 +584,91 @@ class PinkClickHousePersistence:
         }
         self._sink("status_snapshots", row)
 
+    def _write_trade_exit_leg(
+        self, snapshot: Any, decision: Decision, intent: Intent,
+        slot_dict: dict[str, Any], outcome: KernelOutcome | None,
+    ) -> None:
+        """Emit one BLUE-schema-compatible ``trade_exit_legs`` row per exit leg.
+
+        The DITAv2 kernel uses a single slot with sequential exit legs rather
+        than BLUE's chained per-leg trade_ids, so the chain_* columns describe
+        the leg sequence within this one trade (root = trade_id). Per-leg deltas
+        (exit_qty, pnl_leg) are computed against the previous leg's snapshot held
+        in ``self._leg_state`` so each row is isolated, not cumulative.
+        """
+        trade_id = intent.trade_id
+        prev = self._leg_state.get(trade_id) or {
+            "prev_realized": 0.0,
+            "prev_size": _safe_float(slot_dict.get("initial_size", 0.0), 0.0),
+            "prev_leg_id": "",
+        }
+        entry_price = self._slot_entry_price(slot_dict) or _safe_float(intent.reference_price, 0.0)
+        exit_price = _safe_float(intent.reference_price, 0.0) or _safe_float(decision.reference_price, 0.0)
+        side = self._slot_side(slot_dict)
+        if side == TradeSide.FLAT:
+            side = intent.side
+        leverage_val = _safe_float(slot_dict.get("leverage", intent.leverage), 1.0)
+
+        cur_size = self._slot_size(slot_dict)
+        cur_realized = _safe_float(slot_dict.get("realized_pnl", 0.0), 0.0)
+        prev_size = _safe_float(prev.get("prev_size", 0.0), 0.0)
+        prev_realized = _safe_float(prev.get("prev_realized", 0.0), 0.0)
+
+        # active_leg_index is post-fill (already advanced); the leg that just
+        # filled is therefore one behind. Clamp to a valid ratio index.
+        ratios = slot_dict.get("exit_leg_ratios", []) or []
+        leg_index = max(0, int(slot_dict.get("active_leg_index", 0) or 0) - 1)
+        fraction = _safe_float(ratios[leg_index], 0.0) if 0 <= leg_index < len(ratios) else 0.0
+
+        exit_qty = max(0.0, prev_size - cur_size)
+        pnl_leg = cur_realized - prev_realized
+        capital_after = self._capital()
+        capital_before = capital_after - pnl_leg
+        exit_notional = _notional(exit_qty, exit_price or entry_price)
+        remaining_notional = _notional(cur_size, entry_price)
+        denom = abs(exit_qty * entry_price * max(leverage_val, 1e-9))
+        pnl_pct_leg = pnl_leg / denom if denom > 0 else 0.0
+        exit_leg_id = f"{trade_id}:leg{leg_index}"
+
+        self._sink("trade_exit_legs", {
+            "ts": snapshot.timestamp.isoformat(),
+            "date": snapshot.timestamp.date().isoformat(),
+            "strategy": self.config.strategy,
+            "trade_id": trade_id,
+            "chain_root_trade_id": trade_id,
+            "chain_head_leg_id": f"{trade_id}:leg0",
+            "chain_prev_leg_id": str(prev.get("prev_leg_id", "") or ""),
+            "chain_seq": leg_index,
+            "chain_token": trade_id,
+            "chain_mode": "LIVE",
+            "exit_leg_id": exit_leg_id,
+            "exit_seq": leg_index,
+            "command_id": decision.decision_id,
+            "source": "ditav2",
+            "reason": intent.reason,
+            "asset": intent.asset,
+            "side": side.value,
+            "entry_price": entry_price,
+            "exit_price": exit_price,
+            "fraction": fraction,
+            "capital_before": capital_before,
+            "capital_after": capital_after,
+            "exit_notional": exit_notional,
+            "remaining_notional": remaining_notional,
+            "remaining_qty": cur_size,
+            "pnl_pct_leg": pnl_pct_leg,
+            "pnl_leg": pnl_leg,
+            "pnl_realized_total": cur_realized,
+            "bars_held": int(intent.bars_held or 0),
+        })
+
+        # Advance the per-trade leg snapshot for the next leg's delta.
+        self._leg_state[trade_id] = {
+            "prev_realized": cur_realized,
+            "prev_size": cur_size,
+            "prev_leg_id": exit_leg_id,
+        }
+
     def _write_trade_event(
         self, snapshot: Any, decision: Decision, intent: Intent,
         slot_dict: dict[str, Any], outcome: KernelOutcome | None,