Vnpy 中增加TD指标计算及BOLL指标的运行情况

马克序列 Demark Sequential (TD 序列 ) 也算是一个常见的趋势指标,但是Vnpy中默认没有实现。

在Vnpy的安装目录Lib\site-packages\vnpy\trader下的utilty.py的ArrayManager中增加具体实现即可。

    def td(self, array=False):
        close_np = self.close
        close_shift = np.empty_like(self.close)
        close_shift[:4] = 0
        close_shift[4:] = close_np[:-4]
        compare_array = close_np > close_shift
        result = np.empty(len(close_np), int)
        counting_number: int = 0
        for i in range(len(close_np)):
            if np.isnan(close_shift[i]):
                result[i] = 0
            else:
                compare_bool = compare_array[i]
                if compare_bool:
                    if counting_number >= 0:
                        counting_number += 1
                    else:
                        counting_number = 1
                else:
                    if counting_number <= 0:
                        counting_number -= 1
                    else:
                        counting_number = -1
                result[i] = counting_number
        if array:
            return result
        return result[-1]  

调用的时候也很简单,在策略文件直接调用即可。

self.td_value = self.am.td()
self.write_log(f”td_value: {self.td_value}”)

Boll指标用来指示当前的运行情况,

class BollArray(object):
    """
    For:
    1. time series container of bar data
    2. calculating technical indicator value
    """

    def __init__(self, size=100):
        """Constructor"""
        self.count = 0
        self.size = size
        self.inited = False
        self.count = 0
        self.boll_up = 0
        self.boll_down = 0
        self.boll_mid = 0
        self.boll_up_cnt=0
        self.boll_up_brk_cnt=0
        self.boll_down_cnt=0
        self.boll_down_brk_cnt=0
        
        self.boll_up_array = np.zeros(size)
        self.boll_down_array = np.zeros(size)
        self.boll_mid_array = np.zeros(size)
        self.boll_up_cnt_array = np.zeros(size)
        self.boll_up_brk_cnt_array = np.zeros(size)
        self.boll_down_cnt_array = np.zeros(size)
        self.boll_down_brk_cnt_array = np.zeros(size)

    def update_boll(self, bar, am):
        """
        Update new bar data into array manager.
        """
        self.count += 1
        if not self.inited and self.count >= self.size:
            self.inited = True
            
        self.boll_up_array[:-1] = self.boll_up_array[1:]
        self.boll_down_array[:-1] = self.boll_down_array[1:]
        self.boll_mid_array[:-1] = self.boll_mid_array[1:]
        self.boll_up_cnt_array[:-1] = self.boll_up_cnt_array[1:]
        self.boll_up_brk_cnt_array[:-1] = self.boll_up_brk_cnt_array[1:]
        self.boll_down_cnt_array[:-1] = self.boll_down_cnt_array[1:]
        self.boll_down_brk_cnt_array[:-1] = self.boll_down_brk_cnt_array[1:]

        
        self.boll_up, self.boll_down = am.boll(20,2)
        self.boll_mid = (self.boll_up + self.boll_down) / 2
        
        if bar.low_price > self.boll_mid:
            self.boll_up_cnt = self.boll_up_cnt + 1
            self.boll_down_cnt = 0
        else:
            self.boll_up_cnt = 0

        if bar.high_price < self.boll_mid:
            self.boll_down_cnt = self.boll_down_cnt + 1
            self.boll_up_cnt = 0
        else:
            self.boll_down_cnt = 0

        if bar.high_price > self.boll_up:
            self.boll_up_brk_cnt = self.boll_up_brk_cnt + 1
        else:
            self.boll_up_brk_cnt = 0

        if bar.low_price < self.boll_down:
            self.boll_down_brk_cnt = self.boll_down_brk_cnt + 1
        else:
            self.boll_down_brk_cnt = 0
                
        self.boll_up_array[-1] = self.boll_up
        self.boll_down_array[-1] = self.boll_down
        self.boll_mid_array[-1] = self.boll_mid
        self.boll_up_cnt_array[-1] = self.boll_up_cnt
        self.boll_up_brk_cnt_array[-1] = self.boll_up_brk_cnt 
        self.boll_down_cnt_array[-1] = self.boll_down_cnt
        self.boll_down_brk_cnt_array[-1] = self.boll_down_brk_cnt 
        
    def boll(self, array=False):
    
        if array:
            return self.boll_up_array, self.boll_up_cnt_array, self.boll_up_brk_cnt_array,self.boll_down_array,self.boll_down_cnt_array,self.boll_down_brk_cnt_array
        return self.boll_up, self.boll_up_cnt, self.boll_up_brk_cnt,self.boll_down,self.boll_down_cnt,self.boll_down_brk_cnt

运行的时候需要引入类。self.boll1.update_boll(bar,self.am1)传入bar和am数据后即可使用类中的变量值

self.boll1.update_boll(bar,self.am1)
        self.boll_up, self.boll_up_cnt, self.boll_up_brk_cnt,self.boll_down,self.boll_down_cnt,self.boll_down_brk_cnt = self.boll1.boll()