"""
|
下单信号管理
|
"""
|
import time
|
|
from l2 import l2_log
|
from l2.huaxin import l2_huaxin_util
|
from log_module.log import logger_l2_trade_buy, logger_debug
|
from utils import tool
|
|
|
class L2TradeSingleDataProcessor:
|
"""
|
L2逐笔成交数据信号管理
|
"""
|
__latest_sell_data = {}
|
|
__latest_limit_up_sell_list_dict = {}
|
|
__latest_limit_up_sell_order_no_set_dict = {}
|
|
# 主动卖订单号集合
|
__active_sell_order_no_set_dict = {}
|
|
@classmethod
|
def add_l2_delegate_limit_up_sell(cls, code, data):
|
"""
|
添加涨停卖单数据，当前不是涨停成交时才加
|
@param code:
|
@param data:
|
@return:
|
"""
|
if code not in cls.__latest_limit_up_sell_list_dict:
|
cls.__latest_limit_up_sell_list_dict[code] = []
|
cls.__latest_limit_up_sell_list_dict[code].append(data)
|
if code not in cls.__latest_limit_up_sell_order_no_set_dict:
|
cls.__latest_limit_up_sell_order_no_set_dict[code] = set()
|
cls.__latest_limit_up_sell_order_no_set_dict[code].add(data['val']['orderNo'])
|
# 只保留前20的数据
|
if len(cls.__latest_limit_up_sell_list_dict[code]) > 20:
|
cls.__latest_limit_up_sell_list_dict[code] = cls.__latest_limit_up_sell_list_dict[code][-20:]
|
# 删除之前的map
|
for d in cls.__latest_limit_up_sell_list_dict[code][0:-20]:
|
cls.__latest_limit_up_sell_order_no_set_dict[code].discard(d["val"]["orderNo"])
|
|
@classmethod
|
def add_l2_delegate_limit_up_sell_cancel(cls, code, order_no):
|
"""
|
涨停卖撤
|
@param code:
|
@param order_no:
|
@return:
|
"""
|
if code not in cls.__latest_limit_up_sell_order_no_set_dict:
|
return
|
order_no_set = cls.__latest_limit_up_sell_order_no_set_dict[code]
|
if order_no not in order_no_set:
|
return
|
cls.__latest_limit_up_sell_order_no_set_dict[code].discard(order_no)
|
|
for i in range(0, len(cls.__latest_limit_up_sell_list_dict[code])):
|
if cls.__latest_limit_up_sell_list_dict[code][i]['val']['orderNo'] == order_no:
|
cls.__latest_limit_up_sell_list_dict[code].pop(i)
|
break
|
|
@classmethod
|
def get_latest_limit_up_sell_order_count(cls, code):
|
sell_list = cls.__latest_limit_up_sell_list_dict.get(code)
|
if not sell_list:
|
return 0
|
return len(sell_list)
|
|
@classmethod
|
def process_passive_limit_up_sell_data(cls, code, datas, limit_up_price):
|
"""
|
添加涨停被动卖成交数据
|
@param data: 数据格式：(data['SecurityID'], data['TradePrice'], data['TradeVolume'],
|
# data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
|
# data['SellNo'], data['ExecType'])
|
@return:
|
"""
|
try:
|
start_time = time.time()
|
sell_list = cls.__latest_limit_up_sell_list_dict.get(code)
|
if not sell_list:
|
return
|
sell_info = sell_list[-1]
|
for data in datas:
|
if data[6] < data[7] or data[1] != limit_up_price:
|
# 排除主动卖/非涨停卖
|
continue
|
sell_no = data[7]
|
if sell_no != sell_info['val']['orderNo']:
|
continue
|
# 需要判断当前单是否已经成交完成
|
if code not in cls.__latest_sell_data:
|
cls.__latest_sell_data[code] = [sell_no, data[2]]
|
else:
|
if cls.__latest_sell_data[code][0] == sell_no:
|
cls.__latest_sell_data[code][1] += data[2]
|
else:
|
cls.__latest_sell_data[code] = [sell_no, data[2]]
|
sell_info_num = sell_info['val']['num']
|
deal_num = cls.__latest_sell_data[code][1] // 100
|
if sell_info_num == deal_num:
|
use_time = round((time.time() - start_time) * 1000, 3)
|
l2_log.info(code, logger_l2_trade_buy,
|
f"找到最近的被动涨停卖单数据：{sell_info['val']['orderNo']}, 计算耗时：{use_time}ms, 可以触发下单")
|
# 成交完成
|
L2TradeSingleDataManager.set_latest_sell_data(code, data)
|
l2_log.info(code, logger_l2_trade_buy, "被动卖数据处理完毕")
|
break
|
# l2_log.info(code, logger_l2_trade_buy, f"找到最近的被动涨停卖单数据：{data['val']['orderNo']}, 可以触发下单")
|
except Exception as e:
|
logger_debug.exception(e)
|
|
@classmethod
|
def add_active_limit_up_sell_data(cls, data):
|
"""
|
添加主动卖数据
|
@param data:
|
@return:
|
"""
|
try:
|
code = data[0]
|
sell_no = data[7]
|
if code not in cls.__active_sell_order_no_set_dict:
|
cls.__active_sell_order_no_set_dict[code] = set()
|
cls.__active_sell_order_no_set_dict[code].add(sell_no)
|
|
except Exception as e:
|
logger_debug.exception(e)
|
|
|
class L2TradeSingleCallback:
|
"""
|
交易信号回调
|
"""
|
|
def OnTradeSingle(self, code, _type, data):
|
"""
|
交易数据信号回调
|
@param code:
|
@param _type: 类型：0-TYPE_PASSIVE 1-TYPE_ACTIVE
|
@param data: (逐笔成交数据,生效时间)
|
@return:
|
"""
|
|
|
class L2TradeSingleDataManager:
|
__callback = None
|
|
TYPE_PASSIVE = 0
|
TYPE_ACTIVE = 1
|
|
"""
|
买入信号管理
|
"""
|
# 最近的涨停卖被动成交数据
|
__latest_sell_data_dict = {} # 格式为：{code:(逐笔成交数据,生效时间（带ms）)}
|
|
# 由被动向主动卖成交转变的数据
|
__latest_sell_active_deal_data_dict = {} # 格式为：{code:(逐笔成交数据,生效时间（带ms）)}
|
|
@classmethod
|
def set_callback(cls, callback: L2TradeSingleCallback):
|
cls.__callback = callback
|
|
@classmethod
|
def set_latest_sell_data(cls, code, data):
|
"""
|
设置最近成交的涨停卖被动成交数据
|
@param code: 代码
|
@param data: L2逐笔成交数据 数据格式：(data['SecurityID'], data['TradePrice'], data['TradeVolume'],
|
# data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
|
# data['SellNo'], data['ExecType'])
|
@return:
|
"""
|
deal_time = l2_huaxin_util.convert_time(data[3], True)
|
# 生效时间在1s以内
|
cls.__latest_sell_data_dict[code] = (data, tool.trade_time_add_millionsecond(deal_time, 1000))
|
if cls.__callback:
|
cls.__callback.OnTradeSingle(code, cls.TYPE_PASSIVE, cls.__latest_sell_data_dict[code])
|
|
@classmethod
|
def set_sell_passive_to_active_datas(cls, code, passive_data, active_data):
|
"""
|
设置被动卖成交向主动卖成交的转变数据
|
@param code: 代码
|
@param passive_data: 被动卖成交逐笔
|
@param active_data: 主动卖成交逐笔
|
@return:
|
"""
|
l2_log.info(code, logger_l2_trade_buy, f"被动卖变主动卖：{passive_data} => {active_data}")
|
deal_time = l2_huaxin_util.convert_time(passive_data[3], True)
|
# 生效时间在1s以内
|
cls.__latest_sell_active_deal_data_dict[code] = (
|
active_data, tool.trade_time_add_millionsecond(deal_time, 1000))
|
if cls.__callback:
|
cls.__callback.OnTradeSingle(code, cls.TYPE_ACTIVE, cls.__latest_sell_active_deal_data_dict[code])
|
|
@classmethod
|
def get_valid_trade_single(cls, code, latest_time_with_ms):
|
"""
|
获取有效的成交下单信号
|
@param code:
|
@param latest_time_with_ms:
|
@return: （逐笔成交数据, 类型）
|
"""
|
# 如果有最近卖涨停成交完就用最近的数据
|
data = cls.__latest_sell_data_dict.get(code)
|
if data and tool.trade_time_sub_with_ms(latest_time_with_ms, data[1]) <= 0:
|
return data, cls.TYPE_PASSIVE
|
data = cls.__latest_sell_active_deal_data_dict.get(code)
|
if data and tool.trade_time_sub_with_ms(latest_time_with_ms, data[1]) <= 0:
|
return data, cls.TYPE_ACTIVE
|
return None
|
|
@classmethod
|
def is_can_place_order(cls, code, buy_data):
|
"""
|
是否可以下单
|
@param code: 代码
|
@param buy_data: L2逐笔委托大于50w的买入数据
|
@return: (是否可以下单, 原因)
|
"""
|
buy_data_time_with_ms = tool.to_time_with_ms(buy_data['val']['time'], buy_data['val']['tms'])
|
single = cls.get_valid_trade_single(code, buy_data_time_with_ms)
|
if not single:
|
return False, "没有找到可用信号"
|
if single[0][0][6] < buy_data['val']['orderNo']:
|
# 信号位置的成交买单号要小于买入数据的买单号才行
|
return True, f"找到信号：{single}"
|
else:
|
return False, f"买单没有在信号位之后 信号：{single}"
|
|
@classmethod
|
def clear_data(cls, code):
|
if code in cls.__latest_sell_data_dict:
|
cls.__latest_sell_data_dict.pop(code)
|
|
if code in cls.__latest_sell_active_deal_data_dict:
|
cls.__latest_sell_active_deal_data_dict.pop(code)
|
