Table of Contents
write in front
Download the source code and unzip it
Create python project
environment
process
Compile vnpy_ctp source code
Verify availability
Write in front
Visual Studio must be installed in the window system, and C++ compilation is required when installing the source code later.
Download the source code and unzip
GitHub – vnpy/vnpy_ctp: CTP transaction interface for VeighNa framework
Download zip package
Unzip
To be executable in python, there must be a .pyd file. There is no .pyd file in the decompressed folder.
Create python project
Create a new python project and execute it in a new virtual environment.
environment
Operating system: window10 64 bit
Development tools: pycharm
python version: python3.8.10
process
Compile vnpy_ctp source code
Open the terminal panel below the project
cd to the folder where setup.py is located after decompression
Execute python setup.py build
In about 2 to 3 minutes, the compilation is completed, and there will be an extra build folder under the folder where setup.py is located.
Find the operating system and python version under build The corresponding folder, taking this article as an example, the operating system is 64-bit, then the folder name will contain amd64, python version 3.8, and the folder name will contain 3.8, so the folder name of this article is lib.win- amd64-3.8
What we need is the api folder under the build>lib.win-amd64-3.8>vnpy_ctp folder. We copy the api folder to the project directory.
Verify Availability
Create a class called by Md and a class called by Td
import datetime,sys,os,time,pytz
from api import (
TdApi,
MdApi
)
class CtpMdApi(MdApi):
def __init__(self)->None:
super().__init__()
self.reqid:int = 0
self.connect_status: bool = False
self.login_status: bool = False
self.subscribed: set = set()
self.userid: str = ""
self.password: str = ""
self.brokerid: str = ""
self.current_date: str = datetime.date.today().strftime('%Y%m%d')
pass
def connect(self, address: str, userid: str, password: str, brokerid: str)->None:
self.userid = userid
self.password = password
self.brokerid = brokerid
if not self.connect_status:
con_body_path = CTP_MD_CON_DIR
con_body_path = con_body_path.replace('/','')
self.createFtdcMdApi(con_body_path)
self.registerFront(address)
self.init()
self.connect_status = True
pass
pass
def login(self)->None:
ctp_req: dict = {
'UserID':self.userid,
'Password':self.password,
'BrokerID':self.brokerid
}
self.reqid + = 1
self.reqUserLogin(ctp_req,self.reqid)
pass
def subscribe(self,req:dict):
if self.login_status:
self.subscribeMarketData(req['symbol'])
self.subscribed.add(req['symbol'])
pass
def close(self)->None:
if self.connect_status:
self.exit()
pass
def update_date(self)->None:
self.current_date = datetime.date.today().strftime('%Y%m%d')
pass
def onFrontConnected(self)->None:
self.login()
pass
def onFrontDisconnected(self,reason:int)->None:
self.login_status = False
pass
def onRspUserLogin(self,data:dict,error:dict,reqid:int,last:bool)->None:
if not error['ErrorID']:
self.login_status = True
for symbol in self.subscribed:
self.subscribeMarketData(symbol)
pass
else:
print(f"Login to the quotation server failed. {error['ErrorID']}.{error['ErrorMsg']}")
pass
def onRspError(self, error: dict, reqid: int, last: bool)->None:
print('The market interface reported an error.', error['ErrorID'], error['ErrorMsg'])
pass
def onRspSubMarketData(self, data: dict, error: dict, reqid: int, last: bool):
if not error or not error['ErrorID']:
return
print('Quote subscription failed.',error['ErrorID'],error['ErrorMsg'])
pass
def onRtnDepthMarketData(self,data:dict)->None:
if not data['UpdateTime']:
return
print('tick returns',data['InstrumentID'],data['LastPrice'])
pass
pass
class CtpTdApi(TdApi):
def __init__(self)->None:
super().__init__()
self.reqid: int = 0
self.order_ref: int = 0
self.connect_status: bool = False
self.login_status: bool = False
self.auth_status: bool = False
self.login_failed: bool = False
self.auth_failed: bool = False
self.contract_inited: bool = False
self.userid: str = ""
self.password: str = ""
self.brokerid: str = ""
self.auth_code: str = ""
self.appid: str = ""
self.frontid: int = 0
self.sessionid: int = 0
pass
def connect(self,address:str,userid:str,password:str,brokerid:str,auth_code:str,appid:str)->None:
self.userid = userid
self.password = password
self.brokerid = brokerid
self.auth_code = auth_code
self.appid = appid
if not self.connect_status:
con_body_path = CTP_TD_CON_DIR + self.userid + os.path.sep
if not os.path.exists(con_body_path):
os.mkdir(con_body_path)
con_body_path = con_body_path.replace('/','')
self.createFtdcTraderApi(con_body_path)
self.subscribePrivateTopic(0)
self.subscribePublicTopic(0)
self.registerFront(address)
self.init()
self.connect_status = True
pass
else:
self.authenticate()
pass
def authenticate(self)->None:
if self.auth_failed:
return
ctp_req: dict = {
"UserID": self.userid,
"BrokerID": self.brokerid,
"AuthCode": self.auth_code,
"AppID": self.appid
}
self.reqid + = 1
self.reqAuthenticate(ctp_req, self.reqid)
pass
def login(self)->None:
if self.login_failed:
return
ctp_req: dict = {
"UserID": self.userid,
"Password": self.password,
"BrokerID": self.brokerid,
"AppID": self.appid
}
self.reqid + = 1
self.reqUserLogin(ctp_req, self.reqid)
pass
def close(self)->None:
if self.connect_status:
self.exit()
pass
def onFrontConnected(self)->None:
if self.auth_code:
self.authenticate()
else:
self.login()
pass
def onFrontDisconnected(self,reason:int)->None:
self.login_status = False
pass
def onRspAuthenticate(self, data: dict, error: dict, reqid: int, last: bool)->None:
if not error['ErrorID']:
self.auth_status = True
self.login()
else:
self.auth_failed = True
print('Trading server verification failed.',error['ErrorID'],error['ErrorMsg'])
pass
def onRspUserLogin(self,data: dict, error: dict, reqid: int, last: bool)->None:
if not error["ErrorID"]:
self.frontid = data["FrontID"]
self.sessionid = data["SessionID"]
self.login_status = True
# Automatically confirm the settlement document
ctp_req: dict = {
"BrokerID": self.brokerid,
"InvestorID": self.userid
}
self.reqid + = 1
self.reqSettlementInfoConfirm(ctp_req, self.reqid)
else:
self.login_failed = True
print("Trading server login failed", error['ErrorID'], error['ErrorMsg'])
pass
def onRspSettlementInfoConfirm(self,data: dict, error: dict, reqid: int, last: bool)->None:
while True:
self.reqid + = 1
n: int = self.reqQryInstrument({}, self.reqid)
if not n:
break
else:
time.sleep(1)
pass
def onRspQryInstrument(self, data: dict, error: dict, reqid: int, last: bool) -> None:
print(data['ProductClass'],data['InstrumentID'],data['ProductID'],reqid,last)
if last:
self.contract_initiated = True
print('Contract information query completed')
pass
pass
Execute code
if __name__ == '__main__':
investorid = ""
brokerid=""
password= ""
appid= ""
auth_code= "0000000000000000"
md_ip= "180.168.146.187:10211"
trader_ip= "180.168.146.187:10201"
temp_api = CtpTdApi()
address = f"tcp://{trader_ip}"
temp_api.connect(address,investorid,password,brokerid,auth_code,appid)
import keyboard
keyboard.wait('esc')
sys.exit()
pass
Can log in to the td server and query the contract normally, available