fix: preserve recovered live order fills

Author: Henry

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:
@@ -36,27 +36,47 @@ def tracked_fill_baseline(
     previous_filled: float,
     previous_avg: float,
 ) -> 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 freshest cumulative baseline for the tracked exchange leg.
+
+    ``live_fill_sync`` is a progress snapshot, not an authoritative source: a
+    sync can write zero before the executor records the fill in the row.  Keep
+    it as a candidate instead of allowing it to hide newer row or executor
+    state after a worker restart.
+
+    When the executor summary names this exchange order, the pending-order row
+    may aggregate multiple legs.  In that case only the executor summary and
+    sync marker compete; otherwise the row aggregate is a valid candidate.
+    """
+    candidates: List[Tuple[float, float]] = []
+    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():
@@ -131,6 +136,42 @@ def test_live_sent_sync_finalizes_after_restart_without_duplicate_fill(monkeypat
     assert snapshots[0]["exchange_status"] == "filled"
 
 
+def test_stale_zero_sync_marker_cannot_hide_executor_fill_in_row():
+    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="exchange-41",
+        previous_filled=0.1,
+        previous_avg=685.48,
+    )
+
+    assert filled == pytest.approx(0.1)
+    assert avg == pytest.approx(685.48)
+
+
+def test_live_sent_sync_does_not_rebook_fill_hidden_by_stale_marker(monkeypatch):
+    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"),
+    )
+
+    worker._sync_one_live_sent_order(row)
+
+    assert persisted == []
+    assert snapshots[0]["status"] == "filled"
+    assert snapshots[0]["filled"] == pytest.approx(0.1)
+
+
 def test_bitget_precision_normalized_fill_releases_residual_sent_order(monkeypatch):
     class Client:
         def normalize_base_order_size(self, *, symbol, product_type, base_size):