copilot-swe-agent[bot]andgenichin
30bcbe1d12
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
1465 lines
59 KiB
Python
1465 lines
59 KiB
Python
"""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)
|
|
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 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.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
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 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 <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)
|
|
|
|
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_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
|
|
|
|
# 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)
|
|
|
|
# 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
|
|
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)
|
|
# 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
|
|
|
|
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_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)
|
|
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 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, 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)
|
|
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 == "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())
|