Home >  > VNPY源码(一)CTP封装及K线合成

VNPY源码(一)CTP封装及K线合成

1

蜗牛博客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

1CTP客户端开发指南

暧昧帖

本文暂无标签
Comment (1)
Trackback (0)
  1. Andy Google Chrome 88.0.4324.190Windows 7 沙发 2021/02/25 16:25

    我也想只用vnpy的接口,不用他的回测框架。请问你用的vnpy哪个版本做的这个,我用2.0版本没有你写的这些模块啊。

  • 还没有Trackback

发表评论

*

*