RSJ高频波动率择时指标

发布于vn.py

社区公众号【vnpy-community】

原文作者：Bili | 发布时间：2021-07-29

系列前言

经过这两年2.0大版本下的持续迭代，vn.py在易用性方面有了不少提升，现阶段对于许多社区新人来说，可能最难的已经不再是如何安装使用vn.py，而是如何用vn.py开发出优秀的量化策略。

开发策略这件事情确实需要一定的灵感，但更多还是应该参考爱迪生的名言：

天才是1%的灵感加上99%的汗水。

在当今2021年的中文互联网上，其实已经可以找到非常丰富的量化策略研究资料：

券商和期货公司研究所发布的金融工程研报；
知名高校和业内公司发表的量化投资论文；
微信公众号等自媒体上分享的策略研究文章；
各家量化平台线上公开的经典策略源代码。

初学者最快的学习方式，无疑是站在这些巨人的肩膀上，付出99%的汗水掌握前人的已有知识和经验，然后再加上1%的灵感来开发出属于自己的量化策略。

所以在许多社区用户的建议下，我们决定推出这个新的【Veighna量化策略实验室】文章系列，帮助大家尽可能体系化的学习量化策略开发：

找到好的量化研究资料；
在vn.py中复现策略代码；
和原资料对比检验正确性；
加入更多灵感来改进提升。

那么，接下来就是本系列的第一篇正式内容，策略资料来源于vn.py社区用户分享的券商金工研报，作者是财通证券研究所的陶勤英博士。

基本信息

策略原理

RV计算公式

高频已实现波动率（Realized Volatility

）是一种根据高频数据（这里指的是日内分钟线级别的数据，而非Tick级别数据）计算的日内波动率指标，该指标的计算公式为：

RSJ计算公式

由 Bollerslev

等作者（相关论文请参考研报PDF中的内容）研究的【好】与【坏】的波动率，是在RV的基础上将其分解为单独的上涨（好）和下跌（坏）两种情形下的波动率度量。将波动率分解为【好】波动和【坏】波动后，可以基于其相对的差值，去度量日内价格波动的不对称性（RSJ），该指标的计算公式为：

RSJ指标应用

根据原论文中的内容，日内高频数据计算得出的【好】波动率，描述了价格上涨时候的特征（反之【坏】波动率则是描述了价格下跌时候的特征）。

该指标用于横截面的选股策略上时，对未来收益率可以起到反转的预测效果。如果应用到时序类的CTA策略

，则对应的逻辑大概为：

RSJ数值越大，则说明未来越倾向于下跌；

RSJ数值越小，则说明未来越倾向于上涨。

策略核心逻辑

考虑到RSJ指标衡量的是当天日内整体价格波动所反应出来的信息，且每天股票市场的成交量大多集中在开盘和收盘附近的时间，因此我们选择在临近收盘的时间点，计算过去一段时间的RSJ指标，并对隔夜以及第二天日内的价格走势变化进行预测。

具体策略逻辑：

使用5分钟级别的K线来计算RSJ指标；
在每日的14:55计算前一段时间（可调参数）的RSJ指标；
若RSJ指标大于0则发出空头信号，反之则发出多头信号；
基于指标发出的交易信号，在收盘前完成交易（比如14:56）；
持仓到第二天的14:55分，然后重复2-4步骤调整仓位。

该策略的特点：

策略始终在市场中（持有仓位），要么做多，要么做空；
每天最多执行一次交易（如果信号方向变化）。

回测结果

回测数据上，本文中选择使用米筐RQData提供的IH888平滑主力合约数据，在后续的篇幅中我们会尝试更多的品种，在CtaBacktester中的回测配置如下：

本地代码：IH888.CFFEX
K线周期：1分钟
开始日期：2017-7-30
结束日期：2020-7-22
手续费率：0.00003
交易滑点：0.4
合约乘数：300
价格跳动：0.2
回测资金：100W

作为原始版本策略的源码复现，初步回测结果还不错：

资金曲线的形状和财通研报中的结果基本一致（下图中的蓝线），可以认为比较正确的复现了策略逻辑：

策略回测的关键统计结果：

总交易日：715
盈利交易日：370
总收益率：83.73%
年化收益：28.11%
百分比最大回撤：-25.03%

完整代码

