""" 策略参数优化器 — 网格搜索 + 响应面分析 核心功能: - 给定策略类 + 参数空间,自动遍历所有组合 - 每组参数跑一次回测,收集关键指标 - 按指定指标排序,返回最优参数 + 完整响应面 - 支持错误恢复(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