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.account_snapshot import (
    PostgreSQLAccountSnapshotStore,
)
from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore
from trex_core.adapters.persistence.postgresql import (
    PgStrategyStateStore,
    PostgreSQLIntentStore,
    PostgreSQLLedgerOutbox,
    PostgreSQLSlotRepository,
)
from trex_core.adapters.portfolio import (
    PortfolioHubReadAdapter,
    PostgreSQLPauseStore,
    PostgreSQLReconcileWarningStore,
    StubPortfolioRead,
)
from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore
from trex_core.adapters.portfolio.reconcile_notification_store import (
    PostgreSQLReconcileNotificationStore,
)
from trex_core.application.alarm_emitter import AlarmEmitter
from trex_core.application.account_snapshot import AccountSnapshotService
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.intents import BuyIntent
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:
    """Backward-compatible env-only factory; new callers use broker_provider."""
    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_kis_reinit_callback(config: DaemonConfig, broker: BrokerPort):
    """Build a KISBroker reinitialize callback for the day-start credential hook.

    Returns broker.reinitialize if all conditions are met:
      - config.broker == 'kis' (not paper)
      - config.credential_fetch_enabled == True (opt-in)
      - broker has a reinitialize() method (KISBroker)

    Returns None otherwise so SnapshotScheduler skips reinit (backward compat).
    dcy-036 / REQ-113 AC-2 / REQ-115 AC-1.
    """
    if config.broker != "kis":
        return None
    if not config.credential_fetch_enabled:
        return None
    reinit = getattr(broker, "reinitialize", None)
    if reinit is None or not callable(reinit):
        return None
    return reinit


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 persists
        # independently of slot occupancy and is shared by both RPC handlers.
        self.strategy_state_store = PgStrategyStateStore(config.dsn)
        self.slot_service = SlotService(
            self.slot_repo,
            instance_key=config.instance_key,
        )

        self.outbox = PostgreSQLLedgerOutbox(config.dsn)
        self.ledger_skip_store = PostgreSQLLedgerSkipStore(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()
        # BAT-010/011: per-slot watchdog (snapshot RPC writer, safety-sell reader).
        self.watchdog = WatchdogStore()
        self.pause_store = PostgreSQLPauseStore(config.dsn)
        # BAT-050 / REQ-102: durable reconcile-warning store (slot_id PK).
        self.warn_store = PostgreSQLReconcileWarningStore(config.dsn)
        # req-0007 / ac-0024: durable reconcile notification deduplication store.
        self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn)
        # BAT-012: day-scoped kill-switch exclusion store + service.
        self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn)
        self.portfolio_read = _build_portfolio_read(config)
        # REQ-119: one provider owns source resolution and the long-lived broker.
        # Fetched mode never starts from an implicit env broker.
        from trex_core.runtime.broker_provider import build_broker_provider
        self.broker_provider = build_broker_provider(config, read_port=self.portfolio_read)
        self.broker, self.broker_source = self.broker_provider.get_broker_for_read()
        self.account_snapshot_service: AccountSnapshotService | None = None
        if config.account_snapshot.enabled:
            if config.broker != "kis":
                raise RuntimeError(
                    "account snapshot recovery is supported only with broker=kis"
                )
            required = (
                "account_scope_hash",
                "get_balance",
                "get_available_cash",
            )
            if any(not callable(getattr(self.broker, name, None)) for name in required):
                raise RuntimeError(
                    "configured KIS broker lacks account snapshot capabilities"
                )
            self.account_snapshot_service = AccountSnapshotService(
                broker=self.broker,  # type: ignore[arg-type]
                store=PostgreSQLAccountSnapshotStore(config.dsn),
                buy_max_age_sec=config.account_snapshot.buy_max_age_sec,
                sell_max_age_sec=config.account_snapshot.sell_max_age_sec,
                refresh_interval_sec=config.account_snapshot.refresh_interval_sec,
            )

        # 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)

        # dcy-033 / REQ-100: alarm emitter — shares the event_publisher transport
        # (both run on the execution-loop thread; pika thread-safety rule satisfied).
        # Created before services so all services receive the same emitter instance.
        self.alarm_emitter = AlarmEmitter(self.event_publisher)
        # Late-bind alarm callback into event_publisher for adapter-layer alarm codes
        # (KD-1: publish_unavailable_after_reconnect / dlq_publish_failed).
        if hasattr(self.event_publisher, "set_alarm_callback"):
            self.event_publisher.set_alarm_callback(self.alarm_emitter.publish_alarm)

        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,
            alarm_emitter=self.alarm_emitter,  # dcy-033 / REQ-100
            ledger_skip_store=self.ledger_skip_store,
            instance_key=config.instance_key,
        )
        self.kill_switch = KillSwitchService(
            slot_repo=self.slot_repo,
            slot_service=self.slot_service,
            killswitch_store=self.killswitch_store,
            instance_key=config.instance_key,
            alarm_emitter=self.alarm_emitter,  # dcy-033 / REQ-100
        )
        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,
            alarm_emitter=self.alarm_emitter,  # dcy-033 / REQ-100
            warn_store=self.warn_store,  # BAT-050 / REQ-103: durable reconcile-warning
            notification_store=self.notification_store,  # req-0007 / ac-0024
            credential_fetch_enabled=config.credential_fetch_enabled,  # dcy-036 / REQ-113
            credential_refresh=self.broker_provider.refresh_day_start_credential,
            # dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: KISBroker reinit callback.
            # None when broker!=kis or credential_fetch_enabled=False (backward compat).
            kis_reinit_callback=_build_kis_reinit_callback(config, self.broker),
        )
        # dcy-036 / REQ-114: wire entry_allowed_fn after snapshot_scheduler is created.
        # When credential_fetch_enabled=False (default), entry_allowed_fn=None = no gate.
        entry_allowed_fn = (
            self.snapshot_scheduler.get_entry_allowed
            if config.credential_fetch_enabled
            else None
        )
        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
            warn_store=self.warn_store,  # BAT-050 / REQ-104: reconcile-warning gate
            alarm_emitter=self.alarm_emitter,  # req-0008 / ac-0025
            entry_allowed_fn=entry_allowed_fn,  # dcy-036 / REQ-114: day-start gate
            buy_guard_fn=(
                self._account_snapshot_buy_guard
                if self.account_snapshot_service is not None
                else None
            ),
        )
        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,
            alarm_emitter=self.alarm_emitter,  # dcy-033 / REQ-100
            ledger_skip_store=self.ledger_skip_store,
            balance_reader=(
                self.account_snapshot_service.holdings_for_sell
                if self.account_snapshot_service is not None
                else None
            ),
        )

        # 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

    def _account_snapshot_buy_guard(self, intent: BuyIntent) -> str | None:
        """Bound one new buy against fresh cached cash plus pending reservations."""
        service = self.account_snapshot_service
        if service is None:
            return None
        reserved_cash = 0.0
        for slot in self.slot_repo.list_slots(
            self.config.instance_key, status_filter={"pending_buy"}
        ):
            if (
                slot.qty is None
                or slot.qty <= 0
                or slot.limit_price is None
                or slot.limit_price <= 0
            ):
                return "pending_buy_notional_unknown"
            price = float(slot.limit_price)
            if slot.slippage_pct:
                price *= 1.0 + float(slot.slippage_pct) / 100.0
            reserved_cash += float(slot.qty) * price
        price = intent.entry_price
        if price is not None and intent.slippage_pct:
            price *= 1.0 + float(intent.slippage_pct) / 100.0
        return service.buy_guard(
            qty=intent.qty,
            limit_price=price,
            reserved_cash=reserved_cash,
        )

    # ------------------------------------------------------------ event hook
    def _make_event_emitter(self):
        def emit(event_type: str, slot, routing_strategy_id: str | None = None) -> None:
            payload = slot_transition_event_payload(slot, event_type=event_type)
            # routing_strategy_id is supplied when the transition clears
            # slot.strategy_id (e.g. PENDING_SELL -> INACTIVE for slot_filled_sell).
            # Without this hint the routing key would use "_" and the message would
            # be NO_ROUTEd to the DLQ instead of reaching the strategy queue
            # (M2 REL-009 feedback / slot_filled NO_ROUTE fix).
            effective_strategy_id = (
                routing_strategy_id or getattr(slot, "strategy_id", None) or "_"
            )
            rk = topology.event_routing_key(
                event_type, effective_strategy_id, slot.symbol or "_",
            )
            try:
                self.event_publisher.publish_event(rk, payload)
            except Exception:  # noqa: BLE001 - publish failure must not roll back
                # REQ-093: escalate to ERROR so operators can observe via log
                # alert. Include routing key + slot identity for correlation.
                _log.error(
                    "event publish failed (rk=%s slot_id=%s strategy=%s)",
                    rk,
                    getattr(slot, "slot_id", None) or getattr(slot, "id", None),
                    getattr(slot, "strategy_id", None),
                    exc_info=True,
                )
        return emit

    # ------------------------------------------------------------ workers
    def _execution_loop(self) -> None:
        while not self.stop.wait(self.config.execution_interval_sec):
            # REQ-092: keepalive tick — process pending I/O frames on the
            # publish connection before each execution run.  Called in the
            # same thread that owns event_publisher (pika thread-safety rule).
            self.event_publisher.keepalive_tick()
            try:
                self.execution.run_once()
            except Exception:  # noqa: BLE001 - loop must survive transient errors
                _log.exception("execution loop error — worker continues")

    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,
            warn_store=self.warn_store,  # BAT-050 / REQ-104: reconcile_warned field
            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
            broker_quantity_reader=self.account_snapshot_service,
        )
        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 (entry-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")

    def _account_snapshot_loop(self) -> None:
        service = self.account_snapshot_service
        if service is not None:
            service.run_forever(self.stop)

    # ------------------------------------------------------------ lifecycle
    def setup(self) -> None:
        apply_migrations(self.config.dsn)
        if self.account_snapshot_service is not None:
            # First refresh is synchronous: buy admission never starts from a
            # stale row without first attempting the current account.
            self.account_snapshot_service.refresh()
        # 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."
            )
        if self.account_snapshot_service is not None:
            workers.append(("trex-account-snapshot", self._account_snapshot_loop))
        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

    from trex_core.runtime.logging_setup import configure_logging
    configure_logging(level=logging.INFO)  # REQ-084: ISO8601(KST)+level+json+cid
    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()