trex-core/tests/application/test_execution_policy.py

"""Execution policy tests: slippage dispatch + timeout cancel (dcy-022 / REQ-069/070).

Uses ProgrammableBroker from test_execution_cancel for fine-grained control.
Also covers operator cancel_pending_slot (REQ-071).
"""
from __future__ import annotations

import logging
from dataclasses import replace
from datetime import datetime, timedelta, timezone


from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
from trex_core.application.broker_port import BalanceItem, FillStatus, OrderAck, OrderRequest
from trex_core.application.execution import ExecutionService
from trex_core.application.intents import BuyIntent, IntentService, SellIntent
from trex_core.domain.slot import SlotService, SlotState
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository


class _RecordingBroker:
    """Broker that records submitted orders and lets tests control poll responses."""

    def __init__(self):
        self._seq = 0
        self.orders: list[OrderRequest] = []
        self.cancel_calls: list = []
        self.cancel_raises = False
        self.responder = lambda side: FillStatus(
            state="filled", filled_qty=10, avg_price=1000.0,
            execution_id="exec-1", fee=0.0, tax=0.0,
        )
        self.balance: dict[str, int] = {"005930": 10}

    def name(self):
        return "stub"

    def submit_order(self, req: OrderRequest):
        self._seq += 1
        self.orders.append(req)
        return OrderAck(accepted=True, broker_order_id=f"o{self._seq}")

    def poll_fill(self, broker_order_id, *, side, symbol):
        return self.responder(side)

    def cancel_order(self, broker_order_id, *, side, symbol, qty):
        self.cancel_calls.append((broker_order_id, side, symbol, qty))
        if self.cancel_raises:
            raise RuntimeError("broker error")
        return True

    def get_balance(self):
        return [BalanceItem(symbol=s, hldg_qty=q) for s, q in self.balance.items() if q > 0]

    def get_current_price(self, symbol):
        return None


def _wire(broker):
    repo = InMemorySlotRepository()
    svc = SlotService(repo)
    intents = IntentService(svc, repo, InMemoryIntentStore(),
                            resolve_symbol=lambda pid: "005930")
    execution = ExecutionService(svc, repo, broker)
    repo.reconcile_universe("default", {"PF-1": "SYM"})
    return repo, svc, intents, execution


def _buy_to_pending(intents, qty=10, entry_price=1000, timeout_sec=None, slippage_pct=None):
    return intents.submit_buy(BuyIntent(
        intent_id="b1", qty=qty, entry_price=entry_price,
        portfolio_id="PF-1",
        timeout_sec=timeout_sec, slippage_pct=slippage_pct,
    )).slot_id


# ============================================================================
# REQ-070: slippage effective limit price
# ============================================================================

class TestSlippageDispatch:

    def test_buy_slippage_ceil_applied_to_limit(self):
        """buy: effective_limit = ceil(entry_price * (1 + pct/100))."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        _buy_to_pending(intents, entry_price=1000, slippage_pct=0.5)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()
        order = broker.orders[0]
        # ceil(1000 * 1.005) = ceil(1005.0) = 1005
        assert order.limit_price == 1005

    def test_buy_slippage_fractional_ceil(self):
        """ceil(1001 * 1.003) = ceil(1004.003) = 1005."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        _buy_to_pending(intents, entry_price=1001, slippage_pct=0.3)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()
        import math
        expected = int(math.ceil(1001 * 1.003))
        assert broker.orders[0].limit_price == expected

    def test_sell_slippage_floor_applied_to_limit(self):
        """sell: effective_limit = floor(limit_price * (1 - pct/100))."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        # drive to holding first
        broker.responder = lambda side: FillStatus(
            state="filled", filled_qty=10, avg_price=1000.0,
            execution_id="e1", fee=0.0, tax=0.0,
        )
        _buy_to_pending(intents)
        execution.run_once()  # buy filled -> holding
        # now submit sell with slippage
        sid = [s.slot_id for s in repo.list_slots("default")
               if s.status is SlotState.HOLDING][0]
        intents.submit_sell(SellIntent(
            intent_id="s1", slot_id=sid, slippage_pct=1.0, limit_price=2000.0,
        ))
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()  # dispatch sell (pending state already has limit_price)
        sell_orders = [o for o in broker.orders if o.side == "sell"]
        # floor(2000 * 0.99) = floor(1980.0) = 1980
        import math
        expected = int(math.floor(2000 * 0.99))
        assert sell_orders[0].limit_price == expected

    def test_no_slippage_when_slippage_pct_is_none(self):
        """No slippage_pct → effective_limit = entry_price unchanged."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        _buy_to_pending(intents, entry_price=1000, slippage_pct=None)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()
        assert broker.orders[0].limit_price == 1000

    def test_no_slippage_when_market_order_no_entry_price(self):
        """Market order (no entry_price → limit=None) → slippage not applied."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        intents.submit_buy(BuyIntent(
            intent_id="b1", qty=10, entry_price=None,
            portfolio_id="PF-1", slippage_pct=5.0,
        ))
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()
        assert broker.orders[0].limit_price is None

    def test_zero_slippage_uses_base_price(self):
        """slippage_pct=0 → effective_limit = ceil(1000 * 1.0) = 1000."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        _buy_to_pending(intents, entry_price=1000, slippage_pct=0.0)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()
        # 0.0 is falsy → slippage block skipped → no change
        assert broker.orders[0].limit_price == 1000


# ============================================================================
# REQ-069: timeout auto-cancel
# ============================================================================

