RSRS大盘择时怎么用Python实现

用户头像mx_****zqklr
2026-09-07 发布

如果你正在用 Python 做指数择时,RSRS 是一个很适合自己实现的信号:输入过去一段时间的最高价和最低价,通过滚动回归得到斜率,再把斜率标准化,最后根据阈值生成仓位信号。

真正麻烦的地方并不是写出一段 np.polyfit(),而是把 数据窗口、标准化、交易信号和回测时点 对齐。尤其是 RSRS 需要同时使用短窗口计算斜率、长窗口计算 Z-score,如果时间索引处理不严谨,很容易把未来数据带进信号。

下面从公式开始,把一个可用于大盘择时研究的 Python 版本拆开。

RSRS到底在计算什么?

经典 RSRS 的核心,是在过去 N 个交易日里建立:

Hight=α+βLowt+ϵtHigh_t=\alpha+\beta Low_t+\epsilon_t其中:

  • High 是每日最高价;
  • Low 是每日最低价;
  • β 是回归斜率;
  • 是拟合优度。

简单理解,RSRS 不是直接预测指数明天涨多少,而是在观察一段时间内高低价格之间的相对变化关系。

常见实现会进一步对 β 做标准化:

Zt=βt−μtσtZ_t=\frac{\beta_t-\mu_t}{\sigma_t}其中 μσ 来自过去 M 个 β 值。

进一步的修正版还会使用:

RSRSR2=Z×R2RSRS_{R2}=Z\times R^2一些 RSRS 实现则继续乘以 β,形成右偏标准分:

RSRSright=Z×R2×βRSRS_{right}=Z\times R^2\times\beta公开的 RSRS 实现和研究中都可以看到这几种层级。

因此,所谓“RSRS 择时”,实际上可以拆成三个独立步骤:

  1. 滚动回归计算 β;
  2. 使用历史 β 序列计算 Z-score;
  3. 根据 RSRS 信号和阈值决定持仓状态。

第一步:准备大盘历史高低价

以沪深300这类宽基指数作为择时基准时,需要至少准备:

trade_date
high
low

如果后续还要计算收益、净值和回撤,则还需要 close

数据结构最好直接保持为 Pandas DataFrame:

import pandas as pd

price = pd.DataFrame({
    "trade_date": [...],
    "high": [...],
    "low": [...],
    "close": [...],
})

price["trade_date"] = pd.to_datetime(price["trade_date"])
price = price.sort_values("trade_date").reset_index(drop=True)

这里有一个容易忽略的问题:RSRS 本身依赖时间序列顺序。

因此不要直接对一个没有排序的数据表做 rolling,也不要假设 API 返回的数据一定已经按照交易日升序排列。

第二步:用滚动窗口计算RSRS斜率

假设:

  • N = 18
  • 每 18 个交易日做一次高低价回归。

可以先写成一个独立函数:

import numpy as np

def calc_rsrs_beta(high, low, n=18):
    if len(high) < n:
        return np.nan

    y = high[-n:]
    x = low[-n:]

    if np.isnan(x).any() or np.isnan(y).any():
        return np.nan

    beta, _ = np.polyfit(x, y, 1)
    return beta

然后滚动计算:

price["rsrs_beta"] = [
    calc_rsrs_beta(
        price["high"].iloc[:i + 1].values,
        price["low"].iloc[:i + 1].values,
        n=18
    )
    for i in range(len(price))
]

这样得到的 rsrs_beta 是每个交易日只利用截至当天的数据计算出来的斜率。

如果后面要做真正的历史回测,这一点非常重要。

第三步:把β转换成Z-score

有了 β 序列之后,再计算过去 M 个交易日的均值和标准差。

例如:

M = 600

rolling_mean = price["rsrs_beta"].rolling(M).mean()
rolling_std = price["rsrs_beta"].rolling(M).std()

price["rsrs_z"] = (
    price["rsrs_beta"] - rolling_mean
) / rolling_std

这里的 M 不应该被理解成“RSRS 固定必须使用 600 天”。

不同实现会使用不同窗口,公开策略中可以看到 M=200600、更长窗口等参数设置。

更合理的工程方式是把参数放在策略配置里:

N = 18
M = 600
BUY_THRESHOLD = 0.7
SELL_THRESHOLD = -0.7

这样以后做参数研究时,不需要修改指标函数。

