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

Add multifeature direct horizon Torch forecaster

Browse Source
This commit is contained in:
Codex
2026-06-22 07:29:50 +03:00
parent 544b0f4409
commit 42f96f0a39
8 changed files with 537 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crypto_spot_bot/time_series.py
+196 -30
View File
@@ -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: