feat: support exchange-routed equity trading

2026-09-29 17:07:04 +08:00 · 2026-09-14 12:49:49 +08:00
parent ff195f437b
commit d8508a85a4
82 changed files with 5619 additions and 370 deletions
@@ -0,0 +1,162 @@
+"""Opt-in PostgreSQL tests using an isolated schema and real fill writers.
+
+Set QD_TEST_POSTGRES_DSN to a local PostgreSQL with the application schema.
+Only table definitions are copied; account data and brokers are never used.
+"""
+import os
+from concurrent.futures import ThreadPoolExecutor
+from uuid import uuid4
+
+import pytest
+
+from app.services.execution_streams import processor as module
+from app.services.live_trading import records
+from app.services.live_trading.leg_context import LegContext
+from app.services.pending_orders import fill_records
+from app.utils import db_postgres as pg
+
+
+@pytest.fixture
+def projection(monkeypatch):
+    dsn = os.environ.get("QD_TEST_POSTGRES_DSN")
+    if not dsn:
+        pytest.skip("QD_TEST_POSTGRES_DSN is required")
+    import psycopg2
+    from psycopg2 import sql
+    from psycopg2.pool import ThreadedConnectionPool
+
+    schema = "qd_projection_test_" + uuid4().hex
+    admin = psycopg2.connect(dsn)
+    admin.autocommit = True
+    tables = ("pending_orders", "qd_live_order_bindings", "qd_strategy_positions",
+              "qd_strategy_trades", "strategy_order_fills", "strategy_order_intents")
+    pool = None
+    try:
+        with admin.cursor() as cur:
+            cur.execute(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(schema)))
+            for table in tables:
+                cur.execute(sql.SQL("CREATE TABLE {}.{} (LIKE public.{} INCLUDING DEFAULTS INCLUDING INDEXES)").format(
+                    sql.Identifier(schema), sql.Identifier(table), sql.Identifier(table)))
+                # Do not consume production serial sequences through copied defaults.
+                cur.execute(sql.SQL("CREATE SEQUENCE {}.{}").format(sql.Identifier(schema), sql.Identifier(table + "_test_id")))
+                cur.execute(sql.SQL("ALTER TABLE {}.{} ALTER COLUMN id SET DEFAULT nextval(%s)").format(
+                    sql.Identifier(schema), sql.Identifier(table)), (schema + "." + table + "_test_id",))
+        pool = ThreadedConnectionPool(1, 4, dsn, options=f"-c search_path={schema}")
+        monkeypatch.setattr(pg, "_get_connection_pool", lambda: pool)
+        monkeypatch.setattr(pg, "_acquire_conn_with_wait", lambda p: p.getconn())
+
+        def query(statement, params=()):
+            with pg.get_pg_connection() as db:
+                cur = db.cursor()
+                cur.execute(statement, params)
+                rows = cur.fetchall() if cur._cursor.description else []
+                db.commit()
+                cur.close()
+                return rows
+
+        query("""INSERT INTO pending_orders
+            (id,user_id,strategy_id,symbol,signal_type,market_type,execution_mode,status,amount,price,idempotency_key)
+            VALUES (1,1,1,'BTC/USDT','open_long','swap','live','sent',2,100,'projection-test')""")
+        query("""INSERT INTO qd_live_order_bindings
+            (id,credential_id,exchange_id,market_type,owner_type,owner_id,strategy_id,pending_order_id)
+            VALUES (1,1,'binance','swap','pending_order',1,1,1)""")
+        monkeypatch.setattr(module, "load_strategy_configs", lambda *a: {"user_id": 1})
+        monkeypatch.setattr(module, "resolve_exchange_config", lambda *a, **kw: {"exchange_id": "binance", "credential_id": 1})
+        monkeypatch.setattr(module, "bind_instrument_product_contract", lambda cfg, *a, **kw: cfg)
+        monkeypatch.setattr(module, "create_client", lambda *a, **kw: object())
+        monkeypatch.setattr(module, "append_strategy_log", lambda *a, **kw: None)
+        monkeypatch.setattr(records, "_get_user_id_from_strategy", lambda *a: 1)
+        monkeypatch.setattr(fill_records, "resolve_leg_context", lambda **kw: LegContext(
+            credential_id=1, pending_order_id=1, fill_source="private_websocket"))
+        monkeypatch.setattr(fill_records, "invalidate_position_sync_snapshot_for_exchange", lambda **kw: None)
+        processor = module.ExecutionEventProcessor()
+        monkeypatch.setattr(processor, "_fees", lambda *a, **kw: ({"USDT": 0.1}, 0.1))
+        event = {"id": 11, "quantity": 1, "cumulative_quantity": 1, "is_cumulative": True,
+                 "price": 100, "order_status": "partial", "exchange_id": "binance",
+                 "exchange_order_id": "order-1", "exchange_fill_id": "fill-1", "fee_status": "actual"}
+        binding = {"id": 1, "pending_order_id": 1, "strategy_id": 1, "strategy_run_id": 1}
+        yield processor, event, binding, query
+    finally:
+        if pool is not None:
+            pool.closeall()
+        with admin.cursor() as cur:
+            cur.execute(sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format(sql.Identifier(schema)))
+        admin.close()
+
+
+def assert_ledger(query, amount):
+    assert float(query("SELECT filled FROM pending_orders WHERE id=1")[0]["filled"]) == amount
+    assert float(query("SELECT observed_filled FROM qd_live_order_bindings WHERE id=1")[0]["observed_filled"]) == amount
+    positions = query("SELECT size FROM qd_strategy_positions")
+    assert sum(float(r["size"]) for r in positions) == amount
+    trades = query("SELECT id, amount, commission_quote FROM qd_strategy_trades")
+    runtime = query("SELECT id, quantity, commission_quote FROM strategy_order_fills")
+    assert len(trades) == (1 if amount else 0)
+    assert len(runtime) == (1 if amount else 0)
+    if amount:
+        assert float(trades[0]["amount"]) == amount
+        assert float(runtime[0]["quantity"]) == amount
+        assert float(trades[0]["commission_quote"]) == pytest.approx(0.1)
+        assert float(runtime[0]["commission_quote"]) == pytest.approx(0.1)
+
+
+@pytest.mark.parametrize("failure_stage", ["config", "after_position", "after_trade", "runtime_sql"])
+def test_failed_projection_rolls_back_all_writes_and_replays(projection, monkeypatch, failure_stage):
+    processor, event, binding, query = projection
+
+    def fail(*a, **kw):
+        raise RuntimeError("injected projection failure")
+
+    with monkeypatch.context() as patch:
+        if failure_stage == "config":
+            patch.setattr(module, "load_strategy_configs", fail)
+        elif failure_stage == "after_position":
+            patch.setattr(fill_records, "record_trade", fail)
+        elif failure_stage == "after_trade":
+            patch.setattr(fill_records, "_record_runtime_fill", fail)
+        else:
+            original = pg.PostgresCursor.execute
+
+            def execute(cur, statement, params=None):
+                if "INSERT INTO strategy_order_fills" in statement:
+                    return original(cur, "SELECT 1 / 0")
+                return original(cur, statement, params)
+
+            patch.setattr(pg.PostgresCursor, "execute", execute)
+        with pytest.raises(Exception):
+            processor._process_pending_order(event, binding)
+    assert_ledger(query, 0)
+    processor._process_pending_order(event, binding)
+    assert_ledger(query, 1)
+    # Also covers a crash after commit but before marking the event processed.
+    processor._process_pending_order(event, binding)
+    assert_ledger(query, 1)
+
+
+@pytest.mark.parametrize("cumulative", [True, False])
+def test_concurrent_duplicate_event_is_applied_once(projection, cumulative):
+    processor, event, binding, query = projection
+    if not cumulative:
+        event = {**event, "is_cumulative": False, "cumulative_quantity": 0}
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        list(executor.map(lambda _: processor._process_pending_order(event, binding), range(2)))
+    assert_ledger(query, 1)
+
+
+def test_subsequent_partial_fill_keeps_quantity_and_fees_consistent(projection):
+    processor, event, binding, query = projection
+    processor._process_pending_order(event, binding)
+    second = {**event, "id": 12, "cumulative_quantity": 2, "price": 110,
+              "exchange_fill_id": "fill-2", "order_status": "filled"}
+    processor._process_pending_order(second, binding)
+    processor._process_pending_order(second, binding)
+    order = query("SELECT filled, avg_price, status FROM pending_orders WHERE id=1")[0]
+    assert float(order["filled"]) == 2
+    assert float(order["avg_price"]) == 105
+    assert order["status"] == "filled"
+    position = query("SELECT size, entry_price FROM qd_strategy_positions")[0]
+    assert float(position["size"]) == 2
+    assert float(position["entry_price"]) == 105
+    totals = query("SELECT SUM(amount) AS quantity, SUM(commission_quote) AS fee FROM qd_strategy_trades")[0]
+    assert float(totals["quantity"]) == 2
+    assert float(totals["fee"]) == pytest.approx(0.2)