siloqy/prod/clean_arch/dita_v2/test_bingx_bugs.py

"""Regression tests for BingX adapter bugs found during 2026-06-05 review.

Covers:
  - cancel_async: no thread-pool deadlock from async context
  - process_intent_async CANCEL: uses cancel_async (not sync cancel)
  - S2 task guard: no duplicate background refresh tasks per symbol
  - _events_from_submit with None snapshots: correct fill emit
  - cancel() with None snapshots: correct events
  - _filled_size_from_snapshots(None, None): returns 0.0 safely
  - _events_from_cancel ignores before/after completely
  - submit() sync with None snapshots: no snapshot fallback regression

Run:
    python -m pytest test_bingx_bugs.py -v
"""
from __future__ import annotations

import asyncio
import inspect
import itertools
import sys
import threading
from datetime import datetime, timezone
from typing import Any, List
from unittest.mock import AsyncMock, MagicMock, patch

sys.path.insert(0, "/mnt/dolphinng5_predict")

import pytest

from prod.clean_arch.dita_v2.bingx_venue import BingxVenueAdapter
from prod.clean_arch.dita_v2.contracts import (
    KernelCommandType,
    KernelEventKind,
    KernelIntent,
    TradeSide,
    TradeStage,
    VenueEvent,
    VenueEventStatus,
    VenueOrder,
    VenueOrderStatus,
)
from prod.clean_arch.dita_v2.mock_venue import MockVenueAdapter, MockVenueScenario
from prod.clean_arch.dita_v2.rust_backend import ExecutionKernel


# ── helpers ───────────────────────────────────────────────────────────────────

def _make_venue(*, cancel_response=None, submit_response=None) -> BingxVenueAdapter:
    backend = MagicMock()
    backend.cancel = AsyncMock(return_value=cancel_response or {"status": "CANCELED", "orderId": "O1", "clientOrderId": "C1"})
    backend.submit_intent = AsyncMock(return_value=_mock_receipt("FILLED"))
    backend.refresh_state = AsyncMock(return_value=_flat_snap())
    backend.connect = AsyncMock(return_value=True)
    venue = BingxVenueAdapter.__new__(BingxVenueAdapter)
    venue.backend = backend
    venue._event_seq = itertools.count(1)
    venue._snap_lock = threading.Lock()
    venue._snapshot_ready = threading.Event()
    venue._snapshot_ready.set()
    venue._last_snapshot = None
    return venue


def _flat_snap(**kw):
    s = MagicMock()
    s.open_orders = kw.get("open_orders", [])
    s.all_orders = kw.get("all_orders", [])
    s.all_fills = kw.get("all_fills", [])
    s.open_positions = kw.get("open_positions", {})
    s.capital = kw.get("capital", 25000.0)
    return s


def _mock_receipt(status: str = "FILLED", executed_qty: float = 1.0, order_id: str = "O1") -> MagicMock:
    r = MagicMock()
    r.status = status
    r.order_id = order_id
    r.client_order_id = "C1"
    r.price = 100.0
    r.quantity = 1.0
    r.timestamp = datetime.now(timezone.utc)
    r.raw_ack = {
        "status": status,
        "orderId": order_id,
        "clientOrderId": "C1",
        "executedQty": str(executed_qty),
        "avgPrice": "100.0",
    }
    return r


def _make_order(slot_id: int = 0, asset: str = "TRX-USDT") -> VenueOrder:
    return VenueOrder(
        internal_trade_id="t1",
        venue_order_id="O1",
        venue_client_id="C1",
        side=TradeSide.SHORT,
        intended_size=10.0,
        filled_size=0.0,
        average_fill_price=0.0,
        status=VenueOrderStatus.NEW,
        metadata={"slot_id": slot_id, "asset": asset},
    )


def _make_intent(action=KernelCommandType.ENTER, trade_id="t1", slot_id=0,
                 asset="TRX-USDT", side=TradeSide.SHORT, size=10.0) -> KernelIntent:
    return KernelIntent(
        timestamp=datetime.now(timezone.utc),
        intent_id=trade_id,
        trade_id=trade_id,
        slot_id=slot_id,
        asset=asset,
        action=action,
        side=side,
        reason="test",
        target_size=size,
        leverage=1.0,
        reference_price=100.0,
        exit_leg_ratios=(1.0,),
        metadata={},
    )


