#!/usr/bin/env python3 """One-way, fail-closed import of valid legacy local Paper state. The legacy ``hermes_durable.sqlite3`` file is never used for runtime authority. This tool imports inspectable records into the canonical relational schema. All legacy positions are placed in ``manual_review`` and all legacy plans are invalidated, so importing data cannot authorize an execution. """ from __future__ import annotations import argparse import json import os import sqlite3 import time import uuid from dataclasses import asdict, dataclass from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] import sys sys.path.insert(0, str(ROOT / "hermes_overlay")) from trading.domain.schema import apply_migrations, connect # noqa: E402 from trading.domain.audit_repo import append_audit_event # noqa: E402 from trading.plan_integrity import verify_plan_hash # noqa: E402 @dataclass class ImportReport: source: str dry_run: bool positions_valid: int = 0 positions_imported: int = 0 positions_skipped: int = 0 plans_valid: int = 0 plans_imported: int = 0 plans_skipped: int = 0 invalid_records: int = 0 def _positive(value: Any) -> float: number = float(value) if number <= 0: raise ValueError("expected positive number") return number def _valid_position(symbol: str, payload: Any) -> dict[str, Any]: if not isinstance(payload, dict): raise ValueError("legacy position payload must be an object") canonical_symbol = str(payload.get("symbol") or symbol).strip() side = str(payload.get("side") or "").strip().lower() if not canonical_symbol or side not in {"long", "short"}: raise ValueError("legacy position requires symbol and long/short side") contracts = _positive(payload.get("contracts", payload.get("size"))) base_quantity = _positive(payload.get("base_quantity", payload.get("size", contracts))) return { "symbol": canonical_symbol, "side": side, "leverage": max(1, int(payload.get("leverage") or 1)), "contracts": contracts, "base_quantity": base_quantity, "entry_price": _positive(payload.get("entry_price")), "stop_loss": float(payload["stop_loss"]) if payload.get("stop_loss") is not None else None, "take_profit": float(payload["take_profit"]) if payload.get("take_profit") is not None else None, "opened_at": float(payload.get("opened_at") or time.time()), "payload": payload, } def _read_legacy(source: Path) -> tuple[list[dict[str, Any]], list[dict[str, Any]], int]: if not source.is_file(): raise FileNotFoundError(source) positions: list[dict[str, Any]] = [] plans: list[dict[str, Any]] = [] invalid = 0 legacy = sqlite3.connect(f"file:{source}?mode=ro", uri=True) try: tables = {row[0] for row in legacy.execute( "SELECT name FROM sqlite_master WHERE type='table'" )} if not {"positions", "plans"} <= tables: raise ValueError("legacy snapshot is missing required positions/plans tables") for symbol, raw in legacy.execute("SELECT symbol,payload FROM positions"): try: positions.append(_valid_position(str(symbol), json.loads(raw))) except (TypeError, ValueError, json.JSONDecodeError): invalid += 1 for plan_id, raw, plan_hash, created_at, executed in legacy.execute( "SELECT plan_id,payload,plan_hash,created_at,executed FROM plans" ): try: payload = json.loads(raw) if not isinstance(payload, dict) or not verify_plan_hash(payload, str(plan_hash or "")): raise ValueError("invalid legacy plan hash") plans.append({ "plan_id": str(plan_id), "payload": payload, "plan_hash": str(plan_hash), "created_at": float(created_at), "executed": bool(executed), }) except (TypeError, ValueError, json.JSONDecodeError): invalid += 1 finally: legacy.close() return positions, plans, invalid def migrate_legacy_state(home: Path, *, dry_run: bool = True) -> ImportReport: mode = os.environ.get("TRADING_MODE", "paper").strip().lower() or "paper" if mode != "paper": raise RuntimeError("legacy state import is restricted to local Paper mode") home = home.resolve() source = home / "hermes_durable.sqlite3" positions, plans, invalid = _read_legacy(source) report = ImportReport( source=str(source), dry_run=dry_run, positions_valid=len(positions), plans_valid=len(plans), invalid_records=invalid, ) if dry_run: return report previous_home = os.environ.get("HERMES_HOME") os.environ["HERMES_HOME"] = str(home) try: ok, messages = apply_migrations() if not ok: raise RuntimeError("canonical migration failed: " + "; ".join(messages)) conn = connect() if conn is None: raise RuntimeError("canonical Paper database unavailable") now = time.time() try: conn.execute("BEGIN IMMEDIATE") for item in positions: exists = conn.execute( "SELECT 1 FROM positions WHERE account_id=? AND symbol=?", ("legacy-paper", item["symbol"]), ).fetchone() if exists: report.positions_skipped += 1 continue position_id = uuid.uuid4().hex conn.execute( "INSERT INTO positions(" "position_id,account_id,owner_id,symbol,side,leverage,contracts,base_quantity," "entry_price,stop_loss,take_profit,status,opened_at,updated_at,version" ") VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", ( position_id, "legacy-paper", "legacy-operator", item["symbol"], item["side"], item["leverage"], item["contracts"], item["base_quantity"], item["entry_price"], item["stop_loss"], item["take_profit"], "manual_review", item["opened_at"], now, 1, ), ) conn.execute( "INSERT INTO position_events(" "position_id,account_id,owner_id,symbol,event_type,from_status,to_status," "contracts_delta,price,actor,payload,created_at" ") VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", ( position_id, "legacy-paper", "legacy-operator", item["symbol"], "legacy_import", None, "manual_review", item["contracts"], item["entry_price"], "legacy-import-tool", json.dumps(item["payload"], sort_keys=True, default=str), now, ), ) report.positions_imported += 1 for item in plans: exists = conn.execute("SELECT 1 FROM plans WHERE plan_id=?", (item["plan_id"],)).fetchone() if exists: report.plans_skipped += 1 continue payload = dict(item["payload"]) conn.execute( "INSERT INTO plans(" "plan_id,symbol,payload,plan_hash,owner_id,version,created_at,expires_at,executed," "invalidated,created_by,updated_at,status,revision_reason,schema_version," "hash_schema_version,mode,environment,exchange_id,contract_type" ") VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", ( item["plan_id"], str(payload.get("symbol") or "UNKNOWN"), json.dumps(payload, sort_keys=True, default=str), item["plan_hash"], "legacy-operator", int(payload.get("version") or 1), item["created_at"], payload.get("expires_at") if isinstance(payload.get("expires_at"), (int, float)) else None, 1 if item["executed"] else 0, 1, "legacy-import-tool", now, "invalidated", "legacy_import_requires_reanalysis_and_reapproval", int(payload.get("plan_schema_version") or 1), int(payload.get("plan_hash_schema_version") or 1), "paper", "paper", "paper", "perpetual", ), ) report.plans_imported += 1 append_audit_event( "legacy_paper_import", asdict(report), owner_id="legacy-operator", account_id="legacy-paper", correlation_id=uuid.uuid4().hex, created_at=now, conn=conn, ) conn.commit() except Exception: conn.rollback() raise finally: conn.close() finally: if previous_home is None: os.environ.pop("HERMES_HOME", None) else: os.environ["HERMES_HOME"] = previous_home return report def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--home", type=Path, required=True) parser.add_argument("--apply", action="store_true", help="apply import; default is dry-run") args = parser.parse_args() report = migrate_legacy_state(args.home, dry_run=not args.apply) print(json.dumps(asdict(report), sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())