RSRS指标，相对阻力支撑指标的实现，qlib扩展自己的特征表达式

光大证券有一个RSRS指标，回测看来效果很不错，它也是基于突破的逻辑。与传统布林带这样的通道不同，它对$high最高价，与$low最低价之间做一个线性回归，看它的趋势。

只是RSRS指数要计算两个序列的相对斜率，这个指标，qlib没有实现。

——他们想到的，但还是todo。只能自己动手，丰衣足食。

# TODO:
# support pair-wise rolling like `Slope(A, B, N)`
class Slope(Rolling):

正好熟悉一下qlib的自定义指标。

其实过程不复杂，在初始化的时候，把自定义指标注册进去即可。就是下面代码里的“custom_ops”。

provider_uri = "~/.qlib/qlib_data/{}".format('index_data')
qlib.init(provider_uri=provider_uri, region=REG_CN, **{"custom_ops": [RSRS]})

而指标的实现过程如下，两个指标运算，继承自PairOperator，只需要实现这个_load_internal即可。加载数据逻辑是一样的，把se_left，se_right两个序列加载，然后进行计算。

import numpy as np
import pandas as pd
import qlib
from qlib.data import D
from qlib.data.ops import ElemOperator, PairOperator
from qlib.config import REG_CN

import statsmodels.api as sm

class PairSlope(PairOperator):
    def __init__(self, feature_left, feature_right, N):
        self.N = N
        super(PairSlope, self).__init__(feature_left,feature_right)

    def _load_internal(self, instrument, start_index, end_index, freq):
        series_left = self.feature_left.load(instrument, start_index, end_index, freq)
        series_right = self.feature_right.load(instrument, start_index, end_index, freq)

        slope = []
        #R2 = []
        # 计算斜率值
        n = self.N
        for i in range(len(series_left)):
            if i < (self.N - 1):
                continue
            else:
                x = series_right[i - n + 1:i + 1]
                # iloc左闭右开
                x = sm.add_constant(x)
                y = series_left.iloc[i - n + 1:i + 1]
                regr = sm.OLS(y, x)
                res = regr.fit()
                beta = round(res.params[1], 2)  # 斜率指标
                slope.append(beta)
                #R2.append(res.rsquared)
        #df = df.iloc[n - 1:]
        slope = pd.Series(slope)
        return slope
     

if __name__ == '__main__':
    provider_uri = "~/.qlib/qlib_data/{}".format('index_data')
    qlib.init(provider_uri=provider_uri, region=REG_CN, **{"custom_ops": [PairSlope]})

    instruments = ["sh000300"]
    fields = ["PairSlope($high, $close,18)"]
    print(D.features(instruments, fields, start_time="2010-01-01", end_time="2017-12-31", freq="day"))

然后我们需要实现一下pair_slope即可。

qlib自身有一个slope的计算函数。

# TODO:
# support pair-wise rolling like `Slope(A, B, N)`
class Slope(Rolling):
    def __init__(self, feature, N):
        super(Slope, self).__init__(feature, N, "slope")

    def _load_internal(self, instrument, start_index, end_index, freq):
        series = self.feature.load(instrument, start_index, end_index, freq)
        if self.N == 0:
            series = pd.Series(expanding_slope(series.values), index=series.index)
        else:
            series = pd.Series(rolling_slope(series.values, self.N), index=series.index)
        return series

调用了rolling_slope这个函数。

from ._libs.rolling import rolling_slope, rolling_rsquare, rolling_resi
from ._libs.expanding import expanding_slope, expanding_rsquare, expanding_resi

我调用了statsmodels.api线性回归算法计算了，性能可能差一点，后续可以优化。

有了指标，其实策略就很简单了。

由于计算时间较长，我使用了hdf5来缓存，hdf5可以原样保存dataframe的数据，而且性能很好，可以当离线的字典来用，csv大且慢，而且dataframe写csv再读出来，index,column信息会丢失，数据格式float可能会变成str，都要自己去转换。

store = pd.HDFStore('cache.h5')

if 'cache' in store.keys():
    df = store.get(key='cache')
else:
    features = ["$high", "$low", "PairSlope($high, $close,18)"]
    features.append('$close / Ref($close,1) -1')
    names = ['high', 'low', 'RSRS', 'rate']

    # 'sh000905', 'sz399006'
    df = logic.load_data(instruments=['sh000300',], features=features, names=names)

df = df['feature']

df['to_buy'] = df['RSRS'] > 1
df['to_sell'] =df['RSRS'] < 0.8

明天正式来实证RSRS策略，复现光大证券研报。

策略整合并模板化 | 从零实现AI量化回测平台 #5

"积木式"的策略框架 | 从零搭建自己的AI量化平台 #2

搭建回测系统主框架 | 从零搭建自己的AI量化平台 #1