# ── Bug 1: cancel_async — must be awaitable, must not go through thread pool ──

class TestCancelAsync:
    def test_cancel_async_is_coroutine(self):
        venue = _make_venue()
        order = _make_order()
        result = venue.cancel_async(order)
        assert inspect.iscoroutine(result), "cancel_async must return a coroutine"
        asyncio.run(result)

    def test_cancel_async_calls_backend_cancel_directly(self):
        """cancel_async must await backend.cancel — not go through _run/thread pool."""
        venue = _make_venue()
        order = _make_order()

        async def _run():
            events = await venue.cancel_async(order)
            return events

        events = asyncio.run(_run())
        venue.backend.cancel.assert_awaited_once()
        assert len(events) == 1
        assert events[0].kind == KernelEventKind.CANCEL_ACK

    def test_cancel_async_cancel_rejected_maps_correctly(self):
        """Backend returning REJECTED → CANCEL_REJECT event."""
        venue = _make_venue(cancel_response={"status": "REJECTED", "msg": "no such order", "orderId": "O1", "clientOrderId": "C1"})
        order = _make_order()

        events = asyncio.run(venue.cancel_async(order))
        assert events[0].kind == KernelEventKind.CANCEL_REJECT

    def test_cancel_async_rate_limited_maps_correctly(self):
        venue = _make_venue(cancel_response={"status": "RATE_LIMITED", "msg": "too fast", "orderId": "O1", "clientOrderId": "C1"})
        order = _make_order()

        events = asyncio.run(venue.cancel_async(order))
        assert events[0].kind == KernelEventKind.RATE_LIMITED

    def test_cancel_async_backend_none_cancel_returns_empty_response(self):
        """If backend has no cancel method, cancel_async returns CANCELED event with None response."""
        venue = _make_venue()
        del venue.backend.cancel  # remove cancel attribute
        venue.backend.cancel = None
        order = _make_order()

        events = asyncio.run(venue.cancel_async(order))
        # None response → raw={} → status defaults to CANCELED → CANCEL_ACK
        assert events[0].kind == KernelEventKind.CANCEL_ACK

    def test_cancel_async_no_deadlock_from_running_loop(self):
        """cancel_async must not spin up a thread pool — safe inside running loop."""
        venue = _make_venue()
        order = _make_order()

        async def _inner():
            # If cancel_async used _run() with thread pool, this would deadlock
            # because aiohttp is bound to this event loop.
            return await venue.cancel_async(order)

        events = asyncio.run(_inner())
        assert len(events) >= 1  # got events without deadlock


# ── Bug 2: process_intent_async CANCEL path uses cancel_async ─────────────────

