马克序列 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()