siloqy/prod/tests/test_pink_async_fill_pump.py

"""L1 — async-fill pump.

A resting order (LIMIT-style: ACK on submit, no synchronous fill) fills on a
*later* venue reconcile. `PinkDirectRuntime.pump_venue_events()` must drain that
fill into the kernel so capital settles and the FSM advances, persist the result,
and dedup duplicate reconcile events (no double-settle). MockVenue only; no exchange.
"""

from __future__ import annotations

import asyncio
from datetime import datetime, timezone

from prod.clean_arch.dita_v2 import (
    ExecutionKernel,
    InMemoryControlPlane,
    KernelCommandType,
    KernelControlSnapshot,
    KernelEventKind,
    KernelMode,
    KernelVerbosity,
    MemoryKernelJournal,
    MockVenueAdapter,
    MockVenueScenario,
    TradeSide,
    VenueEvent,
    VenueEventStatus,
)
from prod.clean_arch.dita_v2.contracts import KernelIntent, TradeStage
from prod.clean_arch.dita import DecisionConfig, DecisionEngine, IntentEngine
from prod.clean_arch.persistence import PinkClickHousePersistence
from prod.clean_arch.runtime.pink_direct import PinkDirectRuntime
from prod.clean_arch.ports.data_feed import DataFeedPort


class _Sink:
    def __init__(self) -> None:
        self.calls: list[tuple[str, dict]] = []

    def __call__(self, table: str, row: dict) -> None:
        self.calls.append((table, dict(row)))

    def tables(self) -> list[str]:
        return [t for t, _ in self.calls]


class _StubFeed(DataFeedPort):
    async def connect(self) -> bool:
        return True

    async def disconnect(self) -> None:
        pass

    async def get_latest_snapshot(self, symbol):
        return None

    async def subscribe_snapshots(self, callback) -> None:
        pass

    async def get_acb_update(self):
        return None

    def get_latency_ms(self) -> float:
        return 0.0

    def health_check(self) -> bool:
        return True


class _DelayedFillVenue(MockVenueAdapter):
    """MockVenue whose submit ACKs only; queued fills surface on reconcile()."""

    def __init__(self, scenario=None) -> None:
        super().__init__(scenario)
        self._pending: list[VenueEvent] = []

    def queue(self, event: VenueEvent) -> None:
        self._pending.append(event)

    def reconcile(self):
        out, self._pending = list(self._pending), []
        return out


def _mk_runtime():
    # ACK-only: no synchronous fill on submit (resting order).
    venue = _DelayedFillVenue(
        MockVenueScenario(emit_fill_on_submit=False, partial_fill_ratio=0.0, emit_ack_before_fill=True)
    )
    kernel = ExecutionKernel(
        control_plane=InMemoryControlPlane(
            KernelControlSnapshot(mode=KernelMode.DEBUG, verbosity=KernelVerbosity.TRACE)
        ),
        venue=venue,
        journal=MemoryKernelJournal(),
    )
    kernel.account.snapshot.capital = 25_000.0
    kernel.account.snapshot.peak_capital = 25_000.0
    kernel.account.snapshot.equity = 25_000.0
    sink = _Sink()
    cfg = DecisionConfig()
    persistence = PinkClickHousePersistence(kernel.account, sink=sink, v7_sink=sink)
    runtime = PinkDirectRuntime(
        data_feed=_StubFeed(), kernel=kernel,
        decision_engine=DecisionEngine(cfg), intent_engine=IntentEngine(cfg),
        persistence=persistence, market_state_runtime=None,
    )
    return runtime, kernel, venue, sink


def _intent(action, *, size, price, reason="TEST"):
    return KernelIntent(
        timestamp=datetime.now(timezone.utc), intent_id=f"i-{reason}", trade_id="T1",
        slot_id=0, asset="BTCUSDT", side=TradeSide.SHORT, action=action,
        reference_price=price, target_size=size, leverage=2.0, exit_leg_ratios=(1.0,), reason=reason,
    )


