ai-exchange/scripts/run_replay_file.py

from pathlib import Path
from datetime import datetime, timezone
import json
import os
import sys
from typing import Optional

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from src.services.backtest import detect_confirmed_swings_for_replay, iter_historical_replay, load_replay_candles


def main(argv=None, env=None) -> int:
    argv = argv if argv is not None else sys.argv[1:]
    env = os.environ if env is None else env
    if not argv:
        print("Usage: python3 scripts/run_replay_file.py <candles.json|candles.jsonl|candles.csv>")
        return 2

    left_bars = _get_int(env, "REPLAY_LEFT_BARS", 2)
    right_bars = _get_int(env, "REPLAY_RIGHT_BARS", 2)
    min_candles = _get_int(env, "REPLAY_MIN_CANDLES", 0)
    min_swings = _get_int(env, "REPLAY_MIN_SWINGS", 0)
    max_gap_minutes = _get_float(env, "REPLAY_MAX_GAP_MINUTES", 0.0)
    json_output = _get_bool(env, "REPLAY_OUTPUT_JSON", False)
    try:
        candles = load_replay_candles(argv[0])
        frames = list(iter_historical_replay(candles, left_bars=left_bars, right_bars=right_bars))
        swings = detect_confirmed_swings_for_replay(candles, left_bars=left_bars, right_bars=right_bars)
    except (OSError, ValueError) as exc:
        print(f"Replay file invalid: {exc}")
        return 2

    gate_reasons = _evaluate_replay_gate(
        candle_count=len(candles),
        swing_count=len(swings),
        min_candles=min_candles,
        min_swings=min_swings,
        max_gap_minutes=max_gap_minutes,
        observed_max_gap_minutes=compute_max_gap_minutes(candles),
    )
    summary = _build_replay_summary(
        candles=candles,
        frame_count=len(frames),
        swings=swings,
        gate_reasons=gate_reasons,
        gate_config={"min_candles": min_candles, "min_swings": min_swings, "max_gap_minutes": max_gap_minutes},
    )
    if json_output:
        print(json.dumps(summary, ensure_ascii=False, sort_keys=True))
    else:
        _print_replay_summary(summary)
    return 0 if summary["quality_gate"]["passed"] else 2


def _get_int(env, key: str, default: int) -> int:
    value = env.get(key)
    if value is None or value == "":
        return default
    return int(value)


def _get_float(env, key: str, default: float) -> float:
    value = env.get(key)
    if value is None or value == "":
        return default
    return float(value)


def _get_bool(env, key: str, default: bool) -> bool:
    value = env.get(key)
    if value is None or value == "":
        return default
    return str(value).strip().lower() in {"1", "true", "yes", "on"}


def _evaluate_replay_gate(
    candle_count: int,
    swing_count: int,
    min_candles: int = 0,
    min_swings: int = 0,
    max_gap_minutes: float = 0.0,
    observed_max_gap_minutes: Optional[float] = None,
) -> list[str]:
    reasons = []
    if min_candles and candle_count < min_candles:
        reasons.append("candle_sample_too_small")
    if min_swings and swing_count < min_swings:
        reasons.append("swing_sample_too_small")
    if max_gap_minutes and (observed_max_gap_minutes is None or observed_max_gap_minutes > max_gap_minutes):
        reasons.append("candle_gap_too_large")
    return reasons


def _build_replay_summary(candles: list[dict], frame_count: int, swings: list, gate_reasons: list[str], gate_config: dict) -> dict:
    latest = swings[-1] if swings else None
    return {
        "candle_count": len(candles),
        "frame_count": frame_count,
        "sample": build_replay_sample_summary(candles),
        "swing_count": len(swings),
        "swing_counts": {
            "swing_high": sum(1 for swing in swings if swing.kind == "swing_high"),
            "swing_low": sum(1 for swing in swings if swing.kind == "swing_low"),
        },
        "latest_swing": {
            "kind": latest.kind,
            "price": latest.price,
            "source_index": latest.source_index,
            "confirmed_at_index": latest.confirmed_at_index,
        } if latest else None,
        "quality_gate": {
            "passed": len(gate_reasons) == 0,
            "reasons": gate_reasons,
            "gate": gate_config,
        },
    }


def build_replay_sample_summary(candles: list[dict]) -> dict:
    if not candles:
        return {
            "first_ts_open": None,
            "last_ts_open": None,
            "first_close": None,
            "last_close": None,
            "highest_high": None,
            "lowest_low": None,
            "close_return_pct": None,
            "max_gap_minutes": None,
        }
    first = candles[0]
    last = candles[-1]
    first_close = float(first["close"])
    last_close = float(last["close"])
    close_return_pct = ((last_close - first_close) / first_close) if first_close else None
    return {
        "first_ts_open": first.get("ts_open"),
        "last_ts_open": last.get("ts_open"),
        "first_close": first_close,
        "last_close": last_close,
        "highest_high": max(float(candle["high"]) for candle in candles),
        "lowest_low": min(float(candle["low"]) for candle in candles),
        "close_return_pct": close_return_pct,
        "max_gap_minutes": compute_max_gap_minutes(candles),
    }


def compute_max_gap_minutes(candles: list[dict]) -> Optional[float]:
    gaps = []
    previous = None
    for candle in candles:
        current = parse_timestamp(candle.get("ts_open"))
        if current is None:
            return None
        if previous is not None:
            gaps.append(max((current - previous).total_seconds() / 60, 0.0))
        previous = current
    return max(gaps) if gaps else 0.0


def parse_timestamp(value) -> Optional[datetime]:
    if value is None:
        return None
    if isinstance(value, datetime):
        if value.tzinfo is None:
            return value.replace(tzinfo=timezone.utc)
        return value.astimezone(timezone.utc)
    if isinstance(value, (int, float)):
        seconds = value / 1000 if value > 10_000_000_000 else value
        return datetime.fromtimestamp(seconds, tz=timezone.utc)
    if isinstance(value, str):
        normalized = value.strip().replace("Z", "+00:00")
        try:
            parsed = datetime.fromisoformat(normalized)
        except ValueError:
            return None
        if parsed.tzinfo is None:
            return parsed.replace(tzinfo=timezone.utc)
        return parsed.astimezone(timezone.utc)
    return None


def _print_replay_summary(summary: dict) -> None:
    print(f"Replay candles: {summary['candle_count']}")
    print(f"Replay frames: {summary['frame_count']}")
    sample = summary["sample"]
    if sample["first_ts_open"] is not None:
        print(
            f"Replay sample: first_ts={sample['first_ts_open']} last_ts={sample['last_ts_open']} "
            f"close_return_pct={sample['close_return_pct']}"
        )
        print(f"Replay max gap minutes: {sample['max_gap_minutes']}")
    print(f"Replay confirmed swings: {summary['swing_count']}")
    latest = summary["latest_swing"]
    if latest:
        print(
            f"Latest swing: {latest['kind']} price={latest['price']} "
            f"source_index={latest['source_index']} confirmed_at_index={latest['confirmed_at_index']}"
        )
    print(f"Replay quality gate passed: {summary['quality_gate']['passed']}")
    if summary["quality_gate"]["reasons"]:
        print(f"Replay quality gate reasons: {', '.join(summary['quality_gate']['reasons'])}")


if __name__ == "__main__":
    raise SystemExit(main())