#!/usr/bin/python
|
# -*- coding: UTF-8 -*-
|
import json
|
import logging
|
import time
|
|
import socket_util
|
import tool
|
import xmdapi
|
from client_network import SendResponseSkManager
|
|
level1_data_dict = {
|
|
}
|
|
|
def __send_response(sk, msg):
|
msg = socket_util.load_header(msg)
|
sk.sendall(msg)
|
result, header_str = socket_util.recv_data(sk)
|
if result:
|
result_json = json.loads(result)
|
if result_json.get("code") == 0:
|
return True
|
return False
|
|
|
def get_level1_codes():
|
type_ = "get_level1_codes"
|
fdata = json.dumps(
|
{"type": type_, "data": {}})
|
msg = fdata.encode("utf-8")
|
# 发送消息
|
for i in range(3):
|
try:
|
sk = SendResponseSkManager.create_send_response_sk()
|
msg = socket_util.load_header(msg)
|
sk.sendall(msg)
|
result, header_str = socket_util.recv_data(sk)
|
# 读取代码
|
result_json = json.loads(result)
|
if result_json["code"] == 0:
|
codes = result_json["data"]
|
codes_sh = []
|
codes_sz = []
|
for code in codes:
|
if code.find("00") == 0:
|
codes_sz.append(code.encode("utf-8"))
|
else:
|
codes_sh.append(code.encode("utf-8"))
|
return codes_sh, codes_sz
|
except ConnectionResetError:
|
SendResponseSkManager.del_send_response_sk(type_)
|
except BrokenPipeError:
|
SendResponseSkManager.del_send_response_sk(type_)
|
return None, None
|
|
|
class MdSpi(xmdapi.CTORATstpXMdSpi):
|
def __init__(self, api):
|
self.codes_sh, self.codes_sz = get_level1_codes()
|
xmdapi.CTORATstpXMdSpi.__init__(self)
|
self.__api = api
|
|
def OnFrontConnected(self):
|
print("OnFrontConnected")
|
|
# 请求登录，目前未校验登录用户，请求域置空即可
|
login_req = xmdapi.CTORATstpReqUserLoginField()
|
self.__api.ReqUserLogin(login_req, 1)
|
|
def OnRspUserLogin(self, pRspUserLoginField, pRspInfoField, nRequestID):
|
if pRspInfoField.ErrorID == 0:
|
print('Login success! [%d]' % nRequestID)
|
|
'''
|
订阅行情
|
当sub_arr中只有一个"00000000"的合约且ExchangeID填TORA_TSTP_EXD_SSE或TORA_TSTP_EXD_SZSE时，订阅单市场所有合约行情
|
当sub_arr中只有一个"00000000"的合约且ExchangeID填TORA_TSTP_EXD_COMM时，订阅全市场所有合约行情
|
其它情况，订阅sub_arr集合中的合约行情
|
'''
|
|
print(f"订阅数量：sh-{len(self.codes_sh)} sz-{len(self.codes_sz)}")
|
if self.codes_sh:
|
ret = self.__api.SubscribeMarketData(self.codes_sh, xmdapi.TORA_TSTP_EXD_SSE)
|
if ret != 0:
|
print('SubscribeMarketData fail, ret[%d]' % ret)
|
else:
|
print('SubscribeMarketData success, ret[%d]' % ret)
|
|
if self.codes_sz:
|
ret = self.__api.SubscribeMarketData(self.codes_sz, xmdapi.TORA_TSTP_EXD_SZSE)
|
if ret != 0:
|
print('SubscribeMarketData fail, ret[%d]' % ret)
|
else:
|
print('SubscribeMarketData success, ret[%d]' % ret)
|
|
# sub_arr = [b'600004']
|
# ret = self.__api.UnSubscribeMarketData(sub_arr, xmdapi.TORA_TSTP_EXD_SSE)
|
# if ret != 0:
|
# print('UnSubscribeMarketData fail, ret[%d]' % ret)
|
# else:
|
# print('SubscribeMarketData success, ret[%d]' % ret)
|
|
|
else:
|
print('Login fail!!! [%d] [%d] [%s]'
|
% (nRequestID, pRspInfoField.ErrorID, pRspInfoField.ErrorMsg))
|
|
def OnRspSubMarketData(self, pSpecificSecurityField, pRspInfoField):
|
if pRspInfoField.ErrorID == 0:
|
print('OnRspSubMarketData: OK!')
|
else:
|
print('OnRspSubMarketData: Error! [%d] [%s]'
|
% (pRspInfoField.ErrorID, pRspInfoField.ErrorMsg))
|
|
def OnRspUnSubMarketData(self, pSpecificSecurityField, pRspInfoField):
|
if pRspInfoField.ErrorID == 0:
|
print('OnRspUnSubMarketData: OK!')
|
else:
|
print('OnRspUnSubMarketData: Error! [%d] [%s]'
|
% (pRspInfoField.ErrorID, pRspInfoField.ErrorMsg))
|
|
def OnRtnMarketData(self, pMarketDataField):
|
if pMarketDataField.SecurityName.find("S") == 0:
|
return
|
if pMarketDataField.SecurityName.find("ST") >= 0:
|
return
|
close_price = round(pMarketDataField.UpperLimitPrice / 1.1, 2)
|
rate = round((pMarketDataField.LastPrice - close_price) * 100 / close_price, 2)
|
# print(pMarketDataField.SecurityID, pMarketDataField.SecurityName, rate, pMarketDataField.Volume)
|
|
level1_data_dict[pMarketDataField.SecurityID] = (
|
pMarketDataField.SecurityID, pMarketDataField.LastPrice, rate, pMarketDataField.Volume, time.time())
|
# print(
|
# "SecurityID[%s] SecurityName[%s] LastPrice[%.2f] Volume[%d] Turnover[%.2f] BidPrice1[%.2f] BidVolume1[%d] AskPrice1[%.2f] AskVolume1[%d] UpperLimitPrice[%.2f] LowerLimitPrice[%.2f]"
|
# % (pMarketDataField.SecurityID, pMarketDataField.SecurityName, pMarketDataField.LastPrice,
|
# pMarketDataField.Volume,
|
# pMarketDataField.Turnover, pMarketDataField.BidPrice1, pMarketDataField.BidVolume1,
|
# pMarketDataField.AskPrice1,
|
# pMarketDataField.AskVolume1, pMarketDataField.UpperLimitPrice, pMarketDataField.LowerLimitPrice))
|
|
|
def __upload_codes_info(datas):
|
if not tool.is_trade_time():
|
return
|
# 上传数据
|
type_ = "set_target_codes"
|
fdata = json.dumps(
|
{"type": type_, "data": {"data": datas}})
|
msg = fdata.encode("utf-8")
|
|
# 发送消息
|
for i in range(3):
|
try:
|
sk = SendResponseSkManager.get_send_response_sk(type_)
|
if __send_response(sk, msg):
|
break
|
except ConnectionResetError:
|
SendResponseSkManager.del_send_response_sk(type_)
|
|
except BrokenPipeError:
|
SendResponseSkManager.del_send_response_sk(type_)
|
|
|
def run():
|
# 打印接口版本号
|
print(xmdapi.CTORATstpXMdApi_GetApiVersion())
|
|
# 创建接口对象
|
api = xmdapi.CTORATstpXMdApi_CreateTstpXMdApi(xmdapi.TORA_TSTP_MST_MCAST)
|
|
# 创建回调对象
|
spi = MdSpi(api)
|
|
# 注册回调接口
|
api.RegisterSpi(spi)
|
|
# 注册单个行情前置服务地址
|
# api.RegisterFront("tcp://224.224.1.19:7880")
|
# 注册多个行情前置服务地址，用逗号隔开
|
# api.RegisterFront("tcp://10.0.1.101:6402,tcp://10.0.1.101:16402")
|
# 注册名字服务器地址，支持多服务地址逗号隔开
|
# api.RegisterNameServer('tcp://224.224.3.19:7888')
|
# api.RegisterNameServer('tcp://10.0.1.101:52370,tcp://10.0.1.101:62370')
|
api.RegisterMulticast("udp://224.224.1.19:7880", None, "")
|
|
# 启动接口
|
api.Init()
|
|
# 等待程序结束
|
while True:
|
print("数量", len(level1_data_dict))
|
try:
|
if len(level1_data_dict) < 1:
|
continue
|
# 根据涨幅排序
|
list_ = [level1_data_dict[k] for k in level1_data_dict]
|
flist = []
|
for d in list_:
|
if d[2] >= 5:
|
# 涨幅小于5%的需要删除
|
flist.append(d)
|
flist.sort(key=lambda x: x[2], reverse=True)
|
datas = flist[:100]
|
codes = [x[0] for x in datas]
|
print("代码数量:", len(datas),codes)
|
__upload_codes_info(datas)
|
except Exception as e:
|
logging.exception(e)
|
finally:
|
time.sleep(3)
|
# 释放接口对象
|
api.Release()
|
|
|
if __name__ == "__main__":
|
run()
|
