genichinandClaude Sonnet 4.6
6c24bd4c2c
slots RPC 응답에 available_slots: int|null 포함. max_active_slots 미설정 시 null, 설정 시 max(0, M-active). 16 tests passed (AC-6 신규 3개 포함), 423 전체 통과. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
477 lines
18 KiB
Python
477 lines
18 KiB
Python
"""RPC handlers + event payload builders for messaging transport (REQ-021/022).
|
|
|
|
Handlers are transport-agnostic: they accept a parsed `RpcRequest` payload
|
|
and return an `RpcResponse`. Wiring them onto `MessagingTransport.consume_*`
|
|
happens in the runtime layer.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date as date_cls
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
from uuid import UUID
|
|
|
|
from trex_core.application.intents import BuyIntent, IntentService, SellIntent
|
|
from trex_core.application.messaging import RpcRequest, RpcResponse
|
|
from trex_core.application.portfolio import (
|
|
PauseStore,
|
|
SnapshotStore,
|
|
slot_to_dict,
|
|
)
|
|
from trex_core.application.watchdog import WatchdogStore
|
|
from trex_core.domain.slot import (
|
|
SlotNotFoundError,
|
|
SlotRepository,
|
|
SlotService,
|
|
)
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
|
|
# ---- intent RPC handler -----------------------------------------------------
|
|
|
|
def make_intent_rpc_handler(
|
|
intent_service: IntentService,
|
|
*,
|
|
slot_service: Optional[SlotService] = None,
|
|
watchdog_store: Optional[WatchdogStore] = None,
|
|
):
|
|
"""Build the RPC handler for `trex.intent.req`.
|
|
|
|
Payload shape:
|
|
{
|
|
"intent_type": "buy" | "sell",
|
|
"intent_id": "<idempotency key, = correlation_id>",
|
|
... fields per BuyIntent / SellIntent ...
|
|
}
|
|
or (REQ-040):
|
|
{"action": "update_thresholds", "slot_id": ...,
|
|
"stop_loss_price"?, "take_profit_price"?, "forced_close_at"?}
|
|
|
|
Response shape:
|
|
{"intent_id": str, "status": "admitted"|"rejected", "slot_id": str|None, "reason": str|None}
|
|
or for update_thresholds:
|
|
{"status": "updated"|"ignored"|"error", "slot_id": str, "reason"?: str}
|
|
|
|
Dedup uses `intent_log.try_insert(intent_id, ...)` (REQ-007 gate). The
|
|
request's `correlation_id` is reused as `intent_id` so the messaging
|
|
layer and the application share a single idempotency source of truth
|
|
(REQ-023 AC-1).
|
|
"""
|
|
|
|
def handler(req: RpcRequest) -> RpcResponse:
|
|
body = req.payload or {}
|
|
# REQ-040: update_thresholds action (separate from buy/sell intents).
|
|
if str(body.get("action") or "").lower() == "update_thresholds":
|
|
return _handle_update_thresholds(body, slot_service, watchdog_store)
|
|
intent_id = str(body.get("intent_id") or req.correlation_id)
|
|
intent_type = str(body.get("intent_type") or "").lower()
|
|
try:
|
|
if intent_type == "buy":
|
|
# dcy-020 / REQ-061: buy is keyed by slot_id (canonical);
|
|
# portfolio_id is the legacy dual-accept mapping. symbol is NOT
|
|
# a client field — resolved from the universe. trading_date was
|
|
# dropped from the contract at the cutover (BAT-030).
|
|
buy = BuyIntent(
|
|
intent_id=intent_id,
|
|
portfolio_id=_opt_str(body.get("portfolio_id")) or None,
|
|
slot_id=(UUID(str(body["slot_id"])) if body.get("slot_id") else None),
|
|
qty=_opt_float(body.get("qty")),
|
|
# dcy-023 / REQ-007 AC-6: allocated_capital is no longer a
|
|
# contract field — senders may still include it; it is ignored.
|
|
entry_price=_opt_float(body.get("entry_price")),
|
|
strategy_id=_opt_str(body.get("strategy_id")),
|
|
sell_strategy_id=_opt_str(body.get("sell_strategy_id")),
|
|
forced_close_at=_opt_dt(body.get("forced_close_at")),
|
|
stop_loss_price=_opt_decimal(body.get("stop_loss_price")),
|
|
take_profit_price=_opt_decimal(body.get("take_profit_price")),
|
|
# dcy-022 / REQ-068: execution policy (optional, backward-compat).
|
|
timeout_sec=_opt_int(body.get("timeout_sec")),
|
|
slippage_pct=_opt_float(body.get("slippage_pct")),
|
|
)
|
|
d = intent_service.submit_buy(buy)
|
|
elif intent_type == "sell":
|
|
sell = SellIntent(
|
|
intent_id=intent_id,
|
|
slot_id=UUID(str(body["slot_id"])),
|
|
reason=str(body.get("reason") or "discretionary_sell"),
|
|
sell_strategy_id=_opt_str(body.get("sell_strategy_id")),
|
|
# dcy-012 / REQ-007: optional sell limit price; absent => market sell.
|
|
limit_price=_opt_float(body.get("limit_price")),
|
|
# dcy-021 / REQ-062: optional partial-sell qty; absent => sell all.
|
|
qty=_opt_float(body.get("qty")),
|
|
# dcy-022 / REQ-068: execution policy (optional, backward-compat).
|
|
timeout_sec=_opt_int(body.get("timeout_sec")),
|
|
slippage_pct=_opt_float(body.get("slippage_pct")),
|
|
)
|
|
d = intent_service.submit_sell(sell)
|
|
else:
|
|
return RpcResponse(payload={
|
|
"intent_id": intent_id, "status": "rejected",
|
|
"slot_id": None, "reason": f"unknown_intent_type:{intent_type}",
|
|
})
|
|
except (KeyError, ValueError) as exc:
|
|
return RpcResponse(payload={
|
|
"intent_id": intent_id, "status": "rejected",
|
|
"slot_id": None, "reason": f"bad_request:{exc}",
|
|
})
|
|
return RpcResponse(payload={
|
|
"intent_id": d.intent_id,
|
|
"status": d.status,
|
|
"slot_id": (str(d.slot_id) if d.slot_id else None),
|
|
"reason": d.reason,
|
|
})
|
|
|
|
return handler
|
|
|
|
|
|
def _handle_update_thresholds(
|
|
body: dict,
|
|
slot_service: Optional[SlotService],
|
|
watchdog_store: Optional[WatchdogStore],
|
|
) -> RpcResponse:
|
|
"""REQ-040: update stop_loss/take_profit/forced_close_at on a holding slot.
|
|
|
|
Only the keys actually present in the payload are forwarded, so an absent
|
|
field keeps its value while an explicit ``null`` clears it (AC-4 vs AC-5).
|
|
On success the slot's watchdog last_seen_at is reset (AC-3) — receiving an
|
|
update_thresholds is itself a strategy-module liveness signal.
|
|
"""
|
|
if slot_service is None:
|
|
return RpcResponse(payload={"status": "error", "reason": "update_thresholds_unsupported"})
|
|
try:
|
|
slot_id = UUID(str(body["slot_id"]))
|
|
except (KeyError, ValueError) as exc:
|
|
return RpcResponse(payload={"status": "error", "reason": f"bad_request:{exc}"})
|
|
kwargs: dict = {}
|
|
if "stop_loss_price" in body:
|
|
kwargs["stop_loss_price"] = _opt_decimal(body["stop_loss_price"])
|
|
if "take_profit_price" in body:
|
|
kwargs["take_profit_price"] = _opt_decimal(body["take_profit_price"])
|
|
if "forced_close_at" in body:
|
|
kwargs["forced_close_at"] = _opt_dt(body["forced_close_at"])
|
|
actor = str(body.get("actor") or "strategy")
|
|
try:
|
|
status = slot_service.update_thresholds(slot_id, actor=actor, **kwargs)
|
|
except SlotNotFoundError:
|
|
return RpcResponse(payload={
|
|
"status": "ignored", "slot_id": str(slot_id), "reason": "slot_not_found",
|
|
})
|
|
if status == "updated" and watchdog_store is not None:
|
|
watchdog_store.update(str(slot_id)) # AC-3: liveness signal
|
|
payload = {"status": status, "slot_id": str(slot_id)}
|
|
if status == "ignored":
|
|
payload["reason"] = "not_holding" # AC-6
|
|
return RpcResponse(payload=payload)
|
|
|
|
|
|
# ---- slots RPC handler (REQ-073/074) ----------------------------------------
|
|
|
|
def make_slots_rpc_handler(
|
|
slot_repo: SlotRepository,
|
|
*,
|
|
instance_key: str = "default",
|
|
snapshot_store: Optional[SnapshotStore] = None,
|
|
pause_store: Optional[PauseStore] = None,
|
|
watchdog_store: Optional[WatchdogStore] = None,
|
|
watchdog_timeout_sec: int = 120,
|
|
ledger_outbox=None,
|
|
max_active_slots: Optional[int] = None,
|
|
):
|
|
"""Build the RPC handler for `trex.slots.req` (REQ-073/074).
|
|
|
|
Request: {filter?: str, spy?: bool}
|
|
filter: "all"|"inactive"|"pending"|"holding"|"error" (default "all")
|
|
spy=true: watchdog not updated (operator view, REQ-034 CHG-01)
|
|
|
|
Response: {slots: [...], available_slots: int|null, degraded: bool,
|
|
reason: str|null, message: str|null}
|
|
available_slots: capacity hint — null when uncapped, max(0, M-active) when capped
|
|
(REQ-073 AC-6). Does not replace the admission gate.
|
|
"""
|
|
|
|
def handler(req: RpcRequest) -> RpcResponse:
|
|
body = req.payload or {}
|
|
filter_val = str(body.get("filter") or "all").lower()
|
|
spy = bool(body.get("spy", False))
|
|
|
|
# Retrieve and filter slots by status
|
|
slots = slot_repo.list_slots(instance_key)
|
|
if filter_val == "inactive":
|
|
slots = [s for s in slots if s.status.value == "inactive"]
|
|
elif filter_val == "pending":
|
|
slots = [s for s in slots if s.status.value in ("pending_buy", "pending_sell")]
|
|
elif filter_val == "holding":
|
|
slots = [s for s in slots if s.status.value == "holding"]
|
|
elif filter_val == "error":
|
|
slots = [s for s in slots if s.status.value == "error"]
|
|
# "all" or unrecognized → no filter (REQ-073 AC-2)
|
|
|
|
# REQ-034 CHG-01: update watchdog for holding slots unless spy=True
|
|
if not spy and watchdog_store is not None:
|
|
for s in slot_repo.list_holding_slots(instance_key):
|
|
watchdog_store.update(str(s.slot_id))
|
|
|
|
# Build merged slot rows (REQ-074)
|
|
slot_rows = [
|
|
slot_to_dict(
|
|
s,
|
|
snapshot_store=snapshot_store,
|
|
watchdog_store=watchdog_store,
|
|
pause_store=pause_store,
|
|
watchdog_timeout_sec=watchdog_timeout_sec,
|
|
)
|
|
for s in slots
|
|
]
|
|
|
|
# dcy-018 / REQ-029: cash-estimated overlay (best-effort)
|
|
if ledger_outbox is not None:
|
|
try:
|
|
deltas = ledger_outbox.pending_cash_delta()
|
|
if deltas:
|
|
for row in slot_rows:
|
|
pid = row.get("portfolio_id")
|
|
d = deltas.get(pid)
|
|
if d:
|
|
base = row.get("cash_balance") or 0.0
|
|
row["cash_balance"] = base + d
|
|
row["cash_estimated"] = True
|
|
except Exception: # noqa: BLE001 - estimate must never break the response
|
|
pass
|
|
|
|
# Degraded state from snapshot store
|
|
degraded = False
|
|
reason = None
|
|
message = None
|
|
if snapshot_store is not None:
|
|
reason_code, msg = snapshot_store.degraded_state()
|
|
degraded = reason_code is not None
|
|
reason = reason_code.value if reason_code is not None else None
|
|
message = msg
|
|
|
|
if max_active_slots is not None and max_active_slots > 0:
|
|
available_slots = max(0, max_active_slots - slot_repo.count_active(instance_key))
|
|
else:
|
|
available_slots = None
|
|
|
|
return RpcResponse(payload={
|
|
"slots": slot_rows,
|
|
"available_slots": available_slots,
|
|
"degraded": degraded,
|
|
"reason": reason,
|
|
"message": message,
|
|
})
|
|
|
|
return handler
|
|
|
|
|
|
# ---- pause/resume RPC handler (REQ-032 AC-4) --------------------------------
|
|
|
|
def make_pause_rpc_handler(pause_store: PauseStore):
|
|
"""RPC handler for the `trex.pause.req` control queue.
|
|
|
|
Payload: {"action": "pause" | "resume" | "list", "portfolio_id": "..."}
|
|
Response: {"status": "ok"|"error", "paused": [list-of-portfolio_id], "reason"?: str}
|
|
"""
|
|
|
|
def handler(req: RpcRequest) -> RpcResponse:
|
|
body = req.payload or {}
|
|
action = str(body.get("action") or "").lower()
|
|
pid = _opt_str(body.get("portfolio_id"))
|
|
actor = str(body.get("actor") or "operator")
|
|
try:
|
|
if action == "pause":
|
|
if not pid:
|
|
return _err("portfolio_id required")
|
|
pause_store.pause(pid, actor=actor)
|
|
elif action == "resume":
|
|
if not pid:
|
|
return _err("portfolio_id required")
|
|
pause_store.resume(pid)
|
|
elif action == "list":
|
|
pass
|
|
else:
|
|
return _err(f"unknown action: {action!r}")
|
|
except Exception as exc: # noqa: BLE001
|
|
return _err(f"pause store error: {exc.__class__.__name__}")
|
|
return RpcResponse(payload={
|
|
"status": "ok",
|
|
"paused": sorted(pause_store.list_paused()),
|
|
"action": action,
|
|
})
|
|
|
|
return handler
|
|
|
|
|
|
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,
|
|
cancel_pending_fn=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), `cancel_pending` (dcy-022/REQ-071).
|
|
Unknown actions return an error response.
|
|
|
|
cancel_pending_fn: callable(slot_id: str) -> str — injected from ExecutionService.
|
|
"""
|
|
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)
|
|
if action == "cancel_pending":
|
|
return _handle_cancel_pending(body, cancel_pending_fn)
|
|
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_cancel_pending(body: dict, cancel_fn) -> RpcResponse:
|
|
"""dcy-022 / REQ-071: operator backstop cancel for a pending slot."""
|
|
if cancel_fn is None:
|
|
return _err("cancel_pending_unsupported")
|
|
try:
|
|
slot_id = str(body["slot_id"])
|
|
except KeyError:
|
|
return _err("slot_id_required")
|
|
result = cancel_fn(slot_id)
|
|
if result == "ok":
|
|
return RpcResponse(payload={"status": "ok", "slot_id": slot_id})
|
|
if result == "not_pending":
|
|
return RpcResponse(payload={"status": "ignored", "slot_id": slot_id, "reason": "not_pending"})
|
|
if result == "slot_not_found":
|
|
return RpcResponse(payload={"status": "error", "slot_id": slot_id, "reason": "slot_not_found"})
|
|
return RpcResponse(payload={"status": "error", "slot_id": slot_id, "reason": result})
|
|
|
|
|
|
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:
|
|
"""Build the event payload for a slot transition (REQ-023 AC-5).
|
|
|
|
Carries the *current state* + a monotonic sequence (`updated_at` iso) so
|
|
receivers can apply last-wins without ordering guarantees.
|
|
"""
|
|
return {
|
|
"event_type": event_type,
|
|
"slot_id": str(slot.slot_id),
|
|
"strategy_id": slot.strategy_id,
|
|
"symbol": slot.symbol,
|
|
"status": slot.status.value,
|
|
"qty": float(slot.qty) if slot.qty is not None else None,
|
|
"avg_price": float(slot.avg_price) if slot.avg_price is not None else None,
|
|
"portfolio_id": slot.portfolio_id,
|
|
"sequence_at": (slot.updated_at or datetime.now(tz=timezone.utc)).isoformat(),
|
|
}
|
|
|
|
|
|
# ---- helpers ----------------------------------------------------------------
|
|
|
|
def _opt_str(v) -> Optional[str]:
|
|
if v is None or v == "":
|
|
return None
|
|
return str(v)
|
|
|
|
|
|
def _opt_decimal(v):
|
|
if v is None:
|
|
return None
|
|
from decimal import Decimal
|
|
return Decimal(str(v))
|
|
|
|
|
|
def _opt_float(v) -> Optional[float]:
|
|
if v is None:
|
|
return None
|
|
return float(v)
|
|
|
|
|
|
def _opt_int(v) -> Optional[int]:
|
|
if v is None:
|
|
return None
|
|
return int(v)
|
|
|
|
|
|
def _opt_dt(v) -> Optional[datetime]:
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, datetime):
|
|
return v
|
|
return datetime.fromisoformat(str(v))
|
|
|
|
|
|
def _parse_date(v) -> date_cls:
|
|
if isinstance(v, date_cls):
|
|
return v
|
|
return date_cls.fromisoformat(str(v))
|
|
|
|
|
|
def _today() -> date_cls:
|
|
return datetime.now(tz=timezone.utc).date()
|
|
|
|
|