为什么还要乘R²?

单看 β 有一个问题:回归线的拟合质量可能并不好。

因此可以计算:

def calc_rsrs_beta_r2(high, low, n=18):
    if len(high) < n:
        return np.nan, np.nan

    y = np.asarray(high[-n:], dtype=float)
    x = np.asarray(low[-n:], dtype=float)

    if np.isnan(x).any() or np.isnan(y).any():
        return np.nan, np.nan

    beta, alpha = np.polyfit(x, y, 1)
    predicted = alpha + beta * x

    ss_res = np.sum((y - predicted) ** 2)
    ss_tot = np.sum((y - np.mean(y)) ** 2)

    if ss_tot == 0:
        return beta, np.nan

    r2 = 1 - ss_res / ss_tot
    return beta, r2

然后:

price[["rsrs_beta", "r2"]] = [
    calc_rsrs_beta_r2(
        price["high"].iloc[:i + 1].values,
        price["low"].iloc[:i + 1].values,
        n=N
    )
    for i in range(len(price))
]

得到 Z-score 后:

price["rsrs_z_r2"] = price["rsrs_z"] * price["r2"]

如果研究的是右偏标准分,则进一步:

price["rsrs_right"] = (
    price["rsrs_z"]
    * price["r2"]
    * price["rsrs_beta"]
)

这里应该明确区分:这些都是 RSRS 指标的不同构造方式,不应该把它们混成一个“标准答案”。

把指标变成真正的择时信号

指标有了,下一步才是交易逻辑。

例如最简单的状态型策略:

def generate_signal(score, buy_threshold=0.7, sell_threshold=-0.7):
    if pd.isna(score):
        return 0

    if score > buy_threshold:
        return 1

    if score < sell_threshold:
        return 0

    return np.nan

这里返回 NaN 的意义很重要:

当指标处于买卖阈值之间时,不代表必须平仓,而是保持上一交易状态。

因此需要做状态延续:

signal = price["rsrs_right"].apply(generate_signal)

price["position"] = signal.ffill().fillna(0)

这比直接写:

price["position"] = (
    price["rsrs_right"] > 0.7
).astype(int)

更接近典型的阈值型择时逻辑,因为后者会让信号一旦低于 0.7 就立即变成空仓。

回测时最容易犯的错误:当天信号不能偷看当天收盘

这是 RSRS 实现中比公式本身更值得注意的问题。

假设你使用当天的:

High
Low
Close

计算 RSRS,然后又假设在当天收盘价成交,那么就必须明确交易时点。

如果你的策略定义是:

用 T 日收盘之后计算信号,T+1 日执行。

那么应该使用:

price["position"] = signal.ffill().fillna(0)

price["position_lag"] = price["position"].shift(1)

price["market_return"] = price["close"].pct_change()

price["strategy_return"] = (
    price["position_lag"]
    * price["market_return"]
)

核心就在:

shift(1)

它表示今天的收益使用昨天已经产生的仓位。

如果忽略这个时间错位,回测结果可能看起来非常漂亮,但策略实际上使用了尚未发生的收盘信息。

一个最小可用的RSRS研究框架

把前面的步骤组合起来,可以形成:

import numpy as np
import pandas as pd


N = 18
M = 600
BUY_THRESHOLD = 0.7
SELL_THRESHOLD = -0.7


def rsrs_beta_r2(high, low, n=N):
    y = np.asarray(high[-n:], dtype=float)
    x = np.asarray(low[-n:], dtype=float)

    if len(x) < n or np.isnan(x).any() or np.isnan(y).any():
        return np.nan, np.nan

    beta, alpha = np.polyfit(x, y, 1)
    predicted = alpha + beta * x

    ss_res = np.sum((y - predicted) ** 2)
    ss_tot = np.sum((y - y.mean()) ** 2)

    r2 = np.nan if ss_tot == 0 else 1 - ss_res / ss_tot

    return beta, r2


# 1. 计算 beta 和 R²
values = [
    rsrs_beta_r2(
        price["high"].iloc[:i + 1],
        price["low"].iloc[:i + 1],
        N
    )
    for i in range(len(price))
]

price[["beta", "r2"]] = values


# 2. 计算 beta 的 Z-score
price["beta_mean"] = price["beta"].rolling(M).mean()
price["beta_std"] = price["beta"].rolling(M).std()

