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