trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
30bcbe1d12cc93a694236be53372eea8a6e9e428
trex-core/trex_core/application/messaging_handlers.py
T

637 lines
25 KiB
Python
Raw Blame History

"""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, Protocol
from uuid import UUID
from trex_core.application.intents import BuyIntent, IntentService, SellIntent
from trex_core.application.account_snapshot import AccountSnapshotUnavailable
from trex_core.application.broker_port import BalanceItem
from trex_core.application.messaging import RpcRequest, RpcResponse
from trex_core.application.portfolio import (
PauseStore,
ReconcileWarningStore,
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")),
venue=_opt_str(body.get("venue")) or "SOR",
)
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")),
venue=_opt_str(body.get("venue")) or "SOR",
)
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:
payload = {
"intent_id": intent_id, "status": "rejected",
"slot_id": None, "reason": f"bad_request:{exc}",
}
return RpcResponse(payload=payload)
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 RpcResponse(payload=payload)
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) ----------------------------------------
# Maps a `trex.slots.req` filter token to the set of slot status values it
# selects (issue #25 — pushed to the repository via list_slots status_filter).
# "all" and unrecognized tokens map to None (no filter — REQ-073 AC-2).
_FILTER_STATUS: dict[str, set[str]] = {
"inactive": {"inactive"},
"pending": {"pending_buy", "pending_sell"},
"holding": {"holding"},
"error": {"error"},
}
class SlotsBrokerQuantityReader(Protocol):
def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: ...
def make_slots_rpc_handler(
slot_repo: SlotRepository,
*,
instance_key: str = "default",
snapshot_store: Optional[SnapshotStore] = None,
pause_store: Optional[PauseStore] = None,
warn_store: Optional[ReconcileWarningStore] = None,
watchdog_store: Optional[WatchdogStore] = None,
watchdog_timeout_sec: int = 120,
ledger_outbox=None,
max_active_slots: Optional[int] = None,
strategy_state_store=None,
broker_quantity_reader: Optional[SlotsBrokerQuantityReader] = None,
):
"""Build the RPC handler for `trex.slots.req` (REQ-073/074/116).
Request: {filter?: str, spy?: bool, strategy?: str}
filter: "all"|"inactive"|"pending"|"holding"|"error" (default "all")
spy=true: watchdog not updated (operator view, REQ-034 CHG-01)
strategy: exact-match on strategy_name (REQ-116, optional). When specified
only slots whose strategy_name == strategy are returned. AND-combined with
filter. available_slots is always computed before this filter (REQ-073 AC-6).
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))
# REQ-116 (BAT-055): optional strategy filter — exact-match on strategy_name.
# None means no filter (all slots returned regardless of strategy_name).
strategy_filter = _opt_str(body.get("strategy"))
# issue #25: filter by status at the DB (REQ-002/006 list_slots
# status_filter) instead of fetching every row and filtering in Python.
# "all"/unrecognized → None (no filter, REQ-073 AC-2).
status_filter = _FILTER_STATUS.get(filter_val)
slots = slot_repo.list_slots(instance_key, status_filter=status_filter)
# REQ-034 CHG-01: update watchdog for holding slots unless spy=True.
# When the fetched set already contains every holding slot (filter "all"
# or "holding"), reuse it and skip the second connection (issue #25 (2)).
if not spy and watchdog_store is not None:
if status_filter is None or filter_val == "holding":
holding_slots = [s for s in slots if s.status.value == "holding"]
else:
holding_slots = slot_repo.list_holding_slots(instance_key)
for s in holding_slots:
watchdog_store.update(str(s.slot_id))
# Build merged slot rows (REQ-074)
# BAT-050 / REQ-104: warn_store passed to slot_to_dict for reconcile_warned field.
slot_rows = [
slot_to_dict(
s,
snapshot_store=snapshot_store,
watchdog_store=watchdog_store,
pause_store=pause_store,
warn_store=warn_store,
watchdog_timeout_sec=watchdog_timeout_sec,
)
for s in slots
]
broker_qty_by_symbol: dict[str, int] = {}
broker_qty_available = False
broker_qty_captured_at: Optional[datetime] = None
if broker_quantity_reader is not None:
try:
holdings, broker_qty_captured_at = broker_quantity_reader.holdings_for_slots_rpc()
broker_qty_by_symbol = {
str(item.symbol): int(item.hldg_qty)
for item in holdings
}
broker_qty_available = True
except AccountSnapshotUnavailable:
broker_qty_available = False
for row, slot in zip(slot_rows, slots):
row["slot_quantity"] = row.get("qty")
if slot.status.value != "holding":
continue
row["broker_quantity_available"] = broker_qty_available
row["broker_quantity_captured_at"] = (
broker_qty_captured_at.isoformat() if broker_qty_captured_at is not None else None
)
if broker_qty_available:
broker_qty = int(broker_qty_by_symbol.get(str(slot.symbol), 0))
row["qty"] = float(broker_qty)
row["qty_source"] = "broker"
else:
row["qty"] = None
row["qty_source"] = "unavailable"
# dcy-028 / REQ-073/074·079 (C-1): attach opaque strategy_state map per
# row via one batched store read (read piggyback — no extra round trip).
# Only entries matching the slot's current occupancy_seq are surfaced.
if strategy_state_store is not None and 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, occ)
for row, s in zip(slot_rows, slots):
entries = state_map.get(s.slot_id)
if entries:
row["strategy_state"] = {
strat: {
"version": e.version,
"data": e.data,
"updated_at": e.updated_at.isoformat(),
}
for strat, e in entries.items()
}
# 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
# REQ-073 AC-6: available_slots computed from instance-wide active count —
# independent of strategy filter (REQ-116). Calculate before filtering.
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
# REQ-116 (BAT-055): strategy filter — Python-level post-filter after
# available_slots is already computed. AND-combined with status filter.
# Unmatched strategy (no slots) returns [] without error.
if strategy_filter is not None:
slot_rows = [r for r in slot_rows if r.get("strategy_name") == strategy_filter]
return RpcResponse(payload={
"slots": slot_rows,
"available_slots": available_slots,
"degraded": degraded,
"reason": reason,
"message": message,
})
return handler
# ---- strategy-state write RPC handler (dcy-028 / REQ-080) -------------------
def make_strategy_state_set_handler(
slot_repo: SlotRepository,
strategy_state_store,
):
"""Build the RPC handler for `trex.slots.strategy_state.set` (REQ-080).
Request: {slot_id, strategy_id, expected_version:int,
expected_occupancy_seq:int, data:str}
Response (ok): {ok:true, slot_id, strategy_id, version:int}
Response (fail): {ok:false, slot_id, strategy_id, reason, version?|occupancy_seq?}
reason ∈ {occupancy_changed, version_conflict, slot_inactive,
state_too_large, bad_request:<detail>}. The store owns the CAS
(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"}
def handler(req: RpcRequest) -> RpcResponse:
body = req.payload or {}
try:
slot_id = UUID(str(body["slot_id"]))
strategy_id = str(body["strategy_id"])
expected_version = int(body["expected_version"])
expected_occupancy_seq = int(body["expected_occupancy_seq"])
data = body["data"]
if not isinstance(data, str):
raise ValueError("data must be a string")
except (KeyError, ValueError, TypeError) as exc:
return RpcResponse(payload={
"ok": False,
"slot_id": body.get("slot_id"),
"strategy_id": body.get("strategy_id"),
"reason": f"bad_request:{exc}",
})
slot = slot_repo.fetch_slot(slot_id)
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(
slot_id,
strategy_id,
expected_occupancy_seq=expected_occupancy_seq,
expected_version=expected_version,
data=data,
current_occupancy_seq=current_occupancy_seq,
slot_active=slot_active,
)
payload = {
"ok": result.ok,
"slot_id": str(slot_id),
"strategy_id": strategy_id,
}
if result.ok:
payload["version"] = result.version
else:
payload["reason"] = result.reason
if result.version is not None:
payload["version"] = result.version
if result.occupancy_seq is not None:
payload["occupancy_seq"] = result.occupancy_seq
return RpcResponse(payload=payload)
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
if not isinstance(v, int) or isinstance(v, bool):
raise ValueError("integer required")
return 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()
Reference in New Issue View Git Blame Copy Permalink