JoinQuant 基本面分析：基于经营活动现金流（Net Operate Cash Flow）的选股

在聚宽（JoinQuant）平台上，筛选“经营活动现金流净额持续增长”的股票，主要涉及到对财务数据的获取和历史比较。

核心思路

1. 数据获取：使用 get_history_fundamentals API。这个函数可以一次性获取多只股票、多个报告期（年度或季度）的历史财务数据，比循环调用 get_fundamentals 效率更高。
2. 财务指标：关注 cash_flow 表中的 net_operate_cash_flow（经营活动产生的现金流量净额）字段。
3. 筛选逻辑：
   • 设定考察年限（例如过去3年）。
   • 获取指定股票池（如沪深300或全A股）的历史数据。
   • 按股票代码分组，判断每只股票的现金流数值是否满足 $Value_t > Value_{t-1} > Value_{t-2}$ 的严格递增关系。
4. 周期设置：通常基本面策略不需要每天调仓，建议按月或按季度运行。

策略代码实现

以下是一个完整的策略代码示例。该策略每月第一个交易日运行，筛选出过去3年（年度数据）经营活动现金流净额持续增长的沪深300成分股，并进行等权重买入。

# -*- coding: utf-8 -*-
# 导入函数库
from jqdata import *
import pandas as pd

def initialize(context):
    # 设定沪深300作为基准
    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')
    
    # 定义全局变量，保存要操作的股票
    g.buy_list = []
    
    # 设置按月运行，每月第一个交易日进行选股和调仓
    run_monthly(trade_func, monthday=1, time='09:30')

def trade_func(context):
    # 1. 获取股票池：这里以沪深300为例，也可以换成 get_all_securities(['stock']).index.tolist()
    stock_pool = get_index_stocks('000300.XSHG')
    
    # 2. 选股：筛选经营活动现金流净额连续3年增长的股票
    g.buy_list = filter_cash_flow_growth(stock_pool, context.previous_date, years=3)
    
    log.info(f"今日选出 {len(g.buy_list)} 只现金流持续增长的股票")
    
    # 3. 执行调仓
    rebalance(context)

def filter_cash_flow_growth(stock_list, watch_date, years=3):
    """
    筛选经营现金流持续增长的股票
    :param stock_list: 股票列表
    :param watch_date: 观察日期
    :param years: 考察的年数
    :return: 符合条件的股票列表
    """
    # 获取历史财务数据
    # stat_by_year=True 表示获取年度数据，避免季节性波动影响
    # interval='1y' 表示间隔为1年
    # count=years 获取过去N年的数据
    df = get_history_fundamentals(
        security=stock_list,
        fields=[cash_flow.net_operate_cash_flow],
        watch_date=watch_date,
        stat_date=None, # 使用watch_date前发布的报表
        count=years,
        interval='1y',
        stat_by_year=True
    )
    
    if df.empty:
        return []
    
    # 数据预处理：去除空值
    df = df.dropna()
    
    # 筛选逻辑
    selected_stocks = []
    
    # 按股票代码分组处理
    grouped = df.groupby('code')
    
    for code, group in grouped:
        # 如果数据行数不足（可能刚上市不久），跳过
        if len(group) < years:
            continue
        
        # 获取现金流数据并按时间排序（get_history_fundamentals返回通常是按时间倒序或正序，这里确保按statDate升序）
        # 注意：get_history_fundamentals 返回的 statDate 是字符串或日期，需要确保排序正确
        group = group.sort_values(by='statDate', ascending=True)
        
        cash_flows = group['net_operate_cash_flow'].values
        
        # 判断是否严格递增
        # 使用 all() 函数比较数组：后一年 > 前一年
        # cash_flows[1:] 取从第2个元素开始的所有元素
        # cash_flows[:-1] 取从第1个元素到倒数第2个元素
        # 比较对应位置：Year2 > Year1, Year3 > Year2
        if np.all(np.diff(cash_flows) > 0):
            # 额外条件：最近一年现金流必须为正（可选，防止虽然增长但仍是巨额负债减少的情况）
            if cash_flows[-1] > 0:
                selected_stocks.append(code)
                
    return selected_stocks

def rebalance(context):
    """
    调仓函数
    """
    # 获取当前持仓
    current_positions = context.portfolio.positions.keys()
    
    # 卖出不在买入列表中的股票
    for stock in current_positions:
        if stock not in g.buy_list:
            order_target_value(stock, 0)
    
    # 买入新选出的股票
    if len(g.buy_list) > 0:
        # 等权重分配资金
        position_count = len(g.buy_list)
        value_per_stock = context.portfolio.total_value / position_count
        
        for stock in g.buy_list:
            order_target_value(stock, value_per_stock)

代码关键点解析

1. get_history_fundamentals:

   • 这是实现历史比较的核心函数。
   • stat_by_year=True: 这一点非常重要。经营活动现金流受季节性影响很大（例如某些行业年底回款），直接比较季度环比（Q2 vs Q1）往往没有意义。开启此选项可以获取年度报告数据，比较 Year(T) vs Year(T-1)，更能反映企业的长期经营趋势。
   • interval='1y': 配合 stat_by_year，指定获取数据的间隔为1年。
   • count=3: 获取过去3年的年报数据。

2. 增长判断逻辑 (np.diff):

   • 我们先将数据按 statDate（统计日期）升序排列。
   • np.diff(cash_flows) 计算相邻元素的差值。
   • np.all(... > 0) 确保所有的差值都大于0，即实现了严格的持续增长判断。

3. 容错处理:

   • 增加了 dropna() 去除数据缺失的行。
   • 增加了 len(group) < years 判断，剔除上市时间不足或财报缺失的股票。

常见问题 Q&A

Q: 为什么要用 stat_by_year=True 而不是直接取单季度数据？
A: 现金流数据具有累积性和季节性。单季度数据（如Q2）通常是累积值（1月-6月），直接对比Q2和Q1（1月-3月）虽然能算出单季增量，但不同季度的业务模式不同。对比年度数据（2022年报 vs 2021年报）能更准确地反映企业经营能力的提升。

Q: 如果我想筛选“季度同比”增长（例如今年Q1比去年Q1增长）怎么办？
A: 可以将 get_history_fundamentals 的参数改为 interval='1q', count=5，然后取出索引为 0 (去年同期) 和 4 (当前) 的数据进行比较，或者使用 stat_by_year=False 获取连续季度数据后，自行编写逻辑比较 df.iloc[-1] 和 df.iloc[-5]。

Q: 这个策略可以直接实盘吗？
A: 该代码仅作为筛选逻辑的演示（Demo）。实盘策略通常还需要结合估值因子（PE/PB）、技术指标（均线/动量）以及风控模型（止损/仓位管理）来构建完整的投资体系。单纯依靠现金流增长可能会选中估值过高的股票。