trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(reset): cutover 도구 폐기 → trex-core-reset(DB 재초기화) 대체 (REQ-064 CHG-20260610-03)

Browse Source 
실 DB 점검 중 마이그레이션 009 부분 유니크 인덱스가 legacy 풀 데이터(같은
자산이 closed+재진입 holding 으로 portfolio_id 중복)에 적용 불가 → cutover·
데몬 부팅 모두 차단되는 결함 확인. 데이터 보존 cutover 를 폐기하고 DB wipe +
자산 중심 재초기화로 전환(사용자 결정). 빈 테이블에는 009 인덱스가 정상
생성되고 자산 중심에서 (instance_key, portfolio_id)는 설계상 유일.

- trex_core/runtime/reset.py 신규(trex-core-reset): slot-core 7테이블 wipe +
  apply_migrations 재적용 → 자산 중심 스키마 착지 검증. dry-run 기본/--yes 적용,
  파괴적·DB 전역. 보유 carry-over 없음(재초기화 후 reconcile_core_holdings 재편입).
- runtime/cutover.py + trex-core-cutover console script + test_cutover_rehearsal.py 삭제.
- pyproject: trex-core-reset 등록.
- tests/runtime/test_reset.py 신규 5건(PG-gated): wipe+착지·dry-run 무변경·멱등·
  wipe 후 admission·테이블 커버리지.
- change-req REQ-064 CHG-20260610-03(cutover→재초기화 재정의) + srs index 갱신.
- baseline 4문서(data-model/runtime-flows/project-structure/interface-contract)
  + BAT-030 design/planning/index/implementation/verification 도구 서술 reset 으로 교체.

