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.indicators import add_indicators from crypto_spot_bot.models import Candle from crypto_spot_bot.time_series import DEFAULT_TORCH_FEATURES, _feature_matrix, _log_returns @dataclass(slots=True) class PreparedData: train_x: torch.Tensor train_y: torch.Tensor validation_x: torch.Tensor validation_y: torch.Tensor validation_targets: list[float] feature_names: list[str] feature_means: list[float] feature_scales: list[float] target_mean: float target_scale: float train_samples: int validation_samples: int class RecurrentReturnModel(nn.Module): def __init__( self, *, architecture: str, input_size: int, 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=input_size, 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) horizon = args.horizon if args.horizon > 0 else max(1, settings.time_series_forecast_horizon) feature_names = _feature_names_arg(args.features) artifact: dict[str, Any] = { "version": 3, "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, "direct_horizon": True, "feature_names": feature_names, "feature_count": len(feature_names), "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, target_horizon=horizon, feature_names=feature_names, 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"features={result['input_size']} hidden={result['hidden_size']} " f"layers={result['num_layers']} horizon={result['target_horizon']} " f"mae={result['validation_mae_percent']:.5f}% " f"baseline={result['baseline_mae_percent']:.5f}% " f"skill={result['skill']:.4f} dir={result['directional_accuracy']:.3f}" ) 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 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("--features", default=",".join(DEFAULT_TORCH_FEATURES), help="Comma-separated feature names.") 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.") parser.add_argument("--layers", default="2", help="Comma-separated recurrent layer counts.") parser.add_argument("--dropouts", default="0.15", 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 features, targets and predictions.") 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, target_horizon: int, feature_names: list[str], 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) 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): return None best: dict[str, Any] | None = None for lookback in lookbacks: prepared = _prepare_data( candles=candles, feature_names=feature_names, lookback=lookback, target_horizon=target_horizon, 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) 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: if num_layers <= 1 and dropout != 0.0: continue candidate = _fit_candidate( prepared=prepared, architecture=architecture, input_size=len(feature_names), 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, "target_horizon": target_horizon, "direct_horizon": True, "input_size": len(feature_names), "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, "hidden_size": hidden_size, "num_layers": num_layers, "dropout": dropout if num_layers > 1 else 0.0, "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, } score = _candidate_score(row) if best is None or score < _candidate_score(best): best = row if best is None: return None best.pop("validation_mae", None) return best def _prepare_data( *, candles: list[Candle], feature_names: list[str], lookback: int, target_horizon: int, 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): current = closes[end_index] future = closes[end_index + target_horizon] if current <= 0 or future <= 0: continue window = feature_rows[end_index - lookback + 1 : end_index + 1] if len(window) != lookback: continue samples.append((window, math.log(future / current))) if len(samples) < 48: return None validation_window = min(max(16, validation_window), max(16, len(samples) // 3)) train_samples = samples[:-validation_window] validation_samples = samples[-validation_window:] if len(train_samples) < 24 or len(validation_samples) < 8: 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) train_x, train_y = _normalize_samples( train_samples, feature_means=feature_means, feature_scales=feature_scales, target_mean=target_mean, target_scale=target_scale, clip=clip, ) validation_x, validation_y = _normalize_samples( validation_samples, feature_means=feature_means, feature_scales=feature_scales, target_mean=target_mean, target_scale=target_scale, 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), 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], feature_names=feature_names, feature_means=feature_means, feature_scales=feature_scales, target_mean=target_mean, target_scale=target_scale, 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]]: columns = [[] for _ in range(input_size)] for window, _target in samples: for row in window: for index in range(input_size): columns[index].append(float(row[index] if index < len(row) else 0.0)) means: list[float] = [] scales: list[float] = [] for values in columns: if not values: means.append(0.0) scales.append(1.0) continue mean = sum(values) / len(values) deviations = sorted(abs(value - mean) for value in values) mad = deviations[len(deviations) // 2] if deviations else 0.0 mean_abs = sum(deviations) / len(deviations) if deviations else 0.0 means.append(mean) scales.append(max(mad, mean_abs * 0.5, 1e-6)) return means, scales def _normalize_samples( samples: list[tuple[list[list[float]], float]], *, feature_means: list[float], feature_scales: list[float], target_mean: float, target_scale: float, clip: float, ) -> tuple[list[list[list[float]]], list[float]]: input_size = len(feature_means) x_values: list[list[list[float]]] = [] y_values: list[float] = [] for window, target in samples: x_values.append( [ [ _clamp( ((row[index] if index < len(row) else 0.0) - feature_means[index]) / max(feature_scales[index], 1e-8), -clip, clip, ) for index in range(input_size) ] for row in window ] ) y_values.append(_clamp((target - target_mean) / max(target_scale, 1e-8), -clip, clip)) return x_values, y_values def _fit_candidate( *, prepared: PreparedData, architecture: str, input_size: int, 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, input_size=input_size, 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_metrics: dict[str, float] = {"validation_mae": math.inf, "directional_accuracy": 0.0, "buy_precision": 0.0} 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() metrics = _validation_metrics(model, prepared, clip) if metrics["validation_mae"] + 1e-12 < best_metrics["validation_mae"]: best_metrics = metrics 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 { **best_metrics, "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_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)] correct = [ 1.0 for prediction, actual in zip(predictions, prepared.validation_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) if prediction != 0 and actual != 0 ] buy_predictions = [ actual for prediction, actual in zip(predictions, prepared.validation_targets) if prediction > 0 ] buy_wins = [actual for actual in buy_predictions if actual > 0] return { "validation_mae": sum(errors) / len(errors) if errors else math.inf, "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, } def _candidate_score(row: dict[str, Any]) -> float: mae = float(row["validation_mae"]) skill = float(row.get("skill", 0.0)) directional = float(row.get("directional_accuracy", 0.0)) buy_precision = float(row.get("buy_precision", 0.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 ) 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()] 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) if __name__ == "__main__": main()