ptrade 行业 ETF 轮动策略：基于相对强弱 RSRS 指标的实现

在量化交易中，RSRS（阻力支撑相对强度）指标通过对最高价（High）与最低价（Low）进行线性回归，能够有效捕捉标的的动量与择时信号。结合行业 ETF 进行轮动，可以在市场不同阶段自动选择最强势的行业标的，规避个股风险并获得超额收益。

本文将详细介绍如何在 ptrade 量化平台 上通过 Python 实现一个完整的行业 ETF RSRS 轮动策略。

一、RSRS 指标计算原理

1. 线性回归斜率（Slope）：
   对过去 $N$ 日的最高价 $High$ 和最低价 $Low$ 进行 OLS 线性回归：
   $$High_t = \alpha + \beta \cdot Low_t + \epsilon_t$$
   这里的斜率 $\beta$ 即为 RSRS 斜率。

2. RSRS 标准分（Z-Score）：
   为了消除斜率长期均值漂移的影响，取过去 $M$ 日的斜率计算标准化得分：
   $$Z = \frac{\beta - \mu_{\beta}}{\sigma_{\beta}}$$

3. 轮动与择时逻辑：

   • 计算候选行业 ETF 池中各标的的 RSRS Z-score。
   • 过滤出 Z-score 大于买入阈值（如 0.7）且排名前 $K$ 的 ETF 进行买入/持仓。
   • 当持有的 ETF Z-score 跌破卖出阈值（如 -0.7）或不在前列时进行平仓调仓。

二、ptrade 策略实现代码

以下策略代码基于 ptrade API 规范编写，支持日线级别的回测与实盘运行：

import numpy as np
import pandas as pd
import statsmodels.api as sm

def initialize(context):
    # 1. 设置对比基准（沪深300）
    set_benchmark('000300.SS')
    
    # 2. 定义行业 ETF 轮动池
    g.etf_pool = [
        '510300.SS', # 沪深300ETF
        '510500.SS', # 中证500ETF
        '159915.SZ', # 创业板ETF
        '512880.SS', # 证券ETF
        '512010.SS', # 医药ETF
        '512660.SS', # 军工ETF
        '515000.SS', # 科技ETF
        '159928.SZ'  # 消费ETF
    ]
    set_universe(g.etf_pool)
    
    # 3. RSRS 参数设置
    g.N = 18          # 线性回归计算斜率的周期
    g.M = 600         # 斜率标准化(Z-Score)的窗口周期
    g.buy_threshold = 0.7   # 买入阈值
    g.sell_threshold = -0.7 # 卖出阈值
    g.top_k = 2             # 最大同时持有行业 ETF 数量
    
    # 4. 设置定时任务，每天 09:35 执行调仓逻辑
    run_daily(context, rotate_etf, time='09:35')

def get_rsrs_zscore(security):
    """计算单只标的的 RSRS Z-Score"""
    # 获取计算所需的最高价和最低价历史数据
    count = g.N + g.M - 1
    df = get_history(count, '1d', field=['high', 'low'], security_list=security)
    
    if df is None or len(df) < count:
        return None
        
    highs = df['high'].values
    lows = df['low'].values
    
    slopes = []
    # 滚动计算 N 日的线性回归斜率
    for i in range(g.M):
        h = highs[i : i + g.N]
        l = lows[i : i + g.N]
        X = sm.add_constant(l)
        model = sm.OLS(h, X).fit()
        slopes.append(model.params[1])
    
    # 计算当前斜率的 Z-Score
    current_slope = slopes[-1]
    mean_slope = np.mean(slopes)
    std_slope = np.std(slopes)
    
    if std_slope == 0:
        return 0
        
    z_score = (current_slope - mean_slope) / std_slope
    return z_score

def rotate_etf(context):
    """ETF 轮动调仓主逻辑"""
    scores = {}
    
    # 计算所有 ETF 的 RSRS 分数
    for etf in g.etf_pool:
        z_score = get_rsrs_zscore(etf)
        if z_score is not None:
            scores[etf] = z_score
            
    if not scores:
        return
        
    # 按 Z-Score 从大到小排序
    sorted_etfs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    
    # 筛选满足买入条件的 ETF
    target_etfs = [etf for etf, score in sorted_etfs if score > g.buy_threshold][:g.top_k]
    
    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    # 1. 卖出逻辑：不在目标列表中，或 Z-Score 跌破卖出阈值的持仓标的
    for etf in current_positions:
        if etf not in target_etfs or scores.get(etf, 0) < g.sell_threshold:
            order_target(etf, 0)
            log.info(f"[卖出信号] {etf}，当前 RSRS Z-Score: {scores.get(etf, 0):.2f}")
            
    # 2. 买入/调仓逻辑
    if target_etfs:
        # 等权重分配资金
        target_value_per_etf = context.portfolio.portfolio_value / len(target_etfs)
        for etf in target_etfs:
            order_target_value(etf, target_value_per_etf)
            log.info(f"[买入/持有信号] {etf}，当前 RSRS Z-Score: {scores.get(etf, 0):.2f}")

def handle_data(context, data):
    pass

三、策略要点说明与 API 解析

1. 数据获取接口：
   使用 get_history(count, frequency, field, security_list) 获取历史高低价，保证包含 $N + M - 1$ 条周期数据用于斜率序列的标准差计算。

2. 组合再平衡（Rebalance）：

   • 通过 order_target_value(security, value) 实现指定价值的调仓。
   • 使用 context.portfolio.portfolio_value 获取当前资产总市值，以实现按总资产比例等权重配置。

3. 实盘/回测注意事项：

   • 回测模式： 建议使用每日分钟线或日线模式回测，避开开盘集合竞价滑点。
   • 复权处理： ETF 交易无需复权，get_history 中默认即可；如涉及股票交易需关注复权设置。