Codex
329 lines
13 KiB
Python
329 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from crypto_spot_bot.models import Candle
|
|
from crypto_spot_bot.time_series import TimeSeriesForecaster
|
|
|
|
|
|
def _candles_from_returns(returns: list[float]) -> list[Candle]:
|
|
close = 100.0
|
|
candles = [
|
|
Candle(
|
|
timestamp=0,
|
|
open=close,
|
|
high=close * 1.001,
|
|
low=close * 0.999,
|
|
close=close,
|
|
volume=100,
|
|
)
|
|
]
|
|
for index, ret in enumerate(returns, start=1):
|
|
previous = close
|
|
close = close * (2.718281828459045 ** ret)
|
|
candles.append(
|
|
Candle(
|
|
timestamp=index,
|
|
open=previous,
|
|
high=max(previous, close) * 1.001,
|
|
low=min(previous, close) * 0.999,
|
|
close=close,
|
|
volume=100,
|
|
)
|
|
)
|
|
return candles
|
|
|
|
|
|
def _write_torch_gru_artifact(
|
|
path,
|
|
*,
|
|
head_bias: float,
|
|
validation_mae_percent: float = 0.02,
|
|
baseline_mae_percent: float = 0.08,
|
|
skill: float = 0.2,
|
|
) -> None:
|
|
hidden_size = 2
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"version": 2,
|
|
"type": "pytorch_recurrent_forecaster",
|
|
"symbols": {
|
|
"BTCUSDT": {
|
|
"model": "torch_gru",
|
|
"architecture": "gru",
|
|
"lookback": 8,
|
|
"hidden_size": hidden_size,
|
|
"num_layers": 1,
|
|
"mean": 0.0,
|
|
"scale": 0.001,
|
|
"clip": 8.0,
|
|
"validation_mae_percent": validation_mae_percent,
|
|
"baseline_mae_percent": baseline_mae_percent,
|
|
"skill": skill,
|
|
"state_dict": {
|
|
"weight_ih_l0": [[0.0] for _ in range(3 * hidden_size)],
|
|
"weight_hh_l0": [[0.0, 0.0] for _ in range(3 * hidden_size)],
|
|
"bias_ih_l0": [0.0 for _ in range(3 * hidden_size)],
|
|
"bias_hh_l0": [0.0 for _ in range(3 * hidden_size)],
|
|
},
|
|
"head_weight": [0.0, 0.0],
|
|
"head_bias": head_bias,
|
|
},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
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 _write_probabilistic_torch_gru_artifact(path) -> None:
|
|
hidden_size = 2
|
|
input_size = 2
|
|
output_size = 10
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"version": 4,
|
|
"type": "pytorch_recurrent_forecaster",
|
|
"target_horizon": 3,
|
|
"target_horizons": [1, 3],
|
|
"direct_horizon": True,
|
|
"target_transform": "net_return_over_volatility",
|
|
"round_trip_cost": 0.0026,
|
|
"output_layout": ["mean", "q10", "q50", "q90", "logit_up"],
|
|
"feature_count": input_size,
|
|
"feature_names": ["return_1", "range_percent"],
|
|
"symbols": {
|
|
"BTCUSDT": {
|
|
"model": "torch_gru",
|
|
"architecture": "gru",
|
|
"lookback": 8,
|
|
"target_horizon": 3,
|
|
"target_horizons": [1, 3],
|
|
"direct_horizon": True,
|
|
"target_transform": "net_return_over_volatility",
|
|
"round_trip_cost": 0.0026,
|
|
"output_layout": ["mean", "q10", "q50", "q90", "logit_up"],
|
|
"input_size": input_size,
|
|
"output_size": output_size,
|
|
"feature_names": ["return_1", "range_percent"],
|
|
"feature_means": [0.0, 0.0],
|
|
"feature_scales": [0.001, 0.001],
|
|
"target_means": [0.0, 0.0],
|
|
"target_scales": [1.0, 1.0],
|
|
"target_mean": 0.0,
|
|
"target_scale": 1.0,
|
|
"hidden_size": hidden_size,
|
|
"num_layers": 1,
|
|
"clip": 8.0,
|
|
"validation_mae_percent": 0.01,
|
|
"baseline_mae_percent": 0.08,
|
|
"validation_mae_by_horizon": {"1": 0.001, "3": 0.0015},
|
|
"baseline_mae_by_horizon": {"1": 0.002, "3": 0.003},
|
|
"skill": 0.2,
|
|
"attention_pooling": True,
|
|
"attention_weight": [0.0, 0.0],
|
|
"attention_bias": 0.0,
|
|
"context_norm": True,
|
|
"context_norm_weight": [1.0, 1.0],
|
|
"context_norm_bias": [0.0, 0.0],
|
|
"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] for _ in range(output_size)],
|
|
"head_bias": [0.2, 0.05, 0.15, 0.35, 1.0, 0.35, 0.10, 0.30, 0.55, 2.0],
|
|
},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def test_time_series_forecaster_requires_torch_artifact(make_settings, tmp_path) -> None:
|
|
settings = make_settings(
|
|
tmp_path,
|
|
time_series_min_candles=80,
|
|
time_series_forecast_horizon=3,
|
|
)
|
|
returns = []
|
|
value = 0.0003
|
|
for _ in range(140):
|
|
value = 0.00025 + value * 0.55
|
|
returns.append(value)
|
|
|
|
forecast = TimeSeriesForecaster(settings).forecast(_candles_from_returns(returns), symbol="BTCUSDT")
|
|
|
|
assert forecast.usable is False
|
|
assert forecast.model == "none"
|
|
assert forecast.candidates == []
|
|
assert "PyTorch" in forecast.reason
|
|
|
|
|
|
def test_time_series_forecaster_blocks_negative_torch_edge(make_settings, tmp_path) -> None:
|
|
artifact_path = tmp_path / "lstm_forecaster.json"
|
|
_write_torch_gru_artifact(artifact_path, head_bias=-0.8, validation_mae_percent=0.01)
|
|
settings = make_settings(
|
|
tmp_path,
|
|
time_series_min_candles=80,
|
|
time_series_forecast_horizon=3,
|
|
time_series_min_edge_percent=0.03,
|
|
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.expected_return_percent < 0
|
|
assert forecast.probability_up < 0.45
|
|
assert forecast.block_entry is True
|
|
|
|
|
|
def test_time_series_forecaster_ignores_legacy_lstm_artifact(make_settings, tmp_path) -> None:
|
|
artifact_path = tmp_path / "lstm_forecaster.json"
|
|
artifact_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"version": 1,
|
|
"type": "lstm_reservoir_ridge_params",
|
|
"symbols": {
|
|
"BTCUSDT": {"lookback": 10, "units": 3, "ridge": 0.01},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
settings = make_settings(
|
|
tmp_path,
|
|
time_series_min_candles=80,
|
|
time_series_lstm_enabled=True,
|
|
time_series_lstm_model_path=artifact_path,
|
|
)
|
|
returns = [0.00012 if index % 3 else -0.00008 for index in range(140)]
|
|
|
|
forecast = TimeSeriesForecaster(settings).forecast(_candles_from_returns(returns), symbol="BTCUSDT")
|
|
|
|
assert forecast.usable is False
|
|
assert forecast.model == "none"
|
|
assert forecast.candidates == []
|
|
assert "PyTorch" in forecast.reason
|
|
|
|
|
|
def test_time_series_forecaster_reads_torch_gru_artifact(make_settings, tmp_path) -> None:
|
|
artifact_path = tmp_path / "lstm_forecaster.json"
|
|
_write_torch_gru_artifact(artifact_path, head_bias=0.2)
|
|
settings = make_settings(
|
|
tmp_path,
|
|
time_series_min_candles=80,
|
|
time_series_lstm_enabled=True,
|
|
time_series_lstm_model_path=artifact_path,
|
|
)
|
|
returns = [0.00015 if index % 4 else -0.00005 for index in range(140)]
|
|
|
|
forecast = TimeSeriesForecaster(settings).forecast(_candles_from_returns(returns), symbol="BTCUSDT")
|
|
|
|
assert forecast.usable is True
|
|
assert forecast.model == "torch_gru"
|
|
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"
|
|
|
|
|
|
def test_time_series_forecaster_reads_probabilistic_multi_horizon_artifact(make_settings, tmp_path) -> None:
|
|
artifact_path = tmp_path / "lstm_forecaster.json"
|
|
_write_probabilistic_torch_gru_artifact(artifact_path)
|
|
settings = make_settings(
|
|
tmp_path,
|
|
time_series_min_candles=80,
|
|
time_series_forecast_horizon=3,
|
|
time_series_lstm_model_path=artifact_path,
|
|
)
|
|
returns = [0.0002 if index % 5 else -0.00007 for index in range(160)]
|
|
|
|
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 forecast.target_transform == "net_return_over_volatility"
|
|
assert forecast.probability_up > 0.85
|
|
assert forecast.quantile_10_percent <= forecast.quantile_50_percent <= forecast.quantile_90_percent
|
|
assert sorted(forecast.horizon_forecasts) == ["1", "3"]
|
|
assert [item["name"] for item in forecast.feature_snapshot] == ["return_1", "range_percent"]
|
|
assert forecast.feature_snapshot[0]["label"] == "Доходность 1ч"
|
|
assert forecast.feature_snapshot[0]["raw_display"].endswith("%")
|
|
assert "диапазон" in forecast.feature_snapshot[0]["interpretation"]
|