trex-core/trex_core/runtime/daemon.py

"""Slot-core daemon entrypoint (REQ-005, REQ-012/013/014, BAT-007).

Sync + threaded runtime. HTTP `/v1` API has been removed (REQ-025); all
strategy-facing communication runs over RabbitMQ (REQ-021/022/023).

Threads (started in `run()`):
  - intent RPC consumer    — `trex.intent.req`,  prefetch=1, FIFO, dedup via intent_log
  - slots RPC consumer     — `trex.slots.req` (also refreshes the watchdog; BAT-035/REQ-073)
  - control RPC consumer    — `trex.control.req` (pause/resume/list + kill_switch/resume_kill_switch, REQ-032/043)
  - execution loop         — broker dispatch + poll + slot transitions + event emit
  - ledger writer          — ledger outbox drain (REQ-019)
  - snapshot scheduler      — daily portfolio-hub snapshot refresh + universe
                              slot reconcile (REQ-030, dcy-020/REQ-058)
  - safety-sell worker      — forced_close / stop-loss·take-profit / market-close (REQ-037/038/039)

Each consumer/publisher uses its OWN RabbitMQTransport instance because
pika BlockingConnection is not thread-safe across channels.
Console script: `trex-core-daemon`.
"""
from __future__ import annotations

import logging
import signal
import threading
from datetime import time as time_cls

from trex_core.adapters.broker.paper import PaperBroker
from trex_core.adapters.ledger import InMemoryLedger, PortfolioHubLedger
from trex_core.adapters.messaging import (
    InMemoryTransport,
    RabbitMQTransport,
    topology,
)
from trex_core.adapters.persistence.migrate import apply_migrations
from trex_core.adapters.persistence.postgresql import (
    PgStrategyStateStore,
    PostgreSQLIntentStore,
    PostgreSQLLedgerOutbox,
    PostgreSQLSlotRepository,
)
from trex_core.adapters.portfolio import (
    PortfolioHubReadAdapter,
    PostgreSQLPauseStore,
    StubPortfolioRead,
)
from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore
from trex_core.application.broker_port import BrokerPort
from trex_core.application.execution import ExecutionService
from trex_core.application.intents import IntentService
from trex_core.application.ledger_outbox import LedgerWriter
from trex_core.application.ledger_port import LedgerPort
from trex_core.application.messaging import MessagingTransport
from trex_core.application.killswitch import KillSwitchService
from trex_core.application.messaging_handlers import (
    make_control_rpc_handler,
    make_intent_rpc_handler,
    make_slots_rpc_handler,
    make_strategy_state_set_handler,
    slot_transition_event_payload,
)
from trex_core.application.portfolio import (
    PortfolioHubReadPort,
    SnapshotStore,
)
from trex_core.application.safety_sell import SafetySellService
from trex_core.application.snapshot_scheduler import SnapshotScheduler
from trex_core.application.watchdog import WatchdogStore
from trex_core.domain.slot import SlotService
from trex_core.runtime.config import DaemonConfig, resolve_config_toml

_log = logging.getLogger(__name__)


def _build_broker(config: DaemonConfig) -> BrokerPort:
    if config.broker == "kis":
        from trex_core.adapters.broker.kis import KISBroker

        k = config.kis
        token_cache = None
        if k.redis_url:
            from trex_core.adapters.broker.token_cache import RedisTokenCache

            token_cache = RedisTokenCache(k.redis_url)
        return KISBroker(
            k.account_ref, app_key=k.app_key, app_secret=k.app_secret,
            base_url=k.base_url, timeout=k.timeout, allow_live=k.allow_live,
            token_cache=token_cache,
        )
    seq_store = None
    if config.paper.redis_url:
        from trex_core.adapters.broker.seq_store import RedisSeqStore

        seq_store = RedisSeqStore(
            config.paper.redis_url,
            key=f"trex:broker:paper:{config.instance_key}:order-seq",
        )
    return PaperBroker(
        fill_after_polls=config.paper.fill_after_polls,
        qty_per_poll=config.paper.qty_per_poll,
        reject_symbols=set(config.paper.reject_symbols),
        seq_store=seq_store,
    )


