PTrade 基本面选股：EPS 连续增长策略

在PTrade中筛选EPS（每股收益）连续增长的股票，主要依赖于 get_fundamentals 函数来获取财务数据。

核心思路

1. 获取股票池：通常在 before_trading_start 中获取目标股票池（如沪深300、中证500或全A股）。
2. 获取财务数据：使用 get_fundamentals 查询 eps 表中的 basic_eps（基本每股收益）字段。
3. 设置查询参数：
   • report_types='4'：指定查询年报数据，以确保比较的是年度增长，避免季节性波动。
   • start_year 和 end_year：根据当前回测时间，向前推算需要查询的年份区间。
4. 逻辑判断：遍历每只股票的历史EPS数据，判断是否满足 $EPS_{当年} > EPS_{去年} > EPS_{前年}$ 的递增逻辑。

策略代码实现

以下是一个完整的策略示例，该策略会筛选出过去3年EPS连续增长的沪深300成分股，并进行等权买入。

def initialize(context):
    """
    初始化函数
    """
    # 设定筛选连续增长的年数
    g.growth_years = 3 
    # 设定操作的指数代码（这里以沪深300为例）
    g.index_code = '000300.SS'
    # 每日筛选出的股票列表
    g.target_stocks = []

def before_trading_start(context, data):
    """
    盘前处理：每日开盘前筛选股票
    """
    # 1. 获取股票池（建议在盘前获取，不要在initialize中获取）
    stocks = get_index_stocks(g.index_code)
    
    # 2. 确定查询年份
    # 获取当前回测日期的年份
    current_year = context.blotter.current_dt.year
    # 由于年报通常在次年4月底前披露完毕，为了避免未来函数，
    # 我们查询当前年份的前一年作为最近的一个完整财年。
    # 例如：当前是2023年，我们查看2022, 2021, 2020的数据
    last_complete_year = current_year - 1
    start_query_year = last_complete_year - g.growth_years
    
    # 转换年份为字符串
    start_year_str = str(start_query_year)
    end_year_str = str(last_complete_year)
    
    # 3. 获取财务数据
    # 表名: eps (每股指标)
    # 字段: basic_eps (基本每股收益)
    # report_types='4' 代表年报
    # 注意：按年份查询返回的是一个Panel对象（在PTrade Python3环境中表现为多维结构）
    # 索引通常为 [股票代码, 字段名, 日期] 或类似结构，具体取决于PTrade版本，
    # 但通常我们可以通过 df[stock] 获取该股票的时间序列DataFrame
    funda_data = get_fundamentals(
        stocks, 
        'eps', 
        'basic_eps', 
        start_year=start_year_str, 
        end_year=end_year_str, 
        report_types='4'
    )
    
    g.target_stocks = []
    
    # 4. 遍历股票进行筛选
    if funda_data is not None:
        # 这里的funda_data是一个Panel，Items是股票代码
        for stock in stocks:
            if stock in funda_data:
                # 获取该股票的EPS数据 DataFrame
                # 行索引是 end_date (会计周期截止日)
                stock_df = funda_data[stock]
                
                # 确保数据按时间升序排列
                stock_df = stock_df.sort_index()
                
                # 获取 basic_eps 列的数据
                eps_series = stock_df['basic_eps']
                
                # 剔除空值
                eps_values = eps_series.dropna().values
                
                # 数据长度必须满足我们要比较的年数 (例如比较3年增长，至少需要4年的数据来计算增长，或者直接比较绝对值)
                # 这里我们比较绝对值：EPS_2022 > EPS_2021 > EPS_2020
                if len(eps_values) >= g.growth_years:
                    # 截取最近的 N 年数据
                    recent_eps = eps_values[-g.growth_years:]
                    
                    # 判断是否严格递增
                    is_growing = True
                    for i in range(1, len(recent_eps)):
                        if recent_eps[i] <= recent_eps[i-1]:
                            is_growing = False
                            break
                    
                    # 额外条件：最近一年EPS必须大于0（剔除亏损股）
                    if is_growing and recent_eps[-1] > 0:
                        g.target_stocks.append(stock)
    
    log.info("今日筛选出EPS连续增长的股票数量: %s" % len(g.target_stocks))
    # 更新股票池以便handle_data使用
    set_universe(g.target_stocks)

