Artifex/control_plane/trading_studio/indicators/definitions.py
2026-08-18 02:00:53 +07:00

264 lines
11 KiB
Python

"""Typed, data-only definitions for the recovered HS22 indicator inventory."""
from __future__ import annotations
from dataclasses import asdict, dataclass
from enum import StrEnum
from itertools import product
class Role(StrEnum):
LEVEL = "level"
OSC = "osc"
TREND = "trend"
FILTER = "filter"
class ImplementationStatus(StrEnum):
UNIMPLEMENTED = "unimplemented"
ALIAS = "alias"
UNREGISTERED = "unregistered"
@dataclass(frozen=True, slots=True)
class IndicatorDefinition:
indicator_id: int
name: str
role: Role | None
status: ImplementationStatus = ImplementationStatus.UNIMPLEMENTED
alias_of: int | None = None
behavior: str = "refuse_execution"
@dataclass(frozen=True, slots=True)
class IndicatorVariant:
indicator_id: int
period: int
p1: float
role: Role
def asdict(self) -> dict[str, object]:
return asdict(self)
# Recovered from the historical pool definition. This is deliberately data, not
# a dependency on or import of the reference runtime.
_SPECS: tuple[tuple[int, str, Role, tuple[int, ...], tuple[float, ...]], ...] = (
*(
(i, name, Role.LEVEL, tuple(range(start, stop, step)), params)
for i, name, start, stop, step, params in (
(0, "SMA", 5, 51, 3, (0.0,)),
(1, "EMA", 5, 51, 3, (0.0,)),
(2, "WMA", 5, 51, 3, (0.0,)),
(3, "HMA", 6, 51, 2, (0.0,)),
(4, "DEMA", 5, 41, 3, (0.0,)),
(5, "TEMA", 5, 36, 3, (0.0,)),
(6, "KAMA", 8, 36, 3, (0.0,)),
(7, "EHLERS_SS", 6, 31, 2, (0.0,)),
(8, "MCGINLEY", 8, 36, 3, (0.0,)),
(9, "JMA", 8, 36, 4, (-50.0, 0.0, 50.0, 100.0)),
(10, "T3", 8, 31, 3, (0.6, 0.7, 0.8, 0.9)),
(11, "ALMA", 8, 31, 3, (4.0, 6.0, 8.0)),
(12, "ZLEMA", 5, 41, 3, (0.0,)),
(13, "VIDYA", 8, 36, 4, (0.0,)),
(14, "FRAMA", 10, 41, 4, (0.0,)),
(15, "LSMA", 8, 41, 3, (0.0,)),
(16, "SWMA", 5, 31, 3, (0.0,)),
(17, "BB_UPPER", 14, 26, 2, (1.5, 2.0, 2.5)),
(18, "BB_LOWER", 14, 26, 2, (1.5, 2.0, 2.5)),
(19, "SUPERTREND", 8, 16, 2, (2.0, 3.0, 4.0)),
(20, "DONCHIAN_UPPER", 10, 30, 4, (0.0,)),
(21, "DONCHIAN_LOWER", 10, 30, 4, (0.0,)),
(22, "DONCHIAN_MID", 10, 30, 4, (0.0,)),
(23, "KELTNER_UPPER", 14, 26, 2, (1.0, 1.5, 2.0)),
(24, "KELTNER_LOWER", 14, 26, 2, (1.0, 1.5, 2.0)),
)
),
(25, "KELTNER_MID", Role.LEVEL, tuple(range(14, 26, 2)), (0.0,)),
(26, "ICHIMOKU_TENKAN", Role.LEVEL, tuple(range(7, 13, 2)), (0.0,)),
(27, "ICHIMOKU_KIJUN", Role.LEVEL, tuple(range(20, 30, 2)), (0.0,)),
(28, "PSAR", Role.LEVEL, (1,), (0.01, 0.02, 0.03)),
)
# The remaining recovered records use compact rows: id, name, role, periods, p1 values.
_COMPACT_SPECS = (
(30, "RSI", "osc", range(6, 28, 2)),
(31, "STOCH_K", "osc", range(5, 22, 2)),
(32, "STOCH_D", "osc", range(5, 22, 2)),
(33, "CCI", "osc", range(10, 30, 2)),
(34, "WILLIAMS_R", "osc", range(5, 22, 2)),
(35, "ROC", "osc", range(5, 21, 2)),
(36, "CMO", "osc", range(8, 22, 2)),
(37, "TRIX", "osc", range(8, 22, 2)),
(38, "PPO", "osc", range(16, 30, 2)),
(39, "MACD_HIST", "osc", range(20, 32, 2)),
(40, "MFI", "osc", range(10, 22, 2)),
(41, "DPO", "osc", range(10, 26, 2)),
(42, "PCTILE_RANK", "osc", range(14, 51, 6)),
(43, "ZSCORE", "osc", range(14, 51, 6)),
(44, "MINMAX", "osc", range(14, 51, 6)),
*(
(i, name, "osc", periods)
for i, name, periods in (
(90, "MICRO_BODY_RATIO", range(5, 21, 3)),
(91, "MICRO_UPPER_WICK", range(5, 21, 3)),
(92, "MICRO_LOWER_WICK", range(5, 21, 3)),
(93, "MICRO_BUY_PRESSURE", range(5, 21, 3)),
(94, "MICRO_SELL_PRESSURE", range(5, 21, 3)),
(95, "MICRO_STOP_HUNT", range(10, 21, 5)),
(96, "MICRO_FVG", (1,)),
(97, "MICRO_LIQ_SWEEP", range(5, 21, 4)),
(98, "TREND_BREAKOUT", range(10, 31, 5)),
(99, "TREND_PULLBACK", range(10, 31, 5)),
(100, "MOM_VELOCITY", range(5, 21, 3)),
(101, "MOM_ACCELERATION", range(5, 21, 3)),
(102, "MOM_COMPOSITE", range(12, 31, 4)),
(103, "MOM_NORMALIZED_ROC", range(5, 21, 3)),
(104, "CROSS_PRICE_VS_EMA", range(8, 31, 3)),
(105, "CROSS_EMA_SPREAD", range(14, 41, 4)),
(106, "CROSS_MOM_X_VOL", range(8, 21, 3)),
(107, "CROSS_VOL_ADJ_MOM", range(8, 21, 3)),
(108, "NORM_RETURN_ZSCORE", range(14, 51, 6)),
(109, "TIME_BARS_SINCE_HIGH", range(10, 51, 10)),
(110, "TIME_BARS_SINCE_LOW", range(10, 51, 10)),
(116, "REGIME_SKEWNESS", range(20, 61, 10)),
(52, "VOL_ATR_ZSCORE", range(10, 31, 5)),
(59, "VOL_RANGE_PCTILE", range(14, 31, 4)),
(60, "VOL_TR_MOMENTUM", range(10, 31, 5)),
)
),
*(
(i, name, "trend", periods)
for i, name, periods in (
(70, "ADX", range(10, 22, 2)),
(71, "AROON_UP", range(10, 30, 4)),
(72, "AROON_DOWN", range(10, 30, 4)),
(73, "LINREG_SLOPE", range(8, 31, 3)),
(74, "LINREG_R2", range(8, 31, 3)),
(75, "EFFICIENCY", range(8, 31, 3)),
(76, "ANGLE", range(8, 31, 3)),
(77, "DURATION", (1,)),
(78, "HURST", range(20, 61, 10)),
(80, "FRACTAL_DIM", range(20, 61, 10)),
(81, "TREND_PERSIST", range(10, 31, 5)),
(83, "MICRO_TREND_STRUCT", range(5, 21, 3)),
)
),
)
def _variants() -> tuple[IndicatorVariant, ...]:
specs = list(_SPECS)
specs.extend((i, n, Role(r), tuple(p), (0.0,)) for i, n, r, p in _COMPACT_SPECS)
specs.extend(
(
(79, "VARIANCE_RATIO", Role.TREND, tuple(range(20, 61, 10)), (2.0, 4.0, 8.0)),
(82, "AUTOCORR", Role.TREND, tuple(range(20, 51, 10)), (1.0, 3.0, 5.0)),
)
)
specs.extend(
(i, n, Role.FILTER, tuple(p), (0.0,))
for i, n, p in (
(50, "ATR", range(8, 22, 2)),
(51, "ATR_PCT", range(8, 22, 2)),
(53, "VOL_COMPRESSION", range(14, 31, 4)),
(54, "VOL_OF_VOL", range(14, 31, 4)),
(55, "VOL_PARKINSON", range(10, 31, 5)),
(56, "VOL_GARMAN_KLASS", range(10, 31, 5)),
(57, "VOL_ROGERS_SATCHELL", range(10, 31, 5)),
(58, "VOL_REALIZED", range(10, 31, 5)),
(61, "NORM_VOL_ZSCORE", range(14, 51, 6)),
(63, "MICRO_VOL_DELTA", range(5, 21, 3)),
(111, "ENTROPY_SHANNON", range(20, 61, 10)),
(112, "KURTOSIS", range(20, 61, 10)),
(113, "HALFLIFE", range(30, 61, 10)),
(115, "MA_COMPRESSION", range(10, 31, 5)),
(118, "TIME_HOUR_COS", (1,)),
(119, "TIME_SESSION_VOL", (1,)),
)
)
specs.extend(
(
(62, "MICRO_VOL_CLUSTERING", Role.TREND, tuple(range(10, 31, 5)), (0.0,)),
(117, "TIME_HOUR_SIN", Role.TREND, (1,), (0.0,)),
)
)
specs.append((114, "DC_EVENTS", Role.FILTER, tuple(range(20, 51, 10)), (10.0, 20.0, 50.0)))
specs.extend(
(
i,
n,
Role.LEVEL if role == "level" else Role.OSC if role == "osc" else Role.FILTER,
tuple(p),
params,
)
for i, n, role, p, params in (
(120, "PIVOT_CLASSIC", "level", range(30, 361, 30), (0.0,)),
(121, "PIVOT_R1", "level", range(30, 361, 30), (0.0,)),
(122, "PIVOT_S1", "level", range(30, 361, 30), (0.0,)),
(123, "PIVOT_DISTANCE", "osc", range(30, 361, 30), (0.0,)),
(124, "PREV_PERIOD_HIGH", "level", range(30, 361, 30), (0.0,)),
(125, "PREV_PERIOD_LOW", "level", range(30, 361, 30), (0.0,)),
(126, "PREV_PERIOD_CLOSE", "level", range(30, 361, 30), (0.0,)),
(127, "ROLLING_MEDIAN", "level", range(10, 41, 5), (0.0,)),
(128, "LINREG_CHAN_UPPER", "level", range(14, 31, 4), (1.5, 2.0, 2.5)),
(129, "LINREG_CHAN_LOWER", "level", range(14, 31, 4), (1.5, 2.0, 2.5)),
(130, "QUANTILE_UPPER", "level", range(14, 41, 6), (0.0,)),
(131, "QUANTILE_LOWER", "level", range(14, 41, 6), (0.0,)),
(135, "FRESH_BREAKOUT", "osc", range(10, 31, 5), (0.0,)),
(136, "FAILED_BREAKOUT", "osc", range(10, 31, 5), (0.0,)),
(137, "FIRST_PULLBACK", "osc", range(10, 31, 5), (0.0,)),
(138, "VOL_EXPANSION", "osc", range(10, 21, 5), (0.0,)),
(139, "INSIDE_BAR", "osc", (1,), (0.0,)),
(140, "COMPRESS_RELEASE", "osc", range(10, 21, 5), (0.0,)),
(141, "SWEEP_RECLAIM", "osc", range(10, 31, 5), (0.0,)),
(142, "CLOSE_IN_RANGE", "osc", (1,), (0.0,)),
(143, "CLOSE_IN_ROLLING_RANGE", "osc", range(10, 31, 5), (0.0,)),
(144, "DIST_RECENT_HIGH", "osc", range(10, 31, 5), (0.0,)),
(145, "DIST_RECENT_LOW", "osc", range(10, 31, 5), (0.0,)),
(146, "CHANNEL_POSITION", "osc", range(14, 31, 4), (0.0,)),
(147, "OC_VS_PRIOR", "osc", range(10, 31, 5), (0.0,)),
(150, "CHOP_INDEX", "filter", range(10, 31, 5), (0.0,)),
(151, "CANDLE_OVERLAP", "filter", range(10, 31, 5), (0.0,)),
(152, "NOISE_RATIO", "filter", range(10, 31, 5), (0.0,)),
(153, "WICK_INSTABILITY", "filter", range(10, 31, 5), (0.0,)),
(154, "FALSE_BREAK_FREQ", "filter", range(10, 31, 10), (0.0,)),
(155, "SPREAD_PROXY", "filter", range(10, 31, 5), (0.0,)),
(156, "DIR_CLEAN", "filter", range(10, 31, 5), (0.0,)),
(157, "REVERSAL_FREQ", "filter", range(10, 31, 5), (0.0,)),
(158, "MEDIAN_EXCURSION", "filter", range(10, 31, 5), (0.0,)),
(160, "RSI_SLOPE", "osc", range(8, 22, 2), (0.0,)),
(161, "RSI_DIST_50", "osc", range(8, 22, 2), (0.0,)),
(162, "RSI_DIVERGENCE", "osc", range(10, 22, 2), (5.0, 10.0)),
(163, "MACD_SLOPE", "osc", range(20, 30, 2), (0.0,)),
(164, "MACD_DIVERGENCE", "osc", range(20, 30, 2), (0.0,)),
(165, "TIME_SINCE_OB", "osc", range(8, 22, 2), (0.0,)),
(166, "TIME_SINCE_OS", "osc", range(8, 22, 2), (0.0,)),
(167, "EXHAUSTION", "osc", range(10, 22, 2), (0.0,)),
(168, "LHLL_SCORE", "trend", range(5, 21, 3), (0.0,)),
)
)
return tuple(
IndicatorVariant(i, p, q, r) for i, _, r, ps, qs in specs for p, q in product(ps, qs)
)
HISTORICAL_VARIANTS = _variants()
assert len(HISTORICAL_VARIANTS) == 1107
_names = {i: n for i, n, *_ in _SPECS}
_names.update({i: n for i, n, *_ in _COMPACT_SPECS})
_names.update(
{
variant.indicator_id: _names.get(variant.indicator_id, f"ID_{variant.indicator_id}")
for variant in HISTORICAL_VARIANTS
}
)
HISTORICAL_DEFINITIONS = tuple(
IndicatorDefinition(
i,
_names.get(i, f"RESERVED_{i}"),
next((v.role for v in HISTORICAL_VARIANTS if v.indicator_id == i), None),
)
for i in sorted({v.indicator_id for v in HISTORICAL_VARIANTS})
)