def _fill_for(order, *, kind, price, filled, remaining, eid):
    return VenueEvent(
        timestamp=datetime.now(timezone.utc), event_id=eid, trade_id="T1", slot_id=0,
        kind=kind, status=VenueEventStatus.FILLED if kind == KernelEventKind.FULL_FILL else VenueEventStatus.PARTIALLY_FILLED,
        venue_order_id=order.venue_order_id, venue_client_id=order.venue_client_id,
        side=TradeSide.SHORT, asset="BTCUSDT", price=price, size=1.0,
        filled_size=filled, remaining_size=remaining,
    )


def test_resting_entry_fills_via_pump_and_dedups():
    runtime, kernel, venue, sink = _mk_runtime()

    # ENTER rests (ACK only, nothing filled).
    kernel.process_intent(_intent(KernelCommandType.ENTER, size=1.0, price=100.0))
    slot = kernel.slot(0)
    assert slot.fsm_state == TradeStage.ENTRY_WORKING
    assert abs(slot.size) < 1e-9
    entry_order = slot.active_entry_order
    assert entry_order is not None

    # A later reconcile surfaces the fill -> pump settles it.
    venue.queue(_fill_for(entry_order, kind=KernelEventKind.FULL_FILL, price=100.0, filled=1.0, remaining=0.0, eid="EVF1"))
    applied = asyncio.run(runtime.pump_venue_events())
    assert applied == 1
    assert kernel.slot(0).fsm_state == TradeStage.POSITION_OPEN
    assert abs(kernel.slot(0).size - 1.0) < 1e-9
    assert "account_events" in sink.tables() and "position_state" in sink.tables()
    assert "ENTRY_FILLED" in [r["event_type"] for t, r in sink.calls if t == "trade_reconstruction"]

    # Duplicate reconcile event -> kernel dedups; pump applies nothing, no double-settle.
    cap_before = kernel.account.snapshot.capital
    rows_before = len(sink.calls)
    venue.queue(_fill_for(entry_order, kind=KernelEventKind.FULL_FILL, price=100.0, filled=1.0, remaining=0.0, eid="EVF1"))
    applied2 = asyncio.run(runtime.pump_venue_events())
    assert applied2 == 0, "duplicate fill must be deduped by the kernel"
    assert kernel.account.snapshot.capital == cap_before
    assert len(sink.calls) == rows_before, "no rows persisted for a deduped event"


def test_resting_exit_fills_via_pump_settles_capital():
    runtime, kernel, venue, sink = _mk_runtime()

    # Open a position via the pump (entry rests, then fills).
    kernel.process_intent(_intent(KernelCommandType.ENTER, size=1.0, price=100.0))
    venue.queue(_fill_for(kernel.slot(0).active_entry_order, kind=KernelEventKind.FULL_FILL, price=100.0, filled=1.0, remaining=0.0, eid="EVE1"))
    asyncio.run(runtime.pump_venue_events())
    assert kernel.slot(0).fsm_state == TradeStage.POSITION_OPEN
    cap_after_entry = kernel.account.snapshot.capital  # entry does not realize PnL

    # EXIT rests (ACK only), then fills @90 on a later reconcile -> SHORT profit.
    kernel.process_intent(_intent(KernelCommandType.EXIT, size=1.0, price=90.0, reason="TP"))
    exit_order = kernel.slot(0).active_exit_order
    assert exit_order is not None
    venue.queue(_fill_for(exit_order, kind=KernelEventKind.FULL_FILL, price=90.0, filled=1.0, remaining=0.0, eid="EVX1"))
    applied = asyncio.run(runtime.pump_venue_events())
    assert applied == 1
    assert kernel.slot(0).closed
    assert kernel.slot(0).fsm_state == TradeStage.CLOSED
    # SHORT 1.0 @100 -> exit @90, leverage 2 => realized profit > 0; capital rose.
    assert kernel.account.snapshot.capital > cap_after_entry
    tables = sink.tables()
    assert "trade_exit_legs" in tables, "async exit must persist a leg row"
    assert "trade_events" in tables, "async close must persist a terminal trade_event"