PTrade 经典动量策略实现指南（附 Python 源码）

这是一个基于 PTrade 平台的动量策略（Momentum Strategy）实现方案。

策略逻辑说明

动量策略的核心思想是“强者恒强”。本策略的具体逻辑如下：

1. 股票池：以 沪深300指数 (000300.SS) 成分股为基础。
2. 选股因子：计算过去 20个交易日 的涨跌幅（动量）。
3. 过滤条件：剔除停牌、ST股、退市整理期的股票。
4. 持仓数量：持有涨幅最高的 10只 股票。
5. 调仓频率：按月调仓（每月第一个交易日进行调整）。
6. 资金分配：等权重买入（即每只股票买入总资金的 1/10）。

策略代码实现

import numpy as np
import pandas as pd

def initialize(context):
    """
    初始化函数，设置策略参数
    """
    # 设定基准为沪深300
    set_benchmark('000300.SS')
    # 开启真实市价成交模式（回测时更接近真实）
    set_limit_mode('UNLIMITED')
    # 设置手续费（股票万分之三）
    set_commission(commission_ratio=0.0003, min_commission=5.0, type="STOCK")
    # 设置滑点
    set_slippage(slippage=0.002)
    
    # --- 策略参数设置 ---
    # 目标指数：沪深300
    g.index_code = '000300.SS'
    # 持仓数量
    g.top_n = 10
    # 动量计算周期（过去N天的涨幅）
    g.momentum_days = 20
    
    # 记录上一次调仓的月份，用于判断是否跨月
    g.last_month = 0
    
    # 每天开盘前运行，用于判断是否调仓
    run_daily(context, check_rebalance, time='09:30')

def before_trading_start(context, data):
    """
    盘前处理：获取成分股并过滤
    """
    # 获取指数成分股
    stocks = get_index_stocks(g.index_code)
    
    # 过滤掉 ST、停牌、退市的股票
    # filter_stock_by_status 接口会自动剔除 ST, HALT, DELISTING
    g.target_pool = filter_stock_by_status(stocks)
    
    # 设置股票池，保证数据获取的范围
    set_universe(g.target_pool)

def check_rebalance(context):
    """
    判断是否需要调仓（每月第一个交易日）
    """
    current_month = context.blotter.current_dt.month
    
    # 如果当前月份与上一次记录的月份不同，则执行调仓
    if current_month != g.last_month:
        log.info("月份变化，触发调仓逻辑...")
        rebalance(context)
        g.last_month = current_month

def rebalance(context):
    """
    核心调仓逻辑
    """
    if len(g.target_pool) == 0:
        log.warning("今日无有效股票池，跳过调仓")
        return

    # 1. 获取历史价格数据
    # 获取过去 momentum_days + 1 天的数据，以便计算 (今天收盘 - N天前收盘) / N天前收盘
    # 注意：get_history 返回的数据包含今天（回测中视作当前时刻），
    # 但为了计算过去N天的完整涨幅，我们需要足够的数据点。
    # 这里取 count = g.momentum_days + 1
    hist_data = get_history(count=g.momentum_days + 1, frequency='1d', field='close', security_list=g.target_pool)
    
    # 2. 计算动量（涨跌幅）
    momentum_scores = {}
    
    # 遍历股票池计算涨幅
    # 注意：hist_data 的结构可能因 Python 版本或 PTrade 版本略有不同
    # 通常 hist_data 是一个 DataFrame，列是股票代码，或者是一个 Panel
    # 这里使用通用的处理方式
    
    # 如果是 DataFrame 且包含 'code' 列 (Python 3.11+ 风格)
    if isinstance(hist_data, pd.DataFrame) and 'code' in hist_data.columns:
        # 按代码分组计算
        for stock in g.target_pool:
            try:
                # 获取单只股票数据
                stock_df = hist_data[hist_data['code'] == stock]
                if len(stock_df) < g.momentum_days + 1:
                    continue
                
                prices = stock_df['close'].values
                # 计算涨幅: (最新价 - N天前价格) / N天前价格
                # prices[-1] 是最近一个收盘价，prices[0] 是N天前的收盘价
                pct_change = (prices[-1] - prices[0]) / prices[0]
                momentum_scores[stock] = pct_change
            except Exception as e:
                continue
                
    # 如果是 DataFrame 但列索引是股票代码 (Python 3.5 风格)
    elif isinstance(hist_data, pd.DataFrame):
        for stock in g.target_pool:
            if stock in hist_data.columns:
                prices = hist_data[stock].values
                # 剔除含 NaN 的数据
                prices = prices[~np.isnan(prices)]
                
                if len(prices) >= 2: # 至少要有首尾数据
                    pct_change = (prices[-1] - prices[0]) / prices[0]
                    momentum_scores[stock] = pct_change
    
    # 3. 排序并选股
    # 按涨幅从大到小排序
    sorted_stocks = sorted(momentum_scores.items(), key=lambda x: x[1], reverse=True)
    
    # 取前 Top N
    buy_list = [x[0] for x in sorted_stocks[:g.top_n]]
    log.info("本月选中动量股: %s" % buy_list)
    
    # 4. 执行交易
    adjust_position(context, buy_list)

