ptrade 平台如何通过 merge_type 规避财务数据未来函数？（附 Python 源码）

为什么基本面回测容易产生“未来函数”？

在量化投资中，未来函数是指在历史某个时间点使用了该时间点之后才发生的数据或信息。在进行基本面（财务数据）选股回测时，未来函数是一个极易踩中的“大坑”：

1. 财报修正：上市公司在发布季度报或年报后，往往会因为审计、会计准则调整等原因，在几个月甚至几年后对历史财报进行修正或重分类。
2. 回测偏差：如果我们在回测 2020 年的数据时，直接取到了 2023 年修正后的“最新”财务数据，就会导致回测结果异常优异，但实盘时根本无法复制（因为 2020 年时修正数据尚未发布）。

核心解决方案：合理设置 merge_type

ptrade 提供了强大的 get_fundamentals 接口，其关键参数 merge_type 就是为了解决这一痛点而设计的：

• merge_type = None（或不传）：获取首次发布的数据。即使上市公司后期对该笔数据进行了修正，接口也只返回历史首次披露的原样数据。在回测场景中，为了彻底规避未来函数，必须使用此模式。
• merge_type = 1：获取最新发布的数据。返回经过修正后的最新财务数据。此模式适用于研究场景或实盘交易场景，但不适用于历史回测。

接口定义对照：

get_fundamentals(security, table, fields=None, date=None, start_year=None, end_year=None, report_types=None, date_type=None, merge_type=None)

避坑实战：规避未来函数的财务选股策略源码

以下是一个完整的 ptrade 策略示例。该策略在盘前（before_trading_start）获取沪深 300 成分股的资产负债表数据，通过显式不传 merge_type（默认首次发布）来规避未来函数，筛选出总资产排名前 10 的股票，并在盘中进行调仓。

import time

def initialize(context):
    # 设置基准为沪深300
    set_benchmark('000300.SS')
    # 设定运行周期为日线
    g.security = '000300.SS'
    set_universe(g.security)
    
    # 记录持仓天数
    g.days = 0
    # 调仓周期（例如每20个交易日调仓一次）
    g.rebalance_period = 20
    g.target_list = []

def before_trading_start(context, data):
    # 仅在调仓日执行财务选股，降低API调用频率
    if g.days % g.rebalance_period == 0:
        log.info("--- 调仓日基本面选股开始 ---")
        # 获取沪深300成分股
        stocks = get_index_stocks('000300.XBHS')
        
        # 获取当前回测日期的前一交易日，避免当日盘中未公布数据的未来函数
        current_date = context.previous_date.strftime('%Y%m%d')
        
        try:
            # 获取资产负债表中的总资产 (total_assets)
            # 核心：不传 merge_type (或设为 None)，确保获取首次发布数据，彻底消除未来函数！
            df = get_fundamentals(
                security=stocks, 
                table='balance_statement', 
                fields='total_assets', 
                date=current_date,
                merge_type=None  # 显式声明获取首次发布数据
            )
            
            if df is not None and not df.empty:
                # 过滤掉空值并按总资产降序排序
                df_sorted = df.dropna(subset=['total_assets']).sort_values(by='total_assets', ascending=False)
                # 选取总资产最大的前10只股票作为目标池
                g.target_list = list(df_sorted.index[:10])
                log.info("选股成功，目标持仓列表: %s" % g.target_list)
            else:
                log.warning("获取财务数据为空，保持原仓位")
        except Exception as e:
            log.error("获取财务数据异常: %s" % str(e))
            
        # 遵守流控限制，API调用后适当sleep
        time.sleep(1)

def handle_data(context, data):
    # 仅在调仓日进行交易
    if g.days % g.rebalance_period == 0 and len(g.target_list) > 0:
        # 获取当前持仓
        current_positions = list(context.portfolio.positions.keys())
        
        # 1. 卖出不在目标列表中的股票
        for stock in current_positions:
            if stock not in g.target_list:
                order_target(stock, 0)
                log.info("卖出非目标股: %s" % stock)
                
        # 2. 等权重买入目标股票
        cash = context.portfolio.cash
        if len(g.target_list) > 0:
            target_value = cash / len(g.target_list)
            for stock in g.target_list:
                if data[stock].is_open > 0: # 确保未停牌
                    order_target_value(stock, target_value)
                    log.info("买入/调整目标股: %s，目标价值: %.2f" % (stock, target_value))
                    
    g.days += 1

财务数据避坑黄金法则总结

1. 回测必用 merge_type=None：确保历史回测时拿到的是当时历史时点首次披露的数据，防止“后视镜”效应。
2. 使用 context.previous_date 作为查询日期：在 before_trading_start 中获取财务数据时，传入前一交易日日期，防止在交易日当天盘中提前使用了尚未披露的财报数据。
3. 注意 API 流控限制：get_fundamentals 接口每秒调用不得超过 100 次，单次最大调用量 500 条。在多股多字段查询时，务必在代码中加入 time.sleep() 进行限流保护。