from __future__ import annotations import argparse import json import math import sys import time 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 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_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_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, *, architecture: str, input_size: int, 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 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.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) 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: 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) 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))) _progress( f"training started: symbols={len(symbols)} interval={interval} " f"limit={args.limit} epochs={args.epochs}" ) artifact: dict[str, Any] = { "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": 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), "symbols": {}, } total_symbols = len(symbols) for index, symbol in enumerate(symbols, start=1): _progress(f"{symbol}: training started ({index}/{total_symbols})") result = _train_symbol( client=client, symbol=symbol, interval=interval, limit=args.limit, validation_window=args.validation_window, 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), 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, attention_pooling=args.attention_pooling, context_norm=args.context_norm, device=device, seed=args.seed, ) if result is None: _progress(f"{symbol}: skipped, not enough candles or train/validation samples") continue artifact["symbols"][symbol] = result _progress( f"{symbol}: model={result['model']} lookback={result['lookback']} " f"features={result['input_size']} hidden={result['hidden_size']} " 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) 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) _progress(f"saved {output}") def _progress(message: str) -> None: print(message, flush=True) 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("--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.") 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("--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.") 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_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], layers_values: list[int], dropouts: list[float], epochs: int, patience: int, batch_size: int, learning_rate: float, weight_decay: float, clip: float, attention_pooling: bool, context_norm: bool, device: torch.device, seed: int, ) -> dict[str, Any] | None: 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) 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: _progress(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: _progress(f"{symbol}: preparing lookback={lookback}") prepared = _prepare_data( candles=candles, feature_names=feature_names, lookback=lookback, 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[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 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 _progress( f"{symbol}: fitting {architecture} " f"lookback={lookback} hidden={hidden_size} " f"layers={num_layers} dropout={dropout}" ) candidate = _fit_candidate( 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, epochs=epochs, patience=patience, batch_size=batch_size, learning_rate=learning_rate, weight_decay=weight_decay, clip=clip, attention_pooling=attention_pooling, context_norm=context_norm, 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": 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_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, "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_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, 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] if current <= 0: continue window = feature_rows[end_index - lookback + 1 : end_index + 1] if len(window) != lookback: continue 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 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)) 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, train_up = _normalize_samples( train_samples, feature_means=feature_means, feature_scales=feature_scales, target_means=target_means, target_scales=target_scales, clip=clip, ) validation_x, validation_y, validation_up = _normalize_samples( validation_samples, feature_means=feature_means, feature_scales=feature_scales, 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_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_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[TrainingSample], input_size: int) -> tuple[list[float], list[float]]: columns = [[] for _ in range(input_size)] 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)) 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 _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[TrainingSample], *, feature_means: list[float], feature_scales: list[float], target_means: list[float], target_scales: list[float], clip: 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[list[float]] = [] up_values: list[list[float]] = [] for sample in samples: window = sample.window 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_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( *, prepared: PreparedData, architecture: str, input_size: int, output_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, attention_pooling: bool, context_norm: bool, 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, 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) generator = torch.Generator(device="cpu").manual_seed(seed) loader = DataLoader( TensorDataset(prepared.train_x, prepared.train_y, prepared.train_up), 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, batch_up in loader: optimizer.zero_grad(set_to_none=True) 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() 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_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(): 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(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(decision_predictions, decision_targets) if prediction != 0 and actual != 0 ] buy_predictions = [ actual 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, } 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)) 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]: return { key: _round_nested(value.detach().cpu().tolist()) for key, value in model.rnn.state_dict().items() } 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": 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 _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] 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 _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) if __name__ == "__main__": main()