回测的局限性

# ❌ 错误示例：未来函数
signal = df['close'] > df['ma20']
execute_price = df['close']
# 问题：收盘前你不知道收盘价是否高于MA20

# ✅ 正确做法：使用滞后数据
signal = df['close'].shift(1) > df['ma20'].shift(1)
execute_price = df['open']

# ❌ 错误示例：财报日期错误
df['use_earnings'] = df['quarter_end']

# ✅ 正确做法：使用实际公告日期
df['use_earnings'] = df['announcement_date']

def apply_transaction_costs(returns, turnover, fee_rate=0.00025,
                            stamp_tax=0.0005, slippage=0.001):
    """
    对收益率扣除交易成本
    turnover: 日换手率（买卖双边）
    fee_rate: 佣金率（默认万2.5）
    stamp_tax: 印花税（卖出时，默认万5）
    slippage: 滑点（双边，默认千1）
    """
    buy_cost = turnover * fee_rate + turnover * slippage
    sell_cost = turnover * (fee_rate + stamp_tax + slippage)
    total_cost = buy_cost + sell_cost
    net_returns = returns - total_cost
    return net_returns

# 示例：一个日换手率50%的策略，年化成本有多恐怖？
daily_turnover = 0.5
annual_turnover = daily_turnover * 252
cost_per_year = annual_turnover * (0.00025*2 + 0.0005 + 0.001*2)
print(f"年化交易成本: {cost_per_year:.2%}")
# 结果可能高达30-40%，足以吞噬大部分策略收益

请审查以下回测代码，找出所有潜在的未来函数、幸存者偏差、成本低估问题：
[粘贴你的回测代码]
要求：逐行标注问题，给出修改建议，并估算这些问题对回测收益的高估幅度。

请为我设计一个量化策略回测质量检查清单（Checklist），包含：
1. 数据质量检查（10项）
2. 未来函数排查（8项）
3. 成本与流动性检查（6项）
4. 过拟合风险评估（5项）
每项给出检查方法和通过标准。

我有一个策略回测年化收益40%，夏普2.0，最大回撤10%。
请帮我分析：如果这个策略实盘运行，哪些因素会导致收益下降？
请给出各项因素的量化估算（如滑点影响、容量限制、情绪偏差等），
并预测实盘可能的收益区间。

"""回测质量检查器 - 在相信回测结果之前，先运行这个"""
import pandas as pd
import numpy as np

class BacktestAuditor:
    def __init__(self, price_df, signal_df, trades_df):
        self.price = price_df
        self.signal = signal_df
        self.trades = trades_df
    
    def check_lookahead_bias(self):
        """检查未来函数"""
        issues = []
        # 检查信号是否使用了未来数据
        if 'close' in self.signal.columns:
            if (self.signal.index == self.price.index).all():
                issues.append("警告：信号与价格同日期，可能存在未来函数")
        return issues
    
    def check_survivorship_bias(self):
        """检查幸存者偏差"""
        total_stocks = self.price['code'].nunique()
        # 如果股票池在过去几年一成不变，可能存在幸存者偏差
        yearly_counts = self.price.groupby(
            self.price.index.year)['code'].nunique()
        if yearly_counts.std() < total_stocks * 0.05:
            return ["警告：股票池数量年度变化极小，可能存在幸存者偏差"]
        return []
    
    def estimate_costs(self, turnover_estimate=0.3):
        """估算交易成本对收益的影响"""
        fee = 0.00025
        stamp = 0.0005
        slip = 0.001
        daily_cost = turnover_estimate * (fee*2 + stamp + slip*2)
        annual_cost = daily_cost * 252
        return annual_cost
    
    def full_audit(self):
        print("="*50)
        print("回测质量审计报告")
        print("="*50)
        
        lb = self.check_lookahead_bias()
        print(f"
[未来函数检查] 发现 {len(lb)} 个问题")
        for issue in lb:
            print(f"  - {issue}")
        
        sb = self.check_survivorship_bias()
        print(f"
[幸存者偏差检查] 发现 {len(sb)} 个问题")
        for issue in sb:
            print(f"  - {issue}")
        
        cost = self.estimate_costs()
        print(f"
[成本估算] 年化交易成本约: {cost:.2%}")
        print("="*50)

# 使用示例
# auditor = BacktestAuditor(price_df, signal_df, trades_df)
# auditor.full_audit()