1. 被冤枉的通达信函数“未来函数”XMA
提到通达信函数XMA，人们最常见到的词汇是“未来函数”、“欺骗”、“陷阱”、“坑”......等等不好的字眼，仿佛XMA函数是个捉摸不定的未来函数，是你亏损的根源！
其实大家对这个函数不了解，如果你了解了它的实现机理，它的优点和不得已的缺点，扬长避短，是完全可以使用的。
1.1 MA与XMA的区别
未来叙述的方便，先指标一个供MA与XMA计算的数组：
位置：[ A B C D E F H I J K L M N O P Q R S T U ]
数据：[ 1. 2. 3. 4. 5. 6. 7. 8. 9. 10. 11. 12. 13. 14. 15. 16. 17. 18. 19. 20.]
分析周期N=5
1.1.1 滞后的N日MA均价
位置：[ A B C D E F H I J K L M N O P Q R S T U ]
MA5 [nan nan nan nan 3. 4. 5. 6. 7. 8. 9. 10. 11. 12. 13. 14. 15. 16. 17. 18.]
- A = nan
- B = nan
- C = nan
- D = nan
- E = (1+2+3+4+5)/5
- F = (2+3+4+5+6)/5
- G = (3+4+5+6+7)/5
- ... ...
- U = (16+17+18+19+20)/5
1.1.2 XMA的N日均价
位置：[ A B C D E F H I J K L M N O P Q R S T U ]
XMA5:[ 2. 2.5 3. 4. 5. 6. 7. 8. 9. 10. 11. 12. 13. 14. 15. 16. 17. 18. 18.5 19. ]
- A = (1+2+3)/3=2
- B = (1+2+3+4)/4=2.5
- C = (1+2+3+4+5)/5=3
- D = (2+3+4+5+6)/5=4
- E = (3+4+5+6+7)/5=5
- F = (4+5+6+7+8)/5=6
- ... ...
- S = (16+17+18+19+20)/5
- T = (17+18+19+20)/4
- U = (18+19+20)/3
1.1.3 人们期望的N日均价
位置：[ A B C D E F H I J K L M N O P Q R S T U W(21) X(22)]
XMA5:[ nan nan nan nan 5. 6. 7. 8. 9. 10. 11. 12. 13. 14. 15. 16. 17. 18. 19. 20. ]
1.2 MA与XMA的相同与区别
它们的相同点是均值算法都是一样的，都是N个数的算术平均值，但是均值代表的位置是不同的。
N日MA均值是必须满足N个数据，不足时为nan，其所代表的位置是第N日的均值；
N日XMA均值是必须满足N//2+1个数据，不足时为没有意义，其所代表的位置是第N//2+1日的均值，它的计算是用前N//2个数数据、当前日和后N//2个数数据。当数据个数多于N//2个但不够N数的时候，是有几个数算几个数的均线。
以N=5为XMA例：
- A位置时它取A，B，C位置的数，求3个算术均值，
- 位置时它取A，B，C，D位置的数，求4个算术均值，
- 到了C位置时它取A，B，C，D，E位置的数，求5个数的算术均值。
- 到了S位置时它取Q、R、S、T、U位置的数，求5个数的算术均值。
- 到了T位置时它取R、S、T、U位置的数，求4个数的算术均值。
- 到了U位置时它取S、T、U位置的数，求3个数的算术均值。
MA具有确定性，不会受到下一个数据的变化的影响，而其计算出来的结果一定是滞后的。
从这里我们可以看到，XMA其实中历史数据个数大于N的时候，它的结果能够代表人们通常理解的均值，但是因为它用的数据包含当前位置之后N//2日的数据，因为使用预测的数据，XMA随着计算位置的不同，其计算的数据范围发生了变化。
注意：XMA并没有用到预测的数据，它只是在不满足N个计算数据时缩短了N！随着新的数据的到来，它后面N//2个数是会发生变化的，XMA明确而且简单，不神秘！
2. XMA的python实现
以下为用python实现XMA函数代码，已经与通达信的自带函数XMA函数做了严格的验证，数据相同的情况下，XMA的均值是完全相同的，也就是说是可靠移植的，可以信赖的！
如果有怀疑，可以自行验证的。
2.1 XMA()函数
def XMA(src:np.ndarray,N:int,array=False) -> np.ndarray:
""" XMA函数的python实现 """
data_len = len(src)
half_len : int = (N // 2)+(1 if N % 2 else 0)
if data_len < half_len:
out = np.array([np.nan for i in range(data_len)],dtype=float)
if array:
return out
return out[-half_len:] if len(out) else np.array([],dtype=float)
head = np.array([talib.MA(src[0:ilen],ilen)[-1] for ilen in range(half_len,N)])
out = head
if data_len >= N:
body = talib.MA(src,N)[N-1:]
out = np.append(out,body)
tail = np.array([talib.MA(src[-ilen:],ilen)[-1] for ilen in range(N-1,half_len-1,-1)])
out = np.append(out,tail)
if array:
return out
return out[-half_len:]2.2 为vnpy的数组管理器增加xma()成员函数
class DynaArrayManager(ArrayManager):
"""
DynaArrayManager是对ArrayManager的扩展，解决它无法用于临时K线的指标计算问题。
作者：hxxjava
"""
def __init__(self, size: int = 100) -> None:
super().__init__(size)
self.bar_datetimes:List[datetime] = []
def update_bar(self, bar: BarData) -> None:
if not self.bar_datetimes or self.bar_datetimes[-1] < bar.datetime:
self.bar_datetimes.append(bar.datetime)
super().update_bar(bar)
else:
"""
Only Update all arrays in array manager with temporary bar data.
"""
self.open_array[-1] = bar.open_price
self.high_array[-1] = bar.high_price
self.low_array[-1] = bar.low_price
self.close_array[-1] = bar.close_price
self.volume_array[-1] = bar.volume
self.turnover_array[-1] = bar.turnover
self.open_interest_array[-1] = bar.open_interest
# ... ... 其他无关XMA部分略去
def xma(self,select:str="C",N:int=3,array: bool = False) -> np.ndarray:
""" XMA函数的 """
if select.upper() == "C":
src = self.close
elif select.upper() == "O":
src = self.open
elif select.upper() == "H":
src = self.high
elif select.upper() == "L":
src = self.low
else:
assert(False)
data_len = len(src)
half_len : int = (N // 2)+(1 if N % 2 else 0)
if data_len < half_len:
out = np.array([np.nan for i in range(data_len)],dtype=float)
if array:
return out
return out[-half_len:] if len(out) else np.array([],dtype=float)
head = np.array([talib.MA(src[0:ilen],ilen)[-1] for ilen in range(half_len,N)])
out = head
if data_len >= N:
body = talib.MA(src,N)[N-1:]
out = np.append(out,body)
tail = np.array([talib.MA(src[-ilen:],ilen)[-1] for ilen in range(N-1,half_len-1,-1)])
out = np.append(out,tail)
if array:
return out
return out[-half_len:]3. 在vnpy实现显示XMA的指标组件XmaItem
3.1 XmaItem指标组件实现的难点
可以这么说：如果会写XMA的曲线的vnpy显示组件，已经可以写作任何vnpy显示组件了，它不是一般的难！
- 考虑到XMA的后N//2的均值，会随随着新的数据的到来，后N//2个数是会发生变化的特点，XmaItem需要在每次计算新的XMA的值之后，会得到N//2+1个均值结果。我们需要将这个N//2+1个均值替换之前的XMA结果值。
- 还是因为后N//2个数是会不停发生变化，XmaItem指标显示组件，需要不同刷新后N//2+1个K线的显示曲线。否则会看到一个断裂带尾部曲线。
3.2 XmaItem指标显示组件的实现代码
class XmaItem(CandleItem):
""""""
def __init__(self, manager: BarManager,xma_window:int=10):
""""""
super().__init__(manager)
self.line_pen: QtGui.QPen = pg.mkPen(color=(100, 100, 255), width=2)
self.xma_window = xma_window
self.dyna_am = DynaArrayManager(xma_window)
self.xma_data: Dict[int, float] = {}
def update_history(self, history: List[BarData]) -> None:
""" reimpliment of update_history """
for bar in history:
self.dyna_am.update_bar(bar)
super().update_history(history)
def update_bar(self, bar: BarData) -> None:
""" reimpliment of update_bar """
self.dyna_am.update_bar(bar)
super().update_bar(bar)
def _get_xma_value(self, ix: int) -> float:
""" """
max_ix = self._manager.get_count() - 1
if ix < 0 or ix > max_ix:
return np.nan
# When initialize, calculate all rsi value
if not self.xma_data:
bars = self._manager.get_all_bars()
close_data = [bar.close_price for bar in bars]
sma_array = XMA(np.array(close_data), self.xma_window,array=True)
for n, value in enumerate(sma_array):
self.xma_data[n] = value
# Return if already calcualted
if ix != max_ix and ix in self.xma_data:
return self.xma_data[ix]
if self.dyna_am.inited:
values = self.dyna_am.xma(select='C',N=self.xma_window)
vlen = len(values)
# print(f"vlen={vlen},values={list(values)}")
start_ix = ix-vlen+1
tail_idxs = [(start_ix+i,values[i]) for i in range(vlen)]
for idx,xma in tail_idxs:
self.xma_data[idx] = xma
return self.xma_data[ix]
return np.nan
def _draw_bar_picture(self, ix: int, bar: BarData) -> QtGui.QPicture:
""" """
# Create objects
picture = QtGui.QPicture()
painter = QtGui.QPainter(picture)
# Draw XMA Line
xma_value = self._get_xma_value(ix)
last_xma_value = self._get_xma_value(ix - 1)
if not (np.isnan(xma_value) or np.isnan(last_xma_value)):
# Set painter color
painter.setPen(self.line_pen)
# draw XMA line
start_point = QtCore.QPointF(ix-1, last_xma_value)
end_point = QtCore.QPointF(ix, xma_value)
painter.drawLine(start_point, end_point)
# Finish
painter.end()
return picture
def paint(self, painter: QtGui.QPainter, opt: QtWidgets.QStyleOptionGraphicsItem, w: QtWidgets.QWidget) -> None:
"""
Reimplement the paint method of parent class.
This function is called by external QGraphicsView.
"""
# return super().paint(painter, opt, w)
rect = opt.exposedRect
half = (self.xma_window // 2)
min_ix: int = int(rect.left()) - half
min_ix: int = max(min_ix,0)
max_ix: int = int(rect.right())
max_ix: int = min(max_ix, len(self._bar_picutures))
rect_area: tuple = (min_ix, max_ix)
if (
self._to_update
or rect_area != self._rect_area
or not self._item_picuture
):
self._to_update = False
self._rect_area = rect_area
# print(f"XmaItem _draw_item_picture:{min_ix}--{max_ix}")
self._draw_item_picture(min_ix, max_ix)
self._item_picuture.play(painter)
def _draw_item_picture(self, min_ix: int, max_ix: int) -> None:
"""
Draw the picture of item in specific range.
"""
self._item_picuture = QtGui.QPicture()
painter: QtGui.QPainter = QtGui.QPainter(self._item_picuture)
for ix in range(min_ix, max_ix):
bar_picture: QtGui.QPicture = self._bar_picutures[ix]
# if bar_picture is None:
bar: BarData = self._manager.get_bar(ix)
bar_picture = self._draw_bar_picture(ix, bar)
self._bar_picutures[ix] = bar_picture
bar_picture.play(painter)
painter.end()
def get_info_text(self, ix: int) -> str:
""""""
if ix in self.xma_data:
xma_value = self.xma_data[ix]
text = f"XMA({self.xma_window}): {xma_value:.2f}"
else:
text = "XMA({self.xma_window}) -"
return text3.3 XMA指标显示的效果
尾部不断变化的XMA曲线：