class TestCancelAsyncUsedByKernel:
    def test_process_intent_cancel_uses_cancel_async_when_available(self):
        """process_intent_async CANCEL branch must await cancel_async (not sync cancel)
        when the venue exposes cancel_async.

        Strategy: inject a fake slot with an active_exit_order so the cancel
        condition is met, then verify cancel_async was awaited.
        """
        from prod.clean_arch.dita_v2.contracts import VenueOrderStatus

        cancel_async_calls = []

        fake_order = VenueOrder(
            internal_trade_id="t1",
            venue_order_id="O1",
            venue_client_id="C1",
            side=TradeSide.SHORT,
            intended_size=10.0,
            filled_size=0.0,
            average_fill_price=0.0,
            status=VenueOrderStatus.NEW,
            metadata={"slot_id": 0, "asset": "TRX-USDT"},
        )

        async def _fake_cancel_async(order, *, reason=""):
            cancel_async_calls.append(order)
            return [VenueEvent(
                timestamp=datetime.now(timezone.utc),
                event_id="EV-001",
                trade_id="t1",
                slot_id=0,
                kind=KernelEventKind.CANCEL_ACK,
                status=VenueEventStatus.CANCELED,
                venue_order_id="O1",
                venue_client_id="C1",
                side=TradeSide.SHORT,
                asset="TRX-USDT",
                price=0.0, size=0.0, filled_size=0.0, remaining_size=0.0,
                reason=reason, raw_payload={}, metadata={},
            )]

        scenario = MockVenueScenario()
        venue = MockVenueAdapter(scenario=scenario)
        venue.cancel_async = _fake_cancel_async

        kernel = ExecutionKernel(max_slots=1, venue=venue)

        # Inject an active_exit_order via a mock slot view so the cancel condition is met.
        # KernelSlotView.active_exit_order is Rust-backed — patch at the method level.
        mock_slot = MagicMock()
        mock_slot.active_exit_order = fake_order
        mock_slot.active_entry_order = None
        mock_slot.fsm_state = TradeStage.POSITION_OPEN

        with patch.object(kernel, "slot", return_value=mock_slot):
            async def _run():
                cancel_intent = _make_intent(action=KernelCommandType.CANCEL)
                return await kernel.process_intent_async(cancel_intent)

            asyncio.run(_run())

        assert len(cancel_async_calls) > 0, (
            "process_intent_async CANCEL path must call cancel_async, not sync cancel"
        )

    def test_process_intent_cancel_falls_back_to_sync_for_mock(self):
        """If venue has no cancel_async, fall back to sync cancel (mock venue compat)."""
        scenario = MockVenueScenario()
        venue = MockVenueAdapter(scenario=scenario)
        # Confirm MockVenueAdapter has no cancel_async
        assert not hasattr(venue, "cancel_async"), "MockVenueAdapter should not have cancel_async for this test"

        kernel = ExecutionKernel(max_slots=1, venue=venue)

        async def _run():
            enter = _make_intent(action=KernelCommandType.ENTER)
            await kernel.process_intent_async(enter)
            cancel = _make_intent(action=KernelCommandType.CANCEL)
            return await kernel.process_intent_async(cancel)

        # Must not raise even without cancel_async
        outcome = asyncio.run(_run())
        assert outcome is not None


# ── Bug 3: S2 task guard — no duplicate background tasks per symbol ───────────

