代码
from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
class Grid(CtaTemplate):
"""网格交易策略"""
author = "WeiRui"
Grid_distance = 100
fixed_size = 1
max_layers = 4
long_stop = 0
short_stop = 0
parameters = [
"Grid_distance",
"fixed_size",
"max_layers",
]
variables = [
"long_stop",
"short_stop",
"current_layer"
]
def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
"""初始化"""
super().__init__(cta_engine, strategy_name, vt_symbol, setting)
self.bg = BarGenerator(self.on_bar, 15, self.on_15min_bar)
self.am = ArrayManager()
self.initial_price = 0
self.current_layer = 0
def on_init(self):
"""策略初始化"""
self.write_log("策略初始化")
self.load_bar(10)
def on_start(self):
"""策略启动"""
self.write_log("策略启动")
def on_stop(self):
"""策略停止"""
self.write_log("策略停止")
def on_tick(self, tick: TickData):
"""新的tick数据更新"""
self.bg.update_tick(tick)
def on_bar(self, bar: BarData):
"""新的bar数据更新"""
self.bg.update_bar(bar)
def on_15min_bar(self, bar: BarData):
"""15分钟bar数据更新"""
self.cancel_all()
am = self.am
am.update_bar(bar)
if not am.inited:
return
if self.pos == 0:
# 开立初始多头仓位和空头仓位
self.buy(bar.close_price, self.fixed_size)
self.short(bar.close_price, self.fixed_size)
self.initial_price = bar.close_price
self.current_layer = 0
elif bar.close_price >= self.initial_price + (self.current_layer + 1) * self.Grid_distance:
# 平掉多头仓位
self.sell(bar.close_price, self.fixed_size)
# 再次开立一手多单和空单
self.buy(bar.close_price, self.fixed_size)
self.short(bar.close_price, self.fixed_size)
# 更新初始开仓价格和当前网格层
self.initial_price = bar.close_price
self.current_layer += 1
if self.current_layer == self.max_layers:
# 平掉所有仓位
self.cover(bar.close_price, abs(self.pos))
# 重置网格层和初始价格
self.current_layer = 0
self.initial_price = bar.close_price
elif bar.close_price <= self.initial_price - (self.current_layer + 1) * self.Grid_distance:
# 平掉空头仓位
self.cover(bar.close_price, self.fixed_size)
# 再次开立一手多单和空单
self.buy(bar.close_price, self.fixed_size)
self.short(bar.close_price, self.fixed_size)
# 更新初始开仓价格和当前网格层
self.initial_price = bar.close_price
self.current_layer += 1
if self.current_layer == self.max_layers:
# 平掉所有仓位
self.cover(bar.close_price, abs(self.pos))
# 重置网格层和初始价格
self.current_layer = 0
self.initial_price = bar.close_price
# 检查是否回撤到上一层网格
elif bar.close_price <= self.initial_price - self.Grid_distance:
# 平掉所有仓位
self.cover(bar.close_price, abs(self.pos))
# 重置网格层和初始价格
self.current_layer = 0
self.initial_price = bar.close_price
self.put_event() # 发送事件通知其他组件
def on_order(self, order: OrderData):
"""新的订单数据更新"""
pass
def on_trade(self, trade: TradeData):
"""新的成交数据更新"""
self.put_event() # 发送事件通知其他组件
def on_stop_order(self, stop_order: StopOrder):
"""停止单更新"""
pass
