diff --git a/docs/interface-contract.md b/docs/interface-contract.md index 3bb43a5..bc7038b 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -4,8 +4,8 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-06-21 (BAT-049 — dcy-033 알람 채널 trex.alerts 추가) -- Source Discovery / Batch: bootstrap-baseline (pre-Discovery); HTTP API = dcy-002 / BAT-002 +- Last Updated (KST): 2026-07-15 (BAT-057 — operator broker truth hotfix baseline alignment) +- Source Discovery / Batch: bootstrap-baseline (pre-Discovery); HTTP API = dcy-002 / BAT-002; operator broker truth = dcy-037 / BAT-057 - Primary Runtime: daemon (HTTP API) + library (slot 도메인) ## Project Summary @@ -420,6 +420,13 @@ resume_kill_switch 요청·응답 (REQ-046): ## `BrokerPort` Protocol 메서드 (BAT-056 / REQ-118) +### Balance-read failure and reconciliation boundary (BAT-057 / REQ-120/121) + +- `get_balance()`는 검증된 `list[BalanceItem]`만 반환한다. KIS의 token/auth/authorization/transport/response/rejection 실패는 빈 목록으로 축소되지 않고 `BrokerBalanceError`를 raise한다. 안정 code allow-list는 `token_issuance_failed`, `authentication_failed`, `authorization_failed`, `transport_failed`, `malformed_response`, `broker_rejected`다. +- `sync`와 `verify` CLI는 첫 balance read를 완료한 뒤 cache한 결과를 해당 invocation에 전달한다. 이 read 또는 broker credential resolution이 실패하면 exit code `1`이며, `--json` stderr는 `{ "error": {"code": "…", "message": "broker balance unavailable"} }` 형태의 안전한 값만 낸다. upstream payload, credential, account 값은 계약상 출력하지 않는다. +- `reconcile_core_holdings()`의 broker-only 결과는 `unmanaged_position`(대상 instance의 managed slot 없음) 또는 `managed_non_holding_divergence`(동일 instance의 non-holding managed slot 존재)다. `find_managed_slot_by_symbol(instance_key, symbol)` 조회는 instance-scoped이며, 이 두 분류 자체는 slot/warning write나 자동 주문·청산·slot 생성을 하지 않는다. +- 이 변경은 AMQP 또는 portfolio-hub external wire contract와 DB schema를 바꾸지 않는다. + ### cancel_order ``` diff --git a/docs/operations.md b/docs/operations.md index 6497ec4..f0af40e 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -61,6 +61,13 @@ pkill -TERM -f trex-core-daemon - enter/exit 안전장치: 확인 프롬프트 기본 ON, `--yes`(paper 생략), live(`broker=kis`)는 `--yes --live` 이중 ack, `--dry-run` 미리보기. `--json`은 비대화형이라 enter/exit에 `--yes` 필수. - 인스턴스 라우팅: `--config ` 또는 `--instance `, 미지정 시 `default`. +### `sync` / `verify` broker truth 확인과 release gate (BAT-057) + +- `sync`와 `verify`는 broker balance를 먼저 한 번 읽어 해당 실행 안에서 재사용한다. balance/credential read 실패는 안전한 error code와 함께 exit `1`로 종료하며, 빈 broker balance를 추정하지 않는다. 성공한 빈 목록만 flat balance로 취급한다. +- source evidence에는 mode(`fetched`/`env`/`break_glass`)와 비밀 없는 label만 기록한다. fetched mode는 KIS + `credential_fetch_enabled=true`에서만 사용하고, break-glass는 명시적 운영자 환경 설정으로 fetch를 건너뛴다. Paper와 env KIS 경로는 fetch하지 않는다. +- reconciliation에서 `unmanaged_position` 또는 `managed_non_holding_divergence`가 보이면 자동 보정·자동 주문·자동 청산이 없다. broker, ledger, slot evidence를 교차 확인하고 warning/slot 상태를 보존한 채 triage한다. +- remediation은 **dry-run → broker/ledger/slot 교차 확인 → 사장님 명시 승인 → 단일 대상 `sync --live --yes` → 재-reconcile** 순서의 별도 release gate다. 이 문서는 실제 target 선택, live 실행 또는 승인을 부여하지 않는다. + ### `report` — 일일 운영 리포트 (REQ-088 / dcy-029, REQ-110/111 / dcy-035) ```bash .venv/bin/trex-core report # 사람용 표 diff --git a/docs/project-structure.md b/docs/project-structure.md index 5d7d27e..5f0c3d7 100644 --- a/docs/project-structure.md +++ b/docs/project-structure.md @@ -92,6 +92,7 @@ │ └── runtime/ # 데몬 (BAT-002, BAT-004, BAT-006, BAT-007, BAT-008) │ ├── config.py # DaemonConfig(+instance_key, max_active_slots) + KISConfig + PortfolioHubConfig + RabbitMQConfig + SnapshotScheduleConfig + SafetySellConfig + DaemonInstanceConfig.from_toml/from_config (env + per-instance TOML, BAT-013/028/030) │ ├── daemon.py # Daemon class: `--config` TOML 진입점(BAT-013) + setup(migrate/reconcile/declare-topology(instance_key)) + 7 worker threads(instance_key 큐 구독) + WatchdogStore + KillSwitchService + graceful shutdown. _execution_loop 각 틱 시작 시 event_publisher.keepalive_tick() 호출(REQ-092). _make_event_emitter 발행 최종 실패 시 ERROR 로그(REQ-093). +│ ├── broker_provider.py # daemon/CLI 공용 broker source resolution: KIS fetched/env/break-glass와 day-scoped fetched credential 재사용 (BAT-057 / REQ-119) │ ├── reset.py # trex-core-reset: slot-core 테이블 wipe + 자산 중심 재초기화 (REQ-064, BAT-030) │ ├── logging_setup.py # 구조화 로깅(ISO8601 KST·레벨·Loki JSON·correlation id) + configure_logging (BAT-045 / REQ-084, dcy-029) │ └── cli.py # trex-core CLI: pause/resume/paused + universe/slots/enter/exit/cancel + report(REQ-088) (BAT-008/020/046 / REQ-032) diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index 51662d5..789dfdb 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -4,8 +4,8 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-06-21 (BAT-049 — dcy-033 알람 발신 흐름 추가) -- Source Discovery / Batch: bootstrap-baseline (pre-Discovery); 데몬/HTTP 흐름 = dcy-002 / BAT-002 +- Last Updated (KST): 2026-07-15 (BAT-057 — operator broker truth hotfix baseline alignment) +- Source Discovery / Batch: bootstrap-baseline (pre-Discovery); 데몬/HTTP 흐름 = dcy-002 / BAT-002; operator broker truth = dcy-037 / BAT-057 - Primary Runtime: daemon (HTTP API) + library (slot 도메인) ## Project Summary @@ -98,7 +98,13 @@ 1. 신규 자산(portfolio_id 에 슬롯 없음) → `inactive` 영구 슬롯 생성(멱등 — `ON CONFLICT (instance_key, portfolio_id)` partial unique). 2. universe 이탈 + inactive → soft-retire(`retired_at` set; 행 보존, 운영 뷰 `universe_detached`). 이탈 + holding/pending → tombstone(그대로 유지 — safety-sell 경로 보존). 3. 가드: universe 빈 응답이면 reconcile skip(degrade 가드); 직전 대비 ≥50% 축소면 retire 보류(`allow_retire=False`, 대량 소실 가드 — dcy-020 OQ-2 D3). -4. 이어서 `reconcile_core_holdings`(REQ-062): 브로커 잔고 ground truth 로 core holding 검증 — core holding/브로커 0 → slot `error` 격리+WARN, 수량 불일치 → WARN(상태 불변), 브로커 보유/슬롯 없음 → WARN(unmanaged). 자동 청산 없음. +4. 이어서 `reconcile_core_holdings`(REQ-062/121): 브로커 잔고 ground truth 로 core holding 검증 — core holding/브로커 0 → slot `error` 격리+WARN, 수량 불일치 → WARN(상태 불변), 브로커 보유인데 대상 instance에 관리 slot이 없으면 `unmanaged_position`, 동일 instance에 `inactive`/`error` 등 non-holding 관리 slot이 있으면 `managed_non_holding_divergence`를 WARN으로 남긴다. 두 broker-only 분류는 상태·warning persistence·주문·청산·slot 생성의 자동 수정을 하지 않는다. + +### Broker provider and operator balance-read barrier (BAT-057 / REQ-119/120) + +1. daemon과 `sync`/`verify` CLI는 같은 broker provider를 통해 브로커를 얻는다. `broker=kis` 및 `credential_fetch_enabled=true`일 때만 portfolio-hub의 유효한 인증정보를 fetched source로 사용한다. daemon의 day-start refresh는 이미 같은 KST 날짜에 얻은 source가 있으면 이를 재사용한다. +2. 명시적 `TREX_KIS_BROKER_BREAK_GLASS`는 fetched source를 건너뛰고 env KIS 설정을 사용하는 별도 source다. Paper와 `credential_fetch_enabled=false` KIS 경로는 env/config source이며 credential fetch를 하지 않는다. +3. `sync`/`verify`는 실행당 잔고를 한 번 먼저 검증해 cache한 뒤 그 값만 사용한다. KIS 잔고 읽기 실패는 빈 잔고로 해석하거나 후속 DB/slot write로 진행하지 않고 command boundary에서 안전하게 실패한다. 성공한 빈 잔고(`[]`)만 검증된 flat balance다. ## Flow: `포지션 정합 검증` (BAT-008 / REQ-031) diff --git a/tests/adapters/broker/test_kis.py b/tests/adapters/broker/test_kis.py index 26f1e36..1e1ea93 100644 --- a/tests/adapters/broker/test_kis.py +++ b/tests/adapters/broker/test_kis.py @@ -10,7 +10,7 @@ import os import pytest from trex_core.adapters.broker.kis import KISBroker -from trex_core.application.broker_port import OrderRequest +from trex_core.application.broker_port import BrokerBalanceError, OrderRequest _MOCK_URL = "https://openapivts.koreainvestment.com:29443" @@ -75,3 +75,38 @@ def test_sandbox_token_and_submit(): assert ack.broker_order_id else: assert ack.reason # rejection carries a reason (e.g., 장 미개장/잔고부족) + + +def test_get_balance_rejects_failure_instead_of_false_empty(monkeypatch): + """BAT-057: a rejected KIS response is not a successful empty balance.""" + b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL) + monkeypatch.setattr(b, "_headers", lambda _tr: {}) + + class Response: + def raise_for_status(self): pass + def json(self): return {"rt_cd": "1", "output1": []} + + monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", lambda *a, **kw: Response()) + with pytest.raises(BrokerBalanceError, match="broker_rejected"): + b.get_balance() + + +def test_get_balance_accepts_validated_empty_balance(monkeypatch): + b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL) + monkeypatch.setattr(b, "_headers", lambda _tr: {}) + + class Response: + def raise_for_status(self): pass + def json(self): return {"rt_cd": "0", "output1": []} + + monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", lambda *a, **kw: Response()) + assert b.get_balance() == [] + + +def test_get_balance_sanitizes_token_issuance_failure(monkeypatch): + b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL) + monkeypatch.setattr(b, "_headers", lambda _tr: (_ for _ in ()).throw(KeyError("access_token-secret"))) + with pytest.raises(BrokerBalanceError) as raised: + b.get_balance() + assert raised.value.code == "token_issuance_failed" + assert "secret" not in str(raised.value) diff --git a/tests/application/test_holding_divergence.py b/tests/application/test_holding_divergence.py index 1036362..08cbf3d 100644 --- a/tests/application/test_holding_divergence.py +++ b/tests/application/test_holding_divergence.py @@ -1,7 +1,10 @@ """dcy-020 / REQ-062 (BAT-029): core holding ↔ broker divergence reconcile.""" from __future__ import annotations +from dataclasses import replace + from trex_core.application.broker_port import BalanceItem +from trex_core.application.portfolio import InMemoryReconcileWarningStore from trex_core.application.reconciliation import reconcile_core_holdings from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository @@ -36,6 +39,18 @@ def _wire(): return repo, SlotService(repo, instance_key=INST) +class TrackingRepository(InMemorySlotRepository): + """Records writes so reconciliation-only classification stays read-only.""" + + def __init__(self): + super().__init__() + self.update_calls = 0 + + def update_slot(self, slot): + self.update_calls += 1 + return super().update_slot(slot) + + def test_A_core_holding_broker_flat_goes_error(): repo, svc = _wire() slot = _hold(repo, svc, "pf-a", "005930", 10) @@ -66,4 +81,71 @@ def test_B_broker_holds_unmanaged_warns(): repo, svc = _wire() broker = FakeBroker([BalanceItem(symbol="000660", hldg_qty=5)]) # core has none d = reconcile_core_holdings(slot_repo=repo, slot_service=svc, broker=broker, instance_key=INST) - assert d["broker:000660"] == "unmanaged_warning" + assert d["broker:000660"] == "unmanaged_position" + + +def test_broker_held_symbol_with_same_instance_non_holding_slot_is_read_only_divergence(): + repo = TrackingRepository() + svc = SlotService(repo, instance_key=INST) + warn_store = InMemoryReconcileWarningStore() + repo.reconcile_universe(INST, {"pf-a": "000660"}) + slot = repo.find_slot_by_portfolio(INST, "pf-a") + assert slot is not None + writes_before = repo.update_calls + + decisions = reconcile_core_holdings( + slot_repo=repo, slot_service=svc, + broker=FakeBroker([BalanceItem(symbol="000660", hldg_qty=5)]), + instance_key=INST, warn_store=warn_store, + ) + + assert decisions == {"broker:000660": "managed_non_holding_divergence"} + assert repo.fetch_slot(slot.slot_id) == slot + assert repo.update_calls == writes_before + assert warn_store.list_warned() == set() + assert repo.list_slots(INST) == [slot] + + +def test_broker_held_symbol_with_same_instance_error_slot_is_read_only_divergence(): + repo = TrackingRepository() + svc = SlotService(repo, instance_key=INST) + warn_store = InMemoryReconcileWarningStore() + repo.reconcile_universe(INST, {"pf-a": "000660"}) + inactive = repo.find_slot_by_portfolio(INST, "pf-a") + assert inactive is not None + error = replace(inactive, status=SlotState.ERROR) + repo.update_slot(error) + writes_before = repo.update_calls + + decisions = reconcile_core_holdings( + slot_repo=repo, slot_service=svc, + broker=FakeBroker([BalanceItem(symbol="000660", hldg_qty=5)]), + instance_key=INST, warn_store=warn_store, + ) + + assert decisions == {"broker:000660": "managed_non_holding_divergence"} + assert repo.fetch_slot(error.slot_id) == error + assert repo.update_calls == writes_before + assert warn_store.list_warned() == set() + + +def test_other_instance_only_managed_slot_remains_unmanaged_position_without_writes(): + repo = TrackingRepository() + svc = SlotService(repo, instance_key=INST) + warn_store = InMemoryReconcileWarningStore() + repo.reconcile_universe("other-instance", {"pf-a": "000660"}) + other = repo.find_slot_by_portfolio("other-instance", "pf-a") + assert other is not None + writes_before = repo.update_calls + + decisions = reconcile_core_holdings( + slot_repo=repo, slot_service=svc, + broker=FakeBroker([BalanceItem(symbol="000660", hldg_qty=5)]), + instance_key=INST, warn_store=warn_store, + ) + + assert decisions == {"broker:000660": "unmanaged_position"} + assert repo.fetch_slot(other.slot_id) == other + assert repo.update_calls == writes_before + assert warn_store.list_warned() == set() + assert repo.list_slots(INST) == [] diff --git a/tests/application/test_reconcile_warning_blocking.py b/tests/application/test_reconcile_warning_blocking.py index 778f4ff..ad5ef31 100644 --- a/tests/application/test_reconcile_warning_blocking.py +++ b/tests/application/test_reconcile_warning_blocking.py @@ -130,7 +130,7 @@ def test_reconcile_core_holdings_matching_qty_no_warn(caplog): instance_key="default", warn_store=warn_store, ) assert warn_store.list_warned() == set() # S-4: no false positive - assert all(v == "ok" for v in decisions.values() if v != "unmanaged_warning") + assert all(v == "ok" for v in decisions.values() if v != "unmanaged_position") # REQ-103 AC-4: A case (broker 0) => ERROR transition, NOT warn_store diff --git a/tests/runtime/test_broker_provider.py b/tests/runtime/test_broker_provider.py new file mode 100644 index 0000000..2472ec1 --- /dev/null +++ b/tests/runtime/test_broker_provider.py @@ -0,0 +1,83 @@ +"""REQ-119 provider source and refresh lifecycle regressions.""" +from types import SimpleNamespace + +import pytest + +from trex_core.runtime.broker_provider import ( + BrokerCredentialUnavailable, + ConfigBrokerProvider, +) +from trex_core.runtime.config import DaemonConfig, KISConfig, PortfolioHubConfig + + +def _cfg(*, fetched=True, broker="kis"): + return DaemonConfig( + dsn="postgresql://test/test", broker=broker, execution_interval_sec=1, + kis=KISConfig("https://openapivts.example", "env-key", "env-secret", "env-account", 1, False), + portfolio_hub=PortfolioHubConfig("http://hub"), credential_fetch_enabled=fetched, + ) + + +class _Broker: + def __init__(self): self.credentials = [] + def reinitialize(self, credential): self.credentials.append(credential) + + +class _Read: + def __init__(self, ctx=None, error=None): self.ctx, self.error, self.calls = ctx, error, 0 + def fetch_broker_credential(self): + self.calls += 1 + if self.error: + raise self.error + return self.ctx + + +def _ctx(version="v1"): + return SimpleNamespace(trading_enabled=True, broker=SimpleNamespace( + account_ref="synthetic-account", app_key="synthetic-key", app_secret="synthetic-secret", + credential_version=version, + )) + + +def test_fetched_provider_constructs_once_reuses_and_refreshes_without_env_fallback(monkeypatch): + broker = _Broker() + monkeypatch.setattr("trex_core.runtime.broker_provider._make_env_broker", lambda cfg: broker) + read = _Read(_ctx()) + provider = ConfigBrokerProvider(_cfg(), read_port=read) + + got, source = provider.get_broker_for_read() + assert got is broker + assert source.mode == "fetched" + assert source.identity_label == "credential-version:v1" + assert broker.credentials[0].app_key == "synthetic-key" + assert read.calls == 1 + assert provider.refresh_day_start_credential() == source + assert read.calls == 1 # same day cache, no broker/token churn + + +def test_fetched_provider_failure_is_sanitized_and_never_constructs_env_broker(monkeypatch): + monkeypatch.setattr("trex_core.runtime.broker_provider._make_env_broker", lambda cfg: pytest.fail("env fallback")) + provider = ConfigBrokerProvider(_cfg(), read_port=_Read(ctx=None, error=RuntimeError("secret payload"))) + with pytest.raises(BrokerCredentialUnavailable) as raised: + provider.get_broker_for_read() + assert raised.value.code == "portfolio_hub_unavailable" + assert "secret" not in str(raised.value) + + +def test_enabled_paper_never_fetches_kis_credentials(monkeypatch): + read = _Read(_ctx()) + provider = ConfigBrokerProvider(_cfg(broker="paper"), read_port=read) + _, source = provider.get_broker_for_read() + assert source.mode == "env" + assert read.calls == 0 + + +def test_break_glass_is_distinct_and_skips_fetch(monkeypatch): + broker = _Broker() + monkeypatch.setenv("TREX_KIS_BROKER_BREAK_GLASS", "true") + monkeypatch.setattr("trex_core.runtime.broker_provider._make_env_broker", lambda cfg: broker) + read = _Read(_ctx()) + _, source = ConfigBrokerProvider(_cfg(), read_port=read).get_broker_for_read() + assert source.mode == "break_glass" + assert source.identity_label == "operator-break-glass" + assert read.calls == 0 diff --git a/tests/runtime/test_cli_broker_failure.py b/tests/runtime/test_cli_broker_failure.py new file mode 100644 index 0000000..ad3ddd2 --- /dev/null +++ b/tests/runtime/test_cli_broker_failure.py @@ -0,0 +1,53 @@ +"""REQ-120 CLI failure boundary regression tests.""" +import sys +import types +from types import SimpleNamespace + +# This worktree intentionally lacks pika. CLI imports only messaging topology for +# these tests; install a minimal import-time stub without contacting a broker. +_pika = types.ModuleType("pika") +_pika.__path__ = [] +_pika.URLParameters = lambda *a, **k: None +_pika.BlockingConnection = object +_pika.BasicProperties = object +_pika_exceptions = types.ModuleType("pika.exceptions") +_pika_exceptions.AMQPError = Exception +_pika.exceptions = _pika_exceptions +sys.modules.setdefault("pika", _pika) +sys.modules.setdefault("pika.exceptions", _pika_exceptions) + +from trex_core.application.broker_port import BrokerBalanceError # noqa: E402 +from trex_core.runtime import cli # noqa: E402 +from trex_core.runtime.config import DaemonConfig # noqa: E402 + + +def _args(*, as_json=True): + return SimpleNamespace(portfolio_id="PF-1", all=False, json=as_json, dry_run=True, + yes=True, live=False, instance=None, config=None, config_path=None) + + +def test_sync_balance_failure_is_sanitized_and_happens_before_db(monkeypatch, capsys): + class Provider: + def get_broker_for_read(self): + return SimpleNamespace(get_balance=lambda: (_ for _ in ()).throw(BrokerBalanceError("authentication_failed"))), None + monkeypatch.setattr("trex_core.runtime.broker_provider.build_broker_provider", lambda cfg: Provider()) + # A DB object being constructed would prove the boundary arrived too late. + monkeypatch.setattr("trex_core.adapters.portfolio.reconcile_warning_store.PostgreSQLReconcileWarningStore", + lambda *a: (_ for _ in ()).throw(AssertionError("DB touched"))) + assert cli._cmd_sync(_args(), DaemonConfig("dsn", "paper", 1), lambda _: True) == 1 + captured = capsys.readouterr() + assert '"authentication_failed"' in captured.err + assert "broker balance unavailable" in captured.err + + +def test_verify_balance_failure_never_leaks_payload_or_opens_db(monkeypatch, capsys): + class Provider: + def get_broker_for_read(self): + return SimpleNamespace(get_balance=lambda: (_ for _ in ()).throw(BrokerBalanceError("transport_failed"))), None + monkeypatch.setattr("trex_core.runtime.broker_provider.build_broker_provider", lambda cfg: Provider()) + monkeypatch.setattr("trex_core.adapters.persistence.postgresql.PostgreSQLSlotRepository", + lambda *a: (_ for _ in ()).throw(AssertionError("DB touched"))) + assert cli._cmd_verify(_args(), DaemonConfig("dsn", "paper", 1)) == 1 + captured = capsys.readouterr() + assert '"transport_failed"' in captured.err + assert "payload-secret" not in captured.err diff --git a/trex_core/adapters/broker/kis.py b/trex_core/adapters/broker/kis.py index e0633f7..684b528 100644 --- a/trex_core/adapters/broker/kis.py +++ b/trex_core/adapters/broker/kis.py @@ -18,6 +18,7 @@ import httpx from trex_core.application.broker_port import ( BalanceItem, + BrokerBalanceError, FillStatus, OrderAck, OrderRequest, @@ -285,8 +286,9 @@ class KISBroker: """KIS 주식잔고조회 (`TTTC8434R`/`VTTC8434R`). Aggregates `output1[*].pdno` + `hldg_qty`, drops zero-qty rows - (D-2 leftover, Design Decision 7). Network/credential errors map to - an empty list (caller treats as no broker positions). + (D-2 leftover, Design Decision 7). Only a validated `rt_cd=0` response + may return an empty list; token/HTTP/shape/KIS failures raise a sanitized + :class:`BrokerBalanceError` so reconciliation cannot infer false zero. """ cano, product = self._account() tr_id = "VTTC8434R" if self._mock else "TTTC8434R" @@ -298,21 +300,44 @@ class KISBroker: "PRCS_DVSN": "01", "CTX_AREA_FK100": "", "CTX_AREA_NK100": "", } + try: + headers = self._headers(tr_id) + except (httpx.HTTPError, KeyError, TypeError, ValueError): + # Token issuance/header construction is distinct from a rejected + # balance endpoint response and must never look like an empty book. + raise BrokerBalanceError("token_issuance_failed") from None try: resp = httpx.get( f"{self._base_url}/uapi/domestic-stock/v1/trading/inquire-balance", - headers=self._headers(tr_id), params=params, timeout=self._timeout, + headers=headers, params=params, timeout=self._timeout, ) + resp.raise_for_status() payload = resp.json() - except (httpx.HTTPError, ValueError): - return [] + except httpx.HTTPStatusError as exc: + status = exc.response.status_code if exc.response is not None else None + code = "authentication_failed" if status == 401 else ( + "authorization_failed" if status == 403 else "transport_failed" + ) + raise BrokerBalanceError(code) from None + except httpx.HTTPError: + raise BrokerBalanceError("transport_failed") from None + except (ValueError, TypeError): + raise BrokerBalanceError("malformed_response") from None + if not isinstance(payload, dict): + raise BrokerBalanceError("malformed_response") if str(payload.get("rt_cd")) != "0": - return [] - rows = payload.get("output1") or [] + raise BrokerBalanceError("broker_rejected") + rows = payload.get("output1") + if not isinstance(rows, list): + raise BrokerBalanceError("malformed_response") out: list[BalanceItem] = [] for row in rows: + if not isinstance(row, dict): + raise BrokerBalanceError("malformed_response") symbol = str(row.get("pdno") or "").strip() - qty = _safe_int(row.get("hldg_qty")) or 0 + qty = _safe_int(row.get("hldg_qty")) + if qty is None: + raise BrokerBalanceError("malformed_response") if not symbol or qty <= 0: continue out.append(BalanceItem(symbol=symbol, hldg_qty=qty)) diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index fc5f05a..a484759 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -134,6 +134,18 @@ class PostgreSQLSlotRepository: ) return [self._row_to_slot(r) for r in cur.fetchall()] + def find_managed_slot_by_symbol(self, instance_key: str, symbol: str) -> Optional[Slot]: + with self._psycopg.connect(self._dsn) as conn: + with conn.cursor() as cur: + cur.execute( + f"SELECT {_SLOT_COLUMNS} FROM daily_slot " + "WHERE instance_key = %s AND symbol = %s " + "ORDER BY updated_at DESC NULLS LAST, slot_id ASC LIMIT 1", + (instance_key, symbol), + ) + row = cur.fetchone() + return self._row_to_slot(row) if row else None + def reconcile_universe( self, instance_key: str, universe: dict[str, str], *, allow_retire: bool = True, diff --git a/trex_core/application/broker_port.py b/trex_core/application/broker_port.py index adac710..5437fcb 100644 --- a/trex_core/application/broker_port.py +++ b/trex_core/application/broker_port.py @@ -13,6 +13,21 @@ from decimal import Decimal from typing import Any, Optional, Protocol +class BrokerBalanceError(RuntimeError): + """Sanitized failure distinct from a validated empty broker balance.""" + + _CODES = frozenset({ + "token_issuance_failed", "authentication_failed", "authorization_failed", + "transport_failed", "malformed_response", "broker_rejected", + }) + + def __init__(self, code: str) -> None: + if code not in self._CODES: + raise ValueError(f"unsupported broker balance error code: {code}") + self.code = code + super().__init__(f"broker balance unavailable ({code})") + + @dataclass(frozen=True) class OrderRequest: slot_id: str # originating slot (audit/correlation) diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index 69d55e4..4aa470c 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -58,7 +58,8 @@ def reconcile_core_holdings( → set by 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. + Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_position'| + 'managed_non_holding_divergence'} for telemetry. """ broker_qty: dict[str, int] = { item.symbol: int(item.hldg_qty) for item in broker.get_balance() @@ -114,13 +115,23 @@ def reconcile_core_holdings( ) else: decisions[str(slot.slot_id)] = "ok" - for sym, bq in broker_qty.items(): # unmanaged broker position + for sym, bq in broker_qty.items(): # unmatched broker position if bq > 0 and sym not in matched: - decisions[f"broker:{sym}"] = "unmanaged_warning" - _log.warning( - "broker holds %s qty=%d with no core holding slot (unmanaged position)", - sym, bq, - ) + managed = slot_repo.find_managed_slot_by_symbol(instance_key, sym) + if managed is not None: + decisions[f"broker:{sym}"] = "managed_non_holding_divergence" + _log.warning( + "managed non-holding divergence: slot_id=%s portfolio_id=%s " + "symbol=%s status=%s instance_key=%s", + managed.slot_id, managed.portfolio_id, sym, + managed.status.value, instance_key, + ) + else: + decisions[f"broker:{sym}"] = "unmanaged_position" + _log.warning( + "unmanaged position: symbol=%s qty=%d instance_key=%s", + sym, bq, instance_key, + ) return decisions diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index c873e2a..dad3a68 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -51,6 +51,7 @@ class SnapshotScheduler: warn_store: Optional[ReconcileWarningStore] = None, credential_fetch_enabled: bool = False, kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = None, + credential_refresh: Optional[Callable[[], object]] = None, ) -> None: self._read = read_port self._store = store @@ -77,6 +78,8 @@ class SnapshotScheduler: # KISBroker with the fetched day-start credential. None = no reinit (paper # broker or credential_fetch_enabled=False). Set by daemon.py assembly. self._kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = kis_reinit_callback + # REQ-119: provider-owned refresh supersedes the legacy raw fetch hook. + self._credential_refresh = credential_refresh # REQ-113: detect break-glass at init time (env var set by operator). self._break_glass: bool = ( os.environ.get("TREX_KIS_BROKER_BREAK_GLASS", "").lower() @@ -250,7 +253,17 @@ class SnapshotScheduler: # Must run before portfolio snapshot so entry_allowed is set before # the first BuyIntent can arrive from the strategy daemon. if self._credential_fetch_enabled: - self._maybe_fetch_credential() + if self._credential_refresh is None: + self._maybe_fetch_credential() + else: + try: + self._credential_refresh() + self._entry_allowed = True + except Exception as exc: # provider exposes only sanitized failure + _log.error("DAY_START_LOAD_FAILED: broker credential refresh failed: %s (instance=%s)", + exc.__class__.__name__, self._instance_key) + self._entry_allowed = False + self._emit_day_start_alarm(reason="broker_credential_unavailable") if self._daemon_id is None: try: diff --git a/trex_core/domain/slot/in_memory_repository.py b/trex_core/domain/slot/in_memory_repository.py index 22de722..07b43dc 100644 --- a/trex_core/domain/slot/in_memory_repository.py +++ b/trex_core/domain/slot/in_memory_repository.py @@ -89,6 +89,19 @@ class InMemorySlotRepository: if s.instance_key == instance_key and s.status is SlotState.HOLDING ] + def find_managed_slot_by_symbol(self, instance_key: str, symbol: str) -> Optional[Slot]: + candidates = [ + s for s in self._slots.values() + if s.instance_key == instance_key and s.symbol == symbol + ] + # PostgreSQL parity: updated_at DESC NULLS LAST, slot_id ASC. + candidates.sort(key=lambda s: ( + s.updated_at is None, + -s.updated_at.timestamp() if s.updated_at is not None else 0, + str(s.slot_id), + )) + return candidates[0] if candidates else None + def reconcile_universe( self, instance_key: str, universe: dict[str, str], *, allow_retire: bool = True, diff --git a/trex_core/domain/slot/repository.py b/trex_core/domain/slot/repository.py index 1edd6c2..6365b47 100644 --- a/trex_core/domain/slot/repository.py +++ b/trex_core/domain/slot/repository.py @@ -72,6 +72,13 @@ class SlotRepository(Protocol): reconcile; safety-sell / kill-switch sweeps).""" ... + def find_managed_slot_by_symbol(self, instance_key: str, symbol: str) -> Optional[Slot]: + """Return one deterministic managed slot for exact instance+symbol. + + Read-only reconciliation triage; this must not imply remediation. + """ + ... + def reconcile_universe( self, instance_key: str, universe: dict[str, str], *, allow_retire: bool = True, diff --git a/trex_core/runtime/broker_provider.py b/trex_core/runtime/broker_provider.py new file mode 100644 index 0000000..024f714 --- /dev/null +++ b/trex_core/runtime/broker_provider.py @@ -0,0 +1,147 @@ +"""Common broker credential resolution and lifecycle boundary (REQ-119/120).""" +from __future__ import annotations + +import os +from dataclasses import dataclass +from datetime import date, datetime, timedelta, timezone +from typing import Literal, Protocol + +from trex_core.adapters.broker.paper import PaperBroker +from trex_core.application.broker_port import BrokerPort +from trex_core.application.portfolio import PortfolioHubUnavailableError +from trex_core.runtime.config import DaemonConfig + +_KST = timezone(timedelta(hours=9)) + + +@dataclass(frozen=True) +class BrokerSource: + """Non-secret credential-source metadata safe for CLI/log output.""" + mode: Literal["env", "fetched", "break_glass"] + identity_label: str + + +class BrokerCredentialUnavailable(RuntimeError): + _CODES = frozenset({"portfolio_hub_unavailable", "broker_absent", "trading_disabled", "malformed_credential", "broker_initialization_failed"}) + + def __init__(self, code: str) -> None: + if code not in self._CODES: + raise ValueError(f"unsupported broker credential error code: {code}") + self.code = code + super().__init__("broker credential unavailable") + + +class BrokerProvider(Protocol): + def get_broker_for_read(self) -> tuple[BrokerPort, BrokerSource]: ... + def refresh_day_start_credential(self) -> BrokerSource: ... + + +def _break_glass_enabled() -> bool: + return os.environ.get("TREX_KIS_BROKER_BREAK_GLASS", "").lower() in {"1", "true", "yes"} + + +def _make_env_broker(config: DaemonConfig) -> BrokerPort: + if config.broker != "kis": + seq_store = None + if config.paper.redis_url: + from trex_core.adapters.broker.seq_store import RedisSeqStore + seq_store = RedisSeqStore(config.paper.redis_url, key=f"trex:broker:paper:{config.instance_key}:order-seq") + return PaperBroker(fill_after_polls=config.paper.fill_after_polls, qty_per_poll=config.paper.qty_per_poll, + reject_symbols=set(config.paper.reject_symbols), seq_store=seq_store) + from trex_core.adapters.broker.kis import KISBroker + k = config.kis + if k is None: + raise BrokerCredentialUnavailable("broker_initialization_failed") + token_cache = None + if k.redis_url: + from trex_core.adapters.broker.token_cache import RedisTokenCache + token_cache = RedisTokenCache(k.redis_url) + return KISBroker(k.account_ref, app_key=k.app_key, app_secret=k.app_secret, base_url=k.base_url, + timeout=k.timeout, allow_live=k.allow_live, token_cache=token_cache) + + +class ConfigBrokerProvider: + """One provider owns one process/command broker and fetched-source refresh.""" + def __init__(self, config: DaemonConfig, *, read_port=None, clock=None) -> None: + self._config = config + self._read = read_port + self._clock = clock or (lambda: datetime.now(tz=timezone.utc)) + self._broker: BrokerPort | None = None + self._source: BrokerSource | None = None + self._credential_date: date | None = None + + def _read_port(self): + if self._read is None: + if self._config.portfolio_hub is None: + raise BrokerCredentialUnavailable("portfolio_hub_unavailable") + from trex_core.adapters.portfolio.hub_read import PortfolioHubReadAdapter + p = self._config.portfolio_hub + self._read = PortfolioHubReadAdapter(p.base_url, token=p.token or None, timeout=p.timeout) + return self._read + + def _fetch(self) -> BrokerSource: + try: + ctx = self._read_port().fetch_broker_credential() + except PortfolioHubUnavailableError: + raise BrokerCredentialUnavailable("portfolio_hub_unavailable") from None + except BrokerCredentialUnavailable: + raise + except Exception: + raise BrokerCredentialUnavailable("portfolio_hub_unavailable") from None + if ctx is None or getattr(ctx, "broker", None) is None: + raise BrokerCredentialUnavailable("broker_absent") + if not getattr(ctx, "trading_enabled", False): + raise BrokerCredentialUnavailable("trading_disabled") + credential = ctx.broker + if not all(isinstance(getattr(credential, x, None), str) and getattr(credential, x).strip() + for x in ("account_ref", "app_key", "app_secret")): + raise BrokerCredentialUnavailable("malformed_credential") + try: + if self._broker is None: + broker = _make_env_broker(self._config) + reinitialize = getattr(broker, "reinitialize", None) + if not callable(reinitialize): + raise RuntimeError("KIS broker lacks credential reload") + reinitialize(credential) + self._broker = broker + else: + self._broker.reinitialize(credential) # type: ignore[attr-defined] + except Exception: + raise BrokerCredentialUnavailable("broker_initialization_failed") from None + self._credential_date = self._clock().astimezone(_KST).date() + # Version is designated non-secret by the portfolio contract; no account/key material. + return BrokerSource("fetched", f"credential-version:{getattr(credential, 'credential_version', 'unknown')}") + + def get_broker_for_read(self) -> tuple[BrokerPort, BrokerSource]: + if self._config.broker != "kis": + if self._broker is None: + self._broker = _make_env_broker(self._config) + self._source = BrokerSource("env", "paper-config") + return self._broker, self._source + if not self._config.credential_fetch_enabled: + if self._broker is None: + self._broker = _make_env_broker(self._config) + self._source = BrokerSource("env", "kis-config") + return self._broker, self._source + if _break_glass_enabled(): + if self._broker is None: + self._broker = _make_env_broker(self._config) + self._source = BrokerSource("break_glass", "operator-break-glass") + return self._broker, self._source + source = self._fetch() + self._source = source + return self._broker, source # type: ignore[return-value] + + def refresh_day_start_credential(self) -> BrokerSource: + # Paper/env/break-glass paths deliberately do no portfolio-hub call. + if self._config.broker != "kis" or not self._config.credential_fetch_enabled or _break_glass_enabled(): + return self.get_broker_for_read()[1] + today = self._clock().astimezone(_KST).date() + if self._broker is not None and self._credential_date == today and self._source is not None: + return self._source + _, source = self.get_broker_for_read() + return source + + +def build_broker_provider(config: DaemonConfig, *, read_port=None) -> ConfigBrokerProvider: + return ConfigBrokerProvider(config, read_port=read_port) diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index 6cc3a1d..bd54670 100644 --- a/trex_core/runtime/cli.py +++ b/trex_core/runtime/cli.py @@ -329,6 +329,29 @@ def _fail(message: str, as_json: bool) -> None: print(f"error: {message}", file=sys.stderr) +def _broker_read_failure(exc, as_json: bool) -> int: + """REQ-120 command boundary: expose code only, never upstream payloads.""" + code = getattr(exc, "code", "broker_unavailable") + payload = {"error": {"code": code, "message": "broker balance unavailable"}} + if as_json: + print(_json.dumps(payload), file=sys.stderr) + else: + print(f"broker balance unavailable ({code})", file=sys.stderr) + return 1 + + +class _CachedBalanceBroker: + """One validated balance read per operator invocation; delegates everything else.""" + def __init__(self, broker, balance) -> None: + self._broker, self._balance = broker, balance + + def get_balance(self): + return self._balance + + def __getattr__(self, name): + return getattr(self._broker, name) + + # --------------------------------------------------------------------------- # # arg parsing # --------------------------------------------------------------------------- # @@ -446,7 +469,6 @@ def _build_parser() -> argparse.ArgumentParser: def _cmd_sync(args, cfg, confirm) -> int: """REQ-105 (dcy-034, BAT-051): sync slot qty to broker ground truth.""" - import psycopg from trex_core.adapters.portfolio.reconcile_warning_store import ( PostgreSQLReconcileWarningStore, @@ -476,11 +498,18 @@ def _cmd_sync(args, cfg, confirm) -> int: print("aborted") return 1 - warn_store = PostgreSQLReconcileWarningStore(cfg.dsn) + # REQ-119/120: resolve and validate broker truth before opening/enumerating DB. + from trex_core.runtime.broker_provider import ( + BrokerCredentialUnavailable, build_broker_provider, + ) + from trex_core.application.broker_port import BrokerBalanceError + try: + broker_port, _source = build_broker_provider(cfg).get_broker_for_read() + broker_port = _CachedBalanceBroker(broker_port, broker_port.get_balance()) + except (BrokerCredentialUnavailable, BrokerBalanceError) as exc: + return _broker_read_failure(exc, args.json) - # Build broker port - from trex_core.runtime.daemon import _build_broker - broker_port = _build_broker(cfg) + warn_store = PostgreSQLReconcileWarningStore(cfg.dsn) # Build slot_repo + slot_service from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository @@ -498,6 +527,7 @@ def _cmd_sync(args, cfg, confirm) -> int: ) # Determine target slot_ids + import psycopg if args.portfolio_id: with psycopg.connect(cfg.dsn) as conn: with conn.cursor() as cur: @@ -564,7 +594,7 @@ def _cmd_sync(args, cfg, confirm) -> int: print(f" after: status={after.get('status')} " f"qty={after.get('qty')} avg_price={after.get('avg_price')}") if cleared: - print(f" reconcile-warning CLEARED") + print(" reconcile-warning CLEARED") return exit_code @@ -579,12 +609,20 @@ def _cmd_verify(args, cfg) -> int: from trex_core.application.reconciliation import verify_slot_reconcile instance = _resolve_instance(args) + # REQ-120 no-write/no-enumeration barrier also applies to verify. + from trex_core.runtime.broker_provider import ( + BrokerCredentialUnavailable, build_broker_provider, + ) + from trex_core.application.broker_port import BrokerBalanceError + try: + broker_port, _source = build_broker_provider(cfg).get_broker_for_read() + broker_port = _CachedBalanceBroker(broker_port, broker_port.get_balance()) + except (BrokerCredentialUnavailable, BrokerBalanceError) as exc: + return _broker_read_failure(exc, args.json) + warn_store = PostgreSQLReconcileWarningStore(cfg.dsn) slot_repo = PostgreSQLSlotRepository(cfg.dsn) - from trex_core.runtime.daemon import _build_broker - broker_port = _build_broker(cfg) - # Build a minimal snapshot store (we only need asset_quantity for C-case check). # For verify we use a fresh SnapshotStore populated from warn_store metadata. # The hub_qty is read from the snapshot; if snapshot is empty we fall back to 0 @@ -787,7 +825,13 @@ def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int: else None ) _load_dotenv(_toml, getattr(args, "env_file", None)) - cfg = DaemonConfig.from_env() + cfg = ( + DaemonConfig.from_config( + resolve_config_toml(args.config, args.config_path), env_file=args.env_file + ) + if args.config + else DaemonConfig.from_env() + ) confirm = confirm or _default_confirm # REQ-088 (dcy-029): operator report — direct-DB point-in-time aggregate. diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index ca65079..612dd1a 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -74,6 +74,7 @@ _log = logging.getLogger(__name__) def _build_broker(config: DaemonConfig) -> BrokerPort: + """Backward-compatible env-only factory; new callers use broker_provider.""" if config.broker == "kis": from trex_core.adapters.broker.kis import KISBroker @@ -180,7 +181,7 @@ class Daemon: instance_key=config.instance_key, strategy_state_store=self.strategy_state_store, ) - self.broker = _build_broker(config) + self.outbox = PostgreSQLLedgerOutbox(config.dsn) # BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler. # (Created before the ledger writer so option-D write-back can target it.) @@ -193,6 +194,11 @@ class Daemon: # BAT-012: day-scoped kill-switch exclusion store + service. self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn) self.portfolio_read = _build_portfolio_read(config) + # REQ-119: one provider owns source resolution and the long-lived broker. + # Fetched mode never starts from an implicit env broker. + from trex_core.runtime.broker_provider import build_broker_provider + self.broker_provider = build_broker_provider(config, read_port=self.portfolio_read) + self.broker, self.broker_source = self.broker_provider.get_broker_for_read() # Event publisher transport — one dedicated instance for the # execution-loop thread (pika is not thread-safe). @@ -237,6 +243,7 @@ class Daemon: alarm_emitter=self.alarm_emitter, # dcy-033 / REQ-100 warn_store=self.warn_store, # BAT-050 / REQ-103: durable reconcile-warning credential_fetch_enabled=config.credential_fetch_enabled, # dcy-036 / REQ-113 + credential_refresh=self.broker_provider.refresh_day_start_credential, # dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: KISBroker reinit callback. # None when broker!=kis or credential_fetch_enabled=False (backward compat). kis_reinit_callback=_build_kis_reinit_callback(config, self.broker),