共享内存/zeromq通信测试

Administrator

2024-04-18 15ee75f52d3258b14670d0487b36eae15a4e9dbc

 l2/l2_transaction_data_manager.py

@@ -5,18 +5,22 @@

from db import redis_manager
from db.redis_manager_delegate import RedisUtils
from l2 import l2_log
from l2.huaxin import l2_huaxin_util
from l2.l2_data_util import local_today_sellno_map, local_today_datas
from l2.place_order_single_data_manager import L2TradeSingleDataProcessor, L2TradeSingleDataManager

from log_module import async_log_util
from log_module.log import hx_logger_l2_transaction_desc
from log_module.log import hx_logger_l2_transaction_desc, hx_logger_l2_transaction_sell_order, hx_logger_l2_active_sell

from utils import tool


class HuaXinTransactionDataManager:
# 成交数据统计
class HuaXinBuyOrderManager:
    __db = 0
    __instance = None
    __redis_manager = redis_manager.RedisManager(0)


    # 正在成交的订单
    __dealing_order_info_dict = {}
@@ -25,7 +29,7 @@

    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = super(HuaXinTransactionDataManager, cls).__new__(cls, *args, **kwargs)
            cls.__instance = super(HuaXinBuyOrderManager, cls).__new__(cls, *args, **kwargs)
            cls.__load_datas()
        return cls.__instance

@@ -82,3 +86,188 @@
        return None


