Модель${escapeHtml(modelName(forecast.model || '-'))}
+
Горизонт${num(forecast.horizon || 0, 0)}ч
P роста${num((forecast.probability_up || 0) * 100, 1)}%
Ожидание${signedNum(forecast.expected_return_percent, 3)}%
+
Q10/Q50/Q90${signedNum(forecast.quantile_10_percent, 2)} / ${signedNum(forecast.quantile_50_percent, 2)} / ${signedNum(forecast.quantile_90_percent, 2)}%
Волат.${num(forecast.volatility_percent, 3)}%
`;
}
diff --git a/crypto_spot_bot/time_series.py b/crypto_spot_bot/time_series.py
index 4b1a2f9..687944b 100644
--- a/crypto_spot_bot/time_series.py
+++ b/crypto_spot_bot/time_series.py
@@ -2,6 +2,7 @@ from __future__ import annotations
import json
import math
+from bisect import bisect_right
from dataclasses import asdict, dataclass, field
from typing import Any
@@ -13,17 +14,46 @@ 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",
@@ -56,6 +86,13 @@ class TimeSeriesForecast:
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
+ horizon_forecasts: dict[str, Any] = field(default_factory=dict)
candidates: list[dict[str, Any]] = field(default_factory=list)
def as_dict(self) -> dict[str, Any]:
@@ -68,7 +105,14 @@ class TimeSeriesForecaster:
self._lstm_artifact_mtime: float | None = None
self._lstm_artifact: dict[str, Any] = {}
- def forecast(self, candles: list[Candle], symbol: str | None = None) -> TimeSeriesForecast:
+ 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]
@@ -82,7 +126,17 @@ class TimeSeriesForecaster:
artifact = self._load_lstm_artifact()
entry = _torch_recurrent_entry(symbol, artifact)
model = _torch_recurrent_model_name(symbol, artifact)
- feature_rows = _feature_matrix(candles, _feature_names(entry)) if entry else []
+ feature_rows = (
+ _feature_matrix(
+ candles,
+ _feature_names(entry),
+ symbol=symbol,
+ market_candles=market_candles,
+ trend_candles=trend_candles,
+ )
+ if entry
+ else []
+ )
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")
@@ -92,10 +146,79 @@ class TimeSeriesForecaster:
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")),
+ horizon_forecasts=_public_horizon_forecasts(prediction),
+ candidates=[{"model": model, "mae_percent": round(model_mae * 100, 4)}],
+ )
+
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
@@ -145,6 +268,13 @@ class TimeSeriesForecaster:
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")),
+ horizon_forecasts={},
candidates=[{"model": model, "mae_percent": round(model_mae * 100, 4)}],
)
@@ -186,6 +316,13 @@ def _empty_forecast(enabled: bool, reason: str) -> TimeSeriesForecast:
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",
+ horizon_forecasts={},
)
@@ -193,15 +330,58 @@ 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) -> list[list[float]]:
+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) for name in names])
+ rows.append([_feature_value(name, candles, index, candle, context) for name in names])
return rows
-def _feature_value(name: str, candles: list[Candle], index: int, candle: Candle) -> float:
+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":
@@ -210,6 +390,10 @@ def _feature_value(name: str, candles: list[Candle], index: int, candle: Candle)
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":
@@ -222,16 +406,52 @@ def _feature_value(name: str, candles: list[Candle], index: int, candle: Candle)
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
@@ -266,6 +486,143 @@ def _pattern_feature_value(name: str, candles: list[Candle], index: int) -> floa
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 {
@@ -486,7 +843,8 @@ def _torch_recurrent_predict(
*,
feature_rows: list[list[float]] | None = None,
closes: list[float] | None = None,
-) -> float | 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:
@@ -523,11 +881,19 @@ def _torch_recurrent_predict(
)
if hidden is None:
return None
- head_weight = _float_vector(entry.get("head_weight"))
- head_bias = _float_entry(entry, "head_bias", 0.0)
- if len(head_weight) != hidden_size:
+ head_outputs = _torch_head_outputs(hidden, entry, hidden_size)
+ if not head_outputs:
return None
- normalized_prediction = sum(weight * value for weight, value in zip(head_weight, hidden)) + head_bias
+ 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
@@ -543,6 +909,111 @@ def _torch_recurrent_predict(
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"))
@@ -575,6 +1046,7 @@ def _torch_recurrent_hidden(
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):
@@ -587,7 +1059,32 @@ def _torch_recurrent_hidden(
else:
return None
layer_input = h_layers[layer]
- return h_layers[-1]
+ 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(
@@ -705,7 +1202,91 @@ def _is_direct_horizon(entry: dict[str, Any]) -> bool:
def _entry_horizon(entry: dict[str, Any], default: int) -> int:
- return int(_clamp(_float_entry(entry, "target_horizon", float(max(1, default))), 1.0, 96.0))
+ 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:
@@ -762,6 +1343,24 @@ def _horizon_return_scale(closes: list[float], horizon: int) -> float:
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
diff --git a/tests/test_time_series.py b/tests/test_time_series.py
index 9ad6a32..876afce 100644
--- a/tests/test_time_series.py
+++ b/tests/test_time_series.py
@@ -124,6 +124,73 @@ def _write_multifeature_torch_gru_artifact(path, *, head_bias: float) -> None:
)
+def _write_probabilistic_torch_gru_artifact(path) -> None:
+ hidden_size = 2
+ input_size = 2
+ output_size = 10
+ path.write_text(
+ json.dumps(
+ {
+ "version": 4,
+ "type": "pytorch_recurrent_forecaster",
+ "target_horizon": 3,
+ "target_horizons": [1, 3],
+ "direct_horizon": True,
+ "target_transform": "net_return_over_volatility",
+ "round_trip_cost": 0.0026,
+ "output_layout": ["mean", "q10", "q50", "q90", "logit_up"],
+ "feature_count": input_size,
+ "feature_names": ["return_1", "range_percent"],
+ "symbols": {
+ "BTCUSDT": {
+ "model": "torch_gru",
+ "architecture": "gru",
+ "lookback": 8,
+ "target_horizon": 3,
+ "target_horizons": [1, 3],
+ "direct_horizon": True,
+ "target_transform": "net_return_over_volatility",
+ "round_trip_cost": 0.0026,
+ "output_layout": ["mean", "q10", "q50", "q90", "logit_up"],
+ "input_size": input_size,
+ "output_size": output_size,
+ "feature_names": ["return_1", "range_percent"],
+ "feature_means": [0.0, 0.0],
+ "feature_scales": [0.001, 0.001],
+ "target_means": [0.0, 0.0],
+ "target_scales": [1.0, 1.0],
+ "target_mean": 0.0,
+ "target_scale": 1.0,
+ "hidden_size": hidden_size,
+ "num_layers": 1,
+ "clip": 8.0,
+ "validation_mae_percent": 0.01,
+ "baseline_mae_percent": 0.08,
+ "validation_mae_by_horizon": {"1": 0.001, "3": 0.0015},
+ "baseline_mae_by_horizon": {"1": 0.002, "3": 0.003},
+ "skill": 0.2,
+ "attention_pooling": True,
+ "attention_weight": [0.0, 0.0],
+ "attention_bias": 0.0,
+ "context_norm": True,
+ "context_norm_weight": [1.0, 1.0],
+ "context_norm_bias": [0.0, 0.0],
+ "state_dict": {
+ "weight_ih_l0": [[0.0, 0.0] for _ in range(3 * hidden_size)],
+ "weight_hh_l0": [[0.0, 0.0] for _ in range(3 * hidden_size)],
+ "bias_ih_l0": [0.0 for _ in range(3 * hidden_size)],
+ "bias_hh_l0": [0.0 for _ in range(3 * hidden_size)],
+ },
+ "head_weight": [[0.0, 0.0] for _ in range(output_size)],
+ "head_bias": [0.2, 0.05, 0.15, 0.35, 1.0, 0.35, 0.10, 0.30, 0.55, 2.0],
+ },
+ },
+ }
+ ),
+ encoding="utf-8",
+ )
+
+
def test_time_series_forecaster_requires_torch_artifact(make_settings, tmp_path) -> None:
settings = make_settings(
tmp_path,
@@ -233,3 +300,25 @@ def test_time_series_forecaster_reads_multifeature_direct_horizon_artifact(make_
assert forecast.horizon == 3
assert 0.015 <= forecast.expected_return_percent <= 0.025
assert forecast.volatility_model == "direct horizon validation MAE"
+
+
+def test_time_series_forecaster_reads_probabilistic_multi_horizon_artifact(make_settings, tmp_path) -> None:
+ artifact_path = tmp_path / "lstm_forecaster.json"
+ _write_probabilistic_torch_gru_artifact(artifact_path)
+ settings = make_settings(
+ tmp_path,
+ time_series_min_candles=80,
+ time_series_forecast_horizon=3,
+ time_series_lstm_model_path=artifact_path,
+ )
+ returns = [0.0002 if index % 5 else -0.00007 for index in range(160)]
+
+ forecast = TimeSeriesForecaster(settings).forecast(_candles_from_returns(returns), symbol="BTCUSDT")
+
+ assert forecast.usable is True
+ assert forecast.model == "torch_gru"
+ assert forecast.horizon == 3
+ assert forecast.target_transform == "net_return_over_volatility"
+ assert forecast.probability_up > 0.85
+ assert forecast.quantile_10_percent <= forecast.quantile_50_percent <= forecast.quantile_90_percent
+ assert sorted(forecast.horizon_forecasts) == ["1", "3"]
diff --git a/tools/install_windows_torch_retrainer.ps1 b/tools/install_windows_torch_retrainer.ps1
index fa59dae..67ccb58 100644
--- a/tools/install_windows_torch_retrainer.ps1
+++ b/tools/install_windows_torch_retrainer.ps1
@@ -3,9 +3,11 @@ param(
[string]$TaskName = "TradeBot PyTorch Forecaster Retrainer",
[int]$EveryHours = 6,
[string]$Symbols = "BTCUSDT,ETHUSDT,SOLUSDT,LTCUSDT",
- [int]$Limit = 1000,
+ [int]$Limit = 3000,
[int]$Horizon = 0,
+ [string]$Horizons = "",
[string]$Features = "",
+ [string]$ContextSymbols = "",
[int]$FirstRunMinutes = 0
)
@@ -35,9 +37,15 @@ if ($Limit -gt 0) {
if ($Horizon -gt 0) {
$actionArgs += " -Horizon $Horizon"
}
+if ($Horizons) {
+ $actionArgs += " -Horizons `"$Horizons`""
+}
if ($Features) {
$actionArgs += " -Features `"$Features`""
}
+if ($ContextSymbols) {
+ $actionArgs += " -ContextSymbols `"$ContextSymbols`""
+}
$action = New-ScheduledTaskAction -Execute "powershell.exe" -Argument $actionArgs -WorkingDirectory $RepoRoot
$trigger = New-ScheduledTaskTrigger `
-Once `
diff --git a/tools/run_torch_retrain.ps1 b/tools/run_torch_retrain.ps1
index 64dbf4d..5b295ac 100644
--- a/tools/run_torch_retrain.ps1
+++ b/tools/run_torch_retrain.ps1
@@ -8,7 +8,9 @@ param(
[string]$Layers = "",
[string]$Dropouts = "",
[int]$Horizon = 0,
+ [string]$Horizons = "",
[string]$Features = "",
+ [string]$ContextSymbols = "",
[int]$Epochs = 0,
[int]$Patience = 0,
[string]$Interval = "",
@@ -51,17 +53,19 @@ function Resolve-Python {
if (-not $Symbols -and $env:TORCH_RETRAIN_SYMBOLS) { $Symbols = $env:TORCH_RETRAIN_SYMBOLS }
if ($Limit -le 0) {
- $Limit = if ($env:TORCH_RETRAIN_LIMIT) { [int]$env:TORCH_RETRAIN_LIMIT } else { 1000 }
+ $Limit = if ($env:TORCH_RETRAIN_LIMIT) { [int]$env:TORCH_RETRAIN_LIMIT } else { 3000 }
}
-if (-not $Lookbacks) { $Lookbacks = if ($env:TORCH_RETRAIN_LOOKBACKS) { $env:TORCH_RETRAIN_LOOKBACKS } else { "32,64" } }
+if (-not $Lookbacks) { $Lookbacks = if ($env:TORCH_RETRAIN_LOOKBACKS) { $env:TORCH_RETRAIN_LOOKBACKS } else { "64" } }
if (-not $Architectures) { $Architectures = if ($env:TORCH_RETRAIN_ARCHITECTURES) { $env:TORCH_RETRAIN_ARCHITECTURES } else { "lstm,gru" } }
-if (-not $HiddenSizes) { $HiddenSizes = if ($env:TORCH_RETRAIN_HIDDEN_SIZES) { $env:TORCH_RETRAIN_HIDDEN_SIZES } else { "32,64" } }
+if (-not $HiddenSizes) { $HiddenSizes = if ($env:TORCH_RETRAIN_HIDDEN_SIZES) { $env:TORCH_RETRAIN_HIDDEN_SIZES } else { "64,96" } }
if (-not $Layers) { $Layers = if ($env:TORCH_RETRAIN_LAYERS) { $env:TORCH_RETRAIN_LAYERS } else { "2" } }
if (-not $Dropouts) { $Dropouts = if ($env:TORCH_RETRAIN_DROPOUTS) { $env:TORCH_RETRAIN_DROPOUTS } else { "0.15" } }
if ($Horizon -le 0 -and $env:TORCH_RETRAIN_HORIZON) { $Horizon = [int]$env:TORCH_RETRAIN_HORIZON }
+if (-not $Horizons -and $env:TORCH_RETRAIN_HORIZONS) { $Horizons = $env:TORCH_RETRAIN_HORIZONS }
if (-not $Features -and $env:TORCH_RETRAIN_FEATURES) { $Features = $env:TORCH_RETRAIN_FEATURES }
-if ($Epochs -le 0) { $Epochs = if ($env:TORCH_RETRAIN_EPOCHS) { [int]$env:TORCH_RETRAIN_EPOCHS } else { 60 } }
-if ($Patience -le 0) { $Patience = if ($env:TORCH_RETRAIN_PATIENCE) { [int]$env:TORCH_RETRAIN_PATIENCE } else { 10 } }
+if (-not $ContextSymbols -and $env:TORCH_RETRAIN_CONTEXT_SYMBOLS) { $ContextSymbols = $env:TORCH_RETRAIN_CONTEXT_SYMBOLS }
+if ($Epochs -le 0) { $Epochs = if ($env:TORCH_RETRAIN_EPOCHS) { [int]$env:TORCH_RETRAIN_EPOCHS } else { 70 } }
+if ($Patience -le 0) { $Patience = if ($env:TORCH_RETRAIN_PATIENCE) { [int]$env:TORCH_RETRAIN_PATIENCE } else { 8 } }
if (-not $Interval -and $env:TORCH_RETRAIN_INTERVAL) { $Interval = $env:TORCH_RETRAIN_INTERVAL }
if (-not $EnvFile -and $env:TORCH_RETRAIN_ENV) { $EnvFile = $env:TORCH_RETRAIN_ENV }
if (-not $EnvFile -and (Test-Path (Join-Path $RepoRoot ".env"))) { $EnvFile = Join-Path $RepoRoot ".env" }
@@ -94,7 +98,9 @@ try {
if ($Interval) { $trainerArgs += @("--interval", $Interval) }
if ($EnvFile) { $trainerArgs += @("--env", $EnvFile) }
if ($Horizon -gt 0) { $trainerArgs += @("--horizon", $Horizon.ToString()) }
+ if ($Horizons) { $trainerArgs += @("--horizons", $Horizons) }
if ($Features) { $trainerArgs += @("--features", $Features) }
+ if ($ContextSymbols) { $trainerArgs += @("--context-symbols", $ContextSymbols) }
Push-Location $RepoRoot
$pushedLocation = $true
diff --git a/tools/train_torch_recurrent_forecaster.py b/tools/train_torch_recurrent_forecaster.py
index cd73338..346e0fa 100644
--- a/tools/train_torch_recurrent_forecaster.py
+++ b/tools/train_torch_recurrent_forecaster.py
@@ -4,6 +4,7 @@ import argparse
import json
import math
import sys
+import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
@@ -30,22 +31,40 @@ from crypto_spot_bot.models import Candle
from crypto_spot_bot.time_series import DEFAULT_TORCH_FEATURES, _feature_matrix, _log_returns
+OUTPUT_LAYOUT = ("mean", "q10", "q50", "q90", "logit_up")
+QUANTILES = {"q10": 0.10, "q50": 0.50, "q90": 0.90}
+
+
@dataclass(slots=True)
class PreparedData:
train_x: torch.Tensor
train_y: torch.Tensor
+ train_up: torch.Tensor
validation_x: torch.Tensor
validation_y: torch.Tensor
- validation_targets: list[float]
+ validation_up: torch.Tensor
+ validation_targets: list[list[float]]
+ validation_volatility_scales: list[list[float]]
feature_names: list[str]
feature_means: list[float]
feature_scales: list[float]
- target_mean: float
- target_scale: float
+ target_means: list[float]
+ target_scales: list[float]
+ target_horizons: list[int]
+ decision_horizon: int
+ decision_horizon_index: int
train_samples: int
validation_samples: int
+@dataclass(slots=True)
+class TrainingSample:
+ window: list[list[float]]
+ normalized_targets: list[float]
+ raw_targets: list[float]
+ volatility_scales: list[float]
+
+
class RecurrentReturnModel(nn.Module):
def __init__(
self,
@@ -55,6 +74,9 @@ class RecurrentReturnModel(nn.Module):
hidden_size: int,
num_layers: int,
dropout: float,
+ output_size: int,
+ attention_pooling: bool,
+ context_norm: bool,
) -> None:
super().__init__()
recurrent_cls = nn.LSTM if architecture == "lstm" else nn.GRU
@@ -65,11 +87,19 @@ class RecurrentReturnModel(nn.Module):
dropout=dropout if num_layers > 1 else 0.0,
batch_first=True,
)
- self.head = nn.Linear(hidden_size, 1)
+ self.attention = nn.Linear(hidden_size, 1) if attention_pooling else None
+ self.context_norm = nn.LayerNorm(hidden_size) if context_norm else nn.Identity()
+ self.head = nn.Linear(hidden_size, output_size)
def forward(self, values: torch.Tensor) -> torch.Tensor:
output, _state = self.rnn(values)
- return self.head(output[:, -1, :]).squeeze(-1)
+ if self.attention is not None:
+ scores = self.attention(output).squeeze(-1)
+ weights = torch.softmax(scores, dim=1).unsqueeze(-1)
+ context = (output * weights).sum(dim=1)
+ else:
+ context = output[:, -1, :]
+ return self.head(self.context_norm(context))
def main() -> None:
@@ -84,19 +114,27 @@ def main() -> None:
interval = args.interval or settings.base_interval
output = Path(args.output) if args.output else settings.time_series_lstm_model_path
device = _device(args.device)
- horizon = args.horizon if args.horizon > 0 else max(1, settings.time_series_forecast_horizon)
+ decision_horizon = args.horizon if args.horizon > 0 else max(1, settings.time_series_forecast_horizon)
+ target_horizons = _horizons(args.horizons, decision_horizon)
feature_names = _feature_names_arg(args.features)
+ round_trip_cost = max(0.0, 2.0 * (float(settings.taker_fee_rate) + float(settings.slippage_rate)))
artifact: dict[str, Any] = {
- "version": 3,
+ "version": 4,
"type": "pytorch_recurrent_forecaster",
"created_at": datetime.now(timezone.utc).isoformat(),
"trainer": Path(__file__).name,
"interval": interval,
"limit": args.limit,
"validation_window": args.validation_window,
- "target_horizon": horizon,
+ "target_horizon": decision_horizon,
+ "target_horizons": target_horizons,
"direct_horizon": True,
+ "target_transform": "net_return_over_volatility",
+ "target_return": "round_trip_after_cost_log_return",
+ "round_trip_cost": round(round_trip_cost, 10),
+ "output_layout": list(OUTPUT_LAYOUT),
+ "quantiles": list(QUANTILES.values()),
"feature_names": feature_names,
"feature_count": len(feature_names),
"device": str(device),
@@ -110,8 +148,11 @@ def main() -> None:
interval=interval,
limit=args.limit,
validation_window=args.validation_window,
- target_horizon=horizon,
+ target_horizons=target_horizons,
+ decision_horizon=decision_horizon,
feature_names=feature_names,
+ round_trip_cost=round_trip_cost,
+ context_symbols=_strings(args.context_symbols),
architectures=_strings(args.architectures),
lookbacks=_ints(args.lookbacks),
hidden_sizes=_ints(args.hidden_sizes),
@@ -123,6 +164,8 @@ def main() -> None:
learning_rate=args.learning_rate,
weight_decay=args.weight_decay,
clip=args.clip,
+ attention_pooling=args.attention_pooling,
+ context_norm=args.context_norm,
device=device,
seed=args.seed,
)
@@ -133,10 +176,11 @@ def main() -> None:
print(
f"{symbol}: model={result['model']} lookback={result['lookback']} "
f"features={result['input_size']} hidden={result['hidden_size']} "
- f"layers={result['num_layers']} horizon={result['target_horizon']} "
+ f"layers={result['num_layers']} horizons={','.join(map(str, result['target_horizons']))} "
f"mae={result['validation_mae_percent']:.5f}% "
f"baseline={result['baseline_mae_percent']:.5f}% "
- f"skill={result['skill']:.4f} dir={result['directional_accuracy']:.3f}"
+ f"skill={result['skill']:.4f} dir={result['directional_accuracy']:.3f} "
+ f"p_brier={result['probability_brier']:.4f}"
)
output.parent.mkdir(parents=True, exist_ok=True)
@@ -154,7 +198,9 @@ def _parse_args() -> argparse.Namespace:
parser.add_argument("--limit", type=int, default=1000, help="Kline limit per symbol.")
parser.add_argument("--validation-window", type=int, default=120, help="Held-out tail targets used for validation.")
parser.add_argument("--horizon", type=int, default=0, help="Direct forecast horizon in candles. Defaults to TIME_SERIES_FORECAST_HORIZON.")
+ parser.add_argument("--horizons", default="1,3,6,12", help="Comma-separated direct forecast horizons.")
parser.add_argument("--features", default=",".join(DEFAULT_TORCH_FEATURES), help="Comma-separated feature names.")
+ parser.add_argument("--context-symbols", default="BTCUSDT,ETHUSDT", help="Cross-asset context symbols.")
parser.add_argument("--architectures", default="lstm,gru", help="Comma-separated recurrent types: lstm,gru.")
parser.add_argument("--lookbacks", default="32,64", help="Comma-separated sequence lengths.")
parser.add_argument("--hidden-sizes", default="32,64", help="Comma-separated hidden sizes.")
@@ -166,6 +212,8 @@ def _parse_args() -> argparse.Namespace:
parser.add_argument("--learning-rate", type=float, default=0.001, help="AdamW learning rate.")
parser.add_argument("--weight-decay", type=float, default=0.0001, help="AdamW weight decay.")
parser.add_argument("--clip", type=float, default=8.0, help="Clamp normalized features, targets and predictions.")
+ parser.add_argument("--attention-pooling", action=argparse.BooleanOptionalAction, default=True, help="Use exportable attention pooling over recurrent states.")
+ parser.add_argument("--context-norm", action=argparse.BooleanOptionalAction, default=True, help="Use exportable LayerNorm before the forecast head.")
parser.add_argument("--seed", type=int, default=7, help="Random seed.")
parser.add_argument("--threads", type=int, default=0, help="Torch CPU threads; 0 keeps torch default.")
parser.add_argument("--device", default="auto", help="auto, cpu, cuda, or mps.")
@@ -188,8 +236,11 @@ def _train_symbol(
interval: str,
limit: int,
validation_window: int,
- target_horizon: int,
+ target_horizons: list[int],
+ decision_horizon: int,
feature_names: list[str],
+ round_trip_cost: float,
+ context_symbols: list[str],
architectures: list[str],
lookbacks: list[int],
hidden_sizes: list[int],
@@ -201,15 +252,31 @@ def _train_symbol(
learning_rate: float,
weight_decay: float,
clip: float,
+ attention_pooling: bool,
+ context_norm: bool,
device: torch.device,
seed: int,
) -> dict[str, Any] | None:
- candles = client.klines(symbol, interval, limit)
+ candles = _historical_klines(client, symbol, interval, limit)
add_indicators(candles)
closes = [float(candle.close) for candle in candles if candle.close > 0]
returns = _log_returns(closes)
- if len(candles) < max(140, validation_window + max(lookbacks) + target_horizon + 16):
+ max_horizon = max(target_horizons)
+ if len(candles) < max(180, validation_window + max(lookbacks) + max_horizon + 16):
return None
+ market_candles: dict[str, list[Candle]] = {symbol.upper(): candles}
+ for context_symbol in context_symbols:
+ context_symbol = context_symbol.upper()
+ if context_symbol in market_candles:
+ continue
+ try:
+ rows = _historical_klines(client, context_symbol, interval, limit)
+ add_indicators(rows)
+ market_candles[context_symbol] = rows
+ except Exception as exc:
+ print(f"{symbol}: context {context_symbol} skipped: {exc}")
+ trend_candles = _historical_klines(client, symbol, "D", min(max(260, limit // 24 + 260), 1000))
+ add_indicators(trend_candles)
best: dict[str, Any] | None = None
for lookback in lookbacks:
@@ -217,14 +284,21 @@ def _train_symbol(
candles=candles,
feature_names=feature_names,
lookback=lookback,
- target_horizon=target_horizon,
+ target_horizons=target_horizons,
+ decision_horizon=decision_horizon,
+ round_trip_cost=round_trip_cost,
+ market_candles=market_candles,
+ trend_candles=trend_candles,
validation_window=validation_window,
clip=clip,
device=device,
)
if prepared is None:
continue
- baseline_mae = sum(abs(value) for value in prepared.validation_targets) / len(prepared.validation_targets)
+ baseline_mae = (
+ sum(abs(value[prepared.decision_horizon_index]) for value in prepared.validation_targets)
+ / len(prepared.validation_targets)
+ )
for architecture in architectures:
if architecture not in {"lstm", "gru"}:
continue
@@ -237,6 +311,7 @@ def _train_symbol(
prepared=prepared,
architecture=architecture,
input_size=len(feature_names),
+ output_size=len(target_horizons) * len(OUTPUT_LAYOUT),
hidden_size=hidden_size,
num_layers=num_layers,
dropout=dropout,
@@ -246,6 +321,8 @@ def _train_symbol(
learning_rate=learning_rate,
weight_decay=weight_decay,
clip=clip,
+ attention_pooling=attention_pooling,
+ context_norm=context_norm,
device=device,
seed=seed,
)
@@ -256,19 +333,30 @@ def _train_symbol(
"model": f"torch_{architecture}",
"architecture": architecture,
"lookback": lookback,
- "target_horizon": target_horizon,
+ "target_horizon": prepared.decision_horizon,
+ "target_horizons": prepared.target_horizons,
"direct_horizon": True,
+ "target_transform": "net_return_over_volatility",
+ "target_return": "round_trip_after_cost_log_return",
+ "round_trip_cost": round(round_trip_cost, 10),
+ "output_layout": list(OUTPUT_LAYOUT),
+ "quantiles": list(QUANTILES.values()),
"input_size": len(feature_names),
+ "output_size": len(target_horizons) * len(OUTPUT_LAYOUT),
"feature_names": feature_names,
"feature_means": prepared.feature_means,
"feature_scales": prepared.feature_scales,
- "target_mean": prepared.target_mean,
- "target_scale": prepared.target_scale,
- "mean": prepared.target_mean,
- "scale": prepared.target_scale,
+ "target_means": prepared.target_means,
+ "target_scales": prepared.target_scales,
+ "target_mean": prepared.target_means[prepared.decision_horizon_index],
+ "target_scale": prepared.target_scales[prepared.decision_horizon_index],
+ "mean": prepared.target_means[prepared.decision_horizon_index],
+ "scale": prepared.target_scales[prepared.decision_horizon_index],
"hidden_size": hidden_size,
"num_layers": num_layers,
"dropout": dropout if num_layers > 1 else 0.0,
+ "attention_pooling": attention_pooling,
+ "context_norm": context_norm,
"clip": clip,
"validation_mae_percent": validation_mae * 100,
"baseline_mae_percent": baseline_mae * 100,
@@ -292,23 +380,47 @@ def _prepare_data(
candles: list[Candle],
feature_names: list[str],
lookback: int,
- target_horizon: int,
+ target_horizons: list[int],
+ decision_horizon: int,
+ round_trip_cost: float,
+ market_candles: dict[str, list[Candle]],
+ trend_candles: list[Candle],
validation_window: int,
clip: float,
device: torch.device,
) -> PreparedData | None:
closes = [float(candle.close) for candle in candles]
- feature_rows = _feature_matrix(candles, feature_names)
- samples: list[tuple[list[list[float]], float]] = []
- for end_index in range(lookback - 1, len(candles) - target_horizon):
+ feature_rows = _feature_matrix(
+ candles,
+ feature_names,
+ market_candles=market_candles,
+ trend_candles=trend_candles,
+ )
+ max_horizon = max(target_horizons)
+ samples: list[TrainingSample] = []
+ for end_index in range(lookback - 1, len(candles) - max_horizon):
current = closes[end_index]
- future = closes[end_index + target_horizon]
- if current <= 0 or future <= 0:
+ if current <= 0:
continue
window = feature_rows[end_index - lookback + 1 : end_index + 1]
if len(window) != lookback:
continue
- samples.append((window, math.log(future / current)))
+ raw_targets: list[float] = []
+ volatility_scales: list[float] = []
+ normalized_targets: list[float] = []
+ valid = True
+ for horizon in target_horizons:
+ future = closes[end_index + horizon]
+ if future <= 0:
+ valid = False
+ break
+ net_return = math.log(future / current) - round_trip_cost
+ volatility_scale = _target_volatility_scale(candles, closes, end_index, horizon)
+ raw_targets.append(net_return)
+ volatility_scales.append(volatility_scale)
+ normalized_targets.append(net_return / max(volatility_scale, 1e-8))
+ if valid:
+ samples.append(TrainingSample(window, normalized_targets, raw_targets, volatility_scales))
if len(samples) < 48:
return None
@@ -319,45 +431,55 @@ def _prepare_data(
return None
feature_means, feature_scales = _feature_stats(train_samples, len(feature_names))
- train_targets = [target for _, target in train_samples]
- target_mean = sum(train_targets) / len(train_targets)
- target_scale = _return_scale(train_targets)
+ target_means, target_scales = _target_stats(train_samples, len(target_horizons))
+ decision_horizon = decision_horizon if decision_horizon in target_horizons else min(
+ target_horizons,
+ key=lambda value: abs(value - decision_horizon),
+ )
+ decision_horizon_index = target_horizons.index(decision_horizon)
- train_x, train_y = _normalize_samples(
+ train_x, train_y, train_up = _normalize_samples(
train_samples,
feature_means=feature_means,
feature_scales=feature_scales,
- target_mean=target_mean,
- target_scale=target_scale,
+ target_means=target_means,
+ target_scales=target_scales,
clip=clip,
)
- validation_x, validation_y = _normalize_samples(
+ validation_x, validation_y, validation_up = _normalize_samples(
validation_samples,
feature_means=feature_means,
feature_scales=feature_scales,
- target_mean=target_mean,
- target_scale=target_scale,
+ target_means=target_means,
+ target_scales=target_scales,
clip=clip,
)
return PreparedData(
train_x=torch.tensor(train_x, dtype=torch.float32, device=device),
train_y=torch.tensor(train_y, dtype=torch.float32, device=device),
+ train_up=torch.tensor(train_up, dtype=torch.float32, device=device),
validation_x=torch.tensor(validation_x, dtype=torch.float32, device=device),
validation_y=torch.tensor(validation_y, dtype=torch.float32, device=device),
- validation_targets=[target for _, target in validation_samples],
+ validation_up=torch.tensor(validation_up, dtype=torch.float32, device=device),
+ validation_targets=[sample.raw_targets for sample in validation_samples],
+ validation_volatility_scales=[sample.volatility_scales for sample in validation_samples],
feature_names=feature_names,
feature_means=feature_means,
feature_scales=feature_scales,
- target_mean=target_mean,
- target_scale=target_scale,
+ target_means=target_means,
+ target_scales=target_scales,
+ target_horizons=target_horizons,
+ decision_horizon=decision_horizon,
+ decision_horizon_index=decision_horizon_index,
train_samples=len(train_x),
validation_samples=len(validation_x),
)
-def _feature_stats(samples: list[tuple[list[list[float]], float]], input_size: int) -> tuple[list[float], list[float]]:
+def _feature_stats(samples: list[TrainingSample], input_size: int) -> tuple[list[float], list[float]]:
columns = [[] for _ in range(input_size)]
- for window, _target in samples:
+ for sample in samples:
+ window = sample.window
for row in window:
for index in range(input_size):
columns[index].append(float(row[index] if index < len(row) else 0.0))
@@ -377,19 +499,32 @@ def _feature_stats(samples: list[tuple[list[list[float]], float]], input_size: i
return means, scales
+def _target_stats(samples: list[TrainingSample], output_size: int) -> tuple[list[float], list[float]]:
+ means: list[float] = []
+ scales: list[float] = []
+ for index in range(output_size):
+ values = [sample.normalized_targets[index] for sample in samples]
+ mean = sum(values) / len(values) if values else 0.0
+ means.append(mean)
+ scales.append(_return_scale([value - mean for value in values]))
+ return means, scales
+
+
def _normalize_samples(
- samples: list[tuple[list[list[float]], float]],
+ samples: list[TrainingSample],
*,
feature_means: list[float],
feature_scales: list[float],
- target_mean: float,
- target_scale: float,
+ target_means: list[float],
+ target_scales: list[float],
clip: float,
-) -> tuple[list[list[list[float]]], list[float]]:
+) -> tuple[list[list[list[float]]], list[list[float]], list[list[float]]]:
input_size = len(feature_means)
x_values: list[list[list[float]]] = []
- y_values: list[float] = []
- for window, target in samples:
+ y_values: list[list[float]] = []
+ up_values: list[list[float]] = []
+ for sample in samples:
+ window = sample.window
x_values.append(
[
[
@@ -404,8 +539,18 @@ def _normalize_samples(
for row in window
]
)
- y_values.append(_clamp((target - target_mean) / max(target_scale, 1e-8), -clip, clip))
- return x_values, y_values
+ y_values.append(
+ [
+ _clamp(
+ (target - target_means[index]) / max(target_scales[index], 1e-8),
+ -clip,
+ clip,
+ )
+ for index, target in enumerate(sample.normalized_targets)
+ ]
+ )
+ up_values.append([1.0 if target > 0 else 0.0 for target in sample.raw_targets])
+ return x_values, y_values, up_values
def _fit_candidate(
@@ -413,6 +558,7 @@ def _fit_candidate(
prepared: PreparedData,
architecture: str,
input_size: int,
+ output_size: int,
hidden_size: int,
num_layers: int,
dropout: float,
@@ -422,6 +568,8 @@ def _fit_candidate(
learning_rate: float,
weight_decay: float,
clip: float,
+ attention_pooling: bool,
+ context_norm: bool,
device: torch.device,
seed: int,
) -> dict[str, Any]:
@@ -432,12 +580,14 @@ def _fit_candidate(
hidden_size=hidden_size,
num_layers=num_layers,
dropout=dropout,
+ output_size=output_size,
+ attention_pooling=attention_pooling,
+ context_norm=context_norm,
).to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate, weight_decay=weight_decay)
- criterion = nn.SmoothL1Loss(beta=0.5)
generator = torch.Generator(device="cpu").manual_seed(seed)
loader = DataLoader(
- TensorDataset(prepared.train_x, prepared.train_y),
+ TensorDataset(prepared.train_x, prepared.train_y, prepared.train_up),
batch_size=max(1, batch_size),
shuffle=True,
generator=generator,
@@ -449,9 +599,9 @@ def _fit_candidate(
stale_epochs = 0
for epoch in range(1, max(1, epochs) + 1):
model.train()
- for batch_x, batch_y in loader:
+ for batch_x, batch_y, batch_up in loader:
optimizer.zero_grad(set_to_none=True)
- loss = criterion(model(batch_x), batch_y)
+ loss = _forecast_loss(model(batch_x), batch_y, batch_up, len(prepared.target_horizons))
loss.backward()
nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
@@ -474,40 +624,79 @@ def _fit_candidate(
"best_epoch": best_epoch,
"epochs_trained": best_epoch + stale_epochs,
"state_dict": _export_recurrent_state(model),
- "head_weight": _round_list(model.head.weight.detach().cpu().squeeze(0).tolist()),
- "head_bias": round(float(model.head.bias.detach().cpu().item()), 10),
+ "head_weight": _round_nested(model.head.weight.detach().cpu().tolist()),
+ "head_bias": _round_list(model.head.bias.detach().cpu().tolist()),
+ **_export_context_state(model),
}
def _validation_metrics(model: nn.Module, prepared: PreparedData, clip: float) -> dict[str, float]:
model.eval()
with torch.no_grad():
- normalized_predictions = model(prepared.validation_x).detach().cpu().tolist()
- predictions = [
- _clamp(float(prediction), -clip, clip) * prepared.target_scale + prepared.target_mean
- for prediction in normalized_predictions
- ]
- errors = [abs(prediction - actual) for prediction, actual in zip(predictions, prepared.validation_targets)]
+ raw_outputs = model(prepared.validation_x).detach().cpu()
+ outputs = raw_outputs.view(len(prepared.validation_targets), len(prepared.target_horizons), len(OUTPUT_LAYOUT))
+ mean_predictions = outputs[:, :, 0].tolist()
+ logit_predictions = outputs[:, :, 4].tolist()
+ predictions: list[list[float]] = []
+ probabilities: list[list[float]] = []
+ for row_index, row in enumerate(mean_predictions):
+ predicted_row: list[float] = []
+ probability_row: list[float] = []
+ for horizon_index, normalized_prediction in enumerate(row):
+ transformed = (
+ _clamp(float(normalized_prediction), -clip, clip)
+ * prepared.target_scales[horizon_index]
+ + prepared.target_means[horizon_index]
+ )
+ predicted_row.append(transformed * prepared.validation_volatility_scales[row_index][horizon_index])
+ probability_row.append(_sigmoid(float(logit_predictions[row_index][horizon_index])))
+ predictions.append(predicted_row)
+ probabilities.append(probability_row)
+ decision = prepared.decision_horizon_index
+ decision_predictions = [row[decision] for row in predictions]
+ decision_targets = [row[decision] for row in prepared.validation_targets]
+ errors = [abs(prediction - actual) for prediction, actual in zip(decision_predictions, decision_targets)]
correct = [
1.0
- for prediction, actual in zip(predictions, prepared.validation_targets)
+ for prediction, actual in zip(decision_predictions, decision_targets)
if (prediction > 0 and actual > 0) or (prediction < 0 and actual < 0)
]
non_zero = [
1.0
- for prediction, actual in zip(predictions, prepared.validation_targets)
+ for prediction, actual in zip(decision_predictions, decision_targets)
if prediction != 0 and actual != 0
]
buy_predictions = [
actual
- for prediction, actual in zip(predictions, prepared.validation_targets)
+ for prediction, actual in zip(decision_predictions, decision_targets)
if prediction > 0
]
buy_wins = [actual for actual in buy_predictions if actual > 0]
+ by_horizon = {}
+ baseline_by_horizon = {}
+ for horizon_index, horizon in enumerate(prepared.target_horizons):
+ horizon_errors = [
+ abs(row[horizon_index] - actual[horizon_index])
+ for row, actual in zip(predictions, prepared.validation_targets)
+ ]
+ horizon_baseline = [abs(actual[horizon_index]) for actual in prepared.validation_targets]
+ by_horizon[str(horizon)] = sum(horizon_errors) / len(horizon_errors) if horizon_errors else math.inf
+ baseline_by_horizon[str(horizon)] = (
+ sum(horizon_baseline) / len(horizon_baseline)
+ if horizon_baseline
+ else math.inf
+ )
+ probability_errors = [
+ (probabilities[row_index][decision] - (1.0 if target > 0 else 0.0)) ** 2
+ for row_index, target in enumerate(decision_targets)
+ ]
return {
"validation_mae": sum(errors) / len(errors) if errors else math.inf,
+ "validation_mae_by_horizon": by_horizon,
+ "baseline_mae_by_horizon": baseline_by_horizon,
"directional_accuracy": len(correct) / len(non_zero) if non_zero else 0.0,
"buy_precision": len(buy_wins) / len(buy_predictions) if buy_predictions else 0.0,
+ "probability_brier": sum(probability_errors) / len(probability_errors) if probability_errors else 1.0,
}
@@ -516,9 +705,26 @@ def _candidate_score(row: dict[str, Any]) -> float:
skill = float(row.get("skill", 0.0))
directional = float(row.get("directional_accuracy", 0.0))
buy_precision = float(row.get("buy_precision", 0.0))
+ probability_brier = float(row.get("probability_brier", 1.0))
return mae * (1.0 - max(0.0, skill) * 0.05) * (1.0 - max(0.0, directional - 0.5) * 0.03) * (
1.0 - max(0.0, buy_precision - 0.5) * 0.02
- )
+ ) * (1.0 + max(0.0, probability_brier - 0.25) * 0.02)
+
+
+def _forecast_loss(outputs: torch.Tensor, targets: torch.Tensor, up_targets: torch.Tensor, horizon_count: int) -> torch.Tensor:
+ values = outputs.view(outputs.shape[0], horizon_count, len(OUTPUT_LAYOUT))
+ mean_loss = nn.functional.smooth_l1_loss(values[:, :, 0], targets, beta=0.5)
+ quantile_losses = []
+ for offset, name in enumerate(("q10", "q50", "q90"), start=1):
+ quantile = QUANTILES[name]
+ errors = targets - values[:, :, offset]
+ quantile_losses.append(torch.maximum((quantile - 1.0) * errors, quantile * errors).mean())
+ logits = values[:, :, 4]
+ bce = nn.functional.binary_cross_entropy_with_logits(logits, up_targets, reduction="none")
+ probabilities = torch.sigmoid(logits)
+ pt = probabilities * up_targets + (1.0 - probabilities) * (1.0 - up_targets)
+ focal = ((1.0 - pt) ** 2.0 * bce).mean()
+ return mean_loss + 0.35 * sum(quantile_losses) / len(quantile_losses) + 0.15 * focal
def _export_recurrent_state(model: RecurrentReturnModel) -> dict[str, Any]:
@@ -528,6 +734,23 @@ def _export_recurrent_state(model: RecurrentReturnModel) -> dict[str, Any]:
}
+def _export_context_state(model: RecurrentReturnModel) -> dict[str, Any]:
+ exported: dict[str, Any] = {}
+ if model.attention is not None:
+ exported["attention_pooling"] = True
+ exported["attention_weight"] = _round_list(model.attention.weight.detach().cpu().squeeze(0).tolist())
+ exported["attention_bias"] = round(float(model.attention.bias.detach().cpu().item()), 10)
+ else:
+ exported["attention_pooling"] = False
+ if isinstance(model.context_norm, nn.LayerNorm):
+ exported["context_norm"] = True
+ exported["context_norm_weight"] = _round_list(model.context_norm.weight.detach().cpu().tolist())
+ exported["context_norm_bias"] = _round_list(model.context_norm.bias.detach().cpu().tolist())
+ else:
+ exported["context_norm"] = False
+ return exported
+
+
def _device(raw: str) -> torch.device:
value = raw.strip().lower()
if value == "auto":
@@ -554,10 +777,90 @@ def _return_scale(returns: list[float]) -> float:
return max(max(median, mean * 0.5), 1e-5)
+def _target_volatility_scale(candles: list[Candle], closes: list[float], end_index: int, horizon: int) -> float:
+ horizon = max(1, horizon)
+ close = max(closes[end_index], 1e-12)
+ candle = candles[end_index]
+ atr_scale = (candle.atr_14 / close) * math.sqrt(horizon) if candle.atr_14 is not None else 0.0
+ start = max(1, end_index - 96)
+ returns = [
+ math.log(closes[index] / closes[index - 1])
+ for index in range(start, end_index + 1)
+ if closes[index] > 0 and closes[index - 1] > 0
+ ]
+ realized = math.sqrt(sum(value * value for value in returns) / len(returns)) * math.sqrt(horizon) if returns else 0.0
+ return max(atr_scale * 0.7, realized, 0.0005)
+
+
+def _historical_klines(client: BybitClient, symbol: str, interval: str, limit: int) -> list[Candle]:
+ limit = max(1, limit)
+ rows_by_timestamp: dict[int, Candle] = {}
+ end: int | None = None
+ while len(rows_by_timestamp) < limit:
+ page_limit = min(1000, limit - len(rows_by_timestamp))
+ params: dict[str, Any] = {
+ "category": "spot",
+ "symbol": symbol,
+ "interval": interval,
+ "limit": page_limit,
+ }
+ if end is not None:
+ params["end"] = end
+ result = client.public_get("/v5/market/kline", params)
+ page = _parse_kline_rows(result.get("list", []))
+ if not page:
+ break
+ for candle in page:
+ rows_by_timestamp[candle.timestamp] = candle
+ oldest = min(candle.timestamp for candle in page)
+ if end is not None and oldest >= end:
+ break
+ end = oldest - 1
+ if len(page) < page_limit:
+ break
+ time.sleep(0.05)
+ return sorted(rows_by_timestamp.values(), key=lambda item: item.timestamp)[-limit:]
+
+
+def _parse_kline_rows(rows: Any) -> list[Candle]:
+ candles: list[Candle] = []
+ for row in rows or []:
+ if len(row) < 7:
+ continue
+ candles.append(
+ Candle(
+ timestamp=int(row[0]),
+ open=_float(row[1]),
+ high=_float(row[2]),
+ low=_float(row[3]),
+ close=_float(row[4]),
+ volume=_float(row[5]),
+ turnover=_float(row[6]),
+ )
+ )
+ candles.sort(key=lambda item: item.timestamp)
+ return candles
+
+
+def _float(value: Any, default: float = 0.0) -> float:
+ try:
+ return float(value)
+ except (TypeError, ValueError):
+ return default
+
+
def _clamp(value: float, low: float, high: float) -> float:
return max(low, min(high, value))
+def _sigmoid(value: float) -> float:
+ if value >= 40:
+ return 1.0
+ if value <= -40:
+ return 0.0
+ return 1 / (1 + math.exp(-value))
+
+
def _round_nested(value: Any) -> Any:
if isinstance(value, list):
return [_round_nested(item) for item in value]
@@ -580,6 +883,18 @@ def _strings(raw: str) -> list[str]:
return [item.strip().lower() for item in raw.split(",") if item.strip()]
+def _horizons(raw: str, decision_horizon: int) -> list[int]:
+ values = []
+ for value in _ints(raw or ""):
+ if 1 <= value <= 96 and value not in values:
+ values.append(value)
+ decision_horizon = max(1, min(96, int(decision_horizon)))
+ if decision_horizon not in values:
+ values.append(decision_horizon)
+ values.sort()
+ return values
+
+
def _feature_names_arg(raw: str) -> list[str]:
names = [item.strip() for item in raw.split(",") if item.strip()]
return names or list(DEFAULT_TORCH_FEATURES)