"""Standalone CUDA Batch01 band/channel prototype.

This module deliberately has no Artifex imports.  It consumes frozen oracle
artifacts and OHLCV CSV files mounted read-only, writes only its output cache,
and records measured parity rather than asserting it.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import os
import time
from pathlib import Path
from typing import Any

import numpy as np
import torch


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def _load_ohlcv(path: Path, device: torch.device) -> tuple[torch.Tensor, ...]:
    with path.open(newline="", encoding="utf-8") as handle:
        rows = list(csv.DictReader(handle))
    if not rows or not {"close", "high", "low", "volume"}.issubset(rows[0]):
        raise ValueError("CSV must contain close, high, low, and volume columns")
    return tuple(
        torch.as_tensor([float(row[name]) for row in rows], dtype=torch.float64, device=device)
        for name in ("close", "high", "low", "volume")
    )


def _rolling_windows(values: torch.Tensor, period: int) -> torch.Tensor:
    return values.unfold(0, period, 1)


def _empty(values: torch.Tensor) -> torch.Tensor:
    return torch.full_like(values, float("nan"))


def _sma(values: torch.Tensor, period: int) -> torch.Tensor:
    out = _empty(values)
    if values.numel() < period:
        return out
    out[period - 1 :] = _rolling_windows(values, period).mean(dim=1)
    return out


def _ema(values: torch.Tensor, period: int) -> torch.Tensor:
    out = _empty(values)
    if values.numel() < period:
        return out
    alpha = 2.0 / (period + 1)
    out[period - 1] = values[:period].sum() / period
    for index in range(period, values.numel()):
        out[index] = alpha * values[index] + (1.0 - alpha) * out[index - 1]
    return out


def _rma(values: torch.Tensor, period: int) -> torch.Tensor:
    out = _empty(values)
    valid = torch.nonzero(~torch.isnan(values), as_tuple=False).flatten()
    if valid.numel() < period:
        return out
    start = int(valid[period - 1].item())
    out[start] = values[valid[:period]].sum() / period
    alpha = 1.0 / period
    for index in range(start + 1, values.numel()):
        out[index] = out[index - 1] if torch.isnan(values[index]) else alpha * values[index] + (1.0 - alpha) * out[index - 1]
    return out


def _true_range(close: torch.Tensor, high: torch.Tensor, low: torch.Tensor) -> torch.Tensor:
    out = torch.empty_like(close)
    out[0] = high[0] - low[0]
    out[1:] = torch.stack((high[1:] - low[1:], (high[1:] - close[:-1]).abs(), (low[1:] - close[:-1]).abs())).amax(dim=0)
    return out


def _bollinger(close: torch.Tensor, period: int, multipliers: list[float]) -> tuple[torch.Tensor, torch.Tensor]:
    upper, lower = _empty(close).repeat(len(multipliers), 1), _empty(close).repeat(len(multipliers), 1)
    if close.numel() < period:
        return upper, lower
    windows = _rolling_windows(close, period)
    mean = windows.mean(dim=1)
    # Keep reductions on-device and vectorized; sample variance matches the oracle formula.
    standard_deviation = ((windows - mean[:, None]).square().sum(dim=1) / (period - 1)).sqrt()
    multiplier = torch.tensor(multipliers, dtype=torch.float64, device=close.device)
    upper[:, period - 1 :] = mean[None, :] + multiplier[:, None] * standard_deviation[None, :]
    lower[:, period - 1 :] = mean[None, :] - multiplier[:, None] * standard_deviation[None, :]
    return upper, lower


def _donchian(values: torch.Tensor, period: int, upper: bool) -> torch.Tensor:
    out = _empty(values)
    if values.numel() >= period:
        windows = _rolling_windows(values, period)
        out[period - 1 :] = windows.amax(dim=1) if upper else windows.amin(dim=1)
    return out


def _keltner(close: torch.Tensor, high: torch.Tensor, low: torch.Tensor, period: int, multipliers: list[float]) -> tuple[torch.Tensor, torch.Tensor]:
    basis, atr = _ema(close, period), _rma(_true_range(close, high, low), period)
    upper, lower = _empty(close).repeat(len(multipliers), 1), _empty(close).repeat(len(multipliers), 1)
    valid = ~torch.isnan(basis) & ~torch.isnan(atr)
    multiplier = torch.tensor(multipliers, dtype=torch.float64, device=close.device)
    upper[:, valid] = basis[valid] + multiplier[:, None] * atr[valid]
    lower[:, valid] = basis[valid] - multiplier[:, None] * atr[valid]
    return upper, lower


def _supertrend(close: torch.Tensor, high: torch.Tensor, low: torch.Tensor, period: int, multipliers: list[float]) -> torch.Tensor:
    count, length = len(multipliers), close.numel()
    out = _empty(close).repeat(count, 1)
    atr, upper, lower = _rma(_true_range(close, high, low), period), _empty(close).repeat(count), _empty(close).repeat(count)
    multiplier = torch.tensor(multipliers, dtype=torch.float64, device=close.device)
    direction = torch.ones(count, dtype=torch.float64, device=close.device)
    for index in range(period, length):
        midpoint, basic_upper, basic_lower = (high[index] + low[index]) / 2.0, None, None
        basic_upper, basic_lower = midpoint + multiplier * atr[index], midpoint - multiplier * atr[index]
        if index == period:
            upper, lower = basic_upper, basic_lower
            out[:, index] = torch.where(close[index] > lower, lower, upper)
            direction = torch.where(close[index] > out[:, index], 1.0, -1.0)
            continue
        upper = torch.where((close[index - 1] <= upper) & ~torch.isnan(upper), torch.minimum(basic_upper, upper), basic_upper)
        lower = torch.where((close[index - 1] >= lower) & ~torch.isnan(lower), torch.maximum(basic_lower, lower), basic_lower)
        out[:, index] = torch.where(direction == 1.0, torch.where(close[index] >= lower, lower, upper), torch.where(close[index] <= upper, upper, lower))
        direction = torch.where(close[index] > out[:, index], 1.0, -1.0)
    return out


def _psar(close: torch.Tensor, high: torch.Tensor, low: torch.Tensor, steps: list[float]) -> torch.Tensor:
    out = _empty(close).repeat(len(steps), 1)
    if close.numel() < 2:
        return out
    step = torch.tensor(steps, dtype=torch.float64, device=close.device)
    bullish = torch.ones(len(steps), dtype=torch.bool, device=close.device)
    acceleration = torch.full_like(step, 0.02)
    extreme, sar = high[0].repeat(len(steps)), low[0].repeat(len(steps))
    for index in range(1, close.numel()):
        sar = sar + acceleration * (extreme - sar)
        bull_sar = torch.minimum(sar, low[index - 1])
        bear_sar = torch.maximum(sar, high[index - 1])
        if index >= 2:
            bull_sar, bear_sar = torch.minimum(bull_sar, low[index - 2]), torch.maximum(bear_sar, high[index - 2])
        reversal_down, reversal_up = bullish & (low[index] < bull_sar), ~bullish & (high[index] > bear_sar)
        candidate = torch.where(bullish, bull_sar, bear_sar)
        sar = torch.where(reversal_down | reversal_up, extreme, candidate)
        new_bullish = torch.where(reversal_down, False, torch.where(reversal_up, True, bullish))
        new_extreme = torch.where(reversal_down, low[index], torch.where(reversal_up, high[index], extreme))
        rising = new_bullish & ~reversal_up & (high[index] > extreme)
        falling = ~new_bullish & ~reversal_down & (low[index] < extreme)
        new_extreme = torch.where(rising, high[index], torch.where(falling, low[index], new_extreme))
        acceleration = torch.where(reversal_down | reversal_up, 0.02, torch.where(rising | falling, torch.minimum(acceleration + step, torch.full_like(step, 0.2)), acceleration))
        bullish, extreme = new_bullish, new_extreme
        out[:, index] = sar
    return out


def supertrend_trace(close: torch.Tensor, high: torch.Tensor, low: torch.Tensor, period: int, multiplier: float) -> dict[str, torch.Tensor]:
    """Return the GPU state-machine trace for one Supertrend variant.

    The bar loop advances state, while every state operation remains a CUDA
    tensor operation; this is not a CPU or per-variant fallback.
    """
    output = _supertrend(close, high, low, period, [multiplier])[0]
    atr = _rma(_true_range(close, high, low), period)
    upper, lower = _empty(close), _empty(close)
    direction, initialized = torch.zeros_like(close, dtype=torch.int8), torch.zeros_like(close, dtype=torch.bool)
    for index in range(period, close.numel()):
        midpoint = (high[index] + low[index]) / 2.0
        basic_upper, basic_lower = midpoint + multiplier * atr[index], midpoint - multiplier * atr[index]
        if index == period:
            upper[index], lower[index] = basic_upper, basic_lower
        else:
            upper[index] = torch.minimum(basic_upper, upper[index - 1]) if close[index - 1] <= upper[index - 1] else basic_upper
            lower[index] = torch.maximum(basic_lower, lower[index - 1]) if close[index - 1] >= lower[index - 1] else basic_lower
        direction[index] = torch.where(close[index] > output[index], 1, -1)
        initialized[index] = True
    return {"output": output, "atr": atr, "upper": upper, "lower": lower, "direction": direction, "initialized": initialized}


def psar_trace(close: torch.Tensor, high: torch.Tensor, low: torch.Tensor, step: float) -> dict[str, torch.Tensor]:
    """Return the GPU state-machine trace for one PSAR variant."""
    output = _psar(close, high, low, [step])[0]
    bullish = torch.zeros_like(close, dtype=torch.bool)
    extreme, acceleration = _empty(close), _empty(close)
    reversal = torch.zeros_like(close, dtype=torch.bool)
    if close.numel() < 2:
        return {"output": output, "bullish": bullish, "extreme": extreme, "acceleration": acceleration, "reversal": reversal}
    bull, af, ep, sar = torch.tensor(True, device=close.device), torch.tensor(.02, dtype=torch.float64, device=close.device), high[0], low[0]
    for index in range(1, close.numel()):
        sar = sar + af * (ep - sar)
        candidate = torch.minimum(sar, torch.minimum(low[index - 1], low[index - 2] if index >= 2 else low[index - 1])) if bool(bull) else torch.maximum(sar, torch.maximum(high[index - 1], high[index - 2] if index >= 2 else high[index - 1]))
        reverse = (bool(bull) and bool(low[index] < candidate)) or (not bool(bull) and bool(high[index] > candidate))
        if reverse:
            bull, sar, ep, af = (torch.tensor(False, device=close.device), ep, low[index], torch.tensor(.02, dtype=torch.float64, device=close.device)) if bool(bull) else (torch.tensor(True, device=close.device), ep, high[index], torch.tensor(.02, dtype=torch.float64, device=close.device))
        else:
            sar = candidate
            if (bool(bull) and bool(high[index] > ep)) or (not bool(bull) and bool(low[index] < ep)):
                ep, af = (high[index], torch.minimum(af + step, torch.tensor(.2, dtype=torch.float64, device=close.device))) if bool(bull) else (low[index], torch.minimum(af + step, torch.tensor(.2, dtype=torch.float64, device=close.device)))
        bullish[index], extreme[index], acceleration[index], reversal[index] = bull, ep, af, reverse
    return {"output": output, "bullish": bullish, "extreme": extreme, "acceleration": acceleration, "reversal": reversal}


def evaluate_batch(request: dict[str, Any], close: torch.Tensor, high: torch.Tensor, low: torch.Tensor, volume: torch.Tensor) -> dict[str, torch.Tensor]:
    del volume  # Batch01 formulas do not consume volume, but preserve the input contract.
    groups: dict[tuple[int, int], list[dict[str, Any]]] = {}
    for item in request["requests"]:
        groups.setdefault((int(item["indicator_id"]), int(item["period"])), []).append(item)
    outputs: dict[str, torch.Tensor] = {}
    for (indicator, period), items in groups.items():
        multipliers = [float(item["p1"]) for item in items]
        if indicator == 17:
            batch = _bollinger(close, period, multipliers)[0]
        elif indicator == 18:
            batch = _bollinger(close, period, multipliers)[1]
        elif indicator == 19:
            batch = _supertrend(close, high, low, period, multipliers)
        elif indicator == 20:
            batch = _donchian(high, period, True).unsqueeze(0)
        elif indicator == 21:
            batch = _donchian(low, period, False).unsqueeze(0)
        elif indicator == 23:
            batch = _keltner(close, high, low, period, multipliers)[0]
        elif indicator == 24:
            batch = _keltner(close, high, low, period, multipliers)[1]
        elif indicator == 28:
            batch = _psar(close, high, low, multipliers)
        else:
            raise ValueError(f"unsupported indicator: {indicator}")
        for index, item in enumerate(items):
            outputs[str(item["request_id"])] = batch[index]
    return outputs


def _cache_path(cache_dir: Path, data_csv: Path, request_json: Path) -> Path:
    key = hashlib.sha256(f"v1:{_sha256(data_csv)}:{_sha256(request_json)}".encode()).hexdigest()[:24]
    return cache_dir / f"batch01_gpu_{key}.npz"


def _load_or_compute(args: argparse.Namespace, device: torch.device) -> tuple[dict[str, np.ndarray], bool, float]:
    cache = _cache_path(args.cache_dir, args.data_csv, args.request_json)
    request = json.loads(args.request_json.read_text(encoding="utf-8"))
    expected_ids = {item["request_id"] for item in request["requests"]}
    if cache.exists():
        with np.load(cache) as saved:
            if set(saved.files) == expected_ids:
                return {key: saved[key] for key in saved.files}, True, 0.0
    start = time.perf_counter()
    outputs = evaluate_batch(request, *_load_ohlcv(args.data_csv, device))
    torch.cuda.synchronize(device)
    elapsed = time.perf_counter() - start
    host = {key: value.detach().cpu().numpy() for key, value in outputs.items()}
    args.cache_dir.mkdir(parents=True, exist_ok=True)
    np.savez_compressed(cache, **host)
    return host, False, elapsed


def smoke(device: torch.device) -> dict[str, Any]:
    values = torch.arange(1024, dtype=torch.float64, device=device)
    result = (values.square().sum() / values.numel()).item()
    torch.cuda.synchronize(device)
    return {"passed": bool(np.isfinite(result)), "device": str(device), "torch": torch.__version__, "cuda": torch.version.cuda, "value": result}


def _family(indicator_id: int) -> str:
    return {17: "bollinger", 18: "bollinger", 19: "supertrend", 20: "donchian", 21: "donchian", 23: "keltner", 24: "keltner", 28: "psar"}[indicator_id]


def _ordered_bits(values: np.ndarray) -> np.ndarray:
    bits = values.view(np.uint64)
    return np.where(bits >> 63 != 0, ~bits, bits | np.uint64(1 << 63))


def _numeric_measurement(actual: np.ndarray, expected: np.ndarray) -> dict[str, Any]:
    finite = np.isfinite(actual) & np.isfinite(expected)
    difference = np.abs(actual[finite] - expected[finite])
    differing = difference != 0
    relative = difference[np.abs(expected[finite]) != 0] / np.abs(expected[finite][np.abs(expected[finite]) != 0])
    ulps = np.abs(_ordered_bits(actual[finite]).astype(object) - _ordered_bits(expected[finite]).astype(object))
    ulps = np.asarray(ulps, dtype=np.float64)
    return {
        "finite_compared_count": int(finite.sum()),
        "finite_differing_count": int(differing.sum()),
        "max_absolute_error": float(difference.max()) if difference.size else 0.0,
        "max_relative_error": float(relative.max()) if relative.size else 0.0,
        "mae": float(difference.mean()) if difference.size else 0.0,
        "max_ulp": int(ulps.max()) if ulps.size else 0,
        "p50_ulp": float(np.percentile(ulps, 50)) if ulps.size else 0.0,
        "p95_ulp": float(np.percentile(ulps, 95)) if ulps.size else 0.0,
        "p99_ulp": float(np.percentile(ulps, 99)) if ulps.size else 0.0,
    }


def _decision_measurement(close: np.ndarray, actual: np.ndarray, expected: np.ndarray) -> dict[str, Any]:
    valid = np.isfinite(close) & np.isfinite(actual) & np.isfinite(expected)
    oracle_sign = np.sign(close[valid] - expected[valid]).astype(np.int8)
    gpu_sign = np.sign(close[valid] - actual[valid]).astype(np.int8)
    sign_exact = bool(np.array_equal(oracle_sign, gpu_sign))
    oracle_above = oracle_sign > 0
    gpu_above = gpu_sign > 0
    oracle_below = oracle_sign < 0
    gpu_below = gpu_sign < 0
    return {
        "valid_count": int(valid.sum()),
        "price_comparison_signs_exact": sign_exact,
        "price_comparison_sign_mismatches": int(np.count_nonzero(oracle_sign != gpu_sign)),
        "crossings_above_exact": bool(np.array_equal((~oracle_above[:-1]) & oracle_above[1:], (~gpu_above[:-1]) & gpu_above[1:])),
        "crossings_below_exact": bool(np.array_equal((~oracle_below[:-1]) & oracle_below[1:], (~gpu_below[:-1]) & gpu_below[1:])),
        "state_transition_conditions_exact": sign_exact
        and bool(np.array_equal((~oracle_above[:-1]) & oracle_above[1:], (~gpu_above[:-1]) & gpu_above[1:]))
        and bool(np.array_equal((~oracle_below[:-1]) & oracle_below[1:], (~gpu_below[:-1]) & gpu_below[1:])),
    }


def _band_relationship(actual_upper: np.ndarray, actual_lower: np.ndarray, expected_upper: np.ndarray, expected_lower: np.ndarray) -> dict[str, Any]:
    valid = np.isfinite(actual_upper) & np.isfinite(actual_lower) & np.isfinite(expected_upper) & np.isfinite(expected_lower)
    oracle_relation = np.sign(expected_upper[valid] - expected_lower[valid]).astype(np.int8)
    gpu_relation = np.sign(actual_upper[valid] - actual_lower[valid]).astype(np.int8)
    return {"valid_count": int(valid.sum()), "upper_lower_relationship_exact": bool(np.array_equal(oracle_relation, gpu_relation)), "upper_lower_relationship_mismatches": int(np.count_nonzero(oracle_relation != gpu_relation))}


def _write_json(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n", encoding="utf-8")


def _contract_markdown(contract: dict[str, Any]) -> str:
    lines = ["# GPU Feature Parity Contract V1", "", f"Status: `{contract['status']}`.", "", "The contract requires exact structure and NaN placement, exact decision/state-transition conditions, and per-family measured numeric limits. Numeric drift is never accepted when it changes a decision.", "", "## Family Limits", ""]
    for family, limit in contract["family_tolerances"].items():
        lines.append(f"- `{family}`: max abs {limit['max_absolute_error']:.17g}, max rel {limit['max_relative_error']:.17g}, max ULP {limit['max_ulp']}, MAE {limit['mae']:.17g}.")
    lines.extend(["", "## Gates", "", "- Shapes, dtypes, and NaN masks must match exactly.", "- Price-comparison signs, crossings above/below, and state-transition conditions must match exactly.", "- Bollinger and Keltner upper/lower relationships must match exactly.", ""])
    return "\n".join(lines)


def parity_report(args: argparse.Namespace, device: torch.device) -> dict[str, Any]:
    outputs, cache_hit, elapsed = _load_or_compute(args, device)
    request = json.loads(args.request_json.read_text(encoding="utf-8"))
    close = _load_ohlcv(args.data_csv, torch.device("cpu"))[0].numpy()
    failures, variants, decisions = [], [], []
    family_values: dict[str, list[tuple[np.ndarray, np.ndarray]]] = {}
    band_pairs: dict[tuple[int, float], dict[int, tuple[np.ndarray, np.ndarray]]] = {}
    with np.load(args.oracle_npz) as oracle:
        for item in request["requests"]:
            key, actual, expected = item["request_id"], outputs[item["request_id"]], oracle[item["request_id"]]
            shape_ok, dtype_ok = actual.shape == expected.shape, actual.dtype == expected.dtype
            equal = (actual == expected) | (np.isnan(actual) & np.isnan(expected)) if shape_ok else np.array([False])
            mismatches = int(equal.size - equal.sum())
            nan_exact = bool(shape_ok and np.array_equal(np.isnan(actual), np.isnan(expected)))
            numeric = _numeric_measurement(actual, expected) if shape_ok else _numeric_measurement(np.array([], dtype=np.float64), np.array([], dtype=np.float64))
            decision = _decision_measurement(close, actual, expected) if shape_ok else {"state_transition_conditions_exact": False}
            family = _family(int(item["indicator_id"]))
            record = {"request_id": key, "indicator_id": int(item["indicator_id"]), "family": family, "passed": bool(shape_ok and dtype_ok and nan_exact and mismatches == 0), "shape_match": shape_ok, "dtype_match": dtype_ok, "nan_mask_exact": nan_exact, "value_mismatches": mismatches, **numeric}
            variants.append(record)
            decisions.append({"request_id": key, "family": family, **decision})
            family_values.setdefault(family, []).append((actual, expected))
            if int(item["indicator_id"]) in {17, 18, 23, 24}:
                band_pairs.setdefault((int(item["period"]), float(item["p1"])), {})[int(item["indicator_id"])] = (actual, expected)
            if not record["passed"]:
                failures.append(record)
    family_tolerances = {family: _numeric_measurement(np.concatenate([pair[0] for pair in pairs]), np.concatenate([pair[1] for pair in pairs])) for family, pairs in family_values.items()}
    relationships = []
    for (period, multiplier), pair in band_pairs.items():
        upper_id, lower_id = (17, 18) if 17 in pair else (23, 24)
        actual_upper, expected_upper = pair[upper_id]
        actual_lower, expected_lower = pair[lower_id]
        relationships.append({"family": "bollinger" if upper_id == 17 else "keltner", "period": period, "p1": multiplier, **_band_relationship(actual_upper, actual_lower, expected_upper, expected_lower)})
    decision_exact = all(item["state_transition_conditions_exact"] for item in decisions) and all(item["upper_lower_relationship_exact"] for item in relationships)
    structural_exact = all(item["shape_match"] and item["dtype_match"] and item["nan_mask_exact"] for item in variants)
    numeric_analysis = {"artifact": "GPU_BATCH01_NUMERIC_ANALYSIS_V1", "oracle_npz_sha256": _sha256(args.oracle_npz), "data_csv_sha256": _sha256(args.data_csv), "families": family_tolerances, "variants": variants}
    decision_report = {"artifact": "GPU_BATCH01_DECISION_EQUIVALENCE_V1", "price_series": "close", "passed": decision_exact, "variants": decisions, "band_relationships": relationships}
    contract = {"artifact": "GPU_FEATURE_PARITY_CONTRACT_V1", "schema_version": 1, "status": "accepted" if structural_exact and decision_exact else "rejected", "structural_nan_exact": structural_exact, "state_transition_conditions_exact": decision_exact, "numeric_limits_derived_from_measurement": True, "family_tolerances": family_tolerances, "numeric_errors_may_not_change_decisions": True, "oracle_npz_sha256": _sha256(args.oracle_npz), "data_csv_sha256": _sha256(args.data_csv)}
    _write_json(args.output_dir / "gpu_batch01_numeric_analysis_v1.json", numeric_analysis)
    _write_json(args.output_dir / "gpu_batch01_decision_equivalence_v1.json", decision_report)
    _write_json(args.output_dir / "gpu_feature_parity_contract_v1.json", contract)
    (args.output_dir / "gpu_feature_parity_contract_v1.md").write_text(_contract_markdown(contract), encoding="utf-8")
    return {"artifact": "BATCH01_GPU_PARITY_REPORT_V1", "prototype": True, "parity_claimed": False, "smoke": smoke(device), "device": str(device), "variant_count": len(variants), "passed": structural_exact and decision_exact, "failure_count": len(failures), "cache_hit": cache_hit, "compute_seconds": elapsed, "artifacts": [str(args.output_dir / name) for name in ("gpu_batch01_numeric_analysis_v1.json", "gpu_batch01_decision_equivalence_v1.json", "gpu_feature_parity_contract_v1.json", "gpu_feature_parity_contract_v1.md")], "failures": failures}


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--mode", choices=("smoke", "parity"), default="smoke")
    parser.add_argument(
        "--data-csv",
        type=Path,
        default=Path(os.getenv("GPU_FEATURE_DATA_CSV", "/data/binance_btcusdt_spot_2m_180d.csv")),
    )
    parser.add_argument(
        "--oracle-npz",
        type=Path,
        default=Path(os.getenv("GPU_FEATURE_ORACLE_NPZ", "/oracle/batch01_oracle_outputs.npz")),
    )
    parser.add_argument(
        "--request-json",
        type=Path,
        default=Path(os.getenv("GPU_FEATURE_REQUEST_JSON", "/oracle/batch01_oracle_request.json")),
    )
    parser.add_argument("--cache-dir", type=Path, default=Path(os.getenv("GPU_FEATURE_CACHE_DIR", "/cache")))
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path(os.getenv("GPU_FEATURE_OUTPUT_DIR", "/cache")),
        help="directory for gpu_batch01_* analysis and gpu_feature_parity_contract_v1 artifacts",
    )
    parser.add_argument("--report", type=Path)
    args = parser.parse_args()
    if not torch.cuda.is_available():
        raise SystemExit("CUDA GPU is required; torch.cuda.is_available() is false")
    device = torch.device("cuda")
    report = smoke(device) if args.mode == "smoke" else parity_report(args, device)
    encoded = json.dumps(report, indent=2, allow_nan=False) + "\n"
    if args.report:
        args.report.parent.mkdir(parents=True, exist_ok=True)
        args.report.write_text(encoded, encoding="utf-8")
    print(encoded, end="")


if __name__ == "__main__":
    main()