问题描述
解决方案
PEG(市盈率相对盈利增长比率)是一个经典的价值投资指标,由吉姆·斯莱特发明,彼得·林奇发扬光大。其核心思想是在看重市盈率(PE)的同时,结合公司的盈利增长速度(Growth)。
PEG = 市盈率 (PE) / 盈利增长率 (G)
- PEG < 1:通常被认为股票被低估(增长速度高于估值)。
- PEG = 1:估值合理。
- PEG > 1:通常被认为股票被高估。
下面是一个基于聚宽(JoinQuant)平台的 PEG 选股策略实现。
策略逻辑
- 选股范围:全市场股票(剔除 ST、停牌、次新股)。
- 数据获取:获取市盈率(PE-TTM)和净利润增长率(同比增长率)。
- 筛选条件:
- 市盈率 (PE) > 0 且 < 60(剔除亏损和极高估值)。
- 净利润增长率 (G) > 15%(保证成长性)。
- 计算 PEG = PE / G。
- 筛选 PEG < 1 的股票。
- 排序与买入:按 PEG 从小到大排序,买入前 20 只。
- 调仓频率:按月调仓。
策略代码
# -*- coding: utf-8 -*-
from jqdata import *
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.stock_num = 20
# 设定每月第一个交易日进行调仓
run_monthly(trade, 1)
def trade(context):
"""
交易函数:选股并执行买卖
"""
# 1. 获取选股列表
target_list = get_peg_stock_list(context)
# 2. 获取当前持仓
current_holdings = list(context.portfolio.positions.keys())
# 3. 卖出不在目标列表中的股票
for stock in current_holdings:
if stock not in target_list:
order_target_value(stock, 0)
# 4. 买入目标列表中的股票
# 资金分配:将剩余资金+卖出资金平均分配给要买入的股票
# 注意:这里简单处理,实际资金可能因为卖出而变化,使用 order_target_value 调整至目标权重
if len(target_list) > 0:
# 计算每只股票的目标持仓价值(总资产 / 目标持仓数)
# 这里的逻辑是等权重持有
target_value = context.portfolio.total_value / len(target_list)
for stock in target_list:
order_target_value(stock, target_value)
def get_peg_stock_list(context):
"""
核心选股逻辑:计算PEG并筛选
"""
# 获取当前时间
date = context.previous_date
# 查询财务数据
# valuation.pe_ratio: 市盈率(TTM)
# indicator.inc_net_profit_year_on_year: 净利润同比增长率(%)
q = query(
valuation.code,
valuation.pe_ratio,
indicator.inc_net_profit_year_on_year
).filter(
valuation.pe_ratio > 0, # 剔除亏损
valuation.pe_ratio < 60, # 剔除极高估值
indicator.inc_net_profit_year_on_year > 15 # 保证一定的成长性(>15%)
)
df = get_fundamentals(q, date=date)
# 过滤掉停牌、ST、退市的股票
# 获取当前所有未停牌、非ST的股票列表
valid_stocks = filter_paused_and_st_stock(df['code'].tolist(), date)
df = df[df['code'].isin(valid_stocks)]
# 计算 PEG
# 注意:聚宽的增长率是百分数(如 20 代表 20%),PE 是绝对值
# PEG = PE / (增长率数值)
df['peg'] = df['pe_ratio'] / df['inc_net_profit_year_on_year']
# 筛选 PEG 区间 (0 < PEG < 1)
# PEG < 0 意味着增长率为负(前面已过滤)或 PE 为负(前面已过滤),这里再次确保安全
df = df[(df['peg'] > 0) & (df['peg'] < 1)]
# 按照 PEG 从小到大排序
df = df.sort_values(by='peg', ascending=True)
# 取前 N 只股票
stock_list = df['code'].head(g.stock_num).tolist()
return stock_list
def filter_paused_and_st_stock(stock_list, date):
"""
过滤停牌、ST、退市股票
"""
current_data = get_price(stock_list, end_date=date, frequency='daily', fields=['paused'], count=1, panel=False)
# 获取停牌股票代码
if not current_data.empty:
paused_stocks = current_data[current_data['paused'] == 1]['code'].tolist()
else:
paused_stocks = []
# 获取ST和退市股票
is_st = get_extras('is_st', stock_list, start_date=date, end_date=date, df=True)
if not is_st.empty:
st_stocks = is_st[is_st.iloc[0] == True].columns.tolist()
else:
st_stocks = []
# 过滤
return [stock for stock in stock_list if stock not in paused_stocks and stock not in st_stocks]
代码关键点解析
-
数据获取 (
get_fundamentals):- 使用了
valuation.pe_ratio获取市盈率(TTM)。 - 使用了
indicator.inc_net_profit_year_on_year获取净利润同比增长率。这是计算 PEG 中 "G" 的关键。
- 使用了
-
PEG 计算:
- 公式:
df['peg'] = df['pe_ratio'] / df['inc_net_profit_year_on_year']。 - 注意:在聚宽数据库中,增长率如果是 20%,存储的数值是
20而不是0.2。因此直接相除即可得到标准的 PEG 值。
- 公式:
-
风险控制:
- 过滤 ST 和停牌:通过
filter_paused_and_st_stock函数剔除无法交易或风险过大的股票。 - PE 范围限制:限制
0 < PE < 60,避免买入亏损股(PE<0)或市场泡沫过大的股票。 - 增长率门槛:要求增长率大于 15%,确保选出的股票具有真实的成长动力,避免 PEG 低是因为 PE 极低但公司已停止增长的“价值陷阱”。
- 过滤 ST 和停牌:通过
-
交易执行:
- 使用
order_target_value进行调仓,这会自动计算买卖数量,使持仓市值达到目标金额(等权重模型)。
- 使用
常见问题 Q&A
Q: 为什么 PEG 策略有时候会选到周期股?
A: 在周期股(如钢铁、航运)业绩爆发的年份,PE 会很低,增长率 G 会极高,导致 PEG 极低。但这往往是周期的顶点。为了优化,可以在筛选条件中加入行业限制,或者要求连续多个季度的增长。
Q: 如何避免“未来函数”?
A: 代码中使用了 get_fundamentals,在回测模式下,聚宽会自动处理数据披露时间。例如在 4 月 1 日调仓时,系统只会使用 3 月 31 日之前已经公告的财报数据,不会使用 4 月 15 日才发布的财报数据,从而避免未来函数。
Q: 可以把调仓频率改成每周吗?
A: 可以,将 run_monthly(trade, 1) 改为 run_weekly(trade, 1)。但基本面数据(财报)通常按季度更新,过于频繁的调仓可能不会带来显著收益提升,反而增加手续费。