AC-0023: filter C-case quantity mismatch slots from strategy RPC, fail-closed on warn_store failure, symbol-aggregate verify clear

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-08-25 13:17:06 +00:00
committed by GitHub
co-authored by genichin
parent d9205fe151
commit 1179bdfe23
8 changed files with 666 additions and 13 deletions
@@ -0,0 +1,560 @@
"""AC-0023: C-case quantity mismatch — strategy RPC exclusion, fail-closed, and clearing.
Coverage:
AC-1: single portfolio mismatch → slot excluded
AC-2: multi-portfolio same symbol: aggregate matches → both slots included
AC-3: multi-portfolio same symbol: aggregate mismatches → both slots excluded
AC-4: different symbol slots unaffected by C-case warning on another symbol
AC-5: excluded_slots.reconcile_quantity_mismatch count in response
AC-6: warn_store=None → no filtering (backward compat), no excluded_slots key
AC-7: warn_store raises → fail-closed: slots=[], degraded=true, warn_store_unavailable
AC-8: intent admitted before warning is set → rejected at intent time (race)
AC-9: in-flight (terminal) slot transitions not blocked
AC-10: verify_slot_reconcile uses symbol aggregate for multi-portfolio clear
AC-11: verify_slot_reconcile clears all C-case warnings for symbol when resolved
AC-12: C-case warning not auto-cleared by time; only cleared on quantity match
"""
from __future__ import annotations
from trex_core.application.broker_port import BalanceItem
from trex_core.application.intents import BuyIntent, IntentService, SellIntent
from trex_core.application.messaging import RpcRequest
from trex_core.application.messaging_handlers import make_slots_rpc_handler
from trex_core.application.portfolio import (
InMemoryPauseStore,
InMemoryReconcileWarningStore,
PortfolioItem,
SnapshotStore,
)
from trex_core.application.reconciliation import (
reconcile_positions,
verify_slot_reconcile,
)
from trex_core.application.watchdog import WatchdogStore
from trex_core.domain.slot import SlotService, SlotState
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
INSTANCE = "default"
SYM_A = "104620"
SYM_B = "005930"
def _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10, instance_key=INSTANCE):
repo.reconcile_universe(instance_key, {portfolio_id: symbol})
slot = repo.find_slot_by_portfolio(instance_key, portfolio_id)
pending = svc.transit_slot(
slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="test", symbol=symbol,
)
holding = svc.transit_slot(
pending.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="test", avg_price=100.0, qty=float(qty),
)
return holding
def _make_handler(repo, warn_store, snapshot_items=None):
store = SnapshotStore()
if snapshot_items:
store.swap(snapshot_items)
return make_slots_rpc_handler(
repo,
instance_key=INSTANCE,
snapshot_store=store,
pause_store=InMemoryPauseStore(),
warn_store=warn_store,
watchdog_store=WatchdogStore(),
)
def _req(filter=None):
payload = {}
if filter:
payload["filter"] = filter
return RpcRequest(payload=payload, correlation_id="test")
class FakeBroker:
def __init__(self, balances):
self._b = balances
def name(self): return "fake"
def submit_order(self, req): ...
def poll_fill(self, *args, **kw): ...
def cancel_order(self, *args, **kw): ...
def get_balance(self): return list(self._b)
class InMemoryIntentStore:
def __init__(self):
self._db = {}
def try_insert(self, intent_id, intent_type, payload):
if intent_id in self._db:
return False
self._db[intent_id] = {"status": None, "slot_id": None, "reason": None}
return True
def get(self, intent_id):
rec = self._db.get(intent_id)
if rec is None:
return None
class R:
def __init__(self, d): self.__dict__.update(d)
r = R(rec)
r.intent_id = intent_id
return r
def finalize(self, intent_id, status, slot_id, reason):
self._db[intent_id] = {"status": status, "slot_id": slot_id, "reason": reason}
# ---------------------------------------------------------------------------
# AC-1: single portfolio C-case → slot excluded from strategy RPC
# ---------------------------------------------------------------------------
def test_ac1_single_portfolio_c_case_excluded():
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C")
handler = _make_handler(repo, warn_store)
resp = handler(_req(filter="holding"))
assert resp.payload["slots"] == []
assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 1
# ---------------------------------------------------------------------------
# AC-2: multi-portfolio same symbol: KIS 440, A=200, B=240 → aggregate 440 → both included
# ---------------------------------------------------------------------------
def test_ac2_multi_portfolio_aggregate_matches_both_included():
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
# Build two holding slots for same symbol
repo.reconcile_universe(INSTANCE, {"pf-A": SYM_A, "pf-B": SYM_A})
slot_a_base = repo.find_slot_by_portfolio(INSTANCE, "pf-A")
slot_b_base = repo.find_slot_by_portfolio(INSTANCE, "pf-B")
p = svc.transit_slot(slot_a_base.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="test", symbol=SYM_A)
slot_a = svc.transit_slot(p.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="test", avg_price=100.0, qty=200.0)
p = svc.transit_slot(slot_b_base.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="test", symbol=SYM_A)
slot_b = svc.transit_slot(p.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="test", avg_price=100.0, qty=240.0)
# No C-case warnings set (aggregate matches KIS 440)
handler = _make_handler(repo, warn_store)
resp = handler(_req(filter="holding"))
portfolio_ids = {r["portfolio_id"] for r in resp.payload["slots"]}
assert "pf-A" in portfolio_ids
assert "pf-B" in portfolio_ids
assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 0
def test_ac2_multi_portfolio_individual_qty_vs_kis_total_not_compared():
"""AC-2: each portfolio qty (200, 240) != KIS 440, but aggregate 440 == KIS 440 → no block."""
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
repo.reconcile_universe(INSTANCE, {"pf-A": SYM_A, "pf-B": SYM_A})
for pid, qty in [("pf-A", 200.0), ("pf-B", 240.0)]:
base = repo.find_slot_by_portfolio(INSTANCE, pid)
p = svc.transit_slot(base.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="test", symbol=SYM_A)
svc.transit_slot(p.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="test", avg_price=100.0, qty=qty)
# simulate reconcile_positions with broker=440 → aggregate 440 == 440 → no warning set
snapshot_items = [
PortfolioItem(portfolio_id="pf-A", symbol=SYM_A, cash_balance=0, asset_quantity=200.0),
PortfolioItem(portfolio_id="pf-B", symbol=SYM_A, cash_balance=0, asset_quantity=240.0),
]
broker = FakeBroker([BalanceItem(symbol=SYM_A, hldg_qty=440)])
store = SnapshotStore()
store.swap(snapshot_items)
reconcile_positions(store=store, broker=broker, slot_repo=repo,
instance_key=INSTANCE, warn_store=warn_store)
# No warnings should be set
assert warn_store.list_c_case_warned() == set()
handler = _make_handler(repo, warn_store, snapshot_items)
resp = handler(_req(filter="holding"))
assert len(resp.payload["slots"]) == 2
assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 0
# ---------------------------------------------------------------------------
# AC-3: multi-portfolio same symbol: aggregate mismatches → all excluded
# ---------------------------------------------------------------------------
def test_ac3_multi_portfolio_aggregate_mismatch_all_excluded():
"""KIS 440, hub A=200 + B=235 → aggregate 435 ≠ 440 → both slots excluded."""
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
repo.reconcile_universe(INSTANCE, {"pf-A": SYM_A, "pf-B": SYM_A})
for pid, qty in [("pf-A", 200.0), ("pf-B", 235.0)]:
base = repo.find_slot_by_portfolio(INSTANCE, pid)
p = svc.transit_slot(base.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="test", symbol=SYM_A)
svc.transit_slot(p.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="test", avg_price=100.0, qty=qty)
# Set C-case warning for both
snapshot_items = [
PortfolioItem(portfolio_id="pf-A", symbol=SYM_A, cash_balance=0, asset_quantity=200.0),
PortfolioItem(portfolio_id="pf-B", symbol=SYM_A, cash_balance=0, asset_quantity=235.0),
]
broker = FakeBroker([BalanceItem(symbol=SYM_A, hldg_qty=440)])
store = SnapshotStore()
store.swap(snapshot_items)
reconcile_positions(store=store, broker=broker, slot_repo=repo,
instance_key=INSTANCE, warn_store=warn_store)
# Both should be C-case warned
assert len(warn_store.list_c_case_warned()) == 2
handler = _make_handler(repo, warn_store, snapshot_items)
resp = handler(_req(filter="holding"))
assert resp.payload["slots"] == []
assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 2
# ---------------------------------------------------------------------------
# AC-4: other-symbol slots not affected by C-case warning on a different symbol
# ---------------------------------------------------------------------------
def test_ac4_other_symbol_slots_unaffected():
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
repo.reconcile_universe(INSTANCE, {"pf-A": SYM_A, "pf-B": SYM_B})
# pf-A: SYM_A, holding, C-case warned
base_a = repo.find_slot_by_portfolio(INSTANCE, "pf-A")
p = svc.transit_slot(base_a.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="test", symbol=SYM_A)
slot_a = svc.transit_slot(p.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="test", avg_price=100.0, qty=10.0)
warn_store.set_warned(str(slot_a.slot_id), portfolio_id="pf-A", case_code="C")
# pf-B: SYM_B, holding, no warning
base_b = repo.find_slot_by_portfolio(INSTANCE, "pf-B")
p = svc.transit_slot(base_b.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="test", symbol=SYM_B)
svc.transit_slot(p.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="test", avg_price=200.0, qty=5.0)
handler = _make_handler(repo, warn_store)
resp = handler(_req(filter="holding"))
portfolio_ids = {r["portfolio_id"] for r in resp.payload["slots"]}
assert "pf-A" not in portfolio_ids, "C-case warned slot should be excluded"
assert "pf-B" in portfolio_ids, "unwarned symbol slot should be included"
assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 1
# ---------------------------------------------------------------------------
# AC-5: excluded_slots present and count is correct
# ---------------------------------------------------------------------------
def test_ac5_excluded_slots_count_zero_when_no_warnings():
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
_make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
handler = _make_handler(repo, warn_store)
resp = handler(_req(filter="holding"))
assert "excluded_slots" in resp.payload
assert resp.payload["excluded_slots"] == {"reconcile_quantity_mismatch": 0}
# ---------------------------------------------------------------------------
# AC-6: warn_store=None → no filtering, no excluded_slots key (backward compat)
# ---------------------------------------------------------------------------
def test_ac6_no_warn_store_no_filtering_no_excluded_slots_key():
repo = InMemorySlotRepository()
svc = SlotService(repo)
_make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
handler = make_slots_rpc_handler(repo, instance_key=INSTANCE) # no warn_store
resp = handler(_req(filter="holding"))
# No warn_store → no filtering (slot included), no excluded_slots key
assert len(resp.payload["slots"]) == 1
assert "excluded_slots" not in resp.payload
# ---------------------------------------------------------------------------
# AC-7: warn_store raises → fail-closed
# ---------------------------------------------------------------------------
class BrokenWarnStore(InMemoryReconcileWarningStore):
"""Simulates a warn_store that fails on list_c_case_warned."""
def list_c_case_warned(self):
raise RuntimeError("db connection lost")
def test_ac7_warn_store_failure_fail_closed():
repo = InMemorySlotRepository()
svc = SlotService(repo)
_make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
broken_warn_store = BrokenWarnStore()
handler = _make_handler(repo, broken_warn_store)
resp = handler(_req())
assert resp.payload["slots"] == []
assert resp.payload["degraded"] is True
assert resp.payload["reason"] == "warn_store_unavailable"
assert "excluded_slots" in resp.payload
# ---------------------------------------------------------------------------
# AC-8: intent rejected if C-case warning set after last slots RPC (race)
# ---------------------------------------------------------------------------
def test_ac8_intent_rejected_when_c_case_warning_set_after_slots_rpc():
"""Even if strategy saw the slot in a previous RPC, intent is rejected at processing time."""
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
intent_store = InMemoryIntentStore()
intent_svc = IntentService(
svc, repo, intent_store,
instance_key=INSTANCE,
warn_store=warn_store,
)
# Warning set after strategy "saw" the slot
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C")
sell = SellIntent(intent_id="sell-001", slot_id=slot.slot_id, reason="test")
d = intent_svc.submit_sell(sell)
assert d.status == "rejected"
assert d.reason == "reconcile_warned"
# ---------------------------------------------------------------------------
# AC-9: in-flight (holding→pending_sell) transition not blocked
# ---------------------------------------------------------------------------
def test_ac9_in_flight_order_terminal_transition_not_blocked():
"""C-case warning does not block slot state convergence for in-flight orders."""
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C")
# Transition to PENDING_SELL (simulating in-flight cancel) — should not raise
pending_sell = svc.transit_slot(
slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL,
reason="sell", actor="test",
)
assert pending_sell.status == SlotState.PENDING_SELL
# Further transition to INACTIVE (fill/cancel reconciliation) — should not raise
terminal = svc.transit_slot(
pending_sell.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
reason="filled_sell", actor="test",
)
assert terminal.status == SlotState.INACTIVE
# ---------------------------------------------------------------------------
# AC-10: verify_slot_reconcile uses symbol aggregate for multi-portfolio clear
# ---------------------------------------------------------------------------
def test_ac10_verify_uses_symbol_aggregate_for_match():
"""verify uses symbol-level hub aggregate (not single portfolio qty) for comparison."""
from uuid import uuid4
from trex_core.domain.slot.models import Slot
repo = InMemorySlotRepository()
ws = InMemoryReconcileWarningStore()
# Two portfolios for same symbol
slot_a = Slot(slot_id=uuid4(), status=SlotState.HOLDING, symbol=SYM_A,
qty=200.0, portfolio_id="pf-A", instance_key=INSTANCE)
slot_b = Slot(slot_id=uuid4(), status=SlotState.HOLDING, symbol=SYM_A,
qty=240.0, portfolio_id="pf-B", instance_key=INSTANCE)
repo.add_slot(slot_a)
repo.add_slot(slot_b)
# Both warned
ws.set_warned(str(slot_a.slot_id), portfolio_id="pf-A", case_code="C")
ws.set_warned(str(slot_b.slot_id), portfolio_id="pf-B", case_code="C")
store = SnapshotStore()
store.swap([
PortfolioItem(portfolio_id="pf-A", symbol=SYM_A, cash_balance=0, asset_quantity=200.0),
PortfolioItem(portfolio_id="pf-B", symbol=SYM_A, cash_balance=0, asset_quantity=240.0),
])
broker = FakeBroker([BalanceItem(symbol=SYM_A, hldg_qty=440)]) # 200+240=440 == 440 ✓
result = verify_slot_reconcile(
portfolio_id="pf-A",
store=store,
broker=broker,
slot_repo=repo,
instance_key=INSTANCE,
warn_store=ws,
)
assert result["result"] == "cleared"
assert result["hub_qty"] == 440 # aggregate, not just pf-A's 200
# ---------------------------------------------------------------------------
# AC-11: verify clears ALL C-case warnings for the symbol
# ---------------------------------------------------------------------------
def test_ac11_verify_clears_all_symbol_c_case_warnings():
from uuid import uuid4
from trex_core.domain.slot.models import Slot
repo = InMemorySlotRepository()
ws = InMemoryReconcileWarningStore()
slot_a = Slot(slot_id=uuid4(), status=SlotState.HOLDING, symbol=SYM_A,
qty=200.0, portfolio_id="pf-A", instance_key=INSTANCE)
slot_b = Slot(slot_id=uuid4(), status=SlotState.HOLDING, symbol=SYM_A,
qty=240.0, portfolio_id="pf-B", instance_key=INSTANCE)
repo.add_slot(slot_a)
repo.add_slot(slot_b)
ws.set_warned(str(slot_a.slot_id), portfolio_id="pf-A", case_code="C")
ws.set_warned(str(slot_b.slot_id), portfolio_id="pf-B", case_code="C")
store = SnapshotStore()
store.swap([
PortfolioItem(portfolio_id="pf-A", symbol=SYM_A, cash_balance=0, asset_quantity=200.0),
PortfolioItem(portfolio_id="pf-B", symbol=SYM_A, cash_balance=0, asset_quantity=240.0),
])
broker = FakeBroker([BalanceItem(symbol=SYM_A, hldg_qty=440)])
verify_slot_reconcile(
portfolio_id="pf-A",
store=store,
broker=broker,
slot_repo=repo,
instance_key=INSTANCE,
warn_store=ws,
)
# Both pf-A and pf-B warnings cleared
assert not ws.is_warned(str(slot_a.slot_id))
assert not ws.is_warned(str(slot_b.slot_id))
def test_ac11_verify_mismatch_keeps_all_warnings():
"""When verify still mismatches, all warnings are kept."""
from uuid import uuid4
from trex_core.domain.slot.models import Slot
repo = InMemorySlotRepository()
ws = InMemoryReconcileWarningStore()
slot_a = Slot(slot_id=uuid4(), status=SlotState.HOLDING, symbol=SYM_A,
qty=200.0, portfolio_id="pf-A", instance_key=INSTANCE)
slot_b = Slot(slot_id=uuid4(), status=SlotState.HOLDING, symbol=SYM_A,
qty=235.0, portfolio_id="pf-B", instance_key=INSTANCE)
repo.add_slot(slot_a)
repo.add_slot(slot_b)
ws.set_warned(str(slot_a.slot_id), portfolio_id="pf-A", case_code="C")
ws.set_warned(str(slot_b.slot_id), portfolio_id="pf-B", case_code="C")
store = SnapshotStore()
store.swap([
PortfolioItem(portfolio_id="pf-A", symbol=SYM_A, cash_balance=0, asset_quantity=200.0),
PortfolioItem(portfolio_id="pf-B", symbol=SYM_A, cash_balance=0, asset_quantity=235.0),
])
broker = FakeBroker([BalanceItem(symbol=SYM_A, hldg_qty=440)]) # 435 != 440
result = verify_slot_reconcile(
portfolio_id="pf-A",
store=store,
broker=broker,
slot_repo=repo,
instance_key=INSTANCE,
warn_store=ws,
)
assert result["result"] == "rejected"
# Both warnings still present
assert ws.is_warned(str(slot_a.slot_id))
assert ws.is_warned(str(slot_b.slot_id))
# ---------------------------------------------------------------------------
# AC-12: C-case warning NOT auto-cleared by time; only cleared on quantity match
# ---------------------------------------------------------------------------
def test_ac12_c_case_warning_not_auto_cleared():
"""C-case warning persists across multiple slots RPC calls without manual verify."""
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C")
handler = _make_handler(repo, warn_store)
# Call slots RPC multiple times — slot stays excluded (no auto-clear)
for _ in range(3):
resp = handler(_req(filter="holding"))
assert resp.payload["slots"] == []
assert resp.payload["excluded_slots"]["reconcile_quantity_mismatch"] == 1
# Warning still present
assert warn_store.is_warned(str(slot.slot_id))
# ---------------------------------------------------------------------------
# Non-invasive: slots RPC does not mutate broker, ledger, or slot state
# ---------------------------------------------------------------------------
def test_non_invasive_slots_rpc_no_side_effects():
"""Slots RPC with C-case exclusion must not change slot status or broker state."""
repo = InMemorySlotRepository()
svc = SlotService(repo)
warn_store = InMemoryReconcileWarningStore()
slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10)
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C")
handler = _make_handler(repo, warn_store)
handler(_req())
# Slot still in HOLDING state (no transition)
refreshed = repo.fetch_slot(slot.slot_id)
assert refreshed.status == SlotState.HOLDING
assert float(refreshed.qty) == 10.0
# Warning still present (not cleared by RPC)
assert warn_store.is_warned(str(slot.slot_id))