统计大单成交信息

Administrator

2024-12-24 9e64f43ae71007889f4923860a69e557aa3b1002

统计大单成交信息

 constant.py

@@ -55,6 +55,7 @@
def get_path_prefix():
    return 'D:' if is_windows() else '/home/userzjj'


# 订阅L2代码数据
SUBSCRIPT_L2_CODES = set()

CACHE_PATH = f"{'D:' if is_windows() else '/home/userzjj'}/trade_cache"

 log_module/log.py

@@ -239,6 +239,14 @@
                   filter=lambda record: record["extra"].get("name") == "hx_l2_debug",
                   rotation="00:00", compression="zip", enqueue=True)

        logger.add(self.get_hx_path("l2", "transaction_big_buy_order"),
                   filter=lambda record: record["extra"].get("name") == "hx_l2_transaction_big_buy_order",
                   rotation="00:00", compression="zip", enqueue=True)

        logger.add(self.get_hx_path("l2", "transaction_big_sell_order"),
                   filter=lambda record: record["extra"].get("name") == "hx_l2_transaction_big_sell_order",
                   rotation="00:00", compression="zip", enqueue=True)

        logger.add(self.get_hx_path("contact", "debug"),
                   filter=lambda record: record["extra"].get("name") == "hx_contact_debug",
                   rotation="00:00", compression="zip", enqueue=True)
@@ -418,6 +426,10 @@
hx_logger_trade_callback = __mylogger.get_logger("hx_trade_callback")
hx_logger_trade_debug = __mylogger.get_logger("hx_trade_debug")
hx_logger_trade_loop = __mylogger.get_logger("hx_trade_loop")
hx_logger_l2_transaction_big_buy_order = __mylogger.get_logger("hx_l2_transaction_big_buy_order")
hx_logger_l2_transaction_big_sell_order = __mylogger.get_logger("hx_l2_transaction_big_sell_order")



# -------------------------------华鑫本地日志---------------------------------
logger_local_huaxin_l2_transaction = __mylogger.get_logger("local_huaxin_transaction")

 trade/l2_data_manager.py

New file
@@ -0,0 +1,72 @@
from log_module import async_log_util
from log_module.log import hx_logger_l2_transaction_big_buy_order, hx_logger_l2_transaction_big_sell_order
from utils import tool


@tool.singleton
class L2TransactionDataManager:
    def __init__(self):
        self.__latest_buy_order = {}
        self.__big_buy_orders = {}
        self.__latest_sell_order = {}
        self.__big_sell_orders = {}
        self.__load_data()

    def __load_data(self):
        pass

    def get_big_buy_orders(self):
        return self.__big_buy_orders

    def get_big_sell_orders(self):
        return self.__big_sell_orders

    def add_transaction_data(self, code, data):
        """

        :param data:(data['SecurityID'], data['TradePrice'], data['TradeVolume'],
                  data['OrderTime'], data['MainSeq'], data['SubSeq'], data['BuyNo'],
                  data['SellNo'], data['ExecType'])
        :return:
        """
        # 买单号, 卖单号, 成交价格, 成交量
        item = (data[6], data[7], data[1], data[2])
        money = round(item[2] * item[3])
        volume = item[3]
        price = item[2]
        order_time = data["OrderTime"]
        if code not in self.__latest_buy_order:
            # (买单号, 量, 金额, 时间, 最新成交价格)
            self.__latest_buy_order[code] = [item[0], 0, 0, order_time, price]
        if self.__latest_buy_order[code][0] == item[0]:
            self.__latest_buy_order[code][1] += volume
            self.__latest_buy_order[code][2] += money
            self.__latest_buy_order[code][3] = order_time
            self.__latest_buy_order[code][4] = price
        else:
            if self.__latest_buy_order[code][2] >= 5e5:
                d = (
                    self.__latest_buy_order[code][0], self.__latest_buy_order[code][1],
                    self.__latest_buy_order[code][2],
                    self.__latest_buy_order[code][3], self.__latest_buy_order[code][4])
                self.__big_buy_orders[code].append(d)
                async_log_util.info(hx_logger_l2_transaction_big_buy_order, f"{code}#{d}")

            self.__latest_buy_order[code] = [item[0], volume, money, order_time, price]

        if code not in self.__latest_sell_order:
            self.__latest_sell_order[code] = [item[1], 0, 0, order_time, price]
        if self.__latest_sell_order[code][0] == item[1]:
            self.__latest_sell_order[code][1] += volume
            self.__latest_sell_order[code][2] += money
            self.__latest_sell_order[code][3] = order_time
            self.__latest_sell_order[code][4] = price
        else:
            if self.__latest_sell_order[code][2] > 5e5:
                d = (
                    self.__latest_sell_order[code][0], self.__latest_sell_order[code][1],
                    self.__latest_sell_order[code][2],
                    self.__latest_sell_order[code][3], self.__latest_sell_order[code][4])
                self.__big_sell_orders[code].append(d)
                async_log_util.info(hx_logger_l2_transaction_big_sell_order, f"{code}#{d}")
            self.__latest_sell_order[code] = [item[1], volume, money, order_time, price]

 trade/trade_strategy.py

