Merge pull request #72 from fourblocks2020/copilot/bug-fix-portfolio-hub-quantity-discrepancy
AC-0023: Block C-case quantity mismatch slots from strategy RPC; fail-closed on warn-store failure
This commit is contained in:
@@ -7,7 +7,11 @@ status: active
|
||||
|
||||
## Criterion
|
||||
|
||||
`trex.slots.req.<instance>`의 `holding` 슬롯 `qty`는 요청한 instance에 결합된 KIS account broker holdings(`hldg_qty`)를 authoritative source로 사용해야 한다. `daily_slot.qty`와 portfolio-hub 집계값은 각각 `slot_quantity`, `asset_quantity` 진단 필드로 분리 유지한다. broker 수량을 확보하지 못했거나 freshness 허용시간을 초과하면 core/hub mirror fallback 없이 `qty=null`, `qty_source="unavailable"`, `broker_quantity_available=false`, `broker_quantity_captured_at=null`을 반환해 전략이 fail-closed하도록 해야 한다. confirmed broker zero는 `qty=0`, `qty_source="broker"`로 unavailable과 구분되어야 한다. 이 read path는 주문 submit/cancel, ledger write, slot lifecycle mutation을 유발하지 않아야 한다.
|
||||
`trex.slots.req.<instance>`의 **전략용 slots 응답에 포함되는** `holding` 슬롯(AC-0023에 따라 C-case 수량 불일치로 제외된 slot 제외)의 `qty`는 요청한 instance에 결합된 KIS account broker holdings(`hldg_qty`)를 authoritative source로 사용해야 한다. `daily_slot.qty`와 portfolio-hub 집계값은 각각 `slot_quantity`, `asset_quantity` 진단 필드로 분리 유지한다. broker 수량을 확보하지 못했거나 freshness 허용시간을 초과하면 core/hub mirror fallback 없이 `qty=null`, `qty_source="unavailable"`, `broker_quantity_available=false`, `broker_quantity_captured_at=null`을 반환해 전략이 fail-closed하도록 해야 한다. confirmed broker zero는 `qty=0`, `qty_source="broker"`로 unavailable과 구분되어야 한다. 이 read path는 주문 submit/cancel, ledger write, slot lifecycle mutation을 유발하지 않아야 한다.
|
||||
|
||||
## Scope note
|
||||
|
||||
AC-0023에 따라 C-case(`case_code=C`) 수량 불일치로 전략용 slots RPC에서 제외된 slot에는 이 AC의 broker-authoritative `qty` 응답 형식을 적용하지 않는다. AC-0023 제외가 AC-0022 broker-qty 형식보다 우선한다.
|
||||
|
||||
## Verification
|
||||
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
---
|
||||
id: "ac-0023"
|
||||
status: active
|
||||
---
|
||||
|
||||
# C-case quantity mismatch — strategy RPC exclusion and intent blocking
|
||||
|
||||
## Criterion
|
||||
|
||||
KIS 계좌 잔고(`hldg_qty`)와 같은 instance가 관리하는 동일 symbol의 portfolio-hub `asset_quantity` 합계가 일치하지 않으면(`case_code=C` reconcile-warning 활성), 해당 symbol의 모든 `holding` slot을 `trex.slots.req.<instance>` 응답에서 제외해야 한다. 수량 대조는 **instance_key + symbol 단위 합계**로 수행하며, 개별 portfolio 수량을 KIS 종목 합계와 직접 비교해서는 안 된다. 응답에는 `excluded_slots.reconcile_quantity_mismatch` 필드로 제외된 slot 수를 포함한다. warn-store를 읽지 못하면 경고 유무를 모르는 상태로 slot 목록을 반환하지 않고 `degraded=true` / `reason="warn_store_unavailable"`으로 응답해야 한다(fail-closed). 신규 buy/sell intent는 durable warning을 재확인하여 `reconcile_warned` 사유로 거부한다. C-case 경고 해제는 symbol 단위 합계가 다시 일치한 경우에만 수행하며, 해당 symbol에 걸린 모든 C-case 경고를 함께 해제한다. 이 경로는 주문 submit/cancel, ledger write, slot lifecycle mutation을 유발하지 않아야 한다.
|
||||
|
||||
## Verification
|
||||
|
||||
- 성공 조건: KIS 440주, hub portfolio A 200주 + B 240주(합계 440) → A·B 모두 반환되고 `excluded_slots.reconcile_quantity_mismatch=0`.
|
||||
- 불일치 조건: KIS 440주, hub A 200주 + B 235주(합계 435) → A·B 모두 제외되고 `excluded_slots.reconcile_quantity_mismatch=2`. 다른 symbol 정상 slot은 같은 응답에 포함된다.
|
||||
- 단일 portfolio 불일치: hub 단일 portfolio qty != KIS → 해당 symbol 전체 slot 제외.
|
||||
- fail-closed (slots RPC): warn-store가 `list_c_case_warned()` 중 예외를 throw하면 `slots=[], degraded=true, reason="warn_store_unavailable"` 반환.
|
||||
- fail-closed (intent): intent 처리 시 `warn_store.is_warned()` 호출이 실패하면 `reconcile_warning_unavailable` 사유로 거부해야 한다 — fail-open 허용 불가.
|
||||
- intent race: 마지막 slots RPC 이후 경고가 설정된 경우에도 intent 처리 시점의 durable warning을 재확인해 `reconcile_warned`로 거부.
|
||||
- 해제: symbol 합계가 다시 일치하면 해당 symbol의 모든 **C-case 경고만** 함께 해제. 같은 symbol의 B-case 등 다른 case 경고는 보존해야 한다.
|
||||
- 해제 대상: HOLDING 뿐 아니라 모든 상태의 slot에 걸린 C-case 경고를 포함한다.
|
||||
- 해제 실패: 해제 도중 예외가 발생하면 `result='clear_failed'`를 반환하고 완전 성공으로 보고하지 않아야 한다(부분 성공 = 실패).
|
||||
- 시간이 지났다는 이유만으로 자동 해제하지 않음.
|
||||
- 비침습 조건: slots RPC 호출 전후로 slot 상태 전이, broker mutation, ledger write side effect가 없어야 한다.
|
||||
- AC-0022 우선순위: C-case로 제외된 slot에는 broker-authoritative `qty` 응답 형식(AC-0022)을 적용하지 않는다.
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"schema_version": 1,
|
||||
"next_line_number": 7,
|
||||
"next_ac_number": 23
|
||||
"next_ac_number": 24
|
||||
}
|
||||
|
||||
@@ -598,3 +598,61 @@ def test_slot_to_dict_paused_and_warned_independent():
|
||||
row = slot_to_dict(slot, pause_store=pause_store, warn_store=warn_store)
|
||||
assert row["paused"] is True
|
||||
assert row["reconcile_warned"] is True
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# AC-0023 review: warn_store.is_warned() raises → intent fail-closed
|
||||
# ===========================================================================
|
||||
|
||||
class _RaisingWarnStore:
|
||||
"""Stub warn_store whose is_warned always raises (simulates DB unavailable)."""
|
||||
def is_warned(self, slot_id: str) -> bool:
|
||||
raise RuntimeError("DB unavailable")
|
||||
def list_c_case_warned(self) -> set:
|
||||
raise RuntimeError("DB unavailable")
|
||||
|
||||
|
||||
def test_buy_intent_fail_closed_when_warn_store_raises():
|
||||
"""If warn_store.is_warned() raises, buy intent must be rejected (fail-closed)."""
|
||||
repo = InMemorySlotRepository()
|
||||
svc = SlotService(repo)
|
||||
intent_store = InMemoryIntentStore()
|
||||
|
||||
repo.reconcile_universe("default", {"pf-A": "005930"})
|
||||
slot = repo.find_slot_by_portfolio("default", "pf-A")
|
||||
|
||||
intent_svc = IntentService(
|
||||
svc, repo, intent_store,
|
||||
instance_key="default",
|
||||
warn_store=_RaisingWarnStore(),
|
||||
)
|
||||
d = intent_svc.submit_buy(BuyIntent(
|
||||
intent_id="b-fc-001",
|
||||
slot_id=slot.slot_id,
|
||||
entry_price=100.0,
|
||||
qty=10.0,
|
||||
))
|
||||
assert d.status == "rejected"
|
||||
assert d.reason == "reconcile_warning_unavailable"
|
||||
|
||||
|
||||
def test_sell_intent_fail_closed_when_warn_store_raises():
|
||||
"""If warn_store.is_warned() raises, sell intent must be rejected (fail-closed)."""
|
||||
repo = InMemorySlotRepository()
|
||||
svc = SlotService(repo)
|
||||
intent_store = InMemoryIntentStore()
|
||||
|
||||
slot = _make_holding_slot(repo, svc, qty=10)
|
||||
|
||||
intent_svc = IntentService(
|
||||
svc, repo, intent_store,
|
||||
instance_key="default",
|
||||
warn_store=_RaisingWarnStore(),
|
||||
)
|
||||
d = intent_svc.submit_sell(SellIntent(
|
||||
intent_id="s-fc-001",
|
||||
slot_id=slot.slot_id,
|
||||
reason="strategy_exit",
|
||||
))
|
||||
assert d.status == "rejected"
|
||||
assert d.reason == "reconcile_warning_unavailable"
|
||||
|
||||
@@ -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 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)
|
||||
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)
|
||||
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))
|
||||
@@ -265,3 +265,140 @@ def test_verify_empty_snapshot_treats_hub_qty_as_zero():
|
||||
assert result["result"] == "rejected"
|
||||
assert result["hub_qty"] == 0
|
||||
assert result["broker_qty"] == 10
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-0023 comment 3853379970: B-case warning preserved; non-holding C-case cleared
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_verify_c_case_clear_preserves_b_case_warning():
|
||||
"""B-case warning for same symbol must NOT be cleared by a C-case verify clear."""
|
||||
sym = "005930"
|
||||
# Slot A: HOLDING, C-case warning (should be cleared)
|
||||
slot_c = _make_slot(status=SlotState.HOLDING, symbol=sym, pid="pf-c")
|
||||
# Slot B: HOLDING, B-case warning (must be preserved)
|
||||
slot_b = _make_slot(status=SlotState.HOLDING, symbol=sym, pid="pf-b")
|
||||
|
||||
repo = InMemorySlotRepository()
|
||||
repo.add_slot(slot_c)
|
||||
repo.add_slot(slot_b)
|
||||
|
||||
ws = InMemoryReconcileWarningStore()
|
||||
ws.set_warned(str(slot_c.slot_id), portfolio_id="pf-c", case_code="C")
|
||||
ws.set_warned(str(slot_b.slot_id), portfolio_id="pf-b", case_code="B")
|
||||
|
||||
# Aggregate hub qty across both portfolios = 10
|
||||
store = SnapshotStore()
|
||||
store.swap([
|
||||
PortfolioItem(portfolio_id="pf-c", symbol=sym, cash_balance=0.0, asset_quantity=5.0),
|
||||
PortfolioItem(portfolio_id="pf-b", symbol=sym, cash_balance=0.0, asset_quantity=5.0),
|
||||
])
|
||||
broker = FakeBroker({sym: 10})
|
||||
|
||||
result = verify_slot_reconcile(
|
||||
portfolio_id="pf-c",
|
||||
store=store,
|
||||
broker=broker,
|
||||
slot_repo=repo,
|
||||
instance_key=INSTANCE,
|
||||
warn_store=ws,
|
||||
)
|
||||
|
||||
assert result["result"] == "cleared"
|
||||
# C-case slot cleared
|
||||
assert not ws.is_warned(str(slot_c.slot_id))
|
||||
# B-case slot untouched
|
||||
assert ws.is_warned(str(slot_b.slot_id))
|
||||
b_info = ws.get_warning_info(str(slot_b.slot_id))
|
||||
assert b_info is not None and b_info["case_code"] == "B"
|
||||
|
||||
|
||||
def test_verify_c_case_clear_non_holding_slot():
|
||||
"""A non-HOLDING slot with C-case warning for the same symbol must also be cleared."""
|
||||
sym = "005930"
|
||||
slot_holding = _make_slot(status=SlotState.HOLDING, symbol=sym, pid="pf-hold")
|
||||
# Simulate a slot that is INACTIVE (e.g. after sell filled) but still has a C-case warning
|
||||
slot_inactive = _make_slot(status=SlotState.INACTIVE, symbol=sym, pid="pf-inact")
|
||||
|
||||
repo = InMemorySlotRepository()
|
||||
repo.add_slot(slot_holding)
|
||||
repo.add_slot(slot_inactive)
|
||||
|
||||
ws = InMemoryReconcileWarningStore()
|
||||
ws.set_warned(str(slot_holding.slot_id), portfolio_id="pf-hold", case_code="C")
|
||||
ws.set_warned(str(slot_inactive.slot_id), portfolio_id="pf-inact", case_code="C")
|
||||
|
||||
store = SnapshotStore()
|
||||
store.swap([
|
||||
PortfolioItem(portfolio_id="pf-hold", symbol=sym, cash_balance=0.0, asset_quantity=10.0),
|
||||
])
|
||||
broker = FakeBroker({sym: 10})
|
||||
|
||||
result = verify_slot_reconcile(
|
||||
portfolio_id="pf-hold",
|
||||
store=store,
|
||||
broker=broker,
|
||||
slot_repo=repo,
|
||||
instance_key=INSTANCE,
|
||||
warn_store=ws,
|
||||
)
|
||||
|
||||
assert result["result"] == "cleared"
|
||||
assert not ws.is_warned(str(slot_holding.slot_id))
|
||||
assert not ws.is_warned(str(slot_inactive.slot_id))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-0023 comment 3853379979: partial clear failure → clear_failed, not cleared
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class _FailAfterNStore(InMemoryReconcileWarningStore):
|
||||
"""Raises on the Nth+ call to clear_warned (1-indexed)."""
|
||||
|
||||
def __init__(self, fail_after: int = 1) -> None:
|
||||
super().__init__()
|
||||
self._calls = 0
|
||||
self._fail_after = fail_after
|
||||
|
||||
def clear_warned(self, slot_id: str) -> None:
|
||||
self._calls += 1
|
||||
if self._calls > self._fail_after:
|
||||
raise RuntimeError("simulated DB error")
|
||||
super().clear_warned(slot_id)
|
||||
|
||||
|
||||
def test_verify_partial_clear_failure_returns_clear_failed():
|
||||
"""If clear_warned raises mid-loop, result must be 'clear_failed', not 'cleared'."""
|
||||
sym = "005930"
|
||||
slot1 = _make_slot(status=SlotState.HOLDING, symbol=sym, pid="pf-1")
|
||||
slot2 = _make_slot(status=SlotState.HOLDING, symbol=sym, pid="pf-2")
|
||||
|
||||
repo = InMemorySlotRepository()
|
||||
repo.add_slot(slot1)
|
||||
repo.add_slot(slot2)
|
||||
|
||||
# Fails on 2nd clear_warned call
|
||||
ws = _FailAfterNStore(fail_after=1)
|
||||
ws.set_warned(str(slot1.slot_id), portfolio_id="pf-1", case_code="C")
|
||||
ws.set_warned(str(slot2.slot_id), portfolio_id="pf-2", case_code="C")
|
||||
|
||||
store = SnapshotStore()
|
||||
store.swap([
|
||||
PortfolioItem(portfolio_id="pf-1", symbol=sym, cash_balance=0.0, asset_quantity=10.0),
|
||||
PortfolioItem(portfolio_id="pf-2", symbol=sym, cash_balance=0.0, asset_quantity=0.0),
|
||||
])
|
||||
broker = FakeBroker({sym: 10})
|
||||
|
||||
result = verify_slot_reconcile(
|
||||
portfolio_id="pf-1",
|
||||
store=store,
|
||||
broker=broker,
|
||||
slot_repo=repo,
|
||||
instance_key=INSTANCE,
|
||||
warn_store=ws,
|
||||
)
|
||||
|
||||
assert result["result"] == "clear_failed"
|
||||
assert result["hub_qty"] == 10
|
||||
assert result["broker_qty"] == 10
|
||||
assert result["reason"] is not None and "warn_store" in result["reason"]
|
||||
|
||||
@@ -68,6 +68,16 @@ class PostgreSQLReconcileWarningStore:
|
||||
cur.execute("SELECT slot_id FROM slot_reconcile_warning")
|
||||
return {row[0] for row in cur.fetchall()}
|
||||
|
||||
def list_c_case_warned(self) -> set[str]:
|
||||
"""Return slot_id strings with an active case_code='C' warning (AC-0023)."""
|
||||
with self._psycopg.connect(self._dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT slot_id FROM slot_reconcile_warning WHERE case_code = %s",
|
||||
("C",),
|
||||
)
|
||||
return {row[0] for row in cur.fetchall()}
|
||||
|
||||
def get_warning_info(self, slot_id: str) -> dict | None:
|
||||
"""Return {"portfolio_id", "case_code", "actor", "reason"} or None (REQ-105/106)."""
|
||||
with self._psycopg.connect(self._dsn) as conn:
|
||||
|
||||
@@ -271,8 +271,10 @@ class IntentService:
|
||||
return IntentDecision(
|
||||
intent.intent_id, "rejected", None, "reconcile_warned"
|
||||
)
|
||||
except Exception: # noqa: BLE001 - fail-open
|
||||
pass
|
||||
except Exception: # noqa: BLE001 - fail-closed: warn-store down → reject
|
||||
return IntentDecision(
|
||||
intent.intent_id, "rejected", None, "reconcile_warning_unavailable"
|
||||
)
|
||||
|
||||
# dcy-021: a buy on a HOLDING slot is a position-sizing increase.
|
||||
is_increase = slot.status is SlotState.HOLDING
|
||||
@@ -362,8 +364,10 @@ class IntentService:
|
||||
return IntentDecision(
|
||||
intent.intent_id, "rejected", None, "reconcile_warned"
|
||||
)
|
||||
except Exception: # noqa: BLE001 - fail-open
|
||||
pass
|
||||
except Exception: # noqa: BLE001 - fail-closed: warn-store down → reject
|
||||
return IntentDecision(
|
||||
intent.intent_id, "rejected", None, "reconcile_warning_unavailable"
|
||||
)
|
||||
|
||||
if slot.status is not SlotState.HOLDING:
|
||||
return IntentDecision(
|
||||
|
||||
@@ -347,13 +347,53 @@ def make_slots_rpc_handler(
|
||||
if strategy_filter is not None:
|
||||
slot_rows = [r for r in slot_rows if r.get("strategy_name") == strategy_filter]
|
||||
|
||||
return RpcResponse(payload={
|
||||
# AC-0023: exclude holding slots with active case_code='C' reconcile warning.
|
||||
# Fail-closed: if warn_store exists but throws, return degraded=True / no slots.
|
||||
excluded_count = 0
|
||||
if warn_store is not None:
|
||||
try:
|
||||
c_warned_ids = warn_store.list_c_case_warned()
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception("warn_store.list_c_case_warned() failed — fail-closed")
|
||||
return RpcResponse(payload={
|
||||
"slots": [],
|
||||
"available_slots": available_slots,
|
||||
"degraded": True,
|
||||
"reason": "warn_store_unavailable",
|
||||
"message": (
|
||||
"reconcile warning store unavailable; "
|
||||
"returning no slots (fail-closed)"
|
||||
),
|
||||
"excluded_slots": {"reconcile_quantity_mismatch": 0},
|
||||
})
|
||||
if c_warned_ids:
|
||||
# Rebuild slot lookup from the original `slots` list (before strategy
|
||||
# filter reduced slot_rows). slot_rows[i] corresponds to slots[i]
|
||||
# only before filtering; use slot_id stored in each row.
|
||||
slot_by_id = {str(s.slot_id): s for s in slots}
|
||||
filtered_rows: list[dict] = []
|
||||
for row in slot_rows:
|
||||
slot_obj = slot_by_id.get(str(row.get("slot_id", "")))
|
||||
if (
|
||||
slot_obj is not None
|
||||
and str(slot_obj.slot_id) in c_warned_ids
|
||||
and slot_obj.status.value == "holding"
|
||||
):
|
||||
excluded_count += 1
|
||||
else:
|
||||
filtered_rows.append(row)
|
||||
slot_rows = filtered_rows
|
||||
|
||||
base_payload: dict = {
|
||||
"slots": slot_rows,
|
||||
"available_slots": available_slots,
|
||||
"degraded": degraded,
|
||||
"reason": reason,
|
||||
"message": message,
|
||||
})
|
||||
}
|
||||
if warn_store is not None:
|
||||
base_payload["excluded_slots"] = {"reconcile_quantity_mismatch": excluded_count}
|
||||
return RpcResponse(payload=base_payload)
|
||||
|
||||
return handler
|
||||
|
||||
|
||||
@@ -391,6 +391,13 @@ class ReconcileWarningStore(Protocol):
|
||||
"""
|
||||
...
|
||||
|
||||
def list_c_case_warned(self) -> "set[str]":
|
||||
"""Return set of slot_id strings that have an active case_code='C' warning.
|
||||
|
||||
AC-0023: used by strategy slots RPC to exclude C-case mismatched holding slots.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class InMemoryReconcileWarningStore:
|
||||
"""Process-local reconcile-warning store for tests + CI (REQ-102)."""
|
||||
@@ -434,6 +441,10 @@ class InMemoryReconcileWarningStore:
|
||||
info = self._warned.get(slot_id)
|
||||
return dict(info) if info is not None else None
|
||||
|
||||
def list_c_case_warned(self) -> "set[str]":
|
||||
with self._lock:
|
||||
return {sid for sid, info in self._warned.items() if info.get("case_code") == "C"}
|
||||
|
||||
|
||||
def slot_to_dict(
|
||||
s,
|
||||
|
||||
@@ -648,31 +648,51 @@ def verify_slot_reconcile(
|
||||
}
|
||||
|
||||
# Re-run the C-case check: hub qty vs broker qty for this symbol.
|
||||
# AC-0023: use symbol-level aggregate across all portfolios (not single portfolio).
|
||||
sym = slot.symbol or ""
|
||||
broker_qty_map: dict[str, int] = {
|
||||
item.symbol: int(item.hldg_qty) for item in broker.get_balance()
|
||||
}
|
||||
broker_qty = broker_qty_map.get(sym, 0)
|
||||
|
||||
# Hub qty: sum asset_quantity for this portfolio_id from snapshot.
|
||||
# Hub qty: sum asset_quantity for ALL portfolios with this symbol in the snapshot.
|
||||
hub_qty: int = 0
|
||||
snap = store.get(portfolio_id)
|
||||
if snap is not None:
|
||||
hub_qty = int(snap.item.asset_quantity)
|
||||
for a in store.list_assets():
|
||||
if a.item.symbol == sym:
|
||||
hub_qty += int(a.item.asset_quantity)
|
||||
|
||||
match = hub_qty == broker_qty
|
||||
|
||||
if match:
|
||||
# AC-0023: clear C-case warnings for ALL slots (any status) with this symbol.
|
||||
# Only rows with case_code='C' are touched; B-case (or other) rows are preserved.
|
||||
cleared_slots: list[str] = []
|
||||
try:
|
||||
warn_store.clear_warned(slot_id_str)
|
||||
store.set_warning(portfolio_id, False)
|
||||
for s in slot_repo.list_slots(instance_key):
|
||||
if s.symbol != sym:
|
||||
continue
|
||||
sid_str = str(s.slot_id)
|
||||
warn_info = warn_store.get_warning_info(sid_str)
|
||||
if warn_info is None or warn_info.get("case_code") != "C":
|
||||
continue
|
||||
warn_store.clear_warned(sid_str)
|
||||
store.set_warning(s.portfolio_id, False)
|
||||
cleared_slots.append(sid_str)
|
||||
_log.info(
|
||||
"verify: reconcile-warning cleared (C-case resolved) "
|
||||
"slot=%s portfolio_id=%s symbol=%s hub_qty=%d broker_qty=%d",
|
||||
slot_id_str, portfolio_id, sym, hub_qty, broker_qty,
|
||||
"symbol=%s hub_qty=%d broker_qty=%d cleared_slots=%s",
|
||||
sym, hub_qty, broker_qty, cleared_slots,
|
||||
)
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception("verify: failed to clear_warned for slot=%s", slot_id_str)
|
||||
_log.exception("verify: failed to clear C-case warned for symbol=%s", sym)
|
||||
return {
|
||||
"portfolio_id": portfolio_id,
|
||||
"slot_id": slot_id_str,
|
||||
"result": "clear_failed",
|
||||
"hub_qty": hub_qty,
|
||||
"broker_qty": broker_qty,
|
||||
"reason": "warn_store error during C-case clear — warnings NOT fully cleared",
|
||||
}
|
||||
return {
|
||||
"portfolio_id": portfolio_id,
|
||||
"slot_id": slot_id_str,
|
||||
|
||||
Reference in New Issue
Block a user