class TestS2TaskGuard:
    def test_s2_task_stored_on_creation(self):
        """Background refresh task must be stored in _s2_tasks for its duration."""
        from prod.clean_arch.adapters.bingx_direct import BingxDirectExecutionAdapter

        adapter = BingxDirectExecutionAdapter.__new__(BingxDirectExecutionAdapter)
        adapter._s2_tasks = {}
        adapter._state = None
        adapter._state_refreshed_at = 0.0

        refresh_started = asyncio.Event()
        refresh_proceed = asyncio.Event()

        async def _slow_refresh(symbol, include_history=False):
            refresh_started.set()
            await refresh_proceed.wait()
            return MagicMock()

        adapter._refresh_exchange_state = _slow_refresh
        from unittest.mock import MagicMock as MM
        adapter._instrument_venue_symbol = lambda a: a.upper().replace("-", "") + "USDT"

        async def _run():
            # Simulate create_task for background refresh
            task = asyncio.create_task(
                adapter._refresh_state_background("TRX-USDT"),
                name="state_refresh_TRXUSDT",
            )
            adapter._s2_tasks["TRXUSDT"] = task
            task.add_done_callback(lambda _t: adapter._s2_tasks.pop("TRXUSDT", None))

            # While task is running, entry should exist
            await refresh_started.wait()
            assert "TRXUSDT" in adapter._s2_tasks, "Task must be tracked while pending"
            assert not adapter._s2_tasks["TRXUSDT"].done()

            # Let it complete
            refresh_proceed.set()
            await asyncio.sleep(0.01)
            # After completion, entry removed by done callback
            assert "TRXUSDT" not in adapter._s2_tasks, "Done task must be removed from _s2_tasks"

        asyncio.run(_run())

    def test_s2_no_duplicate_task_when_one_pending(self):
        """When a background refresh is already pending, a second submit must not
        create a duplicate task — avoids concurrent REST writes to self._state."""
        from prod.clean_arch.adapters.bingx_direct import BingxDirectExecutionAdapter

        adapter = BingxDirectExecutionAdapter.__new__(BingxDirectExecutionAdapter)
        adapter._s2_tasks = {}
        adapter._state = None
        adapter._state_refreshed_at = 0.0

        call_count = 0

        async def _counting_refresh(symbol, include_history=False):
            nonlocal call_count
            call_count += 1
            await asyncio.sleep(0.05)  # simulate slow refresh
            return MagicMock()

        adapter._refresh_exchange_state = _counting_refresh

        async def _simulate_submit(symbol: str):
            existing = adapter._s2_tasks.get(symbol)
            if existing is None or existing.done():
                task = asyncio.create_task(adapter._refresh_state_background(symbol))
                adapter._s2_tasks[symbol] = task
                task.add_done_callback(lambda _t: adapter._s2_tasks.pop(symbol, None))

        async def _run():
            sym = "TRXUSDT"
            # Fire two rapid "submits" while first refresh is still running
            await _simulate_submit(sym)
            await _simulate_submit(sym)  # should be skipped
            await asyncio.sleep(0.1)    # wait for all tasks

        asyncio.run(_run())
        assert call_count == 1, (
            f"Expected 1 background refresh call, got {call_count}. "
            "Duplicate tasks fire redundant REST calls and cause stale last-writer-wins."
        )

    def test_s2_new_task_created_after_previous_done(self):
        """After a background task completes, next submit should create a new one."""
        from prod.clean_arch.adapters.bingx_direct import BingxDirectExecutionAdapter

        adapter = BingxDirectExecutionAdapter.__new__(BingxDirectExecutionAdapter)
        adapter._s2_tasks = {}
        adapter._state = None
        adapter._state_refreshed_at = 0.0
        call_count = 0

        async def _fast_refresh(symbol, include_history=False):
            nonlocal call_count
            call_count += 1
            return MagicMock()

        adapter._refresh_exchange_state = _fast_refresh

        async def _simulate_submit(symbol: str):
            existing = adapter._s2_tasks.get(symbol)
            if existing is None or existing.done():
                task = asyncio.create_task(adapter._refresh_state_background(symbol))
                adapter._s2_tasks[symbol] = task
                task.add_done_callback(lambda _t: adapter._s2_tasks.pop(symbol, None))

        async def _run():
            sym = "TRXUSDT"
            await _simulate_submit(sym)
            await asyncio.sleep(0.01)   # first task completes
            await _simulate_submit(sym)  # second submit → new task OK
            await asyncio.sleep(0.01)

        asyncio.run(_run())
        assert call_count == 2, f"Expected 2 refresh calls (one per submit after prior done), got {call_count}"


# ── _events_from_submit with None snapshots ───────────────────────────────────

