PTrade 选股因子：小市值轮动策略实现教程

这是一个基于 PTrade 平台的小市值轮动策略。

策略逻辑说明

1. 选股范围：全市场 A 股。
2. 过滤条件：
   • 剔除 ST、*ST 股票。
   • 剔除停牌、退市股票。
   • 剔除上市时间不足 1 年的次新股（避免新股上市初期的剧烈波动）。
   • 剔除涨停、跌停股票（避免无法成交）。
3. 排序因子：总市值（total_value）。
4. 核心逻辑：选取总市值最小的前 N 只股票（例如 10 只）。
5. 交易执行：
   • 每日（或指定周期）进行调仓。
   • 卖出不在持仓列表中的股票。
   • 买入新选出的股票，采用等权重分配资金。

PTrade 策略代码

import datetime
import numpy as np
import pandas as pd

def initialize(context):
    """
    初始化函数，设置策略参数
    """
    # 设定持仓股票数量
    g.stock_num = 10
    
    # 设定调仓频率（这里演示为每日调仓，实际可根据需求改为每周或每月）
    # 设定在每天 10:00 运行，避开开盘集合竞价的剧烈波动
    run_daily(context, rebalance, time='10:00')
    
    # 设置手续费（可选，回测时建议设置）
    # 股票佣金万三，最低5元
    set_commission(commission_ratio=0.0003, min_commission=5.0, type='STOCK')

def rebalance(context):
    """
    调仓主函数
    """
    log.info("开始执行小市值选股逻辑...")
    
    # 1. 获取全市场 A 股代码
    # get_Ashares 返回当前交易日所有在世的 A 股代码列表
    all_stocks = get_Ashares()
    
    # 2. 基础过滤：剔除 ST、停牌、退市股票
    # filter_stock_by_status 默认过滤 ST, HALT, DELISTING
    stocks_filtered = filter_stock_by_status(all_stocks)
    
    # 3. 过滤次新股（上市不足 365 天）
    stocks_filtered = filter_new_stocks(context, stocks_filtered, days=365)
    
    if not stocks_filtered:
        log.warning("过滤后无股票可选")
        return

    # 4. 获取市值数据并排序
    # 查询 valuation 表中的 total_value (总市值)
    # 注意：get_fundamentals 在涉及大量股票时，建议只查询需要的字段
    df = get_fundamentals(
        security=stocks_filtered,
        table='valuation',
        fields=['total_value'],
        date=None  # 回测模式下默认取当前回测日期
    )
    
    if df is None or df.empty:
        log.warning("未获取到财务数据")
        return
    
    # 按总市值从小到大排序
    df = df.sort_values(by='total_value', ascending=True)
    
    # 5. 选取市值最小的前 N 只股票
    # 注意：这里还可以增加一步过滤，剔除当日涨跌停的股票，防止买不进卖不出
    target_list = []
    for stock in df.index:
        if len(target_list) >= g.stock_num:
            break
        
        # 检查是否涨跌停
        limit_info = check_limit(stock)
        # check_limit 返回字典，状态说明：2:触板涨停, 1:涨停, 0:正常, -1:跌停, -2:触板跌停
        # 我们只买入非涨停且非跌停的股票
        status = limit_info.get(stock, 0)
        if status == 0:
            target_list.append(stock)
            
    log.info("今日目标持仓: %s" % target_list)
    
    # 6. 执行交易
    adjust_position(context, target_list)

def filter_new_stocks(context, stock_list, days=365):
    """
    过滤上市时间不足指定天数的股票
    """
    filtered_list = []
    # 批量获取股票基础信息
    infos = get_stock_info(stock_list)
    
    current_dt = context.blotter.current_dt
    
    for stock in stock_list:
        info = infos.get(stock)
        if info and info.get('listed_date'):
            listed_date_str = info['listed_date']
            # 将字符串日期转换为 datetime 对象
            try:
                listed_date = datetime.datetime.strptime(listed_date_str, "%Y-%m-%d")
                # 计算上市天数
                delta = (current_dt - listed_date).days
                if delta > days:
                    filtered_list.append(stock)
            except:
                pass
    return filtered_list

def adjust_position(context, target_list):
    """
    根据目标列表调整持仓
    """
    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    # 1. 卖出不在目标列表中的股票
    for stock in current_positions:
        if stock not in target_list:
            # 检查是否停牌或跌停，如果是则无法卖出，这里简单处理直接下卖单
            # order_target_value 设为 0 即清仓
            order_target_value(stock, 0)
            log.info("卖出: %s" % stock)
    
    # 2. 买入目标列表中的股票
    if len(target_list) > 0:
        # 等权重分配资金
        # 注意：这里使用总资产 portfolio_value 来计算，包含现金和持仓市值
        position_value = context.portfolio.portfolio_value / len(target_list)
        
        for stock in target_list:
            # order_target_value 会自动计算需要买入或卖出的数量，使持仓市值达到目标值
            order_target_value(stock, position_value)
            log.info("调整持仓: %s 到目标市值: %.2f" % (stock, position_value))

def handle_data(context, data):
    """
    盘中运行函数（本策略主要逻辑在 rebalance 中，此处留空即可）
    """
    pass

代码关键点解析

1. get_Ashares():

   • 获取当前回测日期或交易日的所有 A 股代码，这是选股的基础池。

2. filter_stock_by_status(stocks, ...):

   • 这是 PTrade 非常实用的内置函数。默认参数下，它会自动过滤掉 ST 股、停牌股和退市股。这比自己写逻辑判断要高效且准确。

3. get_fundamentals(..., table='valuation', fields=['total_value']):

   • 这是获取财务数据的核心接口。
   • valuation 表包含每日估值数据。
   • total_value 代表总市值。如果你想做“流通市值”轮动，可以将字段改为 float_value。
   • 返回的是一个 DataFrame，索引（index）是股票代码，非常方便排序。

4. check_limit(stock):

   • 在实盘或模拟交易中，涨停买不进，跌停卖不出。在选股阶段剔除这些股票可以避免资金占用或无效下单。

5. order_target_value(stock, value):

   • 这是轮动策略中最常用的下单函数。
   • 它的作用是：无论当前持有多少股，通过买入或卖出，使该股票的持仓市值调整到 value。
   • 如果 value 为 0，则相当于清仓。
   • 如果 value 为总资金的 1/N，则实现了等权重买入。

常见问题 (Q&A)

Q: 为什么选择在 10:00 调仓而不是 09:30?
A: 09:30 开盘时刻市场波动极大，且容易受到集合竞价的影响。10:00 市场情绪相对稳定，且此时大部分股票的涨跌停状态已经明朗，数据获取也更稳定。

Q: 如何修改为每周调仓?
A: 可以在 rebalance 函数开头增加日期判断，或者使用 schedule_function (如果 PTrade 版本支持) 或在 initialize 中利用逻辑控制。最简单的方法是在 rebalance 中判断：

# 仅在周一执行
if context.blotter.current_dt.weekday() != 0:
    return

Q: 策略回测很慢怎么办?
A: 小市值策略涉及全市场选股。

1. 尽量减少 get_stock_info 这种需要循环请求的函数调用次数。
2. get_fundamentals 尽量一次性获取所有股票的数据，而不是在循环中获取。上面的代码已经是一次性获取，是比较高效的写法。

Q: 为什么实盘中买入的小市值股票和回测不一样?
A: 财务数据（如市值）在盘中是动态变化的。回测通常使用前一日收盘数据或当日收盘数据（取决于回测引擎设置），而实盘使用的是实时数据。此外，小市值股票流动性差，滑点影响较大。