sevenhill/TradeBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add probabilistic multi-horizon Torch forecaster

Browse Source
This commit is contained in:
Codex
2026-06-22 22:02:38 +03:00
parent 8ae6d4e3a5
commit a548c0e890
8 changed files with 1114 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+9 -7
View File
@@ -65,17 +65,19 @@ Dashboard: <http://127.0.0.1:8787/>
```powershell
.\.venv\Scripts\python.exe -m pip install torch --index-url https://download.pytorch.org/whl/cpu
.\.venv\Scripts\python.exe tools\train_torch_recurrent_forecaster.py `
--limit 1000 `
--limit 3000 `
--architectures lstm,gru `
--lookbacks 32,64 `
--hidden-sizes 32,64 `
--lookbacks 64 `
--hidden-sizes 64,96 `
--layers 2 `
--dropouts 0.15 `
--horizon 3 `
--epochs 60
--horizons 1,3,6,12 `
--context-symbols BTCUSDT,ETHUSDT `
--epochs 70
```
Новый artifact версии 3 обучается как multifeature direct-horizon модель: вход `input_size=26` включает доходности, форму свечи, объем, ATR%, RSI, MACD histogram, расстояние до EMA50/EMA200 и числовые признаки текущего шаблона пары: score, bullish/bearish/range, pullback, reversal, stabilized drop, breakout/breakdown, fast drop, volume spike и позицию цены в 20-свечном диапазоне. Цель обучается сразу на горизонт `TIME_SERIES_FORECAST_HORIZON`, без умножения one-step прогноза.
Новый artifact версии 4 обучается как probabilistic multi-horizon модель: вход включает доходности, форму свечи, объем, ATR%, realized volatility, RSI/MACD/EMA slopes, 4h/24h rolling trend, дневные EMA-признаки, BTC/ETH cross-asset признаки и числовые признаки текущего шаблона пары. Цель обучается как `future log return - комиссии - проскальзывание`, нормализованная на текущую волатильность. Модель сразу прогнозирует горизонты `1/3/6/12`, quantile-оценки `q10/q50/q90` и `P(up)`.
Файл из `TIME_SERIES_LSTM_MODEL_PATH` читается ботом автоматически, если `TIME_SERIES_FORECAST_ENABLED=true`. В стратегии `torch_forecast` экспортированная PyTorch LSTM/GRU модель является единственным направляющим сигналом для входа и forecast-выхода. Экспортированные модели появляются в dashboard как `PyTorch LSTM` или `PyTorch GRU`; старый легкий reservoir LSTM-кандидат и все встроенные не-torch прогнозы удалены.
@@ -86,9 +88,9 @@ powershell -ExecutionPolicy Bypass -File tools\run_torch_retrain.ps1
powershell -ExecutionPolicy Bypass -File tools\install_windows_torch_retrainer.ps1
```
По умолчанию Windows-расписание переобучает PyTorch `LSTM/GRU` каждые 6 часов с `--limit 1000` на парах `BTCUSDT,ETHUSDT,SOLUSDT,LTCUSDT`. Параметры можно переопределить через env: `TORCH_RETRAIN_SYMBOLS`, `TORCH_RETRAIN_LIMIT`, `TORCH_RETRAIN_LOOKBACKS`, `TORCH_RETRAIN_ARCHITECTURES`, `TORCH_RETRAIN_HIDDEN_SIZES`, `TORCH_RETRAIN_LAYERS`, `TORCH_RETRAIN_DROPOUTS`, `TORCH_RETRAIN_EPOCHS`, `TORCH_RETRAIN_PATIENCE`, `TORCH_RETRAIN_INTERVAL`, `TORCH_RETRAIN_ENV`.
По умолчанию Windows-расписание переобучает PyTorch `LSTM/GRU` каждые 6 часов с `--limit 3000` на парах `BTCUSDT,ETHUSDT,SOLUSDT,LTCUSDT`. Параметры можно переопределить через env: `TORCH_RETRAIN_SYMBOLS`, `TORCH_RETRAIN_LIMIT`, `TORCH_RETRAIN_LOOKBACKS`, `TORCH_RETRAIN_ARCHITECTURES`, `TORCH_RETRAIN_HIDDEN_SIZES`, `TORCH_RETRAIN_LAYERS`, `TORCH_RETRAIN_DROPOUTS`, `TORCH_RETRAIN_HORIZON`, `TORCH_RETRAIN_HORIZONS`, `TORCH_RETRAIN_CONTEXT_SYMBOLS`, `TORCH_RETRAIN_FEATURES`, `TORCH_RETRAIN_EPOCHS`, `TORCH_RETRAIN_PATIENCE`, `TORCH_RETRAIN_INTERVAL`, `TORCH_RETRAIN_ENV`.
Дополнительно для нового multifeature trainer доступны env-переменные `TORCH_RETRAIN_HORIZON` и `TORCH_RETRAIN_FEATURES`.
Внутри recurrent модели используются exportable attention pooling и LayerNorm перед forecast-head; Raspberry Pi по-прежнему исполняет модель из JSON без PyTorch runtime.
## Docker
crypto_spot_bot/bot.py
+2
View File
@@ -266,6 +266,8 @@ class CryptoSpotBot:
forecasts[symbol] = self.forecaster.forecast(
self.market.candles.get(symbol, []),
symbol=symbol,
market_candles=self.market.candles,
trend_candles=self.market.trend_candles.get(symbol, []),
).as_dict()
self.market.forecasts = forecasts
crypto_spot_bot/dashboard.py
+2
View File
@@ -839,8 +839,10 @@ HTML = r"""
}
return `<div class="forecast-line">
<div class="forecast-chip"><b>Модель</b>${escapeHtml(modelName(forecast.model || '-'))}</div>
<div class="forecast-chip"><b>Горизонт</b>${num(forecast.horizon || 0, 0)}ч</div>
<div class="forecast-chip"><b>P роста</b>${num((forecast.probability_up || 0) * 100, 1)}%</div>
<div class="forecast-chip"><b>Ожидание</b><span class="${signedClass(forecast.expected_return_percent || 0)}">${signedNum(forecast.expected_return_percent, 3)}%</span></div>
<div class="forecast-chip"><b>Q10/Q50/Q90</b>${signedNum(forecast.quantile_10_percent, 2)} / ${signedNum(forecast.quantile_50_percent, 2)} / ${signedNum(forecast.quantile_90_percent, 2)}%</div>
<div class="forecast-chip"><b>Волат.</b>${num(forecast.volatility_percent, 3)}%</div>
</div>`;
}
crypto_spot_bot/time_series.py
+611 -12
View File
@@ -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
tests/test_time_series.py
+89
View File
@@ -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"]
tools/install_windows_torch_retrainer.ps1
+9 -1
View File
@@ -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 `
tools/run_torch_retrain.ps1
+11 -5
View File
@@ -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
tools/train_torch_recurrent_forecaster.py
+380 -65
View File
@@ -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"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)