##grid_trade_strategy.py
from vnpy_ctastrategy import (
CtaTemplate,
StopOrder,
TickData,
BarData,
TradeData,
OrderData,
BarGenerator,
ArrayManager,
)
from vnpy.trader.constant import OrderType, Offset, Direction
class GridTradingStrategy(CtaTemplate):
author = "用Python的交易员"
allow_order = False
grid_ask_base: float = 1.00
grid_bid_base: float = 1.00
grid_step: float = 0.00 ## jiacha
grid_step_percent : float= 3.00 ##baifenbi
grid_up : float = 100
grid_down : float = 1
grid_pos : int = 1
parameters = ["grid_up", "grid_down","grid_pos","grid_step","grid_step_percent"]
variables = ["grid_ask_base", "grid_bid_base"]
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)
self.am = ArrayManager(1)
def on_init(self):
"""
Callback when strategy is inited.
"""
self.write_log("策略初始化")
self.load_bar(0)
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.
"""
bid_diff = tick.bid_price_1 - self.grid_bid_base ##平多或者开空的价差(买入价价差)
ask_diff = tick.ask_price_1 - self.grid_ask_base ##开多或者平空的价差(卖出价价差)
bid_percent = 0
ask_percent = 0
if(self.grid_step_percent>0 and self.grid_bid_base!=0 and self.grid_ask_base!=0):
bid_percent = 100* bid_diff/self.grid_bid_base
ask_percent = 100* ask_diff / self.grid_ask_base
# 市场下跌时买多
if (ask_diff <= -1* self.grid_step and self.grid_step > 0 ) or ( self.grid_step_percent>0 and ask_percent<=-1 * self.grid_step_percent):
if self.grid_ask_base==1.00 or self.grid_bid_base==1.00 :
self.allow_order = False
else :
self.allow_order =True
self.grid_ask_base = tick.ask_price_1
self.grid_bid_base= tick.bid_price_1
if self.grid_ask_base >=self.grid_down and self.grid_ask_base<=self.grid_up and self.allow_order:
self.write_log("下跌开多")
##下跌时加多仓(加多仓位)
self.vt_orderid= self.buy(
tick.ask_price_1,
self.grid_pos
)
##to futures 下跌时平空仓(平空获利)
if self.vt_symbol.endswith(".CZCE") or self.vt_symbol.endswith(".DCE") or self.vt_symbol.endswith(".SHFE") and self.allow_order:
self.cover(
tick.ask_price_1,
self.grid_pos
)
# 市场上涨时平多
elif (bid_diff >= self.grid_step and self.grid_step>0) or (self.grid_step_percent>0 and bid_percent >=self.grid_step_percent):
self.write_log("上涨平多:" + str(self.pos))
if self.grid_ask_base==1.00 or self.grid_bid_base==1.00 :
self.allow_order = False
else :
self.allow_order =True
self.grid_bid_base = tick.bid_price_1
self.grid_ask_base=tick.ask_price_1
if self.grid_bid_base >=self.grid_down and self.grid_bid_base<=self.grid_up and self.allow_order:
##上涨时，平多仓
self.write_log("平多:" + str(self.vt_symbol))
self.sell(
tick.bid_price_1,
self.grid_pos
)
##to futures上涨时开空仓
if self.vt_symbol.endswith(".CZCE") or self.vt_symbol.endswith(".DCE") or self.vt_symbol.endswith(".SHFE") and self.allow_order:
self.short(
tick.bid_price_1,
self.grid_pos
)
self.bg.update_tick(tick)
# Update UI
if tick.datetime.second % 5 == 0 :
self.put_event()
def on_bar(self, bar: BarData):
"""
Callback of new bar data update.
"""
am = self.am
am.update_bar(bar)
if not am.inited:
return
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