"""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)