etf动量轮动+大盘择时：年化30%的策略

news2025/12/14 23:10:24

原创文章第111篇，专注“个人成长与财富自由、世界运作的逻辑， AI量化投资”。

今天重点来探索一下elegantRL。

昨天的文章金融强化学习与finRL开发包里介绍了finRL的源码结构，背后的强化学习框架是elegantRL。

聚宽平台上有一个“动量轮动+RSRS”择时，在咱们自己的AI量化平台上复现一下。

01 策略思路

我们之前用的动量是ROC（20）就是20日收益率。这里的动量定义为 20日收盘价的“斜率”——就是线性回归的斜率。“斜率”取最大的K支，构建组合。然后RSRS（18，600）对市场择时，如果RSRS信号为buy，则按上述动量组合调仓，若RSRS信号为SELL，则平仓。

02 动量计算

动量定义为 20日收盘价的“斜率”——就是线性回归的斜率。

qlib原本的表达式，使用了cpython，这里我们使用np.polyfit即可实现。

class Slope(Rolling):

    def __init__(self, feature, N):
        super(Slope, self).__init__(feature, N, "slope")

    def _load_internal(self, instrument):
        def calc_slope(x):
            x = x / x[0]  # 这里做了一个“归一化”
            slope = np.polyfit(range(len(x)), x, 1)[0]
            return slope

        series = self.feature.load(instrument)
        result = series.rolling(self.N, min_periods=2).apply(calc_slope)
        series = pd.Series(result, index=series.index)
        return series

这是单序列的线性回归，RSRS是双序列的线性回归。目前没有找到办法rolling，导致性能很差。

调用的代码如下：

fields += ['Slope($close,20)']
names += ['mom_slope']

fields += ["Ref($close,-1)/$close - 1"]
names += ['label']

all = Dataloader().load_one_df(['000300.SH'], names, fields)

如此即可以实现代码最大程度的复用了。

02 排序算子之topK

按某一个因子的顺序，选择前K个进行持仓的算子。

class SelectTopK:
    def __init__(self, K=1, order_by='order_by', b_ascending=False):
        self.K = K
        self.order_by = order_by
        self.b_ascending = b_ascending

    def __call__(self, context):
        stra = context['strategy']
        features = context['features']

        if self.order_by not in features.columns:
            logger.error('排序字段{}未计算'.format(self.order_by))
            return

        bar = get_current_bar(context)
        if bar is None:
            logger.error('取不到bar')
            return True
        bar.sort_values(self.order_by, ascending=self.b_ascending, inplace=True)

        selected = []
        pre_selected = None
        if 'selected' in context:
            pre_selected = context['selected']
            del context['selected']

        for code in list(bar.code):
            if pre_selected:
                if code in pre_selected:
                    selected.append(code)
            else:
                selected.append(code)
            if len(selected) >= self.K:
                break
        context['selected'] = selected

有了算子之后，都是代码模块。

不择时的收益率如下：

如果把“斜率”换成“20日ROC”，收益率小一点，但回撤大不少。

03 大盘择时

大盘使用沪深300的RSRS给大盘择时。

大盘择时的逻辑为：若择时指标为“买”，则按原计划操作，若大盘择时为“卖”，则全部平仓退场。

class PickTime:
    def __init__(self, benchmark='000300.SH', signal='signal'):
        self.benchmark = benchmark
        #self.buy = self.buy
        self.signal = signal

    def __call__(self, context):
        stra = context['strategy']
        extra = context['extra']
        df = extra[self.benchmark]

        if self.signal not in df.columns:
            logger.error('择时信号不存在')
            return True

        curr_date = stra.get_current_dt()
        if curr_date not in df.index:
            logger.error('日期不存在{}'.format(curr_date))
            return None

        bar = df.loc[curr_date]
        if type(bar) is pd.Series:
            bar = bar.to_frame().T

        if bar[self.signal][0]:
            logger.info('择时信号显示，平仓所有。')
            context['selected'] = []

04 组合成策略

积木式开发不需要写策略代码，这个非常方便，而且所有的算子均可复用。

e = BacktraderEngine(init_cash=1000000, benchmark='399006.SZ', start=datetime(2014, 1, 1))
e.add_features(symbols, names, fields)
e.add_extra('000300.SH', fields=['RSRS($high,$low,18,600)', '$RSRS_beta<0.8'], names=['RSRS', 'signal'])

from engine.strategy.algos import SelectTopK, PickTime, WeightEqually

e.run_algo_strategy([SelectTopK(K=1), PickTime(), WeightEqually()])
e.analysis(pyfolio=False)