203 lines
13 KiB
Python
203 lines
13 KiB
Python
"""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()
|