100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from control_plane.trading_studio.indicators.definitions import IndicatorVariant, Role
|
|
from control_plane.trading_studio.indicators.engine import IndicatorEngine, RefusalCode
|
|
from control_plane.trading_studio.indicators.registry import (
|
|
EXPERIMENTAL_UNIVERSE,
|
|
HISTORICAL_ARTIFACT_PATH,
|
|
HISTORICAL_REGISTRY,
|
|
HS22_COHORT001_V1,
|
|
HS22_HISTORICAL_COMPLETE_V1,
|
|
UNREGISTERED_INDICATORS,
|
|
historical_artifact_bytes,
|
|
historical_artifact_digest,
|
|
)
|
|
from control_plane.trading_studio.indicators.schema import HS22SchemaError, parse_hs22
|
|
|
|
|
|
def test_historical_inventory_counts_role_pools_and_digest_are_stable():
|
|
assert len(HISTORICAL_REGISTRY.definitions) == 145
|
|
assert len(HISTORICAL_REGISTRY.variants) == 1107
|
|
assert {role.value: len(HS22_HISTORICAL_COMPLETE_V1.by_role(role)) for role in Role} == {
|
|
"level": 493,
|
|
"osc": 372,
|
|
"trend": 103,
|
|
"filter": 139,
|
|
}
|
|
assert historical_artifact_digest() == (
|
|
"9be2aa3c56c90d2ffc73b64e857092c320fd57b6bbb77a358611a267b52c5ab2"
|
|
)
|
|
assert (
|
|
hashlib.sha256(HISTORICAL_ARTIFACT_PATH.read_bytes().rstrip()).hexdigest()
|
|
== historical_artifact_digest()
|
|
)
|
|
|
|
|
|
def test_artifact_is_machine_readable_and_matches_registry():
|
|
artifact = json.loads(HISTORICAL_ARTIFACT_PATH.read_text())
|
|
assert artifact["definition_count"] == 145
|
|
assert artifact["variant_count"] == 1107
|
|
assert historical_artifact_bytes() == HISTORICAL_ARTIFACT_PATH.read_bytes().rstrip()
|
|
|
|
|
|
def test_recovered_boundary_grids_are_not_truncated_or_reclassified():
|
|
assert len([v for v in HISTORICAL_REGISTRY.variants if v.indicator_id == 14]) == 8
|
|
assert len([v for v in HISTORICAL_REGISTRY.variants if v.indicator_id == 28]) == 3
|
|
assert len([v for v in HISTORICAL_REGISTRY.variants if v.indicator_id == 41]) == 8
|
|
clustering = [v for v in HISTORICAL_REGISTRY.variants if v.indicator_id == 62]
|
|
assert [v.period for v in clustering] == [10, 15, 20, 25, 30]
|
|
assert {v.role for v in clustering} == {Role.TREND}
|
|
|
|
|
|
def test_aliases_and_unregistered_metadata_are_explicit():
|
|
assert HISTORICAL_REGISTRY.definition(25).alias_of == 1
|
|
assert HISTORICAL_REGISTRY.definition(26).status.value == "alias"
|
|
assert HISTORICAL_REGISTRY.definition(27).status.value == "alias"
|
|
assert "MOM_JERK" in UNREGISTERED_INDICATORS
|
|
|
|
|
|
def test_unknown_and_unsupported_variants_never_fall_back():
|
|
engine = IndicatorEngine()
|
|
unknown = engine.resolve(IndicatorVariant(9999, 14, 0.0, Role.OSC))
|
|
unsupported = engine.resolve(IndicatorVariant(30, 7, 0.0, Role.OSC))
|
|
assert unknown.code == RefusalCode.UNKNOWN_INDICATOR
|
|
assert unknown.all_nan_behavior is True
|
|
assert unsupported.code == RefusalCode.UNSUPPORTED_VARIANT
|
|
|
|
|
|
def test_hs22_parser_requires_registered_role_correct_variants():
|
|
state = parse_hs22(
|
|
{
|
|
"trend": [70, 10, 0],
|
|
"signal": [30, 6, 0],
|
|
"trigger": [0, 5, 0],
|
|
"confirm": [31, 5, 0],
|
|
"volatility": [50, 8, 0],
|
|
}
|
|
)
|
|
assert state.trend.role == Role.TREND
|
|
with pytest.raises(HS22SchemaError, match="no fallback"):
|
|
parse_hs22(
|
|
{
|
|
"trend": [70, 10, 0],
|
|
"signal": [30, 7, 0],
|
|
"trigger": [0, 5, 0],
|
|
"confirm": [31, 5, 0],
|
|
"volatility": [50, 8, 0],
|
|
}
|
|
)
|
|
|
|
|
|
def test_derived_and_experimental_universes_do_not_expand_to_historical_pool():
|
|
assert HS22_COHORT001_V1.namespace == "derived"
|
|
assert HS22_COHORT001_V1.variants == ()
|
|
assert EXPERIMENTAL_UNIVERSE.namespace == "experimental"
|
|
assert EXPERIMENTAL_UNIVERSE.variants == ()
|