def adjust_position(context, buy_list):
    """
    执行具体的买卖操作
    """
    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    # 1. 卖出不在买入列表中的股票
    for stock in current_positions:
        if stock not in buy_list:
            # 检查是否停牌，停牌无法卖出
            if check_limit(stock).get(stock, 0) == 0: # 简单检查，实际交易中系统会自动拒单
                order_target_value(stock, 0)
    
    # 2. 买入/调整目标股票
    if len(buy_list) > 0:
        # 等权重分配资金
        # 注意：这里使用 total_value (总资产) 而不是 cash (可用现金)，以实现动态平衡
        target_value_per_stock = context.portfolio.portfolio_value / len(buy_list)
        
        for stock in buy_list:
            # 检查是否涨停，涨停无法买入 (check_limit 返回 1 或 2 表示涨停)
            limit_status = check_limit(stock).get(stock, 0)
            if limit_status not in [1, 2]: 
                order_target_value(stock, target_value_per_stock)

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

代码关键点解析

1. 数据获取 (get_history)：

   • 我们获取了 g.momentum_days + 1 天的数据。这是因为计算收益率需要由“当前价格”和“N天前价格”两个点确定。
   • 代码中兼容了 PTrade 不同 Python 环境下 get_history 返回数据结构的差异（DataFrame 列索引为代码 vs 列内容包含代码）。

2. 股票过滤 (filter_stock_by_status)：

   • 这是 PTrade 提供的便捷函数，能直接过滤掉 ST、停牌和退市股票，防止买入垃圾股或无法交易的股票。

3. 调仓逻辑 (check_rebalance)：

   • 策略通过比较 current_month 和 g.last_month 来判断是否进入了新的月份。这是一种简单有效的月度调仓实现方式。

4. 交易执行 (order_target_value)：

   • 使用 order_target_value 而不是 order。这个函数会自动计算需要买入或卖出的数量，使该股票的持仓市值达到指定金额。
   • 资金分配使用 context.portfolio.portfolio_value / len(buy_list)，这意味着每次调仓都会根据当前总资产重新平衡权重（Rebalancing），这对于控制风险非常重要。

如何优化此策略？

如果你想进一步增强这个策略，可以考虑以下方向：

• 增加风控：在 rebalance 中加入大盘择时逻辑（例如：如果沪深300指数跌破20日均线，则空仓或半仓）。
• 平滑动量：不要只看20天的涨幅，可以结合 20天、60天、120天 的涨幅加权打分。
• 剔除新股：刚上市的股票波动极大，建议在 before_trading_start 中通过上市日期过滤掉上市不满 60 天的股票。

Q&A

Q: 为什么使用 order_target_value 而不是 order_target?
A: order_target_value 是按金额调整持仓，而 order_target 是按股数调整。在等权重策略中，我们需要每只股票占用相同的资金量（例如每只占总资金的10%），而不是持有相同的股数，因此使用 order_target_value 更方便。

Q: get_history 的 count 为什么要加 1?
A: 假设我们要计算过去 1 天的涨幅，我们需要“今天”和“昨天”两个价格数据。同理，计算过去 20 天的涨幅，我们需要包含今天在内的 21 个数据点，才能取到 prices[-1] (今天) 和 prices[0] (20天前)。

Q: 如何在回测中查看策略表现?
A: 将上述代码复制到 PTrade 的策略编辑框中，设置回测时间段（例如 2022-01-01 至 2023-01-01），资金设为 100万，频率选择“日线”，点击“开始回测”即可。