@@ -23,6 +23,7 @@
    logger_trade_position_api_request, logger_l2_error, hx_logger_l2_transaction, printlog
from trade import huaxin_trade_data_update, huaxin_sell_util, huaxin_trade_api
from trade.huaxin_trade_record_manager import PositionManager, DelegateSellOrderManager
from trade.l2_data_manager import L2TransactionDataManager
from trade.sell_rule_manager import TradeRuleManager, SellRule
from utils import outside_api_command_manager, middle_api_protocol, tool, huaxin_util, socket_util, cb_data_util
from utils.outside_api_command_manager import ActionCallback
@@ -539,6 +540,11 @@
            L2DataProcessor.set_deal_price(code, price, time_str)
            # 获取当前票是否有09:30之前的委托卖
            try:
                for data in datas:
                    L2TransactionDataManager().add_transaction_data(code, data)
            except Exception as e:
                logger_debug.exception(e)
            try:
                delegate_sell_orders = DelegateSellOrderManager.get_delegate_sell_orders(code)
                if delegate_sell_orders:
                    for d in delegate_sell_orders:
@@ -569,8 +575,6 @@
        if code in codes:
            async_log_util.info(logger_debug, f"OnRealTimeBuy1Info：{code}-{buy1_info}")
            # L1DataProcessor.excute_sell_rule(code, buy1_info[3], buy1_info[1], "l2-real")




# 做一些初始化的操作

 utils/kpl_data_manager.py

New file
@@ -0,0 +1,77 @@
import json
import os

import constant
from utils import tool
from utils.tool import singleton


@singleton
class KPLLimitUpDataManager:
    __yesterday_limit_up_codes = set()

    @classmethod
    # 获取最近几天的数据，根据日期倒序返回
    def __get_latest_from_file(cls, type, count, max_day=tool.get_now_date_str()):
        files = os.listdir(constant.CACHE_PATH)
        file_name_list = []
        for f in files:
            if f[10:] == f"_{type}.log":
                file_name_list.append((f.split("_")[0], f))

        file_name_list.sort(key=lambda x: x[0], reverse=True)
        fresults = []
        for file in file_name_list:
            path = f"{constant.CACHE_PATH}/{file[1]}"
            if not os.path.exists(path):
                continue
            with open(path, 'r') as f:
                lines = f.readlines()
                if lines:
                    if int(file[0].replace("-", "")) <= int(max_day.replace("-", "")):
                        fresults.append((file[0], json.loads(lines[0])))
                if len(fresults) >= count:
                    break
        return fresults

    @classmethod
    def __get_current_limit_up_data_records(cls, count, day=tool.get_now_date_str()):
        fresults = []
        datas = cls.__get_latest_from_file("limit_up", count + 2, max_day=day)
        # 移除比今天还大的数据
        fdatas = []
        for d in datas:
            if int(d[0].replace("-", "")) > int(day.replace("-", "")):
                continue
            fdatas.append(d)
        datas = fdatas
        for i in range(len(datas)):
            if datas[i][0] == day:
                continue
            fresults.append(datas[i])
            if len(fresults) >= count:
                break
        return fresults

    @classmethod
    def __get_yesterday_limit_up_codes(cls):
        yesterday_limit_up_data_records = cls.__get_yesterday_current_limit_up_records()
        yesterday_codes = set([x[0] for x in yesterday_limit_up_data_records])
        return yesterday_codes

    @classmethod
    def __get_yesterday_current_limit_up_records(cls):
        yesterday_limit_up_data_records = cls.__get_current_limit_up_data_records(1)[0][1]
        return yesterday_limit_up_data_records

    def __init__(self):
        # 加载昨日涨停
        self.__yesterday_limit_up_codes = self.__get_yesterday_limit_up_codes()

    def get_yesterday_limit_up_codes(self):
        return self.__yesterday_limit_up_codes


if __name__ == '__main__':
    pass
    # print(KPLLimitUpDataManager().get_yesterday_limit_up_codes())

 utils/tool.py

@@ -22,6 +22,22 @@
    return wrapper


def singleton(cls):
    """
    单例装饰器
    @param cls:
    @return:
    """
    instances = {}

    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]

    return get_instance


def get_expire():
    now = int(t.time())
    end = int(t.time()) + 60 * 60 * 24