diff --git a/.env.example b/.env.example index 5064ef9..21690a7 100644 --- a/.env.example +++ b/.env.example @@ -56,7 +56,9 @@ TREND_RSI_MAX=65 TIME_SERIES_FORECAST_ENABLED=true TIME_SERIES_MIN_CANDLES=120 TIME_SERIES_FORECAST_HORIZON=3 -TIME_SERIES_MIN_EDGE_PERCENT=0.04 +TIME_SERIES_MIN_EDGE_PERCENT=0.10 +TIME_SERIES_MIN_PROBABILITY_UP=0.64 +TIME_SERIES_MIN_CONFIDENCE=0.72 TIME_SERIES_MAX_ADJUSTMENT=0.08 TIME_SERIES_LSTM_ENABLED=true TIME_SERIES_LSTM_MODEL_PATH=runtime/lstm_forecaster.json diff --git a/README.md b/README.md index 5023783..4bc6ced 100644 --- a/README.md +++ b/README.md @@ -154,7 +154,9 @@ TREND_RSI_MAX=65 TIME_SERIES_FORECAST_ENABLED=true TIME_SERIES_MIN_CANDLES=120 TIME_SERIES_FORECAST_HORIZON=3 -TIME_SERIES_MIN_EDGE_PERCENT=0.04 +TIME_SERIES_MIN_EDGE_PERCENT=0.10 +TIME_SERIES_MIN_PROBABILITY_UP=0.64 +TIME_SERIES_MIN_CONFIDENCE=0.72 TIME_SERIES_MAX_ADJUSTMENT=0.08 TIME_SERIES_LSTM_ENABLED=true TIME_SERIES_LSTM_MODEL_PATH=runtime/lstm_forecaster.json diff --git a/crypto_spot_bot/config.py b/crypto_spot_bot/config.py index 642d8db..3e0999c 100644 --- a/crypto_spot_bot/config.py +++ b/crypto_spot_bot/config.py @@ -108,6 +108,8 @@ class Settings: time_series_min_candles: int time_series_forecast_horizon: int time_series_min_edge_percent: float + time_series_min_probability_up: float + time_series_min_confidence: float time_series_max_adjustment: float time_series_lstm_enabled: bool time_series_lstm_model_path: Path @@ -185,6 +187,7 @@ def load_settings(env_file: str | Path | None = None) -> Settings: top_symbols_count = len(FIXED_SPOT_SYMBOLS) symbols = FIXED_SPOT_SYMBOLS forecast_enabled_default = strategy_mode == "torch_forecast" + min_signal_confidence = _float_env("MIN_SIGNAL_CONFIDENCE", 0.64) settings = Settings( trading_mode=mode, host=os.getenv("HOST", "127.0.0.1"), @@ -207,7 +210,7 @@ def load_settings(env_file: str | Path | None = None) -> Settings: fast_entry_cooldown_seconds=_int_env("FAST_ENTRY_COOLDOWN_SECONDS", 20), max_entries_per_minute=_int_env("MAX_ENTRIES_PER_MINUTE", 12), websocket_enabled=_bool_env("WEBSOCKET_ENABLED", True), - min_signal_confidence=_float_env("MIN_SIGNAL_CONFIDENCE", 0.64), + min_signal_confidence=min_signal_confidence, max_spread_percent=_float_env("MAX_SPREAD_PERCENT", 0.18), min_24h_turnover_usdt=_float_env("MIN_24H_TURNOVER_USDT", 1000000.0), pattern_analysis_enabled=_bool_env("PATTERN_ANALYSIS_ENABLED", False), @@ -247,7 +250,9 @@ def load_settings(env_file: str | Path | None = None) -> Settings: time_series_forecast_enabled=_bool_env("TIME_SERIES_FORECAST_ENABLED", forecast_enabled_default), time_series_min_candles=_int_env("TIME_SERIES_MIN_CANDLES", 120), time_series_forecast_horizon=_int_env("TIME_SERIES_FORECAST_HORIZON", 3), - time_series_min_edge_percent=_float_env("TIME_SERIES_MIN_EDGE_PERCENT", 0.04), + time_series_min_edge_percent=_float_env("TIME_SERIES_MIN_EDGE_PERCENT", 0.10), + time_series_min_probability_up=_float_env("TIME_SERIES_MIN_PROBABILITY_UP", 0.64), + time_series_min_confidence=_float_env("TIME_SERIES_MIN_CONFIDENCE", 0.72), time_series_max_adjustment=_float_env("TIME_SERIES_MAX_ADJUSTMENT", 0.08), time_series_lstm_enabled=_bool_env("TIME_SERIES_LSTM_ENABLED", True), time_series_lstm_model_path=Path(os.getenv("TIME_SERIES_LSTM_MODEL_PATH", "runtime/lstm_forecaster.json")), diff --git a/crypto_spot_bot/dashboard.py b/crypto_spot_bot/dashboard.py index f598d06..8224c3e 100644 --- a/crypto_spot_bot/dashboard.py +++ b/crypto_spot_bot/dashboard.py @@ -224,6 +224,8 @@ def _safe_config(settings: Settings) -> dict[str, Any]: "time_series_min_candles": settings.time_series_min_candles, "time_series_forecast_horizon": settings.time_series_forecast_horizon, "time_series_min_edge_percent": settings.time_series_min_edge_percent, + "time_series_min_probability_up": settings.time_series_min_probability_up, + "time_series_min_confidence": settings.time_series_min_confidence, "time_series_max_adjustment": settings.time_series_max_adjustment, "time_series_lstm_enabled": settings.time_series_lstm_enabled, "time_series_lstm_model_path": str(settings.time_series_lstm_model_path), @@ -1062,6 +1064,9 @@ HTML = r""" ['Лимит на пару', money(config.max_symbol_exposure_usdt)], ['Риск на сделку', `${num((config.risk_per_trade_percent || 0) * 100, 2)}% equity`], ['RSI входа', `${num(config.trend_rsi_min, 1)} - ${num(config.trend_rsi_max, 1)}`], + ['Torch min edge', `${num(config.time_series_min_edge_percent, 3)}%`], + ['Torch min P роста', `${num((config.time_series_min_probability_up || 0) * 100, 1)}%`], + ['Torch min confidence', `${num(config.time_series_min_confidence, 3)}`], ['Лимит в позициях', money(config.max_total_exposure_usdt)], ['Лимит позиций', `${config.max_open_positions} всего / ${config.max_positions_per_symbol} на пару`], ['Стоп', `${num(config.stop_loss_percent * 100, 2)}%`], diff --git a/crypto_spot_bot/strategy.py b/crypto_spot_bot/strategy.py index e3cd144..a4cf439 100644 --- a/crypto_spot_bot/strategy.py +++ b/crypto_spot_bot/strategy.py @@ -644,7 +644,7 @@ def _torch_forecast_entry_signal( "expected_edge_ok": expected_return >= min_edge, "probability_ok": probability_up >= min_probability, "skill_ok": skill > 0.0, - "confidence_ok": confidence >= settings.min_signal_confidence, + "confidence_ok": confidence >= settings.time_series_min_confidence, "spread_ok": spread_ok, "liquidity_ok": liquidity_ok, "risk_size_ok": position_notional >= settings.min_position_usdt, @@ -661,6 +661,7 @@ def _torch_forecast_entry_signal( "min_edge_percent": min_edge, "probability_up": probability_up, "min_probability_up": min_probability, + "min_confidence": settings.time_series_min_confidence, "skill": skill, "spread_percent": round(ticker.spread_percent, 5), "turnover_24h": ticker.turnover_24h, @@ -766,7 +767,7 @@ def _is_torch_forecast(forecast: dict) -> bool: def _torch_min_probability(settings: Settings) -> float: - return round(_clamp(settings.min_signal_confidence - 0.08, 0.52, 0.68), 4) + return round(_clamp(settings.time_series_min_probability_up, 0.45, 0.75), 4) def _torch_forecast_confidence(settings: Settings, forecast: dict) -> float: diff --git a/tests/conftest.py b/tests/conftest.py index 4100294..71a4779 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -72,7 +72,9 @@ def make_settings(): time_series_forecast_enabled=True, time_series_min_candles=120, time_series_forecast_horizon=3, - time_series_min_edge_percent=0.04, + time_series_min_edge_percent=0.10, + time_series_min_probability_up=0.64, + time_series_min_confidence=0.72, time_series_max_adjustment=0.08, time_series_lstm_enabled=True, time_series_lstm_model_path=tmp_path / "lstm_forecaster.json", diff --git a/tests/test_strategy.py b/tests/test_strategy.py index 818f6d6..b83ba4a 100644 --- a/tests/test_strategy.py +++ b/tests/test_strategy.py @@ -252,8 +252,8 @@ def test_torch_forecast_buys_only_from_positive_torch_edge(make_settings, tmp_pa forecast={ "usable": True, "model": "torch_gru", - "expected_return_percent": 0.24, - "probability_up": 0.63, + "expected_return_percent": 0.36, + "probability_up": 0.66, "skill": 0.22, "block_entry": False, }, diff --git a/tools/calibrate_torch_thresholds.py b/tools/calibrate_torch_thresholds.py new file mode 100644 index 0000000..3b6634a --- /dev/null +++ b/tools/calibrate_torch_thresholds.py @@ -0,0 +1,752 @@ +from __future__ import annotations + +import argparse +import json +import math +import sys +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +try: + import torch + + from tools.train_torch_recurrent_forecaster import RecurrentReturnModel +except ImportError: # pragma: no cover - local calibration can fall back to export inference. + torch = None # type: ignore[assignment] + RecurrentReturnModel = None # type: ignore[assignment] + +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, + _current_volatility_scale, + _entry_horizon, + _entry_output_layout, + _entry_target_horizons, + _feature_matrix, + _float_entry, + _log_returns, + _prediction_cap, + _select_horizon_prediction, + _target_vector, + _torch_recurrent_entry, + _torch_recurrent_model_name, + _torch_recurrent_predict, +) + + +@dataclass(slots=True) +class ForecastRecord: + symbol: str + index: int + timestamp: int + expected_percent: float + probability_up: float + confidence: float + skill: float + q50_percent: float + block_entry: bool + future_net_percent: float + + +@dataclass(slots=True) +class CalibrationResult: + edge: float + probability: float + confidence: float + trades: int + wins: int + win_rate: float + total_net_percent: float + average_net_percent: float + max_drawdown_percent: float + profit_factor: float + score: float + + +def main() -> None: + args = _parse_args() + if torch is not None and args.threads > 0: + torch.set_num_threads(args.threads) + settings = load_settings(args.env) + client = BybitClient(settings) + symbols = _symbols(args.symbols, settings.symbols) + context_symbols = sorted(set(symbols + _symbols(args.context_symbols, ()))) + artifact_path = Path(args.artifact or settings.time_series_lstm_model_path) + artifact = json.loads(artifact_path.read_text(encoding="utf-8")) + horizon = args.horizon if args.horizon > 0 else settings.time_series_forecast_horizon + round_trip_cost = _artifact_round_trip_cost(artifact, settings) + + market_candles: dict[str, list[Candle]] = {} + for symbol in context_symbols: + candles = _historical_klines(client, symbol, settings.base_interval, args.limit) + add_indicators(candles) + market_candles[symbol] = candles + print(f"{symbol}: loaded {len(candles)} {settings.base_interval} candles", flush=True) + + records: list[ForecastRecord] = [] + per_symbol_counts: dict[str, int] = {} + for symbol in symbols: + candles = market_candles.get(symbol) + if not candles: + continue + trend_candles = _historical_klines(client, symbol, settings.trend_interval, args.trend_limit) + add_indicators(trend_candles) + symbol_records = _forecast_records( + symbol=symbol, + candles=candles, + market_candles=market_candles, + trend_candles=trend_candles, + artifact=artifact, + horizon=horizon, + round_trip_cost=round_trip_cost, + min_candles=max(30, settings.time_series_min_candles), + calibration_window=args.calibration_window, + batch_size=args.batch_size, + ) + records.extend(symbol_records) + per_symbol_counts[symbol] = len(symbol_records) + print(f"{symbol}: replay records {len(symbol_records)}", flush=True) + + if not records: + raise SystemExit("No forecast records could be built for calibration.") + + results = _calibrate( + records, + edges=_float_grid(args.edge_grid), + probabilities=_float_grid(args.probability_grid), + confidences=_float_grid(args.confidence_grid), + min_trades=args.min_trades, + horizon=horizon, + ) + if not results: + raise SystemExit("No calibration result produced trades. Use wider grids or more history.") + + print("\nrecords_by_symbol", json.dumps(per_symbol_counts, ensure_ascii=False, sort_keys=True)) + print("artifact", json.dumps(_artifact_summary(artifact), ensure_ascii=False, sort_keys=True)) + print("\nTOP_RESULTS") + for result in results[: min(args.top, len(results))]: + print(_result_line(result)) + + recommended = _choose_recommendation(results, min_trades=args.min_trades) + print("\nRECOMMENDED") + print(_result_line(recommended)) + print( + "env " + f"TIME_SERIES_MIN_EDGE_PERCENT={recommended.edge:.4f} " + f"TIME_SERIES_MIN_PROBABILITY_UP={recommended.probability:.4f} " + f"TIME_SERIES_MIN_CONFIDENCE={recommended.confidence:.4f}" + ) + + if args.output: + payload = { + "artifact": _artifact_summary(artifact), + "records_by_symbol": per_symbol_counts, + "recommended": _result_dict(recommended), + "top_results": [_result_dict(result) for result in results[: args.top]], + } + Path(args.output).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Calibrate TradeBot Torch forecast entry thresholds.") + parser.add_argument("--env", default=None, help="Path to .env file.") + parser.add_argument("--artifact", default="", help="Path to lstm_forecaster.json.") + parser.add_argument("--symbols", default="", help="Comma-separated symbols. Defaults to configured fixed symbols.") + parser.add_argument("--context-symbols", default="BTCUSDT,ETHUSDT", help="Cross-asset context symbols.") + parser.add_argument("--limit", type=int, default=2000, help="Hourly candles per symbol.") + parser.add_argument("--trend-limit", type=int, default=320, help="Daily candles per symbol.") + parser.add_argument("--calibration-window", type=int, default=720, help="Tail records used for calibration.") + parser.add_argument("--horizon", type=int, default=0, help="Forecast horizon to calibrate.") + parser.add_argument("--min-trades", type=int, default=12, help="Minimum non-overlapping trades for recommendation.") + parser.add_argument("--edge-grid", default="0.00,0.02,0.04,0.05,0.06,0.08,0.10", help="Percent edge thresholds.") + parser.add_argument("--probability-grid", default="0.55,0.56,0.57,0.58,0.59,0.60,0.62,0.64,0.66,0.68,0.70", help="P(up) thresholds.") + parser.add_argument("--confidence-grid", default="0.50,0.56,0.60,0.64,0.68,0.72", help="Confidence thresholds.") + parser.add_argument("--top", type=int, default=15, help="How many top results to print and save.") + parser.add_argument("--output", default="", help="Optional JSON output path.") + parser.add_argument("--batch-size", type=int, default=256, help="Torch inference batch size.") + parser.add_argument("--threads", type=int, default=0, help="Torch CPU threads; 0 keeps torch default.") + return parser.parse_args() + + +def _symbols(raw: str, fallback: tuple[str, ...] | list[str]) -> list[str]: + if raw.strip(): + return [item.strip().upper() for item in raw.split(",") if item.strip()] + return [str(item).upper() for item in fallback] + + +def _forecast_records( + *, + symbol: str, + candles: list[Candle], + market_candles: dict[str, list[Candle]], + trend_candles: list[Candle], + artifact: dict[str, Any], + horizon: int, + round_trip_cost: float, + min_candles: int, + calibration_window: int, + batch_size: int, +) -> list[ForecastRecord]: + entry = _torch_recurrent_entry(symbol, artifact) + model = _torch_recurrent_model_name(symbol, artifact) + if not entry or not model: + return [] + feature_names = _feature_names(entry) + feature_rows = _feature_matrix( + candles, + feature_names, + symbol=symbol, + market_candles=market_candles, + trend_candles=trend_candles, + ) + closes = [float(candle.close) for candle in candles] + decision_horizon = _entry_horizon(entry, horizon) + start = max(min_candles, int(float(entry.get("lookback", 64)))) + end = len(candles) - decision_horizon - 1 + if calibration_window > 0: + start = max(start, end - calibration_window) + batched_records = _batch_forecast_records( + symbol=symbol, + candles=candles, + feature_rows=feature_rows, + closes=closes, + entry=entry, + model_name=model, + decision_horizon=decision_horizon, + round_trip_cost=round_trip_cost, + start=start, + end=end, + batch_size=batch_size, + ) + if batched_records is not None: + return batched_records + + records: list[ForecastRecord] = [] + skill = float(entry.get("skill", 0.0) or 0.0) + for index in range(start, max(start, end)): + prediction = _torch_recurrent_predict( + _log_returns(closes[: index + 1]), + symbol, + artifact, + feature_rows=feature_rows[: index + 1], + closes=closes[: index + 1], + candles=candles[: index + 1], + ) + if not isinstance(prediction, dict): + continue + selected = _select_horizon_prediction(prediction, decision_horizon) + if not selected: + continue + expected_return = float(selected.get("expected_return", 0.0)) + probability_up = _clamp(float(selected.get("probability_up", 0.5)), 0.0, 1.0) + q50 = float(selected.get("q50", expected_return)) + expected_percent = (math.exp(expected_return) - 1.0) * 100.0 + q50_percent = (math.exp(q50) - 1.0) * 100.0 + future_log_return = math.log(closes[index + decision_horizon] / closes[index]) - round_trip_cost + future_net_percent = (math.exp(future_log_return) - 1.0) * 100.0 + records.append( + ForecastRecord( + symbol=symbol, + index=index, + timestamp=candles[index].timestamp, + expected_percent=expected_percent, + probability_up=probability_up, + confidence=_forecast_confidence(expected_percent, probability_up, skill, 0.04), + skill=skill, + q50_percent=q50_percent, + block_entry=False, + future_net_percent=future_net_percent, + ) + ) + return records + + +def _batch_forecast_records( + *, + symbol: str, + candles: list[Candle], + feature_rows: list[list[float]], + closes: list[float], + entry: dict[str, Any], + model_name: str, + decision_horizon: int, + round_trip_cost: float, + start: int, + end: int, + batch_size: int, +) -> list[ForecastRecord] | None: + if torch is None or RecurrentReturnModel is None: + return None + horizons = _entry_target_horizons(entry) + if not horizons: + return None + model = _build_torch_model(entry, model_name) + if model is None: + return None + + lookback = int(_clamp(_float_entry(entry, "lookback", 64.0), 4.0, 512.0)) + clip = _clamp(_float_entry(entry, "clip", 8.0), 1.0, 50.0) + input_size = int(_clamp(_float_entry(entry, "input_size", len(feature_rows[-1]) if feature_rows else 1), 1.0, 256.0)) + means = _feature_vector(entry, "feature_means", input_size, 0.0) + scales = _feature_vector(entry, "feature_scales", input_size, 1.0) + indices = [ + index + for index in range(start, max(start, end)) + if index - lookback + 1 >= 0 and index + decision_horizon < len(closes) + ] + if not indices: + return [] + + records: list[ForecastRecord] = [] + skill = float(entry.get("skill", 0.0) or 0.0) + model.eval() + with torch.no_grad(): + for offset in range(0, len(indices), max(1, batch_size)): + batch_indices = indices[offset : offset + max(1, batch_size)] + windows = [ + _normalized_window( + feature_rows[index - lookback + 1 : index + 1], + means=means, + scales=scales, + input_size=input_size, + clip=clip, + ) + for index in batch_indices + ] + batch = torch.tensor(windows, dtype=torch.float32) + outputs = model(batch).detach().cpu().tolist() + for index, output in zip(batch_indices, outputs): + selected = _decode_selected_output( + output, + entry=entry, + candles=candles, + closes=closes, + index=index, + horizon=decision_horizon, + clip=clip, + round_trip_cost=round_trip_cost, + ) + if selected is None: + continue + expected_return = float(selected["expected_return"]) + probability_up = _clamp(float(selected["probability_up"]), 0.0, 1.0) + q50 = float(selected["q50"]) + expected_percent = (math.exp(expected_return) - 1.0) * 100.0 + q50_percent = (math.exp(q50) - 1.0) * 100.0 + future_log_return = math.log(closes[index + decision_horizon] / closes[index]) - round_trip_cost + future_net_percent = (math.exp(future_log_return) - 1.0) * 100.0 + records.append( + ForecastRecord( + symbol=symbol, + index=index, + timestamp=candles[index].timestamp, + expected_percent=expected_percent, + probability_up=probability_up, + confidence=_forecast_confidence(expected_percent, probability_up, skill, 0.04), + skill=skill, + q50_percent=q50_percent, + block_entry=False, + future_net_percent=future_net_percent, + ) + ) + return records + + +def _build_torch_model(entry: dict[str, Any], model_name: str) -> Any | None: + if torch is None or RecurrentReturnModel is None: + return None + architecture = "lstm" if model_name == "torch_lstm" else "gru" if model_name == "torch_gru" else "" + if not architecture: + return None + input_size = int(_clamp(_float_entry(entry, "input_size", 1.0), 1.0, 256.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)) + output_size = int(_clamp(_float_entry(entry, "output_size", 0.0), 1.0, 1024.0)) + model = RecurrentReturnModel( + architecture=architecture, + input_size=input_size, + hidden_size=hidden_size, + num_layers=num_layers, + dropout=0.0, + output_size=output_size, + attention_pooling=bool(entry.get("attention_pooling")), + context_norm=bool(entry.get("context_norm")), + ) + raw_state = entry.get("state_dict") + if not isinstance(raw_state, dict): + return None + state: dict[str, Any] = { + f"rnn.{key}": torch.tensor(value, dtype=torch.float32) + 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("attention_pooling")): + attention_weight = entry.get("attention_weight") + if not isinstance(attention_weight, list): + return None + state["attention.weight"] = torch.tensor([attention_weight], dtype=torch.float32) + state["attention.bias"] = torch.tensor([_float_entry(entry, "attention_bias", 0.0)], dtype=torch.float32) + if bool(entry.get("context_norm")): + context_weight = entry.get("context_norm_weight") + context_bias = entry.get("context_norm_bias") + if not isinstance(context_weight, list) or not isinstance(context_bias, list): + return None + state["context_norm.weight"] = torch.tensor(context_weight, dtype=torch.float32) + state["context_norm.bias"] = torch.tensor(context_bias, dtype=torch.float32) + try: + model.load_state_dict(state, strict=True) + except RuntimeError: + return None + return model + + +def _decode_selected_output( + output: list[float], + *, + entry: dict[str, Any], + candles: list[Candle], + closes: list[float], + index: int, + horizon: int, + clip: float, + round_trip_cost: float, +) -> dict[str, float] | None: + horizons = _entry_target_horizons(entry) + if not horizons: + return None + selected_horizon = horizon if horizon in horizons else min(horizons, key=lambda value: abs(value - horizon)) + horizon_index = horizons.index(selected_horizon) + layout = _entry_output_layout(entry) + group_size = len(layout) + base = horizon_index * group_size + if len(output) < base + group_size: + return None + values = {layout[offset]: float(output[base + offset]) for offset in range(group_size)} + target_means = _target_vector(entry, "target_means", "target_mean", len(horizons), 0.0) + target_scales = _target_vector(entry, "target_scales", "target_scale", len(horizons), 1.0) + history_closes = closes[: index + 1] + history_candles = candles[: index + 1] + volatility_scale = _current_volatility_scale(history_candles, history_closes, selected_horizon) + + def decode(name: str, fallback: float = 0.0) -> float: + normalized = _clamp(float(values.get(name, fallback)), -clip, clip) + transformed = normalized * max(target_scales[horizon_index], 1e-8) + target_means[horizon_index] + if str(entry.get("target_transform", "")) == "net_return_over_volatility": + return transformed * volatility_scale + return transformed + + 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) + return { + "expected_return": _clamp(expected, -cap, cap), + "q50": _clamp(q_values[1], -cap, cap), + "probability_up": _sigmoid(float(values.get("logit_up", 0.0))), + } + + +def _normalized_window( + rows: list[list[float]], + *, + means: list[float], + scales: list[float], + input_size: int, + clip: float, +) -> list[list[float]]: + return [ + [ + _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) + ] + for row in rows + ] + + +def _feature_vector(entry: dict[str, Any], key: str, size: int, default: float) -> list[float]: + raw = entry.get(key) + if isinstance(raw, list) and len(raw) == size: + return [float(value) for value in raw] + return [default for _ in range(size)] + + +def _calibrate( + records: list[ForecastRecord], + *, + edges: list[float], + probabilities: list[float], + confidences: list[float], + min_trades: int, + horizon: int, +) -> list[CalibrationResult]: + results: list[CalibrationResult] = [] + for edge in edges: + for probability in probabilities: + for confidence in confidences: + trades = _selected_trades(records, edge, probability, confidence, horizon) + if not trades: + continue + wins = sum(1 for value in trades if value > 0) + total = sum(trades) + average = total / len(trades) + max_drawdown = _max_drawdown(trades) + gross_profit = sum(value for value in trades if value > 0) + gross_loss = abs(sum(value for value in trades if value < 0)) + profit_factor = gross_profit / gross_loss if gross_loss > 0 else (999.0 if gross_profit > 0 else 0.0) + trade_factor = min(1.0, len(trades) / max(1, min_trades)) + score = average * trade_factor + total * 0.015 - max_drawdown * 0.03 + (wins / len(trades)) * 0.04 + results.append( + CalibrationResult( + edge=edge, + probability=probability, + confidence=confidence, + trades=len(trades), + wins=wins, + win_rate=wins / len(trades), + total_net_percent=total, + average_net_percent=average, + max_drawdown_percent=max_drawdown, + profit_factor=profit_factor, + score=score, + ) + ) + results.sort( + key=lambda item: ( + item.score, + item.average_net_percent, + item.total_net_percent, + item.profit_factor, + item.edge, + item.probability, + item.confidence, + ), + reverse=True, + ) + return results + + +def _selected_trades( + records: list[ForecastRecord], + edge: float, + probability: float, + confidence: float, + horizon: int, +) -> list[float]: + next_allowed_by_symbol: dict[str, int] = {} + trades: list[float] = [] + for record in sorted(records, key=lambda item: (item.timestamp, item.symbol)): + if record.index < next_allowed_by_symbol.get(record.symbol, -1): + continue + dynamic_confidence = _forecast_confidence( + record.expected_percent, + record.probability_up, + record.skill, + edge, + ) + block_entry = ( + record.expected_percent <= -edge + and record.probability_up <= 0.45 + ) or ( + record.q50_percent <= -edge + and record.probability_up <= 0.48 + ) + if ( + not block_entry + and record.expected_percent >= edge + and record.probability_up >= probability + and dynamic_confidence >= confidence + and record.skill > 0.0 + ): + trades.append(record.future_net_percent) + next_allowed_by_symbol[record.symbol] = record.index + max(1, horizon) + return trades + + +def _choose_recommendation(results: list[CalibrationResult], *, min_trades: int) -> CalibrationResult: + viable = [ + result + for result in results + if result.trades >= min_trades + and result.average_net_percent > 0 + and result.total_net_percent > 0 + and result.profit_factor >= 1.05 + ] + return viable[0] if viable else results[0] + + +def _forecast_confidence(expected_return: float, probability_up: float, skill: float, min_edge: float) -> float: + expected_return = max(0.0, expected_return) + skill = max(0.0, skill) + min_edge = max(0.01, min_edge) + edge_strength = _clamp(expected_return / max(min_edge * 4.0, 0.01), 0.0, 1.0) + probability_strength = _clamp((probability_up - 0.50) / 0.25, 0.0, 1.0) + skill_strength = _clamp(skill / 0.35, 0.0, 1.0) + confidence = 0.45 + probability_strength * 0.30 + edge_strength * 0.20 + skill_strength * 0.10 + return round(_clamp(confidence, 0.0, 0.96), 4) + + +def _max_drawdown(values: list[float]) -> float: + equity = 0.0 + peak = 0.0 + drawdown = 0.0 + for value in values: + equity += value + peak = max(peak, equity) + drawdown = max(drawdown, peak - equity) + return drawdown + + +def _artifact_round_trip_cost(artifact: dict[str, Any], settings: Any) -> float: + value = artifact.get("round_trip_cost") + if isinstance(value, (int, float)) and value >= 0: + return float(value) + return 2.0 * (float(settings.taker_fee_rate) + float(settings.slippage_rate)) + + +def _artifact_summary(artifact: dict[str, Any]) -> dict[str, Any]: + return { + "version": artifact.get("version"), + "created_at": artifact.get("created_at"), + "feature_count": artifact.get("feature_count"), + "target_horizon": artifact.get("target_horizon"), + "target_horizons": artifact.get("target_horizons"), + "target_transform": artifact.get("target_transform"), + "symbols": { + symbol: { + "model": row.get("model"), + "lookback": row.get("lookback"), + "hidden_size": row.get("hidden_size"), + "skill": row.get("skill"), + "directional_accuracy": row.get("directional_accuracy"), + } + for symbol, row in (artifact.get("symbols") or {}).items() + if isinstance(row, dict) + }, + } + + +def _result_line(result: CalibrationResult) -> str: + return ( + f"edge={result.edge:.4f} prob={result.probability:.4f} conf={result.confidence:.4f} " + f"trades={result.trades} win={result.win_rate:.3f} " + f"avg={result.average_net_percent:.4f}% total={result.total_net_percent:.4f}% " + f"dd={result.max_drawdown_percent:.4f}% pf={result.profit_factor:.3f} score={result.score:.4f}" + ) + + +def _result_dict(result: CalibrationResult) -> dict[str, Any]: + return { + "edge": result.edge, + "probability": result.probability, + "confidence": result.confidence, + "trades": result.trades, + "wins": result.wins, + "win_rate": result.win_rate, + "total_net_percent": result.total_net_percent, + "average_net_percent": result.average_net_percent, + "max_drawdown_percent": result.max_drawdown_percent, + "profit_factor": result.profit_factor, + "score": result.score, + } + + +def _feature_names(entry: dict[str, Any]) -> list[str]: + names = entry.get("feature_names") + if isinstance(names, list) and names: + return [str(name) for name in names] + return list(DEFAULT_TORCH_FEATURES) + + +def _historical_klines(client: BybitClient, symbol: str, interval: str, limit: int) -> list[Candle]: + limit = max(1, limit) + rows_by_timestamp: dict[int, Candle] = {} + end: int | None = None + while len(rows_by_timestamp) < limit: + page_limit = min(1000, limit - len(rows_by_timestamp)) + params: dict[str, Any] = { + "category": "spot", + "symbol": symbol, + "interval": interval, + "limit": page_limit, + } + if end is not None: + params["end"] = end + result = client.public_get("/v5/market/kline", params) + page = _parse_kline_rows(result.get("list", [])) + if not page: + break + for candle in page: + rows_by_timestamp[candle.timestamp] = candle + oldest = min(candle.timestamp for candle in page) + if end is not None and oldest >= end: + break + end = oldest - 1 + if len(page) < page_limit: + break + time.sleep(0.05) + return sorted(rows_by_timestamp.values(), key=lambda item: item.timestamp)[-limit:] + + +def _parse_kline_rows(rows: Any) -> list[Candle]: + candles: list[Candle] = [] + for row in rows or []: + if len(row) < 7: + continue + candles.append( + Candle( + timestamp=int(row[0]), + open=_float(row[1]), + high=_float(row[2]), + low=_float(row[3]), + close=_float(row[4]), + volume=_float(row[5]), + turnover=_float(row[6]), + ) + ) + candles.sort(key=lambda item: item.timestamp) + return candles + + +def _float_grid(raw: str) -> list[float]: + values = [] + for item in raw.split(","): + if item.strip(): + values.append(float(item.strip())) + return values + + +def _float(value: Any, default: float = 0.0) -> float: + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _sigmoid(value: float) -> float: + if value >= 40: + return 1.0 + if value <= -40: + return 0.0 + return 1.0 / (1.0 + math.exp(-value)) + + +def _clamp(value: float, low: float, high: float) -> float: + return max(low, min(high, value)) + + +if __name__ == "__main__": + main()