76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
"""Strict schema parser and state container for a five-role HS22 selection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
|
|
from .definitions import IndicatorVariant, Role
|
|
from .registry import HS22_HISTORICAL_COMPLETE_V1, IndicatorUniverse
|
|
|
|
_ROLE_ORDER = (Role.TREND, Role.OSC, Role.LEVEL, Role.OSC, Role.FILTER)
|
|
_ROLE_NAMES = ("trend", "signal", "trigger", "confirm", "volatility")
|
|
|
|
|
|
class HS22SchemaError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class HS22State:
|
|
trend: IndicatorVariant
|
|
signal: IndicatorVariant
|
|
trigger: IndicatorVariant
|
|
confirm: IndicatorVariant
|
|
volatility: IndicatorVariant
|
|
tail: tuple[float, ...] = ()
|
|
|
|
def variants(self) -> tuple[IndicatorVariant, ...]:
|
|
return (self.trend, self.signal, self.trigger, self.confirm, self.volatility)
|
|
|
|
|
|
def _parse_variant(
|
|
value: object, expected_role: Role, universe: IndicatorUniverse
|
|
) -> IndicatorVariant:
|
|
if isinstance(value, Mapping):
|
|
try:
|
|
variant = IndicatorVariant(
|
|
int(value["indicator_id"]),
|
|
int(value["period"]),
|
|
float(value.get("p1", 0.0)),
|
|
Role(value["role"]),
|
|
)
|
|
except (KeyError, TypeError, ValueError) as error:
|
|
raise HS22SchemaError("invalid HS22 variant") from error
|
|
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)) and len(value) == 3:
|
|
try:
|
|
variant = IndicatorVariant(int(value[0]), int(value[1]), float(value[2]), expected_role)
|
|
except (TypeError, ValueError) as error:
|
|
raise HS22SchemaError("invalid positional HS22 variant") from error
|
|
else:
|
|
raise HS22SchemaError("HS22 variant must be an object or [id, period, p1]")
|
|
if variant.role != expected_role:
|
|
raise HS22SchemaError(f"HS22 role must be {expected_role.value}")
|
|
if not universe.contains(variant):
|
|
raise HS22SchemaError("unsupported HS22 variant; no fallback is permitted")
|
|
return variant
|
|
|
|
|
|
def parse_hs22(
|
|
value: object, universe: IndicatorUniverse = HS22_HISTORICAL_COMPLETE_V1
|
|
) -> HS22State:
|
|
if isinstance(value, Mapping):
|
|
items = tuple(
|
|
_parse_variant(value.get(name), role, universe)
|
|
for name, role in zip(_ROLE_NAMES, _ROLE_ORDER, strict=True)
|
|
)
|
|
tail = tuple(float(number) for number in value.get("tail", ()))
|
|
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)) and len(value) == 22:
|
|
items = tuple(
|
|
_parse_variant(value[index : index + 3], role, universe)
|
|
for index, role in zip(range(0, 15, 3), _ROLE_ORDER, strict=True)
|
|
)
|
|
tail = tuple(float(number) for number in value[15:])
|
|
else:
|
|
raise HS22SchemaError("HS22 must be a role mapping or exactly 22 positional values")
|
|
return HS22State(*items, tail=tail)
|