simonandClaude Opus 4.7
271a9343a5
7 Sprints 全部完成: Sprint 0: 基础设施 (DataManager + MariaDB) Sprint 1: 因子引擎 (34因子/12分类) Sprint 2: VectorBT 回测 (5策略+截面) Sprint 3: Optuna 优化 (+Walk-Forward) Sprint 4: ML 模型 (LightGBM+CatBoost) Sprint 5: Qwen 情绪因子 (三源新闻+日期对齐) Sprint 6: Agent 系统 (4Agent+日报.md/.html) 生产加固 (15项): Tushare双源fallback, SSH自动恢复, pool_pre_ping, save_daily先删后插, load_dotenv绝对路径, 日报5d/20d修复, RiskAgent改上证指数, 昨日对比+数据截止, mac_report utf8mb4, CLAUDE-*.md 9条已知Bug, demo全参数化, djapi数据源归一化, indexDatas API修正 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
"""
|
|
Optuna 目标函数。
|
|
|
|
将策略实例化 → 回测 → 提取指标，包装为 Optuna objective。
|
|
"""
|
|
|
|
import optuna
|
|
import pandas as pd
|
|
|
|
from backtest.base import BaseStrategy
|
|
from backtest.vectorbt.engine import VectorBTEngine
|
|
from optimizer.space import SearchSpace
|
|
|
|
# 指标提取器：从 BacktestReport 取对应字段
|
|
_METRIC_EXTRACTORS = {
|
|
"sharpe": lambda r: r.sharpe_ratio,
|
|
"cagr": lambda r: r.cagr,
|
|
"calmar": lambda r: r.calmar_ratio,
|
|
"total_return": lambda r: r.total_return,
|
|
"return_over_dd": lambda r: abs(r.total_return / r.max_drawdown) if r.max_drawdown != 0 else 0.0,
|
|
"win_rate": lambda r: r.win_rate,
|
|
"profit_factor": lambda r: r.profit_factor,
|
|
}
|
|
|
|
|
|
class Objective:
|
|
"""
|
|
Optuna 目标函数（可调用）。
|
|
|
|
用法:
|
|
obj = Objective(SMACrossStrategy, sma_cross_space, price_df, factor_df, metric="sharpe")
|
|
study = optuna.create_study(direction="maximize")
|
|
study.optimize(obj, n_trials=100)
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
strategy_class: type[BaseStrategy],
|
|
search_space: SearchSpace,
|
|
price_df: pd.DataFrame,
|
|
factor_df: pd.DataFrame | None = None,
|
|
bt_engine: VectorBTEngine | None = None,
|
|
metric: str = "sharpe",
|
|
):
|
|
self.strategy_class = strategy_class
|
|
self.search_space = search_space
|
|
self.price_df = price_df
|
|
self.factor_df = factor_df if factor_df is not None else price_df
|
|
self.bt_engine = bt_engine or VectorBTEngine()
|
|
self.metric = metric
|
|
self._extractor = _METRIC_EXTRACTORS.get(metric)
|
|
if self._extractor is None:
|
|
raise ValueError(f"不支持的指标: '{metric}'。可选: {list(_METRIC_EXTRACTORS)}")
|
|
|
|
def __call__(self, trial: optuna.Trial) -> float:
|
|
params = self.search_space.suggest(trial)
|
|
|
|
try:
|
|
strategy = self.strategy_class(**params)
|
|
except TypeError:
|
|
# 过滤不匹配的参数
|
|
valid = {k: v for k, v in params.items()
|
|
if k in self.strategy_class.__init__.__code__.co_varnames}
|
|
strategy = self.strategy_class(**valid)
|
|
|
|
report = self.bt_engine.run(strategy, self.price_df, self.factor_df)
|
|
|
|
value = self._extractor(report) # type: ignore
|
|
|
|
# 无效值处理
|
|
if value is None or (isinstance(value, float) and (pd.isna(value) or value == float("inf"))):
|
|
return float("-inf")
|
|
|
|
return float(value)
|