JoinQuant 进阶：基于营业利润增长率的成长型选股策略

在聚宽（JoinQuant）平台上，筛选“营业利润连续增长”的股票，最核心的方法是获取股票过去几年的财务数据，并比较其营业利润（operating_profit）是否呈现逐年递增的趋势。

实现这一逻辑主要有两种方式：

1. 使用 get_history_fundamentals（推荐）：一次性获取多期的历史财务数据，适合处理连续增长的逻辑。
2. 使用 get_fundamentals：查询特定年份的数据，需要自己处理年份逻辑。

下面我将提供一个完整的策略示例，使用 get_history_fundamentals 来筛选过去3年营业利润连续增长的股票，并进行月度调仓。

策略思路

1. 股票池：以沪深300成分股为例（为了回测速度，您可以改为全A股）。
2. 数据获取：获取过去3年的年度财务报表数据。
3. 筛选逻辑：
   • 最近一年营业利润 > 上一年营业利润 > 上上年营业利润。
   • 过滤掉ST股、停牌股、涨跌停股票。
4. 交易执行：每月第一个交易日进行调仓，等权重买入筛选出的股票。

策略代码

# -*- coding: utf-8 -*-
from jqdata import *
import pandas as pd

def initialize(context):
    # 设定基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # 设定交易费率
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 设定每月第一个交易日运行选股函数
    run_monthly(trade_func, monthday=1, time='09:30')

def trade_func(context):
    # 1. 获取初始股票池 (这里以沪深300为例，您可以改为 get_all_securities(['stock']).index.tolist())
    target_list = get_index_stocks('000300.XSHG')
    
    # 2. 筛选营业利润连续3年增长的股票
    growth_stocks = filter_continuous_growth(context, target_list, years=3)
    
    # 3. 进一步过滤 (ST, 停牌, 涨跌停)
    final_list = filter_basic_conditions(context, growth_stocks)
    
    # 4. 执行交易
    adjust_position(context, final_list)

def filter_continuous_growth(context, security_list, years=3):
    """
    筛选营业利润连续增长的股票
    :param security_list: 股票列表
    :param years: 连续增长的年数
    :return: 符合条件的股票列表
    """
    # 获取历史财务数据
    # stat_by_year=True 表示获取年度数据
    # count=years 表示获取过去N年的年报
    # interval='1y' 表示间隔1年
    df = get_history_fundamentals(
        security_list, 
        fields=[income.operating_profit], 
        watch_date=context.previous_date, 
        count=years, 
        interval='1y', 
        stat_by_year=True
    )
    
    if df.empty:
        return []
    
    # 数据处理：将长表转换为宽表，索引为股票代码，列为报告期(statDate)，值为营业利润
    # df原始列: code, pubDate, statDate, operating_profit
    pivot_df = df.pivot(index='code', columns='statDate', values='operating_profit')
    
    # 去除有空值的行（意味着数据不足3年）
    pivot_df = pivot_df.dropna()
    
    # 确保列是按时间排序的
    pivot_df = pivot_df.sort_index(axis=1)
    
    # 核心逻辑：判断是否连续增长
    # diff(axis=1) 计算列与列之间的差值，如果差值都大于0，说明是递增的
    # .min(axis=1) > 0 确保每一年的增量都大于0
    # pivot_df.diff(axis=1).iloc[:, 1:] 忽略第一列的NaN(因为第一列没有前一列可减)
    is_growing = (pivot_df.diff(axis=1).iloc[:, 1:] > 0).all(axis=1)
    
    # 获取符合条件的股票代码列表
    selected_stocks = is_growing[is_growing].index.tolist()
    
    log.info(f"筛选出营业利润连续{years}年增长的股票数量: {len(selected_stocks)}")
    return selected_stocks

def filter_basic_conditions(context, stock_list):
    """
    过滤ST、停牌、涨跌停
    """
    current_data = get_current_data()
    final_stocks = []
    for stock in stock_list:
        # 过滤ST
        if current_data[stock].is_st:
            continue
        # 过滤停牌
        if current_data[stock].paused:
            continue
        # 过滤涨停 (买不进)
        if current_data[stock].last_price >= current_data[stock].high_limit:
            continue
        # 过滤跌停 (卖不出，虽然这里是买入逻辑，但一般也过滤掉)
        if current_data[stock].last_price <= current_data[stock].low_limit:
            continue
        
        final_stocks.append(stock)
    return final_stocks

def adjust_position(context, buy_list):
    """
    调仓逻辑：全仓等权重买入
    """
    # 获取当前持仓
    holding_list = list(context.portfolio.positions.keys())
    
    # 卖出不在买入列表中的股票
    for stock in holding_list:
        if stock not in buy_list:
            order_target_value(stock, 0)
    
    # 买入股票
    if len(buy_list) > 0:
        # 分配资金
        value_per_stock = context.portfolio.total_value / len(buy_list)
        for stock in buy_list:
            order_target_value(stock, value_per_stock)
    else:
        log.info("无符合条件的股票，空仓运行")

代码关键点解析

1. get_history_fundamentals 函数：

   • 这是实现“连续增长”逻辑的神器。
   • fields=[income.operating_profit]：指定查询营业利润字段。
   • stat_by_year=True 和 interval='1y'：强制获取年度报告数据，避免季度数据的季节性波动影响判断。
   • count=3：获取最近3年的年报数据。
   • watch_date=context.previous_date：确保获取的是回测当时已经发布的数据，避免未来函数。

2. 数据透视与比较 (pivot & diff)：

   • get_history_fundamentals 返回的是长格式数据（每一行是一条记录）。
   • 使用 pivot 将其转换为宽格式（行是股票，列是年份）。
   • 使用 diff(axis=1) 计算相邻年份的差值。如果所有差值都大于0，即表示数据是单调递增的。

3. 基本面数据的滞后性：

   • 请注意，年报通常在次年的4月底之前发布。例如在 2023年1月时，最新的完整年报可能还是 2021年的（因为2022年报还没发）。get_history_fundamentals 会自动处理这个问题，只返回当时已公开的数据。

常见问题 Q&A

Q: 如果我想筛选“归母净利润”连续增长怎么办？
A: 将代码中的 income.operating_profit 替换为 income.np_parent_company_owners 即可。

Q: 如果我想筛选“季度”同比增长（例如连续3个季度同比增长）？
A: 需要将 stat_by_year 设为 False，interval 设为 '1q'，并且逻辑会稍微复杂一些（需要比较 2023q1 > 2022q1 而不是 2023q2 > 2023q1），通常建议使用 get_fundamentals 配合自定义查询来实现同比逻辑。但对于年度连续增长，上述代码是最优解。

Q: 为什么选出的股票很少？
A: 连续3年营业利润增长是一个比较严格的条件，尤其是在市场环境不好的年份。您可以尝试将股票池扩大到全A股，或者放宽条件（例如允许其中一年微跌，或者只看最近2年）。