From 7186acb9a1f9b85697f61c8b76c3b98b646266b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9A=D1=83=D1=80=D0=BD=D0=B0=D1=82=20=D0=90=D0=BD=D0=B4?= =?UTF-8?q?=D1=80=D0=B5=D0=B9?= Date: Tue, 14 Jul 2026 07:49:32 +0300 Subject: [PATCH] feat: train forecasts on trade outcomes --- README.md | 8 +- android/TradeBotMonitor/README.md | 2 +- android/TradeBotMonitor/app/build.gradle.kts | 4 +- .../kusoft/tradebotmonitor/MainActivity.kt | 56 +++- .../java/xyz/kusoft/tradebotmonitor/Models.kt | 11 +- .../xyz/kusoft/tradebotmonitor/TradeBotApi.kt | 5 +- crypto_spot_bot/strategy.py | 21 +- crypto_spot_bot/time_series.py | 147 +++++++++- crypto_spot_bot/training_coordination.py | 51 +++- tests/test_calibrate_thresholds.py | 47 ++- tests/test_retrain_runner_contract.py | 23 ++ tests/test_time_series.py | 94 ++++++ tests/test_trade_objective_training.py | 86 ++++++ tests/test_training_coordination.py | 36 ++- tools/calibrate_torch_thresholds.py | 91 +++++- tools/run_torch_retrain.ps1 | 4 + tools/train_torch_recurrent_forecaster.py | 267 +++++++++++++++--- tools/windows_training_agent.py | 8 + 18 files changed, 867 insertions(+), 94 deletions(-) create mode 100644 tests/test_retrain_runner_contract.py create mode 100644 tests/test_trade_objective_training.py diff --git a/README.md b/README.md index 8d9e5f8..1730154 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,7 @@ Dashboard: --epochs 70 ``` -Новый artifact версии 4 обучается как probabilistic multi-horizon модель: вход включает доходности, форму свечи, объем, ATR%, realized volatility, RSI/MACD/EMA slopes, 4h/24h rolling trend, дневные EMA-признаки, BTC/ETH cross-asset признаки и числовые признаки текущего шаблона пары. Цель обучается как `future log return - комиссии - проскальзывание`, нормализованная на текущую волатильность. Модель сразу прогнозирует горизонты `1/3/6/12`, quantile-оценки `q10/q50/q90` и `P(up)`. +Новый artifact версии 6 обучается как торговая multi-task multi-horizon модель: вход включает доходности, форму свечи, объем, ATR%, realized volatility, RSI/MACD/EMA slopes, 4h/24h rolling trend, дневные EMA-признаки, BTC/ETH cross-asset признаки и числовые признаки текущего шаблона пары. Для каждой точки симулируется вход по open следующей свечи; затем до каждого горизонта проверяется, что было достигнуто раньше — take-profit или stop-loss. Денежная цель равна чистому log-PnL при первом барьере либо закрытии по горизонту после комиссий и проскальзывания. Вторая цель — вероятность `P(TP before SL)`. Если одна OHLC-свеча касается обоих барьеров, разметка консервативно считает stop-loss первым. Модель прогнозирует горизонты `3/6/12/24` и quantile-оценки `q10/q50/q90` чистого результата. Последний tail (`--holdout-window`, по умолчанию 1000 samples на символ) полностью исключается из training и early stopping. Между train/validation/holdout оставляется purge по максимальному forecast horizon. Threshold walk-forward и guard работают только на этом untouched holdout; calibration и guard криптографически привязаны к SHA-256 конкретного model artifact. В каждом walk-forward fold торговать могут только пары, которые получили жизнеспособный порог на предшествующей train-части; общий порог больше не возвращает в портфель нестабильные пары. @@ -103,9 +103,9 @@ powershell -ExecutionPolicy Bypass -File tools\install_windows_training_agent.ps По умолчанию Windows-agent обучает отдельную PyTorch `LSTM/GRU` для каждой пары на `6000` часовых свечах. Это не заставляет разнородные активы делить одну архитектуру и один набор recurrent-весов. Прогноз усредняется по seed `7/19`, модели сравниваются на validation-folds, а пороги калибруются отдельно для каждой пары. Ensemble guard выполняется пакетно на GPU, а экспорт не дублирует первый набор весов. Search space использует lookback `32/64/128`, hidden `64/96`, dropout `0.20`, AdamW learning rate `0.0007` и weight decay `0.0005`; untouched holdout и quality gate не ослабляются. Для диагностического pooled-запуска используется ключ `-Pooled`. Параметры можно переопределить через 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_HORIZON`, `TORCH_RETRAIN_HORIZONS`, `TORCH_RETRAIN_CONTEXT_SYMBOLS`, `TORCH_RETRAIN_FEATURES`, `TORCH_RETRAIN_SEED`, `TORCH_RETRAIN_ENSEMBLE_SEEDS`, `TORCH_RETRAIN_SELECTION_FOLDS`, `TORCH_RETRAIN_LEARNING_RATE`, `TORCH_RETRAIN_WEIGHT_DECAY`, `TORCH_RETRAIN_EPOCHS`, `TORCH_RETRAIN_PATIENCE`, `TORCH_RETRAIN_INTERVAL`, `TORCH_RETRAIN_ENV`. -Loss и выбор гиперпараметров учитывают after-cost trading utility и ранговую связь прогноза с будущей доходностью, а не только MAE. В каждом walk-forward fold вероятность `P(up)` калибруется Platt-моделью исключительно на train-части; затем на этой же train-части выбираются глобальные и per-symbol пороги, которые применяются к test-части. Для выбора порога требуется минимум 24 непересекающиеся сделки, а финальный quality gate по-прежнему требует не менее 30 OOS-сделок. Калибратор не имеет fallback на единичные сделки: если минимальная статистика не набрана, кандидат получает `calibration_insufficient` и не может пройти gate. +Loss и выбор гиперпараметров учитывают after-cost trading utility, ошибку ожидаемого чистого PnL, quantile-loss и focal BCE для события `TP before SL`, а не только MAE направления цены. В каждом walk-forward fold вероятность успеха калибруется Platt-моделью исключительно на train-части; затем на этой же train-части выбираются глобальные и per-symbol пороги, которые применяются к test-части. Для выбора порога требуется минимум 24 непересекающиеся сделки, а финальный quality gate по-прежнему требует не менее 30 OOS-сделок. Калибратор не имеет fallback на единичные сделки: если минимальная статистика не набрана, кандидат получает `calibration_insufficient` и не может пройти gate. -Основной decision horizon — `12h`, дополнительные горизонты — `3/6/12/24`. Это согласует прогноз с round-trip cost: при текущих fee/slippage полный вход-выход стоит около `0.26%`, поэтому прежний `3h` target чаще описывал шум, который не покрывал издержки. Threshold search оценивается тем же execution replay со stop-loss, take-profit, ATR trailing и forecast-exit, который используется в walk-forward. `holdout_skill` остаётся только в финальном отчёте и никогда не участвует в фильтрации входов или подборе порогов. +Основной decision horizon — `12h`, дополнительные горизонты — `3/6/12/24`. Размеры обучающих барьеров берутся из `STOP_LOSS_PERCENT` и `TAKE_PROFIT_PERCENT`, а round-trip cost — из fee/slippage настроек. Threshold search оценивается тем же execution replay со stop-loss, take-profit, ATR trailing и forecast-exit, который используется в walk-forward. `holdout_skill` остаётся только в финальном отчёте и никогда не участвует в фильтрации входов или подборе порогов. Если retrain запускается с `-DeployToPi`, после успешного guard он синхронизирует `runtime/lstm_forecaster.json`, `runtime/torch_retrain_guard.json` и `runtime/torch_threshold_calibration.json` на Raspberry Pi через SSH-ключ и перезапускает сервис `tradebot`. Отдельный запуск sync: @@ -113,7 +113,7 @@ Loss и выбор гиперпараметров учитывают after-cost powershell -ExecutionPolicy Bypass -File tools\sync_torch_artifacts_to_pi.ps1 -RemoteHost 192.168.0.185 -RemoteUser sevenhill -RemoteRoot /mnt/data/tradebot ``` -Внутри recurrent модели используются exportable attention pooling и LayerNorm перед forecast-head; Raspberry Pi по-прежнему исполняет модель из JSON без PyTorch runtime. +Внутри recurrent модели используются exportable attention pooling и LayerNorm. После recurrent-контекста добавлена нелинейная GELU-проекция и две отдельные экспортируемые головы: одна для ожидаемого PnL/quantiles, вторая для `P(TP before SL)`. Raspberry Pi по-прежнему исполняет модель из JSON без PyTorch runtime. ## Docker diff --git a/android/TradeBotMonitor/README.md b/android/TradeBotMonitor/README.md index 1d82a97..67506b0 100644 --- a/android/TradeBotMonitor/README.md +++ b/android/TradeBotMonitor/README.md @@ -7,7 +7,7 @@ - Русский интерфейс без bubble/pill-оформления. - Современная биржевая компоновка: список пар, один выбранный график, компактные параметры ниже. - Свечной график 1h: тела свечей, фитили, объём, EMA50, EMA200, последняя цена. -- Параметры Torch: edge, P(up), confidence, skill, quantiles, gate, причина решения. +- Параметры Torch: ожидаемый чистый edge, P(TP + "${signedPercent(replayTotalPercent)} (средняя ${signedPercent(replayAveragePercent)})" + replayTotalPercent != null -> signedPercent(replayTotalPercent) + else -> signedMoney(replay.optDouble("net_pnl", 0.0)) + } + val replayPnlValue = replayTotalPercent ?: replay.optDouble("net_pnl", 0.0) + addView(keyValueLine("Replay P&L", replayPnl, colorForSigned(replayPnlValue)).top(dp(4))) if (isTrainingActive) { addView(disabledActionButton(if (retrainRequestInFlight) "Отправляю команду" else "Обучение уже идет").top(dp(10))) } else { @@ -1163,11 +1178,26 @@ class MainActivity : Activity() { orientation = LinearLayout.VERTICAL addView(keyValueLine("Последний gate", status, if (accepted) palette.green else palette.amber)) addView(text(explanation, 12f, Typeface.NORMAL, palette.text).top(dp(7))) - if (!candidateScore.isNaN() && !currentScore.isNaN()) { - addView(keyValueLine("Score candidate / текущая", "${number(candidateScore, 2)} / ${number(currentScore, 2)}", if (accepted) palette.green else palette.amber).top(dp(7))) + when { + !candidateScore.isNaN() && !currentScore.isNaN() -> + addView(keyValueLine("SCORE кандидата / действующей", "${number(candidateScore, 3)} / ${number(currentScore, 3)}", if (accepted) palette.green else palette.amber).top(dp(7))) + !candidateScore.isNaN() -> + addView(keyValueLine("SCORE кандидата", number(candidateScore, 3), if (accepted) palette.green else palette.amber).top(dp(7))) + !currentScore.isNaN() -> + addView(keyValueLine("SCORE действующей", number(currentScore, 3), palette.muted).top(dp(7))) } } + private fun latestTrainingEvaluation(retrain: JSONObject): JSONObject? { + val latestJob = retrain + .optJSONObject("coordination") + ?.optJSONObject("latest_job") + ?: return null + if (latestJob.optStringClean("status") !in setOf("completed", "failed")) return null + val summary = latestJob.optJSONObject("summary") ?: return null + return summary.takeIf { it.optJSONObject("candidate") != null } + } + private fun liveSettingsBlock(data: BotSnapshot?): View { val config = data?.config ?: JSONObject() return LinearLayout(this).apply { @@ -1812,6 +1842,12 @@ class MainActivity : Activity() { private fun probability(value: Double): String = percent(normalizedProbability(value), 0) + private fun probabilityMetricLabel(signal: SignalData?, forecast: ForecastData?): String = + if ( + signal?.targetTransform == "barrier_net_return" || + forecast?.targetTransform == "barrier_net_return" + ) "P(TP int: def _torch_forecast_confidence(settings: Settings, forecast: dict) -> float: expected_return = max(0.0, _safe_float(forecast.get("expected_return_percent"), 0.0)) - probability_up = _safe_float(forecast.get("probability_up"), 0.5) + probability_up = _forecast_probability(forecast) skill = max(0.0, _safe_float(forecast.get("skill"), 0.0)) min_edge = max(0.01, settings.time_series_min_edge_percent) edge_strength = _clamp(expected_return / max(min_edge * 4.0, 0.01), 0.0, 1.0) @@ -1168,7 +1168,7 @@ def _torch_forecast_position_sizing( symbol=symbol, ) expected_return = max(0.0, _safe_float(forecast.get("expected_return_percent"), 0.0)) - probability_up = _safe_float(forecast.get("probability_up"), 0.5) + probability_up = _forecast_probability(forecast) skill = max(0.0, _safe_float(forecast.get("skill"), 0.0)) min_edge = max(0.01, settings.time_series_min_edge_percent) edge_multiplier = _clamp(expected_return / max(min_edge * 3.0, 0.01), 0.25, 1.15) @@ -1323,7 +1323,7 @@ def _position_risk_multiplier(forecast: dict | None, adaptive: dict | None) -> f multiplier = 1.0 forecast = forecast or {} if forecast.get("usable"): - probability_up = _safe_float(forecast.get("probability_up"), 0.5) + probability_up = _forecast_probability(forecast) volatility_percent = _safe_float(forecast.get("volatility_percent"), 0.0) if probability_up < 0.52: multiplier *= 0.75 @@ -1360,7 +1360,7 @@ def _kelly_position( probability_source = "confidence" probability = confidence_probability if forecast.get("usable"): - probability = _safe_float(forecast.get("probability_up"), confidence_probability) + probability = _forecast_probability(forecast, confidence_probability) probability_source = "forecast" probability = _clamp(probability, 0.0, 1.0) @@ -1661,6 +1661,13 @@ def _rebound_state( } +def _forecast_probability(forecast: dict, default: float = 0.5) -> float: + value = forecast.get("probability_take_profit_first") + if not isinstance(value, (int, float, str)): + value = forecast.get("probability_up") + return _clamp(_safe_float(value, default), 0.0, 1.0) + + def _safe_float(value: object, default: float = 0.0) -> float: try: return float(value) @@ -1783,7 +1790,7 @@ def _forecast_exit_signal( return None skill = _safe_float(forecast.get("skill"), 0.0) expected_return = _safe_float(forecast.get("expected_return_percent"), 0.0) - probability_up = _safe_float(forecast.get("probability_up"), 0.5) + probability_up = _forecast_probability(forecast) min_edge = max(0.0, min_edge_percent) strong_negative = skill > 0.02 and expected_return <= -max(min_edge, 0.03) and probability_up <= 0.44 if not strong_negative: diff --git a/crypto_spot_bot/time_series.py b/crypto_spot_bot/time_series.py index 07e32a5..9a3bcdb 100644 --- a/crypto_spot_bot/time_series.py +++ b/crypto_spot_bot/time_series.py @@ -163,6 +163,7 @@ class TimeSeriesForecast: calibrated_min_edge_percent: float = 0.0 calibrated_min_probability_up: float = 0.0 calibrated_min_confidence: float = 0.0 + probability_take_profit_first: float | None = None def as_dict(self) -> dict[str, Any]: return asdict(self) @@ -254,6 +255,7 @@ class TimeSeriesForecaster: expected_gross_return = float(selected.get("expected_gross_return", expected_return)) expected_price = closes[-1] * math.exp(expected_gross_return) probability_up = _clamp(float(selected.get("probability_up", 0.5)), 0.0, 1.0) + target_transform = str(entry.get("target_transform", "net_return_over_volatility")) model_mae = max(float(selected.get("validation_mae", 0.0)), 1e-9) baseline_mae = max(float(selected.get("baseline_mae", model_mae)), model_mae) uncertainty = max(float(selected.get("uncertainty", model_mae)), 1e-9) @@ -279,12 +281,16 @@ class TimeSeriesForecaster: or (expected_return_percent <= -min_edge and probability_up <= 0.45) or (q50_percent <= -min_edge and probability_up <= 0.48) ) - reason = _reason( - model=model, - expected_return_percent=expected_return_percent, - probability_up=probability_up, - skill=skill, - block_entry=block_entry, + reason = ( + _barrier_reason(model, expected_return_percent, probability_up, skill, block_entry) + if target_transform == "barrier_net_return" + else _reason( + model=model, + expected_return_percent=expected_return_percent, + probability_up=probability_up, + skill=skill, + block_entry=block_entry, + ) ) if not symbol_eligible: reason = "symbol excluded by train-only calibration" @@ -292,7 +298,11 @@ class TimeSeriesForecaster: enabled=True, usable=True, model=model, - volatility_model="probabilistic multi-horizon after-cost quantile", + volatility_model=( + "TP-before-SL multi-task after-cost model" + if target_transform == "barrier_net_return" + else "probabilistic multi-horizon after-cost quantile" + ), expected_return_percent=round(expected_return_percent, 4), expected_price=round(expected_price, 8), volatility_percent=round(volatility_percent, 4), @@ -309,7 +319,7 @@ class TimeSeriesForecaster: quantile_50_percent=round(q50_percent, 4), quantile_90_percent=round(q90_percent, 4), conservative_return_percent=round(conservative_return_percent, 4), - target_transform=str(entry.get("target_transform", "net_return_over_volatility")), + target_transform=target_transform, feature_snapshot=feature_snapshot, horizon_forecasts=_public_horizon_forecasts(prediction), candidates=[{"model": model, "mae_percent": round(model_mae * 100, 4)}], @@ -321,6 +331,11 @@ class TimeSeriesForecaster: calibrated_min_edge_percent=calibrated["edge"], calibrated_min_probability_up=calibrated["probability"], calibrated_min_confidence=calibrated["confidence"], + probability_take_profit_first=( + round(probability_up, 4) + if target_transform == "barrier_net_return" + else None + ), ) direct_horizon = _is_direct_horizon(entry) @@ -1172,6 +1187,47 @@ def _average_ensemble_predictions(predictions: list[float | dict[str, Any]]) -> def _torch_head_outputs(context: list[float], entry: dict[str, Any], hidden_size: int) -> list[float]: context = _apply_context_norm(context, entry) + if entry.get("multitask_head") is True: + hidden_matrix = _float_matrix(entry.get("head_hidden_weight")) + hidden_bias = _float_vector(entry.get("head_hidden_bias")) + if not hidden_matrix or len(hidden_bias) != len(hidden_matrix): + return [] + shared = [ + _gelu(_dot(row, context) + hidden_bias[index]) + for index, row in enumerate(hidden_matrix) + if len(row) == hidden_size + ] + if len(shared) != len(hidden_matrix): + return [] + return_matrix = _float_matrix(entry.get("return_head_weight")) + return_bias = _float_vector(entry.get("return_head_bias")) + event_matrix = _float_matrix(entry.get("event_head_weight")) + event_bias = _float_vector(entry.get("event_head_bias")) + if ( + not return_matrix + or len(return_bias) != len(return_matrix) + or not event_matrix + or len(event_bias) != len(event_matrix) + ): + return [] + return_values = [ + _dot(row, shared) + return_bias[index] + for index, row in enumerate(return_matrix) + if len(row) == len(shared) + ] + event_values = [ + _dot(row, shared) + event_bias[index] + for index, row in enumerate(event_matrix) + if len(row) == len(shared) + ] + if len(return_values) != len(event_values) * 4: + return [] + outputs: list[float] = [] + for horizon_index, event_value in enumerate(event_values): + base = horizon_index * 4 + outputs.extend(return_values[base : base + 4]) + outputs.append(event_value) + return outputs raw_weight = entry.get("head_weight") if isinstance(raw_weight, list) and raw_weight and isinstance(raw_weight[0], list): matrix = _float_matrix(raw_weight) @@ -1242,8 +1298,18 @@ def _decode_multi_horizon_prediction( expected = decode("mean") q_values = sorted([decode("q10", expected), decode("q50", expected), decode("q90", expected)]) - probability_up = _sigmoid(float(values.get("logit_up", 0.0))) + probability_up = _sigmoid( + float(values.get("logit_tp_first", values.get("logit_up", 0.0))) + ) cap = _prediction_cap(closes, horizon, round_trip_cost) + if str(entry.get("target_transform", "")) == "barrier_net_return": + stop_percent = _clamp(_float_entry(entry, "target_stop_loss_percent", 0.04), 0.003, 0.08) + take_percent = _clamp(_float_entry(entry, "target_take_profit_percent", 0.035), 0.003, 0.20) + cap = max( + cap, + abs(math.log(1.0 - stop_percent) - round_trip_cost), + abs(math.log(1.0 + take_percent) - round_trip_cost), + ) expected = _clamp(expected, -cap, cap) q10 = _clamp(q_values[0], -cap, cap) q50 = _clamp(q_values[1], -cap, cap) @@ -1267,6 +1333,11 @@ def _decode_multi_horizon_prediction( "q50": q50, "q90": q90, "probability_up": probability_up, + "probability_take_profit_first": ( + probability_up + if str(entry.get("target_transform", "")) == "barrier_net_return" + else None + ), "volatility_scale": vol_scale, "validation_mae": mae, "baseline_mae": base_mae, @@ -1697,6 +1768,10 @@ def _public_horizon_forecasts(prediction: dict[str, Any]) -> dict[str, Any]: "quantile_50_percent": round((math.exp(float(row.get("q50", 0.0))) - 1) * 100, 4), "quantile_90_percent": round((math.exp(float(row.get("q90", 0.0))) - 1) * 100, 4), } + if isinstance(row.get("probability_take_profit_first"), (int, float)): + public[key]["probability_take_profit_first"] = round( + _clamp(float(row["probability_take_profit_first"]), 0.0, 1.0), 4 + ) return public @@ -1728,6 +1803,10 @@ def _dot(left: list[float], right: list[float]) -> float: return sum(left[index] * right[index] for index in range(min(len(left), len(right)))) +def _gelu(value: float) -> float: + return 0.5 * value * (1.0 + math.erf(value / math.sqrt(2.0))) + + def _return_scale(returns: list[float]) -> float: recent = returns[-120:] if len(returns) > 120 else returns values = sorted(abs(value) for value in recent if math.isfinite(value)) @@ -1772,6 +1851,41 @@ def _prediction_cap(closes: list[float], horizon: int, round_trip_cost: float) - return max(base * 1.5 + round_trip_cost, 0.0005) +def _barrier_outcome( + candles: list[Candle], + *, + end_index: int, + horizon: int, + stop_loss_percent: float, + take_profit_percent: float, + round_trip_cost: float, +) -> tuple[float, float] | None: + """Return after-cost log PnL and whether TP was reached before SL.""" + entry_index = end_index + 1 + exit_index = end_index + max(1, horizon) + if entry_index >= len(candles) or exit_index >= len(candles): + return None + entry = float(candles[entry_index].open) + if entry <= 0: + return None + stop = entry * (1.0 - _clamp(stop_loss_percent, 0.003, 0.08)) + take = entry * (1.0 + _clamp(take_profit_percent, 0.003, 0.20)) + for index in range(entry_index, exit_index + 1): + candle = candles[index] + stop_hit = float(candle.low) <= stop + take_hit = float(candle.high) >= take + # OHLC data cannot reveal intrabar ordering, so ties are resolved + # conservatively as stop-loss first. + if stop_hit: + return math.log(stop / entry) - round_trip_cost, 0.0 + if take_hit: + return math.log(take / entry) - round_trip_cost, 1.0 + terminal = float(candles[exit_index].close) + if terminal <= 0: + return None + return math.log(terminal / entry) - round_trip_cost, 0.0 + + def _sigmoid(value: float) -> float: if value >= 40: return 1.0 @@ -1813,6 +1927,21 @@ def _reason( return f"model {model}: forecast {expected_return_percent:.3f}%, P(up)={probability_up:.2f}, skill={skill:.3f}" +def _barrier_reason( + model: str, + expected_return_percent: float, + probability_take_profit_first: float, + skill: float, + block_entry: bool, +) -> str: + state = "entry blocked" if block_entry else "entry evaluated" + return ( + f"model {model}: expected net {expected_return_percent:.3f}%, " + f"P(TP before SL)={probability_take_profit_first:.2f}, " + f"skill={skill:.3f}; {state}" + ) + + def _normal_cdf(value: float) -> float: return 0.5 * (1 + math.erf(value / math.sqrt(2))) diff --git a/crypto_spot_bot/training_coordination.py b/crypto_spot_bot/training_coordination.py index 3c88724..9c487c4 100644 --- a/crypto_spot_bot/training_coordination.py +++ b/crypto_spot_bot/training_coordination.py @@ -379,11 +379,19 @@ def _safe_parameters(value: Any) -> dict[str, Any]: "layers", "dropouts", "epochs", + "validation_window", "holdout_window", "ensemble_seeds", "selection_folds", "learning_rate", "weight_decay", + "horizon", + "horizons", + "patience", + "context_symbols", + "features", + "seed", + "interval", "pooled", "resume_candidate", } @@ -391,8 +399,12 @@ def _safe_parameters(value: Any) -> dict[str, Any]: for key, low, high in ( ("limit", 500, 20000), ("epochs", 1, 200), + ("validation_window", 64, 2000), ("holdout_window", 64, 1000), ("selection_folds", 1, 12), + ("horizon", 1, 96), + ("patience", 1, 50), + ("seed", 1, 2_147_483_647), ): if key not in result: continue @@ -414,9 +426,19 @@ def _safe_parameters(value: Any) -> dict[str, Any]: if item.strip().lower() in {"lstm", "gru"} ] result["architectures"] = ",".join(architectures) or "lstm,gru" - for key in ("lookbacks", "hidden_sizes", "layers", "dropouts", "ensemble_seeds"): + for key in ( + "lookbacks", + "hidden_sizes", + "layers", + "dropouts", + "ensemble_seeds", + "horizons", + "context_symbols", + "features", + "interval", + ): if key in result: - result[key] = str(result[key])[:200] + result[key] = str(result[key])[: 4000 if key == "features" else 500] for key, low, high in ( ("learning_rate", 0.00001, 0.1), ("weight_decay", 0.0, 0.1), @@ -471,10 +493,27 @@ def _validate_symbol_models(symbols: dict[str, Any]) -> None: raise ValueError(f"candidate model dimensions are invalid: {symbol}") from exc if not 4 <= lookback <= 512 or not 1 <= input_size <= 256 or not 1 <= hidden_size <= 1024: raise ValueError(f"candidate model dimensions are out of range: {symbol}") - if not isinstance(entry.get("state_dict"), dict): - raise ValueError(f"candidate recurrent state is missing: {symbol}") - if not isinstance(entry.get("head_weight"), list) or not isinstance(entry.get("head_bias"), list): - raise ValueError(f"candidate forecast head is missing: {symbol}") + members = entry.get("ensemble_members") + payloads = members if isinstance(members, list) and members else [entry] + for payload in payloads: + if not isinstance(payload, dict) or not isinstance(payload.get("state_dict"), dict): + raise ValueError(f"candidate recurrent state is missing: {symbol}") + merged = {**entry, **payload} + if merged.get("multitask_head") is True: + required = ( + "head_hidden_weight", + "head_hidden_bias", + "return_head_weight", + "return_head_bias", + "event_head_weight", + "event_head_bias", + ) + if any(not isinstance(merged.get(name), list) for name in required): + raise ValueError(f"candidate multitask forecast head is missing: {symbol}") + elif not isinstance(merged.get("head_weight"), list) or not isinstance( + merged.get("head_bias"), list + ): + raise ValueError(f"candidate forecast head is missing: {symbol}") def _compact_now() -> str: diff --git a/tests/test_calibrate_thresholds.py b/tests/test_calibrate_thresholds.py index 5623c8b..d720478 100644 --- a/tests/test_calibrate_thresholds.py +++ b/tests/test_calibrate_thresholds.py @@ -7,12 +7,20 @@ from tools.calibrate_torch_thresholds import ( ForecastRecord, _average_selected_predictions, _apply_platt_calibration, + _build_torch_model, _choose_recommendation, _full_backtest, _fit_platt_calibration, + _record_event_target, _entry_validation_skill, ) -from tools.train_torch_recurrent_forecaster import _ensemble_candidate +from tools.train_torch_recurrent_forecaster import ( + OUTPUT_LAYOUT, + RecurrentReturnModel, + _ensemble_candidate, + _export_head_state, + _export_recurrent_state, +) def _result(*, trades: int, average: float, total: float, profit_factor: float) -> CalibrationResult: @@ -85,6 +93,13 @@ def test_platt_calibration_learns_probability_direction_from_train_records() -> assert calibrated[0].probability_up < calibrated[1].probability_up +def test_barrier_event_target_takes_precedence_over_terminal_profit() -> None: + record = _record(1, 0.8, 3.0) + record.take_profit_first = False + + assert _record_event_target(record) == 0.0 + + def test_entry_quality_never_falls_back_to_holdout_skill() -> None: entry = {"validation_skill": 0.12, "skill": 0.99, "holdout_skill": 0.99} @@ -107,6 +122,36 @@ def test_batched_ensemble_averages_decoded_predictions() -> None: } +def test_calibrator_loads_multitask_head() -> None: + model = RecurrentReturnModel( + architecture="gru", + input_size=2, + hidden_size=4, + num_layers=1, + dropout=0.0, + output_size=len(OUTPUT_LAYOUT), + attention_pooling=False, + context_norm=False, + multitask_head=True, + head_hidden_size=6, + ) + entry = { + "input_size": 2, + "hidden_size": 4, + "num_layers": 1, + "output_size": len(OUTPUT_LAYOUT), + "multitask_head": True, + "head_hidden_size": 6, + "state_dict": _export_recurrent_state(model), + **_export_head_state(model), + } + + loaded = _build_torch_model(entry, "torch_gru") + + assert loaded is not None + assert loaded.multitask_head is True + + def test_multi_seed_export_does_not_duplicate_first_member_weights() -> None: members = [ { diff --git a/tests/test_retrain_runner_contract.py b/tests/test_retrain_runner_contract.py new file mode 100644 index 0000000..ad5460d --- /dev/null +++ b/tests/test_retrain_runner_contract.py @@ -0,0 +1,23 @@ +from pathlib import Path + + +def test_retrain_runner_passes_training_horizon_to_calibrator() -> None: + runner = ( + Path(__file__).resolve().parents[1] / "tools" / "run_torch_retrain.ps1" + ).read_text(encoding="utf-8") + + calibration_start = runner.index("$calibrationBaseArgs = @(") + calibration_end = runner.index("\n )", calibration_start) + calibration_args = runner[calibration_start:calibration_end] + + assert '"--horizon", $Horizon.ToString()' in calibration_args + + +def test_retrain_runner_uses_a_regime_sized_validation_window() -> None: + runner = ( + Path(__file__).resolve().parents[1] / "tools" / "run_torch_retrain.ps1" + ).read_text(encoding="utf-8") + + assert "[int]$ValidationWindow = 0" in runner + assert "else { 720 }" in runner + assert '"--validation-window", $ValidationWindow.ToString()' in runner diff --git a/tests/test_time_series.py b/tests/test_time_series.py index 116d041..ef9214c 100644 --- a/tests/test_time_series.py +++ b/tests/test_time_series.py @@ -2,6 +2,8 @@ from __future__ import annotations import json +import pytest + from crypto_spot_bot.models import Candle from crypto_spot_bot.time_series import TimeSeriesForecaster @@ -191,6 +193,77 @@ def _write_probabilistic_torch_gru_artifact(path) -> None: ) +def _write_barrier_multitask_gru_artifact(path) -> None: + hidden_size = 2 + head_hidden_size = 2 + input_size = 2 + output_size = 5 + path.write_text( + json.dumps( + { + "version": 6, + "type": "pytorch_recurrent_forecaster", + "target_horizon": 3, + "target_horizons": [3], + "direct_horizon": True, + "target_transform": "barrier_net_return", + "event_target": "take_profit_before_stop_loss", + "round_trip_cost": 0.0026, + "output_layout": ["mean", "q10", "q50", "q90", "logit_tp_first"], + "feature_names": ["return_1", "range_percent"], + "symbols": { + "BTCUSDT": { + "model": "torch_gru", + "architecture": "gru", + "lookback": 8, + "target_horizon": 3, + "target_horizons": [3], + "direct_horizon": True, + "target_transform": "barrier_net_return", + "event_target": "take_profit_before_stop_loss", + "target_stop_loss_percent": 0.04, + "target_take_profit_percent": 0.035, + "round_trip_cost": 0.0026, + "output_layout": ["mean", "q10", "q50", "q90", "logit_tp_first"], + "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], + "target_scales": [1.0], + "target_mean": 0.0, + "target_scale": 1.0, + "hidden_size": hidden_size, + "num_layers": 1, + "clip": 8.0, + "validation_mae_by_horizon": {"3": 0.01}, + "baseline_mae_by_horizon": {"3": 0.02}, + "validation_mae_percent": 1.0, + "baseline_mae_percent": 2.0, + "skill": 0.2, + "multitask_head": True, + "head_hidden_size": head_hidden_size, + "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_hidden_weight": [[0.0, 0.0], [0.0, 0.0]], + "head_hidden_bias": [0.0, 0.0], + "return_head_weight": [[0.0, 0.0] for _ in range(4)], + "return_head_bias": [0.01, -0.01, 0.005, 0.02], + "event_head_weight": [[0.0, 0.0]], + "event_head_bias": [1.38629436112], + } + }, + } + ), + encoding="utf-8", + ) + + def test_time_series_forecaster_requires_torch_artifact(make_settings, tmp_path) -> None: settings = make_settings( tmp_path, @@ -428,3 +501,24 @@ def test_time_series_forecaster_reads_probabilistic_multi_horizon_artifact(make_ assert forecast.feature_snapshot[0]["label"] == "Доходность 1ч" assert forecast.feature_snapshot[0]["raw_display"].endswith("%") assert "диапазон" in forecast.feature_snapshot[0]["interpretation"] + + +def test_time_series_forecaster_reads_barrier_multitask_artifact(make_settings, tmp_path) -> None: + artifact_path = tmp_path / "lstm_forecaster.json" + _write_barrier_multitask_gru_artifact(artifact_path) + settings = make_settings( + tmp_path, + time_series_lstm_model_path=artifact_path, + time_series_min_candles=80, + time_series_forecast_horizon=3, + ) + + forecast = TimeSeriesForecaster(settings).forecast( + _candles_from_returns([0.0002] * 140), symbol="BTCUSDT" + ) + + assert forecast.usable is True + assert forecast.target_transform == "barrier_net_return" + assert forecast.expected_return_percent == pytest.approx(1.005, abs=0.01) + assert forecast.probability_take_profit_first == pytest.approx(0.8, abs=0.001) + assert "P(TP before SL)" in forecast.reason diff --git a/tests/test_trade_objective_training.py b/tests/test_trade_objective_training.py new file mode 100644 index 0000000..44206a6 --- /dev/null +++ b/tests/test_trade_objective_training.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import math + +import pytest +import torch + +from crypto_spot_bot.models import Candle +from crypto_spot_bot.time_series import _torch_head_outputs +from tools.train_torch_recurrent_forecaster import ( + OUTPUT_LAYOUT, + RecurrentReturnModel, + _barrier_outcome, + _export_head_state, +) + + +def _candle(index: int, *, open_: float, high: float, low: float, close: float) -> Candle: + return Candle(index, open_, high, low, close, 100.0) + + +def test_barrier_target_uses_next_open_and_marks_take_profit_first() -> None: + candles = [ + _candle(0, open_=90.0, high=101.0, low=89.0, close=100.0), + _candle(1, open_=100.0, high=102.0, low=99.0, close=101.0), + _candle(2, open_=101.0, high=104.0, low=100.0, close=103.0), + ] + + net_return, event = _barrier_outcome( + candles, + end_index=0, + horizon=2, + stop_loss_percent=0.02, + take_profit_percent=0.03, + round_trip_cost=0.002, + ) or (math.nan, math.nan) + + assert event == 1.0 + assert net_return == pytest.approx(math.log(1.03) - 0.002) + + +def test_barrier_target_resolves_same_candle_tie_as_stop_loss() -> None: + candles = [ + _candle(0, open_=100.0, high=101.0, low=99.0, close=100.0), + _candle(1, open_=100.0, high=104.0, low=97.0, close=101.0), + ] + + net_return, event = _barrier_outcome( + candles, + end_index=0, + horizon=1, + stop_loss_percent=0.02, + take_profit_percent=0.03, + round_trip_cost=0.002, + ) or (math.nan, math.nan) + + assert event == 0.0 + assert net_return == pytest.approx(math.log(0.98) - 0.002) + + +def test_multitask_head_export_matches_runtime_inference() -> None: + torch.manual_seed(7) + model = RecurrentReturnModel( + architecture="gru", + input_size=2, + hidden_size=4, + num_layers=1, + dropout=0.0, + output_size=2 * len(OUTPUT_LAYOUT), + attention_pooling=False, + context_norm=False, + multitask_head=True, + head_hidden_size=6, + ) + model.eval() + context = torch.tensor([[0.2, -0.1, 0.4, 0.3]], dtype=torch.float32) + with torch.no_grad(): + shared = model.head_activation(model.head_hidden(context)) + returns = model.return_head(shared).view(1, 2, 4) + events = model.event_head(shared).view(1, 2, 1) + expected = torch.cat((returns, events), dim=2).reshape(-1).tolist() + + entry = {"multitask_head": True, **_export_head_state(model)} + actual = _torch_head_outputs(context[0].tolist(), entry, hidden_size=4) + + assert actual == pytest.approx(expected, abs=2e-6) diff --git a/tests/test_training_coordination.py b/tests/test_training_coordination.py index 14bf8d4..486b312 100644 --- a/tests/test_training_coordination.py +++ b/tests/test_training_coordination.py @@ -6,7 +6,7 @@ import json import pytest -from crypto_spot_bot.training_coordination import TrainingCoordinator +from crypto_spot_bot.training_coordination import TrainingCoordinator, _validate_symbol_models def test_training_coordinator_claims_and_completes_job(tmp_path) -> None: @@ -57,10 +57,15 @@ def test_training_coordinator_sanitizes_independent_training_parameters(tmp_path "parameters": { "pooled": False, "limit": 6000, + "validation_window": 720, "ensemble_seeds": "7,19", "selection_folds": 3, "learning_rate": 0.0007, "weight_decay": 0.0005, + "horizon": 12, + "horizons": "3,6,12,24", + "patience": 8, + "seed": 7, }, } ) @@ -68,10 +73,15 @@ def test_training_coordinator_sanitizes_independent_training_parameters(tmp_path assert requested["job"]["parameters"] == { "pooled": False, "limit": 6000, + "validation_window": 720, "ensemble_seeds": "7,19", "selection_folds": 3, "learning_rate": 0.0007, "weight_decay": 0.0005, + "horizon": 12, + "horizons": "3,6,12,24", + "patience": 8, + "seed": 7, } @@ -148,6 +158,30 @@ def test_training_coordinator_accepts_chunked_artifact_upload(tmp_path) -> None: assert coordinator.status()["latest_job"]["artifacts"][0]["sha256"] == sha256 +def test_model_validation_accepts_multitask_ensemble_members() -> None: + head = { + "state_dict": {"weight_ih_l0": [[0.0]]}, + "head_hidden_weight": [[0.0]], + "head_hidden_bias": [0.0], + "return_head_weight": [[0.0]], + "return_head_bias": [0.0], + "event_head_weight": [[0.0]], + "event_head_bias": [0.0], + } + symbols = { + "BTCUSDT": { + "model": "torch_gru", + "lookback": 8, + "input_size": 2, + "hidden_size": 4, + "multitask_head": True, + "ensemble_members": [head, head], + } + } + + _validate_symbol_models(symbols) + + def test_running_claimed_job_keeps_agent_online_when_heartbeat_is_stale(tmp_path) -> None: coordinator = TrainingCoordinator(tmp_path) coordinator.request_retrain({"source": "android"}) diff --git a/tools/calibrate_torch_thresholds.py b/tools/calibrate_torch_thresholds.py index ff676e7..73615bc 100644 --- a/tools/calibrate_torch_thresholds.py +++ b/tools/calibrate_torch_thresholds.py @@ -28,6 +28,7 @@ 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, + _barrier_outcome, _current_volatility_scale, _entry_horizon, _entry_output_layout, @@ -64,6 +65,7 @@ class ForecastRecord: future_net_percent: float benchmark_entry: bool benchmark_exit: bool + take_profit_first: bool | None = None @dataclass(slots=True) @@ -385,7 +387,22 @@ def _forecast_records( next_open = float(candles[index + 1].open) if next_open <= 0: continue - future_log_return = math.log(closes[index + decision_horizon] / next_open) - round_trip_cost + take_profit_first: bool | None = None + if str(entry.get("target_transform", "")) == "barrier_net_return": + outcome = _barrier_outcome( + candles, + end_index=index, + horizon=decision_horizon, + stop_loss_percent=_float_entry(entry, "target_stop_loss_percent", 0.04), + take_profit_percent=_float_entry(entry, "target_take_profit_percent", 0.035), + round_trip_cost=round_trip_cost, + ) + if outcome is None: + continue + future_log_return, event = outcome + take_profit_first = event >= 0.5 + else: + future_log_return = math.log(closes[index + decision_horizon] / next_open) - round_trip_cost future_net_percent = (math.exp(future_log_return) - 1.0) * 100.0 records.append( ForecastRecord( @@ -407,6 +424,7 @@ def _forecast_records( future_net_percent=future_net_percent, benchmark_entry=_benchmark_entry_signal(candles, trend_candles, index), benchmark_exit=_benchmark_exit_signal(candles, index), + take_profit_first=take_profit_first, ) ) return records @@ -497,7 +515,22 @@ def _batch_forecast_records( next_open = float(candles[index + 1].open) if next_open <= 0: continue - future_log_return = math.log(closes[index + decision_horizon] / next_open) - round_trip_cost + take_profit_first: bool | None = None + if str(entry.get("target_transform", "")) == "barrier_net_return": + outcome = _barrier_outcome( + candles, + end_index=index, + horizon=decision_horizon, + stop_loss_percent=_float_entry(entry, "target_stop_loss_percent", 0.04), + take_profit_percent=_float_entry(entry, "target_take_profit_percent", 0.035), + round_trip_cost=round_trip_cost, + ) + if outcome is None: + continue + future_log_return, event = outcome + take_profit_first = event >= 0.5 + else: + future_log_return = math.log(closes[index + decision_horizon] / next_open) - round_trip_cost future_net_percent = (math.exp(future_log_return) - 1.0) * 100.0 records.append( ForecastRecord( @@ -519,6 +552,7 @@ def _batch_forecast_records( future_net_percent=future_net_percent, benchmark_entry=_benchmark_entry_signal(candles, trend_candles, index), benchmark_exit=_benchmark_exit_signal(candles, index), + take_profit_first=take_profit_first, ) ) return records @@ -557,6 +591,10 @@ def _build_torch_model(entry: dict[str, Any], model_name: str) -> Any | None: output_size=output_size, attention_pooling=bool(entry.get("attention_pooling")), context_norm=bool(entry.get("context_norm")), + multitask_head=bool(entry.get("multitask_head")), + head_hidden_size=int( + _clamp(_float_entry(entry, "head_hidden_size", float(hidden_size)), 8.0, 1024.0) + ), ) raw_state = entry.get("state_dict") if not isinstance(raw_state, dict): @@ -566,12 +604,26 @@ def _build_torch_model(entry: dict[str, Any], model_name: str) -> Any | None: for key, value in raw_state.items() if isinstance(value, list) } - head_weight = entry.get("head_weight") - head_bias = entry.get("head_bias") - if not isinstance(head_weight, list) or not isinstance(head_bias, list): - return None - state["head.weight"] = torch.tensor(head_weight, dtype=torch.float32) - state["head.bias"] = torch.tensor(head_bias, dtype=torch.float32) + if bool(entry.get("multitask_head")): + for artifact_name, state_name in ( + ("head_hidden_weight", "head_hidden.weight"), + ("head_hidden_bias", "head_hidden.bias"), + ("return_head_weight", "return_head.weight"), + ("return_head_bias", "return_head.bias"), + ("event_head_weight", "event_head.weight"), + ("event_head_bias", "event_head.bias"), + ): + value = entry.get(artifact_name) + if not isinstance(value, list): + return None + state[state_name] = torch.tensor(value, dtype=torch.float32) + else: + head_weight = entry.get("head_weight") + head_bias = entry.get("head_bias") + if not isinstance(head_weight, list) or not isinstance(head_bias, list): + return None + state["head.weight"] = torch.tensor(head_weight, dtype=torch.float32) + state["head.bias"] = torch.tensor(head_bias, dtype=torch.float32) if bool(entry.get("attention_pooling")): attention_weight = entry.get("attention_weight") if not isinstance(attention_weight, list): @@ -639,10 +691,20 @@ def _decode_selected_output( expected = decode("mean") q_values = sorted([decode("q10", expected), decode("q50", expected), decode("q90", expected)]) cap = _prediction_cap(history_closes, selected_horizon, round_trip_cost) + if str(entry.get("target_transform", "")) == "barrier_net_return": + stop_percent = _clamp(_float_entry(entry, "target_stop_loss_percent", 0.04), 0.003, 0.08) + take_percent = _clamp(_float_entry(entry, "target_take_profit_percent", 0.035), 0.003, 0.20) + cap = max( + cap, + abs(math.log(1.0 - stop_percent) - round_trip_cost), + abs(math.log(1.0 + take_percent) - round_trip_cost), + ) return { "expected_return": _clamp(expected, -cap, cap), "q50": _clamp(q_values[1], -cap, cap), - "probability_up": _sigmoid(float(values.get("logit_up", 0.0))), + "probability_up": _sigmoid( + float(values.get("logit_tp_first", values.get("logit_up", 0.0))) + ), } @@ -1422,7 +1484,7 @@ def _fit_platt_calibration(records: list[ForecastRecord]) -> dict[str, float]: samples = [ ( math.log(_clamp(record.probability_up, 1e-5, 1.0 - 1e-5) / (1.0 - _clamp(record.probability_up, 1e-5, 1.0 - 1e-5))), - 1.0 if record.future_net_percent > 0 else 0.0, + _record_event_target(record), ) for record in records ] @@ -1446,6 +1508,12 @@ def _fit_platt_calibration(records: list[ForecastRecord]) -> dict[str, float]: return {"slope": round(slope, 8), "intercept": round(intercept, 8), "samples": float(len(samples))} +def _record_event_target(record: ForecastRecord) -> float: + if record.take_profit_first is not None: + return 1.0 if record.take_profit_first else 0.0 + return 1.0 if record.future_net_percent > 0 else 0.0 + + def _apply_platt_calibration( records: list[ForecastRecord], calibration: dict[str, float] ) -> list[ForecastRecord]: @@ -1544,6 +1612,9 @@ def _artifact_summary(artifact: dict[str, Any]) -> dict[str, Any]: "target_horizon": artifact.get("target_horizon"), "target_horizons": artifact.get("target_horizons"), "target_transform": artifact.get("target_transform"), + "event_target": artifact.get("event_target"), + "target_stop_loss_percent": artifact.get("target_stop_loss_percent"), + "target_take_profit_percent": artifact.get("target_take_profit_percent"), "symbols": { symbol: { "model": row.get("model"), diff --git a/tools/run_torch_retrain.ps1 b/tools/run_torch_retrain.ps1 index bc17ca6..73d6c24 100644 --- a/tools/run_torch_retrain.ps1 +++ b/tools/run_torch_retrain.ps1 @@ -18,6 +18,7 @@ param( [double]$WeightDecay = 0, [int]$Epochs = 0, [int]$Patience = 0, + [int]$ValidationWindow = 0, [int]$HoldoutWindow = 0, [string]$Interval = "", [string]$EnvFile = "", @@ -155,6 +156,7 @@ if ($LearningRate -le 0) { $LearningRate = if ($env:TORCH_RETRAIN_LEARNING_RATE) if ($WeightDecay -le 0) { $WeightDecay = if ($env:TORCH_RETRAIN_WEIGHT_DECAY) { [double]$env:TORCH_RETRAIN_WEIGHT_DECAY } else { 0.0005 } } if ($Epochs -le 0) { $Epochs = if ($env:TORCH_RETRAIN_EPOCHS) { [int]$env:TORCH_RETRAIN_EPOCHS } else { 70 } } if ($Patience -le 0) { $Patience = if ($env:TORCH_RETRAIN_PATIENCE) { [int]$env:TORCH_RETRAIN_PATIENCE } else { 8 } } +if ($ValidationWindow -le 0) { $ValidationWindow = if ($env:TORCH_RETRAIN_VALIDATION_WINDOW) { [int]$env:TORCH_RETRAIN_VALIDATION_WINDOW } else { 720 } } if ($HoldoutWindow -le 0) { $HoldoutWindow = if ($env:TORCH_RETRAIN_HOLDOUT_WINDOW) { [int]$env:TORCH_RETRAIN_HOLDOUT_WINDOW } else { 1000 } } if (-not $Interval -and $env:TORCH_RETRAIN_INTERVAL) { $Interval = $env:TORCH_RETRAIN_INTERVAL } if (-not $EnvFile -and $env:TORCH_RETRAIN_ENV) { $EnvFile = $env:TORCH_RETRAIN_ENV } @@ -190,6 +192,7 @@ try { "--dropouts", $Dropouts, "--epochs", $Epochs.ToString(), "--patience", $Patience.ToString(), + "--validation-window", $ValidationWindow.ToString(), "--holdout-window", $HoldoutWindow.ToString(), "--ensemble-seeds", $EnsembleSeeds, "--selection-folds", $SelectionFolds.ToString(), @@ -242,6 +245,7 @@ try { "-u", "tools\calibrate_torch_thresholds.py", "--limit", $Limit.ToString(), + "--horizon", $Horizon.ToString(), "--calibration-window", ([Math]::Min(2400, [Math]::Max(1200, [int]($Limit / 2)))).ToString(), "--min-trades", "24", "--walk-forward-folds", "8", diff --git a/tools/train_torch_recurrent_forecaster.py b/tools/train_torch_recurrent_forecaster.py index 7ed072e..0c21e8b 100644 --- a/tools/train_torch_recurrent_forecaster.py +++ b/tools/train_torch_recurrent_forecaster.py @@ -28,10 +28,18 @@ 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 +from crypto_spot_bot.time_series import ( + DEFAULT_TORCH_FEATURES, + _barrier_outcome, + _feature_matrix, + _log_returns, +) -OUTPUT_LAYOUT = ("mean", "q10", "q50", "q90", "logit_up") +RETURN_OUTPUT_LAYOUT = ("mean", "q10", "q50", "q90") +EVENT_OUTPUT_NAME = "logit_tp_first" +OUTPUT_LAYOUT = (*RETURN_OUTPUT_LAYOUT, EVENT_OUTPUT_NAME) +TARGET_TRANSFORM = "barrier_net_return" QUANTILES = {"q10": 0.10, "q50": 0.50, "q90": 0.90} @@ -44,11 +52,13 @@ class PreparedData: validation_y: torch.Tensor validation_up: torch.Tensor validation_targets: list[list[float]] + validation_event_targets: list[list[float]] validation_volatility_scales: list[list[float]] holdout_x: torch.Tensor holdout_y: torch.Tensor holdout_up: torch.Tensor holdout_targets: list[list[float]] + holdout_event_targets: list[list[float]] holdout_volatility_scales: list[list[float]] holdout_start_timestamp: int feature_names: list[str] @@ -69,6 +79,7 @@ class TrainingSample: window: list[list[float]] normalized_targets: list[float] raw_targets: list[float] + event_targets: list[float] volatility_scales: list[float] timestamp: int @@ -85,6 +96,8 @@ class RecurrentReturnModel(nn.Module): output_size: int, attention_pooling: bool, context_norm: bool, + multitask_head: bool = False, + head_hidden_size: int = 0, ) -> None: super().__init__() recurrent_cls = nn.LSTM if architecture == "lstm" else nn.GRU @@ -97,7 +110,30 @@ class RecurrentReturnModel(nn.Module): ) 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) + self.multitask_head = bool(multitask_head) + self.output_size = output_size + if self.multitask_head: + if output_size % len(OUTPUT_LAYOUT) != 0: + raise ValueError("multitask output size must align with OUTPUT_LAYOUT") + self.horizon_count = output_size // len(OUTPUT_LAYOUT) + projected_size = max(8, int(head_hidden_size or hidden_size)) + self.head_hidden = nn.Linear(hidden_size, projected_size) + self.head_activation = nn.GELU() + self.head_dropout = nn.Dropout(dropout) + self.return_head = nn.Linear( + projected_size, + self.horizon_count * len(RETURN_OUTPUT_LAYOUT), + ) + self.event_head = nn.Linear(projected_size, self.horizon_count) + self.head = None + else: + self.horizon_count = 0 + self.head_hidden = None + self.head_activation = None + self.head_dropout = None + self.return_head = None + self.event_head = None + self.head = nn.Linear(hidden_size, output_size) def forward(self, values: torch.Tensor) -> torch.Tensor: output, _state = self.rnn(values) @@ -107,7 +143,21 @@ class RecurrentReturnModel(nn.Module): context = (output * weights).sum(dim=1) else: context = output[:, -1, :] - return self.head(self.context_norm(context)) + context = self.context_norm(context) + if not self.multitask_head: + assert self.head is not None + return self.head(context) + assert self.head_hidden is not None + assert self.head_activation is not None + assert self.head_dropout is not None + assert self.return_head is not None + assert self.event_head is not None + shared = self.head_dropout(self.head_activation(self.head_hidden(context))) + returns = self.return_head(shared).view( + values.shape[0], self.horizon_count, len(RETURN_OUTPUT_LAYOUT) + ) + events = self.event_head(shared).view(values.shape[0], self.horizon_count, 1) + return torch.cat((returns, events), dim=2).reshape(values.shape[0], self.output_size) def main() -> None: @@ -129,13 +179,15 @@ def main() -> None: feature_names.extend(f"symbol_is_{symbol}" for symbol in symbols) ensemble_seeds = _ints(args.ensemble_seeds) or [args.seed] round_trip_cost = max(0.0, 2.0 * (float(settings.taker_fee_rate) + float(settings.slippage_rate))) + stop_loss_percent = _clamp(float(settings.stop_loss_percent), 0.003, 0.08) + take_profit_percent = _clamp(float(settings.take_profit_percent), 0.003, 0.20) _progress( f"training started: symbols={len(symbols)} interval={interval} " f"limit={args.limit} epochs={args.epochs}" ) artifact: dict[str, Any] = { - "version": 4, + "version": 6, "type": "pytorch_recurrent_forecaster", "created_at": datetime.now(timezone.utc).isoformat(), "trainer": Path(__file__).name, @@ -146,8 +198,12 @@ def main() -> None: "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", + "target_transform": TARGET_TRANSFORM, + "target_return": "first_barrier_or_horizon_after_cost_log_return", + "event_target": "take_profit_before_stop_loss", + "target_stop_loss_percent": round(stop_loss_percent, 8), + "target_take_profit_percent": round(take_profit_percent, 8), + "barrier_tie_policy": "stop_loss_first", "round_trip_cost": round(round_trip_cost, 10), "output_layout": list(OUTPUT_LAYOUT), "quantiles": list(QUANTILES.values()), @@ -160,7 +216,7 @@ def main() -> None: } if args.pooled: - artifact["version"] = 5 + artifact["version"] = 7 artifact["pooled_multi_asset"] = True artifact["symbol_embedding"] = "learned_one_hot_projection" artifact["symbols"] = _train_pooled_symbols( @@ -174,6 +230,8 @@ def main() -> None: decision_horizon=decision_horizon, feature_names=feature_names, round_trip_cost=round_trip_cost, + stop_loss_percent=stop_loss_percent, + take_profit_percent=take_profit_percent, context_symbols=_strings(args.context_symbols), architectures=_strings(args.architectures), lookbacks=_ints(args.lookbacks), @@ -202,6 +260,8 @@ def main() -> None: decision_horizon=decision_horizon, feature_names=feature_names, round_trip_cost=round_trip_cost, + stop_loss_percent=stop_loss_percent, + take_profit_percent=take_profit_percent, device=device, ensemble_seeds=ensemble_seeds, ) @@ -213,8 +273,8 @@ def main() -> None: 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}" + f"skill={result['skill']:.4f} tp_precision={result.get('take_profit_first_precision', 0.0):.3f} " + f"tp_brier={result['probability_brier']:.4f}" ) output.parent.mkdir(parents=True, exist_ok=True) @@ -227,7 +287,8 @@ def main() -> None: def _train_independent_symbols( *, client: BybitClient, symbols: list[str], interval: str, args: argparse.Namespace, target_horizons: list[int], decision_horizon: int, feature_names: list[str], - round_trip_cost: float, device: torch.device, ensemble_seeds: list[int], + round_trip_cost: float, stop_loss_percent: float, take_profit_percent: float, + device: torch.device, ensemble_seeds: list[int], ) -> dict[str, Any]: results: dict[str, Any] = {} total_symbols = len(symbols) @@ -244,6 +305,8 @@ def _train_independent_symbols( decision_horizon=decision_horizon, feature_names=feature_names, round_trip_cost=round_trip_cost, + stop_loss_percent=stop_loss_percent, + take_profit_percent=take_profit_percent, context_symbols=_strings(args.context_symbols), architectures=_strings(args.architectures), lookbacks=_ints(args.lookbacks), @@ -273,6 +336,7 @@ def _train_pooled_symbols( *, client: BybitClient, symbols: list[str], interval: str, limit: int, validation_window: int, holdout_window: int, target_horizons: list[int], decision_horizon: int, feature_names: list[str], round_trip_cost: float, + stop_loss_percent: float, take_profit_percent: 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, @@ -303,6 +367,8 @@ def _train_pooled_symbols( target_horizons=target_horizons, decision_horizon=decision_horizon, round_trip_cost=round_trip_cost, + stop_loss_percent=stop_loss_percent, + take_profit_percent=take_profit_percent, market_candles=market_candles, trend_candles=trend_by_symbol[symbol], validation_window=validation_window, @@ -356,6 +422,7 @@ def _train_pooled_symbols( dropout=dropout if num_layers > 1 else 0.0, attention_pooling=attention_pooling, context_norm=context_norm, input_size=len(feature_names), output_size=len(target_horizons) * len(OUTPUT_LAYOUT), + multitask_head=True, head_hidden_size=hidden_size, ) if best is None or _candidate_score(candidate) < _candidate_score(best): best = candidate @@ -376,7 +443,10 @@ def _train_pooled_symbols( "target_horizon": prepared.decision_horizon, "target_horizons": prepared.target_horizons, "direct_horizon": True, - "target_transform": "net_return_over_volatility", + "target_transform": TARGET_TRANSFORM, + "event_target": "take_profit_before_stop_loss", + "target_stop_loss_percent": stop_loss_percent, + "target_take_profit_percent": take_profit_percent, "round_trip_cost": round(round_trip_cost, 10), "output_layout": list(OUTPUT_LAYOUT), "feature_names": feature_names, @@ -413,6 +483,7 @@ def _fit_pooled_candidate( 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, + multitask_head=True, head_hidden_size=hidden_size, ).to(device) optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate, weight_decay=weight_decay) loader = DataLoader( @@ -440,7 +511,7 @@ def _fit_pooled_candidate( symbol: _validation_metrics(model, prepared, clip) for symbol, prepared in prepared_by_symbol.items() } - score = sum(float(row["validation_mae"]) for row in symbol_rows.values()) / len(symbol_rows) + score = sum(_candidate_score(row) for row in symbol_rows.values()) / len(symbol_rows) if score + 1e-12 < best_score: best_score = score best_epoch = epoch @@ -462,15 +533,15 @@ def _fit_pooled_candidate( for name in ( "validation_mae", "directional_accuracy", "buy_precision", "probability_brier", "holdout_skill", "validation_fold_mae_std", "validation_trade_mean", - "validation_trade_win_rate", + "validation_trade_win_rate", "take_profit_first_accuracy", + "take_profit_first_precision", "take_profit_first_brier", ): values = [float(row[name]) for row in per_symbol.values() if isinstance(row.get(name), (int, float))] aggregate[name] = sum(values) / len(values) if values else 0.0 aggregate.update( best_epoch=best_epoch, epochs_trained=best_epoch + stale, 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_head_state(model), **_export_context_state(model), ) return aggregate @@ -535,6 +606,8 @@ def _train_symbol( decision_horizon: int, feature_names: list[str], round_trip_cost: float, + stop_loss_percent: float, + take_profit_percent: float, context_symbols: list[str], architectures: list[str], lookbacks: list[int], @@ -587,6 +660,8 @@ def _train_symbol( target_horizons=target_horizons, decision_horizon=decision_horizon, round_trip_cost=round_trip_cost, + stop_loss_percent=stop_loss_percent, + take_profit_percent=take_profit_percent, market_candles=market_candles, trend_candles=trend_candles, validation_window=validation_window, @@ -630,6 +705,8 @@ def _train_symbol( clip=clip, attention_pooling=attention_pooling, context_norm=context_norm, + multitask_head=True, + head_hidden_size=hidden_size, device=device, seed=member_seed, selection_folds=selection_folds, @@ -647,8 +724,12 @@ def _train_symbol( "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", + "target_transform": TARGET_TRANSFORM, + "target_return": "first_barrier_or_horizon_after_cost_log_return", + "event_target": "take_profit_before_stop_loss", + "target_stop_loss_percent": stop_loss_percent, + "target_take_profit_percent": take_profit_percent, + "barrier_tie_policy": "stop_loss_first", "round_trip_cost": round(round_trip_cost, 10), "output_layout": list(OUTPUT_LAYOUT), "quantiles": list(QUANTILES.values()), @@ -668,6 +749,8 @@ def _train_symbol( "dropout": dropout if num_layers > 1 else 0.0, "attention_pooling": attention_pooling, "context_norm": context_norm, + "multitask_head": True, + "head_hidden_size": hidden_size, "clip": clip, "validation_mae_percent": validation_mae * 100, "baseline_mae_percent": baseline_mae * 100, @@ -700,6 +783,8 @@ def _prepare_data( target_horizons: list[int], decision_horizon: int, round_trip_cost: float, + stop_loss_percent: float, + take_profit_percent: float, market_candles: dict[str, list[Candle]], trend_candles: list[Candle], validation_window: int, @@ -724,25 +809,35 @@ def _prepare_data( if len(window) != lookback: continue raw_targets: list[float] = [] + event_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: + outcome = _barrier_outcome( + candles, + end_index=end_index, + horizon=horizon, + stop_loss_percent=stop_loss_percent, + take_profit_percent=take_profit_percent, + round_trip_cost=round_trip_cost, + ) + if outcome is None: valid = False break - net_return = math.log(future / current) - round_trip_cost - volatility_scale = _target_volatility_scale(candles, closes, end_index, horizon) + net_return, take_profit_first = outcome + volatility_scale = 1.0 raw_targets.append(net_return) + event_targets.append(take_profit_first) volatility_scales.append(volatility_scale) - normalized_targets.append(net_return / max(volatility_scale, 1e-8)) + normalized_targets.append(net_return) if valid: samples.append( TrainingSample( window, normalized_targets, raw_targets, + event_targets, volatility_scales, candles[end_index].timestamp, ) @@ -803,11 +898,13 @@ def _prepare_data( 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_event_targets=[sample.event_targets for sample in validation_samples], validation_volatility_scales=[sample.volatility_scales for sample in validation_samples], holdout_x=torch.tensor(holdout_x, dtype=torch.float32, device=device), holdout_y=torch.tensor(holdout_y, dtype=torch.float32, device=device), holdout_up=torch.tensor(holdout_up, dtype=torch.float32, device=device), holdout_targets=[sample.raw_targets for sample in holdout_samples], + holdout_event_targets=[sample.event_targets for sample in holdout_samples], holdout_volatility_scales=[sample.volatility_scales for sample in holdout_samples], holdout_start_timestamp=holdout_samples[0].timestamp, feature_names=feature_names, @@ -902,7 +999,7 @@ def _normalize_samples( 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]) + up_values.append(list(sample.event_targets)) return x_values, y_values, up_values @@ -923,6 +1020,8 @@ def _fit_candidate( clip: float, attention_pooling: bool, context_norm: bool, + multitask_head: bool, + head_hidden_size: int, device: torch.device, seed: int, selection_folds: int, @@ -937,6 +1036,8 @@ def _fit_candidate( output_size=output_size, attention_pooling=attention_pooling, context_norm=context_norm, + multitask_head=multitask_head, + head_hidden_size=head_hidden_size, ).to(device) optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate, weight_decay=weight_decay) generator = torch.Generator(device="cpu").manual_seed(seed) @@ -948,7 +1049,14 @@ def _fit_candidate( ) 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_metrics: dict[str, float] = { + "validation_mae": math.inf, + "directional_accuracy": 0.0, + "buy_precision": 0.0, + "probability_brier": 1.0, + "validation_trade_mean": -math.inf, + } + best_score = math.inf best_epoch = 0 stale_epochs = 0 for epoch in range(1, max(1, epochs) + 1): @@ -962,8 +1070,10 @@ def _fit_candidate( metrics = _validation_metrics(model, prepared, clip) metrics.update(_validation_stability_metrics(model, prepared, clip, selection_folds)) - if metrics["validation_mae"] + 1e-12 < best_metrics["validation_mae"]: + score = _candidate_score(metrics) + if score + 1e-12 < best_score: best_metrics = metrics + best_score = score best_epoch = epoch best_state = {key: value.detach().cpu().clone() for key, value in model.state_dict().items()} stale_epochs = 0 @@ -981,8 +1091,7 @@ def _fit_candidate( "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_head_state(model), **_export_context_state(model), } @@ -1006,6 +1115,12 @@ def _ensemble_candidate(members: list[dict[str, Any]], seeds: list[int]) -> dict "validation_fold_mae_worst", "validation_trade_mean", "validation_trade_win_rate", + "take_profit_first_accuracy", + "take_profit_first_precision", + "take_profit_first_brier", + "holdout_take_profit_first_accuracy", + "holdout_take_profit_first_precision", + "holdout_take_profit_first_brier", ) for name in metric_names: values = [float(member[name]) for member in members if isinstance(member.get(name), (int, float))] @@ -1015,6 +1130,12 @@ def _ensemble_candidate(members: list[dict[str, Any]], seeds: list[int]) -> dict "state_dict", "head_weight", "head_bias", + "head_hidden_weight", + "head_hidden_bias", + "return_head_weight", + "return_head_bias", + "event_head_weight", + "event_head_bias", "attention_weight", "attention_bias", "context_norm_weight", @@ -1062,6 +1183,7 @@ def _validation_metrics(model: nn.Module, prepared: PreparedData, clip: float) - model, values=prepared.validation_x, targets=prepared.validation_targets, + event_targets=prepared.validation_event_targets, volatility_scales=prepared.validation_volatility_scales, prepared=prepared, clip=clip, @@ -1086,6 +1208,7 @@ def _validation_stability_metrics( model, values=prepared.validation_x[start:end], targets=prepared.validation_targets[start:end], + event_targets=prepared.validation_event_targets[start:end], volatility_scales=prepared.validation_volatility_scales[start:end], prepared=prepared, clip=clip, @@ -1106,6 +1229,7 @@ def _holdout_metrics(model: nn.Module, prepared: PreparedData, clip: float) -> d model, values=prepared.holdout_x, targets=prepared.holdout_targets, + event_targets=prepared.holdout_event_targets, volatility_scales=prepared.holdout_volatility_scales, prepared=prepared, clip=clip, @@ -1126,6 +1250,9 @@ def _holdout_metrics(model: nn.Module, prepared: PreparedData, clip: float) -> d "holdout_directional_accuracy": metrics["directional_accuracy"], "holdout_buy_precision": metrics["buy_precision"], "holdout_probability_brier": metrics["probability_brier"], + "holdout_take_profit_first_accuracy": metrics["take_profit_first_accuracy"], + "holdout_take_profit_first_precision": metrics["take_profit_first_precision"], + "holdout_take_profit_first_brier": metrics["take_profit_first_brier"], } @@ -1134,6 +1261,7 @@ def _evaluation_metrics( *, values: torch.Tensor, targets: list[list[float]], + event_targets: list[list[float]], volatility_scales: list[list[float]], prepared: PreparedData, clip: float, @@ -1173,12 +1301,13 @@ def _evaluation_metrics( for prediction, actual in zip(decision_predictions, decision_targets) if prediction != 0 and actual != 0 ] + decision_events = [row[decision] for row in event_targets] buy_predictions = [ - actual - for prediction, actual in zip(decision_predictions, decision_targets) + event + for prediction, event in zip(decision_predictions, decision_events) if prediction > 0 ] - buy_wins = [actual for actual in buy_predictions if actual > 0] + buy_wins = [event for event in buy_predictions if event >= 0.5] ranked = sorted( zip(decision_predictions, [row[decision] for row in probabilities], decision_targets), key=lambda item: item[0] * max(0.0, item[1] - 0.5), @@ -1201,16 +1330,36 @@ def _evaluation_metrics( 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) + (probabilities[row_index][decision] - decision_events[row_index]) ** 2 + for row_index in range(len(decision_events)) ] + event_predictions = [1.0 if row[decision] >= 0.5 else 0.0 for row in probabilities] + event_correct = sum( + 1 for prediction, actual in zip(event_predictions, decision_events) if prediction == actual + ) + event_precision_denominator = sum(1 for value in event_predictions if value >= 0.5) + event_true_positives = sum( + 1 + for prediction, actual in zip(event_predictions, decision_events) + if prediction >= 0.5 and actual >= 0.5 + ) + event_accuracy = event_correct / len(decision_events) if decision_events else 0.0 + event_precision = ( + event_true_positives / event_precision_denominator + if event_precision_denominator + else 0.0 + ) + event_brier = sum(probability_errors) / len(probability_errors) if probability_errors else 1.0 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, + "probability_brier": event_brier, + "take_profit_first_accuracy": event_accuracy, + "take_profit_first_precision": event_precision, + "take_profit_first_brier": event_brier, "validation_trade_mean": sum(selected_targets) / len(selected_targets) if selected_targets else 0.0, "validation_trade_win_rate": ( sum(1 for value in selected_targets if value > 0) / len(selected_targets) @@ -1229,10 +1378,19 @@ def _candidate_score(row: dict[str, Any]) -> float: fold_std = max(0.0, float(row.get("validation_fold_mae_std", 0.0))) stability_penalty = 1.0 + min(1.0, fold_std / max(mae, 1e-9)) * 0.25 trade_mean = float(row.get("validation_trade_mean", 0.0)) - trade_penalty = max(0.0, -trade_mean) * 2.0 - max(0.0, trade_mean) * 0.5 - return mae * stability_penalty * (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) + trade_penalty + event_precision = float(row.get("take_profit_first_precision", buy_precision)) + trade_penalty = max(0.0, -trade_mean) * 4.0 - max(0.0, trade_mean) * 0.75 + probability_penalty = mae * max(0.0, probability_brier - 0.20) * 0.5 + event_bonus = mae * max(0.0, event_precision - 0.5) * 0.10 + return ( + mae + * stability_penalty + * (1.0 - max(0.0, skill) * 0.05) + * (1.0 - max(0.0, directional - 0.5) * 0.03) + + probability_penalty + + trade_penalty + - event_bonus + ) def _forecast_loss(outputs: torch.Tensor, targets: torch.Tensor, up_targets: torch.Tensor, horizon_count: int) -> torch.Tensor: @@ -1244,7 +1402,14 @@ def _forecast_loss(outputs: torch.Tensor, targets: torch.Tensor, up_targets: tor 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") + positive_rate = up_targets.mean().detach().clamp(0.05, 0.95) + positive_weight = ((1.0 - positive_rate) / positive_rate).clamp(0.5, 5.0) + bce = nn.functional.binary_cross_entropy_with_logits( + logits, + up_targets, + reduction="none", + pos_weight=positive_weight, + ) probabilities = torch.sigmoid(logits) pt = probabilities * up_targets + (1.0 - probabilities) * (1.0 - up_targets) focal = ((1.0 - pt) ** 2.0 * bce).mean() @@ -1257,8 +1422,8 @@ def _forecast_loss(outputs: torch.Tensor, targets: torch.Tensor, up_targets: tor return ( mean_loss + 0.35 * sum(quantile_losses) / len(quantile_losses) - + 0.15 * focal - + 0.10 * after_cost_utility + + 0.35 * focal + + 0.20 * after_cost_utility + 0.05 * ranking_loss ) @@ -1270,6 +1435,26 @@ def _export_recurrent_state(model: RecurrentReturnModel) -> dict[str, Any]: } +def _export_head_state(model: RecurrentReturnModel) -> dict[str, Any]: + if not model.multitask_head: + assert model.head is not None + return { + "head_weight": _round_nested(model.head.weight.detach().cpu().tolist()), + "head_bias": _round_list(model.head.bias.detach().cpu().tolist()), + } + assert model.head_hidden is not None + assert model.return_head is not None + assert model.event_head is not None + return { + "head_hidden_weight": _round_nested(model.head_hidden.weight.detach().cpu().tolist()), + "head_hidden_bias": _round_list(model.head_hidden.bias.detach().cpu().tolist()), + "return_head_weight": _round_nested(model.return_head.weight.detach().cpu().tolist()), + "return_head_bias": _round_list(model.return_head.bias.detach().cpu().tolist()), + "event_head_weight": _round_nested(model.event_head.weight.detach().cpu().tolist()), + "event_head_bias": _round_list(model.event_head.bias.detach().cpu().tolist()), + } + + def _export_context_state(model: RecurrentReturnModel) -> dict[str, Any]: exported: dict[str, Any] = {} if model.attention is not None: diff --git a/tools/windows_training_agent.py b/tools/windows_training_agent.py index 7fe5f8d..6ef8f2c 100644 --- a/tools/windows_training_agent.py +++ b/tools/windows_training_agent.py @@ -108,11 +108,19 @@ def run_retrain(args: argparse.Namespace, job_id: str, job: dict[str, Any], repo "layers": "-Layers", "dropouts": "-Dropouts", "epochs": "-Epochs", + "validation_window": "-ValidationWindow", "holdout_window": "-HoldoutWindow", "ensemble_seeds": "-EnsembleSeeds", "selection_folds": "-SelectionFolds", "learning_rate": "-LearningRate", "weight_decay": "-WeightDecay", + "horizon": "-Horizon", + "horizons": "-Horizons", + "patience": "-Patience", + "context_symbols": "-ContextSymbols", + "features": "-Features", + "seed": "-Seed", + "interval": "-Interval", } for key, ps_arg in arg_map.items(): value = parameters.get(key)