trex-core/trex_core/runtime/cli.py

"""trex-core operator CLI (REQ-032, dcy-016; REQ-105/106, dcy-034).

Subcommands:
  pause <portfolio_id>    — durably pause a portfolio (Postgres, direct-DB)
  resume <portfolio_id>   — 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 <portfolio_id>    — submit a buy intent (intent RPC; safeguarded)
  exit <slot_id>          — submit a sell intent (intent RPC; safeguarded)
  cancel <slot_id>        — 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)

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).
"""
from __future__ import annotations

import argparse
import json as _json
import sys
import uuid
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

from trex_core.adapters.messaging import topology
from trex_core.adapters.portfolio.pause_store import PostgreSQLPauseStore
from trex_core.runtime.config import (
    DaemonConfig,
    DaemonInstanceConfig,
    _load_dotenv,
    resolve_config_toml,
)

_DEFAULT_TIMEOUT = 10.0


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


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


# --------------------------------------------------------------------------- #
# 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')}\tqty(slot)={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')}\taqty={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 (slot mirror)", s.get("qty")),
        ("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)


# --------------------------------------------------------------------------- #
# 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 <config-dir>/.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")
    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)

    return parser


def _cmd_sync(args, cfg, confirm) -> int:
    """REQ-105 (dcy-034, BAT-051): sync slot qty to broker ground truth."""
    import psycopg

    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 <pid> 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

    warn_store = PostgreSQLReconcileWarningStore(cfg.dsn)

    # Build broker port
    from trex_core.runtime.daemon import _build_broker
    broker_port = _build_broker(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
    if args.portfolio_id:
        with psycopg.connect(cfg.dsn) as conn:
            with 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:
            with 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
    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,
        )
        results.append(res)
        if res.get("error") and not res.get("skipped"):
            exit_code = 1

    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)
            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(f"         reconcile-warning CLEARED")

    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_warning_store import (
        PostgreSQLReconcileWarningStore,
    )
    from trex_core.application.portfolio import SnapshotStore
    from trex_core.application.reconciliation import verify_slot_reconcile

    instance = _resolve_instance(args)
    warn_store = PostgreSQLReconcileWarningStore(cfg.dsn)
    slot_repo = PostgreSQLSlotRepository(cfg.dsn)

    from trex_core.runtime.daemon import _build_broker
    broker_port = _build_broker(cfg)

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

    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 info and info.get("case_code") == "C":
                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
    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,
        )
        results.append(res)
        if res["result"] == "rejected":
            exit_code = 1

    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_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:
        with 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 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_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)

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