From 832dcb5fbb88335b0f6a36d38bb3dfe276f57271 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 24 Aug 2026 14:08:01 +0000 Subject: [PATCH 1/4] Initial plan From 30bcbe1d12cc93a694236be53372eea8a6e9e428 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 24 Aug 2026 14:15:43 +0000 Subject: [PATCH 2/4] feat: use broker-authoritative qty in slots RPC holdings Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .proofline/criteria/ac-0022.md | 17 ++ .proofline/identities.json | 2 +- docs/interface-contract.md | 18 +- docs/runtime-flows.md | 7 +- tests/application/test_account_snapshot.py | 19 +- tests/application/test_slots_handler.py | 3 +- .../test_slots_handler_broker_quantity.py | 187 ++++++++++++++++++ trex_core/application/account_snapshot.py | 13 ++ trex_core/application/messaging_handlers.py | 39 +++- trex_core/runtime/cli.py | 12 +- trex_core/runtime/daemon.py | 1 + 11 files changed, 303 insertions(+), 15 deletions(-) create mode 100644 .proofline/criteria/ac-0022.md create mode 100644 tests/application/test_slots_handler_broker_quantity.py diff --git a/.proofline/criteria/ac-0022.md b/.proofline/criteria/ac-0022.md new file mode 100644 index 0000000..33d607d --- /dev/null +++ b/.proofline/criteria/ac-0022.md @@ -0,0 +1,17 @@ +--- +id: "ac-0022" +status: active +--- + +# strategy slots RPC broker-authoritative holding quantity + +## Criterion + +`trex.slots.req.`의 `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을 유발하지 않아야 한다. + +## Verification + +- 성공 조건: mismatch fixture(`broker=440`, `slot=435`, `hub=435`)에서 holding row가 `qty=440`, `slot_quantity=435`, `asset_quantity=435`, `qty_source="broker"`를 반환한다. broker zero fixture는 `qty=0`으로 유지된다. +- 실패·경계 조건: broker unavailable/stale fixture에서 `qty`가 slot/hub 값으로 대체되면 FAIL이다. 이 경우 `qty=null`과 명시적 unavailable metadata를 반환해야 한다. +- 격리 조건: 다중 instance 테스트에서 각 `trex.slots.req.`는 해당 instance account snapshot만 참조하며 타 instance broker/snapshot/slot 데이터가 섞이지 않아야 한다. +- 비침습 조건: slots RPC 호출 전후로 slot 상태 전이, broker mutation, ledger write side effect가 없어야 한다. diff --git a/.proofline/identities.json b/.proofline/identities.json index b98dc6e..1cc2cc4 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, "next_line_number": 7, - "next_ac_number": 22 + "next_ac_number": 23 } diff --git a/docs/interface-contract.md b/docs/interface-contract.md index 37f54fc..6b70a36 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -88,7 +88,11 @@ { "slot_id": "str (UUID)", "status": "inactive | pending_buy | pending_sell | holding | error", "symbol": "str | null", - "qty": "float | null", // slot mirror (REQ-062) + "qty": "float | null", // holding: broker authority / unavailable=null + "qty_source": "broker | unavailable", // holding-only metadata + "broker_quantity_available": "bool", // holding-only metadata + "broker_quantity_captured_at": "ISO-8601 | null", // holding-only metadata + "slot_quantity": "float | null", // core daily_slot mirror (diagnostic) "asset_quantity": "float | null", // hub authority (REQ-062) "avg_price": "float | null", "entry_price": "float | null", @@ -97,7 +101,7 @@ "universe_detached": "bool", "cash_balance": "float | null", // hub authority "cash_estimated": "bool", // true = pending ledger outbox delta 반영 - "warned": "bool", // qty != asset_quantity 시 true (REQ-062) + "warned": "bool", // slot_quantity != asset_quantity 시 true "watchdog": { // null if watchdog store unavailable "expired": "bool", "last_seen": "ISO-8601 | null" @@ -119,12 +123,16 @@ | 필드 | 권위 소스 | 비고 | |---|---|---| | `slot_id`, `status`, `avg_price`, `entry_price`, `forced_close_at`, `universe_detached` | slot (core) | 슬롯 row 직접 | -| `qty` | slot (core) | 슬롯 mirror — broker position | +| `qty` | broker account snapshot | holding 슬롯에서만 authoritative. unavailable/stale 시 null | +| `qty_source` | 파생(metadata) | holding 슬롯: `broker` \| `unavailable` | +| `broker_quantity_available` | 파생(metadata) | holding 슬롯 broker qty freshness 통과 여부 | +| `broker_quantity_captured_at` | broker account snapshot | holding 슬롯 broker qty capture 시각 | +| `slot_quantity` | slot (core) | 기존 daily_slot.qty mirror(진단용) | | `symbol`, `portfolio_id` | slot (core) | 자산 바인딩 | | `asset_quantity` | hub (SnapshotStore) | portfolio-hub 권위값 (REQ-062) | | `cash_balance` | hub (SnapshotStore) | portfolio-hub 권위값; `cash_estimated=true` 시 ledger outbox delta 추정값 | | `cash_estimated` | ledger outbox | pending delta 있을 때 true (REQ-029) | -| `warned` | 파생 | `qty != asset_quantity` (REQ-074 AC-2) | +| `warned` | 파생 | `slot_quantity != asset_quantity` (REQ-074 AC-2) | | `watchdog` | WatchdogStore | 만료 여부 + last_seen; null=store 미연결 | | `paused` | PauseStore | durable pause overlay (REQ-032) | | `strategy_name` | hub (PortfolioItem via SnapshotStore) | REQ-116 additive. portfolio-hub 정본. hub 미제공 시 null | @@ -132,6 +140,7 @@ | `markets` | hub (PortfolioItem via SnapshotStore) | REQ-116 additive. list[str]; hub 미제공 시 [] | - **slot-origin strategy 필드 없음** (REQ-074 AC-5, REQ-116 additive 재도입 후에도 유지): `strategy_id`, `sell_strategy_id`, `active_strategy_id` 는 응답에 포함되지 않는다. `strategy_name`/`strategy_params`/`markets` 는 PortfolioItem(hub) 출처이므로 AC-5 제외 대상과 구분된다 — REQ-116 Notes 참조. +- **Breaking change (holding `qty`) rollout**: 기존 전략 클라이언트가 `qty=slot mirror`로 가정했다면 `slot_quantity`를 읽도록 전환해야 한다. 권장 경로는 (1) 클라이언트에 `qty_source`/`broker_quantity_available` 체크 추가 → (2) fail-closed (`qty is null`) 처리 배포 → (3) `qty`를 broker 정본으로 사용하도록 cutover 순서다. - `spy=false`(기본): holding 슬롯 응답 후 `WatchdogStore.update(slot_id)` 호출 — 전략 폴링 = 생존 신호. - `spy=true`: watchdog 갱신 건너뜀 — 운영자 CLI(`universe`/`slots`) 가 항상 사용. @@ -645,6 +654,7 @@ trex-core가 `source="trex-core"`, `severity="critical"`로 발행하는 초기 결과불명 주문 탐색을 KIS 신형 일별주문체결조회 (`TTTC0081R`/`VTTC0081R`, `EXCG_ID_DVSN_CD=ALL`)로 단일화했다. - 2026-07-06 :: BAT-056 (dcy-002 FR-6, REQ-117/118) — `OrderRequest.venue: str = "KRX"` additive 추가(기존 위치 인수 순서 불변 — AMQP wire 계약 영향 없음, 브로커 어댑터 계층 DTO까지만). KIS 어댑터: submit_order/cancel_order body에 `EXCG_ID_DVSN_CD` 항상 명시 전송(기존 미전송 → KRX 명시화 견고성 향상), cancel_order/get_current_price 시그니처 venue/market 파라미터 추가(기본값 KRX/"J" — 기존 호출자 무영향). paper broker 시그니처 정합(venue/market 무시). 실측 근거: 2026-07-06 NXT 주문/취소 rt_cd=0 확인. +- 2026-08-24 :: issue #62 — `trex.slots.req` holding row `qty`를 broker holdings snapshot 정본으로 전환. `qty_source`/`broker_quantity_available`/`broker_quantity_captured_at`/`slot_quantity`를 추가하고, broker qty unavailable 또는 stale이면 fail-closed(`qty=null`, no core/hub fallback)로 응답. ## REQ-123 Contract Addendum: cost-basis canonical parsing & sync diagnostics diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index 4d33d37..fe4a41c 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -141,10 +141,11 @@ 1. 요청 payload 디코드: `{filter ∈ {all, inactive, pending, holding, error}, spy: bool}`. 2. `slot_repo.list_slots(instance_key)` → status 기준 서버 필터 적용(`inactive`/`pending`/`holding`/`error`; 미인식 = `all`). 3. `spy=false` 면 `slot_repo.list_holding_slots()` 의 모든 holding 슬롯에 `WatchdogStore.update(slot_id)`. -4. 각 슬롯에 `slot_to_dict(s, snapshot_store=..., watchdog_store=..., pause_store=..., watchdog_timeout_sec=...)` 적용 → 병합 row 산출: - - slot 코어 필드: `slot_id/status/symbol/qty/avg_price/entry_price/portfolio_id/forced_close_at/universe_detached` +4. 각 슬롯에 `slot_to_dict(s, snapshot_store=..., watchdog_store=..., pause_store=..., watchdog_timeout_sec=...)` 적용 후 broker holdings snapshot 오버레이를 적용해 병합 row 산출: + - slot 코어 필드: `slot_id/status/symbol/slot_quantity/avg_price/entry_price/portfolio_id/forced_close_at/universe_detached` + - holding authoritative 수량: `qty`(broker), `qty_source`, `broker_quantity_available`, `broker_quantity_captured_at` (`qty_source=unavailable`이면 `qty=null`, mirror fallback 금지) - hub 권위 필드: `asset_quantity`/`cash_balance` (SnapshotStore) - - 파생: `warned = (qty != asset_quantity)` (REQ-074 AC-2) + - 파생: `warned = (slot_quantity != asset_quantity)` (REQ-074 AC-2) - 추정 현금(REQ-029 OQ-4): `cash_estimated=True` + `cash_balance += pending ledger delta`(best-effort) - 워치독: `watchdog: {expired, last_seen}` inline (WatchdogStore) - 오버레이: `paused` (PauseStore, REQ-032) diff --git a/tests/application/test_account_snapshot.py b/tests/application/test_account_snapshot.py index 7983eab..2592ec8 100644 --- a/tests/application/test_account_snapshot.py +++ b/tests/application/test_account_snapshot.py @@ -1,7 +1,7 @@ """Bounded account-read recovery (dcy-041 / GitHub #45).""" from __future__ import annotations -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta import pytest @@ -37,7 +37,7 @@ class FakeAccountBroker: def _wire(): - now = [datetime(2026, 7, 29, tzinfo=timezone.utc)] + now = [datetime(2026, 7, 29, tzinfo=UTC)] broker = FakeAccountBroker() store = InMemoryAccountSnapshotStore() service = AccountSnapshotService( @@ -162,3 +162,18 @@ def test_invalid_or_missing_buy_notional_fails_closed(): assert service.buy_guard(qty=1, limit_price=None) == ( "account_snapshot_price_required" ) + + +def test_slots_rpc_uses_snapshot_only_with_sell_ttl_freshness(): + broker, _, service, now = _wire() + assert service.refresh() + broker.balance = [BalanceItem("005930", 999)] + + holdings, captured_at = service.holdings_for_slots_rpc() + assert holdings == [BalanceItem("005930", 2)] + assert captured_at == now[0] + + now[0] += timedelta(seconds=3601) + with pytest.raises(AccountSnapshotUnavailable) as raised: + service.holdings_for_slots_rpc() + assert raised.value.code == "expired" diff --git a/tests/application/test_slots_handler.py b/tests/application/test_slots_handler.py index bb5764c..7493982 100644 --- a/tests/application/test_slots_handler.py +++ b/tests/application/test_slots_handler.py @@ -155,7 +155,8 @@ def test_warned_true_when_qty_diverges(): resp = handler(_req(filter="holding")) s = resp.payload["slots"][0] # slot.qty=5 vs hub asset_quantity=3 → divergence - assert s["qty"] == 5.0 + assert s["qty"] is None + assert s["slot_quantity"] == 5.0 assert s["asset_quantity"] == 3.0 assert s["warned"] is True diff --git a/tests/application/test_slots_handler_broker_quantity.py b/tests/application/test_slots_handler_broker_quantity.py new file mode 100644 index 0000000..cfb8e52 --- /dev/null +++ b/tests/application/test_slots_handler_broker_quantity.py @@ -0,0 +1,187 @@ +"""Broker-authoritative qty projection in `trex.slots.req` holding rows.""" +from __future__ import annotations + +from datetime import UTC, datetime + +from trex_core.application.account_snapshot import AccountSnapshotUnavailable +from trex_core.application.broker_port import BalanceItem +from trex_core.application.messaging import RpcRequest +from trex_core.application.messaging_handlers import make_slots_rpc_handler +from trex_core.application.portfolio import PortfolioItem, SnapshotStore +from trex_core.domain.slot import SlotService, SlotState +from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + +class _Reader: + def __init__( + self, + *, + holdings: list[BalanceItem] | None = None, + captured_at: datetime | None = None, + unavailable_code: str | None = None, + ) -> None: + self._holdings = holdings or [] + self._captured_at = captured_at + self._unavailable_code = unavailable_code + + def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: + if self._unavailable_code is not None: + raise AccountSnapshotUnavailable(self._unavailable_code) + if self._captured_at is None: + raise AccountSnapshotUnavailable("not_ready") + return list(self._holdings), self._captured_at + + +def _holding_handler( + *, + broker_reader: _Reader, + instance_key: str = "default", + portfolio_id: str = "pf-A", + symbol: str = "104620", +): + repo = InMemorySlotRepository() + svc = SlotService(repo) + 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, + ) + svc.transit_slot( + pending.slot_id, + SlotState.PENDING_BUY, + SlotState.HOLDING, + reason="filled", + actor="test", + avg_price=100.0, + qty=435.0, + ) + + store = SnapshotStore() + store.swap( + [PortfolioItem(portfolio_id=portfolio_id, symbol=symbol, cash_balance=0, asset_quantity=435)] + ) + handler = make_slots_rpc_handler( + repo, + instance_key=instance_key, + snapshot_store=store, + broker_quantity_reader=broker_reader, + ) + return repo, handler + + +def _req_holding(): + return RpcRequest(payload={"filter": "holding", "spy": True}, correlation_id="corr") + + +def test_holding_qty_uses_broker_authority_with_diagnostic_fields(): + _, handler = _holding_handler( + broker_reader=_Reader( + holdings=[BalanceItem(symbol="104620", hldg_qty=440)], + captured_at=datetime(2026, 8, 24, 12, 0, tzinfo=UTC), + ) + ) + row = handler(_req_holding()).payload["slots"][0] + assert row["qty"] == 440.0 + assert row["qty_source"] == "broker" + assert row["broker_quantity_available"] is True + assert row["broker_quantity_captured_at"] == "2026-08-24T12:00:00+00:00" + assert row["slot_quantity"] == 435.0 + assert row["asset_quantity"] == 435.0 + + +def test_holding_qty_distinguishes_confirmed_broker_zero_from_unavailable(): + _, handler = _holding_handler( + broker_reader=_Reader( + holdings=[], + captured_at=datetime(2026, 8, 24, 12, 0, tzinfo=UTC), + ) + ) + row = handler(_req_holding()).payload["slots"][0] + assert row["qty"] == 0.0 + assert row["qty_source"] == "broker" + assert row["broker_quantity_available"] is True + assert row["slot_quantity"] == 435.0 + + +def test_holding_qty_unavailable_does_not_fallback_to_slot_or_hub(): + _, handler = _holding_handler( + broker_reader=_Reader(unavailable_code="not_ready"), + ) + row = handler(_req_holding()).payload["slots"][0] + assert row["qty"] is None + assert row["qty_source"] == "unavailable" + assert row["broker_quantity_available"] is False + assert row["broker_quantity_captured_at"] is None + assert row["slot_quantity"] == 435.0 + assert row["asset_quantity"] == 435.0 + + +def test_holding_qty_stale_snapshot_is_unavailable(): + _, handler = _holding_handler( + broker_reader=_Reader(unavailable_code="expired"), + ) + row = handler(_req_holding()).payload["slots"][0] + assert row["qty"] is None + assert row["qty_source"] == "unavailable" + assert row["broker_quantity_available"] is False + + +def test_broker_quantity_reader_is_instance_bound_per_handler(): + repo = InMemorySlotRepository() + svc = SlotService(repo) + repo.reconcile_universe("inst-A", {"pf-A": "104620"}) + repo.reconcile_universe("inst-B", {"pf-B": "104620"}) + slot_a = repo.find_slot_by_portfolio("inst-A", "pf-A") + slot_b = repo.find_slot_by_portfolio("inst-B", "pf-B") + pending_a = svc.transit_slot( + slot_a.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, + reason="buy", actor="test", symbol="104620", + ) + pending_b = svc.transit_slot( + slot_b.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, + reason="buy", actor="test", symbol="104620", + ) + svc.transit_slot( + pending_a.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, + reason="filled", actor="test", avg_price=100.0, qty=435.0, + ) + svc.transit_slot( + pending_b.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, + reason="filled", actor="test", avg_price=100.0, qty=435.0, + ) + + store = SnapshotStore() + store.swap( + [ + PortfolioItem(portfolio_id="pf-A", symbol="104620", cash_balance=0, asset_quantity=435), + PortfolioItem(portfolio_id="pf-B", symbol="104620", cash_balance=0, asset_quantity=435), + ] + ) + captured_at = datetime(2026, 8, 24, 12, 0, tzinfo=UTC) + handler_a = make_slots_rpc_handler( + repo, + instance_key="inst-A", + snapshot_store=store, + broker_quantity_reader=_Reader( + holdings=[BalanceItem(symbol="104620", hldg_qty=440)], + captured_at=captured_at, + ), + ) + handler_b = make_slots_rpc_handler( + repo, + instance_key="inst-B", + snapshot_store=store, + broker_quantity_reader=_Reader( + holdings=[BalanceItem(symbol="104620", hldg_qty=111)], + captured_at=captured_at, + ), + ) + row_a = handler_a(_req_holding()).payload["slots"][0] + row_b = handler_b(_req_holding()).payload["slots"][0] + assert row_a["qty"] == 440.0 + assert row_b["qty"] == 111.0 diff --git a/trex_core/application/account_snapshot.py b/trex_core/application/account_snapshot.py index d671863..354c590 100644 --- a/trex_core/application/account_snapshot.py +++ b/trex_core/application/account_snapshot.py @@ -217,6 +217,19 @@ class AccountSnapshotService: ) return list(snapshot.holdings) + def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: + """Return freshness-validated snapshot holdings for strategy slots reads. + + Unlike ``holdings_for_sell()``, this read path never performs a live broker + call and therefore cannot mutate broker state or mask snapshot staleness. + """ + snapshot = self._load_current() + self._age(snapshot.holdings_captured_at, self._sell_max_age) + captured_at = snapshot.holdings_captured_at + if captured_at is None: + raise AccountSnapshotUnavailable("not_ready") + return list(snapshot.holdings), captured_at + def buy_guard( self, *, diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index 8e0add9..afd5fe2 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -9,10 +9,12 @@ from __future__ import annotations import logging from datetime import date as date_cls from datetime import datetime, timezone -from typing import Optional +from typing import Optional, Protocol from uuid import UUID from trex_core.application.intents import BuyIntent, IntentService, SellIntent +from trex_core.application.account_snapshot import AccountSnapshotUnavailable +from trex_core.application.broker_port import BalanceItem from trex_core.application.messaging import RpcRequest, RpcResponse from trex_core.application.portfolio import ( PauseStore, @@ -185,6 +187,10 @@ _FILTER_STATUS: dict[str, set[str]] = { } +class SlotsBrokerQuantityReader(Protocol): + def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: ... + + def make_slots_rpc_handler( slot_repo: SlotRepository, *, @@ -197,6 +203,7 @@ def make_slots_rpc_handler( ledger_outbox=None, max_active_slots: Optional[int] = None, strategy_state_store=None, + broker_quantity_reader: Optional[SlotsBrokerQuantityReader] = None, ): """Build the RPC handler for `trex.slots.req` (REQ-073/074/116). @@ -252,6 +259,36 @@ def make_slots_rpc_handler( for s in slots ] + broker_qty_by_symbol: dict[str, int] = {} + broker_qty_available = False + broker_qty_captured_at: Optional[datetime] = None + if broker_quantity_reader is not None: + try: + holdings, broker_qty_captured_at = broker_quantity_reader.holdings_for_slots_rpc() + broker_qty_by_symbol = { + str(item.symbol): int(item.hldg_qty) + for item in holdings + } + broker_qty_available = True + except AccountSnapshotUnavailable: + broker_qty_available = False + + for row, slot in zip(slot_rows, slots): + row["slot_quantity"] = row.get("qty") + if slot.status.value != "holding": + continue + row["broker_quantity_available"] = broker_qty_available + row["broker_quantity_captured_at"] = ( + broker_qty_captured_at.isoformat() if broker_qty_captured_at is not None else None + ) + if broker_qty_available: + broker_qty = int(broker_qty_by_symbol.get(str(slot.symbol), 0)) + row["qty"] = float(broker_qty) + row["qty_source"] = "broker" + else: + row["qty"] = None + row["qty_source"] = "unavailable" + # dcy-028 / REQ-073/074·079 (C-1): attach opaque strategy_state map per # row via one batched store read (read piggyback — no extra round trip). # Only entries matching the slot's current occupancy_seq are surfaced. diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index 90bc283..1d25a39 100644 --- a/trex_core/runtime/cli.py +++ b/trex_core/runtime/cli.py @@ -152,7 +152,8 @@ def _cmd_universe(args, cfg, rpc) -> int: f"degraded={resp.get('degraded')}") for s in slots: print(f" {s.get('portfolio_id')}\t{s.get('symbol')}\t" - f"qty(hub)={s.get('asset_quantity')}\tqty(slot)={s.get('qty')}") + f"qty(hub)={s.get('asset_quantity')}\t" + f"qty(slot)={s.get('slot_quantity', s.get('qty'))}") return 0 @@ -192,7 +193,8 @@ def _cmd_slots(args, cfg, rpc) -> int: wd = s.get("watchdog") or {} tag = " [WATCHDOG: strategy absent]" if wd.get("expired") else "" print(f" {s.get('slot_id')}\t{s.get('status')}\t{s.get('symbol')}\t" - f"qty={s.get('qty')}\taqty={s.get('asset_quantity')}\t" + f"qty={s.get('qty')}\tsqty={s.get('slot_quantity')}\t" + f"aqty={s.get('asset_quantity')}\t" f"limit={s.get('entry_price')}\tavg={s.get('avg_price')}\t" f"pid={pid}\tpaused={s.get('paused')}\twarned={s.get('warned')}{tag}") return 0 @@ -205,7 +207,11 @@ def _print_slot_detail(s: dict) -> None: ("status", s.get("status")), ("symbol", s.get("symbol")), ("portfolio_id", s.get("portfolio_id")), - ("qty (slot mirror)", s.get("qty")), + ("qty (authoritative)", s.get("qty")), + ("slot_quantity (core mirror)", s.get("slot_quantity")), + ("qty_source", s.get("qty_source")), + ("broker_quantity_available", s.get("broker_quantity_available")), + ("broker_quantity_captured_at", s.get("broker_quantity_captured_at")), ("asset_quantity (hub)", s.get("asset_quantity")), ("warned", s.get("warned")), ("avg_price", s.get("avg_price")), diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index f1fc5b7..5c791b9 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -445,6 +445,7 @@ class Daemon: ledger_outbox=self.outbox, max_active_slots=self.config.max_active_slots, strategy_state_store=self.strategy_state_store, # dcy-028 C-1 + broker_quantity_reader=self.account_snapshot_service, ) transport.consume_rpc_forever( topology.slots_queue(self.config.instance_key), handler, self.stop) From f394f1560d6c10c9832d5430e5b1978296fb786e Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 24 Aug 2026 14:19:26 +0000 Subject: [PATCH 3/4] chore: finalize broker qty metadata consistency Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- docs/interface-contract.md | 12 ++++++------ trex_core/application/account_snapshot.py | 5 ++++- trex_core/application/messaging_handlers.py | 13 +++++++------ 3 files changed, 17 insertions(+), 13 deletions(-) diff --git a/docs/interface-contract.md b/docs/interface-contract.md index 6b70a36..414cfcf 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -89,9 +89,9 @@ "status": "inactive | pending_buy | pending_sell | holding | error", "symbol": "str | null", "qty": "float | null", // holding: broker authority / unavailable=null - "qty_source": "broker | unavailable", // holding-only metadata - "broker_quantity_available": "bool", // holding-only metadata - "broker_quantity_captured_at": "ISO-8601 | null", // holding-only metadata + "qty_source": "broker | unavailable | null", // non-holding=null + "broker_quantity_available": "bool", // non-holding=false + "broker_quantity_captured_at": "ISO-8601 | null", // non-holding=null "slot_quantity": "float | null", // core daily_slot mirror (diagnostic) "asset_quantity": "float | null", // hub authority (REQ-062) "avg_price": "float | null", @@ -124,9 +124,9 @@ |---|---|---| | `slot_id`, `status`, `avg_price`, `entry_price`, `forced_close_at`, `universe_detached` | slot (core) | 슬롯 row 직접 | | `qty` | broker account snapshot | holding 슬롯에서만 authoritative. unavailable/stale 시 null | -| `qty_source` | 파생(metadata) | holding 슬롯: `broker` \| `unavailable` | -| `broker_quantity_available` | 파생(metadata) | holding 슬롯 broker qty freshness 통과 여부 | -| `broker_quantity_captured_at` | broker account snapshot | holding 슬롯 broker qty capture 시각 | +| `qty_source` | 파생(metadata) | holding 슬롯: `broker` \| `unavailable`; non-holding은 null | +| `broker_quantity_available` | 파생(metadata) | holding 슬롯 broker qty freshness 통과 여부; non-holding=false | +| `broker_quantity_captured_at` | broker account snapshot | holding 슬롯 broker qty capture 시각; non-holding=null | | `slot_quantity` | slot (core) | 기존 daily_slot.qty mirror(진단용) | | `symbol`, `portfolio_id` | slot (core) | 자산 바인딩 | | `asset_quantity` | hub (SnapshotStore) | portfolio-hub 권위값 (REQ-062) | diff --git a/trex_core/application/account_snapshot.py b/trex_core/application/account_snapshot.py index 354c590..aeeca08 100644 --- a/trex_core/application/account_snapshot.py +++ b/trex_core/application/account_snapshot.py @@ -222,12 +222,15 @@ class AccountSnapshotService: Unlike ``holdings_for_sell()``, this read path never performs a live broker call and therefore cannot mutate broker state or mask snapshot staleness. + Freshness uses the same ``sell_max_age_sec`` policy (`self._sell_max_age`) + as pending-sell quantity protection so strategy reads and sell dispatch + share one stale-snapshot threshold. """ snapshot = self._load_current() - self._age(snapshot.holdings_captured_at, self._sell_max_age) captured_at = snapshot.holdings_captured_at if captured_at is None: raise AccountSnapshotUnavailable("not_ready") + self._age(captured_at, self._sell_max_age) return list(snapshot.holdings), captured_at def buy_guard( diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index afd5fe2..092c73c 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -273,18 +273,19 @@ def make_slots_rpc_handler( except AccountSnapshotUnavailable: broker_qty_available = False - for row, slot in zip(slot_rows, slots): + for row, slot in zip(slot_rows, slots, strict=True): row["slot_quantity"] = row.get("qty") + row["qty_source"] = None + row["broker_quantity_available"] = False + row["broker_quantity_captured_at"] = None if slot.status.value != "holding": continue - row["broker_quantity_available"] = broker_qty_available - row["broker_quantity_captured_at"] = ( - broker_qty_captured_at.isoformat() if broker_qty_captured_at is not None else None - ) if broker_qty_available: broker_qty = int(broker_qty_by_symbol.get(str(slot.symbol), 0)) row["qty"] = float(broker_qty) row["qty_source"] = "broker" + row["broker_quantity_available"] = True + row["broker_quantity_captured_at"] = broker_qty_captured_at.isoformat() else: row["qty"] = None row["qty_source"] = "unavailable" @@ -296,7 +297,7 @@ def make_slots_rpc_handler( slot_ids = [s.slot_id for s in slots] occ = {s.slot_id: getattr(s, "occupancy_seq", 0) for s in slots} state_map = strategy_state_store.get_for_slots(slot_ids, occ) - for row, s in zip(slot_rows, slots): + for row, s in zip(slot_rows, slots, strict=True): entries = state_map.get(s.slot_id) if entries: row["strategy_state"] = { From d4a968ee1a4e86f2aa23bf8c72822f61a843645c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 25 Aug 2026 12:25:39 +0000 Subject: [PATCH 4/4] fix: normalize snapshot read failures to unavailable in slots RPC path Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- tests/application/test_account_snapshot.py | 21 ++++++++ .../test_slots_handler_broker_quantity.py | 48 ++++++++++++++++++- trex_core/application/account_snapshot.py | 11 ++++- 3 files changed, 77 insertions(+), 3 deletions(-) diff --git a/tests/application/test_account_snapshot.py b/tests/application/test_account_snapshot.py index 2592ec8..7a171e5 100644 --- a/tests/application/test_account_snapshot.py +++ b/tests/application/test_account_snapshot.py @@ -36,6 +36,11 @@ class FakeAccountBroker: return self.cash +class BrokenLoadStore(InMemoryAccountSnapshotStore): + def load(self, account_hash: str) -> AccountSnapshot | None: + raise RuntimeError("db read failed") + + def _wire(): now = [datetime(2026, 7, 29, tzinfo=UTC)] broker = FakeAccountBroker() @@ -177,3 +182,19 @@ def test_slots_rpc_uses_snapshot_only_with_sell_ttl_freshness(): with pytest.raises(AccountSnapshotUnavailable) as raised: service.holdings_for_slots_rpc() assert raised.value.code == "expired" + + +def test_slots_rpc_normalizes_snapshot_store_read_failure(): + broker = FakeAccountBroker() + service = AccountSnapshotService( + broker=broker, + store=BrokenLoadStore(), + buy_max_age_sec=600, + sell_max_age_sec=3600, + clock=lambda: datetime(2026, 7, 29, tzinfo=UTC), + rng=lambda: 0.5, + ) + + with pytest.raises(AccountSnapshotUnavailable) as raised: + service.holdings_for_slots_rpc() + assert raised.value.code == "read_failed" diff --git a/tests/application/test_slots_handler_broker_quantity.py b/tests/application/test_slots_handler_broker_quantity.py index cfb8e52..7a8d571 100644 --- a/tests/application/test_slots_handler_broker_quantity.py +++ b/tests/application/test_slots_handler_broker_quantity.py @@ -2,8 +2,14 @@ from __future__ import annotations from datetime import UTC, datetime +from typing import Protocol -from trex_core.application.account_snapshot import AccountSnapshotUnavailable +from trex_core.application.account_snapshot import ( + AccountSnapshot, + AccountSnapshotService, + AccountSnapshotUnavailable, + InMemoryAccountSnapshotStore, +) from trex_core.application.broker_port import BalanceItem from trex_core.application.messaging import RpcRequest from trex_core.application.messaging_handlers import make_slots_rpc_handler @@ -32,9 +38,29 @@ class _Reader: return list(self._holdings), self._captured_at +class _SlotsReader(Protocol): + def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: ... + + +class _BrokenLoadStore(InMemoryAccountSnapshotStore): + def load(self, account_hash: str) -> AccountSnapshot | None: + raise RuntimeError("db read failed") + + +class _Broker: + def account_scope_hash(self) -> str: + return "a" * 64 + + def get_balance(self) -> list[BalanceItem]: + return [] + + def get_available_cash(self, *, price_check_code: str = "005930") -> int: + return 0 + + def _holding_handler( *, - broker_reader: _Reader, + broker_reader: _SlotsReader, instance_key: str = "default", portfolio_id: str = "pf-A", symbol: str = "104620", @@ -131,6 +157,24 @@ def test_holding_qty_stale_snapshot_is_unavailable(): assert row["broker_quantity_available"] is False +def test_holding_qty_store_read_failure_is_unavailable(): + _, handler = _holding_handler( + broker_reader=AccountSnapshotService( + broker=_Broker(), + store=_BrokenLoadStore(), + buy_max_age_sec=600, + sell_max_age_sec=3600, + clock=lambda: datetime(2026, 8, 24, 12, 0, tzinfo=UTC), + rng=lambda: 0.5, + ), + ) + row = handler(_req_holding()).payload["slots"][0] + assert row["qty"] is None + assert row["qty_source"] == "unavailable" + assert row["broker_quantity_available"] is False + assert row["broker_quantity_captured_at"] is None + + def test_broker_quantity_reader_is_instance_bound_per_handler(): repo = InMemorySlotRepository() svc = SlotService(repo) diff --git a/trex_core/application/account_snapshot.py b/trex_core/application/account_snapshot.py index aeeca08..c61c08c 100644 --- a/trex_core/application/account_snapshot.py +++ b/trex_core/application/account_snapshot.py @@ -49,6 +49,7 @@ class AccountSnapshotUnavailable(RuntimeError): "account_mismatch", "expired", "live_read_failed", + "read_failed", }) def __init__(self, code: str) -> None: @@ -189,7 +190,15 @@ class AccountSnapshotService: account_hash = self._broker.account_scope_hash() if not account_hash: raise AccountSnapshotUnavailable("account_mismatch") - snapshot = self._store.load(account_hash) + try: + snapshot = self._store.load(account_hash) + except Exception as exc: + _log.warning( + "account snapshot load failed: account_hash=%s", + account_hash, + exc_info=True, + ) + raise AccountSnapshotUnavailable("read_failed") from exc if snapshot is None or snapshot.account_hash != account_hash: raise AccountSnapshotUnavailable("account_mismatch") return snapshot