검증: ruff clean, 단위 367 passed, PG+RabbitMQ 통합 411 passed/1 skip(KIS sandbox).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
genichin
2026-06-10 16:08:05 +09:00
co-authored by Claude Opus 4.8
parent 5330f32247
commit c04fd276b3
15 changed files with 335 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trex_core/runtime/cutover.py
-235
View File
@@ -1,235 +0,0 @@
"""Asset-centric cutover tool (dcy-020 / REQ-064, BAT-030).
Full rebuild (design D1): wipe this instance's daily_slot rows and recreate
them from the portfolio-hub universe + the broker balance, with the broker as
the single source of truth for carried-over holdings (consistent with D3).
Procedure (one transaction):
1. preflight — abort if any pending_buy/pending_sell rows exist for the
instance (in-flight orders must converge first; run outside the trading
session).
2. DELETE all of the instance's slot rows.
3. one permanent slot per universe asset, status `inactive`.
4. each broker holding (symbol, qty) is mapped symbol→portfolio_id via the
universe and its slot is set to `holding` (qty = broker quantity;
avg_price is left NULL — the balance API does not carry it — WARN).
5. a broker holding whose symbol is NOT in the universe aborts the cutover
(operator decision; `--allow-unmapped` downgrades to WARN+skip).
6. audit rows in slot_transition_log (reason `cutover_rebuild`).
7. optional `--promote-unique`: portfolio_id SET NOT NULL (only after ALL
instances sharing the DB have been cut over).
The real cutover is OPERATOR-RUN (design D4): `trex-core-cutover --config
<instance> --yes` after a rehearsal against a test DB. Rollback = restore the
pre-cutover backup + previous release (BAT-030 design §4).
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Iterable, Optional
from uuid import uuid4
_log = logging.getLogger(__name__)
class CutoverAbort(RuntimeError):
"""Preflight/mapping condition that must stop the cutover."""
@dataclass
class CutoverReport:
instance_key: str
deleted: int = 0
created: int = 0
holdings: int = 0
unmapped: list[str] = field(default_factory=list)
promoted_unique: bool = False
dry_run: bool = False
def summary(self) -> str:
return (
f"cutover[{self.instance_key}] deleted={self.deleted} "
f"created={self.created} holdings={self.holdings} "
f"unmapped={self.unmapped or '[]'} "
f"promote_unique={self.promoted_unique} dry_run={self.dry_run}"
)
def run_cutover(
dsn: str,
instance_key: str,
universe: Iterable[tuple[str, str]],
balances: Iterable[tuple[str, int]],
*,
allow_unmapped: bool = False,
promote_unique: bool = False,
dry_run: bool = False,
psycopg_module: Any = None,
) -> CutoverReport:
"""Rebuild the instance's slots from `universe` + `balances` atomically.
universe: iterable of (portfolio_id, symbol).
balances: iterable of (symbol, hldg_qty) — broker ground truth.
Raises CutoverAbort on in-flight pending rows or unmapped broker holdings
(unless allow_unmapped). dry_run rolls the transaction back at the end.
"""
if psycopg_module is None:
import psycopg # type: ignore[import-not-found]
psycopg_module = psycopg
uni = [(str(pid), str(sym)) for pid, sym in universe if pid]
by_symbol = {sym: pid for pid, sym in uni}
held = [(str(sym), int(qty)) for sym, qty in balances if int(qty) > 0]
report = CutoverReport(instance_key=instance_key, dry_run=dry_run)
# mapping check before touching the DB (step 5)
unmapped = [sym for sym, _ in held if sym not in by_symbol]
if unmapped:
if not allow_unmapped:
raise CutoverAbort(
f"broker holds symbols outside the universe: {unmapped} "
"(resolve in portfolio-hub or pass --allow-unmapped to skip them)"
)
report.unmapped = unmapped
for sym in unmapped:
_log.warning(
"cutover: broker holding %s has no universe portfolio — SKIPPED "
"(unmanaged position; operator attention)", sym,
)
now = datetime.now(tz=timezone.utc)
with psycopg_module.connect(dsn, autocommit=False) as conn:
try:
with conn.cursor() as cur:
# 1. preflight: no in-flight pending rows for this instance
cur.execute(
"SELECT count(*) FROM daily_slot WHERE instance_key = %s "
"AND status IN ('pending_buy','pending_sell')",
(instance_key,),
)
pending = int(cur.fetchone()[0])
if pending:
raise CutoverAbort(
f"{pending} in-flight pending slot(s) for instance "
f"{instance_key!r} — wait for convergence (run outside "
"the trading session) and retry"
)
# 2. wipe the instance's rows (full rebuild, D1)
cur.execute(
"DELETE FROM daily_slot WHERE instance_key = %s", (instance_key,)
)
report.deleted = cur.rowcount
# 3+4. recreate: one permanent slot per universe asset
holding_pids = {}
for sym, qty in held:
pid = by_symbol.get(sym)
if pid is not None:
holding_pids[pid] = (sym, qty)
for pid, sym in uni:
sid = uuid4()
if pid in holding_pids:
h_sym, h_qty = holding_pids[pid]
cur.execute(
"INSERT INTO daily_slot (slot_id, status, symbol, qty, "
"instance_key, portfolio_id, created_at, updated_at) "
"VALUES (%s, 'holding', %s, %s, %s, %s, %s, %s)",
(sid, h_sym, h_qty, instance_key, pid, now, now),
)
report.holdings += 1
_log.warning(
"cutover: %s carried over as holding qty=%d "
"avg_price=NULL (balance API has no cost basis)",
pid, h_qty,
)
else:
cur.execute(
"INSERT INTO daily_slot (slot_id, status, symbol, "
"instance_key, portfolio_id, created_at, updated_at) "
"VALUES (%s, 'inactive', %s, %s, %s, %s, %s)",
(sid, sym, instance_key, pid, now, now),
)
report.created += 1
# 6. audit trail
cur.execute(
"INSERT INTO slot_transition_log "
"(slot_id, from_state, to_state, ts, reason, actor) "
"VALUES (%s, 'inactive', %s, %s, 'cutover_rebuild', 'cutover')",
(sid, "holding" if pid in holding_pids else "inactive", now),
)
# 7. optional promotion (all instances cut over)
if promote_unique:
cur.execute(
"ALTER TABLE daily_slot ALTER COLUMN portfolio_id SET NOT NULL"
)
report.promoted_unique = True
if dry_run:
conn.rollback()
else:
conn.commit()
except BaseException:
conn.rollback()
raise
return report
def main(argv: Optional[list[str]] = None) -> int: # pragma: no cover - CLI shell
import argparse
from trex_core.adapters.persistence.migrate import apply_migrations
from trex_core.runtime.config import DaemonConfig, resolve_config_toml
from trex_core.runtime.daemon import _build_broker, _build_portfolio_read
logging.basicConfig(level=logging.INFO)
parser = argparse.ArgumentParser(
prog="trex-core-cutover",
description="dcy-020/REQ-064: pool→asset-centric full-rebuild cutover "
"(operator-run; rehearse on a test DB first).",
)
parser.add_argument("--config", required=True, help="instance TOML name/path")
parser.add_argument("--config-path", default=None)
parser.add_argument("--env-file", default=None)
parser.add_argument("--yes", action="store_true",
help="actually apply (otherwise dry-run + rollback)")
parser.add_argument("--allow-unmapped", action="store_true",
help="skip (WARN) broker holdings outside the universe "
"instead of aborting")
parser.add_argument("--promote-unique", action="store_true",
help="portfolio_id SET NOT NULL (only after ALL "
"instances sharing this DB are cut over)")
args = parser.parse_args(argv)
toml_path = resolve_config_toml(args.config, args.config_path)
config = DaemonConfig.from_config(toml_path, env_file=args.env_file)
apply_migrations(config.dsn)
read = _build_portfolio_read(config)
daemon_id = read.whoami()
items = read.get_portfolios(daemon_id)
universe = [(it.portfolio_id, it.symbol) for it in items]
broker = _build_broker(config)
balances = [(b.symbol, b.hldg_qty) for b in broker.get_balance()]
try:
report = run_cutover(
config.dsn, config.instance_key, universe, balances,
allow_unmapped=args.allow_unmapped,
promote_unique=args.promote_unique and args.yes,
dry_run=not args.yes,
)
except CutoverAbort as exc:
_log.error("cutover ABORTED: %s", exc)
return 2
_log.info("%s", report.summary())
if not args.yes:
_log.info("dry-run only — re-run with --yes to apply")
return 0
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())
trex_core/runtime/reset.py
+156
View File
@@ -0,0 +1,156 @@
"""Asset-centric DB (re)initialization tool (dcy-020 / REQ-064, BAT-030).
Replaces the data-preserving pool→permanent cutover: instead of migrating
legacy pool rows, this **wipes the slot-core tables and (re)applies the
migrations** so the database lands on the clean asset-centric schema. Use it
to stand up a new instance's database or to reset a dev / superseded
pool-model database to the asset-centric substrate.
Why wipe instead of migrate (the decision that retired the cutover):
migration 009's partial `UNIQUE(instance_key, portfolio_id) WHERE portfolio_id
IS NOT NULL` cannot be applied to legacy pool data, because in the pool model
the same asset legitimately occupied more than one slot (a `closed` row from a
prior cycle plus a re-entered `holding` row) — so `portfolio_id` repeats and
the unique index collides. On an **empty** table the index applies cleanly and
`(instance_key, portfolio_id)` is unique by construction thereafter.
After reset, the daemon's universe reconcile (REQ-058) builds the permanent
asset slots (`inactive`); broker holdings surface through
`reconcile_core_holdings` (REQ-062) for operator adoption.
Destructive and DB-wide (every instance's rows). OPERATOR-RUN, dry-run by
default; pass `--yes` to apply. Take a backup first.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from typing import Any, Optional
_log = logging.getLogger(__name__)
# slot-core tables owned by trex-core migrations (drop order irrelevant — no FKs).
SLOT_CORE_TABLES: tuple[str, ...] = (
"daily_slot",
"daily_symbol_entries",
"slot_transition_log",
"intent_log",
"ledger_outbox",
"snapshot_pause",
"killswitch_exclusion",
)
@dataclass
class ResetReport:
db: str
wiped: dict[str, int] = field(default_factory=dict) # table -> rows removed
migrations_applied: list[str] = field(default_factory=list)
schema_ok: bool = False
dry_run: bool = True
def summary(self) -> str:
total = sum(self.wiped.values())
verb = "would wipe" if self.dry_run else "wiped"
return (
f"reset[{self.db}] {verb} {total} rows across {len(self.wiped)} tables "
f"{self.wiped} | migrations={len(self.migrations_applied)} "
f"schema_ok={self.schema_ok} dry_run={self.dry_run}"
)
def _existing_tables(cur) -> set[str]:
cur.execute(
"SELECT table_name FROM information_schema.tables "
"WHERE table_schema = 'public' AND table_name = ANY(%s)",
(list(SLOT_CORE_TABLES),),
)
return {r[0] for r in cur.fetchall()}
def run_reset(
dsn: str,
*,
dry_run: bool = True,
psycopg_module: Any = None,
) -> ResetReport:
"""Wipe the slot-core tables and reapply migrations to the asset-centric schema.
dry_run (default): report the row counts that WOULD be wiped, change nothing.
apply (dry_run=False): TRUNCATE existing slot-core tables, then
`apply_migrations(dsn)` so the schema is the clean asset-centric form, then
verify `daily_slot` columns == `EXPECTED_SLOT_FIELDS`.
"""
if psycopg_module is None:
import psycopg # type: ignore[import-not-found]
psycopg_module = psycopg
with psycopg_module.connect(dsn, autocommit=True) as conn:
with conn.cursor() as cur:
cur.execute("SELECT current_database()")
report = ResetReport(db=cur.fetchone()[0], dry_run=dry_run)
existing = _existing_tables(cur)
for t in SLOT_CORE_TABLES:
if t in existing:
cur.execute(f"SELECT count(*) FROM {t}")
report.wiped[t] = int(cur.fetchone()[0])
if dry_run:
return report
if existing:
cur.execute(
f"TRUNCATE {', '.join(existing)} RESTART IDENTITY CASCADE"
)
# Reapply migrations on the now-empty DB: 009's unique index applies cleanly,
# 010 sets the inactive CHECK, 011 drops the pool-model columns.
from trex_core.adapters.persistence.migrate import apply_migrations
report.migrations_applied = apply_migrations(dsn, psycopg_module=psycopg_module)
from trex_core.domain.slot import EXPECTED_SLOT_FIELDS
with psycopg_module.connect(dsn) as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT column_name FROM information_schema.columns "
"WHERE table_name = 'daily_slot'"
)
cols = {r[0] for r in cur.fetchall()}
report.schema_ok = cols == set(EXPECTED_SLOT_FIELDS)
return report
def main(argv: Optional[list[str]] = None) -> int: # pragma: no cover - CLI shell
import argparse
from trex_core.runtime.config import DaemonConfig, resolve_config_toml
logging.basicConfig(level=logging.INFO)
parser = argparse.ArgumentParser(
prog="trex-core-reset",
description="dcy-020/REQ-064: wipe the slot-core tables and reinitialize "
"the DB to the asset-centric schema (dry-run by default). "
"Destructive and DB-wide — back up first.",
)
parser.add_argument("--config", required=True, help="instance TOML name/path (DSN source)")
parser.add_argument("--config-path", default=None)
parser.add_argument("--env-file", default=None)
parser.add_argument("--yes", action="store_true",
help="actually wipe + reinitialize (otherwise dry-run)")
args = parser.parse_args(argv)
toml_path = resolve_config_toml(args.config, args.config_path)
config = DaemonConfig.from_config(toml_path, env_file=args.env_file)
report = run_reset(config.dsn, dry_run=not args.yes)
_log.warning("%s", report.summary())
if not args.yes:
_log.info("dry-run only — re-run with --yes to wipe and reinitialize")
elif not report.schema_ok:
_log.error("schema verification FAILED after reset — inspect manually")
return 1
return 0
if __name__ == "__main__": # pragma: no cover
raise SystemExit(main())