因子模型是否有效，需要通过回测去分析和验证。
这里将探讨量化回测基本原理，解释为什么某些设计合理的，某些危险，以及如何构建回测系统。
1 回测计算
1.1 核心目标
回测(Backtesting)是用历史数据模拟交易策略在过去的表现，以评估其有效性和稳健性。
回测核心原则是：
1）无前视偏差(No Look-ahead Bias)
策略在时点t做出的决策只能使用 t 时刻及之前可获得的信息，不能使用未来数据。
2）可投资性(Investability)
模拟的交易必须能在真实市场中执行，包括流动性、交易成本、停牌等现实约束。
3）一致性(Consistency)
数据处理、收益计算、成本扣除等必须在时间维度上保持一致，避免人为偏差。
任何违反这些原则的回测结果都是不可信的，往往高估策略表现。
1.2 信号与收益对齐
1.2.1 信号生成与使用逻辑
在量化选股中，通常流程为：
1. 在 t 日收盘后，利用截至当日的数据计算因子得分(信号)。
2. 在 t+1 日开盘，根据该信号构建或调整组合。
3. 持有组合至 t+1 日收盘，获得 t+1 日的收益。
因此，t日的信号应该对应t+1日的收益，而不是 t 日当天的收益。
若直接用t日的信号获取t日的收益，就隐含了在t日开盘前就知道了t日收盘后才能算出的信号，这属于典型的前视偏差。
1.2.2 实际处理方式
1）收益平移
通常采用收益平移的方式处理，具体为：
将收益序列向上平移一天，使得在日期t处的收益值实际上是t+1日的收益。这是正确的做法。
示例代码如下
returns = returns.shift(-1)
2）为什么平移必要
设原始收益矩阵 R的行索引为日期，R[t]表示 t 日的日收益率。若信号S[t]在 t 日收盘后可用，则正确的组合收益应为R[t+1]，即持有期从 t 日收盘到 t+1 日收盘。将R向上平移一位后，新的收益序列R_shift[t] = R[t+1]，这样在循环中直接使用R_shift.loc[date]就自然对应了未来一天的收益。
1.3 组合构建与再平衡
1.3.1 等权组合与定期再平衡
1）Top-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 -> T+1 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, index=strat_dates, name="strategy") 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 -> T+1 (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 tomorrow's return for today's 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 T+1) 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, index=strat_dates, name="strategy") 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, returns=None) -> 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 -> T+1 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, index=strat_dates, name="strategy") return self._compute_metrics(strategy_returns, returns)2.2.4_run_direct方法
1）数据对齐与收益率平移
取得分矩阵和收益率矩阵共同的日期和股票。
将无穷值替换为NaN。
returns = returns.shift(-1)，将收益率向前平移一天。
在日期T看到的收益率实际上是T+1日的收益率，从而避免了未来函数，即今天(T)根据收盘数据选股，持有到明天(T+1)获取收益。
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, index=strat_dates, name="strategy") 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
---
