genichinandClaude Opus 4.8
f0ad37747b
BAT-035 (dcy-025): trex.snapshot.req → trex.slots.req 즉시 교체
- topology: SNAPSHOT_REQ_QUEUE 제거, SLOTS_REQ_QUEUE = "trex.slots.req" 추가
- portfolio.py: View/build_snapshot_envelope/SnapshotEnvelope/_asset_to_dict 제거;
slot_to_dict 시그니처 확장(snapshot_store/watchdog_store/pause_store/watchdog_timeout_sec)
— asset_quantity(hub)/cash_balance/cash_estimated/warned/watchdog{}/paused 병합 (REQ-074)
- messaging_handlers.py: make_snapshot_rpc_handler 제거, make_slots_rpc_handler 추가
— filter(all/inactive/pending/holding/error) + spy(bool) 처리; strategy 필드 제외 (REQ-073/074)
- daemon.py: _snapshot_rpc_loop → _slots_rpc_loop, make_slots_rpc_handler 연결
- cli.py: universe/slots → slots_queue + spy=True + server-side filter; buyable→inactive 매핑 (REQ-063)
- rabbitmq.py: snapshot_queue → slots_queue
- docs: interface-contract.md + runtime-flows.md slots.req 계약으로 업데이트 (REQ-075)
- REQ-028: Implemented → Deprecated (코드 삭제 완료)
- 테스트: test_snapshot_handler/selection/rpc_watchdog/messaging_snapshot_consumer 삭제;
test_slots_handler.py(신규, 13 cases), test_killswitch_masking_resume 재작성,
test_ledger_option_d._apply_cash_estimate → slots handler 경유, test_topology/cli_operator 갱신
BAT-036 (dcy-024): per-instance safety_sell_enabled 플래그
- config.py: DaemonInstanceConfig + DaemonConfig 에 safety_sell_enabled: bool = True 추가;
from_toml 파싱, from_config 전파
- daemon.py: start() workers 조건부 — safety_sell_enabled=false 시 trex-safety-sell 미기동 + WARNING
- tests: test_instance_config.py 에 safety_sell_enabled 파싱 3개 케이스 추가
REQ: 073/074/075/021/034/063/048/005 Approved → Implemented
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
88 lines
3.7 KiB
Python
88 lines
3.7 KiB
Python
"""Messaging topology constants + builders (REQ-021, dcy-006 OQ-2).
|
|
|
|
Three patterns:
|
|
1. slots RPC — direct queue `trex.slots.req`, reply via amq.rabbitmq.reply-to
|
|
2. intent RPC — direct queue `trex.intent.req`, reply via amq.rabbitmq.reply-to
|
|
3. events — topic exchange `trex.events`, routing key {event_type}.{strategy_id}.{symbol}
|
|
|
|
Mapping principle: requests carry filter/spy in *payload*; events carry
|
|
strategy_id/symbol in the *routing key* so brokers can filter per-strategy
|
|
queues with patterns like `*.{strategy_id}.#`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
# --- RPC queues (durable for intent; transient-ish for slots but kept
|
|
# durable for simplicity since the queue itself is shared) ---
|
|
SLOTS_REQ_QUEUE = "trex.slots.req" # REQ-073 / dcy-025: replaces snapshot.req
|
|
INTENT_REQ_QUEUE = "trex.intent.req"
|
|
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)
|
|
|
|
# --- multi-instance topology (BAT-013 / dcy-009) ---
|
|
DEFAULT_INSTANCE_KEY = "default"
|
|
DIRECT_EXCHANGE = "trex.direct" # topic, durable — directed per-instance routing (REQ-050)
|
|
BROADCAST_EXCHANGE = "trex.broadcast" # fanout, durable — global control to all instances (REQ-050)
|
|
|
|
|
|
def _instance_queue(base: str, instance_key: str | None) -> str:
|
|
"""`{base}.{instance_key}` for named instances; literal `base` for default.
|
|
|
|
The default special-case keeps existing queue names (and the entire test
|
|
suite + deployed clients) byte-for-byte (REQ-049 FR-5 backward-compat).
|
|
"""
|
|
if instance_key in (None, DEFAULT_INSTANCE_KEY):
|
|
return base
|
|
return f"{base}.{instance_key}"
|
|
|
|
|
|
def intent_queue(instance_key: str | None = DEFAULT_INSTANCE_KEY) -> str:
|
|
return _instance_queue(INTENT_REQ_QUEUE, instance_key)
|
|
|
|
|
|
def slots_queue(instance_key: str | None = DEFAULT_INSTANCE_KEY) -> str:
|
|
return _instance_queue(SLOTS_REQ_QUEUE, instance_key)
|
|
|
|
|
|
def control_queue(instance_key: str | None = DEFAULT_INSTANCE_KEY) -> str:
|
|
return _instance_queue(CONTROL_REQ_QUEUE, instance_key)
|
|
|
|
|
|
def direct_routing_key(instance_key: str, channel_type: str) -> str:
|
|
"""Directed routing key for `trex.direct` (REQ-050): `{instance_key}.{channel_type}`."""
|
|
return f"{instance_key}.{channel_type}"
|
|
|
|
# --- Events ---
|
|
EVENTS_EXCHANGE = "trex.events" # topic, durable
|
|
EVENTS_DLQ_EXCHANGE = "trex.events.dlq" # fanout, durable; receives unroutable/NACKed
|
|
EVENTS_DLQ_QUEUE = "trex.events.dlq" # one collector queue bound to DLQ exchange
|
|
|
|
|
|
def event_routing_key(event_type: str, strategy_id: str, symbol: str) -> str:
|
|
"""Build the routing key for an event (dcy-006 OQ-2 mapping)."""
|
|
# Sanitize tokens — AMQP routing keys cannot contain '.', '*', '#'.
|
|
safe = lambda s: str(s or "_").replace(".", "_").replace("*", "_").replace("#", "_") # noqa: E731
|
|
return f"{safe(event_type)}.{safe(strategy_id)}.{safe(symbol)}"
|
|
|
|
|
|
def strategy_event_queue_name(strategy_id: str) -> str:
|
|
"""Per-strategy durable queue name (catch-up + last-wins)."""
|
|
return f"trex.events.strategy.{strategy_id}"
|
|
|
|
|
|
def strategy_event_binding_pattern(strategy_id: str) -> str:
|
|
"""Bind pattern for a strategy queue — all event types, all symbols.
|
|
|
|
Wildcards: `*` matches one word, `#` matches zero-or-more words.
|
|
"""
|
|
return f"*.{strategy_id}.#"
|
|
|
|
|
|
# --- per-queue arguments (REQ-023 TTL / max-length / drop-oldest) ---
|
|
STRATEGY_QUEUE_ARGS: dict = {
|
|
"x-message-ttl": 3_600_000, # 1h
|
|
"x-max-length": 10_000,
|
|
"x-overflow": "drop-head", # drop-oldest, then accept new (REQ-023 AC-7)
|
|
"x-dead-letter-exchange": EVENTS_DLQ_EXCHANGE,
|
|
}
|