Artifex/gpu_supertrend_v1_2_diagnostic.py

"""V1.2-only semantic diagnostic for CPU and GPU Supertrend traces."""
from __future__ import annotations

import argparse
import csv
import json
from collections.abc import Mapping
from pathlib import Path
from typing import Any

import numpy as np
import torch

from gpu_feature_engine_v1_2 import supertrend_trace
from gpu_feature_parity_contract_v1_1 import deterministic_adversarial_ohlcv
from gpu_feature_parity_contract_v1_2 import PREDICATE_KEYS, SUPERTREND_TRACE_KEYS, cpu_supertrend_trace

ARTIFACT = "GPU_SUPERTREND_ATR_SEMANTIC_DIAGNOSTIC_V1_2"
PERIOD = 10
MULTIPLIERS = (2.0, 3.0, 4.0)
NUMERIC_FIELDS = ("output", "true_range", "atr", "basic_upper", "basic_lower", "upper", "lower")
EXACT_FIELDS = tuple(sorted((*PREDICATE_KEYS, "direction", "direction_transition")))


def _read_ohlcv(path: Path) -> dict[str, np.ndarray]:
    with path.open(newline="", encoding="utf-8") as handle:
        rows = list(csv.DictReader(handle))
    names = ("close", "high", "low")
    if not rows or not set(names).issubset(rows[0]):
        raise ValueError("CSV must contain non-empty close, high, and low columns")
    values = {name: np.asarray([float(row[name]) for row in rows], dtype=np.float64) for name in names}
    if not all(np.isfinite(value).all() for value in values.values()):
        raise ValueError("diagnostic requires finite close, high, and low inputs")
    return values


def _json_number(value: Any) -> float | int | bool | None:
    value = value.item() if isinstance(value, np.generic) else value
    if isinstance(value, (bool, np.bool_)):
        return bool(value)
    if isinstance(value, (int, np.integer)):
        return int(value)
    value = float(value)
    return value if np.isfinite(value) else None


def _first_difference(expected: np.ndarray, actual: np.ndarray) -> dict[str, Any] | None:
    equal = np.equal(expected, actual) | (np.isnan(expected) & np.isnan(actual))
    indices = np.flatnonzero(~equal)
    if not len(indices):
        return None
    index = int(indices[0])
    return {"bar": index, "cpu": _json_number(expected[index]), "gpu": _json_number(actual[index])}


def _ulp(expected: np.ndarray, actual: np.ndarray) -> np.ndarray:
    expected_bits, actual_bits = expected.view(np.uint64), actual.view(np.uint64)
    sign = np.uint64(1 << 63)
    expected_ordered = np.where(expected_bits >> 63 != 0, ~expected_bits, expected_bits | sign)
    actual_ordered = np.where(actual_bits >> 63 != 0, ~actual_bits, actual_bits | sign)
    return np.asarray(np.abs(expected_ordered.astype(object) - actual_ordered.astype(object)), dtype=np.float64)


def _numeric_stats(expected: np.ndarray, actual: np.ndarray) -> dict[str, Any]:
    finite = np.isfinite(expected) & np.isfinite(actual)
    delta = np.abs(actual[finite] - expected[finite])
    nonzero_expected = np.abs(expected[finite]) != 0
    relative = delta[nonzero_expected] / np.abs(expected[finite][nonzero_expected])
    zero_reference_divergence = bool(np.any((expected[finite] == 0) & (actual[finite] != 0)))
    return {
        "first_divergence": _first_difference(expected, actual),
        "finite_pairs": int(finite.sum()),
        "nonfinite_mismatches": int(np.count_nonzero(~(np.equal(expected, actual) | (np.isnan(expected) & np.isnan(actual))) & ~finite)),
        "max_abs": float(delta.max()) if delta.size else 0.0,
        "max_rel": float(relative.max()) if relative.size else 0.0,
        "max_rel_unbounded_at_zero_cpu": zero_reference_divergence,
        "max_ulp": float(_ulp(expected[finite], actual[finite]).max()) if delta.size else 0.0,
    }


