from vnpy_ctastrategy import (
CtaTemplate,
BarGenerator,
ArrayManager,
TickData,
BarData,
OrderData,
TradeData,
StopOrder
)
from typing import Any
class Ma1cdStrategy(CtaTemplate):
author = "黑驴"
fast_window = 6
slow_window = 10
signal_window = 5
fixed_size = 1
fast_ma = 0.0
slow_ma = 0.0
ma1cd = 0.0
signal = 0.0
histogram = 0.0
parameters: list = [
"fast_window",
"slow_window",
"signal_window",
"fixed_size"
]
variables: list = [
"fast_ma",
"slow_ma",
"ma1cd",
"signal",
"histogram"
]
def __init__(
self,
cta_engine: Any,
strategy_name: str,
vt_symbol: str,
setting: dict,
) -> None:
""""""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.bg15 = BarGenerator(self.on_bar, 15, self.on_15min_bar)
self.am15 = ArrayManager()
def on_init(self) -> None:
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(10)
def on_start(self) -> None:
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
def on_stop(self) -> None:
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
def on_tick(self, tick: TickData) -> None:
"""
Callback of new tick data update.
"""
self.bg15.update_tick(tick)
def on_bar(self, bar: BarData) -> None:
"""
Callback of new bar data update.
"""
self.bg15.update_bar(bar)
def on_15min_bar(self, bar: BarData):
""""""
self.cancel_all()
self.am15.update_bar(bar)
if self.am15.inited:
return
self.fast_ma = self.am15.sma(bar.close, self.fast_window)
self.slow_ma = self.am15.sma(bar.close, self.slow_window)
self.ma1cd = self.fast_ma - self.slow_ma
self.signal = self.am15.ema(self.ma1cd, self.signal_window)
self.histogram = self.ma1cd - self.signal
cross_above = self.histogram > 0 and self.am15.cross_above(self.histogram, 0)
cross_below = self.histogram < 0 and self.am15.cross_below(self.histogram, 0)
if cross_above:
if self.pos == 0:
self.buy(bar.close, self.fixed_size, stop=True)
elif self.pos < 0:
self.cover(bar.close, abs(self.pos), stop=True)
self.buy(bar.close, self.fixed_size, stop=True)
elif cross_below:
if self.pos == 0:
self.short(bar.close, self.fixed_size, stop=True)
elif self.pos > 0:
self.sell(bar.close, self.pos, stop=True)
self.short(bar.close, self.fixed_size, stop=True)
self.put_event()
def on_trade(self, trade: TradeData) -> None:
"""
Callback of new trade data update.
"""
self.put_event()
def on_order(self, order: OrderData) -> None:
"""
Callback of new order data update.
"""
pass
def on_stop_order(self, stop_order: StopOrder) -> None:
"""
Callback of stop order update.
"""
pass
