658 lines
29 KiB
Python
658 lines
29 KiB
Python
"""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.account_snapshot import (
|
|
PostgreSQLAccountSnapshotStore,
|
|
)
|
|
from trex_core.adapters.persistence.auto_reconcile import PostgreSQLAutoReconcileStore
|
|
from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore
|
|
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,
|
|
PostgreSQLReconcileWarningStore,
|
|
StubPortfolioRead,
|
|
)
|
|
from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore
|
|
from trex_core.adapters.portfolio.reconcile_notification_store import (
|
|
PostgreSQLReconcileNotificationStore,
|
|
)
|
|
from trex_core.adapters.portfolio.unmanaged_position_store import (
|
|
PostgreSQLUnmanagedPositionStore,
|
|
)
|
|
from trex_core.application.account_snapshot import AccountSnapshotService
|
|
from trex_core.application.alarm_emitter import AlarmEmitter
|
|
from trex_core.application.broker_port import BrokerPort
|
|
from trex_core.application.execution import ExecutionService
|
|
from trex_core.application.intents import BuyIntent, IntentService
|
|
from trex_core.application.killswitch import KillSwitchService
|
|
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.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, instance_key=config.instance_key)
|
|
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, case_code) PK).
|
|
self.warn_store = PostgreSQLReconcileWarningStore(config.dsn)
|
|
self.auto_reconcile_store = PostgreSQLAutoReconcileStore(config.dsn)
|
|
# req-0007 / ac-0024: durable reconcile notification deduplication store.
|
|
self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn)
|
|
self.unmanaged_position_store = PostgreSQLUnmanagedPositionStore(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.reconciliation_broker, self.reconciliation_broker_source = (
|
|
self.broker_provider.get_broker_for_reconciliation()
|
|
)
|
|
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,
|
|
core_holdings_broker=self.reconciliation_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
|
|
unmanaged_position_store=self.unmanaged_position_store,
|
|
cost_basis_port=self.portfolio_read,
|
|
auto_reconcile_store=self.auto_reconcile_store,
|
|
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 captured before transitions that clear owner
|
|
# fields (for SELL, the admission-owned sell_strategy_id). Missing
|
|
# SELL ownership is blocked below instead of synthesizing "_".
|
|
effective_strategy_id = (
|
|
routing_strategy_id or getattr(slot, "strategy_id", None)
|
|
)
|
|
# req-0011/ac-0039 defense in depth: application execution normally
|
|
# skips ownerless SELL events before this hook. Never synthesize the
|
|
# legacy "_" owner if a direct caller violates that boundary.
|
|
if (
|
|
event_type == "slot_filled_sell"
|
|
and not (
|
|
isinstance(effective_strategy_id, str)
|
|
and effective_strategy_id.strip()
|
|
)
|
|
):
|
|
_log.error(
|
|
"sell event publish blocked reason=sell_event_owner_missing "
|
|
"slot_id=%s symbol=%s event_type=%s",
|
|
getattr(slot, "slot_id", None) or getattr(slot, "id", None),
|
|
getattr(slot, "symbol", None),
|
|
event_type,
|
|
)
|
|
return
|
|
effective_strategy_id = effective_strategy_id or "_"
|
|
rk = topology.event_routing_key(
|
|
event_type, effective_strategy_id, slot.symbol or "_",
|
|
)
|
|
try:
|
|
self.event_publisher.publish_event(rk, payload)
|
|
except Exception as exc: # noqa: BLE001 - publish failure must not roll back
|
|
# req-0011/ac-0039: transport failures are observable without
|
|
# rendering raw exception text, traceback, broker payload, or
|
|
# connection details into logs.
|
|
_log.error(
|
|
"event publish failed reason=event_publish_failed "
|
|
"rk=%s slot_id=%s strategy=%s exc_type=%s",
|
|
rk,
|
|
getattr(slot, "slot_id", None) or getattr(slot, "id", None),
|
|
getattr(slot, "strategy_id", None),
|
|
type(exc).__name__,
|
|
)
|
|
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 as exc: # noqa: BLE001 - loop must survive transient errors
|
|
_log.error(
|
|
"ledger writer loop error: reason=ledger_writer_run_once_failed "
|
|
"instance_key=%s exc_type=%s",
|
|
self.config.instance_key,
|
|
type(exc).__name__,
|
|
)
|
|
|
|
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,S110
|
|
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,S110
|
|
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,
|
|
ledger_outbox_metrics_reader=self.ledger_writer,
|
|
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,
|
|
broker_balance_health_reader=self,
|
|
)
|
|
transport.consume_rpc_forever(
|
|
topology.slots_queue(self.config.instance_key), handler, self.stop)
|
|
|
|
def balance_health_for_slots_rpc(self) -> dict[str, dict[str, object]]:
|
|
reconcile = self.snapshot_scheduler.balance_worker_health(
|
|
stale_after_sec=24 * 60 * 60,
|
|
)
|
|
return {
|
|
"execution": self.execution.balance_worker_health(stale_after_sec=600),
|
|
"position_reconciliation": reconcile["position_reconciliation"],
|
|
"core_holding_reconciliation": reconcile["core_holding_reconciliation"],
|
|
}
|
|
|
|
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,S110
|
|
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,S110
|
|
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,S112 - 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,S110
|
|
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
|
|
from trex_core.version import __version__
|
|
configure_logging(level=logging.INFO) # REQ-084: ISO8601(KST)+level+json+cid
|
|
parser = argparse.ArgumentParser(prog="trex-core-daemon")
|
|
parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
|
|
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()
|