def handle_data(context, data):
    """
    盘中交易逻辑
    """
    # 如果没有筛选出股票，清仓
    if len(g.target_stocks) == 0:
        for stock in get_positions():
            order_target(stock, 0)
        return

    # 1. 卖出不在目标列表中的股票
    current_positions = get_positions()
    for stock in current_positions:
        if stock not in g.target_stocks:
            order_target(stock, 0)
            log.info("卖出不再满足增长条件的股票: %s" % stock)
    
    # 2. 买入目标股票（等权分配资金）
    # 获取当前可用资金
    cash = context.portfolio.cash
    # 计算每只股票应分配的资金（简单示例，未考虑手续费预留）
    # 实际交易中建议预留一部分资金防止手续费导致资金不足
    if len(g.target_stocks) > 0:
        # 假设我们持有不超过10只，或者全仓买入筛选出的所有
        # 这里演示全仓等权买入
        position_count = len(context.portfolio.positions)
        available_slots = len(g.target_stocks)
        
        # 简单的资金分配逻辑：总资产 / 目标股票数量
        total_value = context.portfolio.portfolio_value
        target_value_per_stock = total_value / len(g.target_stocks)
        
        for stock in g.target_stocks:
            order_target_value(stock, target_value_per_stock)

代码关键点解析

1. get_fundamentals 的使用：

   • 我们使用了 start_year 和 end_year 参数。这种模式下，PTrade 返回的是一个 pandas.Panel 对象（在某些Python版本中可能表现为字典或多级索引DataFrame）。
   • 代码中通过 funda_data[stock] 来获取单只股票的数据，这是处理 Panel 数据的常用方法。
   • report_types='4' 非常重要，它保证了我们获取的是年报数据。如果使用季报（如一季报、中报），由于季节性因素，直接比较连续季度的EPS往往没有意义（例如很多行业Q4是旺季，Q1是淡季）。

2. 时间处理：

   • last_complete_year = current_year - 1：这是为了防止“未来函数”。在回测中，如果当前是2022年1月，2021年的年报可能还没出。PTrade的 get_fundamentals 在回测模式下，如果数据未发布，通常会填充NaN或不返回。但在编写逻辑时，明确指定查询过去年份是比较稳妥的做法。

3. 增长逻辑：

   • 代码通过循环 recent_eps[i] <= recent_eps[i-1] 来判断是否打破了增长趋势。只要有一年不增长（小于或等于前一年），is_growing 就变为 False。
   • 增加了 recent_eps[-1] > 0 的判断，确保最近一年是盈利的。

4. 交易逻辑：

   • 示例中使用了简单的“持有即轮动”逻辑：每天（或调仓日）检查，不在名单里的卖出，在名单里的买入并调整仓位至等权。

Q&A

Q: 为什么使用年报（report_types='4'）而不是单季度数据？
A: 筛选“连续增长”通常指公司的长期基本面改善。单季度数据容易受季节性波动影响（例如白酒春节旺季、空调夏季旺季）。如果确实需要比较季度增长，通常应使用“单季度同比”（例如今年Q1对比去年Q1），但这需要更复杂的数据处理（如 basic_eps 减去上期累计值）。使用年报是最直接、最稳健的判断长期增长的方法。

Q: 如果我想筛选“净利润”连续增长而不是EPS，怎么改？
A: 只需修改 get_fundamentals 中的表名和字段名。

• 将表名 'eps' 改为 'income_statement'（利润表）。
• 将字段 'basic_eps' 改为 'net_profit'（净利润）或 'np_parent_company_owners'（归母净利润）。

Q: 这个策略在回测中会用到未来数据吗？
A: PTrade的 get_fundamentals 在回测中默认是Point-in-Time（PIT）的。即在2022年3月1日调用该接口，如果某公司2021年年报在3月15日才发布，那么在3月1日是查不到该数据的（会返回NaN或旧数据）。本策略代码中虽然指定了年份，但PTrade底层会处理发布日期的限制。为了更严谨，可以在 before_trading_start 中加入对数据是否为NaN的判断。