"""trex-core operator CLI (REQ-032, dcy-016; REQ-105/106, dcy-034). Subcommands: pause — durably pause a portfolio (Postgres, direct-DB) resume — clear the pause (direct-DB) paused — list paused portfolio_ids (direct-DB) universe — list today's universe assets (snapshot RPC) slots [--status] — list positions/slots (snapshot RPC) enter — submit a buy intent (intent RPC; safeguarded) exit — submit a sell intent (intent RPC; safeguarded) cancel — cancel a pending order (control RPC; safeguarded) sync — sync slot qty to broker ground truth (direct-DB; REQ-105) verify — re-check C-case reconcile-warning (direct-DB; REQ-106) unmanaged list — query durable unmanaged-position lifecycle state unmanaged audit — inspect lifecycle/operator audit trail for one symbol unmanaged record — record explicit operator disposition + evidence reference ledger backfill — evidence-backed historical ledger recovery ledger skips list — query durable skipped-ledger records ledger skips resolve — explicitly resolve a skipped-ledger record universe/slots/enter/exit go through the daemon's RabbitMQ RPC surface so the single-writer admission + invariants are preserved (no direct slot writes, dcy-016 NFR-1). Operator queries carry `actor="operator"` so they do NOT count as a strategy liveness signal (REQ-021/034 watchdog suppression, R-4). enter/exit safeguards (dcy-016 OQ-2): confirmation prompt ON by default; `--yes` skips it (paper); live accounts require `--yes --live` (double ack); `--dry-run` previews the resolved intent without submitting. `--json` is a non-interactive machine mode, so enter/exit with `--json` require `--yes`. sync safeguards (REQ-105): `--yes` required (skip confirmation prompt); live accounts require `--yes --live` (double ack); `--dry-run` previews before/after without writing. sync is direct-DB (no RPC). verify (REQ-106): read-centric (no --yes/--live gate needed). ledger backfill: verifies KIS daily-ccld evidence, reconstructs one ledger trade, enqueues it through ledger_outbox, immediately drives the LedgerWriter path, and resolves matching durable skipped-ledger records. """ from __future__ import annotations import argparse import getpass import json as _json import logging import re import sys import uuid from datetime import UTC, datetime, timedelta, timezone from pathlib import Path from zoneinfo import ZoneInfo from trex_core.adapters.messaging import topology from trex_core.adapters.portfolio.pause_store import PostgreSQLPauseStore from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.ledger_outbox import LedgerWriter, OutboxOutcome from trex_core.application.ledger_port import LedgerTrade from trex_core.application.ledger_skip import ledger_skip_alarm_fields, publish_ledger_skip_alarm from trex_core.version import __version__ from trex_core.runtime.config import ( DaemonConfig, DaemonInstanceConfig, _load_dotenv, resolve_config_toml, ) _DEFAULT_TIMEOUT = 10.0 _DEFAULT_LEDGER_BACKFILL_AUDIT_DIR = "artifacts/validation/ledger_backfill" _DEFAULT_LEDGER_SKIP_RESOLVE_AUDIT_DIR = "artifacts/validation/ledger_skip_resolve" _log = logging.getLogger(__name__) # --------------------------------------------------------------------------- # # instance routing (OQ-1: --config canonical, --instance shortcut/override) # --------------------------------------------------------------------------- # def _resolve_instance(args) -> str: if getattr(args, "instance", None): return args.instance if getattr(args, "config", None): toml = resolve_config_toml(args.config, getattr(args, "config_path", None)) return DaemonInstanceConfig.from_toml(toml).instance_key return topology.DEFAULT_INSTANCE_KEY def _operator_actor() -> str: try: username = getpass.getuser().strip() except Exception: # noqa: BLE001 username = "" return f"cli:{username or 'unknown'}" # --------------------------------------------------------------------------- # # default transport-backed RPC (overridable in tests) # --------------------------------------------------------------------------- # def _default_rpc(cfg: DaemonConfig): """Return a callable(queue, payload, *, correlation_id, timeout) -> dict backed by a real RabbitMQ transport built from config.""" from trex_core.adapters.messaging import InMemoryTransport, RabbitMQTransport mq = cfg.rabbitmq transport = RabbitMQTransport(mq.url, timeout=mq.timeout) if mq else InMemoryTransport() def _rpc(queue: str, payload: dict, *, correlation_id: str, timeout: float) -> dict: try: return transport.call_rpc( queue, payload, correlation_id=correlation_id, timeout=timeout ) finally: transport.close() return _rpc def _default_alarm_emitter(cfg: DaemonConfig): from trex_core.adapters.messaging import InMemoryTransport, RabbitMQTransport mq = getattr(cfg, "rabbitmq", None) transport = RabbitMQTransport(mq.url, timeout=mq.timeout) if mq else InMemoryTransport() return AlarmEmitter(transport), transport def _default_confirm(prompt: str) -> bool: try: return input(f"{prompt} [y/N] ").strip().lower() in ("y", "yes") except EOFError: return False def _broker_mode(cfg: DaemonConfig) -> tuple[str, bool]: """(broker, is_live). live = kis broker with allow_live enabled.""" broker = cfg.broker is_live = broker == "kis" and bool(cfg.kis and cfg.kis.allow_live) return broker, is_live def _emit(data: dict, *, as_json: bool) -> None: if as_json: print(_json.dumps(data, ensure_ascii=False, default=str)) def _emit_pretty_json(data: dict) -> None: print(_json.dumps(data, ensure_ascii=False, indent=2, default=str)) # --------------------------------------------------------------------------- # # RPC subcommand handlers # --------------------------------------------------------------------------- # def _today_iso(cfg: DaemonConfig) -> str: return datetime.now(tz=ZoneInfo(cfg.trading_timezone)).date().isoformat() def _cmd_universe(args, cfg, rpc) -> int: # Universe = admittable slots (filter=inactive, spy=true — operator view) instance = _resolve_instance(args) payload = {"filter": "inactive", "spy": True} resp = rpc(topology.slots_queue(instance), payload, correlation_id=str(uuid.uuid4()), timeout=args.timeout) slots = resp.get("slots", []) if args.json: _emit({"instance": instance, "filter": "inactive", "slots": slots, "degraded": resp.get("degraded"), "reason": resp.get("reason")}, as_json=True) else: print(f"instance={instance} filter=inactive count={len(slots)} " f"degraded={resp.get('degraded')}") for s in slots: print(f" {s.get('portfolio_id')}\t{s.get('symbol')}\t" f"qty(hub)={s.get('asset_quantity')}\t" f"qty(slot)={s.get('slot_quantity', s.get('qty'))}") return 0 def _cmd_slots(args, cfg, rpc) -> int: instance = _resolve_instance(args) status = (args.status or "all").lower() # REQ-063 CHG-20260611-01: map --status to filter; buyable → inactive filter_val = "inactive" if status == "buyable" else status payload = {"filter": filter_val, "spy": True} resp = rpc(topology.slots_queue(instance), payload, correlation_id=str(uuid.uuid4()), timeout=args.timeout) slots = resp.get("slots", []) # --symbol / --slot-id: client-side filters (composable with server-side filter) if args.symbol: slots = [s for s in slots if s.get("symbol") == args.symbol] if args.slot_id: slots = [s for s in slots if s.get("slot_id") == args.slot_id] # watchdog-expired ownership: slots with watchdog.expired=True owned = {s.get("portfolio_id") for s in slots if (s.get("watchdog") or {}).get("expired")} if args.json: _emit({"instance": instance, "filter": filter_val, "slots": slots}, as_json=True) else: header = f"instance={instance} filter={filter_val}" if args.symbol: header += f" symbol={args.symbol}" if args.slot_id: header += f" slot_id={args.slot_id}" print(f"{header} count={len(slots)} " f"watchdog_expired={len([s for s in slots if s.get('portfolio_id') in owned])}") if args.symbol or args.slot_id: for s in slots: _print_slot_detail(s) else: for s in slots: pid = s.get("portfolio_id") wd = s.get("watchdog") or {} tag = " [WATCHDOG: strategy absent]" if wd.get("expired") else "" print(f" {s.get('slot_id')}\t{s.get('status')}\t{s.get('symbol')}\t" f"qty={s.get('qty')}\tsqty={s.get('slot_quantity')}\t" f"aqty={s.get('asset_quantity')}\t" f"limit={s.get('entry_price')}\tavg={s.get('avg_price')}\t" f"pid={pid}\tpaused={s.get('paused')}\twarned={s.get('warned')}{tag}") return 0 def _print_slot_detail(s: dict) -> None: """Verbose per-slot block used when `slots` is filtered (e.g. --symbol).""" wd = s.get("watchdog") or {} fields = [ ("status", s.get("status")), ("symbol", s.get("symbol")), ("portfolio_id", s.get("portfolio_id")), ("qty (authoritative)", s.get("qty")), ("slot_quantity (core mirror)", s.get("slot_quantity")), ("qty_source", s.get("qty_source")), ("broker_quantity_available", s.get("broker_quantity_available")), ("broker_quantity_captured_at", s.get("broker_quantity_captured_at")), ("asset_quantity (hub)", s.get("asset_quantity")), ("warned", s.get("warned")), ("avg_price", s.get("avg_price")), ("entry_price (limit)", s.get("entry_price")), ("cash_balance", s.get("cash_balance")), ("cash_estimated", s.get("cash_estimated")), ("paused", s.get("paused")), ("forced_close_at", s.get("forced_close_at")), ("universe_detached", s.get("universe_detached")), ("watchdog.expired", wd.get("expired")), ("watchdog.last_seen", wd.get("last_seen")), ] print(f"slot {s.get('slot_id')}") for label, value in fields: print(f" {label:<24}: {value}") def _intent_guard(args, cfg, summary: dict, action: str, confirm) -> int | None: """Shared enter/exit safeguard. Returns an exit code to short-circuit, or None to proceed with submission.""" broker, is_live = _broker_mode(cfg) summary = {**summary, "instance": _resolve_instance(args), "broker": broker, "live": is_live, "action": action} if args.dry_run: if args.json: _emit({**summary, "dry_run": True}, as_json=True) else: print(f"[dry-run] {action} {summary}") return 0 # live double-ack gate (OQ-2) if is_live and not args.live: msg = "live account requires --live (with --yes) to confirm a real order" _fail(msg, args.json) return 2 # confirmation: --json is non-interactive → --yes required if not args.yes: if args.json: _fail("--json enter/exit requires --yes (no auto-submit)", True) return 2 prompt = (f"Submit {action} on instance={summary['instance']} " f"broker={broker}{' LIVE' if is_live else ''}: {summary}?") if not confirm(prompt): print("aborted") return 1 return None def _cmd_enter(args, cfg, rpc, confirm) -> int: # REQ-032 AC-7 (CHG-20260610-06, issue #23): qty is required — reject before # RPC so a missing --qty fails immediately instead of admitting a zero-qty order # that the broker rejects silently after the fact. if not args.qty: _fail("--qty is required for enter (no quantity → broker would reject)", args.json) return 1 # REQ-032 AC-6: validate the absolute auto-close timestamp BEFORE submitting, # so a malformed value is rejected at the CLI instead of failing admission. forced_close_at = getattr(args, "forced_close_at", None) if forced_close_at is not None: try: forced_close_at = datetime.fromisoformat(forced_close_at).isoformat() except ValueError: _fail(f"invalid --forced-close-at (expected ISO8601): {args.forced_close_at!r}", args.json) return 1 summary = {"portfolio_id": args.portfolio_id, "qty": args.qty, "entry_price": args.entry_price, "forced_close_at": forced_close_at} short = _intent_guard(args, cfg, summary, "enter(buy)", confirm) if short is not None: return short payload = { "intent_type": "buy", "intent_id": str(uuid.uuid4()), "portfolio_id": args.portfolio_id, "qty": args.qty, "entry_price": args.entry_price, "strategy_id": args.strategy_id, "sell_strategy_id": args.sell_strategy_id, "stop_loss_price": args.stop_loss, "take_profit_price": args.take_profit, } # REQ-032 AC-5/6: only carry forced_close_at when specified (backward compat). if forced_close_at is not None: payload["forced_close_at"] = forced_close_at resp = rpc(topology.intent_queue(_resolve_instance(args)), payload, correlation_id=payload["intent_id"], timeout=args.timeout) return _report_intent(resp, args.json) def _cmd_exit(args, cfg, rpc, confirm) -> int: summary = {"slot_id": args.slot_id, "price": args.price or "market", "reason": args.reason} short = _intent_guard(args, cfg, summary, "exit(sell)", confirm) if short is not None: return short payload = { "intent_type": "sell", "intent_id": str(uuid.uuid4()), "slot_id": args.slot_id, "reason": args.reason, "sell_strategy_id": args.sell_strategy_id, "limit_price": args.price, } resp = rpc(topology.intent_queue(_resolve_instance(args)), payload, correlation_id=payload["intent_id"], timeout=args.timeout) return _report_intent(resp, args.json) def _cmd_cancel(args, cfg, rpc, confirm) -> int: """dcy-022 / REQ-071: cancel a pending order (operator backstop).""" summary = {"slot_id": args.slot_id} short = _intent_guard(args, cfg, summary, "cancel(pending)", confirm) if short is not None: return short payload = { "action": "cancel_pending", "slot_id": args.slot_id, "actor": "operator", } resp = rpc( topology.control_queue(_resolve_instance(args)), payload, correlation_id=str(uuid.uuid4()), timeout=args.timeout, ) if args.json: _emit(resp, as_json=True) return 0 if resp.get("status") == "ok" else 1 status = resp.get("status", "unknown") reason = resp.get("reason", "") if status == "ok": print(f"cancel sent: slot_id={resp.get('slot_id')} — execution loop will confirm on next poll") return 0 if status == "ignored": print(f"ignored: {reason} — slot is not in a pending state") return 0 _fail(f"cancel failed: {reason or status}", False) return 1 def _report_intent(resp: dict, as_json: bool) -> int: if as_json: _emit(resp, as_json=True) else: print(f"status={resp.get('status')} slot_id={resp.get('slot_id')} " f"reason={resp.get('reason')}") # admission rejection is a non-zero exit so scripts can detect it. return 0 if resp.get("status") == "admitted" else 1 def _fail(message: str, as_json: bool) -> None: if as_json: print(_json.dumps({"error": message}), file=sys.stderr) else: print(f"error: {message}", file=sys.stderr) def _broker_read_failure(exc, as_json: bool) -> int: """REQ-120 command boundary: expose code only, never upstream payloads.""" code = getattr(exc, "code", "broker_unavailable") payload = {"error": {"code": code, "message": "broker balance unavailable"}} if as_json: print(_json.dumps(payload), file=sys.stderr) else: print(f"broker balance unavailable ({code})", file=sys.stderr) return 1 class _CachedBalanceBroker: """One validated balance read per operator invocation; delegates everything else.""" def __init__(self, broker, balance) -> None: self._broker, self._balance = broker, balance def get_balance(self): return self._balance def __getattr__(self, name): return getattr(self._broker, name) def _parse_iso_date(raw: str) -> str: return datetime.fromisoformat(f"{raw}T00:00:00+00:00").date().isoformat() def _sanitize_path_component(value: str, *, limit: int = 80) -> str: cleaned = re.sub(r"[^A-Za-z0-9._-]+", "_", str(value or "").strip()).strip("._") return (cleaned or "value")[:limit] def _ledger_backfill_audit_path(args, execution_id: str) -> Path: base = Path(getattr(args, "audit_dir", _DEFAULT_LEDGER_BACKFILL_AUDIT_DIR)) stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%SZ") name = ( f"{stamp}--{_sanitize_path_component(_resolve_instance(args))}" f"--{_sanitize_path_component(execution_id)}.json" ) return base.resolve() / name def _write_ledger_backfill_audit(args, report: dict) -> str: path = _ledger_backfill_audit_path(args, report["request"]["execution_id"]) report["audit_path"] = str(path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(_json.dumps(report, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8") return str(path) def _ledger_skip_resolve_audit_path(args, skip_key: str) -> Path: base = Path(getattr(args, "audit_dir", _DEFAULT_LEDGER_SKIP_RESOLVE_AUDIT_DIR)) stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%SZ") name = ( f"{stamp}--{_sanitize_path_component(_resolve_instance(args))}" f"--{_sanitize_path_component(skip_key)}.json" ) return base.resolve() / name def _write_ledger_skip_resolve_audit(args, report: dict) -> str: path = _ledger_skip_resolve_audit_path(args, report["request"]["skip_key"]) report["audit_path"] = str(path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(_json.dumps(report, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8") return str(path) def _lookup_ledger_backfill_evidence(cfg: DaemonConfig, *, trading_date: str, execution_id: str): from trex_core.runtime.broker_provider import build_broker_provider broker, source = build_broker_provider(cfg).get_broker_for_read() lookup = getattr(broker, "lookup_fill_evidence", None) if not callable(lookup): raise TypeError("configured broker does not support KIS ledger backfill evidence lookup") evidence = lookup(trading_date=trading_date, execution_id=execution_id) return evidence, source def _resolve_ledger_backfill_slot(dsn: str, *, instance: str, symbol: str) -> dict: import psycopg with psycopg.connect(dsn) as conn, conn.cursor() as cur: cur.execute( "SELECT slot_id::text, portfolio_id, status " "FROM daily_slot " "WHERE instance_key = %s AND symbol = %s " "AND portfolio_id IS NOT NULL AND retired_at IS NULL " "ORDER BY updated_at DESC NULLS LAST, slot_id ASC " "LIMIT 2", (instance, symbol), ) rows = cur.fetchall() if not rows: raise RuntimeError(f"no daemon-owned slot found for instance={instance} symbol={symbol}") if len(rows) != 1: raise RuntimeError(f"ambiguous daemon-owned slot mapping for instance={instance} symbol={symbol}") return { "slot_id": str(rows[0][0]), "portfolio_id": str(rows[0][1]), "status": str(rows[0][2]), "symbol": symbol, } def _lookup_outbox_by_execution_id(dsn: str, *, execution_id: str) -> list[dict]: import psycopg with psycopg.connect(dsn) as conn, conn.cursor() as cur: cur.execute( "SELECT correlation_id, asset_id, order_id, execution_id, status, " "attempts, last_outcome, last_error " "FROM ledger_outbox WHERE execution_id = %s " "ORDER BY created_at, correlation_id", (execution_id,), ) rows = cur.fetchall() return [ { "correlation_id": str(cid), "asset_id": str(asset_id), "order_id": str(order_id), "execution_id": str(exec_id), "status": str(status), "attempts": int(attempts or 0), "last_outcome": str(last_outcome) if last_outcome else None, "last_error": str(last_error) if last_error else None, } for cid, asset_id, order_id, exec_id, status, attempts, last_outcome, last_error in rows ] def _normalize_existing_execution(existing_rows: list[dict], trade: LedgerTrade) -> dict | None: if not existing_rows: return None if len(existing_rows) != 1: raise RuntimeError(f"execution_id={trade.execution_id} already maps to multiple outbox rows") existing = existing_rows[0] if existing["correlation_id"] != trade.correlation_id: raise RuntimeError( f"execution_id={trade.execution_id} already belongs to {existing['correlation_id']}" ) return existing def _ledger_outcome_result(outcome: OutboxOutcome | None, existing_status: str | None = None) -> tuple[str, int]: if outcome is OutboxOutcome.COMPLETED: return "completed", 0 if outcome is OutboxOutcome.DUPLICATE_COMPLETED: return "duplicate_completed", 0 if outcome is OutboxOutcome.RETRYABLE_FAILURE: return "retryable_failure", 1 if outcome is OutboxOutcome.CONFLICT_BLOCKED: return "conflict_blocked", 1 if outcome in {OutboxOutcome.PERMANENT_FAILURE, OutboxOutcome.DEAD_LETTERED}: return "delivery_failed", 1 if existing_status == "completed": return "already_recorded", 0 if existing_status: return f"already_present:{existing_status}", 0 return "no_action", 1 # --------------------------------------------------------------------------- # # arg parsing # --------------------------------------------------------------------------- # def _add_common(p) -> None: p.add_argument("--config", help="per-instance TOML name (resolved within --config-path) or path") p.add_argument("--config-path", dest="config_path", default=None, help="config directory (default ./config) for --config + .env") p.add_argument("--instance", help="instance key (shortcut; overrides --config)") p.add_argument("--env-file", dest="env_file", default=None, help="explicit .env path; overrides /.env") p.add_argument("--json", action="store_true", help="machine-readable JSON output") p.add_argument("--timeout", type=float, default=_DEFAULT_TIMEOUT) def _add_intent_guards(p) -> None: p.add_argument("--yes", action="store_true", help="skip confirmation (paper)") p.add_argument("--live", action="store_true", help="confirm a LIVE order (with --yes)") p.add_argument("--dry-run", action="store_true", help="preview intent, do not submit") p.add_argument("--sell-strategy-id", dest="sell_strategy_id", default=None) def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="trex-core", description="trex-core admin CLI") parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") sub = parser.add_subparsers(dest="cmd", required=True) p_pause = sub.add_parser("pause", help="Pause a portfolio_id") p_pause.add_argument("portfolio_id") p_pause.add_argument("--actor", default="cli") p_resume = sub.add_parser("resume", help="Resume a portfolio_id") p_resume.add_argument("portfolio_id") sub.add_parser("paused", help="List paused portfolio_ids") p_uni = sub.add_parser("universe", help="List today's universe (snapshot RPC)") _add_common(p_uni) p_slots = sub.add_parser("slots", help="List positions (snapshot RPC)") p_slots.add_argument("--status", choices=["all", "held", "holding", "buyable", "pending"], default="all") p_slots.add_argument("--symbol", default=None, help="show only the slot(s) bound to this symbol") p_slots.add_argument("--slot-id", dest="slot_id", default=None, help="show only the slot with this slot_id") _add_common(p_slots) p_enter = sub.add_parser("enter", help="Submit a buy intent (intent RPC)") p_enter.add_argument("portfolio_id") p_enter.add_argument("--qty", type=float, default=None) p_enter.add_argument("--entry-price", dest="entry_price", type=float, default=None) p_enter.add_argument("--strategy-id", dest="strategy_id", default=None) p_enter.add_argument("--stop-loss", dest="stop_loss", type=float, default=None) p_enter.add_argument("--take-profit", dest="take_profit", type=float, default=None) p_enter.add_argument( "--forced-close-at", dest="forced_close_at", default=None, help="absolute auto-close timestamp (ISO8601, e.g. 2026-06-09T15:20:00+09:00) " "→ buy intent forced_close_at", ) _add_intent_guards(p_enter) _add_common(p_enter) p_exit = sub.add_parser("exit", help="Submit a sell intent (intent RPC)") p_exit.add_argument("slot_id") p_exit.add_argument("--price", type=float, default=None) p_exit.add_argument("--reason", default="operator_exit") _add_intent_guards(p_exit) _add_common(p_exit) p_cancel = sub.add_parser( "cancel", help="Cancel a pending order — operator backstop (control RPC)" ) p_cancel.add_argument("slot_id", help="slot_id of the pending_buy or pending_sell slot") _add_intent_guards(p_cancel) _add_common(p_cancel) # REQ-088 (dcy-029): operator report — holdings / today fills / realized PnL / # pending. Direct-DB (DB-backed point-in-time aggregate, no new RPC). p_report = sub.add_parser( "report", help="Holdings, today's fills, realized PnL, pending orders (direct-DB)" ) _add_common(p_report) # REQ-105 (dcy-034 / BAT-051): sync — slot qty ← broker (direct-DB). p_sync = sub.add_parser( "sync", help="Sync slot qty to broker ground truth (direct-DB; REQ-105)", ) target_grp = p_sync.add_mutually_exclusive_group() target_grp.add_argument( "--portfolio-id", dest="portfolio_id", default=None, help="target a single portfolio (slot must exist)", ) target_grp.add_argument( "--all", action="store_true", help="sync all warned + inactive slots for this instance", ) p_sync.add_argument("--yes", action="store_true", help="skip confirmation (paper)") p_sync.add_argument("--live", action="store_true", help="confirm a LIVE sync (with --yes)") p_sync.add_argument("--dry-run", dest="dry_run", action="store_true", help="preview before/after without writing") _add_common(p_sync) # REQ-106 (dcy-034 / BAT-051): verify — C-case re-check (read-centric, direct-DB). p_verify = sub.add_parser( "verify", help="Re-check C-case reconcile-warning (read-centric, direct-DB; REQ-106)", ) p_verify.add_argument( "--portfolio-id", dest="portfolio_id", default=None, help="target a single portfolio (omit for all C-warned slots)", ) _add_common(p_verify) p_unmanaged = sub.add_parser( "unmanaged", help="List or annotate durable unmanaged broker positions (direct-DB)", ) p_unmanaged_sub = p_unmanaged.add_subparsers(dest="unmanaged_cmd", required=True) p_unmanaged_list = p_unmanaged_sub.add_parser( "list", help="List unmanaged broker positions with lifecycle state", ) p_unmanaged_list.add_argument( "--status", choices=("active", "resolved", "all"), default="active", ) p_unmanaged_list.add_argument("--symbol", default=None) _add_common(p_unmanaged_list) p_unmanaged_audit = p_unmanaged_sub.add_parser( "audit", help="Show lifecycle/operator audit trail for one unmanaged position", ) p_unmanaged_audit.add_argument("symbol") _add_common(p_unmanaged_audit) p_unmanaged_record = p_unmanaged_sub.add_parser( "record", help="Record an explicit operator disposition for an unmanaged position", ) p_unmanaged_record.add_argument("symbol") p_unmanaged_record.add_argument( "action", choices=("investigating", "approved_exception", "managed_inclusion"), ) p_unmanaged_record.add_argument("--evidence-ref", required=True) p_unmanaged_record.add_argument("--note", default="") _add_common(p_unmanaged_record) p_ledger = sub.add_parser( "ledger", help="Ledger operator tools (direct-DB enqueue + immediate writer path)", ) p_ledger_sub = p_ledger.add_subparsers(dest="ledger_cmd", required=True) p_backfill = p_ledger_sub.add_parser( "backfill", help="Evidence-backed historical ledger backfill (KIS daily-ccld)", ) p_backfill.add_argument("--date", required=True, help="KST trading date (YYYY-MM-DD)") p_backfill.add_argument("--execution-id", dest="execution_id", required=True) p_backfill.add_argument("--yes", action="store_true", help="skip confirmation") p_backfill.add_argument("--live", action="store_true", help="confirm a LIVE ledger mutation (with --yes)") p_backfill.add_argument("--dry-run", dest="dry_run", action="store_true", help="preview the exact ledger trade, do not write") p_backfill.add_argument( "--audit-dir", default=_DEFAULT_LEDGER_BACKFILL_AUDIT_DIR, help="directory for persistent JSON audit reports", ) _add_common(p_backfill) p_skips = p_ledger_sub.add_parser( "skips", help="Query or resolve durable skipped-ledger records", ) p_skips_sub = p_skips.add_subparsers(dest="ledger_skips_cmd", required=True) p_skips_list = p_skips_sub.add_parser( "list", help="List skipped-ledger records by instance/date/portfolio/state", ) p_skips_list.add_argument("--date", default=None, help="UTC date filter (YYYY-MM-DD)") p_skips_list.add_argument("--portfolio-id", dest="portfolio_id", default=None) p_skips_list.add_argument( "--status", choices=("unresolved", "resolved", "all"), default="unresolved", ) p_skips_list.add_argument("--limit", type=int, default=100) _add_common(p_skips_list) p_skips_resolve = p_skips_sub.add_parser( "resolve", help="Mark one skipped-ledger record resolved and emit a resolved alert", ) p_skips_resolve.add_argument("--skip-key", required=True) p_skips_resolve.add_argument("--reason", default="operator_reconciled") p_skips_resolve.add_argument("--yes", action="store_true", help="skip confirmation") p_skips_resolve.add_argument("--live", action="store_true", help="confirm a LIVE ledger skip mutation (with --yes)") p_skips_resolve.add_argument( "--audit-dir", default=_DEFAULT_LEDGER_SKIP_RESOLVE_AUDIT_DIR, help="directory for persistent JSON audit reports", ) _add_common(p_skips_resolve) return parser def _cmd_sync(args, cfg, confirm) -> int: """REQ-105 (dcy-034, BAT-051): sync slot qty to broker ground truth.""" from trex_core.adapters.portfolio.reconcile_notification_store import ( PostgreSQLReconcileNotificationStore, ) from trex_core.adapters.portfolio.reconcile_warning_store import ( PostgreSQLReconcileWarningStore, ) from trex_core.application.reconciliation import sync_slot_to_broker if not args.portfolio_id and not args.all: _fail("--portfolio-id or --all is required", args.json) return 1 broker, is_live = _broker_mode(cfg) instance = _resolve_instance(args) dry_run = getattr(args, "dry_run", False) # live double-ack gate if not dry_run: if is_live and not getattr(args, "live", False): _fail("live account requires --live (with --yes) to confirm a real sync", args.json) return 2 if not getattr(args, "yes", False): if args.json: _fail("--json sync requires --yes (no auto-submit)", True) return 2 prompt = (f"Sync slot(s) to broker on instance={instance} " f"broker={broker}{' LIVE' if is_live else ''}?") if not confirm(prompt): print("aborted") return 1 # REQ-119/120: resolve and validate broker truth before opening/enumerating DB. from trex_core.application.broker_port import BrokerBalanceError from trex_core.runtime.broker_provider import ( BrokerCredentialUnavailable, build_broker_provider, ) try: broker_port, _source = build_broker_provider(cfg).get_broker_for_read() broker_port = _CachedBalanceBroker(broker_port, broker_port.get_balance()) except (BrokerCredentialUnavailable, BrokerBalanceError) as exc: return _broker_read_failure(exc, args.json) warn_store = PostgreSQLReconcileWarningStore(cfg.dsn) notification_store = PostgreSQLReconcileNotificationStore(cfg.dsn) # req-0007 / ac-0024: alarm emitter for resolved notifications. alarm_emitter_sync, _alarm_transport_sync = _default_alarm_emitter(cfg) # Build slot_repo + slot_service from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository from trex_core.domain.slot.service import SlotService slot_repo = PostgreSQLSlotRepository(cfg.dsn) slot_service = SlotService(slot_repo, instance) # Optional cost_basis_port (graceful degrade when not configured) cost_basis_port = None if cfg.portfolio_hub and cfg.portfolio_hub.base_url: from trex_core.adapters.portfolio.hub_read import PortfolioHubReadAdapter cost_basis_port = PortfolioHubReadAdapter( cfg.portfolio_hub.base_url, token=cfg.portfolio_hub.token ) # Determine target slot_ids import psycopg if args.portfolio_id: with psycopg.connect(cfg.dsn) as conn, conn.cursor() as cur: cur.execute( "SELECT slot_id FROM daily_slot " "WHERE instance_key = %s AND portfolio_id = %s LIMIT 1", (instance, args.portfolio_id), ) row = cur.fetchone() if row is None: _fail(f"no slot found for portfolio_id={args.portfolio_id} instance={instance}", args.json) return 1 slot_ids = [str(row[0])] else: # --all: target all warned slots (B case) plus all inactive slots with broker holding warned = warn_store.list_warned() with psycopg.connect(cfg.dsn) as conn, conn.cursor() as cur: cur.execute( "SELECT slot_id FROM daily_slot " "WHERE instance_key = %s AND status IN ('inactive', 'holding')", (instance,), ) all_ids = {str(r[0]) for r in cur.fetchall()} slot_ids = sorted(all_ids | warned) results = [] exit_code = 0 try: for sid in slot_ids: res = sync_slot_to_broker( slot_id=sid, slot_repo=slot_repo, slot_service=slot_service, broker=broker_port, instance_key=instance, warn_store=warn_store, cost_basis_port=cost_basis_port, dry_run=dry_run, alarm_emitter=alarm_emitter_sync, notification_store=notification_store, ) results.append(res) # A skipped result can still be a fail-closed live error (REQ-123). # Do not let the no-write marker hide a failed sync from automation. if res.get("error"): exit_code = 1 finally: _alarm_transport_sync.close() if args.json: print(_json.dumps({"instance": instance, "dry_run": dry_run, "results": results}, ensure_ascii=False, default=str)) else: for res in results: sid = res["slot_id"] sym = res.get("symbol", "") pid = res.get("portfolio_id", "") before = res.get("before", {}) after = res.get("after", {}) skipped = res.get("skipped", False) skip_reason = res.get("skip_reason", "") cleared = res.get("warning_cleared", False) seeded_avg_price = res.get("seeded_avg_price", False) cost_basis_source = res.get("cost_basis_source") cost_basis_reason = res.get("cost_basis_reason") prefix = "[dry-run] " if dry_run else "" if skipped: print(f" {prefix}SKIP slot={sid} symbol={sym} pid={pid} reason={skip_reason}") else: print(f" {prefix}SYNC slot={sid} symbol={sym} pid={pid}") print(f" before: status={before.get('status')} " f"qty={before.get('qty')} avg_price={before.get('avg_price')}") print(f" after: status={after.get('status')} " f"qty={after.get('qty')} avg_price={after.get('avg_price')}") if cleared: print(" reconcile-warning CLEARED") if seeded_avg_price is not None: print(f" cost_basis.seeded={seeded_avg_price}") if cost_basis_source or cost_basis_reason: print(f" cost_basis.source={cost_basis_source} reason={cost_basis_reason}") if res.get("error"): print(f" error={res.get('error')}") return exit_code def _cmd_verify(args, cfg) -> int: """REQ-106 (dcy-034, BAT-051): re-check C-case reconcile-warning (read-centric).""" from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository from trex_core.adapters.portfolio.reconcile_notification_store import ( PostgreSQLReconcileNotificationStore, ) from trex_core.adapters.portfolio.reconcile_warning_store import ( PostgreSQLReconcileWarningStore, ) from trex_core.application.portfolio import SnapshotStore from trex_core.application.reconciliation import verify_slot_reconcile instance = _resolve_instance(args) # REQ-120 no-write/no-enumeration barrier also applies to verify. from trex_core.application.broker_port import BrokerBalanceError from trex_core.runtime.broker_provider import ( BrokerCredentialUnavailable, build_broker_provider, ) try: broker_port, _source = build_broker_provider(cfg).get_broker_for_read() broker_port = _CachedBalanceBroker(broker_port, broker_port.get_balance()) except (BrokerCredentialUnavailable, BrokerBalanceError) as exc: return _broker_read_failure(exc, args.json) warn_store = PostgreSQLReconcileWarningStore(cfg.dsn) notification_store = PostgreSQLReconcileNotificationStore(cfg.dsn) # req-0007 / ac-0024: alarm emitter for resolved notifications. alarm_emitter_verify, _alarm_transport_verify = _default_alarm_emitter(cfg) slot_repo = PostgreSQLSlotRepository(cfg.dsn) # Build a minimal snapshot store (we only need asset_quantity for C-case check). # For verify we use a fresh SnapshotStore populated from warn_store metadata. # The hub_qty is read from the snapshot; if snapshot is empty we fall back to 0 # and the mismatch will surface as "rejected" (correct: re-check fails until # portfolio-hub CLI is run and snapshot is refreshed). This is acceptable for # operator CLI verify (not a real-time path). store = SnapshotStore() # Populate store from portfolio-hub if configured (best-effort). if cfg.portfolio_hub and cfg.portfolio_hub.base_url: try: from trex_core.adapters.portfolio.hub_read import PortfolioHubReadAdapter hub = PortfolioHubReadAdapter(cfg.portfolio_hub.base_url, token=cfg.portfolio_hub.token) daemon_id = hub.whoami() items = hub.get_portfolios(daemon_id) store.swap(items) except Exception: # noqa: BLE001 — degrade: verify with empty store _log.info("verify snapshot preload skipped", exc_info=True) if args.portfolio_id: target_pids = [args.portfolio_id] else: # All C-warned slots c_slots: list[str] = [] for sid in warn_store.list_warned(): info = warn_store.get_warning_info(sid) if not info: continue raw_codes = info.get("case_codes") if isinstance(raw_codes, list): case_codes = {str(code) for code in raw_codes if code} else: case_code = str(info.get("case_code", "")) case_codes = {case_code} if case_code else set() if "C" in case_codes: c_slots.append(info.get("portfolio_id", "")) target_pids = [p for p in c_slots if p] if not target_pids: msg = "no C-warned slots found" if args.json: print(_json.dumps({"instance": instance, "results": [], "message": msg})) else: print(msg) return 0 results = [] exit_code = 0 try: for pid in target_pids: res = verify_slot_reconcile( portfolio_id=pid, store=store, broker=broker_port, slot_repo=slot_repo, instance_key=instance, warn_store=warn_store, alarm_emitter=alarm_emitter_verify, notification_store=notification_store, ) results.append(res) if res["result"] == "rejected": exit_code = 1 finally: _alarm_transport_verify.close() if args.json: print(_json.dumps({"instance": instance, "results": results}, ensure_ascii=False, default=str)) else: for res in results: pid = res["portfolio_id"] sid = res.get("slot_id") result = res["result"] hub_qty = res.get("hub_qty") broker_qty = res.get("broker_qty") reason = res.get("reason", "") if result == "cleared": print(f" CLEARED portfolio_id={pid} slot={sid} " f"hub_qty={hub_qty} broker_qty={broker_qty}") elif result == "rejected": print(f" REJECTED portfolio_id={pid} slot={sid} — {reason}", file=sys.stderr) elif result == "not_c_case": print(f" INFO portfolio_id={pid} slot={sid} — {reason}") elif result == "not_warned": print(f" OK portfolio_id={pid} slot={sid} — already cleared") else: print(f" {result.upper()} portfolio_id={pid} slot={sid} — {reason}") return exit_code def _cmd_unmanaged_list(args, cfg) -> int: from trex_core.adapters.portfolio.unmanaged_position_store import ( PostgreSQLUnmanagedPositionStore, ) instance = _resolve_instance(args) states = PostgreSQLUnmanagedPositionStore(cfg.dsn).list_states( instance_key=instance, lifecycle_state=args.status, symbol=args.symbol, ) payload = { "instance_key": instance, "status": args.status, "count": len(states), "states": states, } if args.json: _emit(payload, as_json=True) else: print( f"instance={instance} status={args.status} count={len(states)} " f"symbol={args.symbol or '*'}" ) for state in states: print( f" {state['symbol']}\tqty={state['quantity']}\t" f"lifecycle={state['lifecycle_state']}\t" f"first_seen={state['first_seen_at']}\tlast_seen={state['last_seen_at']}\t" f"resolved_at={state['resolved_at']}\t" f"operator_action={state.get('operator_action') or '-'}\t" f"evidence_ref={state.get('evidence_ref') or '-'}" ) return 0 def _cmd_unmanaged_audit(args, cfg) -> int: from trex_core.adapters.portfolio.unmanaged_position_store import ( PostgreSQLUnmanagedPositionStore, ) instance = _resolve_instance(args) store = PostgreSQLUnmanagedPositionStore(cfg.dsn) state = store.get_state(instance_key=instance, symbol=args.symbol) if state is None: _fail( f"unmanaged position not found: instance={instance} symbol={args.symbol}", args.json, ) return 1 audit = store.list_audit(instance_key=instance, symbol=args.symbol) payload = { "instance_key": instance, "symbol": args.symbol, "state": state, "audit": audit, } if args.json: _emit(payload, as_json=True) else: print( f"instance={instance} symbol={args.symbol} lifecycle={state['lifecycle_state']} " f"entries={len(audit)}" ) for row in audit: print( f" {row['event_type']}\tat={row['recorded_at']}\t" f"quantity={row.get('quantity')}\taction={row.get('action') or '-'}\t" f"actor={row.get('actor') or '-'}\tevidence_ref={row.get('evidence_ref') or '-'}\t" f"note={row.get('note') or '-'}" ) return 0 def _cmd_unmanaged_record(args, cfg) -> int: from trex_core.adapters.portfolio.unmanaged_position_store import ( PostgreSQLUnmanagedPositionStore, ) instance = _resolve_instance(args) state = PostgreSQLUnmanagedPositionStore(cfg.dsn).record_operator_action( instance_key=instance, symbol=args.symbol, action=args.action, evidence_ref=args.evidence_ref, actor=_operator_actor(), note=args.note, ) if state is None: _fail( f"unmanaged position not found: instance={instance} symbol={args.symbol}", args.json, ) return 1 payload = {"result": "recorded", "state": state} if args.json: _emit(payload, as_json=True) else: print( f"recorded unmanaged position instance={instance} symbol={args.symbol} " f"action={args.action} evidence_ref={args.evidence_ref}" ) return 0 def _cmd_report(args, cfg) -> int: """REQ-088/110/111: holdings / today's fills / WAC realized PnL / pending — direct-DB.""" import psycopg from trex_core.application.operator_report import ( Holding, PendingOrder, TradeRow, apply_wac_to_report, build_report, kst_day_bounds_utc, render_json, render_text, ) from trex_core.application.portfolio import PortfolioHubUnavailableError start_utc, end_utc = kst_day_bounds_utc() holdings: list[Holding] = [] pending: list[PendingOrder] = [] trades: list[TradeRow] = [] with psycopg.connect(cfg.dsn) as conn, conn.cursor() as cur: cur.execute( "SELECT symbol, portfolio_id, qty, avg_price FROM daily_slot " "WHERE status = 'holding' ORDER BY symbol" ) for symbol, pid, qty, avg in cur.fetchall(): holdings.append(Holding(symbol or "", pid or "", float(qty or 0), float(avg or 0))) cur.execute( "SELECT symbol, portfolio_id, status FROM daily_slot " "WHERE status IN ('pending_buy','pending_sell') ORDER BY symbol" ) for symbol, pid, status in cur.fetchall(): pending.append(PendingOrder(symbol or "", pid or "", status)) cur.execute( "SELECT asset_id, side, qty, price, fee, tax FROM ledger_outbox " "WHERE COALESCE(executed_at, created_at) >= %s " "AND COALESCE(executed_at, created_at) < %s", (start_utc, end_utc), ) for asset_id, side, qty, price, fee, tax in cur.fetchall(): trades.append(TradeRow(asset_id, side, float(qty), float(price), float(fee or 0), float(tax or 0))) rep = build_report(holdings, pending, trades) # REQ-110: WAC 실현손익 통합 (portfolio-hub 설정 있으면 WAC 경로) # REQ-111: 미가용 시 degrade (폴백 없음) if cfg.portfolio_hub and cfg.portfolio_hub.base_url: try: from trex_core.adapters.portfolio.hub_read import PortfolioHubReadAdapter hub = PortfolioHubReadAdapter( cfg.portfolio_hub.base_url, token=cfg.portfolio_hub.token ) # OQ-2: 오늘(KST 일 경계) 기본값 _KST = timezone(timedelta(hours=9)) from_date = start_utc.astimezone(_KST).date() from_dt_str = from_date.isoformat() to_dt_str = from_date.isoformat() # 당일 # OQ-4: portfolio_id별 N콜 target_pids = ( {h.portfolio_id for h in holdings} | {p.portfolio_id for p in pending} ) wac_total = 0.0 wac_by_asset_accum: dict[str, float] = {} any_warning = False for pid in sorted(target_pids): result = hub.get_realized_pnl(pid, from_dt_str, to_dt_str) wac_total += result.total_realized_pnl if result.data_completeness_warning: any_warning = True for item in result.items: key = item.symbol or pid wac_by_asset_accum[key] = wac_by_asset_accum.get(key, 0.0) + item.realized_pnl apply_wac_to_report( rep, wac_total=wac_total, wac_by_asset=wac_by_asset_accum, data_completeness_warning=any_warning, ) except PortfolioHubUnavailableError as exc: rep.wac_unavailable = True rep.wac_unavailable_reason = ( f"portfolio-hub unavailable — realized PnL unavailable ({exc})" ) else: # REQ-111 AC-4: base_url 미설정도 동일 degrade rep.wac_unavailable = True rep.wac_unavailable_reason = ( "portfolio-hub unavailable — realized PnL unavailable (base_url not configured)" ) if getattr(args, "json", False): print(_json.dumps(render_json(rep), ensure_ascii=False)) else: print(render_text(rep)) return 0 def _render_ledger_skip_record(record) -> dict[str, object]: return ledger_skip_alarm_fields(record) def _cmd_ledger_skips_list(args, cfg) -> int: from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore instance = _resolve_instance(args) on_date = _parse_iso_date(args.date) if args.date else None records = PostgreSQLLedgerSkipStore(cfg.dsn).list_records( instance_key=instance, portfolio_id=args.portfolio_id, on_date=on_date, status=args.status, limit=args.limit, ) payload = { "instance": instance, "status": args.status, "count": len(records), "records": [_render_ledger_skip_record(record) for record in records], } if args.json: _emit(payload, as_json=True) else: print( f"instance={instance} status={args.status} count={len(records)} " f"portfolio_id={args.portfolio_id or '*'} date={args.date or '*'}" ) for record in records: print( f" {record.skip_key}\t{record.status}\t{record.skip_reason}\t" f"slot={record.slot_id}\tportfolio={record.portfolio_id or '-'}\t" f"execution={record.execution_id or '-'}" ) return 0 def _cmd_ledger_skips_resolve(args, cfg, confirm) -> int: from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore instance = _resolve_instance(args) broker_name, is_live = _broker_mode(cfg) report = { "request": { "instance": instance, "broker": broker_name, "live_account": is_live, "skip_key": args.skip_key, "resolution_reason": args.reason, }, "result": None, "record": None, "audit_path": None, } if is_live and (not getattr(args, "yes", False) or not getattr(args, "live", False)): report["result"] = {"status": "rejected", "reason": "live account requires --yes --live"} report["audit_path"] = _write_ledger_skip_resolve_audit(args, report) _fail("live account requires --yes --live", args.json) if args.json: _emit(report, as_json=True) else: _emit_pretty_json(report) return 2 if not getattr(args, "yes", False): if args.json: report["result"] = {"status": "rejected", "reason": "--json ledger skips resolve requires --yes"} report["audit_path"] = _write_ledger_skip_resolve_audit(args, report) _fail("--json ledger skips resolve requires --yes", True) _emit(report, as_json=True) return 2 prompt = ( f"Resolve skipped ledger record on instance={instance} broker={broker_name}" f"{' LIVE' if is_live else ''}: {args.skip_key}?" ) if not confirm(prompt): report["result"] = {"status": "aborted", "reason": "operator declined confirmation"} report["audit_path"] = _write_ledger_skip_resolve_audit(args, report) print("aborted") _emit_pretty_json(report) return 1 store = PostgreSQLLedgerSkipStore(cfg.dsn) record = store.resolve_by_key( args.skip_key, instance_key=instance, resolution_reason=args.reason, ) if record is None: report["result"] = {"status": "not_found", "reason": f"unresolved ledger skip not found: {args.skip_key}"} report["audit_path"] = _write_ledger_skip_resolve_audit(args, report) _fail(f"unresolved ledger skip not found: {args.skip_key}", args.json) if args.json: _emit(report, as_json=True) else: _emit_pretty_json(report) return 1 emitter, transport = _default_alarm_emitter(cfg) try: publish_ledger_skip_alarm(emitter, record, resolved=True) finally: transport.close() payload = {"result": "resolved", "record": _render_ledger_skip_record(record)} report["result"] = {"status": "resolved"} report["record"] = payload["record"] report["audit_path"] = _write_ledger_skip_resolve_audit(args, report) if args.json: _emit({**payload, "audit_path": report["audit_path"]}, as_json=True) else: print(f"resolved {record.skip_key} reason={record.resolution_reason}") return 0 def _cmd_ledger_backfill(args, cfg, confirm) -> int: from trex_core.adapters.broker.kis import KISEvidenceLookupError from trex_core.adapters.ledger.portfolio_hub import PortfolioHubLedger from trex_core.adapters.persistence.ledger_skip_store import PostgreSQLLedgerSkipStore from trex_core.adapters.persistence.postgresql import PostgreSQLLedgerOutbox from trex_core.runtime.broker_provider import BrokerCredentialUnavailable instance = _resolve_instance(args) broker_name, is_live = _broker_mode(cfg) if broker_name != "kis": _fail("ledger backfill is only supported with the KIS broker", args.json) return 1 try: trading_date = _parse_iso_date(args.date) except ValueError: _fail(f"invalid --date (expected YYYY-MM-DD): {args.date!r}", args.json) return 1 report = { "request": { "instance": instance, "date": trading_date, "execution_id": str(args.execution_id), "dry_run": bool(args.dry_run), "broker": broker_name, "live": bool(is_live), }, "evidence": None, "slot_binding": None, "trade": None, "outbox_precheck": None, "result": None, "resolved_skips": [], "audit_path": None, } try: evidence, broker_source = _lookup_ledger_backfill_evidence( cfg, trading_date=trading_date.replace("-", ""), execution_id=str(args.execution_id) ) report["broker_source"] = { "mode": broker_source.mode, "identity_label": broker_source.identity_label, } report["evidence"] = { "symbol": evidence["symbol"], "side": evidence["side"], "qty": evidence["qty"], "price": evidence["price"], "fee": evidence["fee"], "tax": evidence["tax"], "order_date": trading_date, "order_id": evidence["order_id"], "order_venue": evidence["order_venue"], "execution_venue": evidence["execution_venue"], "executed_at": evidence["executed_at"], "raw": evidence["raw"], } slot_binding = _resolve_ledger_backfill_slot( cfg.dsn, instance=instance, symbol=str(evidence["symbol"]) ) report["slot_binding"] = slot_binding trade = LedgerTrade( asset_id=str(slot_binding["portfolio_id"]), order_id=str(evidence["order_id"]), execution_id=str(args.execution_id), side=str(evidence["side"]), qty=float(evidence["qty"]), price=float(evidence["price"]), fee=float(evidence["fee"]), tax=float(evidence["tax"]), executed_at=evidence["executed_at"], venue=evidence["execution_venue"], order_venue=str(evidence["order_venue"]), ) report["trade"] = { "correlation_id": trade.correlation_id, "asset_id": trade.asset_id, "order_id": trade.order_id, "execution_id": trade.execution_id, "side": trade.side, "qty": trade.qty, "price": trade.price, "fee": trade.fee, "tax": trade.tax, "executed_at": trade.executed_at, "order_venue": trade.order_venue, "execution_venue": trade.venue, } existing = _normalize_existing_execution( _lookup_outbox_by_execution_id(cfg.dsn, execution_id=trade.execution_id), trade, ) report["outbox_precheck"] = { "execution_id": trade.execution_id, "existing": existing, } except (BrokerCredentialUnavailable, KISEvidenceLookupError, RuntimeError, TypeError) as exc: report["result"] = {"status": "rejected", "reason": str(exc)} report["audit_path"] = _write_ledger_backfill_audit(args, report) if args.json: _emit(report, as_json=True) else: _emit_pretty_json(report) return 1 if args.dry_run: report["result"] = { "status": "dry_run", "reason": "no writes performed", } report["audit_path"] = _write_ledger_backfill_audit(args, report) if args.json: _emit(report, as_json=True) else: _emit_pretty_json(report) return 0 if is_live and not getattr(args, "live", False): report["result"] = {"status": "rejected", "reason": "live account requires --live (with --yes)"} report["audit_path"] = _write_ledger_backfill_audit(args, report) _fail("live account requires --live (with --yes)", args.json) if args.json: _emit(report, as_json=True) else: _emit_pretty_json(report) return 2 if not getattr(args, "yes", False): if args.json: report["result"] = {"status": "rejected", "reason": "--json ledger backfill requires --yes"} report["audit_path"] = _write_ledger_backfill_audit(args, report) _fail("--json ledger backfill requires --yes", True) _emit(report, as_json=True) return 2 prompt = ( f"Backfill ledger trade on instance={instance} broker={broker_name}" f"{' LIVE' if is_live else ''}: {report['trade']}?" ) if not confirm(prompt): report["result"] = {"status": "aborted", "reason": "operator declined confirmation"} report["audit_path"] = _write_ledger_backfill_audit(args, report) print("aborted") _emit_pretty_json(report) return 1 outbox = PostgreSQLLedgerOutbox(cfg.dsn, instance_key=instance) if not cfg.portfolio_hub or not cfg.portfolio_hub.base_url: report["result"] = {"status": "rejected", "reason": "portfolio-hub ledger adapter is not configured"} report["audit_path"] = _write_ledger_backfill_audit(args, report) if args.json: _emit(report, as_json=True) else: _emit_pretty_json(report) return 1 writer = LedgerWriter( outbox, PortfolioHubLedger( cfg.portfolio_hub.base_url, token=cfg.portfolio_hub.token or None, timeout=cfg.portfolio_hub.timeout, ), ) existing = (report.get("outbox_precheck") or {}).get("existing") if existing is None: outbox.enqueue(trade) outcome = writer.process_correlation_id(trade.correlation_id) elif existing.get("status") == "pending": outcome = writer.process_correlation_id(trade.correlation_id) else: outcome = None item = outbox.get(trade.correlation_id) result_status, exit_code = _ledger_outcome_result( outcome, existing_status=str(existing["status"]) if isinstance(existing, dict) else None, ) report["result"] = { "status": result_status, "correlation_id": trade.correlation_id, "outcome": outcome.value if outcome is not None else None, "work_item_status": item.status.value if item is not None else None, "attempts": item.attempts if item is not None else None, "last_outcome": item.last_outcome.value if item and item.last_outcome else None, "last_error": item.last_error if item is not None else None, } if exit_code == 0: resolved = PostgreSQLLedgerSkipStore(cfg.dsn).resolve_for_fill( instance_key=instance, slot_id=uuid.UUID(str(slot_binding["slot_id"])), side=trade.side, order_id=trade.order_id, execution_id=trade.execution_id, resolution_reason="manual_ledger_backfill", ) report["resolved_skips"] = [_render_ledger_skip_record(record) for record in resolved] if resolved: emitter, transport = _default_alarm_emitter(cfg) try: for record in resolved: publish_ledger_skip_alarm(emitter, record, resolved=True) finally: transport.close() report["audit_path"] = _write_ledger_backfill_audit(args, report) if args.json: _emit(report, as_json=True) else: _emit_pretty_json(report) return exit_code def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int: """CLI entrypoint. `rpc` and `confirm` are injectable for tests: `rpc(queue, payload, *, correlation_id, timeout) -> dict`, `confirm(prompt) -> bool`. """ args = _build_parser().parse_args(argv) # REQ-010 CHG-20260609-01: load the config-dir .env (or --env-file) before # reading env. from_env() also runs the legacy root-.env fallback; the # config-dir load wins (loaded first, override=False). _toml = ( resolve_config_toml(args.config, getattr(args, "config_path", None)) if getattr(args, "config", None) else None ) _load_dotenv(_toml, getattr(args, "env_file", None)) cfg = ( DaemonConfig.from_config( resolve_config_toml(args.config, args.config_path), env_file=args.env_file ) if args.config else DaemonConfig.from_env() ) confirm = confirm or _default_confirm # REQ-088 (dcy-029): operator report — direct-DB point-in-time aggregate. if args.cmd == "report": return _cmd_report(args, cfg) # REQ-105 (BAT-051): sync — direct-DB, slot qty ← broker. if args.cmd == "sync": return _cmd_sync(args, cfg, confirm) # REQ-106 (BAT-051): verify — direct-DB, C-case re-check (read-centric). if args.cmd == "verify": return _cmd_verify(args, cfg) if args.cmd == "unmanaged" and getattr(args, "unmanaged_cmd", None) == "list": return _cmd_unmanaged_list(args, cfg) if args.cmd == "unmanaged" and getattr(args, "unmanaged_cmd", None) == "audit": return _cmd_unmanaged_audit(args, cfg) if args.cmd == "unmanaged" and getattr(args, "unmanaged_cmd", None) == "record": return _cmd_unmanaged_record(args, cfg) if args.cmd == "ledger" and getattr(args, "ledger_cmd", None) == "backfill": return _cmd_ledger_backfill(args, cfg, confirm) if ( args.cmd == "ledger" and getattr(args, "ledger_cmd", None) == "skips" and getattr(args, "ledger_skips_cmd", None) == "list" ): return _cmd_ledger_skips_list(args, cfg) if ( args.cmd == "ledger" and getattr(args, "ledger_cmd", None) == "skips" and getattr(args, "ledger_skips_cmd", None) == "resolve" ): return _cmd_ledger_skips_resolve(args, cfg, confirm) # direct-DB control commands (pause/resume stay direct-DB, OQ-3). if args.cmd in ("pause", "resume", "paused"): store = PostgreSQLPauseStore(cfg.dsn) if args.cmd == "pause": store.pause(args.portfolio_id, actor=args.actor) print(f"paused {args.portfolio_id}") elif args.cmd == "resume": store.resume(args.portfolio_id) print(f"resumed {args.portfolio_id}") else: for pid in sorted(store.list_paused()): print(pid) return 0 rpc = rpc or _default_rpc(cfg) try: if args.cmd == "universe": return _cmd_universe(args, cfg, rpc) if args.cmd == "slots": return _cmd_slots(args, cfg, rpc) if args.cmd == "enter": return _cmd_enter(args, cfg, rpc, confirm) if args.cmd == "exit": return _cmd_exit(args, cfg, rpc, confirm) if args.cmd == "cancel": return _cmd_cancel(args, cfg, rpc, confirm) except Exception as exc: # noqa: BLE001 - surface RPC/transport failure clearly _fail(f"{type(exc).__name__}: {exc}", getattr(args, "json", False)) return 3 return 0 if __name__ == "__main__": # pragma: no cover sys.exit(main())