price["z"] = (
    price["beta"] - price["beta_mean"]
) / price["beta_std"]


# 3. 计算修正 RSRS
price["rsrs"] = price["z"] * price["r2"]


# 4. 生成持仓信号
raw_signal = np.where(
    price["rsrs"] > BUY_THRESHOLD,
    1,
    np.where(
        price["rsrs"] < SELL_THRESHOLD,
        0,
        np.nan
    )
)

price["position"] = (
    pd.Series(raw_signal, index=price.index)
    .ffill()
    .fillna(0)
)


# 5. 使用下一交易日持仓计算收益
price["position"] = price["position"].shift(1)

price["market_return"] = price["close"].pct_change()

price["strategy_return"] = (
    price["position"]
    * price["market_return"]
)

这段代码只是一个研究框架,不是某个交易平台的内置策略,也没有包含手续费、滑点、停牌、交易时间和实际成交约束。

历史数据怎么接入更适合长期研究?

如果只是学习 RSRS,CSV 或本地 DataFrame 已经足够。

但当研究逐渐变成长期运行的策略时,数据获取本身就会成为维护工作:每天更新历史 K 线、处理代码格式、补数据、统一字段,然后再把数据交给 Pandas。

这时可以把“行情获取”和“指标计算”拆开。

例如当前 QuantDash 的 Python SDK 已公开支持 A 股、ETF、美股和港股,并提供 DataFrame 形式的 K 线数据;PyPI 当前文档也明确给出了 qd.klines.get() 和批量 K 线接口。

安装:

pip install quantdash

API Key 可以通过环境变量管理:

import os

api_key = os.getenv("QUANTDASH_API_KEY")

初始化后获取历史 K 线:

from quantdash import QuantDash

qd = QuantDash()

df = qd.klines.get(
    "600519.SH",
    period="1d",
    count=1000,
    to_dataframe=True
)

QuantDash 当前公开 SDK 还支持 forwardbackwardnone 等复权方式。

对于 RSRS,数据层最重要的不是“换一个 API 就能提高策略收益”,而是让研究代码拿到结构稳定、时间顺序明确、字段可直接进入 Pandas 的行情数据。

指标计算和交易逻辑仍然由你的 Python 代码负责。

RSRS回测前应该先检查什么?

不要急着看年化收益率。

先检查这几个地方:

第一,数据有没有按交易日排序。

df = df.sort_values("trade_date")

第二,N 日回归是否真的只用了过去 N 个交易日。

第三,M 日标准化是否使用了未来 β。

第四,交易信号和实际持仓是否错开。

第五,复权口径是否和回测逻辑一致。

第六,手续费和滑点是否被纳入策略假设。

尤其是 RSRS 这种滚动指标,最容易出现的问题不是 Python 报错,而是代码正常运行、结果也正常,但时间索引实际上已经错位。

RSRS适合解决什么问题?

RSRS 更适合被当成一个大盘仓位信号,而不是万能选股因子。

例如:

指数历史行情
      ↓
N日高低价回归
      ↓
RSRS β
      ↓
M日标准化
      ↓
Z / Z×R² / Z×R²×β
      ↓
阈值判断
      ↓
仓位信号
      ↓
用户自己的回测逻辑

这种拆分方式的好处是,每一层都可以单独检查。

数据错了,查行情层;
指标异常,查回归和 rolling;
信号异常,查阈值;
回测收益异常,先查信号与交易日是否错位。

RSRS 本身并不提供收益保证,参数也不应该因为某一段历史区间表现较好就永久固定。公开研究中已经存在不同 N、M 和阈值组合,说明参数选择本身就是策略研究的一部分。

总结

用 Python 实现 RSRS,真正值得花时间的不是把线性回归写出来,而是把下面这条链路做正确:

历史高低价 → N 日回归 β → M 日标准化 → RSRS 信号 → 下一交易日持仓 → 回测收益。

如果只是学习指标,本地数据完全够用;如果要把研究代码长期运行,行情获取、复权、标的代码和 DataFrame 接入则应该独立成数据层。

QuantDash 在这里真正相关的角色,是提供股票和 ETF 等市场行情数据,并通过 Python SDK 将 K 线直接接入 Pandas;RSRS 指标、择时规则和回测逻辑仍然由研究者自己的 Python 代码完成。

QuantDash 官方资源

评论