trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core-A): reconcile-warning durable store + blocking gate (BAT-050, REQ-102/103/104/107/108)

Browse Source 
B/C 불일치 시 슬롯을 durable 차단(ReconcileWarningStore)하고 전략 intent 경로에
pause∨reconcile_warned 게이트를 추가한다. 안전망(safety-sell/kill-switch/watchdog)
경로는 SlotService 직접 호출로 게이트를 구조적으로 우회(OQ-B 확정).
SlotState enum/전이표/wire 계약 필드는 불변, reconcile_warned 필드만 additive 추가.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5
This commit is contained in:
genichin
2026-06-26 00:43:04 +09:00
co-authored by Claude Sonnet 4.6
parent 35f3147250
commit c193388198
17 changed files with 1410 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trex_core/application/intents.py
+52
View File
@@ -14,6 +14,11 @@ dcy-020 (BAT-030): admission is asset-centric only — the legacy pool path
cutover. Buy is keyed by slot_id (REQ-061 canonical; portfolio_id is the
legacy dual-accept mapping) and admits FCFS under the `max_active_slots` cap
M when configured (no cap when unset).
BAT-050 / REQ-104: reconcile-warning gate added to _admit_buy/_admit_sell.
warn_store is Optional — when None, gate is skipped (backward compat).
Safety-sell / kill-switch / watchdog bypass this gate by design (OQ-B):
those paths call SlotService.sell() directly, not through IntentService.
"""
from __future__ import annotations
@@ -104,6 +109,7 @@ class IntentService:
instance_key: str = "default",
max_active_slots: Optional[int] = None,
default_order_timeout_sec: Optional[int] = None,
warn_store=None, # Optional[ReconcileWarningStore] — BAT-050 / REQ-104
) -> None:
self._svc = slot_service
self._repo = slot_repo
@@ -119,6 +125,9 @@ class IntentService:
# dcy-022 / REQ-068: default timeout when intent omits timeout_sec.
# None = OFF = indefinite wait (legacy behavior). Per-intent value overrides.
self._default_order_timeout_sec = default_order_timeout_sec
# BAT-050 / REQ-104: reconcile-warning gate store (Optional).
# When None, gate is skipped (backward compat, NFR-4).
self._warn_store = warn_store
# ------------------------------------------------------------
def lookup(self, intent_id: str) -> Optional[IntentDecision]:
@@ -176,6 +185,23 @@ class IntentService:
return IntentDecision(intent_id, "rejected", None, "duplicate_in_flight")
return IntentDecision(intent_id, rec.status, rec.slot_id, rec.reason)
def _check_reconcile_warned(self, slot_id) -> Optional[IntentDecision]:
"""BAT-050 / REQ-104: return rejected decision if slot is reconcile-warned.
Returns None if not warned (gate passes). The warn_store check is best-effort:
if the store raises, we allow the intent through (fail-open for resilience).
Safety-sell / kill-switch / watchdog bypass this gate — they call
SlotService.sell() directly without going through IntentService (OQ-B).
"""
if self._warn_store is None:
return None
try:
if self._warn_store.is_warned(str(slot_id)):
return None # placeholder — intent_id unknown here; caller provides it
except Exception: # noqa: BLE001 - fail-open
pass
return None # not warned (or store error — fail-open)
def _admit_buy(self, intent: BuyIntent) -> IntentDecision:
"""dcy-020 / REQ-059·REQ-061 + dcy-021/REQ-066: activate or size up the
asset's permanent slot.
@@ -207,6 +233,19 @@ class IntentService:
return IntentDecision(intent.intent_id, "rejected", None, "slot_id_or_portfolio_required")
if slot is None:
return IntentDecision(intent.intent_id, "rejected", None, miss_reason)
# BAT-050 / REQ-104: reconcile-warning gate (external strategy intent only).
# Safety-sell / kill-switch / watchdog bypass this gate by design (OQ-B):
# those paths call SlotService.sell() directly.
if self._warn_store is not None:
try:
if self._warn_store.is_warned(str(slot.slot_id)):
return IntentDecision(
intent.intent_id, "rejected", None, "reconcile_warned"
)
except Exception: # noqa: BLE001 - fail-open
pass
# dcy-021: a buy on a HOLDING slot is a position-sizing increase.
is_increase = slot.status is SlotState.HOLDING
if slot.status in (SlotState.PENDING_BUY, SlotState.PENDING_SELL):
@@ -260,6 +299,19 @@ class IntentService:
slot = self._repo.fetch_slot(intent.slot_id)
if slot is None:
return IntentDecision(intent.intent_id, "rejected", None, "slot_not_found")
# BAT-050 / REQ-104: reconcile-warning gate for external strategy sell intent.
# Internal safety-sell / kill-switch / watchdog bypass this gate: they call
# SlotService.sell() / transit_slot() directly without IntentService (OQ-B).
if self._warn_store is not None:
try:
if self._warn_store.is_warned(str(slot.slot_id)):
return IntentDecision(
intent.intent_id, "rejected", None, "reconcile_warned"
)
except Exception: # noqa: BLE001 - fail-open
pass
if slot.status is not SlotState.HOLDING:
return IntentDecision(
intent.intent_id, "rejected", None, f"slot_not_holding:{slot.status.value}"
trex_core/application/messaging_handlers.py
+4
View File
@@ -16,6 +16,7 @@ from trex_core.application.intents import BuyIntent, IntentService, SellIntent
from trex_core.application.messaging import RpcRequest, RpcResponse
from trex_core.application.portfolio import (
PauseStore,
ReconcileWarningStore,
SnapshotStore,
slot_to_dict,
)
@@ -220,6 +221,7 @@ def make_slots_rpc_handler(
instance_key: str = "default",
snapshot_store: Optional[SnapshotStore] = None,
pause_store: Optional[PauseStore] = None,
warn_store: Optional[ReconcileWarningStore] = None,
watchdog_store: Optional[WatchdogStore] = None,
watchdog_timeout_sec: int = 120,
ledger_outbox=None,
@@ -261,12 +263,14 @@ def make_slots_rpc_handler(
watchdog_store.update(str(s.slot_id))
# Build merged slot rows (REQ-074)
# BAT-050 / REQ-104: warn_store passed to slot_to_dict for reconcile_warned field.
slot_rows = [
slot_to_dict(
s,
snapshot_store=snapshot_store,
watchdog_store=watchdog_store,
pause_store=pause_store,
warn_store=warn_store,
watchdog_timeout_sec=watchdog_timeout_sec,
)
for s in slots
trex_core/application/portfolio.py
+84 -5
View File
@@ -5,10 +5,13 @@ snapshot store + asset-selection service. Domain stays free of HTTP and
RabbitMQ deps (NFR-1).
The snapshot is split between:
- in-memory `SnapshotStore` universe + cash/position + warnings (process-local;
warning is automatic/temporary, reset on restart)
- durable `PauseStore` portfolio_id -> paused (Postgres; manual/persistent,
survives restart and daily refresh — REQ-032)
- in-memory `SnapshotStore` universe + cash/position + warnings (process-local;
warning is automatic/temporary, reset on restart)
- durable `PauseStore` portfolio_id -> paused (Postgres; manual/persistent,
survives restart and daily refresh — REQ-032)
- durable `ReconcileWarningStore` slot_id -> reconcile-warning (Postgres; automatic,
set by reconcile B/C cases, cleared by sync/verify
— REQ-102, dcy-034)
"""
from __future__ import annotations
@@ -246,6 +249,73 @@ class InMemoryKillSwitchStore:
self._excluded.discard((portfolio_id, trading_date))
# ---------------------------------------------------------------- reconcile-warning store (REQ-102)
# BAT-050 / dcy-034: durable reconcile-warning channel — distinct from pause (manual).
# Keyed by slot_id (TEXT UUID string). case_code: 'B' (slot↔broker qty mismatch)
# or 'C' (ledger↔broker mismatch). Survives restart (PostgreSQL-backed in production).
class ReconcileWarningStore(Protocol):
"""Durable reconcile-warning overlay — slot_id PK (REQ-102, dcy-034)."""
def set_warned(
self,
slot_id: str,
*,
portfolio_id: str,
case_code: str,
actor: str = "reconcile",
reason: str = "",
) -> None:
...
def clear_warned(self, slot_id: str) -> None:
...
def is_warned(self, slot_id: str) -> bool:
...
def list_warned(self) -> "set[str]":
...
class InMemoryReconcileWarningStore:
"""Process-local reconcile-warning store for tests + CI (REQ-102)."""
def __init__(self) -> None:
self._lock = threading.Lock()
# slot_id -> {"portfolio_id", "case_code", "actor", "reason"}
self._warned: dict[str, dict] = {}
def set_warned(
self,
slot_id: str,
*,
portfolio_id: str,
case_code: str,
actor: str = "reconcile",
reason: str = "",
) -> None:
with self._lock:
self._warned[slot_id] = {
"portfolio_id": portfolio_id,
"case_code": case_code,
"actor": actor,
"reason": reason,
}
def clear_warned(self, slot_id: str) -> None:
with self._lock:
self._warned.pop(slot_id, None)
def is_warned(self, slot_id: str) -> bool:
with self._lock:
return slot_id in self._warned
def list_warned(self) -> "set[str]":
with self._lock:
return set(self._warned.keys())
def slot_to_dict(
s,
@@ -253,6 +323,7 @@ def slot_to_dict(
snapshot_store: "Optional[SnapshotStore]" = None,
watchdog_store=None,
pause_store: "Optional[PauseStore]" = None,
warn_store: "Optional[ReconcileWarningStore]" = None,
watchdog_timeout_sec: int = 120,
) -> dict:
"""Merged slot row for trex.slots.req (REQ-073/074).
@@ -260,7 +331,7 @@ def slot_to_dict(
Carries slot/core state + hub asset data + watchdog + overlay flags.
Field-level authority: slot/core = slot_id/status/qty/avg_price/entry_price/
forced_close_at; hub = asset_quantity/cash_balance; watchdog = watchdog{};
overlay = paused/warned. strategy fields excluded (REQ-074 AC-5).
overlay = paused/warned/reconcile_warned. strategy fields excluded (REQ-074 AC-5).
"""
d = {
"slot_id": str(s.slot_id),
@@ -312,6 +383,14 @@ def slot_to_dict(
paused = s.portfolio_id in pause_store.list_paused()
d["paused"] = paused
# BAT-050 / REQ-104: reconcile_warned overlay (ReconcileWarningStore) — additive field.
# Bulk list_warned() called once per RPC (caller pre-fetches; see make_slots_rpc_handler).
# warn_store=None → reconcile_warned=False (backward compatible, NFR-4).
reconcile_warned = False
if warn_store is not None:
reconcile_warned = str(s.slot_id) in warn_store.list_warned()
d["reconcile_warned"] = reconcile_warned
return d
trex_core/application/reconciliation.py
+81 -13
View File
@@ -1,10 +1,15 @@
"""Position reconciliation between portfolio-hub and broker (REQ-031).
"""Position reconciliation between portfolio-hub and broker (REQ-031, REQ-103).
The broker account is single (invariant B) so portfolio-hub quantities are
aggregated to `symbol` before being compared with `BrokerPort.get_balance()`.
Mismatches flip the asset's `warning` overlay; a subsequent successful match
clears it automatically (in-memory only — pause is the durable channel for
manual holds).
BAT-050 / REQ-103: B-case (slot qty mismatch) and C-case (ledger mismatch)
now also set a durable reconcile-warning via ReconcileWarningStore (slot_id
keyed). The in-memory warning overlay is kept for backward compat; the durable
store is additive and Optional (None => warn-log-only, backward compat).
"""
from __future__ import annotations
@@ -13,7 +18,7 @@ from typing import Optional
from trex_core.application.alarm_emitter import AlarmEmitter
from trex_core.application.broker_port import BrokerPort
from trex_core.application.portfolio import SnapshotStore
from trex_core.application.portfolio import ReconcileWarningStore, SnapshotStore
from trex_core.domain.slot.state import SlotState
_log = logging.getLogger(__name__)
@@ -26,16 +31,24 @@ def reconcile_core_holdings(
broker: BrokerPort,
instance_key: str,
alarm_emitter: Optional[AlarmEmitter] = None,
warn_store: Optional[ReconcileWarningStore] = None,
) -> dict[str, str]:
"""dcy-020 / REQ-062 (BAT-029): surface core-slot holding ↔ broker divergence.
"""dcy-020 / REQ-062 (BAT-029) + REQ-103 (BAT-050): surface core-slot holding
vs broker divergence and set durable reconcile-warning for B case.
Broker balance is the reconcile ground truth. Decisions (no silent auto-fix,
no auto-liquidation — issue #15):
- A. core holding but broker flat (qty 0) -> slot to `error` + WARN (operator resolve)
- C. both hold but qty differs -> WARN (mirror is non-authoritative)
- A. core holding but broker flat (qty 0) -> slot to `error` + WARN + CRITICAL alarm
- C. both hold but qty differs (B in REQ-103 naming) -> WARN + durable reconcile-warning
- B. broker holds, no core holding slot -> WARN (unmanaged position)
REQ-103 case codes:
- case_code='B': slot has holding but qty differs from broker (reconcile_core_holdings)
- case_code='C': ledger/hub qty differs from broker (reconcile_positions)
warn_store: Optional — when None, WARN-log-only (backward compat, REQ-103 AC-5).
Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_warning'} for telemetry.
(Liquidation qty = broker authoritative in execution dispatch is BAT-δ.)"""
"""
broker_qty: dict[str, int] = {
item.symbol: int(item.hldg_qty) for item in broker.get_balance()
}
@@ -65,15 +78,32 @@ def reconcile_core_holdings(
)
except Exception: # noqa: BLE001 - never let reconcile crash the loop
_log.exception("failed to error-quarantine slot %s", slot.slot_id)
elif slot.qty is not None and int(slot.qty) != bq: # C — qty mismatch
decisions[str(slot.slot_id)] = "qty_warning"
elif slot.qty is not None and int(slot.qty) != bq: # B case (REQ-103) — qty mismatch
slot_id_str = str(slot.slot_id)
decisions[slot_id_str] = "qty_warning"
# REQ-107: structured WARNING log for B case.
_log.warning(
"qty mismatch: slot=%s symbol=%s core=%s broker=%d (mirror non-authoritative)",
slot.slot_id, sym, slot.qty, bq,
"reconcile_warning set: slot_id=%s portfolio_id=%s symbol=%s "
"case_code=B broker_qty=%d slot_qty=%s actor=reconcile",
slot_id_str, slot.portfolio_id, sym, bq, slot.qty,
)
# REQ-103 / BAT-050: durable reconcile-warning (case_code='B').
if warn_store is not None:
try:
warn_store.set_warned(
slot_id_str,
portfolio_id=slot.portfolio_id or "",
case_code="B",
actor="reconcile",
reason=f"slot_qty={slot.qty} broker_qty={bq} symbol={sym}",
)
except Exception: # noqa: BLE001 - never let reconcile crash the loop
_log.exception(
"failed to set reconcile_warning for slot %s", slot.slot_id
)
else:
decisions[str(slot.slot_id)] = "ok"
for sym, bq in broker_qty.items(): # B — unmanaged broker position
for sym, bq in broker_qty.items(): # unmanaged broker position
if bq > 0 and sym not in matched:
decisions[f"broker:{sym}"] = "unmanaged_warning"
_log.warning(
@@ -83,11 +113,23 @@ def reconcile_core_holdings(
return decisions
def reconcile_positions(*, store: SnapshotStore, broker: BrokerPort) -> dict[str, str]:
"""Run one reconciliation pass.
def reconcile_positions(
*,
store: SnapshotStore,
broker: BrokerPort,
slot_repo=None,
instance_key: str = "default",
warn_store: Optional[ReconcileWarningStore] = None,
) -> dict[str, str]:
"""Run one reconciliation pass (REQ-031 + REQ-103 BAT-050).
Returns a `{portfolio_id: 'warning' | 'cleared' | 'ok'}` map for telemetry.
The map is also reflected onto `store` via `set_warning(...)`.
BAT-050 / REQ-103: when warn_store and slot_repo are provided, C-case mismatches
(ledger/hub qty vs broker) also set durable reconcile-warning on the corresponding
slot (portfolio_id -> slot lookup via slot_repo). When either is None, falls back
to in-memory warning only (backward compat, REQ-103 AC-5).
"""
broker_qty: dict[str, int] = {
item.symbol: int(item.hldg_qty) for item in broker.get_balance()
@@ -117,6 +159,32 @@ def reconcile_positions(*, store: SnapshotStore, broker: BrokerPort) -> dict[str
)
store.set_warning(pid, True)
decisions[pid] = "warning"
# REQ-103 / BAT-050: durable reconcile-warning for C case (case_code='C').
if warn_store is not None and slot_repo is not None:
try:
slot = slot_repo.find_slot_by_portfolio(instance_key, pid)
if slot is not None:
slot_id_str = str(slot.slot_id)
# REQ-107: structured WARNING log for C case.
_log.warning(
"reconcile_warning set: slot_id=%s portfolio_id=%s "
"symbol=%s case_code=C hub_qty=%d broker_qty=%d actor=reconcile",
slot_id_str, pid, symbol, hub_total, broker_total,
)
warn_store.set_warned(
slot_id_str,
portfolio_id=pid,
case_code="C",
actor="reconcile",
reason=(
f"hub_qty={hub_total} broker_qty={broker_total} "
f"symbol={symbol}"
),
)
except Exception: # noqa: BLE001
_log.exception(
"failed to set reconcile_warning for portfolio_id=%s", pid
)
else:
if was_warned:
store.set_warning(pid, False)
trex_core/application/snapshot_scheduler.py
+16 -1
View File
@@ -4,6 +4,8 @@ Wakes at the configured KST time, fetches universe from PortfolioHubReadPort,
swaps the SnapshotStore, then runs reconciliation. Option D `*_after` writes
(REQ-030 AC-2/AC-3) are wired through `SnapshotStore.update_cash_position`
but disabled at the ledger layer until portfolio-hub#8 ships.
BAT-050 / REQ-103: reconcile now accepts warn_store for durable reconcile-warning.
"""
from __future__ import annotations
@@ -18,6 +20,7 @@ from trex_core.application.portfolio import (
DegradedReason,
PortfolioHubReadPort,
PortfolioHubUnavailableError,
ReconcileWarningStore,
SnapshotStore,
)
from trex_core.application.reconciliation import reconcile_positions
@@ -42,6 +45,7 @@ class SnapshotScheduler:
slot_service=None,
instance_key: str = "default",
alarm_emitter: Optional[AlarmEmitter] = None,
warn_store: Optional[ReconcileWarningStore] = None,
) -> None:
self._read = read_port
self._store = store
@@ -59,6 +63,8 @@ class SnapshotScheduler:
self._last_universe_size: Optional[int] = None
# dcy-033 / REQ-100: best-effort alarm emitter.
self._alarm_emitter = alarm_emitter
# BAT-050 / REQ-103: durable reconcile-warning store (Optional).
self._warn_store = warn_store
# ---- callable from a daemon thread loop --------------------------------
def run_forever(self, stop: threading.Event, *, tick_sec: float = 60.0) -> None:
@@ -124,7 +130,14 @@ class SnapshotScheduler:
self._store.swap(items)
try:
reconcile_positions(store=self._store, broker=self._broker)
# BAT-050 / REQ-103: pass warn_store + slot_repo for C-case durable warning.
reconcile_positions(
store=self._store,
broker=self._broker,
slot_repo=self._slot_repo,
instance_key=self._instance_key,
warn_store=self._warn_store,
)
except Exception: # noqa: BLE001
_log.exception("reconciliation pass failed")
if self._slot_repo is not None:
@@ -135,10 +148,12 @@ class SnapshotScheduler:
if self._slot_service is not None:
try:
from trex_core.application.reconciliation import reconcile_core_holdings
# BAT-050 / REQ-103: pass warn_store for B-case durable warning.
reconcile_core_holdings(
slot_repo=self._slot_repo, slot_service=self._slot_service,
broker=self._broker, instance_key=self._instance_key,
alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100
warn_store=self._warn_store, # BAT-050 / REQ-103
)
except Exception: # noqa: BLE001
_log.exception("core-holding divergence reconcile failed")