class TestTimeoutCancel:

    def _buy_and_dispatch(self, intents, execution, repo, timeout_sec):
        """Drive a pending buy into dispatched state, then age the slot."""
        sid = _buy_to_pending(intents, timeout_sec=timeout_sec)
        # keep order in "accepted" (not filled) so poll always lands in else branch
        execution._repo  # just to confirm it's wired
        return sid

    def test_timeout_exceeded_sends_cancel_order(self):
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents, timeout_sec=30)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()  # dispatch (no fill yet)
        assert len(broker.cancel_calls) == 0

        # age the slot past timeout
        s = repo.fetch_slot(sid)
        repo._slots[sid] = replace(
            s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
        )
        r = execution.run_once()
        assert r.timeout_cancelled == 1
        assert len(broker.cancel_calls) == 1

    def test_timeout_not_exceeded_no_cancel(self):
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        _buy_to_pending(intents, timeout_sec=3600)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()  # dispatch
        # do NOT age the slot — timeout_sec is large
        r = execution.run_once()
        assert r.timeout_cancelled == 0
        assert len(broker.cancel_calls) == 0

    def test_no_timeout_sec_no_cancel(self):
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents, timeout_sec=None)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()  # dispatch
        # age the slot
        s = repo.fetch_slot(sid)
        repo._slots[sid] = replace(
            s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=9999),
        )
        r = execution.run_once()
        assert r.timeout_cancelled == 0
        assert len(broker.cancel_calls) == 0

    def test_timeout_cancel_best_effort_on_exception(self, caplog):
        """Broker exception during cancel must not raise — loop continues."""
        broker = _RecordingBroker()
        broker.cancel_raises = True
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents, timeout_sec=1)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()  # dispatch
        s = repo.fetch_slot(sid)
        repo._slots[sid] = replace(
            s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
        )
        with caplog.at_level(logging.WARNING):
            r = execution.run_once()
        # timeout_cancelled NOT incremented when the cancel call raises
        assert r.timeout_cancelled == 0
        assert repo.fetch_slot(sid).status is SlotState.PENDING_BUY  # state unchanged

    def test_timeout_cancel_does_not_change_slot_state(self):
        """Timeout cancel is best-effort: slot stays PENDING_BUY after cancel_order."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents, timeout_sec=1)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()
        s = repo.fetch_slot(sid)
        repo._slots[sid] = replace(
            s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
        )
        execution.run_once()
        # slot still PENDING_BUY — cancel is best-effort; rollback on next poll
        assert repo.fetch_slot(sid).status is SlotState.PENDING_BUY

    def test_timeout_cancel_then_fill_race_handled(self):
        """Cancel sent, but broker fills the order — is_filled path wins on next poll."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents, timeout_sec=1)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()  # dispatch
        # age the slot → cancel sent
        s = repo.fetch_slot(sid)
        repo._slots[sid] = replace(
            s, updated_at=datetime.now(tz=timezone.utc) - timedelta(seconds=60),
        )
        r1 = execution.run_once()
        assert r1.timeout_cancelled == 1
        # next poll: broker says filled anyway (race — fill beat the cancel)
        broker.responder = lambda side: FillStatus(
            state="filled", filled_qty=10, avg_price=1000.0,
            execution_id="exec-race", fee=0.0, tax=0.0,
        )
        r2 = execution.run_once()
        assert r2.filled == 1
        assert repo.fetch_slot(sid).status is SlotState.HOLDING


# ============================================================================
# REQ-071: cancel_pending_slot (operator backstop)
# ============================================================================

class TestCancelPendingSlot:

    def test_ok_sends_cancel_and_returns_ok(self):
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents)
        # dispatch so broker_order_id is set
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()

        result = execution.cancel_pending_slot(sid)
        assert result == "ok"
        assert len(broker.cancel_calls) == 1

    def test_not_pending_returns_not_pending(self):
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents)
        broker.responder = lambda side: FillStatus(
            state="filled", filled_qty=10, avg_price=1000.0,
            execution_id="e1", fee=0.0, tax=0.0,
        )
        execution.run_once()  # fill -> holding
        assert repo.fetch_slot(sid).status is SlotState.HOLDING
        result = execution.cancel_pending_slot(sid)
        assert result == "not_pending"

    def test_slot_not_found_returns_slot_not_found(self):
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        from uuid import uuid4
        result = execution.cancel_pending_slot(uuid4())
        assert result == "slot_not_found"

    def test_not_dispatched_returns_not_dispatched(self):
        """Slot is PENDING_BUY but broker_order_id not yet set."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents)
        # Do NOT call run_once — broker_order_id is still None
        result = execution.cancel_pending_slot(sid)
        assert result == "not_dispatched"

    def test_cancel_fail_returns_cancel_failed(self):
        broker = _RecordingBroker()
        broker.cancel_raises = True
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()  # dispatch

        result = execution.cancel_pending_slot(sid)
        assert result == "cancel_failed"

    def test_cancel_does_not_change_slot_state(self):
        """Operator cancel sends cancel_order; slot stays PENDING_BUY (execution loop transitions)."""
        broker = _RecordingBroker()
        repo, svc, intents, execution = _wire(broker)
        sid = _buy_to_pending(intents)
        broker.responder = lambda side: FillStatus(state="accepted")
        execution.run_once()

        execution.cancel_pending_slot(sid)
        assert repo.fetch_slot(sid).status is SlotState.PENDING_BUY