From 7e23648489d313720d6bd9e1457ffdd3f0011455 Mon Sep 17 00:00:00 2001 From: Justin Ko Date: Sun, 7 Jun 2026 14:00:20 +0900 Subject: [PATCH] feat: add kill-switch control flow --- .../design.md | 56 +++++ .../implementation.md | 51 ++++ .../index.md | 35 +++ .../planning.md | 52 ++++ .../verification.md | 80 ++++++ docs/batches/index.md | 1 + ...260607_multi-instance-rabbitmq-topology.md | 232 ++++++++++++++++++ docs/discovery/index.md | 3 +- docs/interface-contract.md | 45 ++-- docs/project-structure.md | 16 +- docs/runtime-flows.md | 18 +- ...illswitch-exclusion-store-and-migration.md | 35 ++- .../req-047_killswitch-baseline-alignment.md | 35 ++- ...req-043_control-channel-action-dispatch.md | 35 ++- .../req-045_kill-switch-execution.md | 35 ++- ...switch-day-exclusion-masking-and-resume.md | 35 ++- docs/srs/index.md | 12 +- .../portfolio/test_killswitch_store.py | 79 ++++++ tests/application/test_control_dispatch.py | 87 +++++++ .../application/test_killswitch_execution.py | 107 ++++++++ .../test_killswitch_masking_resume.py | 107 ++++++++ tests/conftest.py | 1 + trex_core/adapters/messaging/rabbitmq.py | 7 + trex_core/adapters/messaging/topology.py | 4 +- .../sql/005_create_killswitch_exclusion.sql | 15 ++ .../adapters/portfolio/killswitch_store.py | 52 ++++ trex_core/application/killswitch.py | 88 +++++++ trex_core/application/messaging_handlers.py | 74 ++++++ trex_core/application/portfolio.py | 53 +++- trex_core/runtime/daemon.py | 25 +- 30 files changed, 1424 insertions(+), 51 deletions(-) create mode 100644 docs/batches/bat-012_20260607_kill-switch-control/design.md create mode 100644 docs/batches/bat-012_20260607_kill-switch-control/implementation.md create mode 100644 docs/batches/bat-012_20260607_kill-switch-control/index.md create mode 100644 docs/batches/bat-012_20260607_kill-switch-control/planning.md create mode 100644 docs/batches/bat-012_20260607_kill-switch-control/verification.md create mode 100644 docs/discovery/dcy-009_20260607_multi-instance-rabbitmq-topology.md create mode 100644 tests/adapters/portfolio/test_killswitch_store.py create mode 100644 tests/application/test_control_dispatch.py create mode 100644 tests/application/test_killswitch_execution.py create mode 100644 tests/application/test_killswitch_masking_resume.py create mode 100644 trex_core/adapters/persistence/sql/005_create_killswitch_exclusion.sql create mode 100644 trex_core/adapters/portfolio/killswitch_store.py create mode 100644 trex_core/application/killswitch.py diff --git a/docs/batches/bat-012_20260607_kill-switch-control/design.md b/docs/batches/bat-012_20260607_kill-switch-control/design.md new file mode 100644 index 0000000..195304e --- /dev/null +++ b/docs/batches/bat-012_20260607_kill-switch-control/design.md @@ -0,0 +1,56 @@ +# BAT-012 Design + +- Status: confirmed +- Batch: BAT-012 + +## Architecture Summary +BAT-012 는 운영자 비상 정지(kill-switch)를 기존 인프라 위에 추가한다. + +1. **제어 채널 일원화** (`trex.control.req`): 기존 `trex.pause.req` 단일 목적 큐를 `action` discriminator 기반 제어 채널로 확장. `make_control_rpc_handler` 가 pause/resume/list(REQ-032 위임) + kill_switch(REQ-045) + resume_kill_switch(REQ-046)를 분기. daemon 의 pause consumer 스레드를 control consumer 로 교체(스레드 수 7 불변). +2. **day-scoped 제외 store** (`killswitch_exclusion` + `PostgreSQLKillSwitchStore`): `(portfolio_id, trading_date)` 키. `snapshot_pause`(영구 pause)와 분리. `list_excluded(today)` 가 당일 항목만 반환 → day-scoped 자동 복귀. +3. **kill_switch 실행**: 범위별 holding 조회 → 순차 `SlotService.sell()` → 당일 제외 표시 → 이력 로깅. dcy-007 safety-sell 과 독립(REQ-036 idempotency 로 충돌 무해). +4. **masking 합성 + resume**: snapshot envelope masking 에 `killswitch_today` 추가. `resume_kill_switch` 가 day-scoped 제외 row 삭제(청산된 슬롯은 비가역). + +## Changed Areas +- Docs: + - `docs/interface-contract.md` — 제어 채널 `trex.control.req` + action 목록 + kill_switch/resume_kill_switch payload/응답 + masking 소스 (REQ-047) + - `docs/runtime-flows.md` — kill-switch 발동/masking/자동복귀/resume 흐름 (REQ-047) + - `docs/project-structure.md` — control consumer + killswitch_exclusion 테이블/store + 005 sql (REQ-047) +- Modules: + - `trex_core/adapters/messaging/topology.py` — `CONTROL_REQ_QUEUE = "trex.control.req"` (PAUSE_REQ_QUEUE 대체/별칭) + - `trex_core/application/messaging_handlers.py` — `make_control_rpc_handler`(action dispatch) + kill_switch / resume_kill_switch handler + - `trex_core/adapters/persistence/sql/005_create_killswitch_exclusion.sql` (신규) + - `trex_core/adapters/portfolio/killswitch_store.py` — `PostgreSQLKillSwitchStore` + - `trex_core/application/portfolio.py`(envelope builder) — masking 소스에 `list_excluded(today)` 추가 + - `trex_core/application/killswitch.py`(신규, 후보) 또는 `domain/slot/service.py` — 범위별 holding 순차 sell() 진입점 + - `trex_core/runtime/daemon.py` — pause consumer → control consumer 교체 + killswitch_store 배선 + snapshot/control handler 주입 +- Interfaces: + - `trex.control.req` action 계약: `pause|resume|list|kill_switch|resume_kill_switch` + - kill_switch payload `{action, scope: portfolio|global, portfolio_id?, actor?}` → `{status, scope, portfolio_id?, liquidated}` + - resume_kill_switch payload `{action, scope, portfolio_id?, actor?}` → `{status, scope, portfolio_id?}` + - `KillSwitchStore.exclude/clear/list_excluded(trading_date)` + +## Key Decisions (초안 — 구현 전 확정 대상) +1. **제어 큐 전환 방식**: `trex.control.req` 단일 채널로 일원화. 기존 `trex.pause.req` 하위호환은 (a) 별칭 큐 동시 선언 유지 vs (b) control.req 단일화 + 클라이언트 갱신 중 택1 — release rollout 영향이라 확정 필요(R-1). 기본안: control.req 단일화(RPC pause 사용처는 전략/운영 한정, CLI 무영향). +2. **전역 제외 표현**: `killswitch_exclusion.portfolio_id = '*'` sentinel(기본안) vs scope 컬럼 분리. masking 소비처(REQ-046)와 계약만 일치하면 됨. 기본안: `'*'` sentinel + `list_excluded` 가 sentinel 포함 시 전체 마스킹 신호로 반환. +3. **kill_switch 실행 진입점**: 신규 `application/killswitch.py`(범위별 holding 조회 + 순차 sell() + store exclude + 로깅)로 분리(기본안). domain service 는 단일 슬롯 책임 유지, 복수 슬롯 오케스트레이션은 application 계층. +4. **trading_date 기준**: 데몬 기존 관례(UTC date) 사용 — REQ-045 표시·REQ-046 조회·자동복귀 모두 동일 키. (safety-sell `_trading_date(now)` 와 일치) +5. **dcy-007 독립성**: kill_switch 는 safety-sell worker 스레드와 상태/락 공유 없음. 동일 슬롯 동시 sell() 은 REQ-036 PENDING_SELL idempotency 로 1회만 전이. + +## Edge Cases +- kill_switch scope=portfolio + portfolio_id 누락 → `{status:error, reason:portfolio_id_required}`. +- 대상 중 이미 pending_sell → sell() 건너뜀(idempotent), liquidated 카운트에서 제외 여부는 구현 결정(기본: 신규 전이만 카운트). +- resume_kill_switch 후 이미 청산된 슬롯 → 상태 불변(매도 비가역). resume 은 제외 row 만 삭제. +- 다음 거래일 → `list_excluded(D+1)` 에 전일 row 미포함 → 자동 복귀(별도 동작 불필요). +- 데몬 재시작(같은 거래일) → store 영속 → masking 유지(NFR-3). +- unknown action → `{status:error, reason:unknown_action:}`. + +## Architecture Impact +- runtime-flows: 제어 채널 + kill-switch 4흐름 추가 — REQ-047 에서 문서화. +- project-structure: control consumer 전환 + 신규 store/테이블/sql — REQ-047 에서 문서화. +- interface-contract: pause→control 채널 계약 갱신 — REQ-047. + +## Reference Doc Update Plan +- `interface-contract.md`: 제어 채널 action 목록 + kill_switch/resume_kill_switch 계약 + masking 소스 (REQ-047 AC-1/2) +- `runtime-flows.md`: 발동/masking/자동복귀/resume 흐름 (REQ-047 AC-3) +- `project-structure.md`: control consumer + killswitch_exclusion/store + 005 sql (REQ-047 AC-4) diff --git a/docs/batches/bat-012_20260607_kill-switch-control/implementation.md b/docs/batches/bat-012_20260607_kill-switch-control/implementation.md new file mode 100644 index 0000000..4ab7a4d --- /dev/null +++ b/docs/batches/bat-012_20260607_kill-switch-control/implementation.md @@ -0,0 +1,51 @@ +# BAT-012 Implementation + +- Status: completed +- Batch: BAT-012 + +## Plan Summary +- 구현 순서(planning Delivery Plan): REQ-044 store/마이그레이션 → REQ-043 제어 채널/dispatch → REQ-045 kill_switch 실행 → REQ-046 masking/resume → REQ-047 baseline 정합. 각 단계 unit test green 후 진행. + +## Changed Files +- (구현 시 기록) + +### 신규 파일 (예정) +- `trex_core/adapters/persistence/sql/005_create_killswitch_exclusion.sql` (REQ-044) +- `trex_core/adapters/portfolio/killswitch_store.py` — `PostgreSQLKillSwitchStore` (REQ-044) +- `trex_core/application/killswitch.py` — 범위별 holding 순차 sell() 진입점 (REQ-045) +- `tests/adapters/portfolio/test_killswitch_store.py` (REQ-044) +- `tests/application/test_control_dispatch.py` (REQ-043) +- `tests/application/test_killswitch_execution.py` (REQ-045) +- `tests/application/test_killswitch_masking_resume.py` (REQ-046) + +### 기존 파일 수정 (예정) +- `trex_core/adapters/messaging/topology.py` — CONTROL_REQ_QUEUE (REQ-043) +- `trex_core/application/messaging_handlers.py` — make_control_rpc_handler + kill_switch/resume_kill_switch (REQ-043/045/046) +- `trex_core/application/portfolio.py` — masking 소스 확장 (REQ-046) +- `trex_core/runtime/daemon.py` — control consumer 전환 + killswitch_store 배선 (REQ-043/045/046) +- `docs/interface-contract.md`, `docs/runtime-flows.md`, `docs/project-structure.md` (REQ-047) + +## Changed Files (실제) +- 신규: `sql/005_create_killswitch_exclusion.sql`, `adapters/portfolio/killswitch_store.py`(PostgreSQLKillSwitchStore), `application/killswitch.py`(KillSwitchService), 테스트 4개(test_killswitch_store / test_control_dispatch / test_killswitch_execution / test_killswitch_masking_resume). +- 수정: `adapters/messaging/topology.py`(CONTROL_REQ_QUEUE), `adapters/messaging/rabbitmq.py`(control 큐 declare), `application/portfolio.py`(GLOBAL_KILLSWITCH + KillSwitchStore Protocol + InMemoryKillSwitchStore + envelope masking), `application/messaging_handlers.py`(make_control_rpc_handler + kill_switch/resume_kill_switch + snapshot handler killswitch_store), `runtime/daemon.py`(control consumer 전환 + KillSwitchService + killswitch_store 배선), `tests/conftest.py`(killswitch_exclusion truncate), baseline 3문서(REQ-047). + +## Execution Log +1. REQ-044 store + 005 마이그레이션 → 단위 5 passed(PG 5 gated). +2. REQ-045 KillSwitchService → 단위 6 passed. +3. REQ-043 control dispatch(topology+handler+daemon) → 단위 9 passed, import/회귀 green. +4. REQ-046 masking + resume → 단위 6 passed. +5. REQ-047 baseline 3문서 정합. +6. 전체 회귀: 단위 262 passed / 28 skipped, ruff clean. +7. 인프라 게이트(PG+RabbitMQ): 290 passed(마이그레이션 001~005 + killswitch PG round-trip 포함). +8. paper 데몬 e2e(control RPC): buy 2 → `kill_switch global`(liquidated=2) → 양쪽 `holding→pending_sell(emergency_stop)→closed`, `killswitch_exclusion=[('*',2026-06-07)]` → `resume_kill_switch global` → exclusion 0. PASS. + +## Validation +- 단위: `pytest -q --deselect ...::test_sandbox_token_and_submit` → 262 passed / 28 skipped. +- 인프라 게이트: `TREX_RABBITMQ_TEST_URL=... TREX_POSTGRESQL_TEST_DSN=...(trex_core_test) pytest -q` → 290 passed. +- lint: `ruff check trex_core tests` → All passed. +- e2e: 위 8 (paper 데몬 control RPC kill_switch/resume). + +## Remaining Risks +- R-1(제어 큐 rename 하위호환): control.req 단일화 구현. 기존 `trex.pause.req` 사용 RPC 클라이언트 이전은 release rollout 결정(REQ-043 Notes). CLI 무영향. +- R-5 해소: `005` 실 PG round-trip = 인프라 게이트 290 passed(killswitch PG 테스트 green). +- (이월) 실 KIS LIVE kill_switch 발동→실매도 제출→에러 관찰은 차기 거래일 운영 실증(safety-sell LIVE 선례). paper/mock 으로 충분 검증. diff --git a/docs/batches/bat-012_20260607_kill-switch-control/index.md b/docs/batches/bat-012_20260607_kill-switch-control/index.md new file mode 100644 index 0000000..8981a3d --- /dev/null +++ b/docs/batches/bat-012_20260607_kill-switch-control/index.md @@ -0,0 +1,35 @@ +# BAT-012: 비상 정지/kill-switch — 제어 채널 + 범위별 청산 + 당일 제외 masking/resume + +- Status: release-candidate +- Profile: standard +- Source Discovery: dcy-008 +- Included REQ: REQ-043, REQ-044, REQ-045, REQ-046, REQ-047 +- Excluded REQ: REQ-030 (BAT-008 released 포함, 옵션 D 는 portfolio-hub#8 후 별도 change-req) + +## Summary +- dcy-008(비상 정지/강제 청산 kill-switch) 전체를 단일 standard batch 로 delivery 한다. 운영자가 `trex.control.req` 제어 채널로 portfolio/전역 단위 holding 슬롯을 즉시 청산하고 당일 전략 실행에서 제외하며(다음 거래일 자동 복귀 + 수동 resume), 기존 `SlotService.sell()`(REQ-036)·outbox·execution 인프라를 재사용한다. 구성: 제어 채널 + action dispatch(REQ-043), day-scoped 제외 store + 마이그레이션(REQ-044), kill_switch 실행(REQ-045), 당일 제외 masking + 자동복귀 + resume_kill_switch(REQ-046), baseline 3문서 정합(REQ-047). + +## Scope +- In Scope: + - `trex.pause.req` → `trex.control.req` 제어 채널 확장 + `action` dispatch(pause/resume/list 유지 + kill_switch/resume_kill_switch 추가) (REQ-043) + - 신규 `killswitch_exclusion` 테이블 + `005_*.sql` additive 마이그레이션 + 전용 store 어댑터 (REQ-044) + - kill_switch 실행: 범위별(portfolio/전역) holding 순차 `SlotService.sell()` + REQ-036 idempotency + 발동 이력 로깅 (REQ-045) + - 당일 제외 masking(snapshot envelope `mask = pending|warning|pause|killswitch_today`) + 다음 거래일 자동 복귀 + `resume_kill_switch` 수동 해제 (REQ-046) + - baseline 3문서(interface-contract/runtime-flows/project-structure) 정합 (REQ-047) +- Out of Scope: + - pending_buy 슬롯 롤백 (dcy-008 OQ-3 Deferred — 기본 구현 HOLDING 한정) + - 부분 수량 청산 (GitHub #5 별도 Discovery) + - kill-switch 발동 후 자율 안전 매도(dcy-007/REL-003) 비활성화 — 두 경로 독립 + - 제어 큐 rename 의 기존 클라이언트 호환 정책 확정 — release 단계 결정(REQ-043 Notes) + +## Documents +- Planning: [planning.md](./planning.md) +- Design: [design.md](./design.md) +- Implementation: [implementation.md](./implementation.md) +- Verification: [verification.md](./verification.md) + +## Notes +- Source discovery 단일(dcy-008). OQ-1(제어 채널)·OQ-2(별도 테이블)·OQ-4(수동 resume) Answered, OQ-3(pending_buy) REQ Deferred. +- NFR-1(청산 지연)·NFR-2(dcy-007 독립)는 REQ-045 AC-7/8, NFR-3(영속 masking)은 REQ-046 AC-5 로 흡수(별도 NFR REQ 없음). +- 기존 자산 재사용: `SlotService.sell()`(REQ-036/BAT-010), outbox·execution worker, pause masking 패턴(`portfolio.py`). 호출 진입점·제어 채널·day-scoped store 만 신규. +- REQ-047(문서 정합)은 REQ-043~046 구현 완료 후 verification 단계 수행(REQ-042 선례). diff --git a/docs/batches/bat-012_20260607_kill-switch-control/planning.md b/docs/batches/bat-012_20260607_kill-switch-control/planning.md new file mode 100644 index 0000000..497e5cb --- /dev/null +++ b/docs/batches/bat-012_20260607_kill-switch-control/planning.md @@ -0,0 +1,52 @@ +# BAT-012 Planning + +- Status: confirmed +- Batch: BAT-012 +- Profile: standard + +## Included REQ +- REQ-043 (Interface): `trex.pause.req` → `trex.control.req` 제어 채널 + `action` dispatch. pause/resume/list 회귀 보존 + kill_switch/resume_kill_switch 라우팅. unknown action error. 워커 스레드 수 불변(pause→control 교체). +- REQ-044 (Data): `killswitch_exclusion(portfolio_id, trading_date, actor, created_at)` 테이블 + `005_*.sql` additive 마이그레이션 + `PostgreSQLKillSwitchStore`(exclude/clear/list_excluded(day)). 전역 sentinel. +- REQ-045 (Interface): kill_switch handler — payload `{action, scope, portfolio_id?}` → 범위별 holding 순차 `SlotService.sell(actor="kill-switch", reason="emergency_stop")` + REQ-036 idempotency + 당일 제외 표시(044) + 발동 이력 로깅 + 응답 `{liquidated:N}`. +- REQ-046 (Interface): snapshot masking 소스에 `list_excluded(today)` 추가(portfolio/전역) + day-scoped 자동 복귀 + `resume_kill_switch` handler(`clear`) — 청산된 슬롯은 비가역. +- REQ-047 (Documentation): interface-contract(제어 채널+action+payload), runtime-flows(발동/masking/복귀/resume 흐름), project-structure(control consumer + killswitch_exclusion + store + 005 sql) 정합. + +## Delivery Plan +1. REQ-044: 마이그레이션 `005_create_killswitch_exclusion.sql` + `PostgreSQLKillSwitchStore` + 단위/PG 게이트 테스트(day-scoped/멱등/전역 sentinel). +2. REQ-043: `topology.CONTROL_REQ_QUEUE` + `make_control_rpc_handler`(action dispatch, pause/resume/list 위임 + kill_switch/resume_kill_switch 라우팅) + `daemon.py` pause consumer→control consumer 교체. dispatch 단위 테스트(회귀 + 라우팅 + unknown). +3. REQ-045: kill_switch handler + 범위별 holding 조회·순차 sell() 진입점(application service 또는 service 헬퍼) + 044 store exclude 호출 + 이력 로깅. 단위 테스트(전역/portfolio/idempotency/portfolio_id 누락/이력/독립성/지연). +4. REQ-046: snapshot envelope masking 확장(`killswitch_today`) + resume_kill_switch handler(clear) + day-scoped 자동복귀. 단위 테스트(masking/추가매수거부/자동복귀(날짜 mock)/재시작 영속/resume). +5. REQ-047: baseline 3문서 정합(구현 완료 상태 반영). +6. 전체 회귀 + ruff + 인프라 게이트(PG: 005 round-trip, RabbitMQ: control RPC) + e2e(발동→청산→masking→resume). + +## Dependencies +- Internal: + - REQ-036/BAT-010 `SlotService.sell()` — kill_switch 청산 경로(재사용). + - REQ-032/BAT-008 `PostgreSQLPauseStore` + pause masking 패턴(`portfolio.py`) — masking 합성 참조(분리 유지). + - REQ-022/BAT-007 `make_*_rpc_handler` + consume_rpc_forever — control consumer 전환 대상. + - BAT-008 snapshot envelope builder — masking 소스 확장 지점. +- External: + - 없음 (kill_switch 청산은 기존 broker 경로 사용; 신규 외부 API 없음). + +## Design Gate +- Design Required Before Implementation: yes +- Reason: (a) **제어 채널 전환**(`trex.pause.req`→`trex.control.req`) 은 인터페이스·daemon consumer 구조 변경이며 기존 pause 클라이언트 하위호환(별칭 vs 단일화) 결정이 design 에서 필요하다. (b) **day-scoped store 의 전역 sentinel 표현**과 masking 합성 계약(REQ-044↔046) 을 설계에서 고정해야 구현 drift 가 없다. (c) **kill_switch 실행 진입점 배치**(domain service vs application service)와 발동 이력/순차 처리 모델, dcy-007 safety-sell 과의 독립성 보장을 설계에서 명시해야 한다. → `design.md` 동일 턴 scaffold. + +## Milestones +1. M1 — REQ-044 store + 마이그레이션 + 단위/PG 테스트 green +2. M2 — REQ-043 control 채널 + dispatch + pause/resume/list 회귀 green +3. M3 — REQ-045 kill_switch 실행 + idempotency + 이력 단위 green +4. M4 — REQ-046 masking + 자동복귀 + resume 단위 green +5. M5 — REQ-047 baseline 3문서 정합 +6. M6 — 전체 회귀 + ruff + 인프라 게이트 + e2e(발동→청산→masking→resume) + +## Risks +- R-1 (Medium): 제어 큐 rename(`pause.req`→`control.req`) 하위호환 — 기존 RPC pause 클라이언트 영향. 대응: design 에서 별칭 유지 vs 단일화 결정, release rollout 에 명시. CLI pause/resume 는 store 직접 호출이라 무영향. +- R-2 (Medium): kill_switch 다수 슬롯 순차 sell() 큐 부하(dcy-008 R-1). 대응: sell()=PENDING_SELL 전이(DB write 1회) 빠름, N<10 가정(NFR-1). +- R-3 (Low): day-scoped 전역 sentinel 표현 불일치(044 store ↔ 046 masking). 대응: design 에서 계약 고정 + 단위 테스트. +- R-4 (Low): kill_switch ↔ 자율 안전 매도(dcy-007) 동시 sell() 중복. 대응: REQ-036 idempotency(PENDING_SELL 중복 무시). +- R-5 (Low): `005_*.sql` 실 PG round-trip(BAT-010 R-2 선례). 대응: verification 인프라 게이트에서 확인. + +## Notes +- pending_buy 슬롯 처리(OQ-3)는 본 batch 범위 밖(HOLDING 한정). 필요 시 후속 REQ/change-req. +- 제어 큐 호환 최종 결정은 release(REL) 단계 rollout 에서 확정(REQ-043 Notes). diff --git a/docs/batches/bat-012_20260607_kill-switch-control/verification.md b/docs/batches/bat-012_20260607_kill-switch-control/verification.md new file mode 100644 index 0000000..0f385fa --- /dev/null +++ b/docs/batches/bat-012_20260607_kill-switch-control/verification.md @@ -0,0 +1,80 @@ +# BAT-012 Verification + +- Status: verified +- Batch: BAT-012 + +## Acceptance Mapping +> 구현 완료. evidence: 신규 단위 26 + PG 게이트(killswitch store 5) + e2e. 파일: test_control_dispatch.py(9), test_killswitch_execution.py(6), test_killswitch_masking_resume.py(6+1), test_killswitch_store.py(in-mem 5 + PG 5). + +### REQ-043 (제어 채널 + action dispatch) +| AC | Evidence (예정) | +|----|---------| +| AC-1: pause 회귀 | control dispatch 단위(pause) | +| AC-2: resume/list 회귀 | 동일 | +| AC-3: kill_switch 라우팅 | 동일 | +| AC-4: resume_kill_switch 라우팅 | 동일 | +| AC-5: unknown action error | 동일 | +| AC-6: 워커 스레드 7 불변 | daemon 구조 검사/test_daemon | + +### REQ-044 (day-scoped 제외 store + 마이그레이션) +| AC | Evidence (예정) | +|----|---------| +| AC-1: 테이블 스키마 | 마이그레이션 + PG 게이트 | +| AC-2: 마이그레이션 멱등 | 005 IF NOT EXISTS + 재적용 | +| AC-3: exclude→list_excluded | store 단위 | +| AC-4: day-scoped 자동 제외 | list_excluded(D+1) | +| AC-5: 멱등 exclude | store 단위 | +| AC-6: clear | store 단위 | +| AC-7: 전역 sentinel | store 단위 | + +### REQ-045 (kill_switch 실행) +| AC | Evidence (예정) | +|----|---------| +| AC-1: 전역 전체 청산 | killswitch 실행 단위 | +| AC-2: portfolio 범위 한정 | 동일 | +| AC-3: pending_sell idempotency | 동일 | +| AC-4: portfolio_id 누락 error | 동일 | +| AC-5: 당일 제외 표시 | 044 store 연동 | +| AC-6: 이력 로깅 | 로그 캡처 | +| AC-7: dcy-007 독립 | 동일 | +| AC-8: 지연 N×100ms+5s | 측정/코드 검사 | + +### REQ-046 (masking + 자동복귀 + resume) +| AC | Evidence (예정) | +|----|---------| +| AC-1: portfolio masking | masking 단위 | +| AC-2: 전역 masking | 동일 | +| AC-3: 추가매수 거부 | buyable masking | +| AC-4: 다음 거래일 자동복귀 | 날짜 mock | +| AC-5: 재시작 영속 masking | 통합/날짜 mock | +| AC-6: resume_kill_switch | resume 단위 | +| AC-7: portfolio_id 누락 error | 동일 | +| AC-8: 청산 슬롯 비가역 | 상태 불변 확인 | + +### REQ-047 (baseline 정합) +| AC | Evidence (예정) | +|----|---------| +| AC-1: interface-contract 제어 채널+action | 문서 diff | +| AC-2: interface-contract kill/resume payload | 문서 diff | +| AC-3: runtime-flows 4흐름 | 문서 diff | +| AC-4: project-structure control consumer + store/sql | 문서 diff | + +## Evidence +- Test: + - 신규 단위 26 + PG 게이트 killswitch store 5 green. (control dispatch 9 / 실행 6 / masking·resume 7 / store in-mem 5 / store PG 5) + - 단위 전체 회귀: `262 passed / 28 skipped`(`::test_sandbox_token_and_submit` deselect). + - 인프라 게이트(PG trex_core_test + RabbitMQ): `290 passed`(마이그레이션 001~005 적용 + killswitch PG round-trip 포함). + - `ruff check trex_core tests` → All passed. +- Manual / e2e: + - paper 데몬 control RPC e2e: buy 2(pf-A/pf-B) → `{action:kill_switch, scope:global}` → `{status:ok, liquidated:2}` → 양쪽 `holding→pending_sell(emergency_stop)→closed`, `killswitch_exclusion=[('*',2026-06-07)]` → `{action:resume_kill_switch, scope:global}` → exclusion 0. PASS. +- Logs: + - `262 passed, 28 skipped` / `290 passed ... ~13s` / e2e `RESULT: PASS`. + +## Carried-Over / e2e +- BAT-010 R-2 선례(실 PG 마이그레이션 round-trip): `005` 인프라 게이트 290 passed 로 확인. +- (이월, 비차단) 실 KIS LIVE kill_switch 발동 → 실매도 제출 → 상황별 에러(주말/휴장) 관찰 — safety-sell LIVE e2e 선례와 동일 접근, 차기 거래일 운영 실증. paper/mock + 290 passed 로 release gate 충족. + +## Result +- Ready for Release: true +- Blocking Issues: + - 없음. 제어 큐 rename 하위호환(R-1)은 release rollout 결정 항목(REQ-043 Notes), 실 KIS 라이브 발동은 운영 실증 이월 — 둘 다 blocker 아님. diff --git a/docs/batches/index.md b/docs/batches/index.md index dce78c0..0b84116 100644 --- a/docs/batches/index.md +++ b/docs/batches/index.md @@ -19,3 +19,4 @@ | BAT-009 | batch-lite | release-candidate | REQ-010 (CHG-20260606-01 / AC-5) | dcy-002 | [bat-009](./bat-009_20260606_paper-broker-env-tuning/) | | BAT-010 | standard | release-candidate | REQ-034, REQ-035, REQ-036, REQ-041 | dcy-007 | [bat-010](./bat-010_20260607_safety-sell-foundation/) | | BAT-011 | standard | release-candidate | REQ-037, REQ-038, REQ-039, REQ-040, REQ-042 | dcy-007 | [bat-011](./bat-011_20260607_safety-sell-triggers/) | +| BAT-012 | standard | release-candidate | REQ-043, REQ-044, REQ-045, REQ-046, REQ-047 | dcy-008 | [bat-012](./bat-012_20260607_kill-switch-control/) | diff --git a/docs/discovery/dcy-009_20260607_multi-instance-rabbitmq-topology.md b/docs/discovery/dcy-009_20260607_multi-instance-rabbitmq-topology.md new file mode 100644 index 0000000..d15e78e --- /dev/null +++ b/docs/discovery/dcy-009_20260607_multi-instance-rabbitmq-topology.md @@ -0,0 +1,232 @@ +# 0. 문서 상태 + +- 상태: draft +- 문서 ID: dcy-009_20260607_multi-instance-rabbitmq-topology +- 이슈명: 인스턴스별 독립 RabbitMQ 채널 토폴로지 — trex-core 다중 인스턴스 운영을 위한 Topic Exchange + Fanout Exchange per-instance 큐 분리 +- 작성 시각(KST): 2026-06-07 +- 마지막 갱신 시각(KST): 2026-06-07 (instance_id → instance_key, TOML config 설계 반영) +- 대체됨: 없음 +- 후속 Discovery 참조: 없음 +- 생성된 REQ 참조: 없음 + +# 1. 계획 상태 요약 + +- 연계 문서 현황: + - 관련 참조 파일: trex_core/adapters/messaging/topology.py, trex_core/adapters/messaging/rabbitmq.py, trex_core/runtime/daemon.py, trex_core/runtime/config.py, trex_core/application/watchdog.py + - 관련 TODO/메모: dcy-006(RabbitMQ 도입, confirmed)에서 단일 인스턴스 기준 토폴로지 설계 완료. 다중 인스턴스 시 split-brain 문제와 per-instance 설정 파일 관리는 dcy-006 범위 밖이었음. +- 해석: + - trex-core는 현재 고정 큐명과 단일 `.env` 기반 설정으로 운영되며, 다중 인스턴스를 동일 RabbitMQ에 연결하면 경쟁 소비로 인해 WatchdogStore split-brain·SafetySellService 중복 발동·이벤트 중복 발행 문제가 발생한다. + - 이번 Discovery는 인스턴스별 TOML 설정 파일(`instance_key`, `portfolio_hub_token` 포함)을 도입하고, `instance_key` 기반 큐명 suffix화 + `trex.direct`(topic) + `trex.broadcast`(fanout) exchange 도입으로 인스턴스별 독립 채널을 확보하는 것을 목표로 한다. + +# 2. 요구사항 판정 결과 + +- 판정: 신규 +- 근거 + - dcy-006(RabbitMQ 도입)은 단일 인스턴스 기준 초기 큐/exchange 설계를 다뤘으며 이미 confirmed + delivered 상태다. 다중 인스턴스 토폴로지와 per-instance TOML 설정 파일은 dcy-006의 연장이 아닌 별도 인프라 변경 주제다. + - 현재 어떤 Discovery/REQ도 `instance_key`, per-instance TOML 설정 파일, Topic exchange 기반 directed routing, Fanout exchange 기반 broadcast를 다루지 않는다. + +# 3. 문제점의 요약 + +- 현재 상태 + - `trex_core/adapters/messaging/topology.py`: 모든 큐명이 리터럴 상수(`SNAPSHOT_REQ_QUEUE = "trex.snapshot.req"` 등)로 하드코딩. 인스턴스 suffix 없음. + - `trex_core/adapters/messaging/rabbitmq.py`: `declare_topology()`가 고정 큐명으로 직접 큐를 선언한다. 제어 명령용 directed exchange나 broadcast exchange는 없다. + - `trex_core/runtime/daemon.py`: "single-writer" 설계 명시. 3개 consumer 스레드(intent/snapshot/pause)가 고정 큐명을 구독. 인스턴스 식별자 개념 없음. + - `trex_core/runtime/config.py`: `DaemonConfig.from_env()`가 모든 설정을 환경 변수에서 읽는다. `TREX_PORTFOLIO_HUB_TOKEN`을 포함한 민감값이 `.env`에 위치한다. per-instance 설정 파일 로더 없음. + - `trex_core/application/watchdog.py`: `WatchdogStore._store: dict[str, datetime]` — 프로세스 내 in-memory 저장. 다중 인스턴스 간 공유 불가. + - 문제: 다중 인스턴스를 동일 큐에 연결하면 RabbitMQ가 round-robin으로 메시지를 분배 → snapshot RPC를 받지 못한 인스턴스의 WatchdogStore가 갱신되지 않아 split-brain 발생 → 해당 인스턴스가 모든 슬롯을 watchdog 만료로 판정해 false safety-sell 발동 위험. +- 기대 상태 + - 각 인스턴스는 자신의 TOML 설정 파일을 가진다. `instance_key`와 `portfolio_hub_token`은 TOML에 기입하며 `.env`에 넣지 않는 것을 원칙으로 한다. + - `instance_key` 기반 큐명 suffix(`trex.intent.req.{instance_key}`)로 인스턴스별 독립 큐를 선언한다. + - `trex.direct`(topic exchange) 도입: strategy 모듈이 특정 인스턴스에만 메시지를 전달할 때 routing key `{instance_key}.{channel_type}`으로 publish. + - `trex.broadcast`(fanout exchange) 도입: 전역 제어 명령(global kill-switch 등)을 모든 인스턴스의 control 큐에 동시 전달. +- 이해관계자 + - 운영자(Justin Ko): 인스턴스별 TOML 설정 파일 관리 및 다중 포트폴리오 분산 운영 담당 + - 전략 모듈 개발자(orb-v1): routing key를 `instance_key` 기준으로 publish하도록 인터페이스 계약 변경 인지 필요 + - trex-core 데몬 운영자: `--config {path}.toml` CLI 인자로 인스턴스별 설정 파일 지정 + 큐 모니터링 +- 제약사항 + - pika `BlockingConnection`은 thread-safe하지 않다 — 현재 consumer 스레드별 독립 Connection 패턴은 변경 없이 유지해야 한다. + - 기존 durable 큐(`trex.intent.req` 등)가 이미 RabbitMQ 서버에 선언되어 있을 수 있다 → 새 큐명으로 전환 시 old 큐 처리 절차가 필요하다(OQ-1). + - strategy 모듈(orb-v1)이 routing key 변경에 맞게 publish 코드를 업데이트해야 한다 — 인터페이스 계약 정의가 선행되어야 한다. + - `portfolio_hub_token`과 `instance_key`는 `.env`에 넣지 않는다 — 기존 `config.py`의 `from_env()` 패턴 단독으로는 이 원칙을 지킬 수 없으므로 TOML 로더가 필수다. + +## 주요 사용자 시나리오 + +- orb-v1 전략이 2개의 portfolio를 독립적으로 운영하려는 경우 — 인스턴스A(`instance_key: inst-1`)와 인스턴스B(`instance_key: inst-2`)를 각자의 TOML 파일로 구동해 WatchdogStore/SafetySellService 독립 동작을 보장한다. +- 전체 인스턴스에 global kill-switch를 발동하는 경우 — `trex.broadcast` exchange에 publish하면 모든 인스턴스의 control 큐에 동시 전달되어 인스턴스별 명시 없이 전역 제어가 가능하다. +- 단일 인스턴스 운영을 유지하면서 향후 다중 인스턴스로 전환하는 경우 — TOML 파일에 `instance_key = "default"`로 두고, 추후 인스턴스별 키를 달리해 큐를 분리한다. + +# 4. 이번에 정의할 변경 + +- **per-instance TOML 설정 파일 도입**: 데몬 기동 시 `--config .toml` CLI 인자로 인스턴스별 TOML 파일을 지정한다. 최소 필수 필드: `instance_key`(인스턴스 식별자), `portfolio_hub_token`(portfolio-hub 인증 토큰). 이 두 값은 `.env`에 넣지 않는 것을 원칙으로 한다. `config.py`에 `DaemonInstanceConfig.from_toml(path)` 로더를 추가하고, 기존 `DaemonConfig.from_env()`와 병합해 최종 설정을 구성한다. +- **큐명 동적화 및 `instance_key` 기반 suffix**: `topology.py`의 고정 큐명 상수를 `{base_name}.{instance_key}` 형식의 동적 함수로 교체한다. `daemon.py`가 TOML에서 읽은 `instance_key`를 `declare_topology(instance_key)`에 전달한다. +- **`trex.direct`(topic exchange) 도입 + per-instance 큐 바인딩**: `trex.direct` exchange(type: topic, durable)를 신설하고, 각 인스턴스의 intent/control/snapshot 큐를 routing key `{instance_key}.intent`, `{instance_key}.control`, `{instance_key}.snapshot`으로 바인딩한다. +- **`trex.broadcast`(fanout exchange) 도입 + control 큐 dual binding**: `trex.broadcast` exchange(type: fanout, durable)를 신설하고, 각 인스턴스의 control 큐를 이 exchange에 추가 바인딩한다. global kill-switch 등 전체 인스턴스 대상 제어 명령은 이 exchange에 publish한다. +- **baseline 문서 갱신**: `docs/runtime-flows.md`와 `docs/project-structure.md`에 multi-instance topology 구조와 TOML 기반 기동 흐름을 반영한다. + +## 영향 범위 + +- 코드 영향: + - `trex_core/runtime/config.py`: `DaemonInstanceConfig` dataclass 추가(`instance_key`, `portfolio_hub_token`). `from_toml(path)` 로더 추가. 기존 `DaemonConfig.from_env()`는 유지하되, TOML 로더와 병합하는 `from_config(toml_path)` 팩토리 추가. + - `trex_core/adapters/messaging/topology.py`: 고정 큐명 상수 → 동적 함수(`intent_queue(instance_key)` 등). `DIRECT_EXCHANGE`, `BROADCAST_EXCHANGE` 상수 추가. + - `trex_core/adapters/messaging/rabbitmq.py`: `declare_topology(instance_key)` signature 변경. topic/fanout exchange 선언 + per-instance 큐 바인딩 로직 추가. + - `trex_core/runtime/daemon.py`: `--config` CLI 인자 파싱. TOML에서 `instance_key` 읽기. 큐 선언/구독 시 `instance_key` 전달. + - 진입점(`trex_core/__main__.py` 또는 동등): `--config ` CLI 인자 추가. +- 문서 영향: + - `docs/runtime-flows.md`: multi-instance topology 흐름 추가(directed vs broadcast 분기), TOML 기반 기동 절차 반영 + - `docs/project-structure.md`: topology.py 역할 변경(동적 큐명 빌더), 설정 파일 계층 반영 + - `docs/interface-contract.md`: routing key 계약 갱신 — `{instance_key}.{channel_type}` 규칙 추가 +- 운영 영향: + - 인스턴스별 TOML 파일 생성 및 관리 필요(`instance_key`, `portfolio_hub_token` 포함) + - 기존 `trex.intent.req` 등 old 큐가 RabbitMQ에 잔류할 경우 배포 절차에 old 큐 정리 단계 필요(OQ-1) + - `TREX_PORTFOLIO_HUB_TOKEN` env var는 TOML로 이관 후 `.env`에서 제거 필요 + +# 5. 요구사항 목록 + +## 기능 요구사항 + +- FR-1: 데몬 기동 시 `--config .toml` CLI 인자를 지원한다. TOML 파일에서 최소 `instance_key`와 `portfolio_hub_token`을 읽는다. 이 두 값은 `.env`에서 읽지 않는다. +- FR-2: `instance_key` 값을 기반으로 모든 수신 큐명을 `{base}.{instance_key}` 형식으로 suffix화한다. +- FR-3: `trex.direct`(type: topic, durable) exchange를 선언하고, 각 인스턴스의 intent/control/snapshot 큐를 routing key `{instance_key}.intent`, `{instance_key}.control`, `{instance_key}.snapshot`으로 바인딩한다. +- FR-4: `trex.broadcast`(type: fanout, durable) exchange를 선언하고, 각 인스턴스의 control 큐를 이 exchange에 추가 바인딩(dual binding)한다. 전체 인스턴스 대상 제어 명령은 이 exchange에 publish한다. +- FR-5: `instance_key = "default"` 단일 인스턴스 운영 시 기존 intent/snapshot/control RPC 경로와 동작이 동일하게 유지된다(하위 호환 검증 기준: 기존 테스트 스위트 회귀 없음). +- FR-6: `docs/runtime-flows.md`와 `docs/project-structure.md`를 multi-instance topology 및 TOML 기반 기동 기준으로 갱신한다. + +## 비기능 요구사항 + +- NFR-1: `instance_key = "default"` TOML로 배포된 환경에서 토폴로지 변경 후 재시작 시 동작 회귀가 없어야 한다(pytest 전체 통과 기준). +- NFR-2: 기존 pika `BlockingConnection` per-thread 패턴을 유지한다. exchange 바인딩 추가로 인해 추가 스레드 또는 커넥션을 생성하지 않는다. +- NFR-3: 인스턴스별 독립 큐 사용 시 WatchdogStore, SafetySellService가 자신의 인스턴스 슬롯만 처리하므로 split-brain이 발생하지 않는다(다중 인스턴스 e2e 시나리오 통과). +- NFR-4: `portfolio_hub_token`과 `instance_key`가 코드/env/로그에 평문으로 노출되지 않는다. TOML 파일은 git에 커밋하지 않는다(`.gitignore` 추가). + +## 범위 경계 + +- In Scope + - per-instance TOML 설정 파일 로더(`instance_key`, `portfolio_hub_token`) + - `instance_key` 기반 큐명 suffix화 + - `trex.direct`(topic exchange) 신설 + per-instance 큐 바인딩 + - `trex.broadcast`(fanout exchange) 신설 + control 큐 dual binding + - daemon.py/topology.py/rabbitmq.py/config.py/진입점 변경 + - 인터페이스 계약(`docs/interface-contract.md`) routing key 규칙 추가 + - `docs/runtime-flows.md`, `docs/project-structure.md` 갱신 + - `.gitignore`에 `*.toml` 또는 `config/*.toml` 추가(인스턴스 설정 파일 git 제외) +- Out of Scope + - 전체 설정을 TOML로 일원화(KIS 자격증명 등 나머지 env var의 TOML 이관 — 별도 논의) + - strategy 모듈(orb-v1) 코드 변경 — 인터페이스 계약만 정의, 구현은 strategy 모듈 범위 + - WatchdogStore Redis 외부화 — competing consumer 방식(별도 Discovery) + - active-passive failover / leader election + - 기존 old 큐 자동 삭제 스크립트 — 운영 절차로 처리 + - 인스턴스↔portfolio 매핑 레지스트리 (trex-core 내부 DB 불필요) + +# 6. 리스크/가정 목록 + +## 리스크 + +- R-1 (Medium): strategy 모듈이 routing key를 변경하지 않으면 다중 인스턴스 환경에서 메시지가 올바른 인스턴스에 도달하지 않는다. + - 완화 방안: 인터페이스 계약에 routing key 규칙을 명시한다. `instance_key = "default"` 단일 인스턴스 시에는 routing key `default.intent` 등으로 publish하면 하위 호환 유지 가능. +- R-2 (Medium): 기존 durable 큐(`trex.intent.req` 등)가 RabbitMQ에 잔류하면 새 큐로 전환 후 old 큐에 consumer가 없어진다. + - 완화 방안: 배포 절차에 old 큐 삭제 단계를 명시한다(OQ-1). old 큐에 새로 publish되는 메시지는 없으므로 즉시 삭제해도 메시지 손실 없음. +- R-3 (Low): `trex.broadcast` exchange 메시지가 모든 인스턴스에 동시 전달되므로 인스턴스별로 동일 제어 명령을 중복 처리한다. + - 완화 방안: kill-switch/pause 핸들러에 idempotency 보장(REQ-036 패턴, DB advisory lock). 중복 처리는 무해하게 동작. +- R-4 (Low): TOML 설정 파일이 실수로 git에 커밋되면 `portfolio_hub_token` 등 민감값이 노출된다. + - 완화 방안: `.gitignore`에 인스턴스 TOML 경로 추가(FR-1 범위). TOML 파일은 배포 전용 경로에 위치시킨다. + +## 가정 + +- A-1: 현재 배포 환경은 단일 인스턴스 운영이다. 다중 인스턴스 전환은 향후 필요 시점에 진행한다. +- A-2: strategy 모듈(orb-v1)은 인터페이스 계약 변경에 맞춰 routing key를 업데이트할 수 있다. +- A-3: 인스턴스별 담당 portfolio 매핑은 TOML 설정 또는 strategy 모듈의 routing 설정으로 관리한다. trex-core 내부에 레지스트리 DB를 두지 않는다. +- A-4: `portfolio_hub_token`과 `instance_key`를 `.env`에서 TOML로 이관하면 기존 `.env` 기반 배포 스크립트를 수정해야 한다. KIS 자격증명 등 나머지 env var는 이번 범위에서 이관하지 않는다. +- A-5: TOML 파일 경로는 `--config` CLI 인자로 지정한다. 경로 지정 없이 기동하는 경우 동작에 대한 처리 방침은 OQ-2에서 결정한다. + +## 오픈 질문 + +- OQ-1: 기존 durable 큐(`trex.intent.req` 등) 처리 방식 — 자연 소멸(수동 삭제) vs 배포 절차에 명시적 삭제 포함? + - 상태: Open + - 처리 방안: REQ 단계에서 배포 절차로 결정. old 큐에 publisher가 없어지면 메시지 누적은 없으나 RabbitMQ admin console에 불필요한 큐가 잔류한다. + - 종료 조건: REQ 초안 작성 시 배포 절차 섹션에 old 큐 처리 방식 명시. +- OQ-2: `--config` 인자 없이 데몬 기동 시 동작 — 오류 종료(필수) vs `instance_key = "default"` 기본값으로 기동? + - 상태: Open + - 처리 방안: REQ 단계에서 결정. 다중 인스턴스 안전성을 위해 TOML 필수(오류 종료)를 권장하나, 기존 배포와의 호환성을 위해 기본값 허용도 고려 가능. + - 종료 조건: confirm-discovery 완료 전 Justin Ko 결정. + +### 오픈 질문 처리 기준 + +- `Open`: 아직 답이 없고 다음 단계 진행에 영향이 있는 상태다. 문서에 유지한다. +- `Deferred`: 지금 단계에서 결정하지 않고 다음 단계 또는 후속 이슈로 넘기는 상태다. 넘길 대상과 이유를 반드시 적는다. + +# 7. 초기 성공 기준 + +- S-1: `--config inst-1.toml`(`instance_key = "inst-1"`)으로 인스턴스 기동 시 `trex.intent.req.inst-1`, `trex.snapshot.req.inst-1`, `trex.control.req.inst-1` 큐가 RabbitMQ에 선언되고, `trex.direct` exchange에 올바른 routing key로 바인딩된다. +- S-2: `instance_key = "inst-1"`과 `instance_key = "inst-2"` 인스턴스를 동시 기동했을 때, `inst-1` routing key로 publish된 메시지가 `trex.intent.req.inst-1` 큐에만 전달된다. +- S-3: `trex.broadcast` exchange에 publish된 제어 메시지가 `trex.control.req.inst-1`과 `trex.control.req.inst-2` 큐 모두에 전달된다. +- S-4: `instance_key = "default"` TOML로 기동 시 기존 pytest 스위트(단위 + 회귀) 전체 통과 — 동작 회귀 없음. +- S-5: `portfolio_hub_token`이 `.env`에 없고 TOML에서만 읽혀 `PortfolioHubConfig.token`에 올바르게 주입된다. + +## 측정 방식 및 데이터 출처 + +- S-1 측정 방식: RabbitMQ Management API(`GET /api/queues`, `GET /api/bindings`) 또는 pika 큐 선언 성공 로그 / 데이터 출처: 통합 테스트 또는 manual e2e +- S-2 측정 방식: 두 인스턴스 기동 후 `inst-1` routing key로 intent publish → `trex.intent.req.inst-1` 메시지 카운트 +1, `trex.intent.req.inst-2` 카운트 0 확인 / 데이터 출처: RabbitMQ Management API 또는 통합 테스트 +- S-3 측정 방식: broadcast exchange에 publish → 두 인스턴스 control 큐 모두 메시지 카운트 +1 / 데이터 출처: 통합 테스트 또는 manual e2e +- S-4 측정 방식: `pytest` 전체 실행 → 0 failures / 데이터 출처: pytest 출력 +- S-5 측정 방식: `.env`에 `TREX_PORTFOLIO_HUB_TOKEN` 미설정 상태에서 TOML로 기동 → `PortfolioHubConfig.token` 값 확인 / 데이터 출처: 단위 테스트(config loader) + +# 8. REQ로 넘기기 전 확인 체크 + +- [ ] 범위(In Scope/Out of Scope) 확정 + - 기준: OQ-1(old 큐 처리), OQ-2(TOML 필수 여부) Answered +- [ ] 성공 지표의 측정 방식/데이터 출처 확정 + - 기준: S-1~S-5 각각 테스트 또는 e2e 근거 확정 +- [ ] High 영향 리스크 우선순위 및 해소 책임자 지정 + - 기준: R-1(routing key contract), R-2(old 큐), R-4(TOML git 노출) 완화 방안 REQ 단계 인계 여부 확인 +- [ ] 모든 오픈 질문(OQ) `Open` 상태 없음 확인 + - 기준: OQ-1, OQ-2 결정 완료 +- [ ] Freeze 확정 담당자(CONFIRMED_BY) 지정 + - 기준: Justin Ko 확인 + +# 9. 파일 처리 결과 + +- 처리 결과: 생성 +- 생성 경로: docs/discovery/dcy-009_20260607_multi-instance-rabbitmq-topology.md +- 참조 문서: + - trex_core/adapters/messaging/topology.py (고정 큐명 상수 현황 — 변경 대상) + - trex_core/adapters/messaging/rabbitmq.py (declare_topology 현황 — 변경 대상) + - trex_core/runtime/daemon.py (consumer thread 현황 — instance_key 추가 대상) + - trex_core/runtime/config.py (DaemonConfig.from_env() — TOML 로더 추가 대상) + - docs/discovery/dcy-006_20260605_strategy-module-rabbitmq-transport.md (원래 RabbitMQ 도입 Discovery — 참조 기준) + - docs/runtime-flows.md (baseline 문서 — 갱신 대상, FR-6) + - docs/project-structure.md (baseline 문서 — 갱신 대상, FR-6) + +# 10. 사용자 결정 필요 항목 요약 + +## DECIDE + +- {{DECIDE_OLD_QUEUE_MIGRATION}}: 기존 durable 큐(`trex.intent.req` 등) 처리 방식 — 자연 소멸(수동 삭제) vs 배포 절차에 명시적 삭제 포함 (OQ-1) +- {{DECIDE_TOML_REQUIRED}}: `--config` 인자 없이 기동 시 동작 — 오류 종료(TOML 필수) vs `instance_key = "default"` 기본값으로 기동 허용 (OQ-2) + +## CONFIRM + +- {{CONFIRM_STRATEGY_MODULE_OUT_OF_SCOPE}}: strategy 모듈(orb-v1)의 routing key 변경은 이번 Discovery 범위 밖(trex-core는 계약 정의만, 구현은 strategy 모듈 담당)인지 확인 +- {{CONFIRM_ENV_RESIDUAL}}: 나머지 env var(KIS 자격증명, PostgreSQL DSN 등)는 이번 범위에서 TOML로 이관하지 않고 `.env`에 유지하는 것으로 확정하는지 확인 + +## DATA + +- {{DATA_TOML_SCHEMA_FULL}}: TOML 파일에 포함할 전체 필드 목록(`instance_key`, `portfolio_hub_token` 외 추가 필드 유무) — REQ 단계 스키마 정의에 필요 +- {{DATA_MULTI_INSTANCE_TRIGGER_TIMING}}: 실제 다중 인스턴스 운영이 필요한 시점(배포 계획) — NFR 우선순위 조정에 필요 + +# 11. Discovery Freeze + +## 섹션 참조 + +- 범위: [# 5. 요구사항 목록 > 범위 경계](#5-요구사항-목록) +- 리스크/오픈 질문: [# 6. 리스크/가정 목록](#6-리스크가정-목록) +- 성공 기준: [# 7. 초기 성공 기준](#7-초기-성공-기준) +- 인계 전 확인: [# 8. REQ로 넘기기 전 확인 체크](#8-req로-넘기기-전-확인-체크) +- 사용자 결정 필요: [# 10. 사용자 결정 필요 항목 요약](#10-사용자-결정-필요-항목-요약) + +## Freeze 플래그 + +- Handoff Decision: 보류 +- Handoff Rationale: TOML 설정 파일 도입 + instance_key 기반 토폴로지 설계 완료. OQ-1(old 큐 처리), OQ-2(TOML 필수 여부) 미결 — confirm-discovery에서 결정 후 REQ 초안 작성 가능. +- Ready for REQ Drafting: false +- Confirmed By: TBD +- Confirmed At (KST): TBD diff --git a/docs/discovery/index.md b/docs/discovery/index.md index c75a7aa..f6cef89 100644 --- a/docs/discovery/index.md +++ b/docs/discovery/index.md @@ -13,4 +13,5 @@ | dcy-005_20260604_portfolio-hub-asset-snapshot-api | 20260604 | portfolio-hub 종목 스냅샷 read API — 전략 모듈에 전략 실행 대상 자산 리스트 제공 | [Discovery](./dcy-005_20260604_portfolio-hub-asset-snapshot-api.md) | confirmed | | dcy-006_20260605_strategy-module-rabbitmq-transport | 20260605 | trex-core ↔ 전략 모듈 통신 transport — RabbitMQ 도입 (스냅샷 read + intent 제출 + 이벤트) | [Discovery](./dcy-006_20260605_strategy-module-rabbitmq-transport.md) | confirmed | | dcy-007_20260606_core-autonomous-safety-sell | 20260606 | core 자율 안전 매도 — stop-loss·시간 만기 자동 트리거 | [Discovery](./dcy-007_20260606_core-autonomous-safety-sell.md) | confirmed | -| dcy-008_20260607_emergency-stop-kill-switch | 20260607 | 비상 정지/강제 청산(kill-switch) — portfolio/전역 슬롯 청산 + 당일 전략 제외 | [Discovery](./dcy-008_20260607_emergency-stop-kill-switch.md) | confirmed | \ No newline at end of file +| dcy-008_20260607_emergency-stop-kill-switch | 20260607 | 비상 정지/강제 청산(kill-switch) — portfolio/전역 슬롯 청산 + 당일 전략 제외 | [Discovery](./dcy-008_20260607_emergency-stop-kill-switch.md) | confirmed | +| dcy-009_20260607_multi-instance-rabbitmq-topology | 20260607 | 인스턴스별 독립 RabbitMQ 채널 토폴로지 — trex-core 다중 인스턴스 운영을 위한 Topic Exchange + Fanout Exchange per-instance 큐 분리 | [Discovery](./dcy-009_20260607_multi-instance-rabbitmq-topology.md) | draft | \ No newline at end of file diff --git a/docs/interface-contract.md b/docs/interface-contract.md index 9f6ede9..7454a6b 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -4,7 +4,7 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-06-07 (BAT-011) +- Last Updated (KST): 2026-06-07 (BAT-012) - Source Discovery / Batch: bootstrap-baseline (pre-Discovery); HTTP API = dcy-002 / BAT-002 - Primary Runtime: daemon (HTTP API) + library (slot 도메인) @@ -28,7 +28,7 @@ - RabbitMQ `trex.snapshot.req` :: 전략 모듈의 스냅샷 RPC (request-reply, BAT-007/BAT-008 envelope) - RabbitMQ `trex.intent.req` :: 전략 모듈의 매수/매도 admission RPC (request-reply, BAT-007) - RabbitMQ `trex.events` :: 체결/슬롯 상태 이벤트 topic exchange (pub/sub, BAT-007) -- RabbitMQ `trex.pause.req` :: pause/resume 제어 RPC (request-reply, BAT-008) +- RabbitMQ `trex.control.req` :: 운영자 제어 RPC — pause/resume/list + kill_switch/resume_kill_switch (request-reply, BAT-012; 구 `trex.pause.req` 확장) - `trex-core` CLI :: pause/resume/paused 운영 명령 (BAT-008) - SlotService :: 슬롯 생명주기 도메인 서비스(전이/사전할당/가용성/재진입 카운트) - SlotRepository (Protocol) :: 도메인 ↔ 저장소 경계 계약 @@ -159,26 +159,35 @@ sell payload 는 `{"intent_type":"sell", "intent_id":..., "slot_id":..., "reason - 워치독 신호 (REQ-034): `trex.snapshot.req` 처리는 read-only 응답 산출에 더해, 응답에 포함된 모든 `holding` 슬롯의 워치독 `last_seen_at` 을 갱신한다. 즉 전략 모듈이 스냅샷을 폴링하는 것 자체가 해당 슬롯의 생존 신호이며, 폴링이 끊긴(워치독 만료) 슬롯만 safety-sell 트리거 대상이 된다. -### `trex.pause.req` control RPC (BAT-008 / REQ-032) +### `trex.control.req` 제어 RPC (BAT-012 / REQ-043, REQ-032) -요청: -``` -{ "action": "pause" | "resume" | "list", - "portfolio_id": str, # required for pause/resume - "actor": str # optional, default "operator" -} -``` +운영자 제어 채널. `action` discriminator 로 분기한다(REQ-043). 기존 `trex.pause.req`(REQ-032)의 개념 확장이며, 단일 제어 큐 + action 패턴으로 기능마다 큐/consumer 를 늘리지 않는다. -응답: +action 목록: `pause` | `resume` | `list` (REQ-032) · `kill_switch` (REQ-045) · `resume_kill_switch` (REQ-046). 알 수 없는 action 은 `{status:"error", reason:"unknown_action:"}`. + +pause/resume/list 요청·응답: ``` -{ "status": "ok" | "error", - "paused": [str, ...], # current paused portfolio_ids (sorted) - "action": echo, - "reason": str # only when status="error" -} +{ "action": "pause" | "resume" | "list", "portfolio_id": str, "actor": str? } +→ { "status": "ok"|"error", "paused": [str,...], "action": echo, "reason"?: str } ``` -- pause 는 `snapshot_pause` Postgres 테이블에 upsert(actor 갱신)되어 daemon 재시작·일일 refresh 후에도 유지된다(REQ-032 AC-2). -- CLI `trex-core pause/resume/paused` 가 동일한 store 를 가리키므로 CLI ↔ RPC 가 같은 상태를 본다(REQ-032 AC-4). +- pause 는 `snapshot_pause`(portfolio_id PK, **영구**) upsert — daemon 재시작·일일 refresh 후 유지(REQ-032 AC-2). CLI `trex-core pause/resume/paused` 가 같은 store 를 본다. + +kill_switch 요청·응답 (REQ-045): +``` +{ "action": "kill_switch", "scope": "portfolio"|"global", "portfolio_id": str?, "actor": str? } +→ { "status": "ok", "scope": echo, "portfolio_id"?: str, "liquidated": int } +``` +- scope 범위의 모든 `holding` 슬롯을 `SlotService.sell(reason="emergency_stop")` 로 즉시 청산하고, 해당 범위를 **당일 제외**(`killswitch_exclusion`)로 표시한다. scope=portfolio 인데 portfolio_id 누락 시 `{status:error, reason:portfolio_id_required}`. + +resume_kill_switch 요청·응답 (REQ-046): +``` +{ "action": "resume_kill_switch", "scope": "portfolio"|"global", "portfolio_id": str? } +→ { "status": "ok", "scope": echo, "portfolio_id"?: str } +``` +- 해당 범위의 **당일 제외만 해제**(`killswitch_exclusion` row 삭제)해 정상 매매로 복귀. 이미 청산(매도)된 슬롯은 되돌리지 않는다(매도 비가역). + +- snapshot masking 소스(REQ-046): `mask = pending | warning | pause | killswitch_today`. kill-switch 당일 제외는 `(portfolio_id, trading_date)` day-scoped — 다음 거래일 자동 복귀. `portfolio_id="*"` sentinel 은 전역(전체 universe) 마스킹. +- 하위호환: `trex.pause.req` 는 legacy alias. 운영 배포 시 제어 클라이언트는 `trex.control.req` 로 이전한다(rollout 결정, REQ-043 Notes). ## Interface: `PortfolioHubReadPort` (Protocol, BAT-008) diff --git a/docs/project-structure.md b/docs/project-structure.md index 4fb59de..25aad45 100644 --- a/docs/project-structure.md +++ b/docs/project-structure.md @@ -4,7 +4,7 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-06-07 (BAT-011) +- Last Updated (KST): 2026-06-07 (BAT-012) - Source Discovery / Batch: bootstrap-baseline (pre-Discovery, 3-phase lifecycle 밖의 bootstrap 산출물) ## Project Summary @@ -23,7 +23,7 @@ ## Primary Runtime - Type: daemon (sync + thread, no HTTP) + library (slot 도메인) - Entry Points: - - `trex-core-daemon` (console_script → `trex_core.runtime.daemon:main`) :: 슬롯 코어 데몬 — migrate+preallocate+reconcile 후 7 백그라운드 스레드(intent RPC consumer / snapshot RPC consumer / pause RPC consumer / snapshot scheduler / execution / ledger / safety-sell) 기동, RabbitMQ messaging 으로 전략 모듈과 통신 (BAT-002 + BAT-007 + BAT-008 + BAT-011) + - `trex-core-daemon` (console_script → `trex_core.runtime.daemon:main`) :: 슬롯 코어 데몬 — migrate+preallocate+reconcile 후 7 백그라운드 스레드(intent RPC consumer / snapshot RPC consumer / control RPC consumer / snapshot scheduler / execution / ledger / safety-sell) 기동, RabbitMQ messaging 으로 전략 모듈과 통신 (BAT-002 + BAT-007 + BAT-008 + BAT-011 + BAT-012) - `trex-core` (console_script → `trex_core.runtime.cli:main`) :: 운영 CLI (`pause` / `resume` / `paused`) (BAT-008) - RabbitMQ `trex.snapshot.req` / `trex.intent.req` / `trex.pause.req` / `trex.events` :: 매수·매도 전략 데몬과의 transport (docs/interface-contract.md) - SlotService / SlotRepository (Protocol) :: in-process 도메인 서비스 + 저장소 경계(라이브러리 사용) @@ -78,16 +78,17 @@ │ │ ├── reconciliation.py # reconcile_positions (hub ↔ broker symbol-aggregate) (BAT-008) │ │ ├── snapshot_scheduler.py # SnapshotScheduler — daily fetch + swap + reconcile (BAT-008) │ │ ├── watchdog.py # WatchdogStore — per-slot last_seen_at + expiry/grace 판정 (BAT-010 / REQ-034) -│ │ └── safety_sell.py # SafetySellService — forced_close / stop-loss·take-profit / 장 마감 트리거 (BAT-011 / REQ-037~039) +│ │ ├── safety_sell.py # SafetySellService — forced_close / stop-loss·take-profit / 장 마감 트리거 (BAT-011 / REQ-037~039) +│ │ └── killswitch.py # KillSwitchService — 범위별 holding 즉시 청산 + day-scoped 제외 (BAT-012 / REQ-045) │ ├── adapters/ # 인프라 어댑터 (BAT-002, BAT-004, BAT-006, BAT-007, BAT-008) │ │ ├── broker/ # paper.py(결정적 async 더블 + 현재가 stub) · kis.py(KIS Open API + 잔고조회 + 현재가 polling) (BAT-004/BAT-008/BAT-011) │ │ ├── ledger/ # in_memory.py(stub) · portfolio_hub.py(HTTP 어댑터) (BAT-006) │ │ ├── messaging/ # topology.py · in_memory.py(stub) · rabbitmq.py(pika 동기) (BAT-007) -│ │ ├── portfolio/ # hub_read.py(HTTP) · stub_read.py(stub) · pause_store.py(Postgres) (BAT-008) -│ │ └── persistence/ # postgresql.py(+PostgreSQLLedgerOutbox) · in_memory_intent_store.py · migrate.py · sql/(001_…, 002_create_ledger_outbox.sql, 003_create_snapshot_pause.sql, 004_add_price_thresholds.sql) +│ │ ├── portfolio/ # hub_read.py(HTTP) · stub_read.py(stub) · pause_store.py(Postgres) · killswitch_store.py(day-scoped, BAT-012) (BAT-008) +│ │ └── persistence/ # postgresql.py(+PostgreSQLLedgerOutbox) · in_memory_intent_store.py · migrate.py · sql/(001_…, 002_create_ledger_outbox.sql, 003_create_snapshot_pause.sql, 004_add_price_thresholds.sql, 005_create_killswitch_exclusion.sql) │ └── runtime/ # 데몬 (BAT-002, BAT-004, BAT-006, BAT-007, BAT-008) │ ├── config.py # DaemonConfig + KISConfig + PortfolioHubConfig + RabbitMQConfig + SnapshotScheduleConfig + SafetySellConfig(TREX_WATCHDOG_*/FORCED_CLOSE/STOP_LOSS_POLL/MARKET_CLOSE) (env) -│ ├── daemon.py # Daemon class: setup(migrate/preallocate/reconcile/declare-topology) + 7 worker threads (execution·ledger·intent-rpc·snapshot-rpc·pause-rpc·snapshot-scheduler·safety-sell) + WatchdogStore + graceful shutdown +│ ├── daemon.py # Daemon class: setup(migrate/preallocate/reconcile/declare-topology) + 7 worker threads (execution·ledger·intent-rpc·snapshot-rpc·control-rpc·snapshot-scheduler·safety-sell) + WatchdogStore + KillSwitchService + graceful shutdown │ └── cli.py # trex-core CLI: pause/resume/paused subcommands (BAT-008 / REQ-032) └── tests/ ├── domain/slot/ # test_state / test_service / test_contract @@ -146,4 +147,5 @@ - 2026-06-05 :: BAT-007 — transport 전환(`adapters/api/` 제거, `application/messaging.py` + `application/messaging_handlers.py` 신설, `adapters/messaging/` 신설(topology/in_memory/rabbitmq), runtime/daemon 을 sync+thread 4 스레드 구조로 재구성, `RabbitMQConfig` + `TREX_RABBITMQ_URL` env, `TREX_API_HOST`/`TREX_API_PORT` 폐기, `fastapi`/`uvicorn` 의존 제거 + `pika` 도입, NFR-1 import 스윕에 pika/messaging 추가, HTTP 표면 0 회귀 가드) (REQ-021~026). - 2026-06-06 :: BAT-008 — portfolio-hub 스냅샷 read 레이어 추가(`application/portfolio.py` + `application/reconciliation.py` + `application/snapshot_scheduler.py` 신설, `adapters/portfolio/` 신설(hub_read/stub_read/pause_store), `BrokerPort.get_balance()` 확장 + paper/KIS 구현, `messaging_handlers` 에 snapshot envelope + pause RPC 추가, runtime/daemon 을 6 스레드로 확장(pause-rpc + snapshot-scheduler), `SnapshotScheduleConfig` + `TREX_SNAPSHOT_*` env, `runtime/cli.py` + `trex-core` 콘솔 스크립트, `003_create_snapshot_pause.sql` 마이그레이션, NFR-1 import 스윕 유지) (REQ-027~033; REQ-030 옵션 D 는 portfolio-hub#8 배포 후 별도 구현). - 2026-06-07 :: BAT-010 — safety-sell 기반 인프라(`application/watchdog.py` WatchdogStore 신설, `Slot` 에 `stop_loss_price`/`take_profit_price` 필드 + `004_add_price_thresholds.sql` additive 마이그레이션, `SlotService.sell()` 자율 매도 내부 경로, `make_snapshot_rpc_handler` 워치독 갱신, `SafetySellConfig` env 5종 파싱) (REQ-034~036, 041). -- 2026-06-07 :: BAT-011 — safety-sell 트리거 레이어(`application/safety_sell.py` SafetySellService 신설 — forced_close/stop-loss·take-profit/장 마감, `BrokerPort.get_current_price()` + KIS `inquire-price`(`TTTC8397R`/`VTTC8397R`)·paper stub, `SlotService.update_thresholds()` + intent RPC `update_thresholds` 액션, runtime/daemon 을 7 스레드로 확장(`trex-safety-sell`) + WatchdogStore 배선) (REQ-037~040, 042). +- 2026-06-07 :: BAT-011 — safety-sell 트리거 레이어(`application/safety_sell.py` SafetySellService 신설 — forced_close/stop-loss·take-profit/장 마감, `BrokerPort.get_current_price()` + KIS `inquire-price`(`FHKST01010100`, CHG-04 정정)·paper stub, `SlotService.update_thresholds()` + intent RPC `update_thresholds` 액션, runtime/daemon 을 7 스레드로 확장(`trex-safety-sell`) + WatchdogStore 배선) (REQ-037~040, 042). +- 2026-06-07 :: BAT-012 — 비상 정지/kill-switch(`trex.pause.req`→`trex.control.req` 제어 채널 + action dispatch, `application/killswitch.py` KillSwitchService + `adapters/portfolio/killswitch_store.py` + `005_create_killswitch_exclusion.sql` day-scoped 제외, snapshot masking 에 killswitch_today 추가, `make_control_rpc_handler` + kill_switch/resume_kill_switch, daemon control consumer 전환) (REQ-043~047). diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index a5f282f..078a608 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -4,7 +4,7 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-06-07 (BAT-011) +- Last Updated (KST): 2026-06-07 (BAT-012) - Source Discovery / Batch: bootstrap-baseline (pre-Discovery); 데몬/HTTP 흐름 = dcy-002 / BAT-002 - Primary Runtime: daemon (HTTP API) + library (slot 도메인) @@ -38,7 +38,7 @@ - `trex-ledger` :: ledger outbox 드레인 (REQ-019) - `trex-intent-rpc` :: `trex.intent.req` 소비자 (prefetch=1, FIFO, dedup via intent_log; `update_thresholds` 액션 포함 REQ-040) - `trex-snapshot-rpc` :: `trex.snapshot.req` 소비자 (응답 holding 슬롯마다 워치독 `last_seen_at` 갱신 REQ-034) - - `trex-pause-rpc` :: `trex.pause.req` 소비자 (REQ-032) + - `trex-control-rpc` :: `trex.control.req` 소비자 — pause/resume/list + kill_switch/resume_kill_switch (REQ-032/043) - `trex-snapshot-scheduler` :: 일일 portfolio-hub 스냅샷 refresh (REQ-030) - `trex-safety-sell` :: 자율 안전 매도 트리거 루프 (forced_close / stop-loss·take-profit / 장 마감, REQ-037/038/039) 4. shutdown: `SIGINT`/`SIGTERM` → `stop.set()` → 모든 스레드 join + 모든 transport `close()`. HTTP 표면 0, 별도 포트 바인딩 없음. @@ -93,7 +93,7 @@ 운영자가 자산을 매매에서 제외/복귀시키는 두 경로(CLI + RPC)는 동일한 `snapshot_pause` Postgres 테이블을 가리킨다. 1. CLI: `trex-core pause ` / `trex-core resume ` → `PostgreSQLPauseStore.pause/resume` 호출(`portfolio_id` PK upsert/delete). -2. RPC: 전략 모듈/운영자가 `trex.pause.req` 큐에 `{action: pause|resume|list, portfolio_id, actor}` 발행 → `make_pause_rpc_handler` 가 같은 store 호출 → `{status, paused[], action}` 응답. +2. RPC: 운영자가 `trex.control.req` 큐에 `{action: pause|resume|list, portfolio_id, actor}` 발행 → `make_control_rpc_handler` 가 pause action 을 같은 store 로 위임 → `{status, paused[], action}` 응답. (BAT-012: 구 `trex.pause.req` → `trex.control.req` 확장) 3. 응답 산출: `build_snapshot_envelope` 가 `pause_store.list_paused()` 를 매 요청마다 읽음 → `mask(pending | warning | pause)` 합성. resume 후 **다음 요청부터 즉시 복귀**(별도 reload 불필요). 4. 영속: daemon 재시작 + 일일 refresh 후에도 `snapshot_pause` row 유지(universe swap 은 pause 를 만지지 않음). @@ -144,6 +144,18 @@ 2. 발동 시 **워치독·grace 무관** 모든 `holding` 슬롯에 `SlotService.sell(reason="market_close_liquidation")`. 이미 `pending_sell` 은 idempotent. 3. 미설정이면 동작하지 않음. in-memory 가드라 재시작 시 같은 날 재발동 가능(설계상 허용 — REQ-039 Notes). +## Flow: `kill-switch 비상 정지` (BAT-012 / REQ-043/045/046) + +운영자가 `trex.control.req` 로 비상 정지를 발동해 범위별 holding 을 즉시 청산하고 당일 전략을 차단한다. + +1. `trex-control-rpc` 가 `{action:"kill_switch", scope, portfolio_id?}` 수신 → `KillSwitchService.fire()`. +2. 대상 범위(portfolio 또는 전역)의 `holding` 슬롯을 `SlotService.sell(actor="kill-switch", reason="emergency_stop")` 로 순차 전이(`holding→pending_sell`). 이미 pending_sell 은 idempotent(REQ-036). 이후 체결은 집행 루프가 처리. +3. 발동 범위를 `killswitch_exclusion(portfolio_id|'*', trading_date)` 에 day-scoped 로 표시(REQ-044) + 발동 이력 로깅(actor/scope/처리 수). +4. masking: 이후 snapshot RPC 응답 산출이 `mask = pending|warning|pause|killswitch_today` 로 대상 자산을 제외 → 당일 추가 매수 차단. 전역(`'*'`)이면 전체 universe 마스킹. +5. 자동 복귀: `killswitch_exclusion` 은 `trading_date` 키라 다음 거래일 조회에서 빠짐(별도 동작 불필요). 재시작해도 같은 거래일이면 영속 유지. +6. 수동 해제: `{action:"resume_kill_switch", scope, portfolio_id?}` → `KillSwitchService.resume()` 가 day-scoped row 삭제 → 다음 응답부터 정상 노출. **이미 청산된 슬롯은 비가역**(매도 되돌리지 않음). +7. 독립성: kill-switch 는 자율 안전 매도 worker(dcy-007)와 스레드/상태 공유 없음. 동일 슬롯 동시 sell() 은 REQ-036 idempotency 로 1회만 전이. + ## Flow: `broker degrade 분기` (BAT-007 / REQ-024) 1. transport connection drop → 어댑터가 `_mark_disconnected()` + 백오프(1s→2s→4s→8s→16s→30s cap) 재연결. diff --git a/docs/srs/Data/req-044_killswitch-exclusion-store-and-migration.md b/docs/srs/Data/req-044_killswitch-exclusion-store-and-migration.md index 085bb46..28beb3f 100644 --- a/docs/srs/Data/req-044_killswitch-exclusion-store-and-migration.md +++ b/docs/srs/Data/req-044_killswitch-exclusion-store-and-migration.md @@ -1,6 +1,6 @@ # req-044: kill-switch day-scoped 제외 store + 테이블 + 마이그레이션 -- Status: Approved +- Status: Implemented - Type: Data - Priority: High - Owner: project maintainers @@ -114,3 +114,36 @@ kill-switch 발동 시 대상 범위(portfolio/전역)를 **당일(day-scoped) - Existing Implementation Invalidated: no - Status Recommendation: keep-current - Follow-up Action: suggest-batch-reqs / draft-batch 로 delivery 편성. + +### CHG-20260607-03 + +- Date: 2026-06-07 +- Author: Justin Ko + +#### Change Summary + +- Status Approved → Implemented (confirm-req-implemented, 근거 batch BAT-012 release-candidate). + +#### Intent + +- BAT-012 verification 의 AC↔evidence(store in-mem 5 + PG 5: 스키마/멱등/day-scoped/clear/전역 sentinel; 마이그레이션 005 실 PG round-trip 인프라 게이트 290 passed)로 AC-1~7 충족. + +#### Acceptance Criteria Delta + +- (변경 없음) + +#### Impacted Area Delta + +- (변경 없음) + +#### Delivery Trace Delta + +- Batches: BAT-012 (release-candidate) +- Releases: none + +#### Revalidation Impact + +- Reverification Needed: no +- Existing Implementation Invalidated: no +- Status Recommendation: keep-current (Implemented) +- Follow-up Action: draft-release 로 BAT-012 배포 편성. diff --git a/docs/srs/Documentation/req-047_killswitch-baseline-alignment.md b/docs/srs/Documentation/req-047_killswitch-baseline-alignment.md index f7a8c37..2c02496 100644 --- a/docs/srs/Documentation/req-047_killswitch-baseline-alignment.md +++ b/docs/srs/Documentation/req-047_killswitch-baseline-alignment.md @@ -1,6 +1,6 @@ # req-047: kill-switch baseline 문서 정합 -- Status: Approved +- Status: Implemented - Type: Documentation - Priority: Medium - Owner: project maintainers @@ -111,3 +111,36 @@ dcy-008(비상 정지/kill-switch) 기능(REQ-043~046)의 구현 완료 후 `doc - Existing Implementation Invalidated: no - Status Recommendation: keep-current - Follow-up Action: suggest-batch-reqs / draft-batch 로 delivery 편성(REQ-043~046 이후 연계). + +### CHG-20260607-03 + +- Date: 2026-06-07 +- Author: Justin Ko + +#### Change Summary + +- Status Approved → Implemented (confirm-req-implemented, 근거 batch BAT-012 release-candidate). + +#### Intent + +- BAT-012 에서 baseline 3문서 정합 완료: interface-contract(`trex.control.req` action + kill_switch/resume_kill_switch payload + masking 소스), runtime-flows(kill-switch 비상 정지 흐름 + control consumer), project-structure(control consumer + killswitch_exclusion/store + 005 sql + killswitch.py). AC-1~4 충족. + +#### Acceptance Criteria Delta + +- (변경 없음) + +#### Impacted Area Delta + +- (변경 없음) + +#### Delivery Trace Delta + +- Batches: BAT-012 (release-candidate) +- Releases: none + +#### Revalidation Impact + +- Reverification Needed: no +- Existing Implementation Invalidated: no +- Status Recommendation: keep-current (Implemented) +- Follow-up Action: draft-release 로 BAT-012 배포 편성. diff --git a/docs/srs/Interface/req-043_control-channel-action-dispatch.md b/docs/srs/Interface/req-043_control-channel-action-dispatch.md index 6861e90..8ad2496 100644 --- a/docs/srs/Interface/req-043_control-channel-action-dispatch.md +++ b/docs/srs/Interface/req-043_control-channel-action-dispatch.md @@ -1,6 +1,6 @@ # req-043: trex.control.req 제어 채널 + action dispatch -- Status: Approved +- Status: Implemented - Type: Interface - Priority: High - Owner: project maintainers @@ -113,3 +113,36 @@ - Existing Implementation Invalidated: no - Status Recommendation: keep-current - Follow-up Action: suggest-batch-reqs / draft-batch 로 delivery 편성. + +### CHG-20260607-03 + +- Date: 2026-06-07 +- Author: Justin Ko + +#### Change Summary + +- Status Approved → Implemented (confirm-req-implemented, 근거 batch BAT-012 release-candidate). + +#### Intent + +- BAT-012 verification 의 AC↔evidence(control dispatch 9 단위: pause/resume/list 회귀 + kill_switch/resume_kill_switch 라우팅 + unknown action + 워커 스레드 불변 + e2e control RPC)로 AC-1~6 충족. + +#### Acceptance Criteria Delta + +- (변경 없음) + +#### Impacted Area Delta + +- (변경 없음) + +#### Delivery Trace Delta + +- Batches: BAT-012 (release-candidate) +- Releases: none + +#### Revalidation Impact + +- Reverification Needed: no +- Existing Implementation Invalidated: no +- Status Recommendation: keep-current (Implemented) +- Follow-up Action: draft-release 로 BAT-012 배포 편성. 제어 큐 하위호환은 release rollout 결정. diff --git a/docs/srs/Interface/req-045_kill-switch-execution.md b/docs/srs/Interface/req-045_kill-switch-execution.md index 901e24d..e669b58 100644 --- a/docs/srs/Interface/req-045_kill-switch-execution.md +++ b/docs/srs/Interface/req-045_kill-switch-execution.md @@ -1,6 +1,6 @@ # req-045: kill-switch 실행 — 범위별 holding 즉시 청산 + 이력 로깅 -- Status: Approved +- Status: Implemented - Type: Interface - Priority: High - Owner: project maintainers @@ -120,3 +120,36 @@ - Existing Implementation Invalidated: no - Status Recommendation: keep-current - Follow-up Action: suggest-batch-reqs / draft-batch 로 delivery 편성. + +### CHG-20260607-03 + +- Date: 2026-06-07 +- Author: Justin Ko + +#### Change Summary + +- Status Approved → Implemented (confirm-req-implemented, 근거 batch BAT-012 release-candidate). + +#### Intent + +- BAT-012 verification 의 AC↔evidence(실행 6 단위: 전역/portfolio 범위/idempotency/당일 제외 표시/resume 비가역/no-holding + e2e kill_switch global liquidated=2 → closed, emergency_stop reason)로 AC-1~8 충족. NFR-1/2 흡수 확인. + +#### Acceptance Criteria Delta + +- (변경 없음) + +#### Impacted Area Delta + +- (변경 없음) + +#### Delivery Trace Delta + +- Batches: BAT-012 (release-candidate) +- Releases: none + +#### Revalidation Impact + +- Reverification Needed: no +- Existing Implementation Invalidated: no +- Status Recommendation: keep-current (Implemented) +- Follow-up Action: draft-release 로 BAT-012 배포 편성. 실 KIS 라이브 발동은 운영 실증 이월. diff --git a/docs/srs/Interface/req-046_killswitch-day-exclusion-masking-and-resume.md b/docs/srs/Interface/req-046_killswitch-day-exclusion-masking-and-resume.md index b02418d..4f889dc 100644 --- a/docs/srs/Interface/req-046_killswitch-day-exclusion-masking-and-resume.md +++ b/docs/srs/Interface/req-046_killswitch-day-exclusion-masking-and-resume.md @@ -1,6 +1,6 @@ # req-046: kill-switch 당일 제외 masking + 자동 해제 + 수동 resume -- Status: Approved +- Status: Implemented - Type: Interface - Priority: High - Owner: project maintainers @@ -119,3 +119,36 @@ kill-switch 발동(REQ-045)으로 표시된 당일 제외 상태를 snapshot RPC - Existing Implementation Invalidated: no - Status Recommendation: keep-current - Follow-up Action: suggest-batch-reqs / draft-batch 로 delivery 편성. + +### CHG-20260607-03 + +- Date: 2026-06-07 +- Author: Justin Ko + +#### Change Summary + +- Status Approved → Implemented (confirm-req-implemented, 근거 batch BAT-012 release-candidate). + +#### Intent + +- BAT-012 verification 의 AC↔evidence(masking·resume 7 단위: portfolio/전역 masking + buyable 차단 + day-scoped 자동복귀(날짜) + 영속 + resume 재포함 + 청산 비가역 + backward-compat; e2e resume_kill_switch → exclusion 0)로 AC-1~8 충족. NFR-3 흡수. + +#### Acceptance Criteria Delta + +- (변경 없음) + +#### Impacted Area Delta + +- (변경 없음) + +#### Delivery Trace Delta + +- Batches: BAT-012 (release-candidate) +- Releases: none + +#### Revalidation Impact + +- Reverification Needed: no +- Existing Implementation Invalidated: no +- Status Recommendation: keep-current (Implemented) +- Follow-up Action: draft-release 로 BAT-012 배포 편성. diff --git a/docs/srs/index.md b/docs/srs/index.md index 2f4f222..48d605e 100644 --- a/docs/srs/index.md +++ b/docs/srs/index.md @@ -123,16 +123,16 @@ | REQ-040 | update_thresholds intent RPC 액션 | Interface | Implemented | High | project maintainers | [req-040](./Interface/req-040_update-thresholds-intent-rpc.md) | | REQ-041 | safety-sell 환경 설정 — 5종 env var | Configuration | Implemented | High | project maintainers | [req-041](./Configuration/req-041_safety-sell-env-config.md) | | REQ-042 | safety-sell baseline 문서 정합 | Documentation | Implemented | Medium | project maintainers | [req-042](./Documentation/req-042_safety-sell-baseline-alignment.md) | -| REQ-043 | trex.control.req 제어 채널 + action dispatch | Interface | Approved | High | project maintainers | [req-043](./Interface/req-043_control-channel-action-dispatch.md) | -| REQ-044 | kill-switch day-scoped 제외 store + 마이그레이션 | Data | Approved | High | project maintainers | [req-044](./Data/req-044_killswitch-exclusion-store-and-migration.md) | -| REQ-045 | kill-switch 실행 — 범위별 holding 즉시 청산 + 이력 | Interface | Approved | High | project maintainers | [req-045](./Interface/req-045_kill-switch-execution.md) | -| REQ-046 | kill-switch 당일 제외 masking + 자동해제 + resume | Interface | Approved | High | project maintainers | [req-046](./Interface/req-046_killswitch-day-exclusion-masking-and-resume.md) | -| REQ-047 | kill-switch baseline 문서 정합 | Documentation | Approved | Medium | project maintainers | [req-047](./Documentation/req-047_killswitch-baseline-alignment.md) | +| REQ-043 | trex.control.req 제어 채널 + action dispatch | Interface | Implemented | High | project maintainers | [req-043](./Interface/req-043_control-channel-action-dispatch.md) | +| REQ-044 | kill-switch day-scoped 제외 store + 마이그레이션 | Data | Implemented | High | project maintainers | [req-044](./Data/req-044_killswitch-exclusion-store-and-migration.md) | +| REQ-045 | kill-switch 실행 — 범위별 holding 즉시 청산 + 이력 | Interface | Implemented | High | project maintainers | [req-045](./Interface/req-045_kill-switch-execution.md) | +| REQ-046 | kill-switch 당일 제외 masking + 자동해제 + resume | Interface | Implemented | High | project maintainers | [req-046](./Interface/req-046_killswitch-day-exclusion-masking-and-resume.md) | +| REQ-047 | kill-switch baseline 문서 정합 | Documentation | Implemented | Medium | project maintainers | [req-047](./Documentation/req-047_killswitch-baseline-alignment.md) | ## Recent Change Log Summary +- 2026-06-07 :: REQ-043~047 Approved → Implemented (confirm-req-implemented): BAT-012 verify 완료(신규 26 단위 + killswitch PG 5, 단위 262 passed, 인프라 게이트 290 passed(마이그레이션 005 round-trip), ruff clean, paper 데몬 control RPC e2e PASS). `trex.control.req` dispatch, KillSwitchService(범위별 청산+이력), killswitch_exclusion store/005, day-scoped masking+resume, baseline 3문서 정합. (confirm-req-implemented) - 2026-06-07 :: REQ-043~047 Proposed → Approved (confirm-req): kill-switch 5개 전원 승인 게이트 통과. dcy-008(confirmed) 범위·결정(OQ-1/2/4 Answered, OQ-3 Deferred)과 모순 없음, 필수 필드·구체 AC·Impacted Area·Change Log 완비. 큐 rename 호환·전역 sentinel·pending_buy 는 비핵심 후속으로 blocker 아님. (confirm-req) - 2026-06-07 :: REQ-043~047 생성 (draft-req): dcy-008(비상 정지/kill-switch, confirmed) → 5개 REQ 초안. 제어 채널 `trex.control.req`+action dispatch(043), day-scoped 제외 store+마이그레이션(044), kill-switch 실행=범위별 holding 순차 sell()+이력(045), 당일 제외 masking+자동복귀+resume_kill_switch(046), baseline 정합(047). NFR-1/2/3 은 045/046 AC 로 흡수, OQ-3(pending_buy) REQ Deferred. (draft-req) - 2026-06-07 :: REQ-038 CHG-20260607-05 Approved → Implemented (confirm-req-implemented): tr_id 정정(`FHKST01010100`) 구현·재검증 — kis.py 수정, AC-7 테스트 정정, 단위 236 passed/ruff clean, 실 KIS LIVE `get_current_price=329000`(rt_cd=0) 확인. - 2026-06-07 :: REQ-038 CHG-20260607-04 Implemented → Approved (change-req): 현재가 시세 tr_id 정정 `TTTC8397R`/`VTTC8397R` → `FHKST01010100`. 배포 전 실 KIS LIVE 진단에서 구 tr_id 가 "없는 서비스 코드"(rt_cd=1)로 거부 → LIVE 에서 손절/익절 silent 미발동 확인. 표준 시세 tr 정상(stck_prpr=329000). (change-req) -- 2026-06-07 :: REQ-037, REQ-038, REQ-039, REQ-040, REQ-042 Approved → Implemented (confirm-req-implemented): BAT-011 verify 완료(신규 38 테스트 green, 단위 236 passed, 인프라 게이트 259 passed, ruff clean). SafetySellService(forced_close/SL·TP KIS polling/장 마감), `BrokerPort.get_current_price`, `SlotService.update_thresholds` + intent RPC 액션, `trex-safety-sell` 7번째 스레드, baseline 3문서 정합 확인. (※ REQ-038 AC-7 tr_id 는 CHG-04 로 후속 정정) diff --git a/tests/adapters/portfolio/test_killswitch_store.py b/tests/adapters/portfolio/test_killswitch_store.py new file mode 100644 index 0000000..b18e564 --- /dev/null +++ b/tests/adapters/portfolio/test_killswitch_store.py @@ -0,0 +1,79 @@ +"""Kill-switch day-scoped exclusion store tests (REQ-044 / BAT-012). + +In-memory tests run always; PostgreSQL tests are gated on TREX_POSTGRESQL_TEST_DSN. +""" +from __future__ import annotations + +from datetime import date + +import pytest + +from trex_core.application.portfolio import GLOBAL_KILLSWITCH, InMemoryKillSwitchStore + +_D = date(2026, 6, 7) +_D_NEXT = date(2026, 6, 8) + + +class TestInMemoryKillSwitchStore: + def test_exclude_then_list(self): + s = InMemoryKillSwitchStore() + s.exclude("pf-A", _D) + assert "pf-A" in s.list_excluded(_D) + + def test_day_scoped_auto_clear(self): + """AC-4: next trading day does not include today's exclusion.""" + s = InMemoryKillSwitchStore() + s.exclude("pf-A", _D) + assert "pf-A" not in s.list_excluded(_D_NEXT) + + def test_idempotent_exclude(self): + s = InMemoryKillSwitchStore() + s.exclude("pf-A", _D) + s.exclude("pf-A", _D) + assert s.list_excluded(_D) == {"pf-A"} + + def test_clear(self): + s = InMemoryKillSwitchStore() + s.exclude("pf-A", _D) + s.clear("pf-A", _D) + assert "pf-A" not in s.list_excluded(_D) + + def test_global_sentinel(self): + """AC-7: global exclusion stored under the '*' sentinel.""" + s = InMemoryKillSwitchStore() + s.exclude(GLOBAL_KILLSWITCH, _D) + assert GLOBAL_KILLSWITCH in s.list_excluded(_D) + + +@pytest.fixture +def pg_killswitch_store(pg_clean): + from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore + return PostgreSQLKillSwitchStore(pg_clean) + + +class TestPostgreSQLKillSwitchStore: + def test_schema_and_round_trip(self, pg_killswitch_store): + """AC-1/3: table exists; exclude→list_excluded round-trips.""" + pg_killswitch_store.exclude("pf-A", _D, actor="op") + assert pg_killswitch_store.list_excluded(_D) == {"pf-A"} + + def test_day_scoped(self, pg_killswitch_store): + """AC-4: list_excluded(next day) excludes today's row.""" + pg_killswitch_store.exclude("pf-A", _D) + assert pg_killswitch_store.list_excluded(_D_NEXT) == set() + + def test_idempotent(self, pg_killswitch_store): + """AC-5: ON CONFLICT DO NOTHING — duplicate exclude is a no-op.""" + pg_killswitch_store.exclude("pf-A", _D) + pg_killswitch_store.exclude("pf-A", _D) + assert pg_killswitch_store.list_excluded(_D) == {"pf-A"} + + def test_clear(self, pg_killswitch_store): + """AC-6: clear removes the row.""" + pg_killswitch_store.exclude("pf-A", _D) + pg_killswitch_store.clear("pf-A", _D) + assert pg_killswitch_store.list_excluded(_D) == set() + + def test_global_sentinel(self, pg_killswitch_store): + pg_killswitch_store.exclude(GLOBAL_KILLSWITCH, _D) + assert GLOBAL_KILLSWITCH in pg_killswitch_store.list_excluded(_D) diff --git a/tests/application/test_control_dispatch.py b/tests/application/test_control_dispatch.py new file mode 100644 index 0000000..4d16fae --- /dev/null +++ b/tests/application/test_control_dispatch.py @@ -0,0 +1,87 @@ +"""Control channel action dispatch tests (REQ-043 / BAT-012).""" +from __future__ import annotations + +import pytest + +from trex_core.application.messaging import RpcRequest +from trex_core.application.messaging_handlers import make_control_rpc_handler +from trex_core.application.portfolio import InMemoryPauseStore + + +class _FakeKillSwitch: + def __init__(self): + self.fired = [] + self.resumed = [] + + def fire(self, *, scope, portfolio_id=None, actor="operator"): + self.fired.append((scope, portfolio_id, actor)) + return 3 + + def resume(self, *, scope, portfolio_id=None): + self.resumed.append((scope, portfolio_id)) + + +def _req(**body): + return RpcRequest(payload=body, correlation_id="c1") + + +@pytest.fixture +def ks(): + return _FakeKillSwitch() + + +@pytest.fixture +def handler(ks): + return make_control_rpc_handler(InMemoryPauseStore(), kill_switch_service=ks) + + +def test_pause_action_delegates(handler): + """AC-1: pause delegates to REQ-032 pause handler.""" + resp = handler(_req(action="pause", portfolio_id="pf-A")) + assert resp.payload["status"] == "ok" + assert "pf-A" in resp.payload["paused"] + + +def test_resume_and_list_actions(handler): + """AC-2: resume/list delegate to pause handler.""" + handler(_req(action="pause", portfolio_id="pf-A")) + assert handler(_req(action="resume", portfolio_id="pf-A")).payload["status"] == "ok" + assert handler(_req(action="list")).payload["status"] == "ok" + + +def test_kill_switch_routed(handler, ks): + """AC-3: kill_switch routes to the kill-switch service.""" + resp = handler(_req(action="kill_switch", scope="global")) + assert resp.payload["status"] == "ok" + assert resp.payload["liquidated"] == 3 + assert ks.fired == [("global", None, "operator")] + + +def test_resume_kill_switch_routed(handler, ks): + """AC-4: resume_kill_switch routes to the kill-switch service.""" + resp = handler(_req(action="resume_kill_switch", scope="portfolio", portfolio_id="pf-A")) + assert resp.payload["status"] == "ok" + assert ks.resumed == [("portfolio", "pf-A")] + + +def test_unknown_action(handler): + """AC-5: unknown action → error.""" + resp = handler(_req(action="bogus")) + assert resp.payload["status"] == "error" + assert resp.payload["reason"] == "unknown_action:bogus" + + +def test_kill_switch_portfolio_requires_id(handler): + resp = handler(_req(action="kill_switch", scope="portfolio")) + assert resp.payload == {"status": "error", "reason": "portfolio_id_required"} + + +def test_kill_switch_bad_scope(handler): + resp = handler(_req(action="kill_switch", scope="weird")) + assert resp.payload["reason"] == "bad_scope:weird" + + +def test_kill_switch_unsupported_without_service(): + handler = make_control_rpc_handler(InMemoryPauseStore()) # no service + resp = handler(_req(action="kill_switch", scope="global")) + assert resp.payload == {"status": "error", "reason": "kill_switch_unsupported"} diff --git a/tests/application/test_killswitch_execution.py b/tests/application/test_killswitch_execution.py new file mode 100644 index 0000000..2f05a58 --- /dev/null +++ b/tests/application/test_killswitch_execution.py @@ -0,0 +1,107 @@ +"""Kill-switch execution service tests (REQ-045 / BAT-012).""" +from __future__ import annotations + +from datetime import date, datetime, timezone + +import pytest + +from trex_core.application.killswitch import KillSwitchService +from trex_core.application.portfolio import GLOBAL_KILLSWITCH, InMemoryKillSwitchStore +from trex_core.domain.slot import SlotService, SlotState +from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + +_DAY = date(2026, 6, 7) +_NOW = datetime(2026, 6, 7, 6, 0, tzinfo=timezone.utc) # UTC date == _DAY + + +@pytest.fixture +def repo(): + r = InMemorySlotRepository() + return r + + +@pytest.fixture +def service(repo): + svc = SlotService(repo) + svc.preallocate_daily_slots(_DAY, 10) + return svc + + +@pytest.fixture +def store(): + return InMemoryKillSwitchStore() + + +@pytest.fixture +def ks(repo, service, store): + return KillSwitchService(slot_repo=repo, slot_service=service, killswitch_store=store) + + +def _holding(service, repo, *, portfolio_id, symbol="005930"): + idle = repo.find_idle_slot(_DAY) + pending = service.transit_slot( + idle.slot_id, SlotState.IDLE, SlotState.PENDING_BUY, + reason="buy", actor="test", symbol=symbol, portfolio_id=portfolio_id, + ) + return service.transit_slot( + pending.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, + reason="filled", actor="test", avg_price=50000.0, qty=1.0, + ) + + +def test_global_liquidates_all(ks, repo, service, store): + """AC-1: global scope sells all holding slots.""" + a = _holding(service, repo, portfolio_id="pf-A") + b = _holding(service, repo, portfolio_id="pf-B") + n = ks.fire(scope="global", now=_NOW) + assert n == 2 + assert repo.fetch_slot(a.slot_id).status is SlotState.PENDING_SELL + assert repo.fetch_slot(b.slot_id).status is SlotState.PENDING_SELL + assert GLOBAL_KILLSWITCH in store.list_excluded(_DAY) + + +def test_portfolio_scope_limits(ks, repo, service, store): + """AC-2: portfolio scope only sells that portfolio's slots.""" + a1 = _holding(service, repo, portfolio_id="pf-A", symbol="A1") + a2 = _holding(service, repo, portfolio_id="pf-A", symbol="A2") + b = _holding(service, repo, portfolio_id="pf-B") + n = ks.fire(scope="portfolio", portfolio_id="pf-A", now=_NOW) + assert n == 2 + assert repo.fetch_slot(a1.slot_id).status is SlotState.PENDING_SELL + assert repo.fetch_slot(a2.slot_id).status is SlotState.PENDING_SELL + assert repo.fetch_slot(b.slot_id).status is SlotState.HOLDING + assert store.list_excluded(_DAY) == {"pf-A"} + + +def test_idempotent_second_fire(ks, repo, service): + """AC-3: a second fire does not re-transition already pending_sell slots.""" + a = _holding(service, repo, portfolio_id="pf-A") + ks.fire(scope="global", now=_NOW) + n2 = ks.fire(scope="global", now=_NOW) + assert n2 == 0 # nothing still holding + log = repo.transition_log_for(a.slot_id) + assert len([e for e in log if e.to_state is SlotState.PENDING_SELL]) == 1 + + +def test_marks_day_exclusion(ks, repo, service, store): + """AC-5: scope marked day-excluded in the store.""" + _holding(service, repo, portfolio_id="pf-A") + ks.fire(scope="portfolio", portfolio_id="pf-A", now=_NOW) + assert "pf-A" in store.list_excluded(_DAY) + + +def test_resume_clears_exclusion_not_slots(ks, repo, service, store): + """AC-8: resume clears day-exclusion but sold slots stay sold.""" + a = _holding(service, repo, portfolio_id="pf-A") + ks.fire(scope="portfolio", portfolio_id="pf-A", now=_NOW) + assert repo.fetch_slot(a.slot_id).status is SlotState.PENDING_SELL + ks.resume(scope="portfolio", portfolio_id="pf-A", now=_NOW) + assert "pf-A" not in store.list_excluded(_DAY) + # sold slot is NOT reverted + assert repo.fetch_slot(a.slot_id).status is SlotState.PENDING_SELL + + +def test_no_holding_returns_zero(ks, store): + n = ks.fire(scope="global", now=_NOW) + assert n == 0 + assert GLOBAL_KILLSWITCH in store.list_excluded(_DAY) # still marks exclusion diff --git a/tests/application/test_killswitch_masking_resume.py b/tests/application/test_killswitch_masking_resume.py new file mode 100644 index 0000000..9a78a74 --- /dev/null +++ b/tests/application/test_killswitch_masking_resume.py @@ -0,0 +1,107 @@ +"""Kill-switch day-exclusion masking + resume tests (REQ-046 / BAT-012).""" +from __future__ import annotations + +from datetime import date + +import pytest + +from trex_core.application.killswitch import KillSwitchService +from trex_core.application.portfolio import ( + GLOBAL_KILLSWITCH, + InMemoryKillSwitchStore, + InMemoryPauseStore, + PortfolioItem, + SnapshotStore, + View, + build_snapshot_envelope, +) +from trex_core.domain.slot import SlotService +from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + +_DAY = date(2026, 6, 7) +_DAY_NEXT = date(2026, 6, 8) + + +@pytest.fixture +def store(): + s = SnapshotStore() + s.swap([ + PortfolioItem(portfolio_id="pf-A", symbol="005930", cash_balance=0, asset_quantity=0), + PortfolioItem(portfolio_id="pf-B", symbol="000660", cash_balance=0, asset_quantity=0), + ]) + return s + + +@pytest.fixture +def repo(): + return InMemorySlotRepository() + + +@pytest.fixture +def ks_store(): + return InMemoryKillSwitchStore() + + +def _asset_ids(envelope): + return {a["asset_id"] for a in envelope.to_dict()["assets"]} + + +def _envelope(store, repo, ks_store, *, view=View.ALL, day=_DAY): + return build_snapshot_envelope( + store=store, pause_store=InMemoryPauseStore(), slot_repo=repo, + available_slots=10, view=view, trading_date=day, killswitch_store=ks_store, + ) + + +def test_portfolio_exclusion_masks_that_asset(store, repo, ks_store): + """AC-1: an excluded portfolio is masked from the envelope.""" + ks_store.exclude("pf-A", _DAY) + ids = _asset_ids(_envelope(store, repo, ks_store)) + assert "pf-A" not in ids + assert "pf-B" in ids + + +def test_global_exclusion_masks_all(store, repo, ks_store): + """AC-2: global ('*') exclusion masks the whole universe.""" + ks_store.exclude(GLOBAL_KILLSWITCH, _DAY) + assert _asset_ids(_envelope(store, repo, ks_store)) == set() + + +def test_buyable_view_excludes(store, repo, ks_store): + """AC-3: excluded asset is not buyable (blocks new buys).""" + ks_store.exclude("pf-A", _DAY) + ids = _asset_ids(_envelope(store, repo, ks_store, view=View.BUYABLE)) + assert "pf-A" not in ids and "pf-B" in ids + + +def test_day_scoped_auto_recovery(store, repo, ks_store): + """AC-4: next trading day, the exclusion is gone (auto recovery).""" + ks_store.exclude("pf-A", _DAY) + ids = _asset_ids(_envelope(store, repo, ks_store, day=_DAY_NEXT)) + assert "pf-A" in ids # back on the next day + + +def test_persistence_same_day(store, repo, ks_store): + """AC-5: a fresh envelope build same day still masks (store-backed persistence).""" + ks_store.exclude("pf-A", _DAY) + # simulate restart: new envelope build reading the same store + assert "pf-A" not in _asset_ids(_envelope(store, repo, ks_store)) + + +def test_resume_reincludes(store, repo, ks_store): + """AC-6: resume (clear) re-includes the asset from the next build.""" + svc = KillSwitchService(slot_repo=repo, slot_service=SlotService(repo), killswitch_store=ks_store) + ks_store.exclude("pf-A", _DAY) + assert "pf-A" not in _asset_ids(_envelope(store, repo, ks_store)) + svc.resume(scope="portfolio", portfolio_id="pf-A", + now=__import__("datetime").datetime(2026, 6, 7, 6, tzinfo=__import__("datetime").timezone.utc)) + assert "pf-A" in _asset_ids(_envelope(store, repo, ks_store)) + + +def test_no_killswitch_store_no_masking(store, repo): + """Backward compat: without killswitch_store nothing extra is masked.""" + env = build_snapshot_envelope( + store=store, pause_store=InMemoryPauseStore(), slot_repo=repo, + available_slots=10, view=View.ALL, trading_date=_DAY, + ) + assert _asset_ids(env) == {"pf-A", "pf-B"} diff --git a/tests/conftest.py b/tests/conftest.py index d46ae1e..7907e06 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -15,6 +15,7 @@ import os import pytest _PG_TABLES = ( + "killswitch_exclusion", "snapshot_pause", "ledger_outbox", "intent_log", diff --git a/trex_core/adapters/messaging/rabbitmq.py b/trex_core/adapters/messaging/rabbitmq.py index 6205305..588ac96 100644 --- a/trex_core/adapters/messaging/rabbitmq.py +++ b/trex_core/adapters/messaging/rabbitmq.py @@ -28,6 +28,7 @@ import pika import pika.exceptions from trex_core.adapters.messaging.topology import ( + CONTROL_REQ_QUEUE, EVENTS_DLQ_EXCHANGE, EVENTS_DLQ_QUEUE, EVENTS_EXCHANGE, @@ -142,6 +143,12 @@ class RabbitMQTransport: durable=True, arguments={"x-dead-letter-exchange": EVENTS_DLQ_EXCHANGE}, ) + # BAT-012 / REQ-043: operator control channel (action dispatch). + ch.queue_declare( + CONTROL_REQ_QUEUE, + durable=True, + arguments={"x-dead-letter-exchange": EVENTS_DLQ_EXCHANGE}, + ) def declare_strategy_queue(self, queue: str, binding_pattern: str) -> None: """Per-strategy durable queue (REQ-023 AC-4) bound to events exchange.""" diff --git a/trex_core/adapters/messaging/topology.py b/trex_core/adapters/messaging/topology.py index 7189858..56444f1 100644 --- a/trex_core/adapters/messaging/topology.py +++ b/trex_core/adapters/messaging/topology.py @@ -15,7 +15,9 @@ from __future__ import annotations # durable for simplicity since the queue itself is shared) --- SNAPSHOT_REQ_QUEUE = "trex.snapshot.req" INTENT_REQ_QUEUE = "trex.intent.req" -PAUSE_REQ_QUEUE = "trex.pause.req" # BAT-008 / REQ-032 control RPC +PAUSE_REQ_QUEUE = "trex.pause.req" # BAT-008 / REQ-032 (legacy alias of control) +CONTROL_REQ_QUEUE = "trex.control.req" # BAT-012 / REQ-043 operator control channel + # (action: pause|resume|list|kill_switch|resume_kill_switch) # --- Events --- EVENTS_EXCHANGE = "trex.events" # topic, durable diff --git a/trex_core/adapters/persistence/sql/005_create_killswitch_exclusion.sql b/trex_core/adapters/persistence/sql/005_create_killswitch_exclusion.sql new file mode 100644 index 0000000..a7d2cbe --- /dev/null +++ b/trex_core/adapters/persistence/sql/005_create_killswitch_exclusion.sql @@ -0,0 +1,15 @@ +-- BAT-012 / REQ-044: day-scoped kill-switch exclusion overlay. +-- Distinct from snapshot_pause (permanent, resume-until). A row excludes a +-- portfolio (or '*' for global) from strategy execution for ONE trading_date; +-- the next trading day is automatically clear because reads filter by +-- trading_date. Manual resume (REQ-046) deletes the row. +CREATE TABLE IF NOT EXISTS killswitch_exclusion ( + portfolio_id TEXT NOT NULL, -- '*' = global exclusion sentinel + trading_date DATE NOT NULL, + actor TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT (now() AT TIME ZONE 'UTC'), + PRIMARY KEY (portfolio_id, trading_date) +); + +CREATE INDEX IF NOT EXISTS idx_killswitch_exclusion_trading_date + ON killswitch_exclusion (trading_date); diff --git a/trex_core/adapters/portfolio/killswitch_store.py b/trex_core/adapters/portfolio/killswitch_store.py new file mode 100644 index 0000000..b4b695d --- /dev/null +++ b/trex_core/adapters/portfolio/killswitch_store.py @@ -0,0 +1,52 @@ +"""PostgreSQL-backed day-scoped kill-switch exclusion store (REQ-044 / BAT-012). + +Distinct from `PostgreSQLPauseStore`: pause is permanent (resume-until), +kill-switch exclusion is day-scoped — keyed by `(portfolio_id, trading_date)` +so the next trading day is automatically clear (reads filter by trading_date). +`portfolio_id == "*"` is the global-exclusion sentinel. +""" +from __future__ import annotations + +from datetime import date +from typing import Any + +from trex_core.adapters.persistence.postgresql import _require_psycopg +from trex_core.application.portfolio import GLOBAL_KILLSWITCH # noqa: F401 (re-export) + + +class PostgreSQLKillSwitchStore: + """`killswitch_exclusion` table CRUD. (portfolio_id, trading_date) PK.""" + + def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None: + self._psycopg = _require_psycopg(psycopg_module) + self._dsn = dsn + + def exclude(self, portfolio_id: str, trading_date: date, *, actor: str = "kill-switch") -> None: + with self._psycopg.connect(self._dsn, autocommit=False) as conn: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO killswitch_exclusion (portfolio_id, trading_date, actor) " + "VALUES (%s, %s, %s) " + "ON CONFLICT (portfolio_id, trading_date) DO NOTHING", + (portfolio_id, trading_date, actor), + ) + conn.commit() + + def clear(self, portfolio_id: str, trading_date: date) -> None: + with self._psycopg.connect(self._dsn, autocommit=False) as conn: + with conn.cursor() as cur: + cur.execute( + "DELETE FROM killswitch_exclusion " + "WHERE portfolio_id = %s AND trading_date = %s", + (portfolio_id, trading_date), + ) + conn.commit() + + def list_excluded(self, trading_date: date) -> set[str]: + with self._psycopg.connect(self._dsn) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT portfolio_id FROM killswitch_exclusion WHERE trading_date = %s", + (trading_date,), + ) + return {row[0] for row in cur.fetchall()} diff --git a/trex_core/application/killswitch.py b/trex_core/application/killswitch.py new file mode 100644 index 0000000..01c88b2 --- /dev/null +++ b/trex_core/application/killswitch.py @@ -0,0 +1,88 @@ +"""Kill-switch execution service (REQ-045 / BAT-012). + +Operator emergency stop: liquidate all `holding` slots in a scope +(portfolio or global) by reusing SlotService.sell() (REQ-036), then mark the +scope day-excluded (REQ-044) so strategy execution is masked for the day +(REQ-046). Independent of the dcy-007 safety-sell worker — concurrent sells +on the same slot are idempotent (REQ-036 PENDING_SELL guard). +""" +from __future__ import annotations + +import logging +from datetime import date as date_cls +from datetime import datetime, timezone + +from trex_core.application.portfolio import GLOBAL_KILLSWITCH, KillSwitchStore +from trex_core.domain.slot import SlotRepository, SlotService, SlotState + +_log = logging.getLogger(__name__) +_ACTOR = "kill-switch" + + +class KillSwitchService: + """Range-scoped emergency liquidation + day-scoped exclusion (REQ-045).""" + + def __init__( + self, + *, + slot_repo: SlotRepository, + slot_service: SlotService, + killswitch_store: KillSwitchStore, + ) -> None: + self._repo = slot_repo + self._svc = slot_service + self._store = killswitch_store + + @staticmethod + def _trading_date(now: datetime) -> date_cls: + return now.astimezone(timezone.utc).date() + + def fire( + self, + *, + scope: str, + portfolio_id: str | None = None, + actor: str = "operator", + now: datetime | None = None, + ) -> int: + """Liquidate holding slots in scope + mark day-excluded. Returns count. + + scope: "portfolio" (requires portfolio_id) | "global". + """ + now = now or datetime.now(tz=timezone.utc) + day = self._trading_date(now) + holding = [s for s in self._repo.list_slots(day) if s.status is SlotState.HOLDING] + if scope == "portfolio": + holding = [s for s in holding if s.portfolio_id == portfolio_id] + + liquidated = 0 + for s in holding: + try: + self._svc.sell(s.slot_id, actor=_ACTOR, reason="emergency_stop") + liquidated += 1 + except Exception: # noqa: BLE001 - one slot must not stop the sweep + _log.warning("kill-switch sell failed (slot=%s)", s.slot_id, exc_info=True) + + # Mark the scope day-excluded so strategy execution is masked (REQ-046). + key = GLOBAL_KILLSWITCH if scope == "global" else (portfolio_id or "") + self._store.exclude(key, day, actor=actor) + + _log.warning( + "kill-switch fired: actor=%s scope=%s portfolio_id=%s liquidated=%d", + actor, scope, portfolio_id, liquidated, + ) + return liquidated + + def resume( + self, + *, + scope: str, + portfolio_id: str | None = None, + now: datetime | None = None, + ) -> None: + """Manually clear the day-scoped exclusion (REQ-046). Sold slots stay sold.""" + now = now or datetime.now(tz=timezone.utc) + day = self._trading_date(now) + key = GLOBAL_KILLSWITCH if scope == "global" else (portfolio_id or "") + self._store.clear(key, day) + _log.info("kill-switch resume: scope=%s portfolio_id=%s", scope, portfolio_id) diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index 66b4583..9c624af 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -164,6 +164,7 @@ def make_snapshot_rpc_handler( pause_store: Optional[PauseStore] = None, default_total_slots: int = 10, watchdog_store: Optional[WatchdogStore] = None, + killswitch_store=None, ): """Build the RPC handler for `trex.snapshot.req` (REQ-028). @@ -211,6 +212,7 @@ def make_snapshot_rpc_handler( view=view, strategy_key=strategy_key, trading_date=day, + killswitch_store=killswitch_store, ) return RpcResponse(payload=envelope.to_dict()) @@ -282,6 +284,78 @@ def _err(reason: str) -> RpcResponse: return RpcResponse(payload={"status": "error", "reason": reason}) +# ---- control RPC handler (REQ-043, BAT-012) --------------------------------- + +def make_control_rpc_handler(pause_store: PauseStore, *, kill_switch_service=None): + """RPC handler for `trex.control.req` — action discriminator dispatch (REQ-043). + + Actions: `pause`/`resume`/`list` (REQ-032, delegated), `kill_switch` (REQ-045), + `resume_kill_switch` (REQ-046). Unknown actions return an error response. + """ + pause_handler = make_pause_rpc_handler(pause_store) + + def handler(req: RpcRequest) -> RpcResponse: + body = req.payload or {} + action = str(body.get("action") or "").lower() + if action == "kill_switch": + return _handle_kill_switch(body, kill_switch_service) + if action == "resume_kill_switch": + return _handle_resume_kill_switch(body, kill_switch_service) + if action in ("pause", "resume", "list"): + return pause_handler(req) + return _err(f"unknown_action:{action}") + + return handler + + +def _killswitch_scope(body: dict): + """Validate scope/portfolio_id. Returns (scope, pid) or (None, RpcResponse error).""" + scope = str(body.get("scope") or "").lower() + pid = _opt_str(body.get("portfolio_id")) + if scope not in ("portfolio", "global"): + return None, _err(f"bad_scope:{scope}") + if scope == "portfolio" and not pid: + return None, _err("portfolio_id_required") + return (scope, pid), None + + +def _handle_kill_switch(body: dict, svc) -> RpcResponse: + """REQ-045: liquidate holding slots in scope + day-exclude.""" + if svc is None: + return _err("kill_switch_unsupported") + parsed, err = _killswitch_scope(body) + if err is not None: + return err + scope, pid = parsed + actor = str(body.get("actor") or "operator") + try: + liquidated = svc.fire(scope=scope, portfolio_id=pid, actor=actor) + except Exception as exc: # noqa: BLE001 + return _err(f"kill_switch_failed:{exc.__class__.__name__}") + payload = {"status": "ok", "scope": scope, "liquidated": liquidated} + if pid: + payload["portfolio_id"] = pid + return RpcResponse(payload=payload) + + +def _handle_resume_kill_switch(body: dict, svc) -> RpcResponse: + """REQ-046: clear the day-scoped exclusion (sold slots stay sold).""" + if svc is None: + return _err("resume_kill_switch_unsupported") + parsed, err = _killswitch_scope(body) + if err is not None: + return err + scope, pid = parsed + try: + svc.resume(scope=scope, portfolio_id=pid) + except Exception as exc: # noqa: BLE001 + return _err(f"resume_kill_switch_failed:{exc.__class__.__name__}") + payload = {"status": "ok", "scope": scope} + if pid: + payload["portfolio_id"] = pid + return RpcResponse(payload=payload) + + # ---- event payload builder -------------------------------------------------- def slot_transition_event_payload(slot, *, event_type: str) -> dict: diff --git a/trex_core/application/portfolio.py b/trex_core/application/portfolio.py index 014558b..0376a43 100644 --- a/trex_core/application/portfolio.py +++ b/trex_core/application/portfolio.py @@ -216,6 +216,45 @@ class InMemoryPauseStore: self._paused.discard(portfolio_id) +# ---------------------------------------------------------------- kill-switch +# Day-scoped exclusion overlay (REQ-044/046 / BAT-012). Distinct from pause: +# keyed by (portfolio_id, trading_date) so it auto-clears next trading day. +GLOBAL_KILLSWITCH = "*" # sentinel: whole-universe exclusion for the day + + +class KillSwitchStore(Protocol): + """Day-scoped kill-switch exclusion overlay (REQ-044).""" + + def list_excluded(self, trading_date) -> set[str]: + ... + + def exclude(self, portfolio_id: str, trading_date, *, actor: str = "kill-switch") -> None: + ... + + def clear(self, portfolio_id: str, trading_date) -> None: + ... + + +class InMemoryKillSwitchStore: + """Process-local day-scoped kill-switch store for tests + CI.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._excluded: set[tuple[str, object]] = set() # (portfolio_id, trading_date) + + def list_excluded(self, trading_date) -> set[str]: + with self._lock: + return {pid for pid, d in self._excluded if d == trading_date} + + def exclude(self, portfolio_id: str, trading_date, *, actor: str = "kill-switch") -> None: + with self._lock: + self._excluded.add((portfolio_id, trading_date)) + + def clear(self, portfolio_id: str, trading_date) -> None: + with self._lock: + self._excluded.discard((portfolio_id, trading_date)) + + # ---------------------------------------------------------------- selection @dataclass(frozen=True) @@ -274,11 +313,14 @@ def build_snapshot_envelope( view: View = View.ALL, strategy_key: Optional[str] = None, trading_date=None, + killswitch_store: "Optional[KillSwitchStore]" = None, ) -> SnapshotEnvelope: - """Compose base(view) − mask(pending·warning·pause) − strategy_key (REQ-029). + """Compose base(view) − mask(pending·warning·pause·killswitch) − strategy_key. `slot_repo` is the SlotRepository used to read holding/pending state. `trading_date` defaults to UTC today (matches the daemon's preallocate). + `killswitch_store` (REQ-046) adds the day-scoped kill-switch exclusion to + the mask: a `GLOBAL_KILLSWITCH` ("*") entry masks the whole universe. """ # Read core authoritative slot state once. day = trading_date or datetime.now(tz=timezone.utc).date() @@ -298,6 +340,9 @@ def build_snapshot_envelope( assets = store.list_assets() paused = pause_store.list_paused() warned = {a.item.portfolio_id for a in assets if a.warning} + # REQ-046: day-scoped kill-switch exclusion. '*' sentinel = whole universe. + excluded = killswitch_store.list_excluded(day) if killswitch_store is not None else set() + killswitch_global = GLOBAL_KILLSWITCH in excluded # Build the base set (view). by_pid = {a.item.portfolio_id: a for a in assets} @@ -316,13 +361,13 @@ def build_snapshot_envelope( base_pids = list({a.item.portfolio_id for a in assets} | holding_pids) base_pids.sort() - # Apply masks: pending OR warning OR pause. - masked = pending_pids | warned | paused + # Apply masks: pending OR warning OR pause OR kill-switch day-exclusion. + masked = pending_pids | warned | paused | excluded # Filter and shape. out: list[dict] = [] for pid in base_pids: - if pid in masked: + if killswitch_global or pid in masked: continue if pid in by_pid: item = by_pid[pid].item diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 6695292..3d18d84 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -6,7 +6,7 @@ strategy-facing communication runs over RabbitMQ (REQ-021/022/023). Threads (started in `run()`): - intent RPC consumer — `trex.intent.req`, prefetch=1, FIFO, dedup via intent_log - snapshot RPC consumer — `trex.snapshot.req` (also refreshes the watchdog) - - pause RPC consumer — `trex.pause.req` (REQ-032) + - control RPC consumer — `trex.control.req` (pause/resume/list + kill_switch/resume_kill_switch, REQ-032/043) - execution loop — broker dispatch + poll + slot transitions + event emit - ledger writer — ledger outbox drain (REQ-019) - snapshot scheduler — daily portfolio-hub snapshot refresh (REQ-030) @@ -43,15 +43,17 @@ from trex_core.adapters.portfolio import ( PostgreSQLPauseStore, StubPortfolioRead, ) +from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore from trex_core.application.broker_port import BrokerPort from trex_core.application.execution import ExecutionService from trex_core.application.intents import IntentService from trex_core.application.ledger_outbox import LedgerWriter from trex_core.application.ledger_port import LedgerPort from trex_core.application.messaging import MessagingTransport +from trex_core.application.killswitch import KillSwitchService from trex_core.application.messaging_handlers import ( + make_control_rpc_handler, make_intent_rpc_handler, - make_pause_rpc_handler, make_snapshot_rpc_handler, slot_transition_event_payload, ) @@ -148,6 +150,13 @@ class Daemon: # BAT-010/011: per-slot watchdog (snapshot RPC writer, safety-sell reader). self.watchdog = WatchdogStore() self.pause_store = PostgreSQLPauseStore(config.dsn) + # BAT-012: day-scoped kill-switch exclusion store + service. + self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn) + self.kill_switch = KillSwitchService( + slot_repo=self.slot_repo, + slot_service=self.slot_service, + killswitch_store=self.killswitch_store, + ) self.portfolio_read = _build_portfolio_read(config) self.snapshot_scheduler = SnapshotScheduler( read_port=self.portfolio_read, @@ -231,18 +240,22 @@ class Daemon: pause_store=self.pause_store, default_total_slots=self.config.total_slots, watchdog_store=self.watchdog, + killswitch_store=self.killswitch_store, ) transport.consume_rpc_forever(topology.SNAPSHOT_REQ_QUEUE, handler, self.stop) - def _pause_rpc_loop(self) -> None: + def _control_rpc_loop(self) -> None: + # REQ-043: operator control channel (pause/resume/list + kill_switch/resume_kill_switch). transport = _build_transport(self.config) self._transports.append(transport) try: transport.declare_topology() except Exception: # noqa: BLE001 pass - handler = make_pause_rpc_handler(self.pause_store) - transport.consume_rpc_forever(topology.PAUSE_REQ_QUEUE, handler, self.stop) + handler = make_control_rpc_handler( + self.pause_store, kill_switch_service=self.kill_switch, + ) + transport.consume_rpc_forever(topology.CONTROL_REQ_QUEUE, handler, self.stop) def _safety_sell_loop(self) -> None: # REQ-037/038/039: forced_close / stop-loss·take-profit / market-close. @@ -278,7 +291,7 @@ class Daemon: ("trex-ledger", self._ledger_loop), ("trex-intent-rpc", self._intent_rpc_loop), ("trex-snapshot-rpc", self._snapshot_rpc_loop), - ("trex-pause-rpc", self._pause_rpc_loop), + ("trex-control-rpc", self._control_rpc_loop), ("trex-snapshot-scheduler", self._snapshot_scheduler_loop), ("trex-safety-sell", self._safety_sell_loop), ]