# 卖单统计数据
class HuaXinSellOrderStatisticManager:
    # 最近的大卖单成交，格式:{code:[卖单信息,...]}
    __latest_sell_order_info_list_dict = {}

    # 大单卖单号的集合，格式：{code:{卖单号}}
    __big_sell_order_ids_dict = {}

    # 大卖单的卖单号->卖单信息映射
    __big_sell_order_info_dict = {}

    # 大单列表
    __big_sell_order_info_list_dict = {}

    # 最近的卖单, 格式{code:[卖单号,总手数,价格,（'开始时间',买单号）,（'结束时间',买单号）]}
    __latest_sell_order_dict = {}
    # 最近所有的卖单
    __latest_all_sell_orders_dict = {}

    # 保存最近成交的价格
    __latest_trade_price_dict = {}

    __last_trade_data_dict = {}

    # 返回最近1s的大单卖：(总卖金额,[(卖单号,总手数,价格,（'开始时间',买单号）,（'结束时间',买单号）),...])
    @classmethod
    def add_transaction_datas(cls, code, datas, limit_up_price=None):
        # 是否为主动卖
        def is_active_sell(sell_no, buy_no):
            return sell_no > buy_no

        # q.append((data['SecurityID'], data['TradePrice'], data['TradeVolume'],
        #           data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
        #           data['SellNo'], data['ExecType']))
        if code not in cls.__latest_sell_order_info_list_dict:
            cls.__latest_sell_order_info_list_dict[code] = []
        if code not in cls.__big_sell_order_ids_dict:
            cls.__big_sell_order_ids_dict[code] = set()
        if code not in cls.__big_sell_order_info_dict:
            cls.__big_sell_order_info_dict[code] = {}

        if code not in cls.__big_sell_order_info_list_dict:
            cls.__big_sell_order_info_list_dict[code] = []

        if code not in cls.__latest_all_sell_orders_dict:
            cls.__latest_all_sell_orders_dict[code] = []

        sell_no_map = local_today_sellno_map.get(code)
        total_datas = local_today_datas.get(code)
        if not sell_no_map:
            sell_no_map = {}

        # 保存最近的成交价格:(价格,成交时间)
        cls.__latest_trade_price_dict[code] = (datas[-1][1], datas[-1][3])

        L2TradeSingleDataProcessor.process_passive_limit_up_sell_data(code, datas, limit_up_price)

        for d in datas:
            # 获取当前是否为主动买
            try:
                _is_active_sell = is_active_sell(d[7], d[6])
                if not _is_active_sell:
                    continue

                if d[1] == limit_up_price:
                    # 涨停主动卖
                    L2TradeSingleDataProcessor.add_active_limit_up_sell_data(d)

                # 判断是否是涨停被动变主动
                last_trade_data = cls.__last_trade_data_dict.get(code)
                if last_trade_data and not is_active_sell(last_trade_data[7], last_trade_data[6]) and last_trade_data[
                    1] == limit_up_price:
                    if d[1] == limit_up_price:
                        # 涨停被动变主动
                        L2TradeSingleDataManager.set_sell_passive_to_active_datas(code, last_trade_data, d)
                cls.__latest_sell_order_info_list_dict[code].append(d)
                if code not in cls.__latest_sell_order_dict:
                    cls.__latest_sell_order_dict[code] = [d[7], d[2], d[1], (d[3], d[6]), (d[3], d[6])]
                else:
                    if cls.__latest_sell_order_dict[code][0] == d[7]:
                        cls.__latest_sell_order_dict[code][1] += d[2]
                        cls.__latest_sell_order_dict[code][2] = d[1]
                        cls.__latest_sell_order_dict[code][4] = (d[3], d[6])
                    else:
                        info = cls.__latest_sell_order_dict[code]

                        # 上个卖单成交完成
                        # 封存数据，计算新起点
                        # 大于50w的卖单才会保存
                        # 大于50w加入卖单
                        money = info[1] * info[2]
                        if money >= 500000:
                            # 订单里面有成交是主动卖就算主动卖
                            l2_log.info(code, hx_logger_l2_transaction_sell_order,
                                        f"{cls.__latest_sell_order_dict[code]}")
                            cls.__big_sell_order_ids_dict[code].add(info[0])
                            cls.__big_sell_order_info_dict[code][info[0]] = info
                            cls.__big_sell_order_info_list_dict[code].append(info)
                        # 只保留10w以上的单
                        if money > 100000:
                            cls.__latest_all_sell_orders_dict[code].append(info)
                        if limit_up_price == info[2]:
                            # 将涨停主动卖记入日志
                            l2_log.info(code, hx_logger_l2_active_sell, f"{info}")

                        cls.__latest_sell_order_dict[code] = [d[7], d[2], d[1], (d[3], d[6]), (d[3], d[6])]
            finally:
                cls.__last_trade_data_dict[code] = d

        latest_time = l2_huaxin_util.convert_time(datas[-1][3], with_ms=True)
        min_time = tool.trade_time_add_millionsecond(latest_time, -1000)
        min_time_int = int(min_time.replace(":", "").replace(".", ""))
        # 计算最近1s的大单成交
        total_big_sell_datas = cls.__big_sell_order_info_list_dict.get(code)

        total_sell_info = [0, None]  # 总资金,开始成交信息,结束成交信息
        latest_sell_order_info = cls.__latest_sell_order_dict.get(code)
        if latest_sell_order_info:
            # 不是第一次非主动卖上传
            big_sell_order_ids = cls.__big_sell_order_ids_dict[code]
            # print("大卖单", big_sell_order_ids)
            big_sell_orders = []
            temp_sell_order_ids = set()
            # 统计已经结算出的大单
            for i in range(len(total_big_sell_datas) - 1, -1, -1):
                bd = total_big_sell_datas[i]
                if min_time_int > int(
                        l2_huaxin_util.convert_time(bd[3][0], with_ms=True).replace(":", "").replace(".", "")):
                    break
                if bd[0] != latest_sell_order_info[0]:
                    # 不是最近的成交且不是大单直接过滤
                    if bd[0] not in big_sell_order_ids:
                        continue
                    else:
                        if bd[0] not in temp_sell_order_ids:
                            big_sell_orders.append(cls.__big_sell_order_info_dict[code].get(bd[0]))
                            temp_sell_order_ids.add(bd[0])
                else:
                    # 是最近的但不是大单需要过滤
                    if latest_sell_order_info[1] * latest_sell_order_info[2] < 500000:
                        continue
                    else:
                        if latest_sell_order_info[0] not in temp_sell_order_ids:
                            big_sell_orders.append(latest_sell_order_info)
                            temp_sell_order_ids.add(latest_sell_order_info[0])

                # 统计最近1s的大卖单数据
                total_sell_info[0] += int(bd[1] * bd[2])
            # 统计最近的大单
            if latest_sell_order_info[1] * latest_sell_order_info[2] >= 500000:
                if latest_sell_order_info[0] not in temp_sell_order_ids:
                    # if is_active_sell(latest_sell_order_info[0], latest_sell_order_info[3][1]):
                    big_sell_orders.append(latest_sell_order_info)
                    temp_sell_order_ids.add(latest_sell_order_info[0])
                    total_sell_info[0] += int(latest_sell_order_info[1] * latest_sell_order_info[2])
            big_sell_orders.reverse()
            total_sell_info[1] = big_sell_orders
        return total_sell_info

    # 获取最近成交数据
    @classmethod
    def get_latest_transaction_datas(cls, code, min_sell_order_no=None, min_deal_time=None):
        total_orders = []
        sell_orders = cls.__latest_all_sell_orders_dict.get(code)
        if sell_orders:
            for i in range(len(sell_orders) - 1, -1, -1):
                if min_deal_time and tool.trade_time_sub(min_deal_time,
                                                         l2_huaxin_util.convert_time(sell_orders[i][3][0])) > 0:
                    break

                if min_sell_order_no and min_sell_order_no > sell_orders[i][0]:
                    continue
                total_orders.append(sell_orders[i])
        if code in cls.__latest_sell_order_dict:
            if min_sell_order_no:
                if cls.__latest_sell_order_dict[code][0] >= min_sell_order_no:
                    total_orders.append(cls.__latest_sell_order_dict[code])
            else:
                total_orders.append(cls.__latest_sell_order_dict[code])
        return total_orders

    # 获取最近成交价格信息, 返回格式：（价格,时间）
    @classmethod
    def get_latest_trade_price_info(cls, code):
        return cls.__latest_trade_price_dict.get(code)