diff --git a/README.md b/README.md index ee3529b..16a26c6 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,22 @@ python tools\train_lstm_forecaster.py --symbols BTCUSDT,ETHUSDT,SOLUSDT,XRPUSDT, Файл из `TIME_SERIES_LSTM_MODEL_PATH` читается ботом автоматически. Даже если LSTM-параметры сохранены, сделка меняется только тогда, когда текущая walk-forward проверка в `crypto_spot_bot/time_series.py` показывает качество лучше baseline. +Для более тяжелого локального обучения можно использовать настоящий PyTorch `LSTM/GRU` trainer. PyTorch нужен только на машине обучения; в JSON экспортируются веса, а runtime на Raspberry Pi считает inference обычным Python-кодом: + +```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 ` + --symbols BTCUSDT,ETHUSDT,SOLUSDT,XRPUSDT,LTCUSDT ` + --limit 1000 ` + --architectures lstm,gru ` + --lookbacks 32,64 ` + --hidden-sizes 16,32 ` + --layers 1 ` + --epochs 60 +``` + +Экспортированные модели появляются в dashboard как `torch_lstm` или `torch_gru`; легкий `lstm`-кандидат остается доступен как fallback. + Автопереобучение запускает тот же train-скрипт, пишет лог в `runtime/lstm_retrain.log` и защищается от параллельных запусков: ```powershell @@ -86,7 +102,7 @@ bash tools/run_lstm_retrain.sh bash tools/install_lstm_retrainer_systemd.sh ``` -По умолчанию расписание переобучает каждые 6 часов с `--limit 1000`; Windows-установщик фиксирует пары `BTCUSDT,ETHUSDT,SOLUSDT,XRPUSDT,LTCUSDT`, чтобы первый scheduled run был предсказуемым. Параметры можно переопределить через env: `LSTM_RETRAIN_SYMBOLS`, `LSTM_RETRAIN_LIMIT`, `LSTM_RETRAIN_LOOKBACKS`, `LSTM_RETRAIN_UNITS`, `LSTM_RETRAIN_RIDGES`, `LSTM_RETRAIN_INTERVAL`, `LSTM_RETRAIN_ENV`. +По умолчанию Windows-расписание переобучает PyTorch `LSTM/GRU` каждые 6 часов с `--limit 1000`; Windows-установщик фиксирует пары `BTCUSDT,ETHUSDT,SOLUSDT,XRPUSDT,LTCUSDT`, чтобы первый scheduled run был предсказуемым. Параметры можно переопределить через env: `LSTM_RETRAIN_SYMBOLS`, `LSTM_RETRAIN_LIMIT`, `LSTM_RETRAIN_LOOKBACKS`, `LSTM_RETRAIN_ARCHITECTURES`, `LSTM_RETRAIN_HIDDEN_SIZES`, `LSTM_RETRAIN_LAYERS`, `LSTM_RETRAIN_DROPOUTS`, `LSTM_RETRAIN_EPOCHS`, `LSTM_RETRAIN_PATIENCE`, `LSTM_RETRAIN_INTERVAL`, `LSTM_RETRAIN_ENV`. Для старого легкого trainer можно запустить `tools\run_lstm_retrain.ps1 -Trainer reservoir`. ## Docker diff --git a/crypto_spot_bot/time_series.py b/crypto_spot_bot/time_series.py index 2cbcdad..ffe0c80 100644 --- a/crypto_spot_bot/time_series.py +++ b/crypto_spot_bot/time_series.py @@ -171,6 +171,9 @@ def _validate_candidates( lstm_artifact: dict[str, Any] | None = None, ) -> list[dict[str, float | str]]: models = ["naive", "drift", "ewma", "ar1", "ar3"] + torch_model = _torch_recurrent_model_name(symbol, lstm_artifact or {}) + if torch_model and _can_use_torch_recurrent(returns, symbol, lstm_artifact or {}): + models.append(torch_model) if _can_use_lstm(returns, settings, symbol, lstm_artifact or {}): models.append("lstm") rows: list[dict[str, float | str]] = [] @@ -206,6 +209,8 @@ def _predict_next_return( return _ar_predict(returns, 1) if model == "ar3": return _ar_predict(returns, 3) + if model in {"torch_lstm", "torch_gru"}: + return _torch_recurrent_predict(returns, symbol, lstm_artifact or {}) if model == "lstm" and settings is not None: return _lstm_predict(returns, settings, symbol, lstm_artifact or {}) return 0.0 @@ -285,6 +290,235 @@ def _clean_lstm_params(data: dict[str, Any]) -> dict[str, float | int]: return clean +def _torch_recurrent_model_name(symbol: str | None, lstm_artifact: dict[str, Any]) -> str | None: + entry = _torch_recurrent_entry(symbol, lstm_artifact) + if not entry: + return None + architecture = str(entry.get("architecture", "")).strip().lower() + if architecture in {"lstm", "gru"}: + return f"torch_{architecture}" + model = str(entry.get("model", "")).strip().lower() + return model if model in {"torch_lstm", "torch_gru"} else None + + +def _torch_recurrent_entry(symbol: str | None, lstm_artifact: dict[str, Any]) -> dict[str, Any] | None: + symbols = lstm_artifact.get("symbols") + entry = symbols.get(symbol.upper()) if symbol and isinstance(symbols, dict) else None + if not isinstance(entry, dict): + default = lstm_artifact.get("default") + entry = default if isinstance(default, dict) else None + if not isinstance(entry, dict): + return None + if not isinstance(entry.get("state_dict"), dict): + return None + return entry + + +def _can_use_torch_recurrent(returns: list[float], symbol: str | None, lstm_artifact: dict[str, Any]) -> bool: + entry = _torch_recurrent_entry(symbol, lstm_artifact) + if not entry: + return False + lookback = int(_clamp(_float_entry(entry, "lookback", 0.0), 4.0, 512.0)) + hidden_size = int(_clamp(_float_entry(entry, "hidden_size", 0.0), 1.0, 512.0)) + num_layers = int(_clamp(_float_entry(entry, "num_layers", 1.0), 1.0, 8.0)) + return len(returns) >= lookback + 1 and hidden_size > 0 and num_layers > 0 + + +def _torch_recurrent_predict( + returns: list[float], + symbol: str | None, + lstm_artifact: dict[str, Any], +) -> float: + entry = _torch_recurrent_entry(symbol, lstm_artifact) + model_name = _torch_recurrent_model_name(symbol, lstm_artifact) + if not entry or not model_name: + return _predict_next_return("drift", returns) + lookback = int(_clamp(_float_entry(entry, "lookback", 0.0), 4.0, 512.0)) + hidden_size = int(_clamp(_float_entry(entry, "hidden_size", 0.0), 1.0, 512.0)) + num_layers = int(_clamp(_float_entry(entry, "num_layers", 1.0), 1.0, 8.0)) + mean = _float_entry(entry, "mean", 0.0) + scale = max(_float_entry(entry, "scale", _return_scale(returns)), 1e-8) + clip = _clamp(_float_entry(entry, "clip", 8.0), 1.0, 50.0) + if len(returns) < lookback: + return _predict_next_return("drift", returns) + + normalized = [_clamp((value - mean) / scale, -clip, clip) for value in returns[-lookback:]] + try: + hidden = _torch_recurrent_hidden( + normalized, + entry=entry, + model_name=model_name, + hidden_size=hidden_size, + num_layers=num_layers, + ) + if hidden is None: + return _predict_next_return("drift", returns) + head_weight = _float_vector(entry.get("head_weight")) + head_bias = _float_entry(entry, "head_bias", 0.0) + if len(head_weight) != hidden_size: + return _predict_next_return("drift", returns) + normalized_prediction = sum(weight * value for weight, value in zip(head_weight, hidden)) + head_bias + if not math.isfinite(normalized_prediction): + return _predict_next_return("drift", returns) + prediction = _clamp(normalized_prediction, -clip, clip) * scale + mean + except (IndexError, KeyError, TypeError, ValueError, OverflowError): + return _predict_next_return("drift", returns) + + recent_abs = sorted(abs(value) for value in returns[-48:]) if len(returns) >= 8 else [0.01] + cap = max(recent_abs[int(len(recent_abs) * 0.9)], 0.0002) + return _clamp(prediction, -cap, cap) + + +def _torch_recurrent_hidden( + normalized: list[float], + *, + entry: dict[str, Any], + model_name: str, + hidden_size: int, + num_layers: int, +) -> list[float] | None: + state = entry.get("state_dict") + if not isinstance(state, dict): + return None + h_layers = [[0.0 for _ in range(hidden_size)] for _ in range(num_layers)] + c_layers = [[0.0 for _ in range(hidden_size)] for _ in range(num_layers)] + for value in normalized: + layer_input = [value] + for layer in range(num_layers): + if model_name == "torch_lstm": + next_hidden, next_cell = _torch_lstm_step(layer_input, h_layers[layer], c_layers[layer], state, layer) + h_layers[layer] = next_hidden + c_layers[layer] = next_cell + elif model_name == "torch_gru": + h_layers[layer] = _torch_gru_step(layer_input, h_layers[layer], state, layer) + else: + return None + layer_input = h_layers[layer] + return h_layers[-1] + + +def _torch_lstm_step( + inputs: list[float], + hidden: list[float], + cell: list[float], + state: dict[str, Any], + layer: int, +) -> tuple[list[float], list[float]]: + hidden_size = len(hidden) + gates = _torch_gate_values(inputs, hidden, state, layer, gate_count=4) + input_gate = [_sigmoid(value) for value in gates[0]] + forget_gate = [_sigmoid(value) for value in gates[1]] + cell_gate = [math.tanh(value) for value in gates[2]] + output_gate = [_sigmoid(value) for value in gates[3]] + next_cell = [ + forget_gate[index] * cell[index] + input_gate[index] * cell_gate[index] + for index in range(hidden_size) + ] + next_hidden = [ + output_gate[index] * math.tanh(next_cell[index]) + for index in range(hidden_size) + ] + return next_hidden, next_cell + + +def _torch_gru_step( + inputs: list[float], + hidden: list[float], + state: dict[str, Any], + layer: int, +) -> list[float]: + hidden_size = len(hidden) + weight_ih = _float_matrix(state[f"weight_ih_l{layer}"]) + weight_hh = _float_matrix(state[f"weight_hh_l{layer}"]) + bias_ih = _float_vector(state[f"bias_ih_l{layer}"]) + bias_hh = _float_vector(state[f"bias_hh_l{layer}"]) + + def gate_input(gate: int) -> list[float]: + start = gate * hidden_size + output = [] + for index in range(hidden_size): + row = start + index + output.append(_dot(weight_ih[row], inputs) + bias_ih[row]) + return output + + def gate_hidden(gate: int) -> list[float]: + start = gate * hidden_size + output = [] + for index in range(hidden_size): + row = start + index + output.append(_dot(weight_hh[row], hidden) + bias_hh[row]) + return output + + reset_input = gate_input(0) + update_input = gate_input(1) + new_input = gate_input(2) + reset_hidden = gate_hidden(0) + update_hidden = gate_hidden(1) + new_hidden = gate_hidden(2) + + reset_gate = [_sigmoid(reset_input[index] + reset_hidden[index]) for index in range(hidden_size)] + update_gate = [_sigmoid(update_input[index] + update_hidden[index]) for index in range(hidden_size)] + candidate = [ + math.tanh(new_input[index] + reset_gate[index] * new_hidden[index]) + for index in range(hidden_size) + ] + return [ + (1 - update_gate[index]) * candidate[index] + update_gate[index] * hidden[index] + for index in range(hidden_size) + ] + + +def _torch_gate_values( + inputs: list[float], + hidden: list[float], + state: dict[str, Any], + layer: int, + gate_count: int, +) -> list[list[float]]: + hidden_size = len(hidden) + weight_ih = _float_matrix(state[f"weight_ih_l{layer}"]) + weight_hh = _float_matrix(state[f"weight_hh_l{layer}"]) + bias_ih = _float_vector(state[f"bias_ih_l{layer}"]) + bias_hh = _float_vector(state[f"bias_hh_l{layer}"]) + gates: list[list[float]] = [] + for gate in range(gate_count): + values = [] + start = gate * hidden_size + for index in range(hidden_size): + row = start + index + values.append(_dot(weight_ih[row], inputs) + _dot(weight_hh[row], hidden) + bias_ih[row] + bias_hh[row]) + gates.append(values) + return gates + + +def _float_entry(data: dict[str, Any], key: str, default: float) -> float: + value = data.get(key) + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, str): + try: + return float(value) + except ValueError: + return default + return default + + +def _float_vector(data: Any) -> list[float]: + if not isinstance(data, list): + return [] + return [float(value) for value in data] + + +def _float_matrix(data: Any) -> list[list[float]]: + if not isinstance(data, list): + return [] + return [_float_vector(row) for row in data] + + +def _dot(left: list[float], right: list[float]) -> float: + return sum(left[index] * right[index] for index in range(min(len(left), len(right)))) + + def _lstm_predict( returns: list[float], settings: Settings, diff --git a/requirements-train.txt b/requirements-train.txt new file mode 100644 index 0000000..8a206c3 --- /dev/null +++ b/requirements-train.txt @@ -0,0 +1,5 @@ +# Optional local-only training dependency for tools/train_torch_recurrent_forecaster.py. +# Prefer the CPU wheel on Windows: +# python -m pip install torch --index-url https://download.pytorch.org/whl/cpu +torch>=2.5 +numpy>=2.0 diff --git a/tests/test_time_series.py b/tests/test_time_series.py index de5a01c..aee22dd 100644 --- a/tests/test_time_series.py +++ b/tests/test_time_series.py @@ -122,3 +122,50 @@ def test_time_series_forecaster_reads_lstm_artifact(make_settings, tmp_path) -> assert forecast.usable is True assert any(candidate["model"] == "lstm" for candidate in forecast.candidates) + + +def test_time_series_forecaster_reads_torch_gru_artifact(make_settings, tmp_path) -> None: + artifact_path = tmp_path / "lstm_forecaster.json" + hidden_size = 2 + artifact_path.write_text( + json.dumps( + { + "version": 2, + "type": "pytorch_recurrent_forecaster", + "symbols": { + "BTCUSDT": { + "model": "torch_gru", + "architecture": "gru", + "lookback": 8, + "hidden_size": hidden_size, + "num_layers": 1, + "mean": 0.0, + "scale": 0.001, + "clip": 8.0, + "state_dict": { + "weight_ih_l0": [[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], + "head_bias": 0.2, + }, + }, + } + ), + encoding="utf-8", + ) + settings = make_settings( + tmp_path, + time_series_min_candles=80, + time_series_validation_window=20, + time_series_lstm_enabled=True, + time_series_lstm_model_path=artifact_path, + ) + returns = [0.00015 if index % 4 else -0.00005 for index in range(140)] + + forecast = TimeSeriesForecaster(settings).forecast(_candles_from_returns(returns), symbol="BTCUSDT") + + assert forecast.usable is True + assert any(candidate["model"] == "torch_gru" for candidate in forecast.candidates) diff --git a/tools/install_windows_lstm_retrainer.ps1 b/tools/install_windows_lstm_retrainer.ps1 index 40cfdab..27f5cba 100644 --- a/tools/install_windows_lstm_retrainer.ps1 +++ b/tools/install_windows_lstm_retrainer.ps1 @@ -3,7 +3,10 @@ param( [string]$TaskName = "TradeBot LSTM Retrainer", [int]$EveryHours = 6, [string]$Symbols = "BTCUSDT,ETHUSDT,SOLUSDT,XRPUSDT,LTCUSDT", - [int]$Limit = 1000 + [int]$Limit = 1000, + [ValidateSet("torch", "reservoir")] + [string]$Trainer = "torch", + [int]$FirstRunMinutes = 0 ) $ErrorActionPreference = "Stop" @@ -14,7 +17,7 @@ if (-not (Test-Path $Runner)) { throw "Runner not found: $Runner" } -$actionArgs = "-NoProfile -ExecutionPolicy Bypass -File `"$Runner`"" +$actionArgs = "-NoProfile -ExecutionPolicy Bypass -File `"$Runner`" -Trainer $Trainer" if ($Symbols) { $actionArgs += " -Symbols `"$Symbols`"" } @@ -24,7 +27,7 @@ if ($Limit -gt 0) { $action = New-ScheduledTaskAction -Execute "powershell.exe" -Argument $actionArgs -WorkingDirectory $RepoRoot $trigger = New-ScheduledTaskTrigger ` -Once ` - -At (Get-Date).AddMinutes(5) ` + -At (Get-Date).AddMinutes($(if ($FirstRunMinutes -gt 0) { $FirstRunMinutes } else { $EveryHours * 60 })) ` -RepetitionInterval (New-TimeSpan -Hours $EveryHours) ` -RepetitionDuration (New-TimeSpan -Days 3650) $principal = New-ScheduledTaskPrincipal ` diff --git a/tools/run_lstm_retrain.ps1 b/tools/run_lstm_retrain.ps1 index 7d96a86..5383b22 100644 --- a/tools/run_lstm_retrain.ps1 +++ b/tools/run_lstm_retrain.ps1 @@ -1,10 +1,18 @@ [CmdletBinding()] param( + [ValidateSet("torch", "reservoir")] + [string]$Trainer = "torch", [string]$Symbols = "", [int]$Limit = 0, [string]$Lookbacks = "", [string]$Units = "", [string]$Ridges = "", + [string]$Architectures = "", + [string]$HiddenSizes = "", + [string]$Layers = "", + [string]$Dropouts = "", + [int]$Epochs = 0, + [int]$Patience = 0, [string]$Interval = "", [string]$EnvFile = "" ) @@ -47,9 +55,15 @@ if (-not $Symbols -and $env:LSTM_RETRAIN_SYMBOLS) { $Symbols = $env:LSTM_RETRAIN if ($Limit -le 0) { $Limit = if ($env:LSTM_RETRAIN_LIMIT) { [int]$env:LSTM_RETRAIN_LIMIT } else { 1000 } } -if (-not $Lookbacks) { $Lookbacks = if ($env:LSTM_RETRAIN_LOOKBACKS) { $env:LSTM_RETRAIN_LOOKBACKS } else { "16,32" } } +if (-not $Lookbacks) { $Lookbacks = if ($env:LSTM_RETRAIN_LOOKBACKS) { $env:LSTM_RETRAIN_LOOKBACKS } else { "32,64" } } if (-not $Units) { $Units = if ($env:LSTM_RETRAIN_UNITS) { $env:LSTM_RETRAIN_UNITS } else { "4,6" } } if (-not $Ridges) { $Ridges = if ($env:LSTM_RETRAIN_RIDGES) { $env:LSTM_RETRAIN_RIDGES } else { "0.001" } } +if (-not $Architectures) { $Architectures = if ($env:LSTM_RETRAIN_ARCHITECTURES) { $env:LSTM_RETRAIN_ARCHITECTURES } else { "lstm,gru" } } +if (-not $HiddenSizes) { $HiddenSizes = if ($env:LSTM_RETRAIN_HIDDEN_SIZES) { $env:LSTM_RETRAIN_HIDDEN_SIZES } else { "16,32" } } +if (-not $Layers) { $Layers = if ($env:LSTM_RETRAIN_LAYERS) { $env:LSTM_RETRAIN_LAYERS } else { "1" } } +if (-not $Dropouts) { $Dropouts = if ($env:LSTM_RETRAIN_DROPOUTS) { $env:LSTM_RETRAIN_DROPOUTS } else { "0.0" } } +if ($Epochs -le 0) { $Epochs = if ($env:LSTM_RETRAIN_EPOCHS) { [int]$env:LSTM_RETRAIN_EPOCHS } else { 60 } } +if ($Patience -le 0) { $Patience = if ($env:LSTM_RETRAIN_PATIENCE) { [int]$env:LSTM_RETRAIN_PATIENCE } else { 10 } } if (-not $Interval -and $env:LSTM_RETRAIN_INTERVAL) { $Interval = $env:LSTM_RETRAIN_INTERVAL } if (-not $EnvFile -and $env:LSTM_RETRAIN_ENV) { $EnvFile = $env:LSTM_RETRAIN_ENV } if (-not $EnvFile -and (Test-Path (Join-Path $RepoRoot ".env"))) { $EnvFile = Join-Path $RepoRoot ".env" } @@ -66,14 +80,29 @@ try { } $python = Resolve-Python - $trainerArgs = @( - "-u", - "tools\train_lstm_forecaster.py", - "--limit", $Limit.ToString(), - "--lookbacks", $Lookbacks, - "--units", $Units, - "--ridges", $Ridges - ) + if ($Trainer -eq "torch") { + $trainerArgs = @( + "-u", + "tools\train_torch_recurrent_forecaster.py", + "--limit", $Limit.ToString(), + "--lookbacks", $Lookbacks, + "--architectures", $Architectures, + "--hidden-sizes", $HiddenSizes, + "--layers", $Layers, + "--dropouts", $Dropouts, + "--epochs", $Epochs.ToString(), + "--patience", $Patience.ToString() + ) + } else { + $trainerArgs = @( + "-u", + "tools\train_lstm_forecaster.py", + "--limit", $Limit.ToString(), + "--lookbacks", $Lookbacks, + "--units", $Units, + "--ridges", $Ridges + ) + } if ($Symbols) { $trainerArgs += @("--symbols", $Symbols) } if ($Interval) { $trainerArgs += @("--interval", $Interval) } if ($EnvFile) { $trainerArgs += @("--env", $EnvFile) } diff --git a/tools/train_torch_recurrent_forecaster.py b/tools/train_torch_recurrent_forecaster.py new file mode 100644 index 0000000..f1c681f --- /dev/null +++ b/tools/train_torch_recurrent_forecaster.py @@ -0,0 +1,434 @@ +from __future__ import annotations + +import argparse +import json +import math +import sys +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +try: + import torch + from torch import nn + from torch.utils.data import DataLoader, TensorDataset +except ImportError as exc: # pragma: no cover - exercised on machines without training deps. + raise SystemExit( + "PyTorch is not installed. Install local training deps with: " + "python -m pip install torch --index-url https://download.pytorch.org/whl/cpu" + ) from exc + +from crypto_spot_bot.bybit import BybitClient +from crypto_spot_bot.config import load_settings +from crypto_spot_bot.time_series import _log_returns + + +@dataclass(slots=True) +class PreparedData: + train_x: torch.Tensor + train_y: torch.Tensor + validation_x: torch.Tensor + validation_y: torch.Tensor + validation_returns: list[float] + mean: float + scale: float + train_samples: int + validation_samples: int + + +class RecurrentReturnModel(nn.Module): + def __init__( + self, + *, + architecture: str, + hidden_size: int, + num_layers: int, + dropout: float, + ) -> None: + super().__init__() + recurrent_cls = nn.LSTM if architecture == "lstm" else nn.GRU + self.rnn = recurrent_cls( + input_size=1, + hidden_size=hidden_size, + num_layers=num_layers, + dropout=dropout if num_layers > 1 else 0.0, + batch_first=True, + ) + self.head = nn.Linear(hidden_size, 1) + + def forward(self, values: torch.Tensor) -> torch.Tensor: + output, _state = self.rnn(values) + return self.head(output[:, -1, :]).squeeze(-1) + + +def main() -> None: + args = _parse_args() + if args.threads > 0: + torch.set_num_threads(args.threads) + _seed(args.seed) + + settings = load_settings(args.env) + client = BybitClient(settings) + symbols = _symbols(args.symbols, settings, client) + 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) + + artifact: dict[str, Any] = { + "version": 2, + "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, + "device": str(device), + "symbols": {}, + } + + for symbol in symbols: + result = _train_symbol( + client=client, + symbol=symbol, + interval=interval, + limit=args.limit, + validation_window=args.validation_window, + architectures=_strings(args.architectures), + lookbacks=_ints(args.lookbacks), + hidden_sizes=_ints(args.hidden_sizes), + layers_values=_ints(args.layers), + dropouts=_floats(args.dropouts), + epochs=args.epochs, + patience=args.patience, + batch_size=args.batch_size, + learning_rate=args.learning_rate, + weight_decay=args.weight_decay, + clip=args.clip, + device=device, + seed=args.seed, + ) + if result is None: + print(f"{symbol}: skipped, not enough candles or train/validation samples") + continue + artifact["symbols"][symbol] = result + print( + f"{symbol}: model={result['model']} lookback={result['lookback']} " + f"hidden={result['hidden_size']} layers={result['num_layers']} " + f"mae={result['validation_mae_percent']:.5f}% " + f"baseline={result['baseline_mae_percent']:.5f}% skill={result['skill']:.4f}" + ) + + output.parent.mkdir(parents=True, exist_ok=True) + tmp_output = output.with_name(f"{output.name}.tmp") + tmp_output.write_text(json.dumps(artifact, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + tmp_output.replace(output) + print(f"saved {output}") + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Train PyTorch LSTM/GRU forecast models on Bybit spot candles.") + parser.add_argument("--env", default=None, help="Path to .env file.") + parser.add_argument("--symbols", default="", help="Comma-separated symbols. Defaults to configured or popular pairs.") + parser.add_argument("--interval", default="", help="Bybit kline interval. Defaults to BASE_INTERVAL.") + 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 returns used for validation.") + 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="16,32", help="Comma-separated hidden sizes.") + parser.add_argument("--layers", default="1", help="Comma-separated recurrent layer counts.") + parser.add_argument("--dropouts", default="0.0", help="Comma-separated dropout values; only used with layers > 1.") + parser.add_argument("--epochs", type=int, default=60, help="Maximum epochs per hyperparameter candidate.") + parser.add_argument("--patience", type=int, default=10, help="Early stopping patience in epochs.") + parser.add_argument("--batch-size", type=int, default=64, help="Training batch size.") + 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 returns and predictions to this range.") + 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.") + parser.add_argument("--output", default="", help="Output JSON path. Defaults to TIME_SERIES_LSTM_MODEL_PATH.") + return parser.parse_args() + + +def _symbols(raw: str, settings: Any, client: BybitClient) -> list[str]: + if raw.strip(): + return [item.strip().upper() for item in raw.split(",") if item.strip()] + if settings.symbols: + return list(settings.symbols) + return client.popular_spot_symbols(settings.top_symbols_count) + + +def _train_symbol( + *, + client: BybitClient, + symbol: str, + interval: str, + limit: int, + validation_window: int, + architectures: list[str], + lookbacks: list[int], + hidden_sizes: list[int], + layers_values: list[int], + dropouts: list[float], + epochs: int, + patience: int, + batch_size: int, + learning_rate: float, + weight_decay: float, + clip: float, + device: torch.device, + seed: int, +) -> dict[str, Any] | None: + candles = client.klines(symbol, interval, limit) + closes = [float(candle.close) for candle in candles if candle.close > 0] + returns = _log_returns(closes) + if len(returns) < max(100, validation_window + 80): + return None + + best: dict[str, Any] | None = None + for lookback in lookbacks: + prepared = _prepare_data(returns, lookback, validation_window, clip, device) + if prepared is None: + continue + baseline_mae = sum(abs(value) for value in prepared.validation_returns) / len(prepared.validation_returns) + for architecture in architectures: + if architecture not in {"lstm", "gru"}: + continue + for hidden_size in hidden_sizes: + for num_layers in layers_values: + for dropout in dropouts: + candidate = _fit_candidate( + prepared=prepared, + architecture=architecture, + hidden_size=hidden_size, + num_layers=num_layers, + dropout=dropout, + epochs=epochs, + patience=patience, + batch_size=batch_size, + learning_rate=learning_rate, + weight_decay=weight_decay, + clip=clip, + device=device, + seed=seed, + ) + validation_mae = float(candidate["validation_mae"]) + skill = (baseline_mae - validation_mae) / baseline_mae if baseline_mae > 0 else 0.0 + row = { + **candidate, + "model": f"torch_{architecture}", + "architecture": architecture, + "lookback": lookback, + "hidden_size": hidden_size, + "num_layers": num_layers, + "dropout": dropout, + "mean": prepared.mean, + "scale": prepared.scale, + "clip": clip, + "validation_mae_percent": validation_mae * 100, + "baseline_mae_percent": baseline_mae * 100, + "skill": skill, + "candles": len(candles), + "returns": len(returns), + "train_samples": prepared.train_samples, + "validation_samples": prepared.validation_samples, + } + if best is None or validation_mae < float(best["validation_mae"]): + best = row + if best is None: + return None + best.pop("validation_mae", None) + return best + + +def _prepare_data( + returns: list[float], + lookback: int, + validation_window: int, + clip: float, + device: torch.device, +) -> PreparedData | None: + validation_window = min(max(16, validation_window), max(16, len(returns) // 3)) + split = len(returns) - validation_window + if split <= lookback + 16: + return None + train_returns = returns[:split] + mean = sum(train_returns) / len(train_returns) + scale = _return_scale(train_returns) + normalized = [_clamp((value - mean) / scale, -clip, clip) for value in returns] + train_x: list[list[list[float]]] = [] + train_y: list[float] = [] + validation_x: list[list[list[float]]] = [] + validation_y: list[float] = [] + validation_returns: list[float] = [] + for target_index in range(lookback, len(returns)): + row = [[value] for value in normalized[target_index - lookback : target_index]] + target = normalized[target_index] + if target_index < split: + train_x.append(row) + train_y.append(target) + else: + validation_x.append(row) + validation_y.append(target) + validation_returns.append(returns[target_index]) + if len(train_x) < 24 or len(validation_x) < 8: + return None + return PreparedData( + train_x=torch.tensor(train_x, dtype=torch.float32, device=device), + train_y=torch.tensor(train_y, 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_returns=validation_returns, + mean=mean, + scale=scale, + train_samples=len(train_x), + validation_samples=len(validation_x), + ) + + +def _fit_candidate( + *, + prepared: PreparedData, + architecture: str, + hidden_size: int, + num_layers: int, + dropout: float, + epochs: int, + patience: int, + batch_size: int, + learning_rate: float, + weight_decay: float, + clip: float, + device: torch.device, + seed: int, +) -> dict[str, Any]: + _seed(seed) + model = RecurrentReturnModel( + architecture=architecture, + hidden_size=hidden_size, + num_layers=num_layers, + dropout=dropout, + ).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), + batch_size=max(1, batch_size), + shuffle=True, + generator=generator, + ) + + best_state: dict[str, torch.Tensor] | None = None + best_mae = math.inf + best_epoch = 0 + stale_epochs = 0 + for epoch in range(1, max(1, epochs) + 1): + model.train() + for batch_x, batch_y in loader: + optimizer.zero_grad(set_to_none=True) + loss = criterion(model(batch_x), batch_y) + loss.backward() + nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) + optimizer.step() + + validation_mae = _validation_mae(model, prepared, clip) + if validation_mae + 1e-12 < best_mae: + best_mae = validation_mae + best_epoch = epoch + best_state = {key: value.detach().cpu().clone() for key, value in model.state_dict().items()} + stale_epochs = 0 + else: + stale_epochs += 1 + if stale_epochs >= max(1, patience): + break + + if best_state: + model.load_state_dict(best_state) + return { + "validation_mae": best_mae, + "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), + } + + +def _validation_mae(model: nn.Module, prepared: PreparedData, clip: float) -> float: + model.eval() + with torch.no_grad(): + normalized_predictions = model(prepared.validation_x).detach().cpu().tolist() + errors = [] + for prediction, actual in zip(normalized_predictions, prepared.validation_returns): + raw_prediction = _clamp(float(prediction), -clip, clip) * prepared.scale + prepared.mean + errors.append(abs(raw_prediction - actual)) + return sum(errors) / len(errors) if errors else math.inf + + +def _export_recurrent_state(model: RecurrentReturnModel) -> dict[str, Any]: + return { + key: _round_nested(value.detach().cpu().tolist()) + for key, value in model.rnn.state_dict().items() + } + + +def _device(raw: str) -> torch.device: + value = raw.strip().lower() + if value == "auto": + if torch.cuda.is_available(): + return torch.device("cuda") + if getattr(torch.backends, "mps", None) and torch.backends.mps.is_available(): + return torch.device("mps") + return torch.device("cpu") + return torch.device(value) + + +def _seed(seed: int) -> None: + torch.manual_seed(seed) + if torch.cuda.is_available(): + torch.cuda.manual_seed_all(seed) + + +def _return_scale(returns: list[float]) -> float: + values = sorted(abs(value) for value in returns if math.isfinite(value)) + if not values: + return 0.0005 + median = values[len(values) // 2] + mean = sum(values) / len(values) + return max(max(median, mean * 0.5), 1e-5) + + +def _clamp(value: float, low: float, high: float) -> float: + return max(low, min(high, value)) + + +def _round_nested(value: Any) -> Any: + if isinstance(value, list): + return [_round_nested(item) for item in value] + return round(float(value), 10) + + +def _round_list(values: list[float]) -> list[float]: + return [round(float(value), 10) for value in values] + + +def _ints(raw: str) -> list[int]: + return [int(item.strip()) for item in raw.split(",") if item.strip()] + + +def _floats(raw: str) -> list[float]: + return [float(item.strip()) for item in raw.split(",") if item.strip()] + + +def _strings(raw: str) -> list[str]: + return [item.strip().lower() for item in raw.split(",") if item.strip()] + + +if __name__ == "__main__": + main()