ptrade 财务选股：基于现金流净额与净利润比率的质量选股

在股票量化投资中，仅看公司净利润可能存在财务操纵或应收账款过高的风险。**经营活动产生的现金流量净额与净利润的比率（现金净利润比）**是衡量企业盈利质量的核心指标。当该比率大于 1 时，说明企业的净利润有实打实的现金流入支撑。

本文将介绍如何在 ptrade 平台中使用 get_fundamentals 接口，构建基于盈利质量的量化选股策略。

1. 核心接口说明

在 ptrade 中，获取财务数据主要依赖 get_fundamentals 函数。

• 利润表（income_statement）：获取净利润（net_profit）。
• 现金流量表（cashflow_statement）：获取经营活动产生的现金流量净额（net_operate_cash_flow）。

接口调用示例：

# 获取指定股票的现金流量表数据
cf_df = get_fundamentals(stocks, 'cashflow_statement', fields=['net_operate_cash_flow'], date=current_date)

# 获取指定股票的利润表数据
in_df = get_fundamentals(stocks, 'income_statement', fields=['net_profit'], date=current_date)

2. 选股逻辑设计

1. 全市场/指数成分股筛选：获取全市场 A 股或沪深 300 成分股。
2. 过滤异常股：剔除 ST、停牌及退市股票。
3. 获取财务数据：查询上季度或最新公布的净利润与经营现金流净额。
4. 计算质量指标：
   $$\text{盈利质量比率} = \frac{\text{经营活动现金流量净额}}{\text{净利润}}$$
5. 构建股票池：筛选出净利润大于 0，且盈利质量比率大于 1.2 的优质标的，按比率降序排列选择前 N 只股票。

3. ptrade 完整策略源码

以下为可在 ptrade 回测与交易模块中直接运行的源码：

import pandas as pd
import numpy as np

def initialize(context):
    # 设置基准
    set_benchmark('000300.SS')
    # 设置初始持仓股票数
    g.hold_count = 10
    # 设为每日运行
    run_daily(context, rebalance, time='09:35')

def rebalance(context):
    # 1. 获取全市场 A 股列表
    stocks = get_Ashares()
    
    # 2. 过滤 ST、停牌、退市股票
    stocks = filter_stock_by_status(stocks, filter_type=["ST", "HALT", "DELISTING"])
    
    # 3. 获取上一个交易日日期
    prev_date = str(get_trading_day(-1)).replace('-', '')
    
    # 4. 获取经营现金流净额与净利润
    try:
        cf_data = get_fundamentals(stocks, 'cashflow_statement', fields=['net_operate_cash_flow'], date=prev_date)
        in_data = get_fundamentals(stocks, 'income_statement', fields=['net_profit'], date=prev_date)
    except Exception as e:
        log.error("获取财务数据失败: %s" % e)
        return
    
    # 5. 合并数据并计算比率
    df = pd.merge(cf_data, in_data, left_index=True, right_index=True)
    df = df.dropna()
    
    # 过滤净利润大于0的企业
    df = df[df['net_profit'] > 0]
    
    # 计算现金净利润比率
    df['quality_ratio'] = df['net_operate_cash_flow'] / df['net_profit']
    
    # 筛选比率大于 1.2 的高盈利质量公司
    target_df = df[df['quality_ratio'] > 1.2]
    
    # 按比率降序排序并取前 N 只
    target_df = target_df.sort_values(by='quality_ratio', ascending=False)
    target_stocks = target_df.index[:g.hold_count].tolist()
    
    log.info("今日选股股票池: %s" % target_stocks)
    
    # 6. 调仓卖出不在目标池的股票
    for stock in list(context.portfolio.positions.keys()):
        if stock not in target_stocks:
            order_target(stock, 0)
            log.info("卖出非目标股: %s" % stock)
            
    # 7. 买入目标股票（均分资金）
    if len(target_stocks) > 0:
        cash_per_stock = context.portfolio.cash / len(target_stocks)
        for stock in target_stocks:
            if get_position(stock).amount == 0:
                order_value(stock, cash_per_stock)
                log.info("买入高质量盈利股: %s, 目标金额: %s" % (stock, cash_per_stock))

def handle_data(context, data):
    pass

4. 注意事项与优化建议

1. 数据频次与延时：财务数据通过 HTTP 在线获取，建议在 before_trading_start 或每日定时任务（run_daily）中调用，并加入 try-except 异常保护。
2. 流量限制：若全市场股票量较大，建议分批获取或先通过市值/行业缩小标的范围（如配合 get_index_stocks('000906.XBHS') 中证 800），以避免触发接口限流。
3. 结合估值指标：盈利质量选股可以配合 valuation 表中的 pe_ttm（市盈率）或 pb（市净率）进行二次筛选，剔除估值过高的标的。