最后，秉承vn.py社区的一贯精神：

Talk is cheap, show me your pnl (or code) !

自然必须附上策略的源代码：

from datetime import time

import numpy as np

from vnpy.app.cta_strategy import (    
    CtaTemplate,    
    BarGenerator,    
    ArrayManager,    
    OrderData,    
    TradeData,    
    StopOrder)
from vnpy.trader.object import (    
    BarData,    
    TickData
)


class RsjStrategy(CtaTemplate):    
    """"""    
    author = "Bili"
    
    # 定义参数    
    rsj_window = 12
    
    # 定义变量    
    rsj_value = 0.0
    
    parameters = [        
        "rsj_window"
    ]    
    variables = [        
        "rsj_value"
    ]
    
    def __init__(self, cta_engine, strategy_name, vt_symbol, setting):         
        """"""        
        super().__init__(cta_engine, strategy_name, vt_symbol, setting)
        
        self.bg = BarGenerator(self.on_bar, 5, self.on_5min_bar)     
        self.am = NewArrayManager()
        
    def on_init(self):        
        """        
        策略初始化
        """        
        self.write_log("策略初始化")        
        self.load_bar(10)
        
    def on_start(self):        
        """
        启动策略 
        """        
        self.write_log("策略启动")
        self.put_event()
        
    def on_stop(self):
        """        
        策略停止        
        """        
        self.write_log("策略停止")
        self.put_event()
        
    def on_tick(self, tick: TickData):
        """  
        TICK更新   
        """       
        self.bg.update_tick(tick)
        
    def on_bar(self, bar: BarData):  
        """       
        K线更新        
        """        
        self.bg.update_bar(bar)
        
    def on_5min_bar(self, bar: BarData):
        """5分钟K线更新"""
        # 全撤委托 
        self.cancel_all()
        
        # 缓存K线        
        am = self.am   
        am.update_bar(bar)    
        if not am.inited:     
            return
            
        # 计算技术指标      
        self.rsj_value = self.am.rsj(self.rsj_window)
        
        # 判断交易信号 
        if bar.datetime.time() == time(14, 50):  
            if self.rsj_value > 0:         
                if self.pos > 0:           
                    self.sell(bar.close_price - 10, 1)
                    
                self.short(bar.close_price - 10, 1)      
            elif self.rsj_value < 0:           
                if self.pos < 0:       
                    self.cover(bar.close_price + 10, 1)
                    
                self.buy(bar.close_price + 10, 1)
                
        # 更新图形界面     
        self.put_event()
        
    def on_order(self, order: OrderData):   
        """      
        Callback of new order data update. 
        """        
        pass
        
    def on_trade(self, trade: TradeData):   
        """   
        Callback of new trade data update.   
        """     
        self.put_event()
        
    def on_stop_order(self, stop_order: StopOrder):    
        """      
        Callback of stop order update.   
        """        
        pass


class NewArrayManager(ArrayManager):

    def __init__(self, size=100):    
        """"""  
        super().__init__(size)
        
        self.return_array: np.ndarray = np.zeros(size)
        
    def update_bar(self, bar: BarData) -> None:   
        """更新K线"""   
        # 先调用父类的方法，更新K线    
        super().update_bar(bar)
        # 计算涨跌变化   
        if not self.close_array[-2]:      # 如果尚未初始化上一根收盘价      
            last_return = 0    
        else:      
            last_return = self.close_array[-1] / self.close_array[-2] - 1
            
        # 缓存涨跌变化      
        self.return_array[:-1] = self.return_array[1:]     
        self.return_array[-1] = last_return
        
    def rsj(self, n: int) -> float:      
        """计算RSJ指标"""       
        # 切片出要计算用的收益率数据    
        return_data = self.return_array[-n:] 
        
        # 计算RV       
        rv = np.sum(pow(return_data, 2))
        
        # 计算RV +/-      
        positive_data = np.array([r for r in return_data if r > 0])       
        negative_data = np.array([r for r in return_data if r <= 0])
        
        rv_positive = np.sum(pow(positive_data, 2))    
        rv_negative = np.sum(pow(negative_data, 2))
        
        # 计算RSJ       
        rsj = (rv_positive - rv_negative) / rv    
        return rsj

原文：https://zhuanlan.zhihu.com/p/435526330

蜗牛之路

最新