125 lines
5.3 KiB
Python
125 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from control_plane.trading_studio.indicators.definitions import IndicatorVariant, Role
|
|
from control_plane.trading_studio.indicators.engine import compute_hs22_state, evaluate_indicator
|
|
|
|
|
|
ORACLE = Path(__file__).resolve().parents[1] / "hs22_oracle_v1"
|
|
CSV_SHA256 = "7fb56bb05b0edc2348407765cd693efadf334307c25877dfb52acfe4221e5d00"
|
|
# Spark emitted the oracle on NumPy 2.4.6/aarch64; this Windows runner uses
|
|
# NumPy 2.4.4. The largest observed arithmetic drift is 1.11e-15 (IDs 79/76).
|
|
FLOAT_ATOL = 1.2e-15
|
|
SLOT_ROLE = {
|
|
"trend": Role.TREND,
|
|
"signal": Role.OSC,
|
|
"trigger": Role.LEVEL,
|
|
"confirm": Role.OSC,
|
|
"vol": Role.FILTER,
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def oracle():
|
|
assert sha256((ORACLE / "binance_btcusdt_spot_2m_180d.csv").read_bytes()).hexdigest() == CSV_SHA256
|
|
manifest = json.loads((ORACLE / "required_variants.json").read_text(encoding="utf-8"))
|
|
with (ORACLE / "binance_btcusdt_spot_2m_180d.csv").open(newline="", encoding="utf-8") as handle:
|
|
rows = list(csv.DictReader(handle))
|
|
assert len(rows) == 129_599
|
|
return manifest, tuple(
|
|
np.asarray([float(row[name]) for row in rows], dtype=np.float64)
|
|
for name in ("open", "high", "low", "close", "volume")
|
|
)
|
|
|
|
|
|
def _assert_exact(actual: np.ndarray, expected: np.ndarray) -> None:
|
|
assert actual.dtype == expected.dtype
|
|
assert actual.shape == expected.shape
|
|
assert np.array_equal(np.isnan(actual), np.isnan(expected))
|
|
if np.issubdtype(actual.dtype, np.integer):
|
|
assert np.array_equal(actual, expected)
|
|
else:
|
|
np.testing.assert_allclose(actual, expected, rtol=0.0, atol=FLOAT_ATOL, equal_nan=True)
|
|
|
|
|
|
def _variant(slot: str, item: dict[str, float]) -> IndicatorVariant:
|
|
return IndicatorVariant(int(item["indicator_id"]), int(item["period"]), item["p1"], SLOT_ROLE[slot])
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def native_variants(oracle):
|
|
manifest, (_, high, low, close, volume) = oracle
|
|
outputs = {}
|
|
with np.load(ORACLE / "required_variant_arrays.npz") as expected:
|
|
for combo in manifest["combos"]:
|
|
for slot, item in combo["role_triples"].items():
|
|
variant = _variant(slot, item)
|
|
key = (variant.indicator_id, variant.period, variant.p1)
|
|
actual = evaluate_indicator(variant, close, high, low, volume)
|
|
oracle_key = "v_" + combo["role_variant_fingerprints"][slot]
|
|
_assert_exact(actual, expected[oracle_key])
|
|
outputs[key] = actual
|
|
assert len(outputs) == 80
|
|
return outputs
|
|
|
|
|
|
def test_all_80_required_variant_arrays_match_the_frozen_oracle(native_variants):
|
|
assert len(native_variants) == 80
|
|
|
|
|
|
def _decisions(state: dict[str, np.ndarray], close: np.ndarray, combo: list[float]) -> np.ndarray:
|
|
vol = state["vol"]
|
|
valid = vol[~np.isnan(vol)]
|
|
threshold = np.sort(valid)[min(int(combo[21] / 100.0 * len(valid)), len(valid) - 1)]
|
|
out = np.zeros(len(close), dtype=np.int8)
|
|
signal = state["signal_color"]
|
|
trend = state["trend_color"]
|
|
common = np.isfinite(state["trigger"]) & np.isfinite(state["confirm"]) & np.isfinite(vol)
|
|
common &= vol >= threshold
|
|
flips = (signal != np.roll(signal, 1)) & (signal != 0)
|
|
long = common & flips & (signal == 1) & (trend == 1) & (close > state["trigger"])
|
|
long &= (combo[17] <= state["confirm"]) & (state["confirm"] <= combo[18])
|
|
short = common & flips & (signal == -1) & (trend == -1) & (close < state["trigger"])
|
|
short &= (combo[19] <= state["confirm"]) & (state["confirm"] <= combo[20])
|
|
out[long] = 1
|
|
out[short] = -1
|
|
return out
|
|
|
|
|
|
def test_all_20_hs22_states_and_decisions_match_the_frozen_oracle(oracle):
|
|
manifest, (_, high, low, close, volume) = oracle
|
|
names = ("trend", "signal", "trigger", "confirm", "vol", "trend_color", "signal_color")
|
|
for combo in manifest["combos"]:
|
|
actual = compute_hs22_state(combo["combo"], close, high, low, volume)
|
|
with np.load(ORACLE / combo["state_file"]) as expected:
|
|
for name in names:
|
|
_assert_exact(actual[name], expected[name])
|
|
expected_state = {name: expected[name] for name in names}
|
|
assert np.count_nonzero(_decisions(actual, close, combo["combo"]) != _decisions(expected_state, close, combo["combo"])) == 0
|
|
|
|
|
|
def test_fold1_context_prefix_matches_continuous_history(native_variants, oracle):
|
|
_, (_, high, low, close, volume) = oracle
|
|
fold1_test_stop = 14_400 + 35 + 7_200
|
|
fold1_test = slice(14_400 + 35, fold1_test_stop)
|
|
for (indicator_id, period, p1), full in native_variants.items():
|
|
# Formula output is role-independent; choose the first registered role for this triple.
|
|
for candidate in SLOT_ROLE.values():
|
|
try:
|
|
context = evaluate_indicator(
|
|
IndicatorVariant(indicator_id, period, p1, candidate),
|
|
close[:fold1_test_stop], high[:fold1_test_stop], low[:fold1_test_stop], volume[:fold1_test_stop],
|
|
)
|
|
break
|
|
except ValueError:
|
|
continue
|
|
else:
|
|
raise AssertionError("required variant lost its strict role registration")
|
|
_assert_exact(context[fold1_test], full[fold1_test])
|