def _exact_stats(expected: np.ndarray, actual: np.ndarray) -> dict[str, Any]:
    return {"exact": _first_difference(expected, actual) is None, "first_divergence": _first_difference(expected, actual)}


def _diagnose_one(ohlcv: Mapping[str, np.ndarray], multiplier: float, device: torch.device) -> dict[str, Any]:
    tensors = {name: torch.as_tensor(ohlcv[name], dtype=torch.float64, device=device) for name in ("close", "high", "low")}
    cpu = cpu_supertrend_trace(ohlcv["close"], ohlcv["high"], ohlcv["low"], PERIOD, multiplier)
    gpu_trace = supertrend_trace(tensors["close"], tensors["high"], tensors["low"], PERIOD, multiplier)
    if device.type == "cuda":
        torch.cuda.synchronize(device)
    gpu = {name: value.detach().cpu().numpy() for name, value in gpu_trace.items()}
    missing_cpu = sorted(SUPERTREND_TRACE_KEYS - cpu.keys())
    missing_gpu = sorted(SUPERTREND_TRACE_KEYS - gpu.keys())
    if missing_cpu or missing_gpu:
        raise RuntimeError(f"incomplete V1.2 Supertrend trace: CPU missing {missing_cpu}; GPU missing {missing_gpu}")
    return {
        "period": PERIOD,
        "multiplier": multiplier,
        "numeric": {name: _numeric_stats(cpu[name], gpu[name]) for name in NUMERIC_FIELDS},
        "branch_predicates": {name: _exact_stats(cpu[name], gpu[name]) for name in sorted(PREDICATE_KEYS)},
        "direction": _exact_stats(cpu["direction"], gpu["direction"]),
        "transitions": {"direction_transition": _exact_stats(cpu["direction_transition"], gpu["direction_transition"])},
    }


def diagnose(ohlcv: Mapping[str, np.ndarray], periods: list[int] | None = None, multiplier: float | None = None, device: torch.device | None = None) -> dict[str, Any]:
    """Diagnose one corpus; V1.2 semantics are fixed to period 10 and 2/3/4 multipliers."""
    if periods not in (None, [PERIOD]) or multiplier not in (None, *MULTIPLIERS):
        raise ValueError("V1.2 semantic diagnostic supports only period 10 and multipliers 2, 3, and 4")
    if device is None:
        device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    return {"artifact": ARTIFACT, "schema_version": "1.2", "device": str(device), "bars": len(ohlcv["close"]), "period": PERIOD, "records": [_diagnose_one(ohlcv, value, device) for value in MULTIPLIERS]}


def run(calibration_csv: Path, holdout_csv: Path, *, adversarial_length: int = 256, seed: int = 0, device: torch.device | None = None) -> dict[str, Any]:
    """Produce calibration, holdout, and deterministic adversarial diagnostic corpora."""
    if adversarial_length < PERIOD + 1:
        raise ValueError("adversarial length must be at least 11")
    if device is None:
        if not torch.cuda.is_available():
            raise RuntimeError("CUDA GPU is required; pass device=torch.device('cpu') only for tests")
        device = torch.device("cuda")
    return {"artifact": ARTIFACT, "schema_version": "1.2", "device": str(device), "adversarial": {"generator": "deterministic_adversarial_ohlcv", "length": adversarial_length, "seed": seed}, "corpora": {"calibration": diagnose(_read_ohlcv(calibration_csv), device=device), "holdout": diagnose(_read_ohlcv(holdout_csv), device=device), "adversarial": diagnose(deterministic_adversarial_ohlcv(length=adversarial_length, seed=seed), device=device)}}


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--calibration-csv", type=Path, required=True)
    parser.add_argument("--holdout-csv", type=Path, required=True)
    parser.add_argument("--adversarial-length", type=int, default=256)
    parser.add_argument("--seed", type=int, default=0)
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    payload = run(args.calibration_csv, args.holdout_csv, adversarial_length=args.adversarial_length, seed=args.seed)
    encoded = json.dumps(payload, sort_keys=True, allow_nan=False) + "\n"
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(encoded, encoding="utf-8")
    else:
        print(encoded, end="")


if __name__ == "__main__":
    main()