`from vnpy.app.cta_strategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
from vnpy.trader.constant import Interval
class DoubleMaStrategy(CtaTemplate):
author = "用Python的交易员"
fast_window = 7 # 快线
slow_window = 30 # 慢线
x_min = 15 # 周期
lots = 1 # 开仓手
save = 1 # 止损
startstop = 1 # 开始止盈
stoploss = 1 # 回撤点位
fast_ma0 = 0
fast_ma1 = 0
slow_ma0 = 0
slow_ma1 = 0
price = 0 # tick实时价格
bartime = "" # 时间的显示
avg_buy_price = 0
avg_sell_price = 0
highest = 0
lowest = 0
parameters = ["fast_window", "slow_window", "x_min", "lots", "startstop", "save"]
variables = ["bartime", "price", "fast_ma0", "fast_ma1", "slow_ma0", "slow_ma1","highest","lowest"]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
""""""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.bg_x = BarGenerator(self.on_bar, self.x_min, self.on_x_bar, Interval.MINUTE)
self.am_x = ArrayManager()
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(10)
def on_start(self):
"""
Callback when strategy is started.
"""
self.write_log("策略启动")
self.put_event()
def on_stop(self):
"""
Callback when strategy is stopped.
"""
self.write_log("策略停止")
self.put_event()
def on_tick(self, tick: TickData):
"""
Callback of new tick data update.
"""
self.bg_x.update_tick(tick)
self.price = tick.last_price
self.put_event()
def on_bar(self, bar: BarData):
self.bg_x.update_bar(bar)
def on_x_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
self.cancel_all()
am = self.am_x
am.update_bar(bar)
if not am.inited:
return
self.fast_ma0 = am.close_array[-self.fast_window:-1].mean()
self.fast_ma1 = am.close_array[-self.fast_window - 1:-2].mean()
self.slow_ma0 = am.close_array[-self.slow_window:-1].mean()
self.slow_ma1 = am.close_array[-self.slow_window - 1:-2].mean()
if self.pos == 0:
if self.fast_ma1 < self.slow_ma1 and self.fast_ma0 > self.slow_ma0:
# 金叉
self.buy(bar.close_price, self.lots)
self.highest = bar.close_price
self.avg_buy_price = bar.close_price
elif self.fast_ma1 > self.slow_ma1 and self.fast_ma0 < self.slow_ma0:
# 死叉
self.short(bar.close_price, self.lots)
self.lowest = bar.close_price
self.avg_sell_price = bar.close_price
elif self.pos > 0:
# 多单止损
self.highest = max(bar.high_price, self.highest)
if bar.close_price < self.avg_buy_price - self.save:
self.sell(bar.close_price, self.lots)
elif self.highest > self.avg_buy_price + self.startstop:
if bar.close_price < self.highest - self.stoploss:
self.sell(bar.close_price, self.lots)
# 止盈
elif self.pos < 0:
# 空单止损
self.lowest = min(bar.low_price, self.lowest)
if bar.close_price > self.avg_sell_price + self.save:
self.cover(bar.close_price, self.lots)
elif self.lowest < self.avg_sell_price - self.startstop:
if bar.close_price > self.lowest + self.stoploss:
self.cover(bar.close_price, self.lots)
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
`
