如何使用 ptrade 的 get_fundamentals 获取利润表数据进行选股？

在 ptrade 量化交易平台中，基于基本面进行财务选股是非常常见的策略思路。利用 get_fundamentals 接口，我们可以高效地获取上市公司的三大财务报表数据（资产负债表、利润表、现金流量表）以及各项财务能力指标。本文将重点介绍如何通过 income_statement（利润表）筛选优质股票，并提供完整的策略源码。

一、利润表 (income_statement) 核心字段说明

通过 get_fundamentals 接口查询 table='income_statement' 时，常用的核心利润表指标包括：

二、get_fundamentals 核心调用语法与注意事项

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

1. 查询模式：
   • 按日期查询（推荐回测使用）：传入 date 参数（如 '20230331'），返回指定日期前最近发布的财报数据。若不传 date，回测默认取当前回测日期的最新财报，避免未来函数。
   • 按年份查询：传入 start_year 和 end_year，配合 report_types（1:一季报，2:半年报，3:三季报，4:年报）获取特定季度的历史财报。
2. 调用频次限制：该接口为 HTTP 在线获取，每秒调用不得超过 100 次，单次最大数据返回量为 500 条数据。若对全市场股票进行多字段批量查询，建议加入 time.sleep() 保护或对股票池分批处理。

三、完整策略源码：基于利润表净利润选股策略

以下策略展示了如何在盘前阶段（before_trading_start）通过 get_fundamentals 筛选全市场 A 股中最新财报归母净利润大于 5 亿元且营业收入大于 20 亿元的股票，并在盘中进行轮动配置。

import pandas as pd
import time

def initialize(context):
    # 设置参考基准
    set_benchmark('000300.SS')
    # 每天 09:31 运行主要交易逻辑
    run_daily(context, rebalance, time='9:31')

def before_trading_start(context, data):
    # 1. 获取全市场 A 股列表
    current_date = context.blotter.current_dt.strftime('%Y%m%d')
    all_stocks = get_Ashares(date=current_date)
    
    # 2. 过滤 ST、停牌和退市股票
    filtered_stocks = filter_stock_by_status(all_stocks, filter_type=["ST", "HALT", "DELISTING"])
    
    # 3. 使用 get_fundamentals 查询利润表核心指标
    fields = ['np_parent_company_owners', 'operating_revenue']
    
    # 分批查询以避免触发接口流控限制
    batch_size = 400
    selected_stocks = []
    
    for i in range(0, len(filtered_stocks), batch_size):
        stock_batch = filtered_stocks[i:i + batch_size]
        # 按日期查询模式，获取距离当前日期最近的财报数据
        df_income = get_fundamentals(
            security=stock_batch,
            table='income_statement',
            fields=fields,
            date=current_date
        )
        
        if df_income is not None and not df_income.empty:
            # 筛选条件：归母净利润 > 5亿 (5e8) 且 营业收入 > 20亿 (20e8)
            condition = (
                (df_income['np_parent_company_owners'] > 5e8) &
                (df_income['operating_revenue'] > 20e8)
            )
            matched = df_income[condition].index.tolist()
            selected_stocks.extend(matched)
        
        time.sleep(0.5)  # 避免频繁请求
    
    # 4. 将筛选后的股票池存入全局变量，取前 10 只作为目标持仓
    g.target_stocks = selected_stocks[:10]
    log.info("今日利润表选股筛选结果: %s" % g.target_stocks)
    set_universe(g.target_stocks)

def rebalance(context):
    # 5. 根据筛选结果调仓交易
    target_list = getattr(g, 'target_stocks', [])
    if not target_list:
        return
    
    # 卖出不在目标池中的持仓
    for stock in list(context.portfolio.positions.keys()):
        if stock not in target_list:
            order_target(stock, 0)
            log.info("卖出不在选股池中的股票: %s" % stock)
            
    # 等权重买入目标池股票
    if len(target_list) > 0:
        position_value = context.portfolio.portfolio_value / len(target_list)
        for stock in target_list:
            order_target_value(stock, position_value)
            log.info("调仓买入股票: %s, 目标价值: %.2f" % (stock, position_value))

def handle_data(context, data):
    pass

四、常见问题与开发避坑提示

1. 数据 Nan 填充处理：部分刚上市或未按时披露财报的公司，查询结果可能为 NaN。在做条件判断时，应确保剔除缺失值（如使用 df.dropna() 或 Pandas 条件筛选自动过滤）。
2. 未来函数防范：在回测环境使用 date 参数时，务必传入当前回测日（例如 context.blotter.current_dt）或前一交易日，不要硬编码未来的日期。