统计大卖单

Administrator

2025-05-09 7b40261a51f7c5ace06908abb5347be57557d424

统计大卖单

 huaxin_client/l2_client_test.py

@@ -49,13 +49,16 @@
        self.__big_buy_orders = []
        # 精确的买单信息，{买单号：订单信息}
        self.__big_accurate_buy_order_dict = {}
        self.__big_accurate_sell_order_dict = {}
        self.__latest_sell_order = None
        self.__big_sell_orders = []
        self.big_accurate_buy_order_queue = queue.Queue(maxsize=10240)
        self.big_accurate_sell_order_queue = queue.Queue(maxsize=10240)
        self.big_buy_order_queue = queue.Queue(maxsize=10240)
        self.big_sell_order_queue = queue.Queue(maxsize=10240)
        self.accurate_buy = accurate_buy
        self.__last_accurate_buy_count = 0
        self.__last_accurate_sell_count = 0

    def get_big_buy_orders(self):
        return self.__big_buy_orders
@@ -63,7 +66,7 @@
    def get_big_sell_orders(self):
        return self.__big_sell_orders

    def add_transaction_data_for_accurate_buy(self, data):
    def add_transaction_data_for_accurate(self, data):
        """
        获取精确的买单信息
        @param data:
@@ -100,11 +103,11 @@
            if last_buy_order[2] > 299e4:
                self.big_accurate_buy_order_queue.put_nowait(last_buy_order)
            # 如果数据过多需要移除过长时间的小金额数据
            accurate_buy_count = len(self.__big_accurate_buy_order_dict)
            accurate_buy_count = len(self.__big_accurate_buy_order_dict.keys())
            if accurate_buy_count > 10000 and accurate_buy_count - self.__last_accurate_buy_count > 2000:
                # 超过1w条数据且新增2000条数据
                # 超过1w条数据就要移除30分钟之前的数据
                now_time_int = int(tool.trade_time_add_second(tool.get_now_time_str(), -1800).replace(":", ""))
                now_time_int = int(tool.trade_time_add_second(tool.get_now_time_str(), -3600).replace(":", ""))
                try:
                    remove_order_nos = [x for x in self.__big_accurate_buy_order_dict if
                                        now_time_int - format_timestamp(
@@ -114,6 +117,36 @@
                            self.__big_accurate_buy_order_dict.pop(order_no)
                finally:
                    self.__last_accurate_buy_count = len(self.__big_accurate_buy_order_dict)

        # 统计卖单
        if item[1] not in self.__big_accurate_sell_order_dict:
            # (卖单号, 量, 金额, 时间, 最新成交价格)
            self.__big_accurate_sell_order_dict[item[1]] = [item[1], 0, 0, order_time, price]
        sell_order_info = self.__big_accurate_sell_order_dict[item[1]]
        sell_order_info[1] += volume
        sell_order_info[2] += money
        sell_order_info[3] = order_time
        sell_order_info[4] = price
        if self.__latest_sell_order and self.__latest_sell_order[0] != item[1]:
            # 有可能是大单成交完成， 判断上个订单是否是大单
            last_sell_order = self.__big_accurate_sell_order_dict.get(self.__latest_sell_order[0])
            if last_sell_order[2] > 299e4:
                self.big_accurate_sell_order_queue.put_nowait(last_sell_order)
            # 如果数据过多需要移除过长时间的小金额数据
            accurate_sell_count = len(self.__big_accurate_sell_order_dict.keys())
            if accurate_sell_count > 10000 and accurate_sell_count - self.__last_accurate_sell_count > 2000:
                # 超过1w条数据且新增2000条数据
                # 超过1w条数据就要移除30分钟之前的数据
                now_time_int = int(tool.trade_time_add_second(tool.get_now_time_str(), -3600).replace(":", ""))
                try:
                    remove_order_nos = [x for x in self.__big_accurate_sell_order_dict if
                                        now_time_int - format_timestamp(
                                            self.__big_accurate_sell_order_dict[x][3]) > 0]
                    if remove_order_nos:
                        for order_no in remove_order_nos:
                            self.__big_accurate_sell_order_dict.pop(order_no)
                finally:
                    self.__last_accurate_sell_count = len(self.__big_accurate_sell_order_dict.keys())

    def add_transaction_data(self, data):
        item = (data["BuyNo"], data["SellNo"], data["TradePrice"], data["TradeVolume"])
@@ -129,7 +162,7 @@
        order_time = data["OrderTime"]

        if self.accurate_buy:
            self.add_transaction_data_for_accurate_buy(data)
            self.add_transaction_data_for_accurate(data)

        if not self.__latest_buy_order:
            # (买单号, 量, 金额, 时间, 最新成交价格)
@@ -294,7 +327,8 @@
                    "SellNo": pTransaction['SellNo'],
                    "ExecType": pTransaction['ExecType'].decode()}
            if item["SecurityID"] not in l2_transaction_data_dict:
                l2_transaction_data_dict[item["SecurityID"]] = L2TransactionDataManager(item["SecurityID"],  item["SecurityID"] in self.special_codes)
                l2_transaction_data_dict[item["SecurityID"]] = L2TransactionDataManager(item["SecurityID"], item[
                    "SecurityID"] in self.special_codes)
            l2_transaction_data_dict[item["SecurityID"]].add_transaction_data(item)

    def OnRtnNGTSTick(self, pTick):
@@ -313,7 +347,8 @@
                        "SellNo": pTick['SellNo'],
                        "ExecType": '1'}
                if item["SecurityID"] not in l2_transaction_data_dict:
                    l2_transaction_data_dict[item["SecurityID"]] = L2TransactionDataManager(item["SecurityID"], item["SecurityID"] in self.special_codes)
                    l2_transaction_data_dict[item["SecurityID"]] = L2TransactionDataManager(item["SecurityID"], item[
                        "SecurityID"] in self.special_codes)
                l2_transaction_data_dict[item["SecurityID"]].add_transaction_data(item)
        except Exception as e:
            logger_local_huaxin_l2_subscript.exception(e)
@@ -370,39 +405,46 @@
        logger_system.exception(e)
    while True:
        try:
            # 读取一遍
            for code in l2_transaction_data_dict:
                l2_transaction_data_manager: L2TransactionDataManager = l2_transaction_data_dict[code]
                try:
                    while True:

                while True:
                    if not l2_transaction_data_manager.big_buy_order_queue.empty():
                        result = l2_transaction_data_manager.big_buy_order_queue.get(block=False)
                        if result:
                            _queue.put_nowait((code, 0, result))
                        else:
                            break
                except:
                    pass
                    else:
                        break

                try:
                    while True:
                while True:
                    if not l2_transaction_data_manager.big_accurate_buy_order_queue.empty():
                        result = l2_transaction_data_manager.big_accurate_buy_order_queue.get(block=False)
                        if result:
                            accurate_buy_order_queue.put_nowait((code, 0, result))
                        else:
                            break
                except:
                    pass
                    else:
                        break

                while True:
                    if not l2_transaction_data_manager.big_accurate_sell_order_queue.empty():
                        result = l2_transaction_data_manager.big_accurate_sell_order_queue.get(block=False)
                        if result:
                            accurate_buy_order_queue.put_nowait((code, 1, result))
                    else:
                        break

                try:
                    while True:
                while True:
                    if not l2_transaction_data_manager.big_sell_order_queue.empty():
                        result = l2_transaction_data_manager.big_sell_order_queue.get(block=False)
                        if result:
                            _queue.put_nowait((code, 1, result))
                        else:
                            break
                except:
                    pass
                    else:
                        break
        except:
            pass
        finally:
            time.sleep(1)


if __name__ == "__main__":
    pass

 l2_data_parser.py

@@ -5,12 +5,16 @@
import os
import sys
from db import mysql_data_delegate as mysql_data
from log_module import log_export


def __get_target_codes(day):
    results = mysql_data.Mysqldb().select_all(f"select _code from kpl_limit_up_record where _day='{day}'")
    codes = set([x[0] for x in results])
    return codes
    special_codes_dict = log_export.load_special_codes(day)
    fcodes = set()
    for plate in special_codes_dict:
        codes = special_codes_dict[plate]
        fcodes |= codes
    fcodes


def parse_order_detail(day, code):
@@ -105,7 +109,7 @@
        print("接收的参数", params)
        _type = params[0].strip()
        day = params[1].strip()
        if len(params)>2:
        if len(params) > 2:
            code = params[2].strip()
        else:
            code = None
@@ -117,4 +121,3 @@
            parse_ngtstick(day)
        elif _type == 'MarketData':
            parse_market_data(day)


 l2_test.py

@@ -18,6 +18,7 @@
from third_data.custom_block_in_money_manager import CodeInMoneyManager, BlockInMoneyRankManager
from third_data.history_k_data_manager import HistoryKDataManager
from third_data.history_k_data_util import HistoryKDatasUtils
from trade.buy_radical.block_special_codes_manager import BlockSpecialCodesManager
from utils import tool, middle_api_protocol, global_util
import urllib.parse as urlparse
from urllib.parse import parse_qs
@@ -129,6 +130,10 @@
                if limit_up_price > k_bars[0]["high"]:
                    # 今日涨停价要突破昨日最高价
                    codes.add(code)
        # 获取辨识度的票
        special_codes = BlockSpecialCodesManager().get_origin_code_blocks_dict().keys()
        if special_codes:
            codes |= set(special_codes)
        return codes
    except Exception as e:
        logger_system.exception(e)

 log_module/log_export.py

@@ -845,6 +845,22 @@
    return fdatas


def load_special_codes(date=tool.get_now_date_str()):
    """
    加载之前的收盘价
    @param date:
    @return:
    """
    fdatas = {}
    path = f"{constant.get_path_prefix()}/logs/gp/plates/special_codes.{date}.log"
    lines = __load_file_content(path)
    if lines:
            line = lines[0]
            line = line[line.find(" - ") + 3:]
            return eval(line)
    return None


if __name__ == '__main__':
    line = """
    2025-03-12 14:49:15.028 | DEBUG    | log_module.async_log_util:run_sync:66 - [14:49:14.899602] thread-id=3048 code=600841  L前监控范围：{1477, 1478, 1479, 1480, 1481, 1482, 1486, 1487, 1488, 1489, 1492, 1493, 1495, 1498, 1500} 计算范围：1477-1503

 trade/buy_radical/block_special_codes_manager.py

@@ -61,6 +61,12 @@

        return fdict

    def get_origin_code_blocks_dict(self):
        fdict = {}
        for code in self.__code_blocks_dict:
            fdict[code] = set(self.__code_blocks_dict.get(code))
        return fdict

    def get_temp_code_blocks_dict(self):
        fdict = {}
        for code in self.__temp_code_blocks_dict: