vn.py官网
开源量化社区
Member
avatar
加入于:
帖子: 3115
声望: 181

发布于vn.py社区公众号【vnpy-community】
 
原文作者:Bili | 发布时间:2021-07-29
 

系列前言

 

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

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

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

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

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

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

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

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

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

 

基本信息

 

description
 
研报PDF

 

策略原理

 

RV计算公式

高频已实现波动率(Realized Volatility)是一种根据高频数据(这里指的是日内分钟线级别的数据,而非Tick级别数据)计算的日内波动率指标,该指标的计算公式为:

description

RSJ计算公式

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

description

RSJ指标应用

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

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

  • RSJ数值越大,则说明未来越倾向于下跌;
  • RSJ数值越小,则说明未来越倾向于上涨。

策略核心逻辑

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

具体策略逻辑:

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

该策略的特点:

  1. 策略始终在市场中(持有仓位),要么做多,要么做空;
  2. 每天最多执行一次交易(如果信号方向变化)。

 

代码实现

 

计算RSJ指标

由于talib工具库中没有提供RSJ指标的计算功能,因此我们需要自行实现其计算代码,好在整个公式并不复杂。为了保持代码的简洁清爽,我们选择基于vn.py中内置的时间序列缓存容器ArrayManager,来进行扩展实现RSJ的计算。

首先创建子类NewArrayManager继承父类ArrayManager,用于按时间序列缓存bar数据,提供技术指标的计算。注意在构造函数init中,需要传入一个size参数(缓存数据的长度),将其默认值设为100(表示缓存的K线数量为100根),同时也增加了一个用于缓存收益率序列的return_array数组:

class NewArrayManager(ArrayManager):
    def __init__(self, size=100):
        """"""        
        super().__init__(size)        
        self.return_array: np.ndarray = np.zeros(size)

接下来构造update_bar函数用于更新K线到容器中,除了缓存K线本身的OHLCV数据外,也同时将收益率数据计算好缓存在return_array数组中,方便后续计算RSJ指标:

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

之后即可实现RSJ的计算函数,使用上一步已经缓存了的return_array,结合NumPy的向量化计算功能来保证计算速度:

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

交易信号执行

由于选择了5分钟K线来计算RSJ指标,因此我们需要先使用BarGenerator将1分钟K线合成为5分钟K线,在RsjStrategy类的构造函数下创建NewArrayManager对象am和BarGenerator对象bg,其中bg对象传入了额外的5分钟合成回调函数(self.on_5min_bar):

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()

收到1分钟K线的推送后,将其推送到bg中合成5分钟K线:

def on_bar(self, bar: BarData):        
        """        
        K线更新
        """        
        self.bg.update_bar(bar)

每当有5分钟K线合成后,on_5min_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)

一些注意点:

  • vn.py中的K线采用合成时间段的开始时间戳,因此14:55分合成出的K线其时间戳(datetime)应该为14:50;
  • 当发出做多信号时,策略应该持有1手的多头仓位,此时如果有昨日空头仓位,应该先买入平仓(cover),再买入开仓(buy);
  • 当发出做多信号时,策略应该持有1手的空头仓位,此时如果有昨日多头仓位,应该先卖出平仓(sell),再卖出开仓(short)。

 

回测结果

 

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

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

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

description

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

description

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

  • 总交易日: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

 

Member
avatar
加入于:
帖子: 1
声望: 0

这个系列很有参考价值,希望能够持续更新啊

© 2015-2019 上海韦纳软件科技有限公司
备案服务号:沪ICP备18006526号-3

沪公网安备 31011502017034号