"""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 as exc: # noqa: BLE001 - loop must survive transient errors # Sanitized: emit only the exception type, not the message or # traceback, so credentials / DSN / raw payloads in adapter # exceptions are never written to the log stream. _log.error( "execution loop error — worker continues: exc_type=%s", type(exc).__name__, ) 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 /.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()