"""GPU Batch01 V1.2 full-family, frozen-contract evidence runner.""" from __future__ import annotations import argparse import hashlib import json import os import platform import sys import time from pathlib import Path from typing import Any, Mapping import numpy as np import torch from control_plane.trading_studio.indicators.historical_band_channel import evaluate_band_channel from gpu_batch01_v1_1_runner import read_ohlcv, write_ohlcv_csv from gpu_feature_engine_v1 import evaluate_batch as evaluate_v1_batch 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 ( CALIBRATION_ARTIFACT, calibrate_output, calibrate_supertrend, canonical_bytes, freeze_contract, review_template, role_surface, validate_all_frozen_contract, ) ARTIFACT_PREFIX = "gpu_batch01_v1_2" ATR_LIMIT_KEYS = ("max_absolute_error", "mae") DEFAULT_ATR_LIMITS = {"max_absolute_error": 1.5e-14, "mae": 1.5e-14} def write_json(path: Path, payload: Mapping[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(canonical_bytes(dict(payload)) + b"\n") def _sha256_file(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _atr_limits(atr_limits: Mapping[str, float] | None) -> dict[str, float]: limits = dict(DEFAULT_ATR_LIMITS if atr_limits is None else atr_limits) if set(limits) != set(ATR_LIMIT_KEYS) or any(not np.isfinite(value) or value < 0 for value in limits.values()): raise ValueError("ATR limits must contain finite, non-negative max_absolute_error and mae") return limits def _cpu_outputs(requests: list[Mapping[str, Any]], ohlcv: Mapping[str, np.ndarray]) -> dict[str, np.ndarray]: return { str(item["request_id"]): evaluate_band_channel( int(item["indicator_id"]), ohlcv["close"], ohlcv["high"], ohlcv["low"], ohlcv["volume"], int(item["period"]), float(item["p1"]) ) for item in requests } def _gpu_outputs(requests: list[Mapping[str, Any]], ohlcv: Mapping[str, np.ndarray], device: torch.device) -> tuple[dict[str, np.ndarray], dict[str, tuple[dict[str, np.ndarray], dict[str, np.ndarray]]]]: tensors = tuple(torch.as_tensor(ohlcv[name], dtype=torch.float64, device=device) for name in ("close", "high", "low", "volume")) other = [item for item in requests if int(item["indicator_id"]) != 19] outputs = evaluate_v1_batch({"requests": other}, *tensors) if other else {} traces = {} for item in requests: if int(item["indicator_id"]) != 19: continue key = str(item["request_id"]) actual = supertrend_trace(*tensors[:3], int(item["period"]), float(item["p1"])) # The CPU trace is imported lazily to keep historical output generation explicit. from gpu_feature_parity_contract_v1_2 import cpu_supertrend_trace expected = cpu_supertrend_trace(ohlcv["close"], ohlcv["high"], ohlcv["low"], int(item["period"]), float(item["p1"])) traces[key] = (expected, {name: value.detach().cpu().numpy() for name, value in actual.items()}) outputs[key] = actual["output"] torch.cuda.synchronize(device) return {key: value.detach().cpu().numpy() for key, value in outputs.items()}, traces def _gpu_workload(requests: list[Mapping[str, Any]], ohlcv: Mapping[str, np.ndarray], device: torch.device) -> None: tensors = tuple(torch.as_tensor(ohlcv[name], dtype=torch.float64, device=device) for name in ("close", "high", "low", "volume")) other = [item for item in requests if int(item["indicator_id"]) != 19] if other: evaluate_v1_batch({"requests": other}, *tensors) for item in requests: if int(item["indicator_id"]) == 19: supertrend_trace(*tensors[:3], int(item["period"]), float(item["p1"])) def performance(requests: list[Mapping[str, Any]], ohlcv: Mapping[str, np.ndarray], device: torch.device) -> dict[str, Any]: """Measure V1.2 GPU evaluation with synchronized intervals.""" families: dict[str, list[Mapping[str, Any]]] = {} for item in requests: family = {17: "bollinger", 18: "bollinger", 19: "supertrend", 20: "donchian", 21: "donchian", 23: "keltner", 24: "keltner", 28: "psar"}[int(item["indicator_id"])] families.setdefault(family, []).append(item) results = [] for family, items in sorted(families.items()): timings = [] for _ in range(2): torch.cuda.synchronize(device) started = time.perf_counter() _gpu_workload(items, ohlcv, device) torch.cuda.synchronize(device) timings.append(time.perf_counter() - started) values = len(ohlcv["close"]) * len(items) results.append({"family": family, "variants": len(items), "values": values, "cold_seconds": timings[0], "warm_seconds": timings[1], "cold_values_per_second": values / timings[0] if timings[0] else None, "warm_values_per_second": values / timings[1] if timings[1] else None}) return {"artifact": "GPU_BATCH01_PERFORMANCE_METRICS_V1_2", "schema_version": "1.2", "measurement": "CUDA synchronize before and after each interval; cold is the first family invocation and warm is the second", "families": results} def _markdown(contract: Mapping[str, Any], roles: Mapping[str, Any]) -> str: counts: dict[str, int] = {} for limit in contract["feature_limits"].values(): counts[limit["family"]] = counts.get(limit["family"], 0) + 1 return "\n".join(( "# GPU Batch01 V1.2 Frozen Validation Contract", "", "All 97 observed Batch01 outputs are evaluated against `historical_band_channel`.", "", "## Evidence Classes", "", "- Donchian and PSAR outputs: exact.", "- Bollinger and Keltner outputs: bounded by calibration-only limits.", "- Supertrend final output, bands, direction, transitions, and branch predicates: exact; ATR uses supplied fixed bounds.", "", "## Frozen Coverage", "", *[f"- `{family}`: {count}" for family, count in sorted(counts.items())], "", "## Role Status", "", "Unrecovered strategy roles are `UNVERIFIABLE_FROM_RECOVERED_SOURCE`; they are recorded but are not parity blockers.", f"Role records: {len(roles['records'])}.", "", )) def run(calibration_csv: Path, holdout_csv: Path, request_json: Path, output_dir: Path, *, adversarial_length: int = 256, seed: int = 0, atr_limits: Mapping[str, float] | None = None, device: torch.device | None = None) -> dict[str, Any]: if device is None: if not torch.cuda.is_available(): raise RuntimeError("CUDA GPU is required; no CPU fallback is available") device = torch.device("cuda") atr_limits = _atr_limits(atr_limits) request = json.loads(request_json.read_text(encoding="utf-8")) requests = request.get("requests") if not isinstance(requests, list) or len(requests) != 97: raise ValueError("V1.2 requires the complete 97-request Batch01 manifest") if len({str(item["request_id"]) for item in requests}) != 97: raise ValueError("Batch01 request IDs must be unique") output_dir.mkdir(parents=True, exist_ok=True) calibration, holdout = read_ohlcv(calibration_csv), read_ohlcv(holdout_csv) adversarial = deterministic_adversarial_ohlcv(length=adversarial_length, seed=seed) write_ohlcv_csv(output_dir / f"{ARTIFACT_PREFIX}_adversarial.csv", adversarial) roles = role_surface(requests) write_json(output_dir / "gpu_feature_parity_role_surface_v1_2.json", roles) # Calibration is the sole point at which continuous-family limits are derived. calibration_cpu = _cpu_outputs(requests, calibration) calibration_gpu, calibration_traces = _gpu_outputs(requests, calibration, device) evidence = {"artifact": CALIBRATION_ARTIFACT, "schema_version": "1.2", "feature_count": 97, "records": [ calibrate_supertrend(str(item["request_id"]), *calibration_traces[str(item["request_id"])], atr_limits=atr_limits) if int(item["indicator_id"]) == 19 else calibrate_output(str(item["request_id"]), int(item["indicator_id"]), calibration_cpu[str(item["request_id"])], calibration_gpu[str(item["request_id"])] ) for item in requests ]} contract = freeze_contract(evidence) # Frozen before holdout/adversarial are evaluated. write_json(output_dir / "gpu_feature_parity_calibration_v1_2.json", evidence) write_json(output_dir / "gpu_feature_parity_contract_v1_2.json", contract) (output_dir / "gpu_feature_parity_contract_v1_2.md").write_text(_markdown(contract, roles), encoding="utf-8") np.savez_compressed(output_dir / f"{ARTIFACT_PREFIX}_calibration_cpu_oracle.npz", **calibration_cpu) np.savez_compressed(output_dir / f"{ARTIFACT_PREFIX}_calibration_gpu_outputs.npz", **calibration_gpu) reports = {} for name, corpus in (("calibration", calibration), ("holdout", holdout), ("adversarial", adversarial)): expected = calibration_cpu if name == "calibration" else _cpu_outputs(requests, corpus) actual, traces = (calibration_gpu, calibration_traces) if name == "calibration" else _gpu_outputs(requests, corpus, device) report = validate_all_frozen_contract(contract, expected, actual, traces) report.update({"corpus": name, "feature_count": 97}) reports[name] = report write_json(output_dir / f"gpu_feature_parity_validation_{name}_v1_2.json", report) if name != "calibration": np.savez_compressed(output_dir / f"{ARTIFACT_PREFIX}_{name}_cpu_oracle.npz", **expected) np.savez_compressed(output_dir / f"{ARTIFACT_PREFIX}_{name}_gpu_outputs.npz", **actual) review = review_template() review["required_review"].extend(["all 97 Batch01 requests are represented", "role status remains UNVERIFIABLE_FROM_RECOVERED_SOURCE where lineage is unrecovered"]) write_json(output_dir / "gpu_feature_parity_review_template_v1_2.json", review) write_json(output_dir / "gpu_batch01_performance_metrics_v1_2.json", performance(requests, holdout, device)) provenance = {"artifact": "GPU_BATCH01_RUNTIME_PROVENANCE_V1_2", "schema_version": "1.2", "python": sys.version, "platform": platform.platform(), "torch": torch.__version__, "cuda": torch.version.cuda, "gpu": torch.cuda.get_device_properties(device).name, "image_env": os.getenv("GPU_BATCH01_IMAGE_ENV"), "source_hashes": {path.name: _sha256_file(path) for path in (Path(__file__), Path(__file__).with_name("gpu_batch01_v1_1_runner.py"), Path(__file__).with_name("gpu_feature_engine_v1.py"), Path(__file__).with_name("gpu_feature_engine_v1_2.py"), Path(__file__).with_name("gpu_feature_parity_contract_v1_1.py"), Path(__file__).with_name("gpu_feature_parity_contract_v1_2.py"), Path(__file__).parent / "control_plane/trading_studio/indicators/historical_band_channel.py")}, "data_hashes": {"request_json": _sha256_file(request_json), "calibration_csv": _sha256_file(calibration_csv), "holdout_csv": _sha256_file(holdout_csv), "adversarial_csv": _sha256_file(output_dir / f"{ARTIFACT_PREFIX}_adversarial.csv")}, "adversarial": {"generator": "deterministic_adversarial_ohlcv", "length": adversarial_length, "seed": seed}} write_json(output_dir / "gpu_batch01_runtime_provenance_v1_2.json", provenance) artifacts = ["gpu_feature_parity_calibration_v1_2.json", "gpu_feature_parity_contract_v1_2.json", "gpu_feature_parity_contract_v1_2.md", "gpu_feature_parity_validation_calibration_v1_2.json", "gpu_feature_parity_validation_holdout_v1_2.json", "gpu_feature_parity_validation_adversarial_v1_2.json", "gpu_feature_parity_role_surface_v1_2.json", "gpu_feature_parity_review_template_v1_2.json", "gpu_batch01_performance_metrics_v1_2.json", "gpu_batch01_runtime_provenance_v1_2.json"] manifest = {"artifact": "GPU_BATCH01_V1_2_EVIDENCE_MANIFEST", "schema_version": "1.2", "status": "evidence_generated_not_validated", "feature_count": 97, "contract_sha256": hashlib.sha256(canonical_bytes(contract)).hexdigest(), "atr_limits": atr_limits, "validation_passed": all(report["passed"] for report in reports.values()), "artifacts": artifacts} write_json(output_dir / "gpu_batch01_v1_2_evidence_manifest.json", manifest) return manifest 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("--request-json", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--adversarial-length", type=int, default=256) parser.add_argument("--seed", type=int, default=0) parser.add_argument("--supertrend-atr-max-absolute-error", type=float, default=DEFAULT_ATR_LIMITS["max_absolute_error"]) parser.add_argument("--supertrend-atr-mae", type=float, default=DEFAULT_ATR_LIMITS["mae"]) args = parser.parse_args() print(json.dumps(run(args.calibration_csv, args.holdout_csv, args.request_json, args.output_dir, adversarial_length=args.adversarial_length, seed=args.seed, atr_limits={"max_absolute_error": args.supertrend_atr_max_absolute_error, "mae": args.supertrend_atr_mae}), allow_nan=False)) if __name__ == "__main__": main()