class TestEventsFromSubmitNoneSnapshots:
    def _venue(self):
        return _make_venue()

    def test_filled_market_order_emits_full_fill_no_snapshots(self):
        """MARKET FILLED with None snapshots → ORDER_ACK + FULL_FILL via executedQty."""
        venue = self._venue()
        intent = _make_intent()
        receipt = _mock_receipt("FILLED", executed_qty=10.0)

        events = venue._events_from_submit(intent, receipt, None, None)
        kinds = [e.kind for e in events]
        assert KernelEventKind.ORDER_ACK in kinds
        assert KernelEventKind.FULL_FILL in kinds
        fill = next(e for e in events if e.kind == KernelEventKind.FULL_FILL)
        assert fill.filled_size == pytest.approx(10.0)

    def test_filled_no_executed_qty_falls_back_to_target_size(self):
        """FILLED with no executedQty → fill size falls back to intent.target_size."""
        venue = self._venue()
        intent = _make_intent(size=10.0)
        receipt = _mock_receipt("FILLED", executed_qty=0.0)
        receipt.raw_ack["executedQty"] = "0"

        events = venue._events_from_submit(intent, receipt, None, None)
        fill = next((e for e in events if e.kind == KernelEventKind.FULL_FILL), None)
        assert fill is not None
        assert fill.filled_size == pytest.approx(10.0), "Should fall back to target_size when executedQty=0"

    def test_acked_order_no_fill_event(self):
        """NEW/ACKED order with None snapshots → ORDER_ACK only, no FULL_FILL."""
        venue = self._venue()
        intent = _make_intent()
        receipt = _mock_receipt("NEW", executed_qty=0.0)
        receipt.raw_ack["executedQty"] = "0"
        receipt.raw_ack["status"] = "NEW"

        events = venue._events_from_submit(intent, receipt, None, None)
        kinds = [e.kind for e in events]
        assert KernelEventKind.ORDER_ACK in kinds
        assert KernelEventKind.FULL_FILL not in kinds
        assert KernelEventKind.PARTIAL_FILL not in kinds

    def test_rejected_order_emits_order_reject(self):
        """REJECTED → ORDER_REJECT only."""
        venue = self._venue()
        intent = _make_intent()
        receipt = _mock_receipt("REJECTED", executed_qty=0.0)
        receipt.raw_ack["status"] = "REJECTED"
        receipt.raw_ack["msg"] = "min qty not met"

        events = venue._events_from_submit(intent, receipt, None, None)
        assert len(events) == 1
        assert events[0].kind == KernelEventKind.ORDER_REJECT
        assert "min qty" in events[0].reason

    def test_partial_fill_emits_partial_fill_event(self):
        """PARTIALLY_FILLED → ORDER_ACK + PARTIAL_FILL."""
        venue = self._venue()
        intent = _make_intent(size=10.0)
        receipt = _mock_receipt("PARTIALLY_FILLED", executed_qty=5.0)
        receipt.raw_ack["status"] = "PARTIALLY_FILLED"
        receipt.raw_ack["executedQty"] = "5.0"

        events = venue._events_from_submit(intent, receipt, None, None)
        kinds = [e.kind for e in events]
        assert KernelEventKind.PARTIAL_FILL in kinds
        partial = next(e for e in events if e.kind == KernelEventKind.PARTIAL_FILL)
        assert partial.filled_size == pytest.approx(5.0)
        assert partial.remaining_size == pytest.approx(5.0)

    def test_rate_limited_returns_rate_limited_event(self):
        """RATE_LIMITED → single RATE_LIMITED event, no fills."""
        venue = self._venue()
        intent = _make_intent()
        receipt = _mock_receipt("RATE_LIMITED", executed_qty=0.0)
        receipt.raw_ack = {"status": "RATE_LIMITED", "msg": "too many requests"}

        events = venue._events_from_submit(intent, receipt, None, None)
        assert len(events) == 1
        assert events[0].kind == KernelEventKind.RATE_LIMITED


# ── _filled_size_from_snapshots with None inputs ──────────────────────────────

class TestFilledSizeFromNoneSnapshots:
    def test_both_none_returns_zero(self):
        result = BingxVenueAdapter._filled_size_from_snapshots(None, None, "TRX-USDT")
        assert result == 0.0

    def test_before_none_after_valid(self):
        after = MagicMock()
        after.open_positions = {"TRX-USDT": {"symbol": "TRX-USDT", "positionAmt": "10.0"}}
        result = BingxVenueAdapter._filled_size_from_snapshots(None, after, "TRX-USDT")
        assert result == pytest.approx(10.0)

    def test_both_valid_returns_diff(self):
        before = MagicMock()
        before.open_positions = {"TRX-USDT": {"symbol": "TRX-USDT", "positionAmt": "0.0"}}
        after = MagicMock()
        after.open_positions = {"TRX-USDT": {"symbol": "TRX-USDT", "positionAmt": "10.0"}}
        result = BingxVenueAdapter._filled_size_from_snapshots(before, after, "TRX-USDT")
        assert result == pytest.approx(10.0)


# ── _events_from_cancel ignores before/after completely ──────────────────────