def _build_ledger(config: DaemonConfig) -> LedgerPort:
    """portfolio-hub when configured; otherwise in-memory stub (REQ-017)."""
    pf = config.portfolio_hub
    if pf is None:
        return InMemoryLedger()
    return PortfolioHubLedger(pf.base_url, token=pf.token or None, timeout=pf.timeout)


def _build_transport(config: DaemonConfig) -> MessagingTransport:
    """RabbitMQ when configured; otherwise in-memory stub (REQ-022)."""
    mq = config.rabbitmq
    if mq is None:
        # Stub: lets the daemon boot in environments without a broker (CI,
        # local dev). REQ-024 degrade policy still applies — call_rpc raises.
        return InMemoryTransport()
    return RabbitMQTransport(mq.url, timeout=mq.timeout)


def _build_portfolio_read(config: DaemonConfig) -> PortfolioHubReadPort:
    """portfolio-hub HTTP read adapter when configured; otherwise stub (REQ-027)."""
    pf = config.portfolio_hub
    if pf is None:
        return StubPortfolioRead()
    return PortfolioHubReadAdapter(
        pf.base_url, token=pf.token or None, timeout=pf.timeout,
    )


def _parse_kst_time(spec: str) -> time_cls:
    parts = spec.split(":")
    h = int(parts[0])
    m = int(parts[1]) if len(parts) > 1 else 0
    return time_cls(hour=h, minute=m)


