PTrade 量化实战：如何计算 PEG 指标并筛选成长股？

在 PTrade 中实现 PEG（市盈率相对盈利增长比率）选股策略，核心在于获取**市盈率（PE）和净利润增长率（G）**数据，并计算 $PEG = PE / G$。

通常认为 PEG < 1 表示股票被低估，PEG 越低越好。

策略实现思路

1. 股票池：选取沪深300或中证500作为基础股票池。
2. 获取数据：
   • PE (市盈率)：使用 valuation 表中的 pe_ttm（滚动市盈率），比静态市盈率更能反映近期状况。
   • G (增长率)：使用 growth_ability 表中的 np_parent_company_yoy（归属母公司股东的净利润同比增长率）。
3. 数据清洗：
   • 剔除 PE < 0 的亏损股。
   • 剔除 增长率 < 0 的股票（PEG 无法有效评价负增长）。
   • 剔除停牌、ST 股票。
4. 计算 PEG：$PEG = \frac{PE_{TTM}}{GrowthRate}$。
5. 选股与交易：选取 PEG 最小的前 N 只股票进行等权持仓，按月调仓。

PTrade 策略代码

以下是完整的策略代码，您可以直接复制到 PTrade 回测环境中运行。

import pandas as pd
import numpy as np

def initialize(context):
    """
    策略初始化函数
    """
    # 设定基准为沪深300
    set_benchmark('000300.SS')
    # 设定滑点
    set_slippage(slippage=0.002)
    # 设定佣金
    set_commission(commission_ratio=0.0003, min_commission=5.0)
    
    # 全局变量设置
    g.index_code = '000300.SS'  # 股票池：沪深300
    g.stock_num = 10            # 持仓数量
    g.rebalance_month = 0       # 记录上一次调仓的月份
    
    # 设定按日运行，每天开盘时检查是否需要调仓
    run_daily(context, rebalance_strategy, time='09:30')

def before_trading_start(context, data):
    """
    盘前处理：获取成分股，过滤掉停牌和ST股
    """
    # 获取指数成分股
    stocks = get_index_stocks(g.index_code)
    
    # 过滤停牌和ST股票
    # 获取股票状态：ST, HALT(停牌), DELISTING(退市)
    # filter_stock_by_status 会返回剔除上述状态后的股票列表
    g.target_universe = filter_stock_by_status(stocks, filter_type=["ST", "HALT", "DELISTING"])

def rebalance_strategy(context):
    """
    核心调仓逻辑
    """
    # 获取当前月份
    current_month = context.blotter.current_dt.month
    
    # 判断是否换月，如果月份未变则不调仓
    if current_month == g.rebalance_month:
        return
    
    # 更新调仓月份记录
    g.rebalance_month = current_month
    log.info("开始进行月度调仓，当前时间: %s" % context.blotter.current_dt)

    # 1. 获取财务数据
    # pe_ttm: 市盈率(TTM)
    # np_parent_company_yoy: 归属母公司股东的净利润同比增长(%)
    
    # 注意：get_fundamentals 单次查询有数量限制，沪深300可以直接查，
    # 如果是全A股，建议分批查询或使用 query 语句（PTrade部分版本支持）
    # 这里演示直接查询列表
    
    df_val = get_fundamentals(g.target_universe, 'valuation', ['pe_ttm'], 
                              date=context.previous_date)
    
    df_growth = get_fundamentals(g.target_universe, 'growth_ability', ['np_parent_company_yoy'], 
                                 date=context.previous_date)
    
    # 如果数据获取失败，则跳过本次调仓
    if df_val is None or df_growth is None or df_val.empty or df_growth.empty:
        log.warning("财务数据获取失败，跳过本次调仓")
        return

    # 2. 数据合并与处理
    # 将两个DataFrame合并，索引通常是股票代码
    df = pd.concat([df_val, df_growth], axis=1, join='inner')
    
    # 剔除无效数据 (NaN)
    df = df.dropna()
    
    # 3. 筛选逻辑
    # 剔除 PE <= 0 的股票 (亏损股)
    df = df[df['pe_ttm'] > 0]
    
    # 剔除 增长率 <= 0 的股票 (负增长，PEG失效)
    # 注意：PTrade中增长率通常是百分比数值，例如 20.5 代表 20.5%
    df = df[df['np_parent_company_yoy'] > 0]
    
    # 4. 计算 PEG
    # PEG = PE / (增长率数值)
    # 这里的增长率是否除以100取决于个人定义，只要排序标准统一即可。
    # 彼得林奇通常直接用 PE / 增长率数值 (例如 PE 20, 增长 20%, PEG = 1)
    df['peg'] = df['pe_ttm'] / df['np_parent_company_yoy']
    
    # 5. 排序并选取 PEG 最小的前 N 只
    df = df.sort_values(by='peg', ascending=True)
    
    # 取前 g.stock_num 只股票代码
    target_stocks = df.index[:g.stock_num].tolist()
    log.info("本期选中股票 (PEG最小): %s" % target_stocks)
    
    # 6. 执行交易
    adjust_position(context, target_stocks)