class TestEventsFromCancelIgnoresSnapshots:
    def _venue(self):
        return _make_venue()

    def test_cancel_ack_with_none_snapshots(self):
        venue = self._venue()
        order = _make_order()
        events = venue._events_from_cancel(order, {"status": "CANCELED", "orderId": "O1"}, None, None)
        assert len(events) == 1
        assert events[0].kind == KernelEventKind.CANCEL_ACK

    def test_cancel_ack_with_garbage_snapshots_same_result(self):
        """Passing arbitrary objects as before/after must produce the same events
        as passing None — confirming they're truly ignored."""
        venue = self._venue()
        order = _make_order()

        events_null = venue._events_from_cancel(order, {"status": "CANCELED"}, None, None)
        events_junk = venue._events_from_cancel(order, {"status": "CANCELED"}, "JUNK", 42)
        assert events_null[0].kind == events_junk[0].kind
        assert events_null[0].status == events_junk[0].status

    def test_cancel_reject_with_none_snapshots(self):
        venue = self._venue()
        order = _make_order()
        events = venue._events_from_cancel(order, {"status": "REJECTED", "msg": "already filled"}, None, None)
        assert events[0].kind == KernelEventKind.CANCEL_REJECT

    def test_cancel_rate_limited_with_none_snapshots(self):
        venue = self._venue()
        order = _make_order()
        events = venue._events_from_cancel(order, {"status": "RATE_LIMITED", "msg": "slow down"}, None, None)
        assert events[0].kind == KernelEventKind.RATE_LIMITED


# ── connect() no double refresh ───────────────────────────────────────────────

class TestConnectNoDoubleRefresh:
    def test_connect_calls_backend_connect_once_no_extra_snapshot(self):
        """connect() must call backend.connect() exactly once and not call
        refresh_state() a second time (the redundant _backend_snapshot was removed)."""
        backend = MagicMock()
        backend.connect = AsyncMock(return_value=True)
        backend.refresh_state = AsyncMock(return_value=_flat_snap())
        venue = BingxVenueAdapter.__new__(BingxVenueAdapter)
        venue.backend = backend
        venue._event_seq = itertools.count(1)
        venue._snap_lock = threading.Lock()
        venue._snapshot_ready = threading.Event()
        venue._snapshot_ready.set()
        venue._last_snapshot = None

        result = venue.connect()
        assert result is True
        # backend.connect called once
        backend.connect.assert_called_once()
        # refresh_state must NOT have been called from connect() itself
        backend.refresh_state.assert_not_called()

    def test_connect_without_backend_connect_is_noop(self):
        """If backend has no connect(), connect() returns True without crashing."""
        backend = MagicMock(spec=[])  # empty spec — no attributes
        venue = BingxVenueAdapter.__new__(BingxVenueAdapter)
        venue.backend = backend
        venue._event_seq = itertools.count(1)
        venue._snap_lock = threading.Lock()
        venue._snapshot_ready = threading.Event()
        venue._snapshot_ready.set()
        venue._last_snapshot = None

        result = venue.connect()
        assert result is True


# ── submit() sync path with None snapshots ────────────────────────────────────

class TestSubmitSyncNoneSnapshots:
    def test_submit_sync_filled_emits_full_fill(self):
        """Sync submit() with None snapshots must still emit FULL_FILL for FILLED receipt."""
        backend = MagicMock()
        backend.submit_intent = MagicMock(return_value=_mock_receipt("FILLED", 10.0))
        venue = BingxVenueAdapter.__new__(BingxVenueAdapter)
        venue.backend = backend
        venue._event_seq = itertools.count(1)
        venue._snap_lock = threading.Lock()
        venue._snapshot_ready = threading.Event()
        venue._snapshot_ready.set()
        venue._last_snapshot = None

        intent = _make_intent(size=10.0)
        # Patch _call_backend to return the receipt directly
        from unittest.mock import patch as _patch
        with _patch.object(venue, "_call_backend", return_value=_mock_receipt("FILLED", 10.0)):
            with _patch.object(venue, "_legacy_intent", return_value=MagicMock()):
                events = venue.submit(intent)

        kinds = [e.kind for e in events]
        assert KernelEventKind.FULL_FILL in kinds


