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 @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)) full_backtest = _full_backtest(records, recommended, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings) 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, ) print("\nWALK_FORWARD") print(json.dumps(walk_forward["summary"], 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, "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("--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.") parser.add_argument("--walk-forward-folds", type=int, default=4, help="Threshold walk-forward folds.") 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, 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, ) ) 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, 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, ) ) 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, ) -> 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 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 = ( record.close <= position["highest"] - record.atr * atr_multiplier if record.atr > 0 and position["highest"] > position["entry_price"] else False ) weak_forecast = ( record.expected_percent < thresholds.edge or record.probability_up < thresholds.probability or record.skill <= 0.0 ) exit_reason = "" if 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": rows[-50:]} def _walk_forward( records: list[ForecastRecord], *, edges: list[float], probabilities: list[float], confidences: list[float], min_trades: int, horizon: int, folds: int, ) -> 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] = [] 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_trades = _selected_trades(test, selected.edge, selected.probability, selected.confidence, horizon) all_test_trades.extend(test_trades) rows.append( { "fold": fold, "train_records": len(train), "test_records": len(test), "thresholds": _result_dict(selected), "test": _stats(test_trades), } ) summary = _stats(all_test_trades) summary["status"] = "ok" if summary["trades"] >= min_trades and summary["avg_net_percent"] > 0 else "warn" return {"summary": summary, "folds": rows} 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 _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 _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 _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()