class Daemon:
    """Thread-orchestrating daemon (no HTTP)."""

    def __init__(self, config: DaemonConfig) -> None:
        self.config = config
        self.stop = threading.Event()
        self._threads: list[threading.Thread] = []
        self._transports: list[MessagingTransport] = []

        # Persistence + services
        self.slot_repo = PostgreSQLSlotRepository(config.dsn)
        self.intent_store = PostgreSQLIntentStore(config.dsn)
        # dcy-028 / REQ-079·081 (BAT-042): opaque strategy-state store. Injected
        # into the slot service so close (→INACTIVE) deletes the slot's state.
        self.strategy_state_store = PgStrategyStateStore(config.dsn)
        self.slot_service = SlotService(
            self.slot_repo,
            instance_key=config.instance_key,
            strategy_state_store=self.strategy_state_store,
        )
        self.intent_service = IntentService(
            self.slot_service, self.slot_repo, self.intent_store,
            resolve_symbol=self._resolve_symbol,
            instance_key=config.instance_key,
            max_active_slots=config.max_active_slots,  # dcy-020: asset-centric gate
            default_order_timeout_sec=config.default_order_timeout_sec,  # dcy-022
        )
        self.broker = _build_broker(config)
        self.outbox = PostgreSQLLedgerOutbox(config.dsn)
        # BAT-008: portfolio-hub read + snapshot store + durable pause + scheduler.
        # (Created before the ledger writer so option-D write-back can target it.)
        self.snapshot_store = SnapshotStore()
        self.ledger_writer = LedgerWriter(
            self.outbox, _build_ledger(config),
            snapshot_store=self.snapshot_store,  # dcy-018 / REQ-030: option D write-back
            retry_base_sec=config.ledger_retry_base_sec,
            retry_cap_sec=config.ledger_retry_cap_sec,
            max_attempts=config.ledger_max_attempts,
        )
        # BAT-010/011: per-slot watchdog (snapshot RPC writer, safety-sell reader).
        self.watchdog = WatchdogStore()
        self.pause_store = PostgreSQLPauseStore(config.dsn)
        # BAT-012: day-scoped kill-switch exclusion store + service.
        self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn)
        self.kill_switch = KillSwitchService(
            slot_repo=self.slot_repo,
            slot_service=self.slot_service,
            killswitch_store=self.killswitch_store,
            instance_key=config.instance_key,
        )
        self.portfolio_read = _build_portfolio_read(config)
        self.snapshot_scheduler = SnapshotScheduler(
            read_port=self.portfolio_read,
            store=self.snapshot_store,
            broker=self.broker,
            schedule_kst=_parse_kst_time(config.snapshot_schedule.kst_time),
            # dcy-020 / REQ-058·062 (BAT-030): universe slot reconcile (creates/
            # retires the permanent asset slots) + core-holding divergence
            # reconcile — always on (asset-centric is the only model).
            slot_repo=self.slot_repo,
            slot_service=self.slot_service,
            instance_key=config.instance_key,
        )

        # Event publisher transport — one dedicated instance for the
        # execution-loop thread (pika is not thread-safe).
        self.event_publisher: MessagingTransport = _build_transport(config)
        self._transports.append(self.event_publisher)

        self.execution = ExecutionService(
            self.slot_service, self.slot_repo, self.broker,
            ledger_outbox=self.outbox,
            event_emitter=self._make_event_emitter(),
            stale_pending_threshold_sec=config.pending_stale_threshold_sec,
            instance_key=config.instance_key,
        )

        # BAT-011: safety-sell trigger service (forced_close / SL-TP / market-close).
        self.safety_sell = SafetySellService(
            slot_repo=self.slot_repo,
            slot_service=self.slot_service,
            broker=self.broker,
            watchdog=self.watchdog,
            config=config.safety_sell,
            instance_key=config.instance_key,
        )

    # ------------------------------------------------------------ buy symbol resolve
    def _resolve_symbol(self, portfolio_id: str):
        """dcy-011 / REQ-007: resolve symbol from the universe snapshot for a
        portfolio_id at buy admission (None → unknown_portfolio reject).

        SnapshotStore.get returns a SnapshotAsset wrapping the PortfolioItem;
        the symbol lives on `.item.symbol`."""
        asset = self.snapshot_store.get(portfolio_id)
        return asset.item.symbol if asset is not None else None

    # ------------------------------------------------------------ event hook
    def _make_event_emitter(self):
        def emit(event_type: str, slot) -> None:
            payload = slot_transition_event_payload(slot, event_type=event_type)
            rk = topology.event_routing_key(
                event_type, slot.strategy_id or "_", slot.symbol or "_",
            )
            try:
                self.event_publisher.publish_event(rk, payload)
            except Exception:  # noqa: BLE001 - publish failure must not roll back
                _log.warning("event publish failed (rk=%s)", rk, exc_info=True)
        return emit

    # ------------------------------------------------------------ workers
    def _execution_loop(self) -> None:
        while not self.stop.wait(self.config.execution_interval_sec):
            try:
                self.execution.run_once()
            except Exception:  # noqa: BLE001 - loop must survive transient errors
                continue

    def _ledger_loop(self) -> None:
        while not self.stop.wait(self.config.ledger_interval_sec):
            try:
                self.ledger_writer.run_once()
            except Exception:  # noqa: BLE001
                continue

    def _intent_rpc_loop(self) -> None:
        transport = _build_transport(self.config)
        self._transports.append(transport)
        try:
            transport.declare_topology(self.config.instance_key)
        except Exception:  # noqa: BLE001
            pass
        handler = make_intent_rpc_handler(
            self.intent_service,
            slot_service=self.slot_service,
            watchdog_store=self.watchdog,
        )
        transport.consume_rpc_forever(
            topology.intent_queue(self.config.instance_key), handler, self.stop)

    def _slots_rpc_loop(self) -> None:
        transport = _build_transport(self.config)
        self._transports.append(transport)
        try:
            transport.declare_topology(self.config.instance_key)
        except Exception:  # noqa: BLE001
            pass
        handler = make_slots_rpc_handler(
            self.slot_repo,
            instance_key=self.config.instance_key,
            snapshot_store=self.snapshot_store,
            pause_store=self.pause_store,
            watchdog_store=self.watchdog,
            watchdog_timeout_sec=self.config.safety_sell.watchdog_timeout_sec,
            ledger_outbox=self.outbox,
            max_active_slots=self.config.max_active_slots,
            strategy_state_store=self.strategy_state_store,  # dcy-028 C-1
        )
        transport.consume_rpc_forever(
            topology.slots_queue(self.config.instance_key), handler, self.stop)

    def _strategy_state_set_loop(self) -> None:
        # dcy-028 / REQ-080: opaque strategy-state write RPC (occupancy_seq+version CAS).
        transport = _build_transport(self.config)
        self._transports.append(transport)
        try:
            transport.declare_topology(self.config.instance_key)
        except Exception:  # noqa: BLE001
            pass
        handler = make_strategy_state_set_handler(
            self.slot_repo, self.strategy_state_store,
        )
        transport.consume_rpc_forever(
            topology.strategy_state_set_queue(self.config.instance_key), handler, self.stop)

    def _control_rpc_loop(self) -> None:
        # REQ-043: operator control channel (pause/resume/list + kill_switch/resume_kill_switch).
        transport = _build_transport(self.config)
        self._transports.append(transport)
        try:
            transport.declare_topology(self.config.instance_key)
        except Exception:  # noqa: BLE001
            pass
        handler = make_control_rpc_handler(
            self.pause_store,
            kill_switch_service=self.kill_switch,
            cancel_pending_fn=self.execution.cancel_pending_slot,  # dcy-022/REQ-071
        )
        transport.consume_rpc_forever(
            topology.control_queue(self.config.instance_key), handler, self.stop)

    def _safety_sell_loop(self) -> None:
        # REQ-037/038/039: forced_close / stop-loss·take-profit / market-close.
        while not self.stop.wait(self.safety_sell.base_tick_sec):
            try:
                self.safety_sell.run_once()
            except Exception:  # noqa: BLE001 - loop must survive transient errors
                continue

    def _snapshot_scheduler_loop(self) -> None:
        try:
            self.snapshot_scheduler.run_forever(
                self.stop, tick_sec=self.config.snapshot_schedule.tick_sec,
            )
        except Exception:  # noqa: BLE001
            _log.exception("snapshot scheduler loop crashed")

    # ------------------------------------------------------------ lifecycle
    def setup(self) -> None:
        apply_migrations(self.config.dsn)
        # dcy-020 (BAT-030): the daily pool preallocation is gone — permanent
        # asset slots are created/retired by the snapshot scheduler's universe
        # reconcile (boot fetch runs immediately in its thread).
        self.execution.reconcile()  # REQ-014 restart recovery
        try:
            self.event_publisher.declare_topology(self.config.instance_key)
        except Exception:  # noqa: BLE001
            _log.warning("event publisher topology declare failed", exc_info=True)

    def start(self) -> None:
        workers = [
            ("trex-execution", self._execution_loop),
            ("trex-ledger", self._ledger_loop),
            ("trex-intent-rpc", self._intent_rpc_loop),
            ("trex-slots-rpc", self._slots_rpc_loop),
            ("trex-strategy-state-set", self._strategy_state_set_loop),
            ("trex-control-rpc", self._control_rpc_loop),
            ("trex-snapshot-scheduler", self._snapshot_scheduler_loop),
        ]
        if self.config.safety_sell_enabled:
            workers.append(("trex-safety-sell", self._safety_sell_loop))
        else:
            _log.warning(
                "safety_sell_enabled=false — trex-safety-sell worker not started. "
                "Autonomous safety-sell will NOT run."
            )
        for name, target in workers:
            t = threading.Thread(target=target, name=name, daemon=True)
            t.start()
            self._threads.append(t)

    def shutdown(self) -> None:
        self.stop.set()
        for t in self._threads:
            t.join(
                timeout=max(self.config.execution_interval_sec, self.config.ledger_interval_sec) * 3
            )
        for tr in self._transports:
            try:
                tr.close()
            except Exception:  # noqa: BLE001
                pass

    def run(self) -> None:  # pragma: no cover - process entrypoint
        self.setup()
        self.start()
        signal.signal(signal.SIGINT, lambda *_: self.stop.set())
        signal.signal(signal.SIGTERM, lambda *_: self.stop.set())
        try:
            while not self.stop.is_set():
                self.stop.wait(1.0)
        finally:
            self.shutdown()


def main() -> None:  # pragma: no cover - process entrypoint
    import argparse

    logging.basicConfig(level=logging.INFO)
    parser = argparse.ArgumentParser(prog="trex-core-daemon")
    parser.add_argument(
        "--config", required=True,
        help="per-instance TOML name (resolved within --config-path) or explicit "
             "path. Required — no default (REQ-048 / dcy-009 OQ-2).",
    )
    parser.add_argument(
        "--config-path", default=None,
        help="config directory (default ./config) for --config name resolution "
             "and the matching .env (REQ-048/010 CHG-20260609-01).",
    )
    parser.add_argument(
        "--env-file", default=None,
        help="explicit .env path; overrides <config-dir>/.env (REQ-010 CHG-20260609-01).",
    )
    args = parser.parse_args()
    toml_path = resolve_config_toml(args.config, args.config_path)
    config = DaemonConfig.from_config(toml_path, env_file=args.env_file)
    Daemon(config).run()


if __name__ == "__main__":  # pragma: no cover
    main()