# ── cancel() dead-branch audit ────────────────────────────────────────────────

class TestCancelBranchAudit:
    def test_cancel_uses_cancel_not_cancel_order(self):
        """BingxDirectExecutionAdapter has cancel(), not cancel_order().
        The cancel_order branch was dead code — confirm cancel() uses the right branch."""
        backend = MagicMock()
        # Has cancel but not cancel_order
        backend.cancel = AsyncMock(return_value={"status": "CANCELED", "orderId": "O1", "clientOrderId": "C1"})
        del backend.cancel_order  # ensure cancel_order doesn't exist
        venue = BingxVenueAdapter.__new__(BingxVenueAdapter)
        venue.backend = backend
        venue._event_seq = itertools.count(1)
        venue._snap_lock = threading.Lock()
        venue._snapshot_ready = threading.Event()
        venue._snapshot_ready.set()
        venue._last_snapshot = None

        order = _make_order()
        # Sync cancel goes through _call_backend → _run
        # With asyncio not running, _run calls asyncio.run() directly
        events = venue.cancel(order)
        assert len(events) >= 1
        assert events[0].kind == KernelEventKind.CANCEL_ACK

    def test_cancel_no_cancel_attribute_falls_to_third_branch(self):
        """If backend has neither cancel_order nor cancel, raise RuntimeError for live backend."""
        backend = MagicMock(spec=[])  # no methods
        venue = BingxVenueAdapter.__new__(BingxVenueAdapter)
        venue.backend = backend
        venue._event_seq = itertools.count(1)
        venue._snap_lock = threading.Lock()
        venue._snapshot_ready = threading.Event()
        venue._snapshot_ready.set()
        venue._last_snapshot = None

        order = _make_order()
        with pytest.raises(RuntimeError, match="cancel surface"):
            venue.cancel(order)


class TestHttpSigningBodyNoDuplicateSignature:
    """Regression: http.py was appending signature= twice in POST body.

    build_signed_params injects 'signature' into the returned dict.
    canonical_query(payload) then serialised it, then
    f"{canonical}&signature={payload['signature']}" appended it again.
    BingX received body with two signature= fields.  Fix: exclude
    'signature' from canonical before appending.
    """

    def test_no_duplicate_signature_in_post_body(self):
        from prod.bingx.signing import build_signed_params, canonical_query
        import uuid

        secret = "testsecret1234567890"
        params = {
            "symbol": "TRX-USDT",
            "side": "SELL",
            "positionSide": "BOTH",
            "type": "MARKET",
            "quantity": "10.0",
            "clientOrderId": uuid.uuid4().hex,
        }
        signed = build_signed_params(params, secret, recv_window_ms=5000)

        # Replicate fixed http.py body construction
        canonical = canonical_query({k: v for k, v in signed.items() if k != "signature"})
        body = f"{canonical}&signature={signed['signature']}"

        assert body.count("signature") == 1, (
            "POST body must contain exactly one 'signature=' field"
        )
        # signature must be the last field (appended, not embedded)
        assert body.endswith(f"&signature={signed['signature']}"), (
            "signature must be the final field in the POST body"
        )

    def test_canonical_without_signature_matches_hmac_input(self):
        """The canonical query we send must match exactly what HMAC was computed over."""
        from prod.bingx.signing import build_signed_params, canonical_query, sign_query
        import uuid

        secret = "anothertestsecret99"
        params = {
            "symbol": "ETH-USDT",
            "side": "BUY",
            "type": "MARKET",
            "quantity": "1.0",
            "clientOrderId": uuid.uuid4().hex,
        }
        signed = build_signed_params(params, secret, recv_window_ms=5000)

        # The string HMAC was computed over (inside build_signed_params)
        signed_without_sig = {k: v for k, v in signed.items() if k != "signature"}
        expected_hmac_input = canonical_query(signed_without_sig)

        # Re-derive HMAC from the canonical we're about to send
        recomputed_sig = sign_query(secret, expected_hmac_input)
        assert recomputed_sig == signed["signature"], (
            "canonical without signature must reproduce the HMAC"
        )