diff --git a/README.md b/README.md index 0425baa..d8536e7 100644 --- a/README.md +++ b/README.md @@ -68,11 +68,15 @@ Dashboard: --limit 1000 ` --architectures lstm,gru ` --lookbacks 32,64 ` - --hidden-sizes 16,32 ` - --layers 1 ` + --hidden-sizes 32,64 ` + --layers 2 ` + --dropouts 0.15 ` + --horizon 3 ` --epochs 60 ``` +Новый artifact версии 3 обучается как multifeature direct-horizon модель: вход `input_size=14` включает доходности, форму свечи, объем, ATR%, RSI, MACD histogram и расстояние до EMA50/EMA200; цель обучается сразу на горизонт `TIME_SERIES_FORECAST_HORIZON`, без умножения one-step прогноза. + Файл из `TIME_SERIES_LSTM_MODEL_PATH` читается ботом автоматически, если `TIME_SERIES_FORECAST_ENABLED=true`. В стратегии `torch_forecast` экспортированная PyTorch LSTM/GRU модель является единственным направляющим сигналом для входа и forecast-выхода. Экспортированные модели появляются в dashboard как `PyTorch LSTM` или `PyTorch GRU`; старый легкий reservoir LSTM-кандидат и все встроенные не-torch прогнозы удалены. Автопереобучение на Windows запускает PyTorch trainer, пишет лог в `runtime/torch_retrain.log` и защищается от параллельных запусков: @@ -84,6 +88,8 @@ powershell -ExecutionPolicy Bypass -File tools\install_windows_torch_retrainer.p По умолчанию Windows-расписание переобучает PyTorch `LSTM/GRU` каждые 6 часов с `--limit 1000` на парах `BTCUSDT,ETHUSDT,SOLUSDT`. Параметры можно переопределить через 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`. +Дополнительно для нового multifeature trainer доступны env-переменные `TORCH_RETRAIN_HORIZON` и `TORCH_RETRAIN_FEATURES`. + ## Docker ```bash diff --git a/crypto_spot_bot/dashboard.py b/crypto_spot_bot/dashboard.py index a9c5009..12a7709 100644 --- a/crypto_spot_bot/dashboard.py +++ b/crypto_spot_bot/dashboard.py @@ -291,9 +291,42 @@ def _time_series_model_artifact(settings: Settings) -> dict[str, Any]: "created_at": data.get("created_at", ""), "symbol_count": len(rows), "models": models, + "feature_count": _artifact_feature_count(data, rows), + "target_horizon": _artifact_target_horizon(data, rows), + "direct_horizon": _artifact_direct_horizon(data, rows), } +def _artifact_feature_count(data: dict[str, Any], rows: list[Any]) -> int: + feature_count = data.get("feature_count") + if isinstance(feature_count, int): + return feature_count + counts = [ + int(row.get("input_size", 0)) + for row in rows + if isinstance(row, dict) and isinstance(row.get("input_size"), int) + ] + return max(counts) if counts else 1 + + +def _artifact_target_horizon(data: dict[str, Any], rows: list[Any]) -> int: + horizon = data.get("target_horizon") + if isinstance(horizon, int): + return horizon + horizons = [ + int(row.get("target_horizon", 0)) + for row in rows + if isinstance(row, dict) and isinstance(row.get("target_horizon"), int) + ] + return max(horizons) if horizons else 0 + + +def _artifact_direct_horizon(data: dict[str, Any], rows: list[Any]) -> bool: + if bool(data.get("direct_horizon")): + return True + return any(isinstance(row, dict) and bool(row.get("direct_horizon")) for row in rows) + + def _forecast_model_label(model: str, *, torch_artifact: bool = False) -> str: normalized = model.strip().lower() if normalized in {"torch_lstm", "lstm"} and torch_artifact: diff --git a/crypto_spot_bot/time_series.py b/crypto_spot_bot/time_series.py index 2af394a..814b9c0 100644 --- a/crypto_spot_bot/time_series.py +++ b/crypto_spot_bot/time_series.py @@ -9,6 +9,24 @@ from crypto_spot_bot.config import Settings from crypto_spot_bot.models import Candle +DEFAULT_TORCH_FEATURES = ( + "return_1", + "return_3", + "return_6", + "range_percent", + "body_percent", + "upper_wick_percent", + "lower_wick_percent", + "volume_change", + "volume_ratio", + "atr_percent", + "rsi_centered", + "macd_hist_percent", + "ema50_gap_percent", + "ema200_gap_percent", +) + + @dataclass(slots=True) class TimeSeriesForecast: enabled: bool @@ -40,31 +58,45 @@ class TimeSeriesForecaster: def forecast(self, candles: list[Candle], symbol: str | None = None) -> TimeSeriesForecast: if not self.settings.time_series_forecast_enabled: - return _empty_forecast(False, "прогноз временных рядов выключен") + return _empty_forecast(False, "time-series forecast is disabled") closes = [float(candle.close) for candle in candles if candle.close > 0] min_candles = max(30, self.settings.time_series_min_candles) if len(closes) < min_candles: - return _empty_forecast(True, "недостаточно свечей для PyTorch прогноза") + return _empty_forecast(True, "not enough candles for PyTorch forecast") returns = _log_returns(closes) if len(returns) < 20: - return _empty_forecast(True, "недостаточно доходностей для PyTorch прогноза") + return _empty_forecast(True, "not enough returns for PyTorch forecast") artifact = self._load_lstm_artifact() - model = _torch_recurrent_model_name(symbol, artifact) - if not model or not _can_use_torch_recurrent(returns, symbol, artifact): - return _empty_forecast(True, "нет валидной PyTorch LSTM/GRU модели для пары") entry = _torch_recurrent_entry(symbol, artifact) - prediction = _torch_recurrent_predict(returns, symbol, artifact) - if entry is None or prediction is None: - return _empty_forecast(True, "PyTorch LSTM/GRU модель не смогла построить прогноз") + model = _torch_recurrent_model_name(symbol, artifact) + feature_rows = _feature_matrix(candles, _feature_names(entry)) 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") - horizon = max(1, self.settings.time_series_forecast_horizon) - expected_return = prediction * horizon + prediction = _torch_recurrent_predict( + returns, + symbol, + artifact, + feature_rows=feature_rows, + closes=closes, + ) + if entry is None or prediction is None: + return _empty_forecast(True, "PyTorch LSTM/GRU model could not build a forecast") + + 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 expected_price = closes[-1] * math.exp(expected_return) model_mae = _torch_validation_mae(entry, returns) baseline_mae = max(_float_entry(entry, "baseline_mae_percent", model_mae * 100) / 100, model_mae) - uncertainty_one_step = max(model_mae, _return_scale(returns) * 0.25, 1e-9) - uncertainty = uncertainty_one_step * math.sqrt(horizon) + if direct_horizon: + uncertainty = max(model_mae, _horizon_return_scale(closes, horizon) * 0.25, 1e-9) + volatility_model = "direct horizon validation MAE" + else: + uncertainty_one_step = max(model_mae, _return_scale(returns) * 0.25, 1e-9) + uncertainty = uncertainty_one_step * math.sqrt(horizon) + volatility_model = "one-step validation MAE scaled by horizon" volatility_percent = uncertainty * 100 expected_return_percent = (math.exp(expected_return) - 1) * 100 probability_up = _normal_cdf(expected_return / max(uncertainty, 1e-9)) @@ -89,7 +121,7 @@ class TimeSeriesForecaster: enabled=True, usable=True, model=model, - volatility_model="torch validation MAE", + volatility_model=volatility_model, expected_return_percent=round(expected_return_percent, 4), expected_price=round(expected_price, 8), volatility_percent=round(volatility_percent, 4), @@ -149,6 +181,60 @@ 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]]: + names = list(feature_names or DEFAULT_TORCH_FEATURES) + rows: list[list[float]] = [] + for index, candle in enumerate(candles): + rows.append([_feature_value(name, candles, index, candle) for name in names]) + return rows + + +def _feature_value(name: str, candles: list[Candle], index: int, candle: Candle) -> float: + close = max(float(candle.close), 1e-12) + previous = candles[index - 1] if index >= 1 else candle + if name == "return_1": + return _log_change(candle.close, previous.close) + if name == "return_3": + 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 == "range_percent": + return _safe_feature((candle.high - candle.low) / close) + if name == "body_percent": + return _safe_feature((candle.close - candle.open) / close) + if name == "upper_wick_percent": + return _safe_feature((candle.high - max(candle.open, candle.close)) / close) + if name == "lower_wick_percent": + return _safe_feature((min(candle.open, candle.close) - candle.low) / close) + if name == "volume_change": + 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 == "atr_percent": + return _safe_feature(candle.atr_14 / close) if candle.atr_14 is not None else 0.0 + 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 == "macd_hist_percent": + return _safe_feature(candle.macd_hist / close) if candle.macd_hist is not None else 0.0 + 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 + return 0.0 + + +def _log_change(current: float, previous: float) -> float: + if current <= 0 or previous <= 0: + return 0.0 + return _safe_feature(math.log(current / previous)) + + +def _safe_feature(value: float) -> float: + if not math.isfinite(value): + return 0.0 + return _clamp(float(value), -50.0, 50.0) + + def _torch_recurrent_model_name(symbol: str | None, artifact: dict[str, Any]) -> str | None: entry = _torch_recurrent_entry(symbol, artifact) if not entry: @@ -175,20 +261,32 @@ def _torch_recurrent_entry(symbol: str | None, artifact: dict[str, Any]) -> dict return entry -def _can_use_torch_recurrent(returns: list[float], symbol: str | None, artifact: dict[str, Any]) -> bool: +def _can_use_torch_recurrent( + returns: list[float], + symbol: str | None, + artifact: dict[str, Any], + feature_rows: list[list[float]] | None = None, +) -> bool: entry = _torch_recurrent_entry(symbol, 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 + if hidden_size <= 0 or num_layers <= 0: + return False + if _is_direct_horizon(entry): + return bool(feature_rows and len(feature_rows) >= lookback) + return len(returns) >= lookback + 1 def _torch_recurrent_predict( returns: list[float], symbol: str | None, artifact: dict[str, Any], + *, + feature_rows: list[list[float]] | None = None, + closes: list[float] | None = None, ) -> float | None: entry = _torch_recurrent_entry(symbol, artifact) model_name = _torch_recurrent_model_name(symbol, artifact) @@ -197,16 +295,28 @@ def _torch_recurrent_predict( 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 None + direct_horizon = _is_direct_horizon(entry) + + if direct_horizon: + rows = feature_rows or [] + if len(rows) < lookback: + return None + sequence = _normalize_feature_rows(rows[-lookback:], entry, clip) + target_mean = _float_entry(entry, "target_mean", 0.0) + target_scale = max(_float_entry(entry, "target_scale", _return_scale(returns)), 1e-8) + else: + mean = _float_entry(entry, "mean", 0.0) + scale = max(_float_entry(entry, "scale", _return_scale(returns)), 1e-8) + if len(returns) < lookback: + return None + sequence = [[_clamp((value - mean) / scale, -clip, clip)] for value in returns[-lookback:]] + target_mean = mean + target_scale = scale - normalized = [_clamp((value - mean) / scale, -clip, clip) for value in returns[-lookback:]] try: hidden = _torch_recurrent_hidden( - normalized, + sequence, entry=entry, model_name=model_name, hidden_size=hidden_size, @@ -221,17 +331,40 @@ def _torch_recurrent_predict( normalized_prediction = sum(weight * value for weight, value in zip(head_weight, hidden)) + head_bias if not math.isfinite(normalized_prediction): return None - prediction = _clamp(normalized_prediction, -clip, clip) * scale + mean + prediction = _clamp(normalized_prediction, -clip, clip) * target_scale + target_mean except (IndexError, KeyError, TypeError, ValueError, OverflowError): return None - 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) + if direct_horizon and closes: + horizon = _entry_horizon(entry, 1) + recent_abs = sorted(abs(value) for value in _horizon_log_returns(closes, horizon)[-48:]) + else: + 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)] if recent_abs else 0.0, 0.0002) return _clamp(prediction, -cap, cap) +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")) + input_size = int(_clamp(_float_entry(entry, "input_size", len(rows[-1]) if rows else 1), 1.0, 256.0)) + if len(means) != input_size: + means = [0.0 for _ in range(input_size)] + if len(scales) != input_size: + scales = [1.0 for _ in range(input_size)] + normalized = [] + for row in rows: + normalized.append( + [ + _clamp(((row[index] if index < len(row) else 0.0) - means[index]) / max(scales[index], 1e-8), -clip, clip) + for index in range(input_size) + ] + ) + return normalized + + def _torch_recurrent_hidden( - normalized: list[float], + sequence: list[list[float]], *, entry: dict[str, Any], model_name: str, @@ -243,8 +376,8 @@ 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)] - for value in normalized: - layer_input = [value] + for row in sequence: + layer_input = list(row) 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) @@ -359,6 +492,23 @@ def _torch_validation_mae(entry: dict[str, Any], returns: list[float]) -> float: return _return_scale(returns) +def _feature_names(entry: dict[str, Any] | None) -> list[str]: + if not entry: + return list(DEFAULT_TORCH_FEATURES) + names = entry.get("feature_names") + if isinstance(names, list) and names: + return [str(name) for name in names] + return list(DEFAULT_TORCH_FEATURES) + + +def _is_direct_horizon(entry: dict[str, Any]) -> bool: + return bool(entry.get("direct_horizon")) or "target_horizon" in entry + + +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)) + + def _float_entry(data: dict[str, Any], key: str, default: float) -> float: value = data.get(key) if isinstance(value, (int, float)): @@ -397,6 +547,22 @@ def _return_scale(returns: list[float]) -> float: return max(max(median, mean * 0.5), 1e-5) +def _horizon_log_returns(closes: list[float], horizon: int) -> list[float]: + horizon = max(1, horizon) + values = [] + for index in range(0, len(closes) - horizon): + current = closes[index] + future = closes[index + horizon] + if current > 0 and future > 0: + values.append(math.log(future / current)) + return values + + +def _horizon_return_scale(closes: list[float], horizon: int) -> float: + values = _horizon_log_returns(closes, horizon) + return _return_scale(values) if values else 0.0005 + + def _sigmoid(value: float) -> float: if value >= 40: return 1.0 @@ -434,8 +600,8 @@ def _reason( block_entry: bool, ) -> str: if block_entry: - return f"модель {model}: ожидаемое движение вниз {expected_return_percent:.3f}%, P(рост)={probability_up:.2f}" - return f"модель {model}: прогноз {expected_return_percent:.3f}%, P(рост)={probability_up:.2f}, skill={skill:.3f}" + return f"model {model}: expected move down {expected_return_percent:.3f}%, P(up)={probability_up:.2f}" + return f"model {model}: forecast {expected_return_percent:.3f}%, P(up)={probability_up:.2f}, skill={skill:.3f}" def _normal_cdf(value: float) -> float: diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py index a30e346..450f28e 100644 --- a/tests/test_dashboard.py +++ b/tests/test_dashboard.py @@ -46,4 +46,7 @@ def test_safe_config_summarizes_torch_forecast_artifact(make_settings, tmp_path) "created_at": "2026-06-20T18:15:05+00:00", "symbol_count": 2, "models": ["PyTorch GRU", "PyTorch LSTM"], + "feature_count": 1, + "target_horizon": 0, + "direct_horizon": False, } diff --git a/tests/test_time_series.py b/tests/test_time_series.py index 49baebc..9ad6a32 100644 --- a/tests/test_time_series.py +++ b/tests/test_time_series.py @@ -77,6 +77,53 @@ def _write_torch_gru_artifact( ) +def _write_multifeature_torch_gru_artifact(path, *, head_bias: float) -> None: + hidden_size = 2 + input_size = 2 + path.write_text( + json.dumps( + { + "version": 3, + "type": "pytorch_recurrent_forecaster", + "target_horizon": 3, + "direct_horizon": True, + "feature_count": input_size, + "feature_names": ["return_1", "range_percent"], + "symbols": { + "BTCUSDT": { + "model": "torch_gru", + "architecture": "gru", + "lookback": 8, + "target_horizon": 3, + "direct_horizon": True, + "input_size": input_size, + "feature_names": ["return_1", "range_percent"], + "feature_means": [0.0, 0.0], + "feature_scales": [0.001, 0.001], + "target_mean": 0.0, + "target_scale": 0.001, + "hidden_size": hidden_size, + "num_layers": 1, + "clip": 8.0, + "validation_mae_percent": 0.01, + "baseline_mae_percent": 0.08, + "skill": 0.2, + "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], + "head_bias": head_bias, + }, + }, + } + ), + encoding="utf-8", + ) + + def test_time_series_forecaster_requires_torch_artifact(make_settings, tmp_path) -> None: settings = make_settings( tmp_path, @@ -166,3 +213,23 @@ def test_time_series_forecaster_reads_torch_gru_artifact(make_settings, tmp_path assert forecast.candidates == [{"model": "torch_gru", "mae_percent": 0.02}] assert forecast.expected_return_percent > 0 assert forecast.probability_up > 0.5 + + +def test_time_series_forecaster_reads_multifeature_direct_horizon_artifact(make_settings, tmp_path) -> None: + artifact_path = tmp_path / "lstm_forecaster.json" + _write_multifeature_torch_gru_artifact(artifact_path, head_bias=0.2) + settings = make_settings( + tmp_path, + time_series_min_candles=80, + time_series_forecast_horizon=3, + 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 forecast.model == "torch_gru" + assert forecast.horizon == 3 + assert 0.015 <= forecast.expected_return_percent <= 0.025 + assert forecast.volatility_model == "direct horizon validation MAE" diff --git a/tools/install_windows_torch_retrainer.ps1 b/tools/install_windows_torch_retrainer.ps1 index ed3b4c0..4a9e2a7 100644 --- a/tools/install_windows_torch_retrainer.ps1 +++ b/tools/install_windows_torch_retrainer.ps1 @@ -4,6 +4,8 @@ param( [int]$EveryHours = 6, [string]$Symbols = "BTCUSDT,ETHUSDT,SOLUSDT", [int]$Limit = 1000, + [int]$Horizon = 0, + [string]$Features = "", [int]$FirstRunMinutes = 0 ) @@ -30,6 +32,12 @@ if ($Symbols) { if ($Limit -gt 0) { $actionArgs += " -Limit $Limit" } +if ($Horizon -gt 0) { + $actionArgs += " -Horizon $Horizon" +} +if ($Features) { + $actionArgs += " -Features `"$Features`"" +} $action = New-ScheduledTaskAction -Execute "powershell.exe" -Argument $actionArgs -WorkingDirectory $RepoRoot $trigger = New-ScheduledTaskTrigger ` -Once ` diff --git a/tools/run_torch_retrain.ps1 b/tools/run_torch_retrain.ps1 index 3baaf0a..64dbf4d 100644 --- a/tools/run_torch_retrain.ps1 +++ b/tools/run_torch_retrain.ps1 @@ -7,6 +7,8 @@ param( [string]$HiddenSizes = "", [string]$Layers = "", [string]$Dropouts = "", + [int]$Horizon = 0, + [string]$Features = "", [int]$Epochs = 0, [int]$Patience = 0, [string]$Interval = "", @@ -53,9 +55,11 @@ if ($Limit -le 0) { } if (-not $Lookbacks) { $Lookbacks = if ($env:TORCH_RETRAIN_LOOKBACKS) { $env:TORCH_RETRAIN_LOOKBACKS } else { "32,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 { "16,32" } } -if (-not $Layers) { $Layers = if ($env:TORCH_RETRAIN_LAYERS) { $env:TORCH_RETRAIN_LAYERS } else { "1" } } -if (-not $Dropouts) { $Dropouts = if ($env:TORCH_RETRAIN_DROPOUTS) { $env:TORCH_RETRAIN_DROPOUTS } else { "0.0" } } +if (-not $HiddenSizes) { $HiddenSizes = if ($env:TORCH_RETRAIN_HIDDEN_SIZES) { $env:TORCH_RETRAIN_HIDDEN_SIZES } else { "32,64" } } +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 $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 $Interval -and $env:TORCH_RETRAIN_INTERVAL) { $Interval = $env:TORCH_RETRAIN_INTERVAL } @@ -89,6 +93,8 @@ try { if ($Symbols) { $trainerArgs += @("--symbols", $Symbols) } if ($Interval) { $trainerArgs += @("--interval", $Interval) } if ($EnvFile) { $trainerArgs += @("--env", $EnvFile) } + if ($Horizon -gt 0) { $trainerArgs += @("--horizon", $Horizon.ToString()) } + if ($Features) { $trainerArgs += @("--features", $Features) } Push-Location $RepoRoot $pushedLocation = $true diff --git a/tools/train_torch_recurrent_forecaster.py b/tools/train_torch_recurrent_forecaster.py index f1c681f..cd73338 100644 --- a/tools/train_torch_recurrent_forecaster.py +++ b/tools/train_torch_recurrent_forecaster.py @@ -25,7 +25,9 @@ except ImportError as exc: # pragma: no cover - exercised on machines without t from crypto_spot_bot.bybit import BybitClient from crypto_spot_bot.config import load_settings -from crypto_spot_bot.time_series import _log_returns +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) @@ -34,9 +36,12 @@ class PreparedData: train_y: torch.Tensor validation_x: torch.Tensor validation_y: torch.Tensor - validation_returns: list[float] - mean: float - scale: float + 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 @@ -46,6 +51,7 @@ class RecurrentReturnModel(nn.Module): self, *, architecture: str, + input_size: int, hidden_size: int, num_layers: int, dropout: float, @@ -53,7 +59,7 @@ class RecurrentReturnModel(nn.Module): super().__init__() recurrent_cls = nn.LSTM if architecture == "lstm" else nn.GRU self.rnn = recurrent_cls( - input_size=1, + input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, dropout=dropout if num_layers > 1 else 0.0, @@ -78,15 +84,21 @@ 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) + feature_names = _feature_names_arg(args.features) artifact: dict[str, Any] = { - "version": 2, + "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": {}, } @@ -98,6 +110,8 @@ def main() -> None: 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), @@ -118,9 +132,11 @@ def main() -> None: artifact["symbols"][symbol] = result print( f"{symbol}: model={result['model']} lookback={result['lookback']} " - f"hidden={result['hidden_size']} layers={result['num_layers']} " + 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}% skill={result['skill']:.4f}" + 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) @@ -136,18 +152,20 @@ def _parse_args() -> argparse.Namespace: 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("--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="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("--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 returns and predictions to this range.") + 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.") @@ -170,6 +188,8 @@ def _train_symbol( interval: str, limit: int, validation_window: int, + target_horizon: int, + feature_names: list[str], architectures: list[str], lookbacks: list[int], hidden_sizes: list[int], @@ -185,26 +205,38 @@ def _train_symbol( 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(returns) < max(100, validation_window + 80): + 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(returns, lookback, validation_window, clip, device) + 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_returns) / len(prepared.validation_returns) + 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, @@ -224,11 +256,19 @@ def _train_symbol( "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, - "mean": prepared.mean, - "scale": prepared.scale, + "dropout": dropout if num_layers > 1 else 0.0, "clip": clip, "validation_mae_percent": validation_mae * 100, "baseline_mae_percent": baseline_mae * 100, @@ -238,7 +278,8 @@ def _train_symbol( "train_samples": prepared.train_samples, "validation_samples": prepared.validation_samples, } - if best is None or validation_mae < float(best["validation_mae"]): + score = _candidate_score(row) + if best is None or score < _candidate_score(best): best = row if best is None: return None @@ -247,54 +288,131 @@ def _train_symbol( def _prepare_data( - returns: list[float], + *, + candles: list[Candle], + feature_names: list[str], lookback: int, + target_horizon: 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: + 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 - 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: + + 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_returns=validation_returns, - mean=mean, - scale=scale, + 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, @@ -310,6 +428,7 @@ def _fit_candidate( _seed(seed) model = RecurrentReturnModel( architecture=architecture, + input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, dropout=dropout, @@ -325,7 +444,7 @@ def _fit_candidate( ) best_state: dict[str, torch.Tensor] | None = None - best_mae = math.inf + 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): @@ -337,9 +456,9 @@ def _fit_candidate( 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 + 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 @@ -351,7 +470,7 @@ def _fit_candidate( if best_state: model.load_state_dict(best_state) return { - "validation_mae": best_mae, + **best_metrics, "best_epoch": best_epoch, "epochs_trained": best_epoch + stale_epochs, "state_dict": _export_recurrent_state(model), @@ -360,15 +479,46 @@ def _fit_candidate( } -def _validation_mae(model: nn.Module, prepared: PreparedData, clip: float) -> float: +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() - 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 + 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]: @@ -430,5 +580,10 @@ 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()