from __future__ import annotations import json import math from bisect import bisect_right from dataclasses import asdict, dataclass, field from typing import Any from crypto_spot_bot.config import Settings from crypto_spot_bot.models import Candle DEFAULT_TORCH_FEATURES = ( "return_1", "return_3", "return_6", "return_12", "return_24", "range_percent", "body_percent", "upper_wick_percent", "lower_wick_percent", "volume_change", "volume_ratio", "volume_percentile_20", "atr_percent", "atr_ratio_20", "realized_volatility_12", "realized_volatility_24", "rsi_centered", "rsi_slope_6", "macd_hist_percent", "macd_hist_slope_3", "ema50_gap_percent", "ema200_gap_percent", "ema20_slope_6", "ema50_slope_12", "ema200_slope_24", "ema50_ema200_gap_percent", "range_position_50", "trend_return_4h", "trend_return_24h", "daily_close_ema200_gap_percent", "daily_ema50_ema200_gap_percent", "daily_ema50_slope", "btc_return_1", "btc_return_3", "btc_return_6", "btc_return_24", "eth_return_1", "eth_return_3", "eth_return_6", "eth_return_24", "relative_btc_return_3", "relative_eth_return_3", "btc_eth_return_spread_3", "pattern_score", "pattern_bullish", "pattern_bearish", "pattern_range", "pattern_pullback", "pattern_oversold_reversal", "pattern_stabilized_drop", "pattern_breakout", "pattern_breakdown", "pattern_fast_drop", "pattern_volume_spike", "pattern_range_position_20", ) FEATURE_DESCRIPTIONS: dict[str, tuple[str, str, str]] = { "return_1": ("Цена", "Доходность 1ч", "Изменение цены закрытия за последнюю 1h свечу."), "return_3": ("Цена", "Доходность 3ч", "Изменение цены закрытия за последние 3 часовые свечи."), "return_6": ("Цена", "Доходность 6ч", "Изменение цены закрытия за последние 6 часовых свечей."), "return_12": ("Цена", "Доходность 12ч", "Изменение цены закрытия за последние 12 часовых свечей."), "return_24": ("Цена", "Доходность 24ч", "Изменение цены закрытия за последние 24 часовые свечи."), "range_percent": ("Свеча", "Диапазон свечи", "Размер high-low последней свечи относительно цены закрытия."), "body_percent": ("Свеча", "Тело свечи", "Разница close-open относительно цены закрытия; знак показывает цвет свечи."), "upper_wick_percent": ("Свеча", "Верхняя тень", "Насколько далеко цена уходила выше тела свечи."), "lower_wick_percent": ("Свеча", "Нижняя тень", "Насколько далеко цена уходила ниже тела свечи."), "volume_change": ("Объем", "Изменение объема", "Изменение объема последней свечи относительно предыдущей."), "volume_ratio": ("Объем", "Объем к MA20", "Отклонение текущего объема от средней за 20 свечей."), "volume_percentile_20": ("Объем", "Процентиль объема", "Позиция текущего объема среди последних 20 свечей."), "atr_percent": ("Волатильность", "ATR14 %", "Средний торговый диапазон ATR14 относительно цены."), "atr_ratio_20": ("Волатильность", "ATR к среднему", "Отклонение текущего ATR от среднего ATR за 20 свечей."), "realized_volatility_12": ("Волатильность", "Реализованная вола 12ч", "Фактическая волатильность доходностей за 12 свечей."), "realized_volatility_24": ("Волатильность", "Реализованная вола 24ч", "Фактическая волатильность доходностей за 24 свечи."), "rsi_centered": ("Индикаторы", "RSI14 от 50", "RSI14, приведенный к центру 50: выше нуля сильнее покупатели."), "rsi_slope_6": ("Индикаторы", "Наклон RSI 6ч", "Изменение RSI14 за последние 6 свечей."), "macd_hist_percent": ("Индикаторы", "MACD histogram", "MACD histogram относительно цены; знак показывает импульс."), "macd_hist_slope_3": ("Индикаторы", "Наклон MACD hist", "Изменение MACD histogram за последние 3 свечи."), "ema50_gap_percent": ("EMA/тренд", "Цена к EMA50", "Расстояние цены закрытия до EMA50."), "ema200_gap_percent": ("EMA/тренд", "Цена к EMA200", "Расстояние цены закрытия до EMA200."), "ema20_slope_6": ("EMA/тренд", "Наклон EMA20", "Изменение EMA20 за последние 6 свечей."), "ema50_slope_12": ("EMA/тренд", "Наклон EMA50", "Изменение EMA50 за последние 12 свечей."), "ema200_slope_24": ("EMA/тренд", "Наклон EMA200", "Изменение EMA200 за последние 24 свечи."), "ema50_ema200_gap_percent": ("EMA/тренд", "EMA50 к EMA200", "Расстояние EMA50 относительно EMA200."), "range_position_50": ("Цена", "Позиция в диапазоне 50ч", "Где текущая цена внутри high-low диапазона последних 50 свечей."), "trend_return_4h": ("Цена", "Тренд 4ч", "Изменение цены за последние 4 свечи."), "trend_return_24h": ("Цена", "Тренд 24ч", "Изменение цены за последние 24 свечи."), "daily_close_ema200_gap_percent": ("Дневной тренд", "D цена к EMA200", "Расстояние дневного close до дневной EMA200."), "daily_ema50_ema200_gap_percent": ("Дневной тренд", "D EMA50 к EMA200", "Расстояние дневной EMA50 относительно дневной EMA200."), "daily_ema50_slope": ("Дневной тренд", "D наклон EMA50", "Изменение дневной EMA50 за последние несколько дневных свечей."), "btc_return_1": ("BTC/ETH контекст", "BTC 1ч", "Изменение BTCUSDT за последнюю 1h свечу."), "btc_return_3": ("BTC/ETH контекст", "BTC 3ч", "Изменение BTCUSDT за последние 3 часа."), "btc_return_6": ("BTC/ETH контекст", "BTC 6ч", "Изменение BTCUSDT за последние 6 часов."), "btc_return_24": ("BTC/ETH контекст", "BTC 24ч", "Изменение BTCUSDT за последние 24 часа."), "eth_return_1": ("BTC/ETH контекст", "ETH 1ч", "Изменение ETHUSDT за последнюю 1h свечу."), "eth_return_3": ("BTC/ETH контекст", "ETH 3ч", "Изменение ETHUSDT за последние 3 часа."), "eth_return_6": ("BTC/ETH контекст", "ETH 6ч", "Изменение ETHUSDT за последние 6 часов."), "eth_return_24": ("BTC/ETH контекст", "ETH 24ч", "Изменение ETHUSDT за последние 24 часа."), "relative_btc_return_3": ("BTC/ETH контекст", "Сила к BTC 3ч", "Доходность пары за 3 часа минус доходность BTC за 3 часа."), "relative_eth_return_3": ("BTC/ETH контекст", "Сила к ETH 3ч", "Доходность пары за 3 часа минус доходность ETH за 3 часа."), "btc_eth_return_spread_3": ("BTC/ETH контекст", "BTC-ETH 3ч", "Разница 3-часовой доходности BTC и ETH."), "pattern_score": ("Шаблон", "Оценка шаблона", "Числовая оценка текущего рыночного шаблона от 0 до 1."), "pattern_bullish": ("Шаблон", "Бычий шаблон", "Флаг, что текущий шаблон похож на бычий."), "pattern_bearish": ("Шаблон", "Медвежий шаблон", "Флаг, что текущий шаблон похож на медвежий."), "pattern_range": ("Шаблон", "Флэтовый шаблон", "Флаг, что рынок похож на диапазон/флэт."), "pattern_pullback": ("Шаблон", "Откат", "Флаг отката внутри восходящего движения."), "pattern_oversold_reversal": ("Шаблон", "Перепроданный разворот", "Флаг возможного разворота после перепроданности."), "pattern_stabilized_drop": ("Шаблон", "Падение стабилизировалось", "Флаг замедления падения и попытки стабилизации."), "pattern_breakout": ("Шаблон", "Пробой вверх", "Флаг пробоя верхней части диапазона с объемом."), "pattern_breakdown": ("Шаблон", "Пробой вниз", "Флаг пробоя нижней части диапазона с объемом."), "pattern_fast_drop": ("Шаблон", "Быстрое падение", "Флаг резкого снижения или сильной перепроданности."), "pattern_volume_spike": ("Шаблон", "Всплеск объема", "Флаг объема заметно выше обычного."), "pattern_range_position_20": ("Шаблон", "Позиция в диапазоне 20ч", "Где цена внутри high-low диапазона последних 20 свечей."), } @dataclass(slots=True) class TimeSeriesForecast: enabled: bool usable: bool model: str volatility_model: str expected_return_percent: float expected_price: float volatility_percent: float probability_up: float confidence_adjustment: float block_entry: bool validation_mae_percent: float baseline_mae_percent: float skill: float horizon: int reason: str expected_gross_return_percent: float quantile_10_percent: float quantile_50_percent: float quantile_90_percent: float conservative_return_percent: float target_transform: str feature_snapshot: list[dict[str, Any]] = field(default_factory=list) horizon_forecasts: dict[str, Any] = field(default_factory=dict) candidates: list[dict[str, Any]] = field(default_factory=list) quality_gate_passed: bool | None = None quality_gate: dict[str, Any] = field(default_factory=dict) def as_dict(self) -> dict[str, Any]: return asdict(self) class TimeSeriesForecaster: def __init__(self, settings: Settings): self.settings = settings self._lstm_artifact_mtime: float | None = None self._lstm_artifact: dict[str, Any] = {} self._calibration_mtime: float | None = None self._quality_gate: dict[str, Any] = {} def forecast( self, candles: list[Candle], symbol: str | None = None, *, market_candles: dict[str, list[Candle]] | None = None, trend_candles: list[Candle] | None = None, ) -> TimeSeriesForecast: if not self.settings.time_series_forecast_enabled: return _empty_forecast(False, "time-series forecast is disabled") closes = [float(candle.close) for candle in candles if candle.close > 0] min_candles = max(30, self.settings.time_series_min_candles) if len(closes) < min_candles: return _empty_forecast(True, "not enough candles for PyTorch forecast") returns = _log_returns(closes) if len(returns) < 20: return _empty_forecast(True, "not enough returns for PyTorch forecast") artifact = self._load_lstm_artifact() quality_gate = self._load_quality_gate() quality_gate_passed = _quality_gate_passed(quality_gate) entry = _torch_recurrent_entry(symbol, artifact) model = _torch_recurrent_model_name(symbol, artifact) clip = _clamp(_float_entry(entry or {}, "clip", 8.0), 1.0, 50.0) feature_rows = ( _feature_matrix( candles, _feature_names(entry), symbol=symbol, market_candles=market_candles, trend_candles=trend_candles, ) if entry else [] ) feature_snapshot = _feature_snapshot(feature_rows, entry, clip) if not model or not _can_use_torch_recurrent(returns, symbol, artifact, feature_rows): return _empty_forecast(True, "no valid PyTorch LSTM/GRU model for symbol") prediction = _torch_recurrent_predict( returns, symbol, artifact, feature_rows=feature_rows, closes=closes, candles=candles, ) if entry is None or prediction is None: return _empty_forecast(True, "PyTorch LSTM/GRU model could not build a forecast") if isinstance(prediction, dict): selected = _select_horizon_prediction( prediction, _entry_horizon(entry, self.settings.time_series_forecast_horizon), ) if not selected: return _empty_forecast(True, "PyTorch LSTM/GRU model could not select a forecast horizon") expected_return = float(selected["expected_return"]) expected_gross_return = float(selected.get("expected_gross_return", expected_return)) expected_price = closes[-1] * math.exp(expected_gross_return) probability_up = _clamp(float(selected.get("probability_up", 0.5)), 0.0, 1.0) model_mae = max(float(selected.get("validation_mae", 0.0)), 1e-9) baseline_mae = max(float(selected.get("baseline_mae", model_mae)), model_mae) uncertainty = max(float(selected.get("uncertainty", model_mae)), 1e-9) volatility_percent = uncertainty * 100 expected_return_percent = (math.exp(expected_return) - 1) * 100 expected_gross_return_percent = (math.exp(expected_gross_return) - 1) * 100 q10_percent = (math.exp(float(selected.get("q10", expected_return))) - 1) * 100 q50_percent = (math.exp(float(selected.get("q50", expected_return))) - 1) * 100 q90_percent = (math.exp(float(selected.get("q90", expected_return))) - 1) * 100 skill = _clamp(_float_entry(entry, "skill", 0.0), -1.0, 1.0) horizon = int(selected.get("horizon", _entry_horizon(entry, self.settings.time_series_forecast_horizon))) min_edge = max(0.0, self.settings.time_series_min_edge_percent) confidence_adjustment = _confidence_adjustment( expected_return_percent=expected_return_percent, probability_up=probability_up, skill=skill, min_edge=min_edge, max_adjustment=self.settings.time_series_max_adjustment, ) conservative_return_percent = min(expected_return_percent, q50_percent) block_entry = bool( (expected_return_percent <= -min_edge and probability_up <= 0.45) or (q50_percent <= -min_edge and probability_up <= 0.48) ) reason = _reason( model=model, expected_return_percent=expected_return_percent, probability_up=probability_up, skill=skill, block_entry=block_entry, ) return TimeSeriesForecast( enabled=True, usable=True, model=model, volatility_model="probabilistic multi-horizon after-cost quantile", expected_return_percent=round(expected_return_percent, 4), expected_price=round(expected_price, 8), volatility_percent=round(volatility_percent, 4), probability_up=round(probability_up, 4), confidence_adjustment=round(confidence_adjustment, 4), block_entry=block_entry, validation_mae_percent=round(model_mae * 100, 4), baseline_mae_percent=round(baseline_mae * 100, 4), skill=round(skill, 4), horizon=horizon, reason=reason, expected_gross_return_percent=round(expected_gross_return_percent, 4), quantile_10_percent=round(q10_percent, 4), quantile_50_percent=round(q50_percent, 4), quantile_90_percent=round(q90_percent, 4), conservative_return_percent=round(conservative_return_percent, 4), target_transform=str(entry.get("target_transform", "net_return_over_volatility")), feature_snapshot=feature_snapshot, horizon_forecasts=_public_horizon_forecasts(prediction), candidates=[{"model": model, "mae_percent": round(model_mae * 100, 4)}], quality_gate_passed=quality_gate_passed, quality_gate=quality_gate, ) direct_horizon = _is_direct_horizon(entry) horizon = _entry_horizon(entry, self.settings.time_series_forecast_horizon) expected_return = prediction if direct_horizon else prediction * horizon expected_price = closes[-1] * math.exp(expected_return) model_mae = _torch_validation_mae(entry, returns) baseline_mae = max(_float_entry(entry, "baseline_mae_percent", model_mae * 100) / 100, model_mae) if direct_horizon: uncertainty = max(model_mae, _horizon_return_scale(closes, horizon) * 0.25, 1e-9) volatility_model = "direct horizon validation MAE" else: uncertainty_one_step = max(model_mae, _return_scale(returns) * 0.25, 1e-9) uncertainty = uncertainty_one_step * math.sqrt(horizon) volatility_model = "one-step validation MAE scaled by horizon" volatility_percent = uncertainty * 100 expected_return_percent = (math.exp(expected_return) - 1) * 100 probability_up = _normal_cdf(expected_return / max(uncertainty, 1e-9)) skill = _clamp(_float_entry(entry, "skill", 0.0), -1.0, 1.0) min_edge = max(0.0, self.settings.time_series_min_edge_percent) confidence_adjustment = _confidence_adjustment( expected_return_percent=expected_return_percent, probability_up=probability_up, skill=skill, min_edge=min_edge, max_adjustment=self.settings.time_series_max_adjustment, ) block_entry = bool(expected_return_percent <= -min_edge and probability_up <= 0.45) reason = _reason( model=model, expected_return_percent=expected_return_percent, probability_up=probability_up, skill=skill, block_entry=block_entry, ) return TimeSeriesForecast( enabled=True, usable=True, model=model, volatility_model=volatility_model, expected_return_percent=round(expected_return_percent, 4), expected_price=round(expected_price, 8), volatility_percent=round(volatility_percent, 4), probability_up=round(probability_up, 4), confidence_adjustment=round(confidence_adjustment, 4), block_entry=block_entry, validation_mae_percent=round(model_mae * 100, 4), baseline_mae_percent=round(baseline_mae * 100, 4), skill=round(skill, 4), horizon=horizon, reason=reason, expected_gross_return_percent=round(expected_return_percent, 4), quantile_10_percent=round(expected_return_percent - volatility_percent, 4), quantile_50_percent=round(expected_return_percent, 4), quantile_90_percent=round(expected_return_percent + volatility_percent, 4), conservative_return_percent=round(expected_return_percent, 4), target_transform=str(entry.get("target_transform", "direct_log_return")), feature_snapshot=feature_snapshot, horizon_forecasts={}, candidates=[{"model": model, "mae_percent": round(model_mae * 100, 4)}], quality_gate_passed=quality_gate_passed, quality_gate=quality_gate, ) def _load_lstm_artifact(self) -> dict[str, Any]: if not self.settings.time_series_lstm_enabled: return {} path = self.settings.time_series_lstm_model_path try: stat = path.stat() except OSError: self._lstm_artifact_mtime = None self._lstm_artifact = {} return {} if self._lstm_artifact_mtime == stat.st_mtime: return self._lstm_artifact try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): data = {} self._lstm_artifact = data if isinstance(data, dict) else {} self._lstm_artifact_mtime = stat.st_mtime return self._lstm_artifact def _load_quality_gate(self) -> dict[str, Any]: path = self.settings.time_series_lstm_model_path.parent / "torch_threshold_calibration.json" try: stat = path.stat() except OSError: self._calibration_mtime = None self._quality_gate = {} return {} if self._calibration_mtime == stat.st_mtime: return self._quality_gate try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): data = {} validation = data.get("validation") if isinstance(data, dict) else {} self._quality_gate = validation if isinstance(validation, dict) else {} self._calibration_mtime = stat.st_mtime return self._quality_gate def _empty_forecast(enabled: bool, reason: str) -> TimeSeriesForecast: return TimeSeriesForecast( enabled=enabled, usable=False, model="none", volatility_model="none", expected_return_percent=0.0, expected_price=0.0, volatility_percent=0.0, probability_up=0.5, confidence_adjustment=0.0, block_entry=False, validation_mae_percent=0.0, baseline_mae_percent=0.0, skill=0.0, horizon=0, reason=reason, expected_gross_return_percent=0.0, quantile_10_percent=0.0, quantile_50_percent=0.0, quantile_90_percent=0.0, conservative_return_percent=0.0, target_transform="none", feature_snapshot=[], horizon_forecasts={}, candidates=[], quality_gate_passed=None, quality_gate={}, ) def _quality_gate_passed(quality_gate: dict[str, Any]) -> bool | None: if not quality_gate: return None if "passed" in quality_gate: return bool(quality_gate.get("passed")) status = str(quality_gate.get("status", "")).strip().lower() if status in {"pass", "passed", "ok"}: return True if status in {"fail", "failed", "warn"}: return False return None def _log_returns(closes: list[float]) -> list[float]: return [math.log(closes[index] / closes[index - 1]) for index in range(1, len(closes))] def _feature_matrix( candles: list[Candle], feature_names: list[str] | tuple[str, ...] | None = None, *, symbol: str | None = None, market_candles: dict[str, list[Candle]] | None = None, trend_candles: list[Candle] | None = None, ) -> list[list[float]]: names = list(feature_names or DEFAULT_TORCH_FEATURES) context = _feature_context( candles, symbol=symbol, market_candles=market_candles, trend_candles=trend_candles, ) rows: list[list[float]] = [] for index, candle in enumerate(candles): rows.append([_feature_value(name, candles, index, candle, context) for name in names]) return rows def _feature_context( candles: list[Candle], *, symbol: str | None, market_candles: dict[str, list[Candle]] | None, trend_candles: list[Candle] | None, ) -> dict[str, Any]: market_candles = market_candles or {} normalized_market = {key.upper(): value for key, value in market_candles.items()} context_indexes = { key: {candle.timestamp: index for index, candle in enumerate(rows)} for key, rows in normalized_market.items() } trend_rows = trend_candles or [] trend_timestamps = [candle.timestamp for candle in trend_rows] trend_positions = [ bisect_right(trend_timestamps, candle.timestamp) - 1 if trend_timestamps else -1 for candle in candles ] return { "symbol": (symbol or "").upper(), "market_candles": normalized_market, "context_indexes": context_indexes, "trend_candles": trend_rows, "trend_positions": trend_positions, } def _feature_value(name: str, candles: list[Candle], index: int, candle: Candle, context: dict[str, Any]) -> float: close = max(float(candle.close), 1e-12) previous = candles[index - 1] if index >= 1 else candle if name == "return_1": return _log_change(candle.close, previous.close) if name == "return_3": return _log_change(candle.close, candles[index - 3].close) if index >= 3 else 0.0 if name == "return_6": return _log_change(candle.close, candles[index - 6].close) if index >= 6 else 0.0 if name == "return_12": return _log_change(candle.close, candles[index - 12].close) if index >= 12 else 0.0 if name == "return_24": return _log_change(candle.close, candles[index - 24].close) if index >= 24 else 0.0 if name == "range_percent": return _safe_feature((candle.high - candle.low) / close) if name == "body_percent": return _safe_feature((candle.close - candle.open) / close) if name == "upper_wick_percent": return _safe_feature((candle.high - max(candle.open, candle.close)) / close) if name == "lower_wick_percent": return _safe_feature((min(candle.open, candle.close) - candle.low) / close) if name == "volume_change": return _log_change(max(candle.volume, 1e-12), max(previous.volume, 1e-12)) if name == "volume_ratio": return _safe_feature((candle.volume / candle.volume_ma_20) - 1.0) if candle.volume_ma_20 else 0.0 if name == "volume_percentile_20": return _rolling_percentile([row.volume for row in candles], index, 20) if name == "atr_percent": return _safe_feature(candle.atr_14 / close) if candle.atr_14 is not None else 0.0 if name == "atr_ratio_20": return _ratio_to_recent_mean( [row.atr_14 if row.atr_14 is not None else 0.0 for row in candles], index, 20, ) if name == "realized_volatility_12": return _realized_volatility(candles, index, 12) if name == "realized_volatility_24": return _realized_volatility(candles, index, 24) if name == "rsi_centered": return _safe_feature((candle.rsi_14 - 50.0) / 50.0) if candle.rsi_14 is not None else 0.0 if name == "rsi_slope_6": return _indicator_slope(candles, index, "rsi_14", 6, divisor=50.0) if name == "macd_hist_percent": return _safe_feature(candle.macd_hist / close) if candle.macd_hist is not None else 0.0 if name == "macd_hist_slope_3": return _indicator_price_slope(candles, index, "macd_hist", 3) if name == "ema50_gap_percent": return _safe_feature((candle.close - candle.ema_50) / close) if candle.ema_50 is not None else 0.0 if name == "ema200_gap_percent": return _safe_feature((candle.close - candle.ema_200) / close) if candle.ema_200 is not None else 0.0 if name == "ema20_slope_6": return _ema_slope(candles, index, "ema_20", 6) if name == "ema50_slope_12": return _ema_slope(candles, index, "ema_50", 12) if name == "ema200_slope_24": return _ema_slope(candles, index, "ema_200", 24) if name == "ema50_ema200_gap_percent": if candle.ema_50 is not None and candle.ema_200 is not None and candle.ema_200 > 0: return _safe_feature((candle.ema_50 - candle.ema_200) / candle.ema_200) return 0.0 if name == "range_position_50": return _range_position(candles, index, 50) if name == "trend_return_4h": return _log_change(candle.close, candles[index - 4].close) if index >= 4 else 0.0 if name == "trend_return_24h": return _log_change(candle.close, candles[index - 24].close) if index >= 24 else 0.0 if name.startswith("daily_"): return _daily_feature_value(name, context, index) if name.startswith("btc_") or name.startswith("eth_") or name.startswith("relative_"): return _cross_asset_feature_value(name, candles, index, candle, context) if name.startswith("pattern_"): return _pattern_feature_value(name, candles, index) return 0.0 def _pattern_feature_value(name: str, candles: list[Candle], index: int) -> float: pattern = _pattern_snapshot(candles, index) if name == "pattern_score": return pattern["score"] if name == "pattern_bullish": return pattern["bullish"] if name == "pattern_bearish": return pattern["bearish"] if name == "pattern_range": return pattern["range"] if name == "pattern_pullback": return pattern["pullback"] if name == "pattern_oversold_reversal": return pattern["oversold_reversal"] if name == "pattern_stabilized_drop": return pattern["stabilized_drop"] if name == "pattern_breakout": return pattern["breakout"] if name == "pattern_breakdown": return pattern["breakdown"] if name == "pattern_fast_drop": return pattern["fast_drop"] if name == "pattern_volume_spike": return pattern["volume_spike"] if name == "pattern_range_position_20": return pattern["range_position_20"] return 0.0 def _rolling_percentile(values: list[float], index: int, window: int) -> float: start = max(0, index - window + 1) sample = [float(value) for value in values[start : index + 1] if math.isfinite(float(value))] if not sample: return 0.5 current = float(values[index]) below_or_equal = sum(1 for value in sample if value <= current) return _clamp(below_or_equal / len(sample), 0.0, 1.0) def _ratio_to_recent_mean(values: list[float], index: int, window: int) -> float: current = float(values[index]) if index < len(values) else 0.0 start = max(0, index - window + 1) sample = [float(value) for value in values[start : index + 1] if math.isfinite(float(value)) and value > 0] if current <= 0 or not sample: return 0.0 mean = sum(sample) / len(sample) return _safe_feature((current / mean) - 1.0) if mean > 0 else 0.0 def _realized_volatility(candles: list[Candle], index: int, window: int) -> float: if index < 1: return 0.0 start = max(1, index - window + 1) returns = [ _log_change(candles[position].close, candles[position - 1].close) for position in range(start, index + 1) ] if not returns: return 0.0 return _safe_feature(math.sqrt(sum(value * value for value in returns) / len(returns))) def _indicator_slope(candles: list[Candle], index: int, attr: str, steps: int, *, divisor: float) -> float: if index < steps: return 0.0 current = getattr(candles[index], attr) previous = getattr(candles[index - steps], attr) if current is None or previous is None or divisor <= 0: return 0.0 return _safe_feature((float(current) - float(previous)) / divisor) def _indicator_price_slope(candles: list[Candle], index: int, attr: str, steps: int) -> float: if index < steps: return 0.0 current = getattr(candles[index], attr) previous = getattr(candles[index - steps], attr) close = max(float(candles[index].close), 1e-12) if current is None or previous is None: return 0.0 return _safe_feature((float(current) - float(previous)) / close) def _ema_slope(candles: list[Candle], index: int, attr: str, steps: int) -> float: if index < steps: return 0.0 current = getattr(candles[index], attr) previous = getattr(candles[index - steps], attr) if current is None or previous is None or previous <= 0: return 0.0 return _safe_feature(math.log(float(current) / float(previous))) def _range_position(candles: list[Candle], index: int, window: int) -> float: start = max(0, index - window + 1) rows = candles[start : index + 1] if not rows: return 0.5 high = max(row.high for row in rows) low = min(row.low for row in rows) width = high - low if width <= 0: return 0.5 return _clamp((candles[index].close - low) / width, 0.0, 1.0) def _daily_feature_value(name: str, context: dict[str, Any], index: int) -> float: trend_rows: list[Candle] = context.get("trend_candles") or [] trend_positions: list[int] = context.get("trend_positions") or [] if index >= len(trend_positions): return 0.0 trend_index = trend_positions[index] if trend_index < 0 or trend_index >= len(trend_rows): return 0.0 trend = trend_rows[trend_index] if name == "daily_close_ema200_gap_percent": if trend.ema_200 is not None and trend.ema_200 > 0: return _safe_feature((trend.close - trend.ema_200) / trend.ema_200) return 0.0 if name == "daily_ema50_ema200_gap_percent": if trend.ema_50 is not None and trend.ema_200 is not None and trend.ema_200 > 0: return _safe_feature((trend.ema_50 - trend.ema_200) / trend.ema_200) return 0.0 if name == "daily_ema50_slope": previous_index = trend_index - 5 if previous_index >= 0 and trend.ema_50 is not None and trend_rows[previous_index].ema_50: return _safe_feature(math.log(trend.ema_50 / trend_rows[previous_index].ema_50)) return 0.0 def _cross_asset_feature_value( name: str, candles: list[Candle], index: int, candle: Candle, context: dict[str, Any], ) -> float: if name == "btc_eth_return_spread_3": return _safe_feature( _context_return("BTCUSDT", 3, candle.timestamp, context) - _context_return("ETHUSDT", 3, candle.timestamp, context) ) if name.startswith("btc_return_"): steps = int(name.rsplit("_", 1)[1]) return _context_return("BTCUSDT", steps, candle.timestamp, context) if name.startswith("eth_return_"): steps = int(name.rsplit("_", 1)[1]) return _context_return("ETHUSDT", steps, candle.timestamp, context) if name == "relative_btc_return_3": return _safe_feature((_log_change(candle.close, candles[index - 3].close) if index >= 3 else 0.0) - _context_return("BTCUSDT", 3, candle.timestamp, context)) if name == "relative_eth_return_3": return _safe_feature((_log_change(candle.close, candles[index - 3].close) if index >= 3 else 0.0) - _context_return("ETHUSDT", 3, candle.timestamp, context)) return 0.0 def _context_return(symbol: str, steps: int, timestamp: int, context: dict[str, Any]) -> float: rows = (context.get("market_candles") or {}).get(symbol) indexes = (context.get("context_indexes") or {}).get(symbol) if not rows or not indexes: return 0.0 index = indexes.get(timestamp) if index is None or index < steps: return 0.0 return _log_change(rows[index].close, rows[index - steps].close) def _pattern_snapshot(candles: list[Candle], index: int) -> dict[str, float]: if index < 29: return { "score": 0.0, "bullish": 0.0, "bearish": 0.0, "range": 0.0, "pullback": 0.0, "oversold_reversal": 0.0, "stabilized_drop": 0.0, "breakout": 0.0, "breakdown": 0.0, "fast_drop": 0.0, "volume_spike": 0.0, "range_position_20": 0.5, } window = candles[: index + 1] latest = window[-1] previous = window[-2] high20 = max(candle.high for candle in window[-20:]) low20 = min(candle.low for candle in window[-20:]) width20 = max(0.0, high20 - low20) range_position_20 = _clamp((latest.close - low20) / width20, 0.0, 1.0) if width20 else 0.5 close_3 = window[-4].close if len(window) >= 4 else window[0].close close_10 = window[-11].close if len(window) >= 11 else window[0].close close_20 = window[-21].close if len(window) >= 21 else window[0].close ret_3 = _percent_change(latest.close, close_3) ret_10 = _percent_change(latest.close, close_10) ret_20 = _percent_change(latest.close, close_20) body = abs(latest.close - latest.open) lower_wick = max(0.0, min(latest.open, latest.close) - latest.low) atr_percent = (latest.atr_14 / latest.close * 100) if latest.atr_14 and latest.close else 0.0 volume_ratio = ( latest.volume / latest.volume_ma_20 if latest.volume_ma_20 and latest.volume_ma_20 > 0 else 0.0 ) ema_gap_percent = ( (latest.ema_50 - latest.ema_200) / latest.ema_200 * 100 if latest.ema_50 and latest.ema_200 else 0.0 ) uptrend = bool( latest.ema_20 and latest.ema_50 and latest.ema_200 and latest.ema_20 >= latest.ema_50 >= latest.ema_200 and latest.close >= latest.ema_50 ) downtrend = bool( latest.ema_20 and latest.ema_50 and latest.ema_200 and latest.ema_20 <= latest.ema_50 <= latest.ema_200 and latest.close <= latest.ema_50 ) pullback = bool( latest.ema_20 and uptrend and latest.close <= latest.ema_20 * 1.012 and latest.rsi_14 is not None and 35 <= latest.rsi_14 <= 58 ) oversold_reversal = bool( latest.rsi_14 is not None and latest.rsi_14 <= 35 and latest.close > previous.close and lower_wick >= body * 1.2 ) stabilized_drop = _pattern_stabilized_drop( candles=window, latest=latest, previous=previous, ret_3=ret_3, ret_10=ret_10, ret_20=ret_20, atr_percent=atr_percent, volume_ratio=volume_ratio, lower_wick=lower_wick, body=body, ) breakout = bool(latest.close >= high20 * 0.995 and volume_ratio >= 1.15 and latest.close > latest.open) breakdown = bool(latest.close <= low20 * 1.005 and volume_ratio >= 1.1 and latest.close < latest.open) fast_drop = bool(ret_3 <= -max(1.2, atr_percent * 1.8) or (latest.rsi_14 or 100) <= 25) range_market = bool(abs(ret_20) <= max(0.8, atr_percent * 1.2) and abs(ema_gap_percent) <= 0.35) volume_spike = bool(volume_ratio >= 1.6) score = 0.50 if fast_drop and breakdown: score = 0.18 elif breakdown: score = 0.24 elif pullback: score = 0.76 elif oversold_reversal: score = 0.68 elif stabilized_drop: score = 0.58 elif breakout: score = 0.72 elif uptrend: score = 0.64 elif range_market: score = 0.48 elif downtrend: score = 0.28 bullish = float(pullback or oversold_reversal or stabilized_drop or breakout or uptrend) bearish = float((fast_drop and breakdown) or breakdown or downtrend) return { "score": score, "bullish": bullish, "bearish": bearish, "range": float(range_market), "pullback": float(pullback), "oversold_reversal": float(oversold_reversal), "stabilized_drop": float(stabilized_drop), "breakout": float(breakout), "breakdown": float(breakdown), "fast_drop": float(fast_drop), "volume_spike": float(volume_spike), "range_position_20": range_position_20, } def _percent_change(current: float, previous: float) -> float: return ((current - previous) / previous * 100) if previous else 0.0 def _pattern_stabilized_drop( *, candles: list[Candle], latest: Candle, previous: Candle, ret_3: float, ret_10: float, ret_20: float, atr_percent: float, volume_ratio: float, lower_wick: float, body: float, ) -> bool: recent_drop = ret_10 <= -max(0.35, atr_percent * 1.1) or ret_20 <= -max(0.6, atr_percent * 1.6) if not recent_drop or latest.rsi_14 is None or latest.rsi_14 > 52: return False recent_lows = [candle.low for candle in candles[-5:-1]] no_new_low = bool(recent_lows) and latest.low >= min(recent_lows) * 0.999 bounce_from_low = ((latest.close - min(candle.low for candle in candles[-6:])) / latest.close * 100) if latest.close else 0.0 body_base = max(body, latest.close * 0.0001) absorption = lower_wick >= body_base * 0.6 or bounce_from_low >= max(0.08, atr_percent * 0.3) momentum_stabilized = latest.close >= previous.close or abs(ret_3) <= max(0.25, atr_percent * 0.8) or no_new_low volume_present = volume_ratio >= 0.55 continuing_drop = latest.close < previous.close and not no_new_low and ret_3 <= -max(0.6, atr_percent * 1.2) return bool(momentum_stabilized and absorption and volume_present and not continuing_drop) def _log_change(current: float, previous: float) -> float: if current <= 0 or previous <= 0: return 0.0 return _safe_feature(math.log(current / previous)) def _safe_feature(value: float) -> float: if not math.isfinite(value): return 0.0 return _clamp(float(value), -50.0, 50.0) def _torch_recurrent_model_name(symbol: str | None, artifact: dict[str, Any]) -> str | None: entry = _torch_recurrent_entry(symbol, artifact) if not entry: return None architecture = str(entry.get("architecture", "")).strip().lower() if architecture in {"lstm", "gru"}: return f"torch_{architecture}" model = str(entry.get("model", "")).strip().lower() return model if model in {"torch_lstm", "torch_gru"} else None def _torch_recurrent_entry(symbol: str | None, artifact: dict[str, Any]) -> dict[str, Any] | None: if artifact.get("type") != "pytorch_recurrent_forecaster": return None symbols = artifact.get("symbols") entry = symbols.get(symbol.upper()) if symbol and isinstance(symbols, dict) else None if not isinstance(entry, dict): default = artifact.get("default") entry = default if isinstance(default, dict) else None if not isinstance(entry, dict): return None if not isinstance(entry.get("state_dict"), dict): return None return entry def _can_use_torch_recurrent( returns: list[float], symbol: str | None, artifact: dict[str, Any], feature_rows: list[list[float]] | None = None, ) -> bool: entry = _torch_recurrent_entry(symbol, artifact) if not entry: return False lookback = int(_clamp(_float_entry(entry, "lookback", 0.0), 4.0, 512.0)) hidden_size = int(_clamp(_float_entry(entry, "hidden_size", 0.0), 1.0, 512.0)) num_layers = int(_clamp(_float_entry(entry, "num_layers", 1.0), 1.0, 8.0)) if hidden_size <= 0 or num_layers <= 0: return False if _is_direct_horizon(entry): return bool(feature_rows and len(feature_rows) >= lookback) return len(returns) >= lookback + 1 def _torch_recurrent_predict( returns: list[float], symbol: str | None, artifact: dict[str, Any], *, feature_rows: list[list[float]] | None = None, closes: list[float] | None = None, candles: list[Candle] | None = None, ) -> float | dict[str, Any] | None: entry = _torch_recurrent_entry(symbol, artifact) model_name = _torch_recurrent_model_name(symbol, artifact) if not entry or not model_name: return None lookback = int(_clamp(_float_entry(entry, "lookback", 0.0), 4.0, 512.0)) hidden_size = int(_clamp(_float_entry(entry, "hidden_size", 0.0), 1.0, 512.0)) num_layers = int(_clamp(_float_entry(entry, "num_layers", 1.0), 1.0, 8.0)) clip = _clamp(_float_entry(entry, "clip", 8.0), 1.0, 50.0) direct_horizon = _is_direct_horizon(entry) if direct_horizon: rows = feature_rows or [] if len(rows) < lookback: return None sequence = _normalize_feature_rows(rows[-lookback:], entry, clip) target_mean = _float_entry(entry, "target_mean", 0.0) target_scale = max(_float_entry(entry, "target_scale", _return_scale(returns)), 1e-8) else: mean = _float_entry(entry, "mean", 0.0) scale = max(_float_entry(entry, "scale", _return_scale(returns)), 1e-8) if len(returns) < lookback: return None sequence = [[_clamp((value - mean) / scale, -clip, clip)] for value in returns[-lookback:]] target_mean = mean target_scale = scale try: hidden = _torch_recurrent_hidden( sequence, entry=entry, model_name=model_name, hidden_size=hidden_size, num_layers=num_layers, ) if hidden is None: return None head_outputs = _torch_head_outputs(hidden, entry, hidden_size) if not head_outputs: return None if _entry_target_horizons(entry): return _decode_multi_horizon_prediction( outputs=head_outputs, entry=entry, returns=returns, closes=closes or [], candles=candles or [], clip=clip, ) normalized_prediction = head_outputs[0] if not math.isfinite(normalized_prediction): return None prediction = _clamp(normalized_prediction, -clip, clip) * target_scale + target_mean except (IndexError, KeyError, TypeError, ValueError, OverflowError): return None if direct_horizon and closes: horizon = _entry_horizon(entry, 1) recent_abs = sorted(abs(value) for value in _horizon_log_returns(closes, horizon)[-48:]) else: recent_abs = sorted(abs(value) for value in returns[-48:]) if len(returns) >= 8 else [0.01] cap = max(recent_abs[int(len(recent_abs) * 0.9)] if recent_abs else 0.0, 0.0002) return _clamp(prediction, -cap, cap) def _torch_head_outputs(context: list[float], entry: dict[str, Any], hidden_size: int) -> list[float]: context = _apply_context_norm(context, entry) raw_weight = entry.get("head_weight") if isinstance(raw_weight, list) and raw_weight and isinstance(raw_weight[0], list): matrix = _float_matrix(raw_weight) bias = _float_vector(entry.get("head_bias")) if len(bias) != len(matrix): bias = [0.0 for _ in matrix] return [ _dot(row, context) + bias[index] for index, row in enumerate(matrix) if len(row) == hidden_size ] head_weight = _float_vector(raw_weight) head_bias = _float_entry(entry, "head_bias", 0.0) if len(head_weight) != hidden_size: return [] return [sum(weight * value for weight, value in zip(head_weight, context)) + head_bias] def _apply_context_norm(context: list[float], entry: dict[str, Any]) -> list[float]: weight = _float_vector(entry.get("context_norm_weight")) bias = _float_vector(entry.get("context_norm_bias")) if not weight or len(weight) != len(context): return context if len(bias) != len(context): bias = [0.0 for _ in context] mean = sum(context) / len(context) variance = sum((value - mean) ** 2 for value in context) / len(context) denominator = math.sqrt(variance + 1e-5) return [ ((value - mean) / denominator) * weight[index] + bias[index] for index, value in enumerate(context) ] def _decode_multi_horizon_prediction( *, outputs: list[float], entry: dict[str, Any], returns: list[float], closes: list[float], candles: list[Candle], clip: float, ) -> dict[str, Any] | None: horizons = _entry_target_horizons(entry) if not horizons: return None layout = _entry_output_layout(entry) group_size = len(layout) if len(outputs) < len(horizons) * group_size: return None target_means = _target_vector(entry, "target_means", "target_mean", len(horizons), 0.0) target_scales = _target_vector(entry, "target_scales", "target_scale", len(horizons), _return_scale(returns)) validation_mae = _target_vector(entry, "validation_mae_by_horizon", "validation_mae_percent", len(horizons), 0.0) baseline_mae = _target_vector(entry, "baseline_mae_by_horizon", "baseline_mae_percent", len(horizons), 0.0) round_trip_cost = max(0.0, _float_entry(entry, "round_trip_cost", 0.0)) result: dict[str, Any] = {"horizons": {}} for horizon_index, horizon in enumerate(horizons): base = horizon_index * group_size values = {layout[offset]: outputs[base + offset] for offset in range(group_size)} vol_scale = _current_volatility_scale(candles, closes, horizon) def decode(name: str, fallback: float = 0.0) -> float: normalized = _clamp(float(values.get(name, fallback)), -clip, clip) transformed = normalized * max(target_scales[horizon_index], 1e-8) + target_means[horizon_index] if str(entry.get("target_transform", "")) == "net_return_over_volatility": return transformed * vol_scale return transformed expected = decode("mean") q_values = sorted([decode("q10", expected), decode("q50", expected), decode("q90", expected)]) probability_up = _sigmoid(float(values.get("logit_up", 0.0))) cap = _prediction_cap(closes, horizon, round_trip_cost) expected = _clamp(expected, -cap, cap) q10 = _clamp(q_values[0], -cap, cap) q50 = _clamp(q_values[1], -cap, cap) q90 = _clamp(q_values[2], -cap, cap) mae = validation_mae[horizon_index] if mae > 1.0: mae = mae / 100 base_mae = baseline_mae[horizon_index] if base_mae > 1.0: base_mae = base_mae / 100 if mae <= 0: mae = max(_horizon_return_scale(closes, horizon), 1e-9) if base_mae <= 0: base_mae = max(mae, _horizon_return_scale(closes, horizon)) uncertainty = max(abs(q90 - q10) * 0.5, mae, 1e-9) result["horizons"][str(horizon)] = { "horizon": horizon, "expected_return": expected, "expected_gross_return": expected + round_trip_cost, "q10": q10, "q50": q50, "q90": q90, "probability_up": probability_up, "volatility_scale": vol_scale, "validation_mae": mae, "baseline_mae": base_mae, "uncertainty": uncertainty, } return result def _normalize_feature_rows(rows: list[list[float]], entry: dict[str, Any], clip: float) -> list[list[float]]: means = _float_vector(entry.get("feature_means")) scales = _float_vector(entry.get("feature_scales")) input_size = int(_clamp(_float_entry(entry, "input_size", len(rows[-1]) if rows else 1), 1.0, 256.0)) if len(means) != input_size: means = [0.0 for _ in range(input_size)] if len(scales) != input_size: scales = [1.0 for _ in range(input_size)] normalized = [] for row in rows: normalized.append( [ _clamp(((row[index] if index < len(row) else 0.0) - means[index]) / max(scales[index], 1e-8), -clip, clip) for index in range(input_size) ] ) return normalized def _feature_snapshot( feature_rows: list[list[float]], entry: dict[str, Any] | None, clip: float, ) -> list[dict[str, Any]]: if not entry or not feature_rows: return [] names = _feature_names(entry) latest = feature_rows[-1] normalized_rows = _normalize_feature_rows([latest], entry, clip) normalized = normalized_rows[-1] if normalized_rows else [] means = _float_vector(entry.get("feature_means")) scales = _float_vector(entry.get("feature_scales")) snapshot: list[dict[str, Any]] = [] for index, name in enumerate(names): raw_value = float(latest[index]) if index < len(latest) else 0.0 model_value = float(normalized[index]) if index < len(normalized) else 0.0 group, label, meaning = FEATURE_DESCRIPTIONS.get( name, ("Прочее", name, "Технический входной признак модели."), ) snapshot.append( { "name": name, "label": label, "group": group, "raw_value": round(raw_value, 10), "raw_display": _feature_raw_display(name, raw_value), "model_value": round(model_value, 4), "model_display": f"{model_value:+.2f}", "mean": round(float(means[index]), 10) if index < len(means) else 0.0, "scale": round(float(scales[index]), 10) if index < len(scales) else 1.0, "meaning": meaning, "interpretation": _feature_interpretation(name, raw_value, model_value), } ) return snapshot def _feature_raw_display(name: str, value: float) -> str: if _feature_is_log_percent(name): return f"{(math.exp(value) - 1) * 100:+.3f}%" if _feature_is_linear_percent(name): return f"{value * 100:+.3f}%" if name in {"rsi_centered"}: return f"RSI {value * 50 + 50:.1f}" if name in {"rsi_slope_6"}: return f"{value * 50:+.2f} RSI" if name in {"volume_percentile_20", "range_position_50", "pattern_range_position_20"}: return f"{value * 100:.1f}%" if name.startswith("pattern_") and name != "pattern_score": return "да" if value >= 0.5 else "нет" if name == "pattern_score": return f"{value:.2f}" return f"{value:+.4f}" def _feature_interpretation(name: str, value: float, model_value: float) -> str: norm = _model_value_text(model_value) if name.startswith("pattern_") and name != "pattern_score" and name != "pattern_range_position_20": state = "шаблон активен" if value >= 0.5 else "шаблон не активен" return f"{state}; {norm}." if name in {"volume_percentile_20", "range_position_50", "pattern_range_position_20"}: if value >= 0.8: state = "значение находится в верхней части диапазона" elif value <= 0.2: state = "значение находится в нижней части диапазона" else: state = "значение около середины диапазона" return f"{state}; {norm}." if name in {"volume_ratio", "atr_ratio_20"}: state = "выше среднего" if value > 0 else "ниже среднего" if value < 0 else "около среднего" return f"{state}; {norm}." if name == "rsi_centered": rsi = value * 50 + 50 if rsi >= 65: state = "RSI высокий" elif rsi <= 35: state = "RSI низкий" else: state = "RSI в средней зоне" return f"{state}; {norm}." if _feature_is_log_percent(name) or _feature_is_linear_percent(name) or name.endswith("_slope"): if value > 0: state = "положительное значение" elif value < 0: state = "отрицательное значение" else: state = "нейтральное значение" return f"{state}; {norm}." if name == "pattern_score": if value >= 0.65: state = "шаблон скорее поддерживает long" elif value <= 0.35: state = "шаблон скорее против long" else: state = "шаблон нейтральный" return f"{state}; {norm}." return norm + "." def _model_value_text(value: float) -> str: magnitude = abs(value) if magnitude >= 2.0: return "для модели это сильное отклонение от обучающей нормы" if magnitude >= 1.0: return "для модели это заметное отклонение от обучающей нормы" return "для модели это близко к обычному диапазону" def _feature_is_log_percent(name: str) -> bool: return ( name.startswith("return_") or name.startswith("btc_return_") or name.startswith("eth_return_") or name.startswith("relative_") or name in { "volume_change", "trend_return_4h", "trend_return_24h", "ema20_slope_6", "ema50_slope_12", "ema200_slope_24", "daily_ema50_slope", "btc_eth_return_spread_3", } ) def _feature_is_linear_percent(name: str) -> bool: return name in { "range_percent", "body_percent", "upper_wick_percent", "lower_wick_percent", "volume_ratio", "atr_percent", "atr_ratio_20", "realized_volatility_12", "realized_volatility_24", "macd_hist_percent", "macd_hist_slope_3", "ema50_gap_percent", "ema200_gap_percent", "ema50_ema200_gap_percent", "daily_close_ema200_gap_percent", "daily_ema50_ema200_gap_percent", } def _torch_recurrent_hidden( sequence: list[list[float]], *, entry: dict[str, Any], model_name: str, hidden_size: int, num_layers: int, ) -> list[float] | None: state = entry.get("state_dict") if not isinstance(state, dict): return None h_layers = [[0.0 for _ in range(hidden_size)] for _ in range(num_layers)] c_layers = [[0.0 for _ in range(hidden_size)] for _ in range(num_layers)] top_outputs: list[list[float]] = [] for row in sequence: layer_input = list(row) for layer in range(num_layers): if model_name == "torch_lstm": next_hidden, next_cell = _torch_lstm_step(layer_input, h_layers[layer], c_layers[layer], state, layer) h_layers[layer] = next_hidden c_layers[layer] = next_cell elif model_name == "torch_gru": h_layers[layer] = _torch_gru_step(layer_input, h_layers[layer], state, layer) else: return None layer_input = h_layers[layer] top_outputs.append(list(layer_input)) if not top_outputs: return None if bool(entry.get("attention_pooling")): return _attention_context(top_outputs, entry, hidden_size) return top_outputs[-1] def _attention_context(outputs: list[list[float]], entry: dict[str, Any], hidden_size: int) -> list[float] | None: weight = _float_vector(entry.get("attention_weight")) if len(weight) != hidden_size: return outputs[-1] if outputs else None bias = _float_entry(entry, "attention_bias", 0.0) scores = [_dot(weight, row) + bias for row in outputs] if not scores: return None max_score = max(scores) exps = [math.exp(_clamp(score - max_score, -50.0, 50.0)) for score in scores] total = sum(exps) if total <= 0: return outputs[-1] attention = [value / total for value in exps] return [ sum(attention[row_index] * outputs[row_index][hidden_index] for row_index in range(len(outputs))) for hidden_index in range(hidden_size) ] def _torch_lstm_step( inputs: list[float], hidden: list[float], cell: list[float], state: dict[str, Any], layer: int, ) -> tuple[list[float], list[float]]: hidden_size = len(hidden) gates = _torch_gate_values(inputs, hidden, state, layer, gate_count=4) input_gate = [_sigmoid(value) for value in gates[0]] forget_gate = [_sigmoid(value) for value in gates[1]] cell_gate = [math.tanh(value) for value in gates[2]] output_gate = [_sigmoid(value) for value in gates[3]] next_cell = [ forget_gate[index] * cell[index] + input_gate[index] * cell_gate[index] for index in range(hidden_size) ] next_hidden = [ output_gate[index] * math.tanh(next_cell[index]) for index in range(hidden_size) ] return next_hidden, next_cell def _torch_gru_step( inputs: list[float], hidden: list[float], state: dict[str, Any], layer: int, ) -> list[float]: hidden_size = len(hidden) weight_ih = _float_matrix(state[f"weight_ih_l{layer}"]) weight_hh = _float_matrix(state[f"weight_hh_l{layer}"]) bias_ih = _float_vector(state[f"bias_ih_l{layer}"]) bias_hh = _float_vector(state[f"bias_hh_l{layer}"]) def gate_input(gate: int) -> list[float]: start = gate * hidden_size output = [] for index in range(hidden_size): row = start + index output.append(_dot(weight_ih[row], inputs) + bias_ih[row]) return output def gate_hidden(gate: int) -> list[float]: start = gate * hidden_size output = [] for index in range(hidden_size): row = start + index output.append(_dot(weight_hh[row], hidden) + bias_hh[row]) return output reset_input = gate_input(0) update_input = gate_input(1) new_input = gate_input(2) reset_hidden = gate_hidden(0) update_hidden = gate_hidden(1) new_hidden = gate_hidden(2) reset_gate = [_sigmoid(reset_input[index] + reset_hidden[index]) for index in range(hidden_size)] update_gate = [_sigmoid(update_input[index] + update_hidden[index]) for index in range(hidden_size)] candidate = [ math.tanh(new_input[index] + reset_gate[index] * new_hidden[index]) for index in range(hidden_size) ] return [ (1 - update_gate[index]) * candidate[index] + update_gate[index] * hidden[index] for index in range(hidden_size) ] def _torch_gate_values( inputs: list[float], hidden: list[float], state: dict[str, Any], layer: int, gate_count: int, ) -> list[list[float]]: hidden_size = len(hidden) weight_ih = _float_matrix(state[f"weight_ih_l{layer}"]) weight_hh = _float_matrix(state[f"weight_hh_l{layer}"]) bias_ih = _float_vector(state[f"bias_ih_l{layer}"]) bias_hh = _float_vector(state[f"bias_hh_l{layer}"]) gates: list[list[float]] = [] for gate in range(gate_count): values = [] start = gate * hidden_size for index in range(hidden_size): row = start + index values.append(_dot(weight_ih[row], inputs) + _dot(weight_hh[row], hidden) + bias_ih[row] + bias_hh[row]) gates.append(values) return gates def _torch_validation_mae(entry: dict[str, Any], returns: list[float]) -> float: mae_percent = _float_entry(entry, "validation_mae_percent", 0.0) if mae_percent > 0: return mae_percent / 100 return _return_scale(returns) def _feature_names(entry: dict[str, Any] | None) -> list[str]: if not entry: return list(DEFAULT_TORCH_FEATURES) names = entry.get("feature_names") if isinstance(names, list) and names: return [str(name) for name in names] return list(DEFAULT_TORCH_FEATURES) def _is_direct_horizon(entry: dict[str, Any]) -> bool: return bool(entry.get("direct_horizon")) or "target_horizon" in entry def _entry_horizon(entry: dict[str, Any], default: int) -> int: horizons = _entry_target_horizons(entry) requested = int(_clamp(_float_entry(entry, "target_horizon", float(max(1, default))), 1.0, 96.0)) if horizons: if requested in horizons: return requested return min(horizons, key=lambda value: abs(value - requested)) return requested def _entry_target_horizons(entry: dict[str, Any]) -> list[int]: raw = entry.get("target_horizons") if not isinstance(raw, list): return [] horizons = [] for value in raw: try: horizon = int(value) except (TypeError, ValueError): continue if 1 <= horizon <= 96 and horizon not in horizons: horizons.append(horizon) return horizons def _entry_output_layout(entry: dict[str, Any]) -> list[str]: raw = entry.get("output_layout") if isinstance(raw, list) and raw: return [str(value) for value in raw] return ["mean", "q10", "q50", "q90", "logit_up"] def _target_vector( entry: dict[str, Any], plural_key: str, scalar_key: str, size: int, default: float, ) -> list[float]: raw = entry.get(plural_key) if isinstance(raw, dict): values = [] for horizon in _entry_target_horizons(entry): values.append(float(raw.get(str(horizon), default))) if len(values) == size: return values if isinstance(raw, list) and len(raw) == size: return [float(value) for value in raw] scalar = _float_entry(entry, scalar_key, default) return [scalar for _ in range(size)] def _select_horizon_prediction(prediction: dict[str, Any], horizon: int) -> dict[str, Any] | None: horizons = prediction.get("horizons") if not isinstance(horizons, dict) or not horizons: return None key = str(horizon) selected = horizons.get(key) if isinstance(selected, dict): return selected numeric = [] for raw_key, value in horizons.items(): try: numeric.append((abs(int(raw_key) - horizon), value)) except (TypeError, ValueError): continue numeric.sort(key=lambda item: item[0]) return numeric[0][1] if numeric and isinstance(numeric[0][1], dict) else None def _public_horizon_forecasts(prediction: dict[str, Any]) -> dict[str, Any]: horizons = prediction.get("horizons") if not isinstance(horizons, dict): return {} public: dict[str, Any] = {} for key, row in horizons.items(): if not isinstance(row, dict): continue public[key] = { "expected_return_percent": round((math.exp(float(row.get("expected_return", 0.0))) - 1) * 100, 4), "probability_up": round(_clamp(float(row.get("probability_up", 0.5)), 0.0, 1.0), 4), "quantile_10_percent": round((math.exp(float(row.get("q10", 0.0))) - 1) * 100, 4), "quantile_50_percent": round((math.exp(float(row.get("q50", 0.0))) - 1) * 100, 4), "quantile_90_percent": round((math.exp(float(row.get("q90", 0.0))) - 1) * 100, 4), } return public def _float_entry(data: dict[str, Any], key: str, default: float) -> float: value = data.get(key) if isinstance(value, (int, float)): return float(value) if isinstance(value, str): try: return float(value) except ValueError: return default return default def _float_vector(data: Any) -> list[float]: if not isinstance(data, list): return [] return [float(value) for value in data] def _float_matrix(data: Any) -> list[list[float]]: if not isinstance(data, list): return [] return [_float_vector(row) for row in data] def _dot(left: list[float], right: list[float]) -> float: return sum(left[index] * right[index] for index in range(min(len(left), len(right)))) def _return_scale(returns: list[float]) -> float: recent = returns[-120:] if len(returns) > 120 else returns values = sorted(abs(value) for value in recent if math.isfinite(value)) if not values: return 0.0005 median = values[len(values) // 2] mean = sum(values) / len(values) return max(max(median, mean * 0.5), 1e-5) def _horizon_log_returns(closes: list[float], horizon: int) -> list[float]: horizon = max(1, horizon) values = [] for index in range(0, len(closes) - horizon): current = closes[index] future = closes[index + horizon] if current > 0 and future > 0: values.append(math.log(future / current)) return values def _horizon_return_scale(closes: list[float], horizon: int) -> float: values = _horizon_log_returns(closes, horizon) return _return_scale(values) if values else 0.0005 def _current_volatility_scale(candles: list[Candle], closes: list[float], horizon: int) -> float: horizon = max(1, horizon) latest = candles[-1] if candles else None close = closes[-1] if closes else (latest.close if latest else 0.0) atr_scale = 0.0 if latest and latest.atr_14 is not None and close > 0: atr_scale = (latest.atr_14 / close) * math.sqrt(horizon) realized = _horizon_return_scale(closes, horizon) one_step = _return_scale(_log_returns(closes)) * math.sqrt(horizon) if len(closes) > 2 else 0.0 return max(atr_scale * 0.7, realized, one_step, 0.0005) def _prediction_cap(closes: list[float], horizon: int, round_trip_cost: float) -> float: values = sorted(abs(value) for value in _horizon_log_returns(closes, horizon)[-96:]) base = values[int(len(values) * 0.9)] if values else 0.0 return max(base * 1.5 + round_trip_cost, 0.0005) def _sigmoid(value: float) -> float: if value >= 40: return 1.0 if value <= -40: return 0.0 return 1 / (1 + math.exp(-value)) def _confidence_adjustment( *, expected_return_percent: float, probability_up: float, skill: float, min_edge: float, max_adjustment: float, ) -> float: edge = abs(expected_return_percent) - min_edge if edge <= 0: return 0.0 direction = 1.0 if expected_return_percent > 0 and probability_up >= 0.55 else -1.0 if direction < 0 and probability_up > 0.45: return 0.0 strength = _clamp(edge / max(min_edge, 0.05), 0.0, 1.0) probability_strength = _clamp(abs(probability_up - 0.5) / 0.25, 0.0, 1.0) skill_strength = _clamp((skill + 0.03) / 0.18, 0.25, 1.0) return direction * _clamp(max_adjustment, 0.0, 0.18) * strength * probability_strength * skill_strength def _reason( *, model: str, expected_return_percent: float, probability_up: float, skill: float, block_entry: bool, ) -> str: if block_entry: return f"model {model}: expected move down {expected_return_percent:.3f}%, P(up)={probability_up:.2f}" return f"model {model}: forecast {expected_return_percent:.3f}%, P(up)={probability_up:.2f}, skill={skill:.3f}" def _normal_cdf(value: float) -> float: return 0.5 * (1 + math.erf(value / math.sqrt(2))) def _clamp(value: float, low: float, high: float) -> float: return max(low, min(high, value))