蜗牛博客VNPY学习记录：
VN.PY 2.0学习记录一（如何回测）
VN.PY 2.0学习记录二（策略开发）
Vn.py学习记录三（米筐教程)
VN.PY 2.0学习记录四（多线程、多进程）
Vn.py学习记录五–交易时间段及Widgets
Vn.py学习记录六（无界面模拟盘）
Vn.py学习记录七（V2.0.5版本）
Vnpy学习记录八（R-Breaker及pickle）
Vn.py学习记录九（事件驱动引擎）
VN.PY学习记录十（源码概述）
VNPY学习记录11（微信+Vscode）
VNPY学习记录12（父子进程、回调函数）
VNPY学习记录13（部署到云服务器，实现自动交易）
一、其他
def onBar(self,bar):
#更新策略执行的时间（用于回测时记录发生时间）
#从datasource拿到的bar.datetime是bar的结束时间，比如：9:00-9:05的time就是9:05=curDatetime因为策略中已经将9:05调整为9:00，所以这里要加回来。
self.curDatetime = bar.datetime + timedelta(seconds = self.lineM5.barTimeInterval)
#2 计算交易时间和平仓时间
self.__timeWindow(bar.datetime)
#推送tick到15分钻K线
self.lineM5.addBar(bar)
def __timeWindow(self,dt):
"""交易与平仓窗口"""
self.tradeWindow = False
self.closeWindow = False
self.openWindow = False
#开市期，波动较大，用于判断止损止盈，或开仓
if (dt.hour == 9 or dt.hour == 21) and dt.minute < 2:
self.openWindow = Ture
#日盘
if dt.hour == 9 and dt.minute >= 0:
self.tradeWindow = Ture
return
if dt.hour == 10:
if dt.minute <= 15 or dt.minute >= 30:
self.tradeWindow = Ture
return
if dt.hour == 11 and dt.minute <= 30:
self.tradeWindow = Ture
return
if dt.hour == 13 and dt.minute >= 30:
self.tradeWindow = Ture
return
if dt.hour == 14:
if dt.minute < 59:
self.tradeWindow = Ture
return
if dt.minute == 59: #日盘平仓
self.closeWindow = Ture
return
#夜盘
if dt.hour == 21 and dt.minute >= 0:
self.tradeWindow = Ture
return
#上期 贵金属 次日 02:30
if self.shortSymbol in NIGHT_MARKET_SQ1:
if dt.hour == 22 or dt.hour == 23 or dt.hour == 0 or dt.hour == 1:
self.tradeWindow = Ture
return
if dt.hour == 2:
if dt.minute < 29: #收市前29分钟
self.tradeWindow = Ture
return
if dt.minute == 29: #夜盘平仓
self.closeWindow = Ture
return
return
#上期 有色金额 黑色金额 沥青 次日 01:00
if self.shortSymbol in NIGHT_MARKET_SQ2:
if dt.hour == 22 or dt.hour == 23:
self.tradeWindow = Ture
return
if dt.hour == 0:
if dt.minute < 59: #收市前59分钟
self.tradeWindow = Ture
return
if dt.minute == 59: #夜盘平仓
self.closeWindow = Ture
return
return
#上期 天然橡胶 23:00
if self.shortSymbol in NIGHT_MARKET_SQ3:
if dt.hour ==22:
if dt.minute < 59: #收市前1分钟
self.tradeWindow = Ture
return
if dt.minute == 59: #夜盘平仓
self.closeWindow = Ture
return
return
#也可以写死
def __dailyClosecheck(self,bar):
"""
每天收盘前检查，如果是亏损单，则平掉
"""
if self.position.pos == 0 and self.entrust == 0:
return False
if bar.time not in ['14:45:00','14:50:00','14:55:00','22:45:00','22:50:00','22:55:00',]:
return False
#撤销未成交的订单
if len(self.uncompletedOrders) > 0:
for order in self.uncompletedOrders.keys():
self.writeCtaLog("{}，收盘前15分钟，仍未成交，取消委托单：{}".format(bar.datetime,order))
self.cancelOrder(str(order))
self.uncompletedOrders.clear()
self.entrust = 0
#强制平仓
if self.position.pos > 0 and bar.close < self.policey.entryPrice + self.atr:
self.writeCtaLog("强制日内平亏损多仓")
#降低两个滑点
orderid = self.sell(price = bar.close-2*self.minDiff, volume = self.inputSS, orderTime = self.curDatetime)
if orderid:
#更新下单时间
self.lastOrderTime = self.curDatetime
return Ture
if self.position.pos < 0 and bar.close > self.policey.entryPrice - self.atr:
self.writeCtaLog("强制日内平亏损空仓")
#降低两个滑点
orderid = self.over(price = bar.close+2*self.minDiff, volume = self.inputSS, orderTime = self.curDatetime)
if orderid:
#更新下单时间
self.lastOrderTime = self.curDatetime
return Ture
return Ture
二、备留后用：
# 日盘平仓时间
if nowTime >= datetime.time(hour=14, minute=56) and nowTime < datetime.time(hour=15, minute=00):
self.cancelAll() # 撤销所有未成交委托
if self.pos > 0:
self.sell(bar.close-5,abs(self.pos)) # 超价卖平仓
elif self.pos < 0:
self.cover(bar.close+5,abs(self.pos)) # 超价买平仓
# 夜盘平仓时间
if nowTime >= datetime.time(hour=23, minute=26) and nowTime < datetime.time(hour=23, minute=30):
self.cancelAll() # 撤销所有未成交委托
if self.pos > 0:
self.sell(bar.close-5,abs(self.pos)) # 超价卖平仓
elif self.pos < 0:
self.cover(bar.close+5,abs(self.pos)) # 超价买平仓
其中：
import datetime print(datetime.time(hour=14, minute=56))
的执行结果就是：14:56:00
三、Widgets开发
1.首先准备图片以及文件
2.demoEngine.py
from vnpy.event import Event
from vnpy.trader.vtEvent import EVENT_TICK
from vnpy.trader.vtObject import VtSubscribeReq
EVENT_DEMO_LOG = "eDemolog"
class DemoEngine(object):
def __init__(self,mainEngine,eventEngine):
self.mainEngine = mainEngine
self.eventEngine = eventEngine
sefl.symbol = 'rb1805'
self.priceDict = {}
self.registerEvent()
def subscribeData(self):
req = VtSubscribeReq()
req.symbol = self.symbol
slef.mainEngine.subscribe(req,'CTP')
def registerEvent(self):
sefl.eventEngine.register(EVENT_TICK,self.processTickEvent)
def processTickEvent(self,event):
tick = event.dict_['data']
last = tick.lastPrice
if self.priceDict:
if last >= self.priceDict['ask']:
result = "外盘"
elif last <= self.priceDict['bid']:
result = "内盘"
else:
result = "中性"
self.writeLog("最新Tick的成交价为s%,交易方向为：s%" %(last,result))
self.priceDict['bid'] = tick.bidPrice
self.priceDict['ask'] = tick.askPrice
def writeLog(self,msg):
event = Event(EVENT_DEMO_LOG)
event.dict_['data'] = msg
self.eventEngine.put(event)
3.uiDemoWidget.py
from datetime import datetime
from .demoEngine import EVENT_DEMO_LOG
from vnpy.event import event
from vnpy.trader.uiQt import QtCore,QtWidgets
class ClassName(QtWidgets.Qwidget):
signalLog = QtCore.Signal(type(Event()))
"""docstring for ClassName"""
def __init__(self, demoEngine,eventEngine,parent=None):
super(DemoWidget, self).__init__(parent)
self.demoEngine = demoEngine
self.eventEngine = eventEngine
self.initUi()
self.registerEvent()
def initUi(self):
self.setWindowTitle("DemoAPP")
self.botton = QtWidgets.QPushButton("订阅行情")
self.button.click.connect(self.demoEngine.subscribeData)
self.logMoniter = QtWidgets.QTextEdit()
self.logMoniter.setReadOnly(True)
#横向排版
hbox = QtWidgets.QHBoxLayout()
hbox = addWidget(self.button)
hbox.addStretch()
#竖向排版
vbox = QtWidgets.QVBoxLayout()
vbox = addLayout(hbox)
vbox.addWidget(self.logMoniter)
self.selfLayout(vbox)
def registerEvent():
self.signalLog.connect(self.processLogEvent)
sefl.eventEngine.register(EVENT_DEMO_LOG,self.signalLog.emit)
def processLogEvent(self,event):
msg = event.dict_['data']
t = datetime.now()
mas = str(t)+ " " + msg
self.logMoniter.append(msg)
4. init.py
from demoEngine import DemoEngine from uiDemoWidget import DemoWidget app_name = "DemoAPP" appDisplayName = "DemoAPP" appEngine = DemoEngine appWidget = DemoWidget appIco = "demo.ico"
然后再修改run.py，就可以跑起来测试了。