因子模型是否有效需要通过回测去分析和验证。这里将探讨量化回测基本原理解释为什么某些设计合理的某些危险以及如何构建回测系统。1 回测计算1.1 核心目标回测(Backtesting)是用历史数据模拟交易策略在过去的表现以评估其有效性和稳健性。回测核心原则是1无前视偏差(No Look-ahead Bias)策略在时点t做出的决策只能使用 t 时刻及之前可获得的信息不能使用未来数据。2可投资性(Investability)模拟的交易必须能在真实市场中执行包括流动性、交易成本、停牌等现实约束。3一致性(Consistency)数据处理、收益计算、成本扣除等必须在时间维度上保持一致避免人为偏差。任何违反这些原则的回测结果都是不可信的往往高估策略表现。1.2 信号与收益对齐1.2.1 信号生成与使用逻辑在量化选股中通常流程为1. 在 t 日收盘后利用截至当日的数据计算因子得分(信号)。2. 在 t1 日开盘根据该信号构建或调整组合。3. 持有组合至 t1 日收盘获得 t1 日的收益。因此t日的信号应该对应t1日的收益而不是 t 日当天的收益。若直接用t日的信号获取t日的收益就隐含了在t日开盘前就知道了t日收盘后才能算出的信号这属于典型的前视偏差。1.2.2 实际处理方式1收益平移通常采用收益平移的方式处理具体为将收益序列向上平移一天使得在日期t处的收益值实际上是t1日的收益。这是正确的做法。示例代码如下returns returns.shift(-1)2为什么平移必要设原始收益矩阵 R的行索引为日期R[t]表示 t 日的日收益率。若信号S[t]在 t 日收盘后可用则正确的组合收益应为R[t1]即持有期从 t 日收盘到 t1 日收盘。将R向上平移一位后新的收益序列R_shift[t] R[t1]这样在循环中直接使用R_shift.loc[date]就自然对应了未来一天的收益。1.3 组合构建与再平衡1.3.1 等权组合与定期再平衡1Top-N等权持有策略每次调仓选择得分最高的N只股票每只分配1/N的权重持有至下一个调仓日。这是一种简单且经典的量化选股方式。2定期再平衡在两次调仓之间组合权重会因股票价格变动而偏离初始等权。但实际可能选择不进行每日再平衡而是等到下一个调仓日才恢复等权。这种假设下非调仓日的组合收益等于所有持仓股票当日收益的简单平均。若组合在期初等权配置于N只股票且期间不交易则期末组合收益为各股票收益的简单平均(忽略股息等)。因此可以在非调仓日直接对持仓股票收益取平均是合理的。1.3.2 调仓日收益的处理在调仓日交易发生在收盘时(或次日开盘因此当天的收益通常有两种处理方式1方式一先按旧持仓计算当天收益再在收盘时换仓成本在当天扣除。2方式二在当天开盘即换仓则当天收益按新持仓计算并扣除成本。在是回测时可能采用简化处理的方式比如当天收盘后根据信号决定新持仓但当天收益用新持仓的未来一天收益计算(因为收益已平移)且成本从未来一天收益中扣除。这在时间上略有不一致但影响较小。更严谨做法是在调仓日当天仍持有旧组合收益按旧组合计算成本在当天扣除次日开始按新组合计算收益。1.4 交易成本建模原理交易成本是回测中必须考虑的重要因素忽视成本会显著高估策略表现。常见的成本模型有1固定比例成本按交易金额的一定比例收取如佣金、印花税、冲击成本。2换手率模型成本 换手率 × 单边成本比例。换手率通常定义为组合中发生变动的资产比例。1.4.1 单边与双边换手当组合从旧持仓变为新持仓时需要卖出不再持有的股票并买入新加入的股票。单边换手率仅考虑一侧(买入或卖出)的资产变动比例而双边换手率是两侧之和。若买卖成本费率相同总成本 双边换手率 × 费率 2 × 单边换手率 × 费率。1.4.2 首次建仓成本首次建仓时所有股票都是新买入的换手率应为 100%(买入侧。若使用双边换手则总成本更高。1.4.3 成本扣除时点成本应在交易发生时从组合价值中扣除。在日频回测中若假设交易在收盘时执行则成本应从当日净值中扣除。实际计算也可能将成本从次日的收益中扣除相当于将成本推迟一天轻微影响净值曲线。1.5 数据对齐与缺失值处理原理1.5.1 日期与股票的交集由于停牌、退市等原因某些股票在某些日期可能没有交易数据(价格或收益。另外因子得分也可能存在缺失值。回测时必须对齐数据只使用同时具有信号和收益的样本否则会引入偏差。通常使用intersection对齐日期和股票。在计算收益时若某只持仓股票当日无收益数据代码选择将其排除在平均之外(相当于该股票当日收益为 0。更严谨的做法是保留该股票并假定收益为 0或者将其权重重新分配给其他股票以反映实际无法交易的情况。直接排除会略微改变组合权重但通常影响很小。1.5.2 幸存者偏差如果使用历史数据只包含当前存活股票剔除了已退市股票会产生幸存者偏差高估策略表现。使用者应确保数据包含退市股票。在处理停牌时若简单跳过也可能引入偏差。1.6 绩效评估指标1.6.1 年化收益率与波动率1年化收益率将累计收益转换为按年复利计算的收益率。公式是几何年化适用于复利情形。2年化波动率日收益率标准差乘以基于收益率独立同分布的假设将日波动率放大到年尺度。1.6.2 夏普比率夏普比率衡量每单位总风险(波动率)所获得的超额回报即超过无风险利率。公式 “年化收益率/年化波动率” 是简化形式未减去无风险利率。它适用于评估策略的风险调整后表现。1.6.3 最大回撤与卡玛比率1最大回撤从历史最高点到随后最低点的最大跌幅反映策略可能面临的最大亏损。2卡玛比率年化收益率/最大回撤绝对值衡量单位最大回撤带来的收益。常用于评估趋势跟踪或高波动策略。1.6.4 信息比率信息比率衡量策略相对于基准的超额收益的稳定性。公式为年化超额收益/年化跟踪误差(超额收益的标准差。它考察的是主动管理能力而非绝对收益。2 回测代码示例2.1 代码示例这里给出一个简化测回测示例带以帮助理解和分析验证回测计算过程。# ────────────────────────────────────────────────────────── # Backtest engine # ────────────────────────────────────────────────────────── class BacktestEngine: Long-only backtest with periodic rebalancing. Parameters ---------- n_stocks : number of stocks to hold (top N by composite score) rebalance_days : trading days between rebalances transaction_cost : cost per trade (e.g. 0.002 20bps) benchmark : optional benchmark returns Series def __init__( self, n_stocks: int 30, rebalance_days: int 20, transaction_cost: float 0.002, benchmark: Optional[pd.Series] None, ): self.n_stocks n_stocks self.rebalance_days rebalance_days self.transaction_cost transaction_cost self.benchmark benchmark def run( self, composite_score: pd.DataFrame, prices: pd.DataFrame, returns: pd.DataFrame None, ) - Dict: Run backtest. Parameters ---------- composite_score : DataFrame (datetime x stock or datetime x 1) If multi-column (one per stock), uses directly. If single column, uses as ranking signal. prices : datetime x stock price DataFrame Returns ------- Dict with strategy_returns, benchmark_returns, trades, metrics if returns is None: returns prices.pct_change() # Ensure composite is per-stock if composite_score.shape[1] 1: # Single composite column: use as ranking, select top N return self._run_ranking(composite_score.iloc[:, 0], returns) else: # Already per-stock scores return self._run_direct(composite_score, returns) def _run_ranking( self, scores, returns: pd.DataFrame, ) - Dict: Rank-based approach: select top N stocks by score. if isinstance(scores, pd.Series) and isinstance(scores.index, pd.MultiIndex): score_df scores.unstack() elif isinstance(scores, pd.DataFrame): if isinstance(scores.index, pd.MultiIndex): score_df scores.iloc[:, 0].unstack() else: # datetime x stock DataFrame — use directly if scores.shape[1] 1: score_df scores.iloc[:, 0].to_frame() # This is a single-column DataFrame with stocks as columns score_df scores else: score_df scores else: raise ValueError(Scores must be a MultiIndex Series, DataFrame, or datetime×stock matrix) score_df score_df.replace([np.inf, -np.inf], np.nan) dates score_df.index.intersection(returns.index) # Forward-shift returns: entry at value on date T earns T - T1 return. fwd_returns returns.shift(-1) portfolio_values [1.0] current_holdings set() trade_dates [] daily_returns [] for i, date in enumerate(dates[1:], 1): if i % self.rebalance_days ! 0: # Hold previous portfolio if current_holdings: available current_holdings.intersection(returns.columns) if available: day_ret fwd_returns.loc[date][list(available)].mean() else: day_ret 0 else: day_ret 0 portfolio_values.append(portfolio_values[-1] * (1 day_ret)) daily_returns.append(day_ret) continue # Rebalance if isinstance(score_df, pd.DataFrame) and not isinstance(score_df.index, pd.MultiIndex): # Already datetime x stock scores_today score_df.loc[date].dropna() else: scores_today score_df.loc[date].dropna() top_stocks scores_today.nlargest(self.n_stocks).index.tolist() # Calculate turnover cost if current_holdings: turnover len(set(top_stocks) - current_holdings) / max(self.n_stocks, 1) cost turnover * self.transaction_cost else: cost 0 trade_dates.append(date) current_holdings set(top_stocks) available set(top_stocks).intersection(returns.columns) if available: day_ret returns.loc[date][list(available)].mean() - cost else: day_ret -cost portfolio_values.append(portfolio_values[-1] * (1 day_ret)) daily_returns.append(day_ret) strat_dates dates[1:len(daily_returns)1] strategy_returns pd.Series(daily_returns, indexstrat_dates, namestrategy) return self._compute_metrics(strategy_returns, returns) def _run_direct( self, scores: pd.DataFrame, returns: pd.DataFrame, ) - Dict: Top-N stock selection with periodic rebalancing. At each rebalance, select top-N by score, equal-weight hold. Factor values on date T use data up to T close, so they can only predict returns from T - T1 (shift returns forward by 1 day). # Align indices common_dates scores.index.intersection(returns.index) common_stocks scores.columns.intersection(returns.columns) scores scores.loc[common_dates, common_stocks] returns returns.loc[common_dates, common_stocks] scores scores.replace([np.inf, -np.inf], np.nan) # Shift returns: what matters is tomorrows return for todays selection returns returns.shift(-1) portfolio_values [1.0] daily_returns [] trade_dates [] current_holdings [] for i, date in enumerate(scores.index): scores_today scores.loc[date].dropna() rets_today returns.loc[date].dropna() # Rebalance every rebalance_days if i % self.rebalance_days 0 or len(current_holdings) 0: if len(scores_today) self.n_stocks: top_stocks scores_today.nlargest(self.n_stocks).index.tolist() elif len(scores_today) 0: top_stocks scores_today.index.tolist() else: top_stocks [] # Calculate turnover cost if current_holdings and len(top_stocks) 0: n_turnover len(set(top_stocks) - set(current_holdings)) turnover_rate n_turnover / max(self.n_stocks, 1) cost turnover_rate * self.transaction_cost elif not current_holdings and len(top_stocks) 0: cost self.transaction_cost # initial entry cost else: cost 0 if cost 0: trade_dates.append(date) current_holdings top_stocks # Compute day return (already shifted to T1) if current_holdings: available [s for s in current_holdings if s in rets_today.index] if available: day_ret rets_today[available].mean() if date in trade_dates and len(trade_dates) 0 and trade_dates[-1] date: day_ret - cost else: day_ret 0 else: day_ret 0 portfolio_values.append(portfolio_values[-1] * (1 day_ret)) daily_returns.append(day_ret) strat_dates scores.index[:len(daily_returns)] strategy_returns pd.Series(daily_returns, indexstrat_dates, namestrategy) return self._compute_metrics(strategy_returns, returns) def _compute_metrics( self, strategy_returns: pd.Series, all_returns: pd.DataFrame, ) - Dict: Compute comprehensive performance metrics. ret strategy_returns.dropna() if len(ret) 10: return {strategy_returns: ret, metrics: {}} # Basic metrics total_return (1 ret).prod() - 1 n_years len(ret) / 252 ann_return (1 total_return) ** (1 / max(n_years, 1/252)) - 1 ann_vol ret.std() * np.sqrt(252) sharpe ann_return / (ann_vol 1e-10) # Max drawdown cum (1 ret).cumprod() running_max cum.cummax() dd (cum - running_max) / running_max max_dd dd.min() calmar ann_return / (abs(max_dd) 1e-10) # Win rate win_rate (ret 0).sum() / len(ret) # Benchmark benchmark_stats None if self.benchmark is not None: bm_aligned self.benchmark.loc[ret.index].dropna() if len(bm_aligned) 0: bm_ann bm_aligned.mean() * 252 bm_vol bm_aligned.std() * np.sqrt(252) bm_sharpe bm_ann / (bm_vol 1e-10) excess ret - bm_aligned info_ratio excess.mean() * 252 / (excess.std() * np.sqrt(252) 1e-10) benchmark_stats { benchmark_ann_return: bm_ann, benchmark_ann_vol: bm_vol, benchmark_sharpe: bm_sharpe, excess_return: ann_return - bm_ann, information_ratio: info_ratio, } metrics { total_return: total_return, annual_return: ann_return, annual_volatility: ann_vol, sharpe_ratio: sharpe, max_drawdown: max_dd, calmar_ratio: calmar, win_rate: win_rate, n_trading_days: len(ret), n_years: n_years, } if benchmark_stats: metrics.update(benchmark_stats) return { strategy_returns: ret, cumulative: cum, metrics: metrics, drawdown: dd, }2.2 分段说明2.2.1 初始化方法__init__n_stocks持仓股票数量即每次调仓时选择综合得分最高的前N只股票。rebalance_days调仓周期(交易日)。例如20表示每20个交易日调仓一次。transaction_cost单边交易成本比例。例如0.002表示 0.2%(20 bps。benchmark可选基准收益率序列(日收益率用于计算超额收益和信息比率等指标。def __init__( self, n_stocks: int 30, rebalance_days: int 20, transaction_cost: float 0.002, benchmark: Optional[pd.Series] None, ): self.n_stocks n_stocks self.rebalance_days rebalance_days self.transaction_cost transaction_cost self.benchmark benchmark2.2.2run方法如果调用者没有提供returns则用prices.pct_change()计算每日收益率。判断composite_score的列数只有 1 列调用_run_ranking并将该列作为scores参数传入一个Series。多于 1 列调用_run_direct将整个DataFrame传入。def run(self, composite_score, prices, returnsNone) - Dict: if returns is None: returns prices.pct_change() if composite_score.shape[1] 1: return self._run_ranking(composite_score.iloc[:, 0], returns) else: return self._run_direct(composite_score, returns)2.2.3_run_ranking方法1输出处理将scores统一转换成日期×股票的二维DataFrame其中行为日期列为股票值为因子得分。如果传入MultiIndex Series(索引为(date, stock)unstack()可以得到正确的日期×股票矩阵。如果传入的是普通DataFrame且索引为MultiIndex则取第一列并unstack()。如果传入的是普通DataFrame且索引不是MultiIndex则直接使用该DataFrame。但此时如果scores.shape[1]1会先执行score_df scores.iloc[:, 0].to_frame()随后又被score_df scores覆盖因此最终score_df仍是原始的scores仅一列。_run_ranking实际上只适用于索引为(date, stock)的MultiIndex数据或者已经被展开为日期×股票的DataFrame(但该DataFrame通常不应只有一列否则无法选出多只股票。if isinstance(scores, pd.Series) and isinstance(scores.index, pd.MultiIndex): score_df scores.unstack() elif isinstance(scores, pd.DataFrame): if isinstance(scores.index, pd.MultiIndex): score_df scores.iloc[:, 0].unstack() else: if scores.shape[1] 1: score_df scores.iloc[:, 0].to_frame() score_df scores # 注意这行覆盖了上一行最终就是 scores 本身 else: score_df scores else: raise ValueError(...) score_df score_df.replace([np.inf, -np.inf], np.nan)2回测主循环dates是得分矩阵和收益率矩阵共同覆盖的日期集合。fwd_returns主要用于预防前视偏差循环从dates[1:]开始跳过第一个日期(第0天因为第0天没有前一天收益组合初始值为 1。enumerate(..., 1)使i从 1 开始便于用i% rebalance_days判断是否为调仓日。dates score_df.index.intersection(returns.index) # Forward-shift returns: entry at value on date T earns T - T1 return. fwd_returns returns.shift(-1) portfolio_values [1.0] current_holdings set() trade_dates [] daily_returns [] for i, date in enumerate(dates[1:], 1): ...非调仓日i % rebalance_days ! 0如果当前持仓不为空取这些持仓与收益率列的交集(避免某些股票停牌或已退市。当天的组合收益等于这些股票当日收益率的简单平均等权。如果没有可用股票或没有持仓收益为 0。用portfolio_values[-1] * (1 day_ret)更新组合净值并将日收益记录到daily_returns。不收取任何交易成本因为此时没有调仓。if current_holdings: available current_holdings.intersection(returns.columns) if available: day_ret fwd_returns.loc[date][list(available)].mean() else: day_ret 0 else: day_ret 0 portfolio_values.append(portfolio_values[-1] * (1 day_ret)) daily_returns.append(day_ret) continue调仓日i % rebalance_days 0取出当天的因子得分(一个Series索引为股票值为得分。用nlargest(self.n_stocks)选出得分最高的 N 只股票。注意score_df此时应为日期×股票矩阵所以score_df.loc[date]会得到一行。# 获取当天分数 if isinstance(score_df, pd.DataFrame) and not isinstance(score_df.index, pd.MultiIndex): scores_today score_df.loc[date].dropna() else: scores_today score_df.loc[date].dropna() top_stocks scores_today.nlargest(self.n_stocks).index.tolist()如果已有持仓计算新进入组合的股票数量(set(top_stocks)-current_holdings)长度。换手率 新进入数量 /n_stocks。成本 换手率 × 单边成本比例。如果是首次建仓current_holdings为空成本设为 0。这里为简化计算做了如下假设实际应该会更复杂换手率只考虑了买入的新股票没有考虑卖出的旧股票低估了实际交易成本(双边。分母固定n_stocks实际持仓数可能因停牌等小于n_stocks导致换手率计算偏差。首次建仓成本被忽略不符合实际。# 计算换手成本 if current_holdings: turnover len(set(top_stocks) - current_holdings) / max(self.n_stocks, 1) cost turnover * self.transaction_cost else: cost 0记录调仓日期更新持仓为新的top_stocks。用当天的收益率(returns.loc[date])计算新持仓的平均收益再减去交易成本。如果新持仓中没有可交易的股票则收益为-cost(即纯成本损失。_run_ranking应该已经假设将收益序列向前平移一天。trade_dates.append(date) current_holdings set(top_stocks) available set(top_stocks).intersection(returns.columns) if available: day_ret returns.loc[date][list(available)].mean() - cost else: day_ret -cost然后更新组合净值和日收益。portfolio_values.append(portfolio_values[-1] * (1 day_ret)) daily_returns.append(day_ret)最后构造策略收益序列由于循环从dates[1:]开始长度可能与daily_returns一致。用dates[1:len(daily_returns)1]作为索引构造Series。最后调用_compute_metrics计算绩效指标。strat_dates dates[1:len(daily_returns)1] strategy_returns pd.Series(daily_returns, indexstrat_dates, namestrategy) return self._compute_metrics(strategy_returns, returns)2.2.4_run_direct方法1数据对齐与收益率平移取得分矩阵和收益率矩阵共同的日期和股票。将无穷值替换为NaN。returns returns.shift(-1)将收益率向前平移一天。在日期T看到的收益率实际上是T1日的收益率从而避免了未来函数即今天(T)根据收盘数据选股持有到明天(T1)获取收益。common_dates scores.index.intersection(returns.index) common_stocks scores.columns.intersection(returns.columns) scores scores.loc[common_dates, common_stocks] returns returns.loc[common_dates, common_stocks] scores scores.replace([np.inf, -np.inf], np.nan) returns returns.shift(-1)2回测主循环遍历得分矩阵的每一个日期。scores_today当天所有股票的得分(去除 NaN。rets_today当天对应的明天的收益率(已经shift(-1)所以是未来一天的收益。调仓条件每rebalance_days天一次或者当前持仓为空(初始状态。portfolio_values [1.0] daily_returns [] trade_dates [] current_holdings [] for i, date in enumerate(scores.index): scores_today scores.loc[date].dropna() rets_today returns.loc[date].dropna() # 调仓条件 if i % self.rebalance_days 0 or len(current_holdings) 0: ...选股与成本计算根据得分数量选择前 N 只股票如果不足 N 只但有股票则全部选取否则空仓换手成本计算已有持仓仅计算新进入的股票数量除以n_stocks再乘以单边成本。初始建仓收取一笔固定的transaction_cost而不是按买入股票数量比例收费。其他情况成本为 0。如果成本大于 0记录调仓日期。更新持仓。这里存在的问题是换手成本仍然只计算了买入侧卖出侧未计。初始建仓成本固定为一笔而不是按买入股票数量比例这可能导致成本失真。当持仓数量小于n_stocks时分母仍为n_stocks会低估换手率。if len(scores_today) self.n_stocks: top_stocks scores_today.nlargest(self.n_stocks).index.tolist() elif len(scores_today) 0: top_stocks scores_today.index.tolist() else: top_stocks [] if current_holdings and len(top_stocks) 0: n_turnover len(set(top_stocks) - set(current_holdings)) turnover_rate n_turnover / max(self.n_stocks, 1) cost turnover_rate * self.transaction_cost elif not current_holdings and len(top_stocks) 0: cost self.transaction_cost # 初始建仓固定成本 else: cost 0 if cost 0: trade_dates.append(date) current_holdings top_stocks计算当日收益如果持仓不为空取其中在rets_today中有收益数据的股票计算简单平均收益。如果当天发生了调仓date是trade_dates的最后一个元素则从当天收益中扣除交易成本。如果没有可用股票或空仓收益为 0。更新组合净值和日收益。需要注意的是这里使用的rets_today是明天的收益所以成本是在明天的收益中扣除的实际上成本发生在调仓日收盘时但这里提前扣在次日的收益中存在一天的时间差但对整体结果影响较小。更严谨的做法是在调仓日的净值中扣除成本而不是在次日收益中扣除。if current_holdings: available [s for s in current_holdings if s in rets_today.index] if available: day_ret rets_today[available].mean() if date in trade_dates and len(trade_dates) 0 and trade_dates[-1] date: day_ret - cost else: day_ret 0 else: day_ret 0 portfolio_values.append(portfolio_values[-1] * (1 day_ret)) daily_returns.append(day_ret)构造策略收益序列使用scores.index的前len(daily_returns)个日期作为索引。由于循环遍历了所有日期len(daily_returns)应等于len(scores.index)。注意这里的returns已经是被shift(-1)过的但_compute_metrics并不使用它所以不影响。strat_dates scores.index[:len(daily_returns)] strategy_returns pd.Series(daily_returns, indexstrat_dates, namestrategy) return self._compute_metrics(strategy_returns, returns)2.2.5_compute_metrics方法去除策略收益中的NaN。如果有效数据少于 10 个交易日则只返回策略收益序列不计算指标。这时为了防止样本过小导致指标不稳定。def _compute_metrics(self, strategy_returns, all_returns) - Dict: ret strategy_returns.dropna() if len(ret) 10: return {strategy_returns: ret, metrics: {}}基础收益与风险指标总收益率即复利累计收益。年化收益率将总收益按年数开方年数 交易日数/252。若年数小于1/252即不足一天则按 1 天年化。年化波动率日收益率标准差 × √252。夏普比率年化收益率/年化波动率分母加极小值避免除零。total_return (1 ret).prod() - 1 n_years len(ret) / 252 ann_return (1 total_return) ** (1 / max(n_years, 1/252)) - 1 ann_vol ret.std() * np.sqrt(252) sharpe ann_return / (ann_vol 1e-10)最大回撤与卡玛比率累计净值曲线cum。滚动最大值running_max。回撤序列dd (当前净值 - 历史最高) / 历史最高。最大回撤max_dd为dd的最小值负值。卡玛比率 年化收益率/|最大回撤|衡量每单位回撤带来的收益。cum (1 ret).cumprod() running_max cum.cummax() dd (cum - running_max) / running_max max_dd dd.min() calmar ann_return / (abs(max_dd) 1e-10)胜率日收益大于 0 的天数占比。win_rate (ret 0).sum() / len(ret)基准对比指标将基准收益与策略收益对齐到ret.index。计算基准的年化收益. (简单日收益均值 × 252、年化波动、夏普。计算超额收益序列excess ret - bm_aligned。信息比率 年化超额收益 / 年化跟踪误差年化跟踪误差 超额收益标准差 × √252。excess_return为策略年化收益减去基准年化收益。if self.benchmark is not None: bm_aligned self.benchmark.loc[ret.index].dropna() if len(bm_aligned) 0: bm_ann bm_aligned.mean() * 252 bm_vol bm_aligned.std() * np.sqrt(252) bm_sharpe bm_ann / (bm_vol 1e-10) excess ret - bm_aligned info_ratio excess.mean() * 252 / (excess.std() * np.sqrt(252) 1e-10) benchmark_stats { benchmark_ann_return: bm_ann, benchmark_ann_vol: bm_vol, benchmark_sharpe: bm_sharpe, excess_return: ann_return - bm_ann, information_ratio: info_ratio, }返回结果最终返回包含策略收益、累计净值、指标和回撤的字典。metrics { total_return: total_return, annual_return: ann_return, annual_volatility: ann_vol, sharpe_ratio: sharpe, max_drawdown: max_dd, calmar_ratio: calmar, win_rate: win_rate, n_trading_days: len(ret), n_years: n_years, } if benchmark_stats: metrics.update(benchmark_stats) return { strategy_returns: ret, cumulative: cum, metrics: metrics, drawdown: dd, }reference---