Codex
753 lines
28 KiB
Python
753 lines
28 KiB
Python
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()
|