diff --git a/backend_api_python/app/services/pending_order_worker.py b/backend_api_python/app/services/pending_order_worker.py index b1e4af8ac..4e3fc7134 100644 --- a/backend_api_python/app/services/pending_order_worker.py +++ b/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 diff --git a/backend_api_python/app/services/pending_orders/fill_records.py b/backend_api_python/app/services/pending_orders/fill_records.py index 25b9ce55e..a5103561f 100644 --- a/backend_api_python/app/services/pending_orders/fill_records.py +++ b/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, diff --git a/backend_api_python/app/services/pending_orders/sent_order_recovery.py b/backend_api_python/app/services/pending_orders/sent_order_recovery.py index d77030f4a..17170dd88 100644 --- a/backend_api_python/app/services/pending_orders/sent_order_recovery.py +++ b/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,17 +35,42 @@ 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 {} @@ -53,10 +78,18 @@ def tracked_fill_baseline( 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]) diff --git a/backend_api_python/tests/test_pending_order_worker_live_sync.py b/backend_api_python/tests/test_pending_order_worker_live_sync.py index 7c21f9a81..c45626a1f 100644 --- a/backend_api_python/tests/test_pending_order_worker_live_sync.py +++ b/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)