def adjust_position(context, target_stocks):
    """
    执行具体的买卖操作
    """
    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    # 卖出不在目标列表中的股票
    for stock in current_positions:
        if stock not in target_stocks:
            order_target_value(stock, 0)
            log.info("卖出: %s" % stock)
            
    # 买入目标股票
    if len(target_stocks) > 0:
        # 等权分配资金
        # 注意：这里简单使用总资产/股票数量，实际交易中可能需要预留现金
        position_value = context.portfolio.portfolio_value / len(target_stocks)
        
        for stock in target_stocks:
            order_target_value(stock, position_value)
            log.info("买入/调整: %s, 目标金额: %.2f" % (stock, position_value))

def handle_data(context, data):
    """
    必须实现的函数，这里不需要做分钟级操作
    """
    pass

代码关键点解析

1. 数据获取 (get_fundamentals)：

   • valuation 表提供了 pe_ttm（滚动市盈率）。
   • growth_ability 表提供了 np_parent_company_yoy（净利润增长率）。
   • 注意：在回测中，date 参数建议传入 context.previous_date（前一交易日），以确保获取的是当时已发布的数据，避免未来函数。

2. PEG 计算逻辑：

   • 代码中使用了 df['pe_ttm'] / df['np_parent_company_yoy']。
   • 必须过滤掉 pe_ttm <= 0 和 np_parent_company_yoy <= 0 的数据，因为负值的 PEG 在选股逻辑中没有意义（例如：亏损且负增长的股票 PEG 可能是正数，但这显然不是好股票）。

3. 调仓频率：

   • 策略使用 g.rebalance_month 变量来控制按月调仓。每当 context.blotter.current_dt.month 发生变化时，触发一次调仓。

4. 风险控制：

   • 使用 filter_stock_by_status 剔除了 ST、停牌和退市股票，防止买入无法交易或风险过大的标的。

常见问题 (Q&A)

Q: 为什么使用 pe_ttm 而不是静态 PE？
A: 静态 PE 基于去年的年报，时效性较差。pe_ttm（滚动市盈率）包含了最近四个季度的财务数据，更能反映公司当前的估值水平。

Q: 净利润增长率是百分比还是小数？
A: 在 PTrade 的 get_fundamentals 返回的数据中，np_parent_company_yoy 通常是百分比数值（例如 30.5 代表 30.5%）。计算 PEG 时直接相除即可保持量级一致（即 PE=30, 增长=30, PEG=1）。

Q: 如何避免“未来函数”？
A: 在 get_fundamentals 中不指定 date 时，回测引擎通常会默认使用当前回测时间点已知的数据。但在代码中显式使用 date=context.previous_date 是一种更严谨的写法，确保使用的是昨天收盘后的静态数据。

Q: 如果我想加上 PEG < 1 的硬性过滤条件怎么办？
A: 您可以在排序前添加一行代码：

df = df[df['peg'] < 1.0]

但这可能导致选出的股票数量不足 g.stock_num，需要根据实际情况调整。