4. PEG指标选股策略

任务目标

实现一个基于PEG（市盈率相对盈利增长比率）的选股策略：筛选PEG小于1的成长股，并在A股或指定股票池上回测该策略相对基准（如沪深300）的超额收益。PEG = PE(TTM) / 盈利增长率（预期或历史复合增长率）。本案例采用历史3年净利润复合增长率作为盈利增长率的代理指标，以降低未来预期数据的不确定性。

环境与依赖

需要Python 3.8+，并安装以下库：pandas、numpy、akshare（用于获取财务与行情数据）。回测部分自行实现简化逻辑，不依赖第三方回测框架。

数据获取与预处理

使用akshare获取指定股票池的PE(TTM)、净利润数据及收盘价。核心处理逻辑如下：

import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def get_financial_data(symbols):
    """获取股票池的PE和净利润数据，返回DataFrame"""
    all_data = []
    for sym in symbols:
        try:
            # 获取个股财务指标（含净利润）
            fin = ak.stock_financial_abstract(symbol=sym)
            # 获取PE(TTM)
            spot = ak.stock_individual_spot_xq(symbol=sym)
            pe_ttm = spot[spot['item'] == '市盈率(TTM)']['value'].values[0]
            
            # 提取近4年净利润（每年年报）
            profit = fin[fin['报告期'].str.contains('12-31')]['净利润'].head(4).astype(float)
            if len(profit) < 4:
                continue
            # 计算3年复合增长率
            cagr = (profit.iloc[0] / profit.iloc[-1]) ** (1/3) - 1
            if cagr <= 0:
                continue
            peg = pe_ttm / (cagr * 100)
            all_data.append({'symbol': sym, 'pe': pe_ttm, 'cagr': cagr, 'peg': peg})
        except Exception as e:
            print(f"跳过 {sym}: {e}")
            continue
    return pd.DataFrame(all_data)

def select_stocks(data, peg_threshold=1.0):
    """筛选PEG小于阈值的股票"""
    return data[data['peg'] < peg_threshold].copy()

回测逻辑实现

回测周期设定为2023年1月1日至2023年12月31日，每月第一个交易日调仓，等权持有筛选出的股票，基准为沪深300指数。核心回测代码如下：

def backtest_strategy(symbols, start_date='2023-01-01', end_date='2023-12-31'):
    """简化月度调仓回测，返回策略与基准净值序列"""
    # 获取交易日历
    trade_dates = ak.tool_trade_date_hist_sina()
    trade_dates = trade_dates['trade_date'].astype(str)
    dates = [d for d in trade_dates if start_date <= d <= end_date]
    # 每月第一个交易日
    month_first = []
    for d in dates:
        if not month_first or d[:7] != month_first[-1][:7]:
            month_first.append(d)
    
    # 获取基准（沪深300）数据
    bench = ak.index_zh_a_hist(symbol='000300', period='daily', start_date=start_date, end_date=end_date)
    bench = bench.set_index('日期')
    
    # 获取股票池日线数据（简化：使用前复权收盘价）
    price_data = {}
    for sym in symbols:
        try:
            df = ak.stock_zh_a_hist(symbol=sym, period='daily', start_date=start_date, end_date=end_date, adjust='qfq')
            price_data[sym] = df.set_index('日期')['收盘']
        except:
            continue
    
    # 逐月调仓
    nav = 1.0
    nav_series = []
    bench_nav = 1.0
    bench_series = []
    current_holdings = []
    
    for i, date in enumerate(month_first):
        # 获取当月财务数据并选股
        if i == 0 or i % 3 == 0:  # 每季度更新财务数据
            fin_data = get_financial_data(symbols)
            selected = select_stocks(fin_data)
            current_holdings = selected['symbol'].tolist()
        
        # 计算当月收益（从当月第一个交易日到次月第一个交易日）
        end_idx = i+1 if i+1 < len(month_first) else len(dates)-1
        period_start = date
        period_end = month_first[i+1] if i+1 < len(month_first) else dates[-1]
        
        # 策略收益
        if current_holdings:
            returns = []
            for sym in current_holdings:
                if sym in price_data and period_start in price_data[sym].index and period_end in price_data[sym].index:
                    ret = price_data[sym][period_end] / price_data[sym][period_start] - 1
                    returns.append(ret)
            if returns:
                nav *= (1 + np.mean(returns))
        
        # 基准收益
        if period_start in bench.index and period_end in bench.index:
            bench_nav *= (1 + bench.loc[period_end, '收盘'] / bench.loc[period_start, '收盘'] - 1)
        
        nav_series.append({'date': period_end, 'strategy': nav})
        bench_series.append({'date': period_end, 'benchmark': bench_nav})
    
    return pd.DataFrame(nav_series).set_index('date'), pd.DataFrame(bench_series).set_index('date')

运行与验证

# 定义股票池（示例：沪深300成分股前20只，实际使用时可替换为完整列表）
sample_symbols = ['600519', '000858', '601318', '600036', '000333', '600900', '601166', '600276', '000651', '600887']

# 执行回测
strategy_nav, bench_nav = backtest_strategy(sample_symbols)

# 计算超额收益
result = pd.concat([strategy_nav, bench_nav], axis=1).dropna()
result['excess'] = result['strategy'] - result['benchmark']
print("策略最终净值:", result['strategy'].iloc[-1])
print("基准最终净值:", result['benchmark'].iloc[-1])
print("超额收益:", result['excess'].iloc[-1])

# 验证条件：策略净值 > 基准净值，即超额收益为正
if result['excess'].iloc[-1] > 0:
    print("✅ 策略跑赢基准，PEG选股有效")
else:
    print("❌ 策略未跑赢基准，需调整参数或股票池")

验证方法与判定标准

• 运行方式：直接执行上述Python脚本，确保网络可访问akshare数据源。
• 通过条件：脚本正常输出策略净值、基准净值和超额收益数值，且超额收益为正（即策略净值大于基准净值）。
• 预期输出示例（基于给定股票池和2023年数据，实际数值因数据源更新而异）：策略最终净值约1.05，基准净值约0.98，超额收益约0.07。该输出仅为逻辑推导的示例，非实测结果。

常见问题

• 数据获取失败：akshare接口可能因网络或频率限制失败，可增加重试机制或更换数据源。
• 净利润为负：计算CAGR时需保证首尾年净利润为正，否则跳过该股票。
• 调仓频率：本案例采用月度调仓、季度更新财务数据，实际使用中可根据需要调整。