230 lines
7.5 KiB
Python
230 lines
7.5 KiB
Python
"""
|
|
策略参数优化器 — 网格搜索 + 响应面分析
|
|
|
|
核心功能:
|
|
- 给定策略类 + 参数空间,自动遍历所有组合
|
|
- 每组参数跑一次回测,收集关键指标
|
|
- 按指定指标排序,返回最优参数 + 完整响应面
|
|
- 支持错误恢复(0 信号/异常参数不会中断搜索)
|
|
|
|
用法:
|
|
from optimizer import StrategyOptimizer
|
|
from strategies.sma_cross import SmaCrossStrategy
|
|
|
|
opt = StrategyOptimizer(metric="sharpe_ratio")
|
|
result = opt.optimize(
|
|
strategy_class=SmaCrossStrategy,
|
|
df=df,
|
|
param_grid={"fast": [5,10,15], "slow": [20,30,40]},
|
|
)
|
|
print(result.best_params) # {"fast": 10, "slow": 30}
|
|
result.export("optimization.csv")
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import itertools
|
|
import os
|
|
from typing import Type
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import raptorbt
|
|
|
|
from strategies.base import Strategy
|
|
|
|
# 这些指标的值越小越好
|
|
_MINIMIZE_METRICS = {"max_drawdown_pct"}
|
|
|
|
|
|
class OptimizationResult:
|
|
"""优化结果容器"""
|
|
|
|
def __init__(self, results: pd.DataFrame, metric: str, direction: str, param_names: list):
|
|
self.results = results
|
|
self.metric = metric
|
|
self.direction = direction
|
|
self.param_names = param_names
|
|
|
|
# 找最优行
|
|
valid = results[results[metric].notna()]
|
|
if len(valid) == 0:
|
|
self.best_params = {}
|
|
self.best_score = np.nan
|
|
return
|
|
|
|
if direction == "maximize":
|
|
best_idx = valid[metric].idxmax()
|
|
else:
|
|
best_idx = valid[metric].idxmin()
|
|
best_row = results.loc[best_idx]
|
|
self.best_params = {p: best_row[p] for p in param_names}
|
|
self.best_score = best_row[metric]
|
|
|
|
def top_n(self, n=10) -> pd.DataFrame:
|
|
"""返回前 N 个参数组合"""
|
|
valid = self.results[self.results[self.metric].notna()]
|
|
if len(valid) == 0:
|
|
return valid
|
|
if self.direction == "maximize":
|
|
return valid.nlargest(n, self.metric)
|
|
return valid.nsmallest(n, self.metric)
|
|
|
|
def export(self, path: str):
|
|
"""导出完整结果到 CSV"""
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
self.results.to_csv(path, index=False, encoding="utf-8-sig")
|
|
|
|
def summary(self) -> str:
|
|
params_str = ", ".join(f"{k}={v}" for k, v in self.best_params.items())
|
|
n = len(self.results)
|
|
valid = self.results[self.metric].notna().sum()
|
|
failed = n - valid
|
|
return (
|
|
f"优化完成: {n} 个组合, {valid} 个有效, {failed} 个失败\n"
|
|
f"最优参数: {params_str}\n"
|
|
f"最优 {self.metric}: {self.best_score:.4f}"
|
|
)
|
|
|
|
def to_dict(self) -> dict:
|
|
"""转为可 JSON 序列化的字典 (含最优参数 + Top 10 组合)"""
|
|
top = self.top_n(10)
|
|
# 处理 NaN, 让 json.dumps 能序列化
|
|
top_records = top.to_dict(orient="records")
|
|
return {
|
|
"metric": self.metric,
|
|
"direction": self.direction,
|
|
"n_combinations": len(self.results),
|
|
"n_valid": int(self.results[self.metric].notna().sum()),
|
|
"n_failed": int(self.results[self.metric].isna().sum()),
|
|
"best_params": self.best_params,
|
|
"best_score": None if pd.isna(self.best_score) else self.best_score,
|
|
"top_10": [
|
|
{k: (None if pd.isna(v) else v) for k, v in r.items()}
|
|
for r in top_records
|
|
],
|
|
}
|
|
|
|
|
|
class StrategyOptimizer:
|
|
"""
|
|
策略参数网格搜索优化器
|
|
|
|
参数:
|
|
metric: 优化的目标指标 (如 sharpe_ratio, total_return_pct)
|
|
direction: "maximize" 或 "minimize", None 时自动推断
|
|
"""
|
|
|
|
def __init__(self, metric: str = "sharpe_ratio", direction: str | None = None):
|
|
self.metric = metric
|
|
if direction is None:
|
|
direction = "minimize" if metric in _MINIMIZE_METRICS else "maximize"
|
|
self.direction = direction
|
|
|
|
def optimize(
|
|
self,
|
|
strategy_class: Type[Strategy],
|
|
df: pd.DataFrame,
|
|
param_grid: dict,
|
|
symbol: str = "OPT",
|
|
verbose: bool = True,
|
|
) -> OptimizationResult:
|
|
"""
|
|
执行网格搜索
|
|
|
|
参数:
|
|
strategy_class: Strategy 子类
|
|
df: K 线数据 DataFrame
|
|
param_grid: {参数名: [可选值列表]}
|
|
symbol: 回测标的标签
|
|
verbose: 是否打印进度
|
|
|
|
返回:
|
|
OptimizationResult
|
|
"""
|
|
param_names = list(param_grid.keys())
|
|
param_values = list(param_grid.values())
|
|
combinations = list(itertools.product(*param_values))
|
|
total = len(combinations)
|
|
|
|
results = []
|
|
for i, combo in enumerate(combinations):
|
|
params = dict(zip(param_names, combo))
|
|
row = self._run_single(strategy_class, params, df, symbol)
|
|
row["combo_id"] = i
|
|
results.append(row)
|
|
|
|
if verbose and ((i + 1) % 50 == 0 or i + 1 == total):
|
|
print(f" 进度: {i+1}/{total} ({100*(i+1)/total:.0f}%)")
|
|
|
|
results_df = pd.DataFrame(results)
|
|
return OptimizationResult(results_df, self.metric, self.direction, param_names)
|
|
|
|
def _run_single(self, strategy_class, params: dict, df, symbol) -> dict:
|
|
"""执行单次回测,返回指标字典"""
|
|
row = {**params, "error": ""}
|
|
|
|
# 实例化策略
|
|
try:
|
|
strategy = strategy_class(**params)
|
|
except Exception as e:
|
|
row[self.metric] = np.nan
|
|
row["error"] = f"init: {e}"
|
|
row["total_trades"] = 0
|
|
return row
|
|
|
|
# 检查预热期
|
|
warmup = strategy.warmup_bars()
|
|
if warmup >= len(df):
|
|
row[self.metric] = np.nan
|
|
row["error"] = "warmup >= data"
|
|
row["total_trades"] = 0
|
|
return row
|
|
|
|
# 生成信号
|
|
try:
|
|
signals = strategy.generate_signals(df)
|
|
except Exception as e:
|
|
row[self.metric] = np.nan
|
|
row["error"] = f"signals: {e}"
|
|
row["total_trades"] = 0
|
|
return row
|
|
|
|
n_entries = int(signals.entries.sum())
|
|
row["n_entries"] = n_entries
|
|
if n_entries == 0:
|
|
row[self.metric] = np.nan
|
|
row["error"] = "0 signals"
|
|
row["total_trades"] = 0
|
|
return row
|
|
|
|
# 执行回测
|
|
try:
|
|
config = strategy.build_config()
|
|
arr = Strategy.to_arrays(df)
|
|
result = raptorbt.run_single_backtest(
|
|
timestamps=arr["timestamps"],
|
|
open=arr["open"], high=arr["high"], low=arr["low"], close=arr["close"],
|
|
volume=arr["volume"],
|
|
entries=signals.entries, exits=signals.exits,
|
|
direction=signals.direction, weight=1.0, symbol=symbol,
|
|
config=config,
|
|
)
|
|
m = result.metrics
|
|
row[self.metric] = getattr(m, self.metric, np.nan)
|
|
# 收集常用指标
|
|
row["total_trades"] = m.total_trades
|
|
row["total_return_pct"] = m.total_return_pct
|
|
row["sharpe_ratio"] = m.sharpe_ratio
|
|
row["sortino_ratio"] = getattr(m, "sortino_ratio", np.nan)
|
|
row["max_drawdown_pct"] = m.max_drawdown_pct
|
|
row["win_rate_pct"] = m.win_rate_pct
|
|
row["profit_factor"] = m.profit_factor
|
|
row["expectancy"] = getattr(m, "expectancy", np.nan)
|
|
except Exception as e:
|
|
row[self.metric] = np.nan
|
|
row["error"] = f"backtest: {e}"
|
|
row["total_trades"] = 0
|
|
|
|
return row
|