"""
|
华鑫交易结果处理器
|
"""
|
import copy
|
import json
|
import time
|
|
from log_module import async_log_util
|
from log_module.log import logger_trade, hx_logger_trade_debug
|
from trade.huaxin.huaxin_trade_record_manager import TradeOrderIdManager
|
from utils import huaxin_util
|
import concurrent.futures
|
|
|
class HuaxinOrderEntity:
|
def __init__(self, code, orderStatus, orderRef, accountID, orderSysID, direction=None, insertDate=None,
|
insertTime=None,
|
acceptTime=None):
|
self.code = code
|
self.orderStatus = orderStatus
|
self.orderRef = orderRef
|
self.accountID = accountID
|
self.orderSysID = orderSysID
|
self.insertTime = insertTime
|
self.insertDate = insertDate
|
self.acceptTime = acceptTime
|
self.direction = direction
|
|
|
class CancelOrderManager:
|
__canceling_order_dict = {}
|
__recancel_order_count = {}
|
|
__instance = None
|
|
def __new__(cls, *args, **kwargs):
|
if not cls.__instance:
|
cls.__instance = super(CancelOrderManager, cls).__new__(cls, *args, **kwargs)
|
return cls.__instance
|
|
# 开始撤单
|
def start_cancel(self, code, order_ref, order_sys_id):
|
if code not in self.__canceling_order_dict:
|
self.__canceling_order_dict[code] = set()
|
self.__canceling_order_dict[code].add(json.dumps((order_ref, order_sys_id, int(time.time() * 1000))))
|
|
def __cancel_finish(self, code, order_ref, order_sys_id):
|
if code not in self.__canceling_order_dict:
|
return
|
if not self.__canceling_order_dict[code]:
|
return
|
infos = copy.deepcopy(self.__canceling_order_dict[code])
|
for info in infos:
|
_info = json.loads(info)
|
if _info[0] == order_ref or _info[1] == order_sys_id:
|
# 匹配到目标数据
|
self.__canceling_order_dict[code].discard(info)
|
|
def __add_recancel_count(self, code, order_ref, order_sys_id):
|
key = f"{code}_{order_ref}_{order_sys_id}"
|
if key not in self.__recancel_order_count:
|
self.__recancel_order_count[key] = 0
|
self.__recancel_order_count[key] += 1
|
|
def __can_recancel(self, code, order_ref, order_sys_id):
|
key = f"{code}_{order_ref}_{order_sys_id}"
|
if key not in self.__recancel_order_count:
|
return True
|
return self.__recancel_order_count[key] < 2
|
|
# 撤单成功
|
def cancel_success(self, code, order_ref, order_sys_id):
|
self.__cancel_finish(code, order_ref, order_sys_id)
|
|
# 买入成功
|
def buy_success(self, code, order_ref, order_sys_id):
|
self.__cancel_finish(code, order_ref, order_sys_id)
|
|
# 传入重新下单
|
def run(self, re_cancel_method):
|
while True:
|
try:
|
if self.__canceling_order_dict:
|
for code in self.__canceling_order_dict:
|
infos = self.__canceling_order_dict[code]
|
infos = copy.deepcopy(infos)
|
for info in infos:
|
_info = json.loads(info)
|
timestamp = _info[2]
|
# 查询是否还能重新撤单
|
if not self.__can_recancel(code, _info[0], _info[1]):
|
self.__canceling_order_dict[code].discard(info)
|
continue
|
|
if time.time() * 1000 - timestamp > 100:
|
async_log_util.info(logger_trade, f"{code}触发重新撤单：{info}")
|
# 100ms后才进行
|
self.__add_recancel_count(code, _info[0], _info[1])
|
re_cancel_method(1, code, _info[1], orderRef=_info[0], recancel=True)
|
time.sleep(0.05)
|
except:
|
pass
|
|
|
class TradeResultProcessor:
|
__TradeOrderIdManager = TradeOrderIdManager()
|
__processed_keys = set()
|
__thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=50)
|
# 订单索引号字典
|
order_ref_dict = {}
|
|
# 处理买单
|
@classmethod
|
def process_buy_order(cls, order: HuaxinOrderEntity):
|
"""
|
处理买单
|
@param order:
|
@return: 是否需要撤单
|
"""
|
|
# 处理下单成功
|
def process_order_success(order_: HuaxinOrderEntity, delay_s=0.0):
|
if delay_s > 0:
|
time.sleep(delay_s)
|
# TODO 处理下单成功
|
|
# 只处理买入单
|
if order.direction != str(huaxin_util.TORA_TSTP_D_Buy):
|
return False
|
|
# 同一订单号只有状态变化了才需要处理
|
key = f"{order.insertDate}_{order.code}_{order.orderSysID}_{order.orderStatus}"
|
if key in cls.__processed_keys:
|
return False
|
try:
|
async_log_util.info(hx_logger_trade_debug, f"处理华鑫订单：{key}")
|
cls.__processed_keys.add(key)
|
if huaxin_util.is_can_cancel(order.orderStatus):
|
# 设置下单成功
|
process_order_success(order)
|
elif huaxin_util.is_canceled(order.orderStatus) or huaxin_util.is_deal(
|
order.orderStatus):
|
# 已经撤单/已经成交，需要处理临时保存的系统订单号
|
cls.__TradeOrderIdManager.remove_order_id(order.code,
|
order.accountID,
|
order.orderSysID)
|
if huaxin_util.is_deal(order.orderStatus):
|
# TODO 处理成交
|
pass
|
elif huaxin_util.is_canceled(order.orderStatus):
|
CancelOrderManager().cancel_success(order.code, order.orderRef, order.orderSysID)
|
except Exception as e:
|
async_log_util.exception(hx_logger_trade_debug, e)
|
return False
|
|
@classmethod
|
def get_huaxin_order_by_order_ref(cls, order_ref) -> HuaxinOrderEntity:
|
return cls.order_ref_dict.get(order_ref)
|
|
@classmethod
|
def get_huaxin_sell_order_by_code(cls, code):
|
results = []
|
for k in cls.order_ref_dict:
|
entity = cls.order_ref_dict[k]
|
if entity.code == code and entity.direction == huaxin_util.TORA_TSTP_D_Sell:
|
results.append(entity)
|
return results
|
|
@classmethod
|
def order_success(cls, order: HuaxinOrderEntity):
|
async_log_util.info(hx_logger_trade_debug, f"处理华鑫订单下单成功：{order.code}, {order.orderRef}, {order.orderSysID}")
|
# 加入系统订单号
|
cls.__TradeOrderIdManager.add_order_id(order.code, order.accountID, order.orderSysID)
|
# 删除临时订单号
|
cls.__TradeOrderIdManager.remove_order_ref(order.code, order.orderRef)
|
return None
|
|
@classmethod
|
def cancel_order_success(cls, code, accountId, orderSysID):
|
cls.__TradeOrderIdManager.remove_order_id(code, accountId, orderSysID)
|
|
|
if __name__ == "__main__":
|
CancelOrderManager().start_cancel("000333", 1, "123123")
|
# CancelOrderManager().cancel_success("000333", 1, "123123")
|
# CancelOrderManager().buy_success("000333", 1, "123123")
|
