fix: prevent duplicate fill booking in live order reconciliation (stale live_fill_sync marker)

backend_api_python/app/services/pending_order_worker.py

@@ -50,7 +50,7 @@
 from app.services.live_trading.base import LiveTradingError, is_file_descriptor_exhausted
 from app.services.pending_orders.fill_records import (
     persist_strategy_fill, proportional_spot_position_fill_quantity,
-    trade_close_reason_from_payload,
+    persisted_order_fill_baseline, trade_close_reason_from_payload,
 )
 from app.services.pending_orders.fee_reconciliation import (
     backfill_zero_commission_trades,
@@ -923,11 +923,18 @@ def _sync_one_live_sent_order(self, row: Dict[str, Any]) -> None:
         exchange_status = str(exchange_status or "unknown").strip().lower()
         previous_filled = max(0.0, float(row.get("filled") or 0.0))
         previous_avg = max(0.0, float(row.get("avg_price") or 0.0))
+        persisted_filled, persisted_avg = persisted_order_fill_baseline(
+            exchange_order_id=exchange_order_id,
+            order_intent_id=int(payload.get("order_intent_id") or row.get("order_intent_id") or 0),
+            exchange_id=exchange_id,
+        )
         tracked_previous_filled, tracked_previous_avg = tracked_fill_baseline(
             row,
             exchange_order_id=exchange_order_id,
             previous_filled=previous_filled,
             previous_avg=previous_avg,
+            persisted_filled=persisted_filled,
+            persisted_avg=persisted_avg,
         )
         delta = cumulative_filled - tracked_previous_filled
         aggregate_filled = previous_filled

backend_api_python/app/services/pending_orders/fill_records.py

@@ -66,6 +66,61 @@ def proportional_spot_position_fill_quantity(
     )
 
 
+def persisted_order_fill_baseline(
+    *,
+    exchange_order_id: str,
+    order_intent_id: int = 0,
+    exchange_id: str = "",
+) -> Tuple[float, float]:
+    """Sum the fills already persisted in the ledger for one exchange order.
+
+    Returns ``(filled_qty, avg_price)`` with a notional-weighted avg (0.0
+    when nothing is recorded). Serves as the authoritative duplicate-fill
+    guard: the live fill sync must never re-book quantity the ledger
+    already holds for the same exchange order, even when the pending-order
+    row lost its fill baseline (restart, stale snapshot marker).
+    """
+    exchange_order_id = str(exchange_order_id or "").strip()
+    intent_id = int(order_intent_id or 0)
+    venue = str(exchange_id or "").strip().lower()
+    if not exchange_order_id and intent_id <= 0:
+        return 0.0, 0.0
+    try:
+        with get_db_connection() as db:
+            cur = db.cursor()
+            if exchange_order_id:
+                cur.execute(
+                    """
+                    SELECT COALESCE(SUM(quantity), 0) AS filled,
+                           COALESCE(SUM(notional), 0) AS notional
+                    FROM strategy_order_fills
+                    WHERE exchange_order_id = %s
+                      AND (%s = '' OR exchange_id = %s)
+                    """,
+                    (exchange_order_id, venue, venue),
+                )
+            else:
+                cur.execute(
+                    """
+                    SELECT COALESCE(SUM(quantity), 0) AS filled,
+                           COALESCE(SUM(notional), 0) AS notional
+                    FROM strategy_order_fills
+                    WHERE order_intent_id = %s
+                    """,
+                    (intent_id,),
+                )
+            baseline_row = cur.fetchone() or {}
+            cur.close()
+    except Exception:
+        logger.debug("persisted fill baseline unavailable", exc_info=True)
+        return 0.0, 0.0
+    filled = max(0.0, float(baseline_row.get("filled") or 0.0))
+    notional = max(0.0, float(baseline_row.get("notional") or 0.0))
+    if filled <= 0.0:
+        return 0.0, 0.0
+    return filled, (notional / filled if notional > 0.0 else 0.0)
+
+
 def persist_strategy_fill(
     *,
     strategy_id: int,

backend_api_python/app/services/pending_orders/sent_order_recovery.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import json
-from typing import Any, Dict, Tuple
+from typing import Any, Dict, List, Tuple
 
 
 def normalize_live_order_status(status: str) -> str:
@@ -35,28 +35,61 @@ def tracked_fill_baseline(
     exchange_order_id: str,
     previous_filled: float,
     previous_avg: float,
+    persisted_filled: float = 0.0,
+    persisted_avg: float = 0.0,
 ) -> Tuple[float, float]:
-    """Return the cumulative baseline for the currently tracked exchange leg."""
-    tracked_filled = max(0.0, float(previous_filled or 0.0))
-    tracked_avg = max(0.0, float(previous_avg or 0.0))
+    """Return the cumulative baseline for the currently tracked exchange leg.
+
+    Sources, cross-checked by taking the LARGEST fill count (its avg price
+    travels with it):
+
+    - executor ``market_summary`` when it names this exchange order — the
+      row then tracks legs separately (e.g. a maker/limit leg booked next
+      to this market leg), so the row aggregate must NOT join the race;
+    - otherwise the row aggregate (``filled``/``avg_price``), which for a
+      single-leg order IS this leg's fill;
+    - the sync's own ``live_fill_sync`` progress marker — it can go stale
+      (a marker written before the fill landed survives the executor's
+      update), so it never short-circuits the fresher sources;
+    - the persisted ledger sum (``persisted_filled``, from
+      strategy_order_fills) — authoritative whenever it is ahead of the
+      row, so reconciliation cannot re-book quantity the ledger already
+      holds (observed live as a doubled position after a restart).
+    """
+    ledger = (
+        max(0.0, float(persisted_filled or 0.0)),
+        max(0.0, float(persisted_avg or 0.0)),
+    )
+    candidates: List[Tuple[float, float]] = [ledger]
+    executor_baseline: Tuple[float, float] | None = None
     try:
         previous_response = json.loads(str(row.get("exchange_response_json") or "{}")) or {}
         sync_state = previous_response.get("live_fill_sync") or {}
         if isinstance(sync_state, dict) and "tracked_filled" in sync_state:
-            return (
-                max(0.0, float(sync_state.get("tracked_filled") or 0.0)),
-                max(0.0, float(sync_state.get("tracked_avg_price") or 0.0)),
+            candidates.append(
+                (
+                    max(0.0, float(sync_state.get("tracked_filled") or 0.0)),
+                    max(0.0, float(sync_state.get("tracked_avg_price") or 0.0)),
+                )
             )
         executor_raw = ((previous_response.get("phases") or {}).get("executor") or {})
         market_summary = executor_raw.get("market_summary") or {}
         if (
             isinstance(market_summary, dict)
             and str(market_summary.get("exchange_order_id") or "") == str(exchange_order_id or "")
         ):
-            return (
+            executor_baseline = (
                 max(0.0, float(market_summary.get("filled_qty") or 0.0)),
                 max(0.0, float(market_summary.get("avg_price") or 0.0)),
             )
+            candidates.append(executor_baseline)
     except Exception:
         pass
-    return tracked_filled, tracked_avg
+    if executor_baseline is None:
+        candidates.append(
+            (
+                max(0.0, float(previous_filled or 0.0)),
+                max(0.0, float(previous_avg or 0.0)),
+            )
+        )
+    return max(candidates, key=lambda item: item[0])

backend_api_python/tests/test_pending_order_worker_live_sync.py

@@ -2,13 +2,18 @@
 
 import ast
 import inspect
+import json
 import textwrap
 
 import pytest
 
 from app.services import pending_order_worker as worker_module
 from app.services.live_trading.adapters import LiveOrderPhaseAdapter
-from app.services.pending_orders.sent_order_recovery import is_final_fill, normalize_live_order_status
+from app.services.pending_orders.sent_order_recovery import (
+    is_final_fill,
+    normalize_live_order_status,
+    tracked_fill_baseline,
+)
 
 
 def test_live_order_adapter_call_uses_only_supported_constructor_keywords():
@@ -333,3 +338,70 @@ def get_order_status(self, order_id):
 )
 def test_live_order_status_normalization(raw, expected):
     assert normalize_live_order_status(raw) == expected
+
+
+def test_tracked_fill_baseline_takes_the_largest_known_source():
+    """A stale live_fill_sync marker (written before the fill landed) must
+    not blank the baseline: the executor fill in the row and the persisted
+    ledger both know about the fill, so the largest source wins."""
+    row = {
+        "exchange_response_json": json.dumps(
+            {"live_fill_sync": {"tracked_filled": 0.0, "tracked_avg_price": 0.0}}
+        )
+    }
+    filled, avg = tracked_fill_baseline(
+        row,
+        exchange_order_id="o1",
+        previous_filled=0.1,
+        previous_avg=685.48,
+        persisted_filled=0.1,
+        persisted_avg=685.48,
+    )
+    assert filled == pytest.approx(0.1)
+    assert avg == pytest.approx(685.48)
+
+
+def test_tracked_fill_baseline_ledger_beats_partial_stale_marker():
+    row = {
+        "exchange_response_json": json.dumps(
+            {"live_fill_sync": {"tracked_filled": 0.05, "tracked_avg_price": 100.0}}
+        )
+    }
+    filled, avg = tracked_fill_baseline(
+        row,
+        exchange_order_id="o1",
+        previous_filled=0.0,
+        previous_avg=0.0,
+        persisted_filled=0.1,
+        persisted_avg=101.0,
+    )
+    assert filled == pytest.approx(0.1)
+    assert avg == pytest.approx(101.0)
+
+
+def test_live_sent_sync_ignores_stale_zero_fill_sync_marker(monkeypatch):
+    """Regression for a live double-booked fill: the row carried a
+    live_fill_sync marker with tracked_filled=0 from a pre-fill sync, the
+    executor then recorded the real 0.1 fill, and the next reconciliation
+    re-booked the full 0.1 (position 0.1 -> 0.2 in the ledger while the
+    exchange held 0.1). The ledger baseline must clamp the delta to zero."""
+    row = _row(filled=0.1, avg_price=685.48)
+    row["exchange_response_json"] = json.dumps(
+        {"live_fill_sync": {"tracked_filled": 0.0, "tracked_avg_price": 0.0}}
+    )
+    worker, snapshots, persisted = _worker(
+        monkeypatch,
+        row,
+        exchange_fill=(0.1, 685.48, "filled"),
+    )
+    monkeypatch.setattr(
+        worker_module,
+        "persisted_order_fill_baseline",
+        lambda **kwargs: (0.1, 685.48),
+    )
+
+    worker._sync_one_live_sent_order(row)
+
+    assert persisted == []
+    assert snapshots[0]["status"] == "filled"
+    assert snapshots[0]["filled"] == pytest.approx(0.1)