蜗牛博客VNPY源码学习系列文章
VNPY源码（一）CTP封装及K线合成
VNPY源码（二）API获取行情和script_trader
VNPY源码（三）主引擎MainEngine
VNPY源码（四）DataRecorder
VNPY源码（五）CtaEngine实盘引擎
VNPY源码（六）BacktesterEngine回测引擎
VNPY源码（七）限价单与停止单
VNPY源码（八）VNPY的数据流
一、接收行情、并由Tick生成1分钟K线的Demo
# -*- coding: utf-8 -*-
import thostmduserapi as mdapi
class CFtdcMdSpi(mdapi.CThostFtdcMdSpi):
tapi=''
def __init__(self,tapi):
mdapi.CThostFtdcMdSpi.__init__(self)
self.tapi=tapi
def OnFrontConnected(self) -> "void":
#这个是在执行了join函数后执行的。
#它是底层dll文件调用的
#当客户端与交易托管系统建立起通信连接时（还未登录前），该方法被调用。本方法在完成初始化后调用，可以在其中完成用户登录任务。
print("1.当客户端与交易托管系统建立起通信连接时（还未登录前），OnFrontConnected方法被调用")
print ("OnFrontConnected")
loginfield = mdapi.CThostFtdcReqUserLoginField()
loginfield.BrokerID="9999"
loginfield.UserID="1xxxx1"
loginfield.Password="axxxx4"
loginfield.UserProductInfo="python dll"
self.tapi.ReqUserLogin(loginfield,0)
def OnRspUserLogin(self, pRspUserLogin: 'CThostFtdcRspUserLoginField', pRspInfo: 'CThostFtdcRspInfoField', nRequestID: 'int', bIsLast: 'bool') -> "void":
#登录请求响应，当ReqUserLogin后，该方法被调用。
print ("2.登录请求响应，当ReqUserLogin后，OnRspUserLogin方法被调用")
print ("OnRspUserLogin")
print ("SessionID=",pRspUserLogin.SessionID)
print ("ErrorID=",pRspInfo.ErrorID)
print ("ErrorMsg=",pRspInfo.ErrorMsg)
# ret=self.tapi.SubscribeMarketData([b"ru1910",b"rb1909",b"au1911",b"ag1912"],4)
ret=self.tapi.SubscribeMarketData([b"ag1912"],1)
def OnRtnDepthMarketData(self, pDepthMarketData: 'CThostFtdcDepthMarketDataField') -> "void":
#深度行情通知，当SubscribeMarketData订阅行情后，行情通知由此推送。
print("3.深度行情通知，当SubscribeMarketData订阅行情后，行情通知由此推送")
# print ("OnRtnDepthMarketData")
# print ("InstrumentID=",pDepthMarketData.InstrumentID)
# print ("LastPrice=",pDepthMarketData.LastPrice)
print("当前代码为{}，价格为{}".format(pDepthMarketData.InstrumentID,pDepthMarketData.LastPrice))
print ("当前时间=",pDepthMarketData.UpdateTime)
self.chuli(pDepthMarketData)
print(pDepthMarketData)
def OnRspSubMarketData(self, pSpecificInstrument: 'CThostFtdcSpecificInstrumentField', pRspInfo: 'CThostFtdcRspInfoField', nRequestID: 'int', bIsLast: 'bool') -> "void":
#订阅行情应答，调用SubscribeMarketData后，通过此接口返回
print("4.订阅行情应答，调用SubscribeMarketData后，通过此接口返回")
print ("OnRspSubMarketData")
print ("InstrumentID=",pSpecificInstrument.InstrumentID)
print ("ErrorID=",pRspInfo.ErrorID)
print ("ErrorMsg=",pRspInfo.ErrorMsg)
def chuli(self,tickshuju,huancun=[]):
zongfen=tickshuju.UpdateTime[:-2]+"00"
if huancun==[]:
huancun.append(tickshuju.TradingDay)
huancun.append(zongfen)
huancun.append(float(tickshuju.LastPrice))
print(huancun)
print(huancun[1])
print(zongfen)
print(zongfen!=huancun[1])
else:
if zongfen!=huancun[1]:#xieshuju(huancun)
print("hi"*100)
print(huancun)
xieshuju(huancun)
huancun[0]=tickshuju.TradingDay
huancun[1]=tickshuju.UpdateTime
huancun[2]=float(tickshuju.LastPrice)
else:
a=tickshuju.LastPrice
c=max(a,huancun[2])
huancun[2]=c
print("over"*100)
def xieshuju(shujuliebiao):
f2=open("haha.txt","a")
f2.write(str(shujuliebiao))
f2.close()
def main():
#调用thostmduserapi文件中的CThostFtdcMdApi_CreateFtdcMdApi函数，它输出的是CThostFtdcMdApi实例
mduserapi=mdapi.CThostFtdcMdApi_CreateFtdcMdApi()
print("1"*50)
#实例化一个CFtdcMdSpi类，将上面的CThostFtdcMdApi作为参数传入。
#执行CFtdcMdSpi的init方法，其实是调用父类CThostFtdcMdSpi的方法。
#父类CThostFtdcMdSpi的init的方法，如果self.__class__ == CThostFtdcMdSpi，_self = None，否则_self = self
# 将上面的_self传入到_thostmduserapi.new_CThostFtdcMdSpi(_self, ) ，执行从c编译文件中引入的_thostmduserapi里面的CThostFtdcMdSpi_swiginit（）功能
# 将_thostmduserapi.delete_CThostFtdcMdSpi赋值给__swig_destroy__。
mduserspi=CFtdcMdSpi(mduserapi)
print("2"*50)
#注意是第一步输出的mduserapi，执行CThostFtdcMdApi类的RegisterFront函数，这个函数返回的是“void"？
# mduserapi.RegisterFront("tcp://101.230.209.178:53313")
print("3"*50)
'''以下是7*24小时环境'''
mduserapi.RegisterFront("tcp://180.168.146.187:10131")
#传入的参数是 CThostFtdcTraderSpi
mduserapi.RegisterSpi(mduserspi)
print("4"*50)
#执行CThostFtdcMdApi的Init函数（不是init)，返回一个void
mduserapi.Init()
print("5"*50)
#执行CThostFtdcMdApi的join函数，返回一个int。
mduserapi.Join()
print("6"*50)
x =OnRtnDepthMarketData()
dwbtest(x)
if __name__ == '__main__':
main()
效果展示：
二、Emu
今天想使用vnpy的BarGenerator的时候，发现相关代码有一句from enum import Enum。
枚举类型可以看作是一种标签或是一系列常量的集合，通常用于表示某些特定的有限集合，例如星期、月份、状态等。
使用普通类可直接实现枚举：
class color():
YELLOW = 1
RED = 2
GREEN = 3
PINK = 4
# 访问枚举项
print(color.YELLOW) # 1
虽然这样是可以解决问题的，但是并不严谨，也不怎么安全，比如：
1、枚举类中，不应该存在key相同的枚举项（类变量）
2、不允许在类外直接修改枚举项的值
使用枚举
# 导入枚举类
from enum import Enum
# 继承枚举类
class color(Enum):
YELLOW = 1
BEOWN = 1
# 注意BROWN的值和YELLOW的值相同，这是允许的，此时的BROWN相当于YELLOW的别名
RED = 2
GREEN = 3
PINK = 4
可参考：https://www.cnblogs.com/-beyond/p/9777329.html
三、dataclasses
Python 3.7 的新特性。
Dataclasses 是 Python 的类（LCTT 译注：更准确的说，它是一个模块），适用于存储数据对象。
那么它有什么特点呢？它们可以与同一类型的其他对象进行比较。例如：一个数字可以是 greater than（大于）、less than（小于） 或 equal（等于） 另一个数字。这样是不是很方便？
如果你想要将类转换为dataclasses，只需要使用dataclass作装饰器即可。
@dataclass
class BaseData:
"""
Any data object needs a gateway_name as source
and should inherit base data.
"""
gateway_name: str
四、Callable
有一条from typing import Callable。
动态语言写起来非常爽，便是后期的维护是一个坑，python3.5之后，PEP484为python引入了类型注解，让Python能够像静态语言一样支持类型声明。比如：
#一个典型的函数注释例子，为参数加上了类型
def greeting(name: str) -> str:
return 'Hello ' + name
python3.6之后，引入了对变量类型的声明
# 声明primes是列表，其元素的类型为int。并给primes赋值一个空列表。
primes: List[int] = []
# 声明captain的类型是str，注意这里无初始值
captain: str
# 声明stats是类变量，类型是字典（key的类型为str、value的类型为int）。并给stats赋值一个空字典。
class Starship:
stats: ClassVar[Dict[str, int]] = {}
目前在Python标准库中，有一个暂不稳定的库：typing。typing提供了最基本的声明类型，用户可以像引入普通库一样将其中的类型引入文档中使用，同时提供了NewType()函数让用户能够很方便地创建自定义类型。此模块的完整规范记录在PEP484中。
回调函数可以使用类似Callable[[Arg1Type, Arg2Type],ReturnType]的类型注释：
from typing import Callable
def feeder(get_next_item: Callable[[], str]) -> None:
# Body
def async_query(on_success: Callable[[int], None],
on_error: Callable[[int, Exception], None]) -> None:
# Body
可参考：https://www.cnblogs.com/lynsyklate/p/7594082.html
五、python新特性：
https://www.cnblogs.com/animalize/p/5633215.html
参考资料：
python ctp API: https://github.com/nicai0609/Python-CTPAPI
我也想只用vnpy的接口，不用他的回测框架。请问你用的vnpy哪个版本做的这个，我用2.0版本没有你写的这些模块啊。