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 close: float atr: float expected_percent: float probability_up: float confidence: float skill: float q50_percent: float block_entry: bool future_net_percent: float benchmark_entry: bool benchmark_exit: bool @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, full_backtest = _choose_replay_recommendation( results, records, min_trades=args.min_trades, min_full_replay_trades=args.min_full_replay_trades, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings, ) print("\nRECOMMENDED") print(_result_line(recommended)) print("\nFULL_REPLAY") print(_stats_line(full_backtest)) walk_forward = _walk_forward( 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, folds=args.walk_forward_folds, round_trip_cost=round_trip_cost, settings=settings, ) benchmark = _benchmark_walk_forward( records, horizon=horizon, folds=args.walk_forward_folds, round_trip_cost=round_trip_cost, settings=settings, ) validation = _quality_gate( walk_forward=walk_forward, benchmark=benchmark, min_oos_trades=args.min_oos_trades, min_oos_symbols=args.min_oos_symbols, max_symbol_share=args.max_oos_symbol_share, min_oos_folds=args.min_oos_folds_with_trades, min_profit_factor=args.min_oos_profit_factor, min_benchmark_edge=args.min_benchmark_edge_percent, ) print("\nWALK_FORWARD") print(json.dumps(walk_forward["summary"], ensure_ascii=False, sort_keys=True)) print("\nBENCHMARK") print(json.dumps(benchmark["summary"], ensure_ascii=False, sort_keys=True)) print("\nQUALITY_GATE") print(json.dumps(validation, ensure_ascii=False, sort_keys=True)) 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), "full_replay": full_backtest, "walk_forward": walk_forward, "benchmark": benchmark, "validation": validation, "probability_calibration": _probability_calibration(records), "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("--min-full-replay-trades", type=int, default=8, help="Prefer recommendations with at least this many full replay trades.") 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.45,0.47,0.50,0.52,0.54,0.55,0.56,0.58,0.60,0.62,0.64,0.66,0.68,0.70", help="P(up) thresholds.") parser.add_argument("--confidence-grid", default="0.40", 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.") parser.add_argument("--walk-forward-folds", type=int, default=8, help="Threshold walk-forward folds.") parser.add_argument("--min-oos-trades", type=int, default=30, help="Minimum out-of-sample walk-forward trades for a valid model.") parser.add_argument("--min-oos-symbols", type=int, default=2, help="Minimum symbols with out-of-sample trades.") parser.add_argument("--max-oos-symbol-share", type=float, default=0.75, help="Reject if one symbol contributes more than this share of out-of-sample trades.") parser.add_argument("--min-oos-folds-with-trades", type=int, default=2, help="Minimum walk-forward folds that must produce trades.") parser.add_argument("--min-oos-profit-factor", type=float, default=1.10, help="Minimum out-of-sample profit factor.") parser.add_argument("--min-benchmark-edge-percent", type=float, default=0.0, help="Required total-net percent advantage over the benchmark.") 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, trend_candles=trend_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, close=closes[index], atr=float(candles[index].atr_14 or 0.0), 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, benchmark_entry=_benchmark_entry_signal(candles, trend_candles, index), benchmark_exit=_benchmark_exit_signal(candles, index), ) ) return records def _batch_forecast_records( *, symbol: str, candles: list[Candle], trend_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, close=closes[index], atr=float(candles[index].atr_14 or 0.0), 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, benchmark_entry=_benchmark_entry_signal(candles, trend_candles, index), benchmark_exit=_benchmark_exit_signal(candles, index), ) ) 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 _full_backtest( records: list[ForecastRecord], thresholds: CalibrationResult, *, horizon: int, round_trip_cost: float, settings: Any, detail_limit: int = 50, ) -> dict[str, Any]: positions: dict[str, dict[str, Any]] = {} trades: list[float] = [] rows: list[dict[str, Any]] = [] max_hold = max(12, horizon * 8) stop_loss_percent = max(0.003, min(0.08, float(settings.stop_loss_percent))) * 100.0 stop_loss_exit_enabled = bool(getattr(settings, "stop_loss_exit_enabled", True)) atr_multiplier = max(0.5, min(10.0, float(settings.atr_trailing_multiplier))) for record in sorted(records, key=lambda item: (item.timestamp, item.symbol)): position = positions.get(record.symbol) if position is not None: position["highest"] = max(position["highest"], record.close) net_percent = _net_percent(position["entry_price"], record.close, round_trip_cost) held = record.index - int(position["entry_index"]) atr_stop_level = ( position["highest"] - record.atr * atr_multiplier if record.atr > 0 and position["highest"] > position["entry_price"] else None ) atr_stop = bool( atr_stop_level is not None and record.close <= atr_stop_level and (stop_loss_exit_enabled or atr_stop_level > position["entry_price"]) ) weak_forecast = ( record.expected_percent < thresholds.edge or record.probability_up < thresholds.probability or record.skill <= 0.0 ) exit_reason = "" if stop_loss_exit_enabled and net_percent <= -stop_loss_percent: exit_reason = "stop_loss" elif atr_stop: exit_reason = "atr_trailing_stop" elif (record.expected_percent <= 0.0 or record.probability_up <= 0.50 or _candidate_blocks(record, thresholds.edge)): exit_reason = "forecast_negative" elif weak_forecast and net_percent >= 0: exit_reason = "forecast_weak_profit_lock" elif held >= max_hold: exit_reason = "max_hold" if exit_reason: trades.append(net_percent) rows.append( { "symbol": record.symbol, "entry_timestamp": position["timestamp"], "exit_timestamp": record.timestamp, "net_percent": round(net_percent, 4), "reason": exit_reason, "held_bars": held, "entry_probability": round(float(position["probability_up"]), 4), "entry_expected_percent": round(float(position["expected_percent"]), 4), } ) positions.pop(record.symbol, None) continue if record.symbol in positions: continue if _candidate_allows(record, thresholds.edge, thresholds.probability, thresholds.confidence): positions[record.symbol] = { "entry_price": record.close, "entry_index": record.index, "timestamp": record.timestamp, "highest": record.close, "probability_up": record.probability_up, "expected_percent": record.expected_percent, } for symbol, position in list(positions.items()): tail = next((record for record in reversed(records) if record.symbol == symbol), None) if tail is None: continue net_percent = _net_percent(position["entry_price"], tail.close, round_trip_cost) trades.append(net_percent) rows.append( { "symbol": symbol, "entry_timestamp": position["timestamp"], "exit_timestamp": tail.timestamp, "net_percent": round(net_percent, 4), "reason": "end_of_replay", "held_bars": tail.index - int(position["entry_index"]), "entry_probability": round(float(position["probability_up"]), 4), "entry_expected_percent": round(float(position["expected_percent"]), 4), } ) return { **_stats(trades), "trades_detail": _limited_rows(rows, detail_limit), "symbol_breakdown": _symbol_breakdown(rows), } def _benchmark_backtest( records: list[ForecastRecord], *, horizon: int, round_trip_cost: float, settings: Any, detail_limit: int = 50, ) -> dict[str, Any]: positions: dict[str, dict[str, Any]] = {} trades: list[float] = [] rows: list[dict[str, Any]] = [] max_hold = max(12, horizon * 8) stop_loss_percent = max(0.003, min(0.08, float(settings.stop_loss_percent))) * 100.0 stop_loss_exit_enabled = bool(getattr(settings, "stop_loss_exit_enabled", True)) atr_multiplier = max(0.5, min(10.0, float(settings.atr_trailing_multiplier))) for record in sorted(records, key=lambda item: (item.timestamp, item.symbol)): position = positions.get(record.symbol) if position is not None: position["highest"] = max(position["highest"], record.close) net_percent = _net_percent(position["entry_price"], record.close, round_trip_cost) held = record.index - int(position["entry_index"]) atr_stop_level = ( position["highest"] - record.atr * atr_multiplier if record.atr > 0 and position["highest"] > position["entry_price"] else None ) atr_stop = bool( atr_stop_level is not None and record.close <= atr_stop_level and (stop_loss_exit_enabled or atr_stop_level > position["entry_price"]) ) exit_reason = "" if stop_loss_exit_enabled and net_percent <= -stop_loss_percent: exit_reason = "stop_loss" elif atr_stop: exit_reason = "atr_trailing_stop" elif record.benchmark_exit: exit_reason = "benchmark_exit" elif held >= max_hold: exit_reason = "max_hold" if exit_reason: trades.append(net_percent) rows.append( { "symbol": record.symbol, "entry_timestamp": position["timestamp"], "exit_timestamp": record.timestamp, "net_percent": round(net_percent, 4), "reason": exit_reason, "held_bars": held, } ) positions.pop(record.symbol, None) continue if record.symbol in positions: continue if record.benchmark_entry: positions[record.symbol] = { "entry_price": record.close, "entry_index": record.index, "timestamp": record.timestamp, "highest": record.close, } for symbol, position in list(positions.items()): tail = next((record for record in reversed(records) if record.symbol == symbol), None) if tail is None: continue net_percent = _net_percent(position["entry_price"], tail.close, round_trip_cost) trades.append(net_percent) rows.append( { "symbol": symbol, "entry_timestamp": position["timestamp"], "exit_timestamp": tail.timestamp, "net_percent": round(net_percent, 4), "reason": "end_of_replay", "held_bars": tail.index - int(position["entry_index"]), } ) return { **_stats(trades), "trades_detail": _limited_rows(rows, detail_limit), "symbol_breakdown": _symbol_breakdown(rows), } def _walk_forward( records: list[ForecastRecord], *, edges: list[float], probabilities: list[float], confidences: list[float], min_trades: int, horizon: int, folds: int, round_trip_cost: float, settings: Any, ) -> dict[str, Any]: ordered = sorted(records, key=lambda item: item.timestamp) if folds < 2 or len(ordered) < folds * 20: return {"summary": {"status": "insufficient"}, "folds": []} timestamps = sorted({record.timestamp for record in ordered}) fold_size = max(1, len(timestamps) // folds) rows = [] all_test_trades: list[float] = [] all_test_rows: list[dict[str, Any]] = [] for fold in range(1, folds): test_start = timestamps[fold * fold_size] test_end = timestamps[(fold + 1) * fold_size - 1] if fold < folds - 1 else timestamps[-1] train = [record for record in ordered if record.timestamp < test_start] test = [record for record in ordered if test_start <= record.timestamp <= test_end] train_results = _calibrate( train, edges=edges, probabilities=probabilities, confidences=confidences, min_trades=max(4, min_trades // 2), horizon=horizon, ) if not train_results: continue selected = _choose_recommendation(train_results, min_trades=max(4, min_trades // 2)) test_backtest = _full_backtest( test, selected, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings, detail_limit=0, ) test_rows = test_backtest.get("trades_detail", []) test_trades = [float(row.get("net_percent", 0.0) or 0.0) for row in test_rows if isinstance(row, dict)] all_test_trades.extend(test_trades) all_test_rows.extend(test_rows) rows.append( { "fold": fold, "train_records": len(train), "test_records": len(test), "thresholds": _result_dict(selected), "test": {key: value for key, value in test_backtest.items() if key != "trades_detail"}, } ) summary = _stats(all_test_trades) summary["status"] = "ok" if summary["trades"] >= min_trades and summary["avg_net_percent"] > 0 else "warn" return { "summary": summary, "symbol_breakdown": _symbol_breakdown(all_test_rows), "folds_with_trades": sum(1 for row in rows if int((row.get("test") or {}).get("trades", 0) or 0) > 0), "folds": rows, } def _benchmark_walk_forward( records: list[ForecastRecord], *, horizon: int, folds: int, round_trip_cost: float, settings: Any, ) -> dict[str, Any]: ordered = sorted(records, key=lambda item: item.timestamp) if folds < 2 or len(ordered) < folds * 20: return {"summary": {"status": "insufficient"}, "symbol_breakdown": [], "folds_with_trades": 0, "folds": []} timestamps = sorted({record.timestamp for record in ordered}) fold_size = max(1, len(timestamps) // folds) rows = [] all_test_rows: list[dict[str, Any]] = [] for fold in range(1, folds): test_start = timestamps[fold * fold_size] test_end = timestamps[(fold + 1) * fold_size - 1] if fold < folds - 1 else timestamps[-1] test = [record for record in ordered if test_start <= record.timestamp <= test_end] test_backtest = _benchmark_backtest( test, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings, detail_limit=0, ) test_rows = test_backtest.get("trades_detail", []) all_test_rows.extend(test_rows) rows.append( { "fold": fold, "test_records": len(test), "test": {key: value for key, value in test_backtest.items() if key != "trades_detail"}, } ) trades = [float(row.get("net_percent", 0.0) or 0.0) for row in all_test_rows if isinstance(row, dict)] summary = _stats(trades) summary["status"] = "ok" if summary["trades"] > 0 else "no_trades" return { "name": "trend_macd_baseline", "summary": summary, "symbol_breakdown": _symbol_breakdown(all_test_rows), "folds_with_trades": sum(1 for row in rows if int((row.get("test") or {}).get("trades", 0) or 0) > 0), "folds": rows, } def _quality_gate( *, walk_forward: dict[str, Any], benchmark: dict[str, Any], min_oos_trades: int, min_oos_symbols: int, max_symbol_share: float, min_oos_folds: int, min_profit_factor: float, min_benchmark_edge: float, ) -> dict[str, Any]: summary = walk_forward.get("summary") if isinstance(walk_forward.get("summary"), dict) else {} benchmark_summary = benchmark.get("summary") if isinstance(benchmark.get("summary"), dict) else {} breakdown = walk_forward.get("symbol_breakdown") if isinstance(walk_forward.get("symbol_breakdown"), list) else [] symbols_with_trades = sum(1 for row in breakdown if int(row.get("trades", 0) or 0) > 0) max_share = max((float(row.get("trade_share", 0.0) or 0.0) for row in breakdown), default=0.0) oos_total = float(summary.get("total_net_percent", 0.0) or 0.0) benchmark_total = float(benchmark_summary.get("total_net_percent", 0.0) or 0.0) checks = [ _gate_check("oos_trades", int(summary.get("trades", 0) or 0), min_oos_trades, int(summary.get("trades", 0) or 0) >= min_oos_trades), _gate_check("oos_symbols", symbols_with_trades, min_oos_symbols, symbols_with_trades >= min_oos_symbols), _gate_check("max_symbol_share", round(max_share, 4), max_symbol_share, max_share <= max_symbol_share if breakdown else False), _gate_check("folds_with_trades", int(walk_forward.get("folds_with_trades", 0) or 0), min_oos_folds, int(walk_forward.get("folds_with_trades", 0) or 0) >= min_oos_folds), _gate_check("oos_avg_net_positive", float(summary.get("avg_net_percent", 0.0) or 0.0), "> 0", float(summary.get("avg_net_percent", 0.0) or 0.0) > 0), _gate_check("oos_profit_factor", float(summary.get("profit_factor", 0.0) or 0.0), min_profit_factor, float(summary.get("profit_factor", 0.0) or 0.0) >= min_profit_factor), _gate_check("beats_benchmark_total", round(oos_total - benchmark_total, 4), min_benchmark_edge, (oos_total - benchmark_total) > min_benchmark_edge), ] passed = all(bool(row["passed"]) for row in checks) return { "status": "pass" if passed else "fail", "passed": passed, "checks": checks, "oos_summary": summary, "benchmark_summary": benchmark_summary, "symbol_breakdown": breakdown, } def _gate_check(name: str, value: Any, required: Any, passed: bool) -> dict[str, Any]: return {"name": name, "value": value, "required": required, "passed": bool(passed)} def _probability_calibration(records: list[ForecastRecord]) -> dict[str, Any]: buckets: dict[str, list[ForecastRecord]] = {} for record in records: low = max(0.0, min(0.95, int(record.probability_up * 20) / 20)) key = f"{low:.2f}-{low + 0.05:.2f}" buckets.setdefault(key, []).append(record) rows = [] for key in sorted(buckets): items = buckets[key] wins = sum(1 for item in items if item.future_net_percent > 0) avg_probability = sum(item.probability_up for item in items) / len(items) rows.append( { "bucket": key, "samples": len(items), "avg_probability": round(avg_probability, 4), "actual_win_rate": round(wins / len(items), 4), "avg_future_net_percent": round(sum(item.future_net_percent for item in items) / len(items), 4), } ) return {"samples": len(records), "buckets": rows} def _candidate_blocks(record: ForecastRecord, edge: float) -> bool: return ( record.expected_percent <= -edge and record.probability_up <= 0.45 ) or ( record.q50_percent <= -edge and record.probability_up <= 0.48 ) def _candidate_allows(record: ForecastRecord, edge: float, probability: float, confidence: float) -> bool: dynamic_confidence = _forecast_confidence(record.expected_percent, record.probability_up, record.skill, edge) return ( not _candidate_blocks(record, edge) and record.expected_percent >= edge and record.probability_up >= probability and dynamic_confidence >= confidence and record.skill > 0.0 ) def _benchmark_entry_signal(candles: list[Candle], trend_candles: list[Candle], index: int) -> bool: if index <= 0 or index >= len(candles): return False previous = candles[index - 1] current = candles[index] rsi = current.rsi_14 return bool( _daily_trend_ok(trend_candles, current.timestamp) and _macd_crossed_up(previous, current) and current.ema_50 is not None and current.close > current.ema_50 and rsi is not None and 45.0 <= rsi <= 65.0 ) def _benchmark_exit_signal(candles: list[Candle], index: int) -> bool: if index <= 0 or index >= len(candles): return False previous = candles[index - 1] current = candles[index] return bool(_macd_crossed_down(previous, current) or (current.ema_50 is not None and current.close < current.ema_50)) def _daily_trend_ok(trend_candles: list[Candle], timestamp: int) -> bool: for candle in reversed(trend_candles): if candle.timestamp > timestamp: continue return bool( candle.ema_50 is not None and candle.ema_200 is not None and candle.close > candle.ema_200 and candle.ema_50 > candle.ema_200 ) return False def _macd_crossed_up(previous: Candle, current: Candle) -> bool: if None in (previous.macd, previous.macd_signal, current.macd, current.macd_signal): return False return bool(previous.macd <= previous.macd_signal and current.macd > current.macd_signal) def _macd_crossed_down(previous: Candle, current: Candle) -> bool: if None in (previous.macd, previous.macd_signal, current.macd, current.macd_signal): return False return bool(previous.macd >= previous.macd_signal and current.macd < current.macd_signal) def _net_percent(entry_price: float, exit_price: float, round_trip_cost: float) -> float: if entry_price <= 0 or exit_price <= 0: return 0.0 return (math.exp(math.log(exit_price / entry_price) - round_trip_cost) - 1.0) * 100.0 def _limited_rows(rows: list[dict[str, Any]], detail_limit: int) -> list[dict[str, Any]]: if detail_limit <= 0: return rows return rows[-detail_limit:] def _symbol_breakdown(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: by_symbol: dict[str, list[float]] = {} for row in rows: symbol = str(row.get("symbol", "")) if not symbol: continue by_symbol.setdefault(symbol, []).append(float(row.get("net_percent", 0.0) or 0.0)) total_trades = sum(len(values) for values in by_symbol.values()) result = [] for symbol in sorted(by_symbol): values = by_symbol[symbol] stats = _stats(values) result.append( { "symbol": symbol, **stats, "trade_share": round(len(values) / total_trades, 4) if total_trades else 0.0, } ) return result def _stats(values: list[float]) -> dict[str, Any]: wins = sum(1 for value in values if value > 0) total = sum(values) gross_profit = sum(value for value in values if value > 0) gross_loss = abs(sum(value for value in values if value < 0)) profit_factor = gross_profit / gross_loss if gross_loss > 0 else (999.0 if gross_profit > 0 else 0.0) return { "trades": len(values), "wins": wins, "win_rate": round(wins / len(values), 4) if values else 0.0, "total_net_percent": round(total, 4), "avg_net_percent": round(total / len(values), 4) if values else 0.0, "max_drawdown_percent": round(_max_drawdown(values), 4), "profit_factor": round(profit_factor, 4), } 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 if _candidate_allows(record, edge, probability, confidence): 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 _choose_replay_recommendation( results: list[CalibrationResult], records: list[ForecastRecord], *, min_trades: int, min_full_replay_trades: int, horizon: int, round_trip_cost: float, settings: Any, ) -> tuple[CalibrationResult, dict[str, Any]]: fallback = _choose_recommendation(results, min_trades=min_trades) fallback_replay = _full_backtest(records, fallback, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings) if min_full_replay_trades <= 0: return fallback, fallback_replay viable: list[tuple[CalibrationResult, dict[str, Any]]] = [] for result in results: if result.trades < min(4, min_trades): continue if result.average_net_percent <= 0 or result.total_net_percent <= 0 or result.profit_factor < 1.05: continue replay = _full_backtest(records, result, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings) if int(replay.get("trades", 0) or 0) < min_full_replay_trades: continue if float(replay.get("avg_net_percent", 0.0) or 0.0) <= 0: continue if float(replay.get("profit_factor", 0.0) or 0.0) < 1.05: continue viable.append((result, replay)) if not viable: return fallback, fallback_replay viable.sort( key=lambda item: ( item[0].score, float(item[1].get("avg_net_percent", 0.0) or 0.0), float(item[1].get("total_net_percent", 0.0) or 0.0), int(item[1].get("trades", 0) or 0), item[0].confidence, ), reverse=True, ) return viable[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 _stats_line(stats: dict[str, Any]) -> str: return ( f"trades={stats.get('trades', 0)} win={stats.get('win_rate', 0):.3f} " f"avg={stats.get('avg_net_percent', 0):.4f}% total={stats.get('total_net_percent', 0):.4f}% " f"dd={stats.get('max_drawdown_percent', 0):.4f}% pf={stats.get('profit_factor', 0):.3f}" ) 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()