ptrade 平台如何根据成长能力指标筛选营业收入和净利润高增长的股票？（附 Python 源码）

策略背景与选股逻辑

在基本面量化投资中，成长能力是评估企业未来扩张潜力的核心维度。通过筛选营业收入和净利润均保持高速增长的“双高”企业，可以有效剔除业绩下滑或陷入停滞的标的，锁定具备高成长红利的优质股票。

在 ptrade 平台中，我们可以通过 get_fundamentals 接口调用 growth_ability（成长能力表）中的以下两个核心字段：

1. operating_revenue_grow_rate：营业收入同比增长（%）
2. net_profit_grow_rate：净利润同比增长（%）

核心 API 介绍

获取财务数据接口

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

• table: 传入 'growth_ability' 表示成长能力表。
• fields: 传入 ['operating_revenue_grow_rate', 'net_profit_grow_rate']。
• date: 查询日期，不传默认取当前回测/交易日期。

策略 Python 源码实现

以下是一个完整的 ptrade 财务选股策略示例。策略在每天开盘前（before_trading_start）获取全市场 A 股，并筛选出营业收入同比增长 > 20% 且 净利润同比增长 > 20% 的股票，在盘中进行等权买入。

import pandas as pd
import datetime

def initialize(context):
    # 设置基准为沪深300
    set_benchmark('000300.SS')
    # 每天 9:30 运行选股与调仓逻辑
    run_daily(context, rebalance, time='9:30')
    g.security_list = []

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、停牌、退市股票
    active_stocks = filter_stock_by_status(all_stocks, ["ST", "HALT", "DELISTING"])
    
    # 3. 获取成长能力财务数据
    # 注意：财务数据单次调用最大限制，若股票较多，建议分批获取或在 before_trading_start 中处理
    fields = ['operating_revenue_grow_rate', 'net_profit_grow_rate']
    try:
        df_growth = get_fundamentals(active_stocks, 'growth_ability', fields=fields, date=current_date)
        
        if df_growth is not None and not df_growth.empty:
            # 4. 筛选双高增长股票：营业收入增长率 > 20% 且 净利润增长率 > 20%
            condition = (df_growth['operating_revenue_grow_rate'] > 20.0) & (df_growth['net_profit_grow_rate'] > 20.0)
            selected_df = df_growth[condition]
            
            # 按营业收入增长率降序排序，取前 10 只作为目标持仓
            g.security_list = list(selected_df.sort_values(by='operating_revenue_grow_rate', ascending=False).index[:10])
            log.info(f"今日筛选出的高成长股票池: {g.security_list}")
        else:
            g.security_list = []
    except Exception as e:
        log.error(f"获取财务数据异常: {e}")
        g.security_list = []

def rebalance(context):
    """
    盘中调仓逻辑：卖出不在新股票池中的持仓，等权买入新股票池中的股票
    """
    if not g.security_list:
        return
        
    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    # 1. 卖出不在新股票池中的股票
    for stock in current_positions:
        if stock not in g.security_list:
            order_target(stock, 0)
            log.info(f"卖出非成长股: {stock}")
            
    # 2. 计算等权买入资金
    target_num = len(g.security_list)
    if target_num == 0:
        return
        
    # 分配每只股票的目标价值
    total_value = context.portfolio.portfolio_value
    target_value_per_stock = total_value / target_num
    
    # 3. 买入或调整到目标仓位
    for stock in g.security_list:
        order_target_value(stock, target_value_per_stock)
        log.info(f"调仓/买入成长股: {stock}，目标市值: {target_value_per_stock:.2f}")

def handle_data(context, data):
    pass

注意事项

1. 流控限制：get_fundamentals 接口每秒调用不得超过 100 次，单次最大调用量为 500 条数据。若全市场股票较多，建议分批查询或先通过行业、市值等技术指标初筛缩小股票池后再调用财务接口。
2. 未来函数规避：在回测中，get_fundamentals 默认使用首次发布的数据（merge_type=None），能有效避免未来数据（如财报修正）对回测真实性的干扰。