"""
|
交易API
|
"""
|
import json
|
import logging
|
import random
|
import time
|
|
from socket_manager import ClientSocketManager
|
from utils import socket_util
|
|
TRADE_DIRECTION_BUY = 1
|
TRADE_DIRECTION_SELL = 2
|
|
# 数据操作
|
OPERRATE_SET = 1 # 设置
|
OPERRATE_DELETE = 2 # 删除
|
OPERRATE_GET = 3 # 获取
|
|
# 代码名单类型
|
CODE_LIST_WHITE = "white"
|
CODE_LIST_BLACK = "black"
|
CODE_LIST_WANT = "want"
|
CODE_LIST_PAUSE_BUY = "pause_buy"
|
|
# 类型
|
API_TYPE_TRADE = "trade" # 交易
|
API_TYPE_TRADE_STATE = "trade_state" # 交易状态
|
API_TYPE_TRADE_MODE = "trade_mode" # 交易模式
|
API_TYPE_CODE_LIST = "code_list" # 代码名单
|
API_TYPE_EXPORT_L2 = "export_l2" # 导出L2数据
|
API_TYPE_INIT = "init" # 初始化
|
API_TYPE_REFRESH_TRADE_DATA = "refresh_trade_data" # 交易数据刷新
|
API_TYPE_CODE_ATRRIBUTE = "code_attribute" # 代码属性
|
API_TYPE_CODE_TRADE_STATE = "code_trade_state" # 代码交易状态
|
API_TYPE_GET_ENV = "get_env" # 获取环境信息
|
# 超时时间2s
|
TIMEOUT = 5.0
|
|
# 等待响应的request_id
|
__request_response_dict = {}
|
|
|
def __get_request_id(type):
|
return f"r_{type}_{round(time.time() * 10000)}_{random.randint(0, 100000)}"
|
|
|
# 网络请求
|
def __request(client_type, data):
|
client = ClientSocketManager.acquire_client(client_type)
|
if not client:
|
raise Exception("无可用的交易client")
|
request_id = __get_request_id(data["type"])
|
try:
|
root_data = {"type": "cmd",
|
"data": data,
|
"request_id": request_id}
|
root_data = socket_util.encryp_client_params_sign(root_data)
|
# print("请求前对象", root_data)
|
# 添加请求头
|
client[1].sendall(socket_util.load_header(json.dumps(root_data).encode(encoding='utf-8')))
|
result = client[1].recv(1024)
|
# print("请求发送成功", result.decode(encoding='utf-8'))
|
|
except BrokenPipeError as e:
|
ClientSocketManager.del_client(client[0])
|
raise e
|
except Exception as e:
|
logging.exception(e)
|
raise e
|
return request_id, client
|
|
|
def __read_response(client, request_id, blocking, timeout=TIMEOUT):
|
if blocking:
|
start_time = time.time()
|
try:
|
while True:
|
time.sleep(0.005)
|
if request_id in __request_response_dict:
|
# 获取到了响应内容
|
result = __request_response_dict.pop(request_id)
|
return result
|
if time.time() - start_time > timeout:
|
# 读取内容超时才会释放
|
ClientSocketManager.release_client(client[0])
|
raise Exception(f"读取内容超时: request_id={request_id}")
|
finally:
|
pass
|
|
return None
|
|
|
def set_response(client_id, request_id, response):
|
if client_id and request_id:
|
# 主动触发
|
__request_response_dict[request_id] = response
|
ClientSocketManager.release_client(client_id)
|
else:
|
# 被动触发
|
pass
|
|
|
def parseResponse(data_str):
|
if not data_str:
|
raise Exception("反馈内容为空")
|
res = data_str
|
if type(res) == str:
|
res = json.loads(data_str)
|
res = res['data']
|
if res['code'] != 0:
|
raise Exception(res['msg'])
|
return res['data']
|
|
|
# 下单委托
|
# direction 1-买 2-卖
|
# code:代码
|
# volume：交易量
|
# price：价格（如果是卖时不传价格就按照5挡价卖）
|
# blocking是否阻塞进程
|
def trade_order(direction, code, volume, price, price_type=2, blocking=True):
|
print("客户端", ClientSocketManager.socket_client_dict)
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_TRADE, "trade_type": 1,
|
"direction": direction,
|
"code": code,
|
"volume": volume,
|
"price_type": price_type,
|
"price": price, "sinfo": f"order_{code}_{round(time.time() * 1000)}"})
|
|
return __read_response(client, request_id, blocking)
|
|
|
# 取消下单
|
def trade_cancel_order(direction, code, accountID, orderSysID, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_TRADE, "trade_type": 2,
|
"direction": direction,
|
"code": code,
|
"accountID": accountID,
|
"orderSysID": orderSysID,
|
"sinfo": f"cancel_order_{code}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# 设置交易状态
|
def set_trade_state(state, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_TRADE_STATE, "operate": OPERRATE_SET,
|
"state": state,
|
"sinfo": f"cb_{API_TYPE_TRADE_STATE}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# 获取交易状态
|
def get_trade_state(blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_TRADE_STATE, "operate": OPERRATE_GET,
|
"sinfo": f"cb_{API_TYPE_TRADE_STATE}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# 设置交易模式
|
def set_trade_mode(mode, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_TRADE_MODE, "operate": OPERRATE_SET, "mode": mode,
|
"sinfo": f"cb_{API_TYPE_TRADE_MODE}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# 获取交易模式
|
def get_trade_mode(blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_TRADE_MODE, "operate": OPERRATE_GET,
|
"sinfo": f"cb_{API_TYPE_TRADE_MODE}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# -----代码名单操作----
|
def add_code_list(code, code_list_type, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_CODE_LIST, "code_list_type": code_list_type, "code": code,
|
"operate": OPERRATE_SET,
|
"sinfo": f"cb_{API_TYPE_CODE_LIST}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
def remove_code_list(code, code_list_type, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_CODE_LIST, "code_list_type": code_list_type, "code": code,
|
"operate": OPERRATE_DELETE,
|
"sinfo": f"cb_{API_TYPE_CODE_LIST}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
def get_code_list(code_list_type, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_CODE_LIST, "code_list_type": code_list_type,
|
"operate": OPERRATE_GET,
|
"sinfo": f"cb_{API_TYPE_CODE_LIST}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# -----导出L2数据----
|
def export_l2_data(code, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_EXPORT_L2, "code": code,
|
"sinfo": f"cb_{API_TYPE_CODE_LIST}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# -----每日初始化----
|
def everyday_init(blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_INIT,
|
"sinfo": f"cb_{API_TYPE_CODE_LIST}_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# 刷新交易数据
|
def refresh_trade_data(type, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_REFRESH_TRADE_DATA, "ctype": type,
|
"sinfo": f"update_trade_data_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# 获取代码属性
|
def get_code_attribute(code, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_CODE_ATRRIBUTE, "code": code,
|
"sinfo": f"update_trade_data_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# 获取代码交易状态
|
def get_code_trade_state(code, blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_CODE_TRADE_STATE, "code": code,
|
"sinfo": f"update_trade_data_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
# 获取环境信息
|
def get_env_info(blocking=True):
|
request_id, client = __request(ClientSocketManager.CLIENT_TYPE_TRADE,
|
{"type": API_TYPE_GET_ENV,
|
"sinfo": f"update_trade_data_{round(time.time() * 1000)}"})
|
return __read_response(client, request_id, blocking)
|
|
|
if __name__ == "__main__":
|
d = {"id": "123123"}
|
print(d.pop("id"))
|
