scripts/migrate_legacy_paper_state.py

#!/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())