111 lines
5.8 KiB
Python
111 lines
5.8 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from gpu_feature_parity_contract_v1_1 import (
|
|
ARTIFACT,
|
|
FrozenContractError,
|
|
SCENARIOS,
|
|
calibrate,
|
|
compare_stateful_trace,
|
|
corpus_manifest,
|
|
cpu_psar_trace,
|
|
cpu_supertrend_trace,
|
|
deterministic_adversarial_ohlcv,
|
|
freeze_contract,
|
|
historical_cpu_oracle,
|
|
nan_gap_semantics_manifest,
|
|
validate_frozen_contract,
|
|
)
|
|
|
|
|
|
def _evidence():
|
|
ohlcv = deterministic_adversarial_ohlcv(length=96, seed=7)
|
|
requests = [{"request_id": "super", "indicator_id": 19, "period": 8, "p1": 2.0}]
|
|
oracle = {"super": cpu_supertrend_trace(ohlcv["close"], ohlcv["high"], ohlcv["low"], 8, 2.0)["output"]}
|
|
corpus = corpus_manifest({"adversarial": ohlcv}, {"generator": "deterministic_adversarial_ohlcv", "seed": 7})
|
|
calibration = calibrate(oracle, oracle, requests, ohlcv["close"], corpus)
|
|
return ohlcv, oracle, freeze_contract(calibration)
|
|
|
|
|
|
def test_adversarial_generator_is_deterministic_and_manifest_has_provenance():
|
|
first = deterministic_adversarial_ohlcv(length=96, seed=3)
|
|
second = deterministic_adversarial_ohlcv(length=96, seed=3)
|
|
assert all(np.array_equal(first[key], second[key]) for key in first)
|
|
manifest = corpus_manifest({"edge": first}, {"source": "test"})
|
|
assert manifest["scenario_categories"] == list(SCENARIOS)
|
|
assert manifest["provenance"]["source"] == "test"
|
|
assert {"constant", "nearly_constant_tiny_variance", "high_variance", "alternating", "uptrend", "downtrend", "flat_breakout", "breakout_flat", "threshold_equality", "threshold_one_ulp_above", "threshold_one_ulp_below", "repeated_equality", "zero_range", "tiny_range", "abrupt_atr", "warmup"} == set(SCENARIOS)
|
|
|
|
|
|
def test_frozen_contract_validates_and_rejects_injected_numeric_nan_and_decision_violations():
|
|
ohlcv, oracle, contract = _evidence()
|
|
assert contract["artifact"] == ARTIFACT
|
|
assert {"max_ulp", "p50_ulp", "p95_ulp", "p99_ulp", "p99_9_ulp"}.issubset(contract["feature_limits"]["super"])
|
|
assert {"max_ulp", "p50_ulp", "p95_ulp", "p99_ulp", "p99_9_ulp"}.issubset(contract["family_limits"]["supertrend"])
|
|
assert validate_frozen_contract(contract, oracle, oracle, ohlcv["close"])["passed"]
|
|
numeric = {"super": oracle["super"].copy()}; numeric["super"][-1] += 1.0
|
|
assert not validate_frozen_contract(contract, numeric, oracle, ohlcv["close"])["passed"]
|
|
nan = {"super": oracle["super"].copy()}; nan["super"][-1] = np.nan
|
|
assert not validate_frozen_contract(contract, nan, oracle, ohlcv["close"])["passed"]
|
|
decision = {"super": oracle["super"].copy()}; index = np.flatnonzero(np.isfinite(decision["super"]))[-1]; decision["super"][index] = ohlcv["close"][index] + 1.0
|
|
assert not validate_frozen_contract(contract, decision, oracle, ohlcv["close"])["passed"]
|
|
|
|
|
|
def test_frozen_contract_rejects_a_deliberate_ulp_breach():
|
|
ohlcv, oracle, contract = _evidence()
|
|
breach = {"super": oracle["super"].copy()}
|
|
index = np.flatnonzero(np.isfinite(breach["super"]))[-1]
|
|
breach["super"][index] = np.nextafter(breach["super"][index], np.inf)
|
|
result = validate_frozen_contract(contract, breach, oracle, ohlcv["close"])
|
|
assert not result["passed"]
|
|
assert result["records"][0]["max_ulp"] > contract["feature_limits"]["super"]["max_ulp"]
|
|
assert "max_ulp" in result["records"][0]["numeric_violations"]
|
|
|
|
|
|
def test_validation_refuses_unfrozen_or_derived_contracts():
|
|
ohlcv, oracle, contract = _evidence()
|
|
contract["status"] = "accepted"
|
|
with pytest.raises(FrozenContractError):
|
|
validate_frozen_contract(contract, oracle, oracle, ohlcv["close"])
|
|
|
|
|
|
def test_trace_comparer_requires_every_discrete_state_key_and_value():
|
|
ohlcv = deterministic_adversarial_ohlcv(length=96)
|
|
trace = cpu_psar_trace(ohlcv["close"], ohlcv["high"], ohlcv["low"], .02)
|
|
assert compare_stateful_trace(trace, trace, trace_type="psar")["exact"]
|
|
changed = dict(trace); changed["reversal"] = trace["reversal"].copy(); changed["reversal"][-1] = ~changed["reversal"][-1]
|
|
assert compare_stateful_trace(trace, changed, trace_type="psar")["mismatched_keys"] == ["reversal"]
|
|
assert not compare_stateful_trace(trace, {"output": trace["output"]}, trace_type="psar")["exact"]
|
|
|
|
|
|
def test_decisions_do_not_cross_nan_gaps():
|
|
close = np.array([1., 2., 3.])
|
|
oracle = {"super": np.array([0., np.nan, 4.])}
|
|
calibration = calibrate(oracle, oracle, [{"request_id": "super", "indicator_id": 19}], close, corpus_manifest({"edge": deterministic_adversarial_ohlcv(length=96)}, {}))
|
|
contract = freeze_contract(calibration)
|
|
assert validate_frozen_contract(contract, oracle, oracle, close)["passed"]
|
|
|
|
|
|
def test_historical_cpu_oracle_accepts_arbitrary_csv(tmp_path):
|
|
ohlcv = deterministic_adversarial_ohlcv(length=96)
|
|
path = tmp_path / "input.csv"
|
|
with path.open("w", newline="", encoding="utf-8") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=["close", "high", "low", "volume"]); writer.writeheader()
|
|
writer.writerows({key: ohlcv[key][index] for key in ohlcv} for index in range(96))
|
|
result = historical_cpu_oracle(path, [{"request_id": "psar", "indicator_id": 28, "period": 1, "p1": .02}])
|
|
assert result["outputs"]["psar"].shape == (96,)
|
|
assert result["source"]["evaluator"] == "evaluate_band_channel"
|
|
|
|
|
|
def test_nan_gap_semantics_are_manifested_and_rejected_before_historical_evaluation(tmp_path):
|
|
manifest = nan_gap_semantics_manifest()
|
|
assert manifest["historical_input_policy"] == "reject_non_finite_ohlcv"
|
|
assert manifest["internal_nan_gap_semantics"] == "undefined"
|
|
path = tmp_path / "nan_gap.csv"
|
|
path.write_text("close,high,low,volume\n100,101,99,1000\nnan,101,99,1000\n", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="NaN/gap semantics are undefined"):
|
|
historical_cpu_oracle(path, [{"request_id": "psar", "indicator_id": 28, "period": 1, "p1": .02}])
|