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

feat(strategy-state): persist instance-scoped slot maps

Browse Source
This commit is contained in:
Hermes Agent
2026-08-29 15:11:43 +09:00
parent 3f94c19162
commit 4699de61cc
18 changed files with 301 additions and 646 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/interface-contract.md
+10 -7
View File
@@ -74,11 +74,14 @@
``` ```
{ "filter": "all" | "inactive" | "pending" | "holding" | "error", # default "all" { "filter": "all" | "inactive" | "pending" | "holding" | "error", # default "all"
"spy": true | false, # default false; true = watchdog 갱신 안 함(운영자 조회) "spy": true | false, # default false; true = watchdog 갱신 안 함(운영자 조회)
"strategy": str # optional (REQ-116 BAT-055) — exact-match on strategy_name "strategy": str, # optional (REQ-116 BAT-055) — exact-match on strategy_name
"strategy_instance_id": str # required non-empty string when spy=false; ignored for spy=true filtering
} }
``` ```
- `filter`: 서버 측 슬롯 status 필터. `"inactive"` 는 buyable 슬롯(universe에 바인딩된 매수 가능 슬롯). 미인식 값은 `"all"` 로 처리. - `filter`: 서버 측 슬롯 status 필터. `"inactive"` 는 buyable 슬롯(universe에 바인딩된 매수 가능 슬롯). 미인식 값은 `"all"` 로 처리.
- `spy=true`: 응답에 포함된 holding 슬롯의 워치독 `last_seen_at` 을 갱신하지 않는다(운영자 CLI 조회가 전략 생존 신호로 오인되지 않도록 — REQ-034 CHG-01). - `spy=true`: 응답에 포함된 holding 슬롯의 워치독 `last_seen_at` 을 갱신하지 않는다(운영자 CLI 조회가 전략 생존 신호로 오인되지 않도록 — REQ-034 CHG-01).
- `spy=false`: `strategy_instance_id`는 required non-empty string이다. 누락·비문자열·빈 값은 `{slots:[], available_slots:null, degraded:false, reason:"bad_request:strategy_instance_id", message:"strategy_instance_id must be a non-empty string"}`를 반환하며 watchdog을 갱신하지 않는다.
- strategy-state visibility: `spy=false`는 exact requested entry 하나를 `strategy_state: {version,data,updated_at}`로만 반환하고 missing entry는 생략한다. `spy=true`는 supplied ID와 무관하게 complete `{strategy_instance_id: {version,data,updated_at}}` map을 반환한다.
- `strategy` (REQ-116 additive): 지정 시 `strategy_name == strategy` 인 슬롯 행만 반환. 대소문자 exact-match. 미지정 시 기존과 완전 동일. `filter` 와 AND 조합. 매칭 없으면 `slots=[]`(에러 없음). `available_slots` 는 strategy 필터와 무관하게 인스턴스 전체 기준으로 계산(REQ-073 AC-6 불변). - `strategy` (REQ-116 additive): 지정 시 `strategy_name == strategy` 인 슬롯 행만 반환. 대소문자 exact-match. 미지정 시 기존과 완전 동일. `filter` 와 AND 조합. 매칭 없으면 `slots=[]`(에러 없음). `available_slots` 는 strategy 필터와 무관하게 인스턴스 전체 기준으로 계산(REQ-073 AC-6 불변).
코어가 reply-to 큐로 보내는 응답 payload: 코어가 reply-to 큐로 보내는 응답 payload:
@@ -524,21 +527,21 @@ get_current_price(symbol: str, market: str = "J") -> Optional[Decimal]
- pika BlockingConnection 은 thread-safe 가 아니므로 consumer/publisher 마다 자기 connection/channel 을 가져야 한다. - pika BlockingConnection 은 thread-safe 가 아니므로 consumer/publisher 마다 자기 connection/channel 을 가져야 한다.
- 토폴로지 선언은 멱등하지만 attribute(durable, args) 가 기존 큐와 다르면 RabbitMQ 가 PRECONDITION_FAILED 로 거부 → batch 마이그레이션 시 큐 재선언 절차 필요. - 토폴로지 선언은 멱등하지만 attribute(durable, args) 가 기존 큐와 다르면 RabbitMQ 가 PRECONDITION_FAILED 로 거부 → batch 마이그레이션 시 큐 재선언 절차 필요.
## Interface: `trex.slots.strategy_state.set` RPC (BAT-042 / dcy-028, REQ-080) ## Interface: `trex.slots.strategy_state.set` RPC (REQ-0009)
- Type: external-service (strategy daemon ↔ core write RPC) - Type: external-service (strategy daemon ↔ core write RPC)
- Consumers / Producers: 전략 데몬(produce) ↔ core 핸들러(consume). per-instance 큐(`{base}.{instance_key}`), direct reply-to, JSON. - Consumers / Producers: 전략 데몬(produce) ↔ core 핸들러(consume). per-instance 큐(`{base}.{instance_key}`), direct reply-to, JSON.
- Purpose: 슬롯의 opaque strategy-state 를 occupancy_seq+version CAS 로 write. read 는 `trex.slots.req` 에 piggyback(별도 RPC 아님). - Purpose: 슬롯별 opaque strategy-instance state entry를 version CAS로 write. read는 `trex.slots.req`에 piggyback(별도 RPC 아님).
### Inputs ### Inputs
- `{slot_id, strategy_id, expected_version:int, expected_occupancy_seq:int, data:str}`. `correlation_id = "{slot_id}:{strategy_id}:{expected_version}"`(멱등), delivery persistent. `data` 는 opaque UTF-8(core 무해석), byte size cap(기본 4KB ceiling, 운영 권장 1~2KB). - `{slot_id, strategy_instance_id, expected_version:int, data:str}`. `strategy_instance_id`는 required non-empty string이며 `correlation_id`는 저장 멱등성 key가 아니다. `data`는 opaque UTF-8(core 무해석), byte size cap(기본 4KB ceiling, 운영 권장 1~2KB).
### Outputs / Outcomes ### Outputs / Outcomes
- 성공: `{ok:true, slot_id, strategy_id, version:<new>}`. - 성공: `{ok:true, slot_id, strategy_instance_id, version:<new>}`.
- 실패 reason 전체 집합: `occupancy_changed`(+`occupancy_seq`) / `version_conflict`(+`version`) / `slot_inactive` / `state_too_large` / `bad_request:<detail>`. 소비자는 미인식 reason 에 fail-closed. - 실패 reason 전체 집합: `version_conflict`(+`version`) / `slot_inactive` / `state_too_large` / `bad_request:<detail>`. 소비자는 미인식 reason에 fail-closed.
### Compatibility Rules ### Compatibility Rules
- read 표면(`trex.slots.req`)의 각 슬롯 행에 additive 로 `occupancy_seq`(int)와, 보유 시 `strategy_state` 맵(`{<strategy_id>:{version,data,updated_at}}`)이 실린다(REQ-073/074 CHG). 기존 소비자는 unknown 필드 무시로 하위호환. `strategy_state` 는 strategy_* reasoning 필드(strategy_id 등, REQ-074 AC-5 제외 대상)와 구분되는 opaque 상태 맵이다. - slot strategy-state는 occupancy와 독립된 slot-scoped `data[strategy_instance_id] = {version,data,updated_at}` map이다. close/reopen은 entry를 삭제하거나 version을 초기화하지 않는다. Entry에는 `strategy_id`나 `occupancy_seq` metadata가 없으며, Slot 자체의 `occupancy_seq` semantics는 별도로 유지된다.
## Shared Constraints ## Shared Constraints
docs/runtime-flows.md
+11 -12
View File
@@ -34,11 +34,12 @@
1. `trex-core-daemon --config <path>.toml` 기동 → `DaemonConfig.from_config(toml)` = `from_env()`(TREX_POSTGRESQL_DSN/`TREX_RABBITMQ_URL` 등) + TOML(`instance_key`, `portfolio_hub_token`, `trading_timezone`, `max_active_slots`) 병합. `--config` 필수(없으면 오류 종료, BAT-013/REQ-048). `instance_key` 로 큐명·라우팅·DB 행 스코프 분리(REQ-056). 1. `trex-core-daemon --config <path>.toml` 기동 → `DaemonConfig.from_config(toml)` = `from_env()`(TREX_POSTGRESQL_DSN/`TREX_RABBITMQ_URL` 등) + TOML(`instance_key`, `portfolio_hub_token`, `trading_timezone`, `max_active_slots`) 병합. `--config` 필수(없으면 오류 종료, BAT-013/REQ-048). `instance_key` 로 큐명·라우팅·DB 행 스코프 분리(REQ-056).
2. setup: `apply_migrations(dsn)`(멱등; 010 inactive CHECK·011 컬럼 drop 포함) → `execution.reconcile()`(REQ-014) → 이벤트 publisher 의 `declare_topology()`. (일일 preallocation 은 BAT-030 에서 폐기 — 슬롯 생성은 universe reconcile 이 담당.) 2. setup: `apply_migrations(dsn)`(멱등; 010 inactive CHECK·011 컬럼 drop 포함) → `execution.reconcile()`(REQ-014) → 이벤트 publisher 의 `declare_topology()`. (일일 preallocation 은 BAT-030 에서 폐기 — 슬롯 생성은 universe reconcile 이 담당.)
3. start: 7개 백그라운드 스레드 동시 기동 3. start: 8개 백그라운드 스레드 동시 기동
- `trex-execution` :: 브로커 dispatch/poll 루프 (REQ-013/014; 매도 수량 브로커 정본 REQ-062). **각 틱 시작 시 `event_publisher.keepalive_tick()` 호출** (REQ-092 — pika 스레드 안전 규칙: 발행 연결 소유 스레드에서만 I/O). 틱 간격(기본 1s) < heartbeat/2(15s) → 서버 측 idle 종료 예방. - `trex-execution` :: 브로커 dispatch/poll 루프 (REQ-013/014; 매도 수량 브로커 정본 REQ-062). **각 틱 시작 시 `event_publisher.keepalive_tick()` 호출** (REQ-092 — pika 스레드 안전 규칙: 발행 연결 소유 스레드에서만 I/O). 틱 간격(기본 1s) < heartbeat/2(15s) → 서버 측 idle 종료 예방.
- `trex-ledger` :: ledger outbox 드레인 (REQ-019) - `trex-ledger` :: ledger outbox 드레인 (REQ-019)
- `trex-intent-rpc` :: `trex.intent.req` 소비자 (prefetch=1, FIFO, dedup via intent_log; `update_thresholds` 액션 포함 REQ-040) - `trex-intent-rpc` :: `trex.intent.req` 소비자 (prefetch=1, FIFO, dedup via intent_log; `update_thresholds` 액션 포함 REQ-040)
- `trex-slots-rpc` :: `trex.slots.req` 소비자 (spy=false 시 holding 슬롯 워치독 `last_seen_at` 갱신 REQ-034; `safety_sell_enabled=false` 시 `trex-safety-sell` 미기동 BAT-036) - `trex-slots-rpc` :: `trex.slots.req` 소비자 (spy=false 시 holding 슬롯 워치독 `last_seen_at` 갱신 REQ-034; `safety_sell_enabled=false` 시 `trex-safety-sell` 미기동 BAT-036)
- `trex-strategy-state-set` :: `trex.slots.strategy_state.set` 소비자 (slot-scoped strategy-instance state version CAS)
- `trex-control-rpc` :: `trex.control.req` 소비자 — pause/resume/list + kill_switch/resume_kill_switch (REQ-032/043) - `trex-control-rpc` :: `trex.control.req` 소비자 — pause/resume/list + kill_switch/resume_kill_switch (REQ-032/043)
- `trex-snapshot-scheduler` :: 일일 portfolio-hub 스냅샷 refresh + **universe 슬롯 reconcile**(REQ-058) + holding 불일치 reconcile(REQ-062) - `trex-snapshot-scheduler` :: 일일 portfolio-hub 스냅샷 refresh + **universe 슬롯 reconcile**(REQ-058) + holding 불일치 reconcile(REQ-062)
- `trex-safety-sell` :: 자율 안전 매도 트리거 루프 (forced_close / stop-loss·take-profit / 장 마감, REQ-037/038/039) - `trex-safety-sell` :: 자율 안전 매도 트리거 루프 (forced_close / stop-loss·take-profit / 장 마감, REQ-037/038/039)
@@ -138,7 +139,7 @@
`trex.slots.req` 메시지 처리 (`make_slots_rpc_handler`). `trex.slots.req` 메시지 처리 (`make_slots_rpc_handler`).
1. 요청 payload 디코드: `{filter ∈ {all, inactive, pending, holding, error}, spy: bool}`. 1. 요청 payload 디코드: `{filter ∈ {all, inactive, pending, holding, error}, spy: bool, strategy_instance_id?: str}`. `spy=false` 전략 요청은 non-empty string `strategy_instance_id`가 필수이며, malformed 값은 slots/state 노출과 watchdog 갱신 없이 `bad_request:strategy_instance_id` 기존 envelope으로 종료한다. `spy=true` 운영자 요청은 ID를 요구하거나 supplied ID로 결과를 필터하지 않는다.
2. `slot_repo.list_slots(instance_key)` → status 기준 서버 필터 적용(`inactive`/`pending`/`holding`/`error`; 미인식 = `all`). 2. `slot_repo.list_slots(instance_key)` → status 기준 서버 필터 적용(`inactive`/`pending`/`holding`/`error`; 미인식 = `all`).
3. `spy=false` 면 `slot_repo.list_holding_slots()` 의 모든 holding 슬롯에 `WatchdogStore.update(slot_id)`. 3. `spy=false` 면 `slot_repo.list_holding_slots()` 의 모든 holding 슬롯에 `WatchdogStore.update(slot_id)`.
4. 각 슬롯에 `slot_to_dict(s, snapshot_store=..., watchdog_store=..., pause_store=..., watchdog_timeout_sec=...)` 적용 후 broker holdings snapshot 오버레이를 적용해 병합 row 산출: 4. 각 슬롯에 `slot_to_dict(s, snapshot_store=..., watchdog_store=..., pause_store=..., watchdog_timeout_sec=...)` 적용 후 broker holdings snapshot 오버레이를 적용해 병합 row 산출:
@@ -149,7 +150,7 @@
- 추정 현금(REQ-029 OQ-4): `cash_estimated=True` + `cash_balance += pending ledger delta`(best-effort) - 추정 현금(REQ-029 OQ-4): `cash_estimated=True` + `cash_balance += pending ledger delta`(best-effort)
- 워치독: `watchdog: {expired, last_seen}` inline (WatchdogStore) - 워치독: `watchdog: {expired, last_seen}` inline (WatchdogStore)
- 오버레이: `paused` (PauseStore, REQ-032) - 오버레이: `paused` (PauseStore, REQ-032)
- strategy 필드 없음 (REQ-074 AC-5) - strategy-state: `spy=false`면 요청 `strategy_instance_id`와 정확히 일치하는 `{version,data,updated_at}` entry 하나만(없으면 생략), `spy=true`면 `{strategy_instance_id: {version,data,updated_at}}` 전체 map
5. degraded 상태 동봉: `SnapshotStore.degraded_state()` → `{degraded, reason, message}`. 5. degraded 상태 동봉: `SnapshotStore.degraded_state()` → `{degraded, reason, message}`.
6. 응답: `{slots[], degraded, reason, message}` — `slots` 는 항상 배열, null 금지. 6. 응답: `{slots[], degraded, reason, message}` — `slots` 는 항상 배열, null 금지.
@@ -346,17 +347,15 @@ timeout 미지정·취소 실패로 잔여 고착된 pending 슬롯을 처분:
6. slot_transition_log 에 전이 기록 1행을 append 하고 갱신된 슬롯을 반환한다. 6. slot_transition_log 에 전이 기록 1행을 append 하고 갱신된 슬롯을 반환한다.
7. 외부 I/O(브로커 등)는 lock 밖에서 호출자가 수행한다(lock 보유 중 금지). 7. 외부 I/O(브로커 등)는 lock 밖에서 호출자가 수행한다(lock 보유 중 금지).
## Flow: `slot strategy-state 저장소` (BAT-042 / dcy-028, REQ-079~082) ## Flow: `slot strategy-state 저장소` (REQ-0009)
전략 데몬(예: trex-tsl)이 슬롯 점유 동안 누적하는 opaque 상태(예: TSL peak)를 슬롯 수명에 묶어 server-side 로 보관한다. core 는 `data` 를 해석하지 않는다(opaque UTF-8 + byte size cap 만 검증). 전략 상태는 슬롯 점유와 독립된 opaque 상태다. core는 inner `data`를 해석하지 않고 UTF-8 byte size cap만 검증하며, 하나의 슬롯별 map `data[strategy_instance_id] = {version,data,updated_at}`를 보관한다.
1. **read (piggyback, C-1)**: 데몬이 매 사이클 `trex.slots.req` 로 슬롯을 읽으면 각 행에 `occupancy_seq`(int)와, 보유 시 `strategy_state` 맵(`{<strategy_id>: {version,data,updated_at}}`)이 함께 실린다. 맵은 핸들러가 반환 slot_id 들로 `StrategyStateStore.get_for_slots()` 1회 batched 조회 후 group by 해 조립한다(추가 왕복 없음). 슬롯 현재 occupancy_seq 와 다른 stale 엔트리는 노출하지 않는다. 1. **read (piggyback)**: `spy=false` 전략 요청은 required non-empty `strategy_instance_id`의 exact entry만 각 slot 행의 single `strategy_state` object로 받으며, 없으면 field가 생략된다. 다른 ID나 전체 map은 노출하지 않는다. `spy=true` 운영자 요청은 ID 없이도 각 slot의 complete instance map을 받고 watchdog을 갱신하지 않는다.
2. **evaluate (pure)**: 데몬은 행의 `strategy_state[strategy_id]` 를 read-only 입력으로 `evaluate()` 에 넣는다(상태 write 는 evaluate 밖, 엔진이 수행). 2. **write (별도 RPC)**: `trex.slots.strategy_state.set` payload는 `{slot_id, strategy_instance_id, expected_version, data}`다. active slot, size limit, 요청 entry version만 CAS한다. entry가 없으면 `expected_version=0`, 성공하면 그 entry version만 증가한다. `version_conflict`, `slot_inactive`, `state_too_large`, `bad_request:*`는 write를 적용하지 않는다.
3. **write (별도 RPC, C-2)**: 상태가 바뀌면 `trex.slots.strategy_state.set` 으로 `{slot_id, strategy_id, expected_version, expected_occupancy_seq, data}` 를 보낸다. core 는 슬롯의 현재 occupancy_seq·active 여부를 조회해 CAS 검증: 점유 불일치 → `occupancy_changed`, version 불일치 → `version_conflict`, inactive/미존재 → `slot_inactive`, size 초과 → `state_too_large`, 일치(없으면 expected_version=0) → commit + version+1. malformed → `bad_request`. 동일 입력 재전송은 무부작용. 3. **lifecycle/concurrency**: `→INACTIVE`와 재점유는 map을 삭제하거나 숨기지 않으며 version도 초기화하지 않는다. Slot 자체의 `occupancy_seq`는 기존대로 유지하지만 strategy state persistence/read/write에는 사용하지 않는다. 서로 다른 instance entry의 concurrent 성공 write는 모두 보존된다.
4. **occupancy 가드 (C-2b, R-1)**: 슬롯은 monotonic `occupancy_seq` 를 갖고 새 점유 시작(`inactive→pending_buy`)마다 ++. 영구 슬롯 재사용 시 read v0→close→reopen→write v0 의 ABA(옛 점유 계산이 새 점유에 바인딩)를 occupancy_seq CAS 가 구조적으로 차단한다.
5. **close 삭제 (C-3)**: 슬롯이 어떤 경로로든 `→INACTIVE`(전략 매도/safety-sell/forced-close/operator) 되면 `transit_slot` 단일 전이 지점에서 해당 slot_id 의 모든 strategy_state 를 삭제한다. 재사용 슬롯은 새 occupancy_seq + 빈 상태로 시작 — 전략/데몬은 cleanup 책임이 없다(슬롯이 source of truth + GC).
저장은 별도 `slot_strategy_state` 테이블(복합 PK `(slot_id, strategy_id)`), 슬롯 본체에는 `occupancy_seq` 컬럼 1개만 추가(full 3계층 분할은 후속 Discovery). full 3계층(occupancy 테이블 분할)·다중 writer 동시성·장기 보존은 범위 밖. 저장은 별도 `slot_strategy_state` 테이블의 slot-scoped JSON map이며 entry에는 identity metadata를 중복 저장하지 않는다. core-side instance registry, legacy mapping/backfill, caller 인증은 없다.
## Flow Constraints ## Flow Constraints
tests/adapters/persistence/test_postgresql_repository.py
+20 -19
View File
@@ -155,8 +155,10 @@ def test_cancel_attempt_reservation_is_compare_and_set(pg_clean):
assert repo.fetch_slot(decision.slot_id).cancel_state == "requested" assert repo.fetch_slot(decision.slot_id).cancel_state == "requested"
# dcy-028 / REQ-079·081 (BAT-042): strategy-state store + occupancy_seq column # REQ-0009: slot-scoped strategy-instance state map.
def test_strategy_state_store_pg(pg_clean): def test_strategy_state_store_pg(pg_clean):
from concurrent.futures import ThreadPoolExecutor
from trex_core.adapters.persistence.postgresql import PgStrategyStateStore from trex_core.adapters.persistence.postgresql import PgStrategyStateStore
from trex_core.application import strategy_state as ss from trex_core.application import strategy_state as ss
@@ -169,31 +171,30 @@ def test_strategy_state_store_pg(pg_clean):
reason="buy", actor="t") reason="buy", actor="t")
svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="t", qty=10, avg_price=100) reason="filled", actor="t", qty=10, avg_price=100)
# occupancy_seq persisted r1 = store.set(slot.slot_id, "instance-a", expected_version=0, data="peak", slot_active=True)
held = repo.fetch_slot(slot.slot_id)
assert held.occupancy_seq == 1
# first-create
r1 = store.set(slot.slot_id, "tsl_v1", expected_version=0,
expected_occupancy_seq=1, data="peak", current_occupancy_seq=1,
slot_active=True)
assert r1.ok and r1.version == 1 assert r1.ok and r1.version == 1
# version conflict r2 = store.set(slot.slot_id, "instance-a", expected_version=0, data="stale", slot_active=True)
r2 = store.set(slot.slot_id, "tsl_v1", expected_version=0,
expected_occupancy_seq=1, data="stale", current_occupancy_seq=1,
slot_active=True)
assert not r2.ok and r2.reason == ss.REASON_VERSION_CONFLICT assert not r2.ok and r2.reason == ss.REASON_VERSION_CONFLICT
# batched read missing = store.set(slot.slot_id, "instance-missing", expected_version=1, data="stale", slot_active=True)
got = store.get_for_slots([slot.slot_id], {slot.slot_id: 1}) assert not missing.ok and missing.reason == ss.REASON_VERSION_CONFLICT and missing.version == 0
assert got[slot.slot_id]["tsl_v1"].data == "peak" store.set(slot.slot_id, "instance-b", expected_version=0, data="other", slot_active=True)
got = store.get_for_slots([slot.slot_id])
assert got[slot.slot_id]["instance-a"].data == "peak"
assert got[slot.slot_id]["instance-b"].data == "other"
with ThreadPoolExecutor(max_workers=2) as executor:
writes = list(executor.map(
lambda item: store.set(slot.slot_id, item, expected_version=0, data=item, slot_active=True),
("instance-c", "instance-d"),
))
assert all(write.ok for write in writes)
assert {"instance-c", "instance-d"} <= set(store.get_for_slots([slot.slot_id])[slot.slot_id])
# close deletes state (C-3)
svc.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, svc.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL,
reason="sell", actor="t") reason="sell", actor="t")
svc.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, svc.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
reason="sold", actor="t") reason="sold", actor="t")
assert store.get_for_slots([slot.slot_id], {slot.slot_id: 1}) == {} assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 1
# reuse bumps occupancy_seq
reused = svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reused = svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="t") reason="buy", actor="t")
assert reused.occupancy_seq == 2 assert reused.occupancy_seq == 2
tests/application/test_ledger_option_d.py
+1 -1
View File
@@ -107,7 +107,7 @@ def test_apply_cash_estimate_overlays_pending_delta():
repo = InMemorySlotRepository() repo = InMemorySlotRepository()
repo.reconcile_universe("default", {"PF-1": "SYM"}) repo.reconcile_universe("default", {"PF-1": "SYM"})
handler = make_slots_rpc_handler(repo, snapshot_store=store, ledger_outbox=outbox) handler = make_slots_rpc_handler(repo, snapshot_store=store, ledger_outbox=outbox)
resp = handler(RpcRequest(payload={"filter": "all"}, correlation_id="test")) resp = handler(RpcRequest(payload={"filter": "all", "strategy_instance_id": "test-instance"}, correlation_id="test"))
slots = resp.payload["slots"] slots = resp.payload["slots"]
pf1 = next(s for s in slots if s["portfolio_id"] == "PF-1") pf1 = next(s for s in slots if s["portfolio_id"] == "PF-1")
assert pf1["cash_estimated"] is True assert pf1["cash_estimated"] is True
tests/application/test_reconcile_warning_blocking.py
+2 -2
View File
@@ -693,7 +693,7 @@ def _wire_with_warn(warned_slot_ids=None):
def _req(filter=None): def _req(filter=None):
return RpcRequest(payload={"filter": filter or "all"}, correlation_id="t") return RpcRequest(payload={"filter": filter or "all", "spy": True}, correlation_id="t")
# REQ-104 AC-5: reconcile_warned=True for warned slot # REQ-104 AC-5: reconcile_warned=True for warned slot
@@ -756,7 +756,7 @@ def test_slots_rpc_existing_warned_field_unchanged():
reason="filled", actor="test", avg_price=100.0, qty=10.0) reason="filled", actor="test", avg_price=100.0, qty=10.0)
warn_store = InMemoryReconcileWarningStore() warn_store = InMemoryReconcileWarningStore()
handler = make_slots_rpc_handler(repo, snapshot_store=store, warn_store=warn_store) handler = make_slots_rpc_handler(repo, snapshot_store=store, warn_store=warn_store)
resp = handler(RpcRequest(payload={"filter": "holding"}, correlation_id="t")) resp = handler(RpcRequest(payload={"filter": "holding", "spy": True}, correlation_id="t"))
row = resp.payload["slots"][0] row = resp.payload["slots"][0]
# slot.qty=10 != hub asset_quantity=5 => warned=True (existing field) # slot.qty=10 != hub asset_quantity=5 => warned=True (existing field)
assert row["warned"] is True assert row["warned"] is True
tests/application/test_slots_handler.py
+2
View File
@@ -43,6 +43,8 @@ def _req(filter=None, spy=False):
payload["filter"] = filter payload["filter"] = filter
if spy: if spy:
payload["spy"] = True payload["spy"] = True
else:
payload["strategy_instance_id"] = "test-instance"
return RpcRequest(payload=payload, correlation_id="test-corr") return RpcRequest(payload=payload, correlation_id="test-corr")
tests/application/test_slots_handler_quantity_mismatch.py
+1
View File
@@ -74,6 +74,7 @@ def _make_handler(repo, warn_store, snapshot_items=None):
def _req(filter=None): def _req(filter=None):
payload = {} payload = {}
payload["spy"] = True
if filter: if filter:
payload["filter"] = filter payload["filter"] = filter
return RpcRequest(payload=payload, correlation_id="test") return RpcRequest(payload=payload, correlation_id="test")
tests/application/test_slots_handler_strategy.py
+2
View File
@@ -67,6 +67,8 @@ def _req(filter=None, spy=False, strategy=None):
payload["filter"] = filter payload["filter"] = filter
if spy: if spy:
payload["spy"] = True payload["spy"] = True
else:
payload["strategy_instance_id"] = "test-instance"
if strategy is not None: if strategy is not None:
payload["strategy"] = strategy payload["strategy"] = strategy
return RpcRequest(payload=payload, correlation_id="test-corr") return RpcRequest(payload=payload, correlation_id="test-corr")
tests/application/test_slots_handler_strategy_state.py
+39 -72
View File
@@ -1,93 +1,60 @@
"""slots.req read piggyback of strategy_state + occupancy_seq (dcy-028 C-1). """Strategy and operator strategy-state visibility contracts for slots.req."""
REQ-073 AC-7 / REQ-074 AC-6 / S-1: each slot row carries occupancy_seq and,
when present, a strategy_state map assembled from the store in one batched read.
strategy_* reasoning fields remain absent.
"""
from __future__ import annotations from __future__ import annotations
import pytest
from trex_core.application.messaging import RpcRequest from trex_core.application.messaging import RpcRequest
from trex_core.application.messaging_handlers import make_slots_rpc_handler from trex_core.application.messaging_handlers import make_slots_rpc_handler
from trex_core.application.strategy_state import InMemoryStrategyStateStore from trex_core.application.strategy_state import InMemoryStrategyStateStore
from trex_core.application.watchdog import WatchdogStore
from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot import SlotService, SlotState
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
INST = "default"
def _wire(): def _wire():
repo = InMemorySlotRepository() repo, store, watchdog = InMemorySlotRepository(), InMemoryStrategyStateStore(), WatchdogStore()
store = InMemoryStrategyStateStore() service = SlotService(repo)
service = SlotService(repo, instance_key=INST, strategy_state_store=store) repo.reconcile_universe("default", {"pf": "SYM"})
repo.reconcile_universe(INST, {"pf-0": "SYM"}) slot = repo.list_slots("default")[0]
slot = repo.list_slots(INST)[0] service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t")
service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="fill", actor="t", qty=1, avg_price=1)
reason="buy", actor="t") store.set(slot.slot_id, "instance-a", expected_version=0, data="a", slot_active=True)
service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, store.set(slot.slot_id, "instance-b", expected_version=0, data="b", slot_active=True)
reason="filled", actor="t", qty=10, avg_price=100) return slot, store, watchdog, make_slots_rpc_handler(repo, strategy_state_store=store, watchdog_store=watchdog)
handler = make_slots_rpc_handler(repo, instance_key=INST,
strategy_state_store=store)
return repo, store, service, slot, handler
def _rows(handler): def _call(handler, payload):
return handler(RpcRequest(payload={}, correlation_id="c")).payload["slots"] return handler(RpcRequest(payload=payload, correlation_id="c")).payload
def test_row_carries_occupancy_seq(): def test_strategy_receives_only_exact_requested_entry():
_, _, _, slot, handler = _wire() _, _, _, handler = _wire()
row = _rows(handler)[0] payload = _call(handler, {"strategy_instance_id": "instance-a"})
assert row["occupancy_seq"] == 1 state = payload["slots"][0]["strategy_state"]
assert state["data"] == "a" and set(state) == {"version", "data", "updated_at"}
assert "instance-b" not in repr(payload)
def test_row_carries_strategy_state_when_present(): def test_strategy_omits_absent_entry():
_, store, _, slot, handler = _wire() _, _, _, handler = _wire()
store.set(slot.slot_id, "tsl_v1", expected_version=0, expected_occupancy_seq=1, row = _call(handler, {"strategy_instance_id": "absent"})["slots"][0]
data='{"peak":100}', current_occupancy_seq=1, slot_active=True)
row = _rows(handler)[0]
assert "strategy_state" in row
assert row["strategy_state"]["tsl_v1"]["version"] == 1
assert row["strategy_state"]["tsl_v1"]["data"] == '{"peak":100}'
assert "updated_at" in row["strategy_state"]["tsl_v1"]
def test_row_omits_strategy_state_when_absent():
_, _, _, _, handler = _wire()
row = _rows(handler)[0]
assert "strategy_state" not in row assert "strategy_state" not in row
def test_no_strategy_reasoning_fields(): @pytest.mark.parametrize("identity", [None, "", False, 1, []])
"""REQ-074 AC-5 (REQ-116 additive 재도입 후에도 유지): slot-origin strategy fields absent. def test_malformed_strategy_request_has_exact_safe_envelope(identity):
_, _, watchdog, handler = _wire()
strategy_params / strategy_name / markets are now present (REQ-116 PortfolioItem-origin); assert _call(handler, {"strategy_instance_id": identity}) == {
slot-origin strategy_id / sell_strategy_id / active_strategy_id remain absent. "slots": [], "available_slots": None, "degraded": False,
""" "reason": "bad_request:strategy_instance_id",
_, store, _, slot, handler = _wire() "message": "strategy_instance_id must be a non-empty string",
store.set(slot.slot_id, "tsl_v1", expected_version=0, expected_occupancy_seq=1, }
data="x", current_occupancy_seq=1, slot_active=True) assert watchdog.last_seen("unused") is None
row = _rows(handler)[0]
# slot-origin fields must be absent (REQ-074 AC-5 unchanged)
for k in ("strategy_id", "sell_strategy_id", "active_strategy_id"):
assert k not in row
# PortfolioItem-origin fields now present (REQ-116 additive)
assert "strategy_params" in row # {} (safe default, no snapshot_store)
assert "strategy_name" in row # None
assert "markets" in row # []
def test_reused_slot_shows_no_state(): def test_operator_without_or_with_id_receives_full_map_without_watchdog_update():
"""After a full close + reuse cycle, no prior-occupancy state surfaces slot, _, watchdog, handler = _wire()
(close-delete + read-side occupancy filter). occupancy_seq advances to 2.""" for payload in ({"spy": True}, {"spy": True, "strategy_instance_id": "instance-a"}):
repo, store, service, slot, handler = _wire() state = _call(handler, payload)["slots"][0]["strategy_state"]
store.set(slot.slot_id, "tsl_v1", expected_version=0, expected_occupancy_seq=1, assert set(state) == {"instance-a", "instance-b"}
data="old", current_occupancy_seq=1, slot_active=True) assert watchdog.last_seen(str(slot.slot_id)) is None
service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL,
reason="sell", actor="t")
service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
reason="sold", actor="t")
service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="t") # occupancy_seq -> 2
row = _rows(handler)[0]
assert row["occupancy_seq"] == 2
assert "strategy_state" not in row
tests/application/test_strategy_state_set_handler.py
+29 -99
View File
@@ -1,10 +1,8 @@
"""strategy_state.set write RPC handler tests (dcy-028 / REQ-080, BAT-042). """strategy_state.set strategy-instance wire contract tests."""
Covers the full reject reason set + success + bad_request, with the handler
supplying the slot's current occupancy_seq / active-ness to the store CAS.
"""
from __future__ import annotations from __future__ import annotations
import pytest
from trex_core.application import strategy_state as ss from trex_core.application import strategy_state as ss
from trex_core.application.messaging import RpcRequest from trex_core.application.messaging import RpcRequest
from trex_core.application.messaging_handlers import make_strategy_state_set_handler from trex_core.application.messaging_handlers import make_strategy_state_set_handler
@@ -12,109 +10,41 @@ from trex_core.application.strategy_state import InMemoryStrategyStateStore
from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot import SlotService, SlotState
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
INST = "default"
def _wire(): def _wire():
repo = InMemorySlotRepository() repo, store = InMemorySlotRepository(), InMemoryStrategyStateStore()
store = InMemoryStrategyStateStore() service = SlotService(repo)
service = SlotService(repo, instance_key=INST, strategy_state_store=store) repo.reconcile_universe("default", {"pf": "SYM"})
repo.reconcile_universe(INST, {"pf-0": "SYM"}) slot = repo.list_slots("default")[0]
slot = repo.list_slots(INST)[0] service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t")
# activate to holding (occupancy_seq -> 1) service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="fill", actor="t", qty=1, avg_price=1)
service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, return repo, store, service, slot, make_strategy_state_set_handler(repo, store)
reason="buy", actor="t")
service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING,
reason="filled", actor="t", qty=10, avg_price=100)
handler = make_strategy_state_set_handler(repo, store)
return repo, store, service, slot, handler
def _req(slot_id, *, ev=0, eo=1, data="peak", strat="tsl_v1"): def _req(slot_id, instance="instance-a", version=0, data="state"):
return RpcRequest(payload={ return RpcRequest(payload={"slot_id": str(slot_id), "strategy_instance_id": instance, "expected_version": version, "data": data}, correlation_id="same-correlation-id")
"slot_id": str(slot_id), "strategy_id": strat,
"expected_version": ev, "expected_occupancy_seq": eo, "data": data,
}, correlation_id=f"{slot_id}:{strat}:{ev}")
def test_first_write_ok(): def test_write_uses_instance_identity_and_entry_cas():
_, _, _, slot, handler = _wire() _, _, _, slot, handler = _wire()
resp = handler(_req(slot.slot_id, ev=0)) assert handler(_req(slot.slot_id)).payload == {"ok": True, "slot_id": str(slot.slot_id), "strategy_instance_id": "instance-a", "version": 1}
assert resp.payload["ok"] is True and resp.payload["version"] == 1 conflict = handler(_req(slot.slot_id, version=0, data="stale")).payload
assert conflict["reason"] == ss.REASON_VERSION_CONFLICT and conflict["version"] == 1
def test_version_conflict(): @pytest.mark.parametrize("identity", [None, "", 1, True, []])
_, _, _, slot, handler = _wire() def test_malformed_instance_id_is_rejected_without_write(identity):
handler(_req(slot.slot_id, ev=0)) # version -> 1 _, store, _, slot, handler = _wire()
resp = handler(_req(slot.slot_id, ev=0, data="stale")) response = handler(_req(slot.slot_id, identity)).payload
assert resp.payload["ok"] is False assert response["ok"] is False
assert resp.payload["reason"] == ss.REASON_VERSION_CONFLICT assert response["reason"].startswith("bad_request:")
assert resp.payload["version"] == 1 assert store.get_for_slots([slot.slot_id]) == {}
def test_occupancy_changed(): def test_inactive_and_oversize_writes_do_not_change_existing_entry():
_, _, _, slot, handler = _wire()
# daemon read occupancy 0 (before activation) but slot is occupancy 1 now
resp = handler(_req(slot.slot_id, ev=0, eo=0))
assert resp.payload["ok"] is False
assert resp.payload["reason"] == ss.REASON_OCCUPANCY_CHANGED
assert resp.payload["occupancy_seq"] == 1
def test_slot_inactive():
repo, store, service, slot, handler = _wire() repo, store, service, slot, handler = _wire()
# close the slot handler(_req(slot.slot_id))
service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason="sell", actor="t")
reason="sell", actor="t") service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, reason="sold", actor="t")
service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, assert handler(_req(slot.slot_id, version=1)).payload["reason"] == ss.REASON_SLOT_INACTIVE
reason="sold", actor="t") assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 1
resp = handler(_req(slot.slot_id, ev=0, eo=1))
assert resp.payload["ok"] is False
assert resp.payload["reason"] == ss.REASON_SLOT_INACTIVE
def test_unknown_slot_is_slot_inactive():
from uuid import uuid4
_, _, _, _, handler = _wire()
resp = handler(_req(uuid4(), ev=0, eo=1))
assert resp.payload["ok"] is False
assert resp.payload["reason"] == ss.REASON_SLOT_INACTIVE
def test_state_too_large():
_, _, _, slot, handler = _wire()
big = "a" * (ss._STATE_CAP_BYTES + 1)
resp = handler(_req(slot.slot_id, ev=0, data=big))
assert resp.payload["ok"] is False
assert resp.payload["reason"] == ss.REASON_STATE_TOO_LARGE
def test_bad_request_missing_field():
_, _, _, slot, handler = _wire()
resp = handler(RpcRequest(payload={"slot_id": str(slot.slot_id)},
correlation_id="c"))
assert resp.payload["ok"] is False
assert resp.payload["reason"].startswith("bad_request:")
def test_bad_request_non_string_data():
_, _, _, slot, handler = _wire()
resp = handler(RpcRequest(payload={
"slot_id": str(slot.slot_id), "strategy_id": "tsl_v1",
"expected_version": 0, "expected_occupancy_seq": 1, "data": 123,
}, correlation_id="c"))
assert resp.payload["ok"] is False
assert resp.payload["reason"].startswith("bad_request:")
def test_reason_set_is_enumerated():
"""All non-ok reasons must be within the documented set (REQ-080 AC-6)."""
allowed_prefixes = {
ss.REASON_OCCUPANCY_CHANGED, ss.REASON_VERSION_CONFLICT,
ss.REASON_SLOT_INACTIVE, ss.REASON_STATE_TOO_LARGE, "bad_request",
}
_, _, _, slot, handler = _wire()
handler(_req(slot.slot_id, ev=0))
bad = handler(_req(slot.slot_id, ev=0, data="x")) # version_conflict
reason = bad.payload["reason"].split(":")[0]
assert reason in allowed_prefixes
tests/application/test_strategy_state_store.py
+35 -100
View File
@@ -1,118 +1,53 @@
"""InMemoryStrategyStateStore CAS tests (dcy-028 / REQ-079, BAT-042). """Slot-scoped strategy-instance state store contract tests."""
Covers the 6 CAS outcomes (occupancy+version commit / version_conflict /
occupancy_changed / slot_inactive / state_too_large / first-create), idempotent
re-send, batched read with stale-occupancy filtering, and delete-on-close.
"""
from __future__ import annotations from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor
from uuid import uuid4 from uuid import uuid4
import pytest
from trex_core.application import strategy_state as ss from trex_core.application import strategy_state as ss
from trex_core.application.strategy_state import InMemoryStrategyStateStore from trex_core.application.strategy_state import InMemoryStrategyStateStore
@pytest.fixture def _set(store, slot_id, instance="instance-a", *, version=0, data="x", active=True):
def store():
return InMemoryStrategyStateStore()
def _set(store, slot_id, *, ev=0, eo=1, data="x", cur_occ=1, active=True, strat="tsl_v1"):
return store.set( return store.set(
slot_id, strat, slot_id, instance, expected_version=version, data=data, slot_active=active
expected_version=ev, expected_occupancy_seq=eo, data=data,
current_occupancy_seq=cur_occ, slot_active=active,
) )
class TestCas: def test_entry_versions_are_isolated():
def test_first_create(self, store): store, slot_id = InMemoryStrategyStateStore(), uuid4()
sid = uuid4() assert _set(store, slot_id, "a", data="a1").version == 1
r = _set(store, sid, ev=0, eo=1, cur_occ=1, data='{"peak":100}') assert _set(store, slot_id, "b", data="b1").version == 1
assert r.ok and r.version == 1 assert _set(store, slot_id, "a", version=1, data="a2").version == 2
entries = store.get_for_slots([slot_id])[slot_id]
def test_commit_increments_version(self, store): assert entries["a"].version == 2 and entries["a"].data == "a2"
sid = uuid4() assert entries["b"].version == 1 and entries["b"].data == "b1"
_set(store, sid, ev=0) assert not hasattr(entries["a"], "strategy_id")
r = _set(store, sid, ev=1, data="next") assert not hasattr(entries["a"], "occupancy_seq")
assert r.ok and r.version == 2
def test_version_conflict_does_not_store(self, store):
sid = uuid4()
_set(store, sid, ev=0, data="first") # version=1
r = _set(store, sid, ev=0, data="stale") # expects 0, stored 1
assert not r.ok and r.reason == ss.REASON_VERSION_CONFLICT and r.version == 1
# original preserved
got = store.get_for_slots([sid], {sid: 1})[sid]["tsl_v1"]
assert got.data == "first" and got.version == 1
def test_occupancy_changed_does_not_store(self, store):
sid = uuid4()
# daemon read occupancy_seq=1 but slot is now occupancy 2
r = _set(store, sid, ev=0, eo=1, cur_occ=2)
assert not r.ok and r.reason == ss.REASON_OCCUPANCY_CHANGED and r.occupancy_seq == 2
assert store.get_for_slots([sid], {sid: 2}) == {}
def test_slot_inactive_does_not_store(self, store):
sid = uuid4()
r = _set(store, sid, ev=0, active=False)
assert not r.ok and r.reason == ss.REASON_SLOT_INACTIVE
assert store.get_for_slots([sid], {sid: 1}) == {}
def test_state_too_large(self, store, monkeypatch):
sid = uuid4()
big = "a" * (ss._STATE_CAP_BYTES + 1)
r = _set(store, sid, ev=0, data=big)
assert not r.ok and r.reason == ss.REASON_STATE_TOO_LARGE
def test_idempotent_resend_is_conflict_not_double_apply(self, store):
sid = uuid4()
r1 = _set(store, sid, ev=0, data="v1") # -> version 1
assert r1.ok and r1.version == 1
# re-send identical expected_version=0 -> rejected (already advanced)
r2 = _set(store, sid, ev=0, data="v1")
assert not r2.ok and r2.reason == ss.REASON_VERSION_CONFLICT
class TestAbaGuard: def test_conflict_inactive_and_size_limit_preserve_entries(monkeypatch):
def test_turnover_first_write_rejected(self, store): store, slot_id = InMemoryStrategyStateStore(), uuid4()
"""dcy-028 R-1 / S-4: read v0 of occupancy 1, slot reused as occupancy 2, _set(store, slot_id, "a", data="saved")
stale write against occupancy 1 must not bind to occupancy 2.""" assert _set(store, slot_id, "a", data="stale").reason == ss.REASON_VERSION_CONFLICT
sid = uuid4() assert _set(store, slot_id, "b", active=False).reason == ss.REASON_SLOT_INACTIVE
# occupancy 2 is current; daemon computed against occupancy 1 (eo=1) assert _set(store, slot_id, "b", data="x" * (ss._STATE_CAP_BYTES + 1)).reason == ss.REASON_STATE_TOO_LARGE
r = _set(store, sid, ev=0, eo=1, cur_occ=2, data="stale-peak") entries = store.get_for_slots([slot_id])[slot_id]
assert not r.ok and r.reason == ss.REASON_OCCUPANCY_CHANGED assert set(entries) == {"a"} and entries["a"].data == "saved"
# new occupancy starts clean
assert store.get_for_slots([sid], {sid: 2}) == {}
class TestBatchedRead: def test_concurrent_different_entries_both_survive():
def test_get_for_slots_groups_by_slot_and_strategy(self, store): store, slot_id = InMemoryStrategyStateStore(), uuid4()
a, b = uuid4(), uuid4() with ThreadPoolExecutor(max_workers=2) as executor:
_set(store, a, ev=0, cur_occ=1, strat="tsl_v1", data="A") results = list(executor.map(lambda key: _set(store, slot_id, key, data=key), ("a", "b")))
_set(store, a, ev=0, cur_occ=1, strat="buy_v1", data="A2") assert all(result.ok for result in results)
_set(store, b, ev=0, cur_occ=1, strat="tsl_v1", data="B") assert set(store.get_for_slots([slot_id])[slot_id]) == {"a", "b"}
out = store.get_for_slots([a, b], {a: 1, b: 1})
assert set(out[a]) == {"tsl_v1", "buy_v1"}
assert out[a]["tsl_v1"].data == "A"
assert out[b]["tsl_v1"].data == "B"
def test_stale_occupancy_entries_invisible(self, store):
sid = uuid4()
_set(store, sid, ev=0, cur_occ=1, data="old") # stored at occupancy 1
# slot is now occupancy 2 -> the occupancy-1 entry must not surface
assert store.get_for_slots([sid], {sid: 2}) == {}
def test_empty_slot_ids(self, store):
assert store.get_for_slots([], {}) == {}
class TestDelete: def test_same_entry_competing_cas_has_one_winner():
def test_delete_all_removes_every_strategy(self, store): store, slot_id = InMemoryStrategyStateStore(), uuid4()
sid = uuid4() _set(store, slot_id, "a", data="first")
_set(store, sid, ev=0, strat="tsl_v1") with ThreadPoolExecutor(max_workers=2) as executor:
_set(store, sid, ev=0, strat="buy_v1") results = list(executor.map(lambda data: _set(store, slot_id, "a", version=1, data=data), ("x", "y")))
store.delete_all(sid) assert sum(result.ok for result in results) == 1
assert store.get_for_slots([sid], {sid: 1}) == {} assert store.get_for_slots([slot_id])[slot_id]["a"].version == 2
tests/domain/slot/test_occupancy_seq_and_close_delete.py
+16 -113
View File
@@ -1,120 +1,23 @@
"""Slot occupancy_seq bump + close-delete hook (dcy-028 / REQ-081, BAT-042). """Slot occupancy remains intact while strategy state survives close/reopen."""
S-3: every close path (→INACTIVE) deletes the slot's strategy_state and a reused
slot starts with a new occupancy_seq + empty state. occupancy_seq is stable
across one holding episode and bumps on inactive→pending_buy.
"""
from __future__ import annotations from __future__ import annotations
from trex_core.application.strategy_state import InMemoryStrategyStateStore from trex_core.application.strategy_state import InMemoryStrategyStateStore
from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot import SlotService, SlotState
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
INST = "default"
def test_close_preserves_state_and_reopen_continues_version():
def _seed(repo): repo, store = InMemorySlotRepository(), InMemoryStrategyStateStore()
repo.reconcile_universe(INST, {"pf-0": "SYM"}) service = SlotService(repo, strategy_state_store=store)
return repo.list_slots(INST)[0] repo.reconcile_universe("default", {"pf": "SYM"})
slot = repo.list_slots("default")[0]
service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t")
def _service(repo, store=None): service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="fill", actor="t", qty=1, avg_price=1)
return SlotService(repo, instance_key=INST, strategy_state_store=store) assert store.set(slot.slot_id, "instance", expected_version=0, data="old", slot_active=True).version == 1
service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason="sell", actor="t")
inactive = service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, reason="sold", actor="t")
def _activate_to_holding(service, slot_id): assert inactive.occupancy_seq == 1
service.transit_slot(slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance"].data == "old"
reason="buy", actor="t") reopened = service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t")
return service.transit_slot(slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, assert reopened.occupancy_seq == 2
reason="filled", actor="t", qty=10, avg_price=100) assert store.set(slot.slot_id, "instance", expected_version=1, data="new", slot_active=True).version == 2
class TestOccupancySeq:
def test_bumps_on_inactive_to_pending_buy(self):
repo = InMemorySlotRepository()
service = _service(repo)
slot = _seed(repo)
assert slot.occupancy_seq == 0
updated = service.transit_slot(
slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY,
reason="buy", actor="t")
assert updated.occupancy_seq == 1
def test_stable_across_holding_episode(self):
repo = InMemorySlotRepository()
service = _service(repo)
slot = _seed(repo)
held = _activate_to_holding(service, slot.slot_id)
assert held.occupancy_seq == 1
sold = service.transit_slot(slot.slot_id, SlotState.HOLDING,
SlotState.PENDING_SELL, reason="sell", actor="t")
assert sold.occupancy_seq == 1 # unchanged within the episode
def test_increments_on_reuse(self):
repo = InMemorySlotRepository()
service = _service(repo)
slot = _seed(repo)
_activate_to_holding(service, slot.slot_id)
service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL,
reason="sell", actor="t")
service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
reason="sold", actor="t")
# reuse: occupancy_seq bumps from 1 to 2
reused = service.transit_slot(slot.slot_id, SlotState.INACTIVE,
SlotState.PENDING_BUY, reason="buy", actor="t")
assert reused.occupancy_seq == 2
class TestCloseDelete:
def _setup(self):
repo = InMemorySlotRepository()
store = InMemoryStrategyStateStore()
service = _service(repo, store)
slot = _seed(repo)
_activate_to_holding(service, slot.slot_id)
# daemon writes state during holding (occupancy_seq=1)
store.set(slot.slot_id, "tsl_v1", expected_version=0,
expected_occupancy_seq=1, data="peak", current_occupancy_seq=1,
slot_active=True)
assert store.get_for_slots([slot.slot_id], {slot.slot_id: 1})
return repo, store, service, slot
def test_sell_fill_close_deletes_state(self):
repo, store, service, slot = self._setup()
service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL,
reason="sell", actor="t")
service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
reason="sold", actor="t")
assert store.get_for_slots([slot.slot_id], {slot.slot_id: 1}) == {}
def test_reused_slot_starts_empty_with_new_occupancy(self):
repo, store, service, slot = self._setup()
service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL,
reason="sell", actor="t")
service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE,
reason="sold", actor="t")
reused = service.transit_slot(slot.slot_id, SlotState.INACTIVE,
SlotState.PENDING_BUY, reason="buy", actor="t")
assert reused.occupancy_seq == 2
assert store.get_for_slots([slot.slot_id], {slot.slot_id: 2}) == {}
def test_error_close_path_also_deletes(self):
repo, store, service, slot = self._setup()
# holding → error → inactive (operator resolve) also clears state
service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.ERROR,
reason="err", actor="t")
service.transit_slot(slot.slot_id, SlotState.ERROR, SlotState.INACTIVE,
reason="resolve", actor="t")
assert store.get_for_slots([slot.slot_id], {slot.slot_id: 1}) == {}
def test_no_store_injected_is_noop(self):
# backward-compat: service without store does not raise on close
repo = InMemorySlotRepository()
service = _service(repo, store=None)
slot = _seed(repo)
_activate_to_holding(service, slot.slot_id)
service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL,
reason="sell", actor="t")
out = service.transit_slot(slot.slot_id, SlotState.PENDING_SELL,
SlotState.INACTIVE, reason="sold", actor="t")
assert out.status is SlotState.INACTIVE
trex_core/adapters/persistence/postgresql.py
+48 -70
View File
@@ -28,7 +28,7 @@ from trex_core.application.ports import IntentRecord
from trex_core.application.strategy_state import ( from trex_core.application.strategy_state import (
SetResult, SetResult,
StrategyStateEntry, StrategyStateEntry,
_cas, _STATE_CAP_BYTES,
) )
from trex_core.domain.slot import Slot, SlotNotFoundError, SlotState, SlotTransitionRecord from trex_core.domain.slot import Slot, SlotNotFoundError, SlotState, SlotTransitionRecord
@@ -701,13 +701,7 @@ class PostgreSQLLedgerOutbox:
class PgStrategyStateStore: class PgStrategyStateStore:
"""slot_strategy_state-backed StrategyStateStore (dcy-028 / REQ-079). """PostgreSQL slot-scoped strategy-instance map store."""
Stores opaque per-strategy state keyed by (slot_id, strategy_id). The CAS
decision is the same `_cas` used by the in-memory double; occupancy_seq and
slot_active are supplied by the caller (fetched from the slot) so the store
only owns the slot_strategy_state table.
"""
def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None: def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
self._psycopg = _require_psycopg(psycopg_module) self._psycopg = _require_psycopg(psycopg_module)
@@ -716,71 +710,65 @@ class PgStrategyStateStore:
def set( def set(
self, self,
slot_id: UUID, slot_id: UUID,
strategy_id: str, strategy_instance_id: str,
*, *,
expected_occupancy_seq: int,
expected_version: int, expected_version: int,
data: str, data: str,
current_occupancy_seq: int,
slot_active: bool, slot_active: bool,
) -> SetResult: ) -> SetResult:
if not slot_active:
return SetResult(ok=False, reason="slot_inactive")
if len(data.encode("utf-8")) > _STATE_CAP_BYTES:
return SetResult(ok=False, reason="state_too_large")
now = datetime.now(tz=timezone.utc)
with self._psycopg.connect(self._dsn, autocommit=False) as conn: with self._psycopg.connect(self._dsn, autocommit=False) as conn:
try: try:
with conn.cursor() as cur: with conn.cursor() as cur:
cur.execute( cur.execute(
"SELECT occupancy_seq, version, data, updated_at " "INSERT INTO slot_strategy_state (slot_id, data) "
"FROM slot_strategy_state " "SELECT %s, jsonb_build_object(%s, jsonb_build_object("
"WHERE slot_id = %s AND strategy_id = %s FOR UPDATE", "'version', 1, 'data', %s, 'updated_at', %s)) WHERE %s = 0 "
(slot_id, strategy_id), "ON CONFLICT (slot_id) DO UPDATE SET data = jsonb_set("
) "slot_strategy_state.data, ARRAY[%s], jsonb_build_object("
row = cur.fetchone() "'version', %s + 1, 'data', %s, 'updated_at', %s), true) "
existing = ( "WHERE COALESCE((slot_strategy_state.data -> %s ->> 'version')::bigint, 0) = %s "
StrategyStateEntry( "RETURNING data -> %s ->> 'version'",
strategy_id=strategy_id,
occupancy_seq=int(row[0]),
version=int(row[1]),
data=row[2],
updated_at=row[3],
)
if row is not None
else None
)
result = _cas(
existing,
expected_occupancy_seq=expected_occupancy_seq,
expected_version=expected_version,
data=data,
current_occupancy_seq=current_occupancy_seq,
slot_active=slot_active,
)
if not result.ok:
conn.rollback()
return result
cur.execute(
"INSERT INTO slot_strategy_state "
"(slot_id, strategy_id, occupancy_seq, version, data, updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s) "
"ON CONFLICT (slot_id, strategy_id) DO UPDATE SET "
"occupancy_seq = EXCLUDED.occupancy_seq, "
"version = EXCLUDED.version, data = EXCLUDED.data, "
"updated_at = EXCLUDED.updated_at",
( (
slot_id, slot_id,
strategy_id, strategy_instance_id,
current_occupancy_seq,
result.version,
data, data,
datetime.now(tz=timezone.utc), now,
expected_version,
strategy_instance_id,
expected_version,
data,
now,
strategy_instance_id,
expected_version,
strategy_instance_id,
), ),
) )
row = cur.fetchone()
if row is None:
cur.execute(
"SELECT COALESCE((data -> %s ->> 'version')::bigint, 0) "
"FROM slot_strategy_state WHERE slot_id = %s",
(strategy_instance_id, slot_id),
)
conflict = cur.fetchone()
return SetResult(
ok=False,
reason="version_conflict",
version=int(conflict[0]) if conflict is not None else 0,
)
conn.commit() conn.commit()
return result return SetResult(ok=True, version=int(row[0]))
except BaseException: except BaseException:
conn.rollback() conn.rollback()
raise raise
def get_for_slots( def get_for_slots(
self, slot_ids: list[UUID], current_occupancy_seq: dict[UUID, int] self, slot_ids: list[UUID]
) -> dict[UUID, dict[str, StrategyStateEntry]]: ) -> dict[UUID, dict[str, StrategyStateEntry]]:
if not slot_ids: if not slot_ids:
return {} return {}
@@ -788,26 +776,16 @@ class PgStrategyStateStore:
with self._psycopg.connect(self._dsn) as conn: with self._psycopg.connect(self._dsn) as conn:
with conn.cursor() as cur: with conn.cursor() as cur:
cur.execute( cur.execute(
"SELECT slot_id, strategy_id, occupancy_seq, version, data, updated_at " "SELECT slot_id, data "
"FROM slot_strategy_state WHERE slot_id = ANY(%s)", "FROM slot_strategy_state WHERE slot_id = ANY(%s)",
(list(slot_ids),), (list(slot_ids),),
) )
for row in cur.fetchall(): for row in cur.fetchall():
sid = row[0] if isinstance(row[0], UUID) else UUID(str(row[0])) sid = row[0] if isinstance(row[0], UUID) else UUID(str(row[0]))
# stale-occupancy entries are invisible (dcy-028 FR-3 defense) for instance_id, entry in row[1].items():
if int(row[2]) != current_occupancy_seq.get(sid): out.setdefault(sid, {})[instance_id] = StrategyStateEntry(
continue version=int(entry["version"]),
out.setdefault(sid, {})[row[1]] = StrategyStateEntry( data=entry["data"],
strategy_id=row[1], updated_at=datetime.fromisoformat(entry["updated_at"]),
occupancy_seq=int(row[2]), )
version=int(row[3]),
data=row[4],
updated_at=row[5],
)
return out return out
def delete_all(self, slot_id: UUID) -> None:
with self._psycopg.connect(self._dsn, autocommit=False) as conn:
with conn.cursor() as cur:
cur.execute("DELETE FROM slot_strategy_state WHERE slot_id = %s", (slot_id,))
conn.commit()
trex_core/adapters/persistence/sql/027_replace_slot_strategy_state_with_instance_map.sql
+18
View File
@@ -0,0 +1,18 @@
-- REQ-0009: state is one opaque map per slot, keyed by strategy_instance_id.
-- The old strategy_id/occupancy-scoped rows cannot be assigned an instance ID;
-- no mapping or backfill is permitted by this contract.
DO $$
BEGIN
IF EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name = 'slot_strategy_state'
AND column_name = 'strategy_id'
) THEN
DROP TABLE slot_strategy_state;
END IF;
END $$;
CREATE TABLE slot_strategy_state (
slot_id UUID PRIMARY KEY,
data JSONB NOT NULL DEFAULT '{}'::jsonb
);
trex_core/application/messaging_handlers.py
+38 -32
View File
@@ -207,7 +207,7 @@ def make_slots_rpc_handler(
): ):
"""Build the RPC handler for `trex.slots.req` (REQ-073/074/116). """Build the RPC handler for `trex.slots.req` (REQ-073/074/116).
Request: {filter?: str, spy?: bool, strategy?: str} Request: {filter?: str, spy?: bool, strategy?: str, strategy_instance_id?: str}
filter: "all"|"inactive"|"pending"|"holding"|"error" (default "all") filter: "all"|"inactive"|"pending"|"holding"|"error" (default "all")
spy=true: watchdog not updated (operator view, REQ-034 CHG-01) spy=true: watchdog not updated (operator view, REQ-034 CHG-01)
strategy: exact-match on strategy_name (REQ-116, optional). When specified strategy: exact-match on strategy_name (REQ-116, optional). When specified
@@ -224,6 +224,15 @@ def make_slots_rpc_handler(
body = req.payload or {} body = req.payload or {}
filter_val = str(body.get("filter") or "all").lower() filter_val = str(body.get("filter") or "all").lower()
spy = bool(body.get("spy", False)) spy = bool(body.get("spy", False))
strategy_instance_id = body.get("strategy_instance_id")
if not spy and (not isinstance(strategy_instance_id, str) or not strategy_instance_id):
return RpcResponse(payload={
"slots": [],
"available_slots": None,
"degraded": False,
"reason": "bad_request:strategy_instance_id",
"message": "strategy_instance_id must be a non-empty string",
})
# REQ-116 (BAT-055): optional strategy filter — exact-match on strategy_name. # REQ-116 (BAT-055): optional strategy filter — exact-match on strategy_name.
# None means no filter (all slots returned regardless of strategy_name). # None means no filter (all slots returned regardless of strategy_name).
strategy_filter = _opt_str(body.get("strategy")) strategy_filter = _opt_str(body.get("strategy"))
@@ -290,24 +299,29 @@ def make_slots_rpc_handler(
row["qty"] = None row["qty"] = None
row["qty_source"] = "unavailable" row["qty_source"] = "unavailable"
# dcy-028 / REQ-073/074·079 (C-1): attach opaque strategy_state map per # Strategy requests receive only their exact entry; operator requests
# row via one batched store read (read piggyback — no extra round trip). # receive the complete slot map without changing watchdog behavior.
# Only entries matching the slot's current occupancy_seq are surfaced.
if strategy_state_store is not None and slots: if strategy_state_store is not None and slots:
slot_ids = [s.slot_id for s in slots] slot_ids = [s.slot_id for s in slots]
occ = {s.slot_id: getattr(s, "occupancy_seq", 0) for s in slots} state_map = strategy_state_store.get_for_slots(slot_ids)
state_map = strategy_state_store.get_for_slots(slot_ids, occ)
for row, s in zip(slot_rows, slots, strict=True): for row, s in zip(slot_rows, slots, strict=True):
entries = state_map.get(s.slot_id) entries = state_map.get(s.slot_id)
if entries: if entries:
row["strategy_state"] = { if spy:
strat: { row["strategy_state"] = {
"version": e.version, instance_id: {
"data": e.data, "version": e.version,
"updated_at": e.updated_at.isoformat(), "data": e.data,
"updated_at": e.updated_at.isoformat(),
}
for instance_id, e in entries.items()
}
elif (entry := entries.get(strategy_instance_id)) is not None:
row["strategy_state"] = {
"version": entry.version,
"data": entry.data,
"updated_at": entry.updated_at.isoformat(),
} }
for strat, e in entries.items()
}
# dcy-018 / REQ-029: cash-estimated overlay (best-effort) # dcy-018 / REQ-029: cash-estimated overlay (best-effort)
if ledger_outbox is not None: if ledger_outbox is not None:
@@ -406,15 +420,12 @@ def make_strategy_state_set_handler(
): ):
"""Build the RPC handler for `trex.slots.strategy_state.set` (REQ-080). """Build the RPC handler for `trex.slots.strategy_state.set` (REQ-080).
Request: {slot_id, strategy_id, expected_version:int, Request: {slot_id, strategy_instance_id, expected_version:int, data:str}
expected_occupancy_seq:int, data:str} Response (ok): {ok:true, slot_id, strategy_instance_id, version:int}
Response (ok): {ok:true, slot_id, strategy_id, version:int} Response (fail): {ok:false, slot_id, strategy_instance_id, reason, version?}
Response (fail): {ok:false, slot_id, strategy_id, reason, version?|occupancy_seq?}
reason ∈ {occupancy_changed, version_conflict, slot_inactive, The store owns entry-version CAS; the handler supplies active-ness and never
state_too_large, bad_request:<detail>}. The store owns the CAS interprets opaque data.
(REQ-079); the handler supplies the slot's current occupancy_seq and
active-ness and never interprets `data` (opaque).
""" """
_ACTIVE_STATES = {"pending_buy", "holding", "pending_sell"} _ACTIVE_STATES = {"pending_buy", "holding", "pending_sell"}
@@ -422,9 +433,10 @@ def make_strategy_state_set_handler(
body = req.payload or {} body = req.payload or {}
try: try:
slot_id = UUID(str(body["slot_id"])) slot_id = UUID(str(body["slot_id"]))
strategy_id = str(body["strategy_id"]) strategy_instance_id = body["strategy_instance_id"]
if not isinstance(strategy_instance_id, str) or not strategy_instance_id:
raise ValueError("strategy_instance_id must be a non-empty string")
expected_version = int(body["expected_version"]) expected_version = int(body["expected_version"])
expected_occupancy_seq = int(body["expected_occupancy_seq"])
data = body["data"] data = body["data"]
if not isinstance(data, str): if not isinstance(data, str):
raise ValueError("data must be a string") raise ValueError("data must be a string")
@@ -432,27 +444,23 @@ def make_strategy_state_set_handler(
return RpcResponse(payload={ return RpcResponse(payload={
"ok": False, "ok": False,
"slot_id": body.get("slot_id"), "slot_id": body.get("slot_id"),
"strategy_id": body.get("strategy_id"), "strategy_instance_id": body.get("strategy_instance_id"),
"reason": f"bad_request:{exc}", "reason": f"bad_request:{exc}",
}) })
slot = slot_repo.fetch_slot(slot_id) slot = slot_repo.fetch_slot(slot_id)
slot_active = slot is not None and slot.status.value in _ACTIVE_STATES slot_active = slot is not None and slot.status.value in _ACTIVE_STATES
current_occupancy_seq = getattr(slot, "occupancy_seq", 0) if slot else 0
result = strategy_state_store.set( result = strategy_state_store.set(
slot_id, slot_id,
strategy_id, strategy_instance_id,
expected_occupancy_seq=expected_occupancy_seq,
expected_version=expected_version, expected_version=expected_version,
data=data, data=data,
current_occupancy_seq=current_occupancy_seq,
slot_active=slot_active, slot_active=slot_active,
) )
payload = { payload = {
"ok": result.ok, "ok": result.ok,
"slot_id": str(slot_id), "slot_id": str(slot_id),
"strategy_id": strategy_id, "strategy_instance_id": strategy_instance_id,
} }
if result.ok: if result.ok:
payload["version"] = result.version payload["version"] = result.version
@@ -460,8 +468,6 @@ def make_strategy_state_set_handler(
payload["reason"] = result.reason payload["reason"] = result.reason
if result.version is not None: if result.version is not None:
payload["version"] = result.version payload["version"] = result.version
if result.occupancy_seq is not None:
payload["occupancy_seq"] = result.occupancy_seq
return RpcResponse(payload=payload) return RpcResponse(payload=payload)
return handler return handler
trex_core/application/portfolio.py
+1 -2
View File
@@ -564,7 +564,7 @@ def slot_to_dict(
"forced_close_at": s.forced_close_at.isoformat() if s.forced_close_at else None, "forced_close_at": s.forced_close_at.isoformat() if s.forced_close_at else None,
"universe_detached": getattr(s, "retired_at", None) is not None, "universe_detached": getattr(s, "retired_at", None) is not None,
# dcy-028/REQ-081 (BAT-042): per-occupancy guard counter — the strategy # dcy-028/REQ-081 (BAT-042): per-occupancy guard counter — the strategy
# daemon echoes this as expected_occupancy_seq on strategy_state writes. # Slot occupancy remains domain state; strategy-state writes do not use it.
"occupancy_seq": getattr(s, "occupancy_seq", 0), "occupancy_seq": getattr(s, "occupancy_seq", 0),
} }
@@ -638,4 +638,3 @@ def _sanitize(message: str) -> str:
break break
return out[:200] return out[:200]
trex_core/application/strategy_state.py
+28 -110
View File
@@ -1,27 +1,4 @@
"""Slot strategy-state store (dcy-028 / REQ-079, BAT-042). """Slot-scoped opaque strategy-instance state."""
A strategy daemon (e.g. trex-tsl) accumulates opaque state across a slot's
occupancy — for a Trailing Stop Loss the peak price since entry. `evaluate()`
must stay pure, so the state cannot live in the daemon (lost on restart) nor in
an external store (orphan-cleanup problem). Since the state's lifecycle matches
the slot occupancy exactly, trex-core stores it server-side keyed by
`(slot_id, strategy_id)`, round-trips it on the slots snapshot (read), accepts
version-CAS writes via a dedicated RPC, and deletes it when the slot closes.
trex-core never interprets `data` (opaque UTF-8 string). It validates only
(1) valid UTF-8 and (2) a byte size cap — encoding (JSON/base64) is the
strategy's business (dcy-028 OQ-1).
Occupancy guard (dcy-028 R-1 / REQ-081): a permanent slot is reused, so two
occupancies of the same slot both start at version 0. A write computed against
occupancy X must not land on a later occupancy Y. The slot carries a monotonic
`occupancy_seq`; the CAS rejects a write whose `expected_occupancy_seq` differs
from the slot's current value (`occupancy_changed`). The version CAS alone
cannot catch this at the first write (0 vs 0).
Single-writer assumption: one daemon operates a given slot·strategy. The CAS is
for retry / duplicate / occupancy-turnover safety, not multi-writer concurrency.
"""
from __future__ import annotations from __future__ import annotations
import os import os
@@ -31,13 +8,8 @@ from datetime import datetime, timezone
from typing import Optional, Protocol from typing import Optional, Protocol
from uuid import UUID from uuid import UUID
# dcy-028 OQ-1: hard ceiling for the opaque `data` blob, measured in UTF-8
# bytes. Operators target 1~2KB for the known TSL use case; 4KB is the cap.
_STATE_CAP_BYTES = int(os.environ.get("TREX_STRATEGY_STATE_CAP_BYTES", "4096")) _STATE_CAP_BYTES = int(os.environ.get("TREX_STRATEGY_STATE_CAP_BYTES", "4096"))
# Reject reason tokens (dcy-028 / REQ-080). The full enumerated set; consumers
# fail-closed on anything unrecognized.
REASON_OCCUPANCY_CHANGED = "occupancy_changed"
REASON_VERSION_CONFLICT = "version_conflict" REASON_VERSION_CONFLICT = "version_conflict"
REASON_SLOT_INACTIVE = "slot_inactive" REASON_SLOT_INACTIVE = "slot_inactive"
REASON_STATE_TOO_LARGE = "state_too_large" REASON_STATE_TOO_LARGE = "state_too_large"
@@ -45,10 +17,8 @@ REASON_STATE_TOO_LARGE = "state_too_large"
@dataclass(frozen=True) @dataclass(frozen=True)
class StrategyStateEntry: class StrategyStateEntry:
"""One stored opaque state value for a (slot_id, strategy_id).""" """One opaque entry in a slot's strategy_instance_id map."""
strategy_id: str
occupancy_seq: int
version: int version: int
data: str data: str
updated_at: datetime updated_at: datetime
@@ -56,64 +26,20 @@ class StrategyStateEntry:
@dataclass(frozen=True) @dataclass(frozen=True)
class SetResult: class SetResult:
"""Outcome of a CAS set. ok=True carries the new version; else `reason`."""
ok: bool ok: bool
version: Optional[int] = None version: Optional[int] = None
reason: Optional[str] = None reason: Optional[str] = None
occupancy_seq: Optional[int] = None # current slot occupancy_seq on conflict
def _byte_len(data: str) -> int: def _byte_len(data: str) -> int:
return len(data.encode("utf-8")) return len(data.encode("utf-8"))
class StrategyStateStore(Protocol):
def set(
self,
slot_id: UUID,
strategy_id: str,
*,
expected_occupancy_seq: int,
expected_version: int,
data: str,
current_occupancy_seq: int,
slot_active: bool,
) -> SetResult:
"""occupancy_seq + version CAS upsert of opaque `data` (dcy-028 FR-2)."""
...
def get_for_slots(
self, slot_ids: list[UUID], current_occupancy_seq: dict[UUID, int]
) -> dict[UUID, dict[str, StrategyStateEntry]]:
"""Batched read for the snapshot (C-1). Returns slot_id -> {strategy_id: entry}.
Only entries whose stored occupancy_seq matches the slot's current
occupancy_seq are returned (stale entries from a prior occupancy are
invisible — defense in depth alongside close-delete, dcy-028 FR-3)."""
...
def delete_all(self, slot_id: UUID) -> None:
"""Delete all strategy_state for a slot (close hook, dcy-028 C-3)."""
...
def _cas( def _cas(
existing: Optional[StrategyStateEntry], existing: Optional[StrategyStateEntry], *, expected_version: int, data: str, slot_active: bool
*,
expected_occupancy_seq: int,
expected_version: int,
data: str,
current_occupancy_seq: int,
slot_active: bool,
) -> SetResult: ) -> SetResult:
"""Pure CAS decision shared by all store implementations (dcy-028 FR-2)."""
if not slot_active: if not slot_active:
return SetResult(ok=False, reason=REASON_SLOT_INACTIVE) return SetResult(ok=False, reason=REASON_SLOT_INACTIVE)
if expected_occupancy_seq != current_occupancy_seq:
return SetResult(
ok=False, reason=REASON_OCCUPANCY_CHANGED, occupancy_seq=current_occupancy_seq
)
if _byte_len(data) > _STATE_CAP_BYTES: if _byte_len(data) > _STATE_CAP_BYTES:
return SetResult(ok=False, reason=REASON_STATE_TOO_LARGE) return SetResult(ok=False, reason=REASON_STATE_TOO_LARGE)
stored_version = existing.version if existing is not None else 0 stored_version = existing.version if existing is not None else 0
@@ -122,62 +48,54 @@ def _cas(
return SetResult(ok=True, version=stored_version + 1) return SetResult(ok=True, version=stored_version + 1)
class StrategyStateStore(Protocol):
def set(
self,
slot_id: UUID,
strategy_instance_id: str,
*,
expected_version: int,
data: str,
slot_active: bool,
) -> SetResult: ...
def get_for_slots(self, slot_ids: list[UUID]) -> dict[UUID, dict[str, StrategyStateEntry]]: ...
class InMemoryStrategyStateStore: class InMemoryStrategyStateStore:
"""Process-local strategy-state store (paper / tests / explicit double).""" """Process-local slot-scoped state-map store."""
def __init__(self) -> None: def __init__(self) -> None:
# (slot_id, strategy_id) -> StrategyStateEntry self._store: dict[UUID, dict[str, StrategyStateEntry]] = {}
self._store: dict[tuple[UUID, str], StrategyStateEntry] = {}
self._lock = threading.Lock() self._lock = threading.Lock()
def set( def set(
self, self,
slot_id: UUID, slot_id: UUID,
strategy_id: str, strategy_instance_id: str,
*, *,
expected_occupancy_seq: int,
expected_version: int, expected_version: int,
data: str, data: str,
current_occupancy_seq: int,
slot_active: bool, slot_active: bool,
) -> SetResult: ) -> SetResult:
with self._lock: with self._lock:
existing = self._store.get((slot_id, strategy_id)) entries = self._store.setdefault(slot_id, {})
existing = entries.get(strategy_instance_id)
result = _cas( result = _cas(
existing, existing, expected_version=expected_version, data=data, slot_active=slot_active
expected_occupancy_seq=expected_occupancy_seq,
expected_version=expected_version,
data=data,
current_occupancy_seq=current_occupancy_seq,
slot_active=slot_active,
) )
if not result.ok: if not result.ok:
if not entries:
self._store.pop(slot_id, None)
return result return result
self._store[(slot_id, strategy_id)] = StrategyStateEntry( entries[strategy_instance_id] = StrategyStateEntry(
strategy_id=strategy_id,
occupancy_seq=current_occupancy_seq,
version=result.version, # type: ignore[arg-type] version=result.version, # type: ignore[arg-type]
data=data, data=data,
updated_at=datetime.now(tz=timezone.utc), updated_at=datetime.now(tz=timezone.utc),
) )
return result return result
def get_for_slots( def get_for_slots(self, slot_ids: list[UUID]) -> dict[UUID, dict[str, StrategyStateEntry]]:
self, slot_ids: list[UUID], current_occupancy_seq: dict[UUID, int]
) -> dict[UUID, dict[str, StrategyStateEntry]]:
wanted = set(slot_ids) wanted = set(slot_ids)
out: dict[UUID, dict[str, StrategyStateEntry]] = {}
with self._lock: with self._lock:
for (sid, strat), entry in self._store.items(): return {sid: dict(entries) for sid, entries in self._store.items() if sid in wanted}
if sid not in wanted:
continue
# stale-occupancy entries are invisible (dcy-028 FR-3 defense)
if entry.occupancy_seq != current_occupancy_seq.get(sid):
continue
out.setdefault(sid, {})[strat] = entry
return out
def delete_all(self, slot_id: UUID) -> None:
with self._lock:
for key in [k for k in self._store if k[0] == slot_id]:
del self._store[key]
trex_core/domain/slot/service.py
-7
View File
@@ -52,8 +52,6 @@ class SlotService:
# does not serialize transitions across instances (NFR-1). # does not serialize transitions across instances (NFR-1).
self._instance_key = instance_key self._instance_key = instance_key
self._lock_id = slot_advisory_lock_id(instance_key) self._lock_id = slot_advisory_lock_id(instance_key)
# dcy-028/REQ-081 (C-3): when injected, all strategy_state for a slot is
# deleted on close (→INACTIVE). None keeps legacy behaviour (no store).
self._strategy_state_store = strategy_state_store self._strategy_state_store = strategy_state_store
@property @property
@@ -155,11 +153,6 @@ class SlotService:
extra=extra or {}, extra=extra or {},
) )
) )
# dcy-028/REQ-081 (C-3): close deletes the slot's strategy_state so a
# reused slot starts empty. Covers every close path (strategy sell,
# safety-sell, forced-close, operator) since all funnel to →INACTIVE.
if to_state is SlotState.INACTIVE and self._strategy_state_store is not None:
self._strategy_state_store.delete_all(slot_id)
return updated return updated
@staticmethod @staticmethod