genichinandClaude Opus 4.8
f0ad37747b
BAT-035 (dcy-025): trex.snapshot.req → trex.slots.req 즉시 교체
- topology: SNAPSHOT_REQ_QUEUE 제거, SLOTS_REQ_QUEUE = "trex.slots.req" 추가
- portfolio.py: View/build_snapshot_envelope/SnapshotEnvelope/_asset_to_dict 제거;
slot_to_dict 시그니처 확장(snapshot_store/watchdog_store/pause_store/watchdog_timeout_sec)
— asset_quantity(hub)/cash_balance/cash_estimated/warned/watchdog{}/paused 병합 (REQ-074)
- messaging_handlers.py: make_snapshot_rpc_handler 제거, make_slots_rpc_handler 추가
— filter(all/inactive/pending/holding/error) + spy(bool) 처리; strategy 필드 제외 (REQ-073/074)
- daemon.py: _snapshot_rpc_loop → _slots_rpc_loop, make_slots_rpc_handler 연결
- cli.py: universe/slots → slots_queue + spy=True + server-side filter; buyable→inactive 매핑 (REQ-063)
- rabbitmq.py: snapshot_queue → slots_queue
- docs: interface-contract.md + runtime-flows.md slots.req 계약으로 업데이트 (REQ-075)
- REQ-028: Implemented → Deprecated (코드 삭제 완료)
- 테스트: test_snapshot_handler/selection/rpc_watchdog/messaging_snapshot_consumer 삭제;
test_slots_handler.py(신규, 13 cases), test_killswitch_masking_resume 재작성,
test_ledger_option_d._apply_cash_estimate → slots handler 경유, test_topology/cli_operator 갱신
BAT-036 (dcy-024): per-instance safety_sell_enabled 플래그
- config.py: DaemonInstanceConfig + DaemonConfig 에 safety_sell_enabled: bool = True 추가;
from_toml 파싱, from_config 전파
- daemon.py: start() workers 조건부 — safety_sell_enabled=false 시 trex-safety-sell 미기동 + WARNING
- tests: test_instance_config.py 에 safety_sell_enabled 파싱 3개 케이스 추가
REQ: 073/074/075/021/034/063/048/005 Approved → Implemented
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
454 lines
19 KiB
Python
454 lines
19 KiB
Python
"""trex-core operator CLI (REQ-032, dcy-016).
|
|
|
|
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)
|
|
|
|
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`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json as _json
|
|
import sys
|
|
import uuid
|
|
from datetime import datetime
|
|
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)
|
|
|
|
return parser
|
|
|
|
|
|
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
|
|
|
|
# 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())
|