trex-core/trex_core/application/messaging_handlers.py

"""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,
):
    """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: [...], degraded: bool, reason: str|null, message: str|null}
    Each slot row carries slot/core state + hub asset data + watchdog + overlay
    flags (strategy fields excluded — REQ-074 AC-5).
    """

    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

        return RpcResponse(payload={
            "slots": slot_rows,
            "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()