inspect_results.py

#!/usr/bin/env python3
"""Inspect replay result folders and flag stale summary files."""

from __future__ import annotations

import argparse
import datetime as dt
import json
from pathlib import Path


CURRENT_SCHEMA_VERSION = 2
REQUIRED_COMPARISON_FIELDS = [
    "settlement_only_pnl_pct",
    "settlement_only_pnl_quote",
    "settlement_only_value_quote",
    "avellaneda_vs_settlement_only_pct",
    "avellaneda_vs_settlement_only_quote",
    "avellaneda_vs_settlement_only_result",
]


def load_summary(path: Path) -> dict:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise SystemExit(f"Could not parse {path}: {exc}") from exc


def add_derived_fields(summary: dict) -> bool:
    """Backfill fields that can be derived without rerunning the replay.

    This only handles arithmetic fields. If the summary is missing settlement
    values entirely, the replay must be rerun from candles + settlements.
    """

    changed = False
    initial_value = summary.get("initial_value_quote")
    settlement_pnl = summary.get("settlement_only_pnl_quote")
    final_value = summary.get("final_value_quote")
    settlement_value = summary.get("settlement_only_value_quote")

    if "settlement_only_pnl_pct" not in summary and initial_value and settlement_pnl is not None:
        summary["settlement_only_pnl_pct"] = settlement_pnl / initial_value * 100
        changed = True

    if "avellaneda_vs_settlement_only_pct" not in summary and initial_value and final_value is not None and settlement_value is not None:
        summary["avellaneda_vs_settlement_only_pct"] = (final_value - settlement_value) / initial_value * 100
        changed = True

    if "avellaneda_vs_settlement_only_quote" not in summary and final_value is not None and settlement_value is not None:
        summary["avellaneda_vs_settlement_only_quote"] = final_value - settlement_value
        changed = True

    if "avellaneda_vs_settlement_only_result" not in summary and final_value is not None and settlement_value is not None:
        summary["avellaneda_vs_settlement_only_result"] = "better" if final_value > settlement_value else "worse" if final_value < settlement_value else "same"
        changed = True

    if summary.get("summary_schema_version") != CURRENT_SCHEMA_VERSION:
        summary["summary_schema_version"] = CURRENT_SCHEMA_VERSION
        changed = True

    return changed


def status_for(summary: dict) -> tuple[str, list[str]]:
    missing = [field for field in REQUIRED_COMPARISON_FIELDS if field not in summary]
    if missing:
        return "OLD", missing
    if summary.get("summary_schema_version") != CURRENT_SCHEMA_VERSION:
        return "OLD", ["summary_schema_version"]
    return "OK", []


def format_number(value: object, digits: int = 4) -> str:
    return f"{value:.{digits}f}" if isinstance(value, int | float) else "-"


def format_span(summary: dict) -> str:
    start_value = summary.get("start")
    end_value = summary.get("end")
    if not start_value or not end_value:
        return "-"
    try:
        start = dt.datetime.fromisoformat(start_value.replace("Z", "+00:00"))
        end = dt.datetime.fromisoformat(end_value.replace("Z", "+00:00"))
    except ValueError:
        return f"{start_value} -> {end_value}"

    days = (end - start).total_seconds() / 86_400
    if start.date() == end.date():
        return f"{start.date()} ({days:.2f}d)"
    return f"{start.date()}..{end.date()} ({days:.2f}d)"


def solver_address_for(summary: dict) -> str:
    if summary.get("solver_address"):
        return summary["solver_address"]
    solver_addresses = summary.get("solver_addresses") or []
    if len(solver_addresses) == 1:
        return solver_addresses[0]
    fetch_metadata = summary.get("fetch_metadata") or {}
    return fetch_metadata.get("solver_address", "-")


def main() -> None:
    parser = argparse.ArgumentParser(description="List replay summaries and flag old result folders.")
    parser.add_argument("--results-dir", type=Path, default=Path("results"))
    parser.add_argument("--fix-derived", action="store_true", help="Backfill comparison fields that can be derived from existing summary values.")
    args = parser.parse_args()

    summaries = sorted(args.results_dir.glob("*/summary.json"))
    if not summaries:
        raise SystemExit(f"No summary.json files found under {args.results_dir}")

    for path in summaries:
        summary = load_summary(path)
        if args.fix_derived and add_derived_fields(summary):
            path.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
        status, missing = status_for(summary)
        result = summary.get("avellaneda_vs_settlement_only_result", "-")
        avellaneda_delta = format_number(summary.get("avellaneda_vs_settlement_only_pct"))
        reward_quote = format_number(summary.get("solver_reward_quote"))
        reward_budget_after = format_number(summary.get("reward_budget_after_avellaneda_quote"))
        settlement_pct = format_number(summary.get("settlement_only_pnl_pct"))
        strategy_pct = format_number(summary.get("pnl_pct"))
        span = format_span(summary)
        solver_address = solver_address_for(summary)
        suffix = f" missing={','.join(missing)}" if missing else ""
        print(
            f"{status} {path.parent} "
            f"span={span} "
            f"solver={solver_address} "
            f"settlement_only={settlement_pct}% "
            f"strategy={strategy_pct}% "
            f"avellaneda_vs_settlement={avellaneda_delta}% "
            f"solver_reward={reward_quote} "
            f"reward_budget_after_avellaneda={reward_budget_after} "
            f"result={result}{suffix}"
        )


if __name__ == "__main__":
    main()