L前3s囊括一次/增加L2数据接口/接口输出交易信息

Administrator

2023-10-24 60d02096d9dc4bc9cb54f8cf81becd10a9dd0c61

L前3s囊括一次/增加L2数据接口/接口输出交易信息

 huaxin_client/l2_data_manager.py

@@ -208,10 +208,11 @@
                    l2_data_callback.OnL2Order(code, udatas, int(time.time() * 1000))
                    # l2_data_transaction_protocol.send_l2_order_detail(pipe, _mmap, code, udatas)
                    use_time = int((time.time() - start_time) * 1000)
                    if use_time > 20:
                    if use_time > 10:
                        async_log_util.info(logger_local_huaxin_l2_upload, f"{code}-上传代码耗时：{use_time}ms")
                time.sleep(0.01)

                else:
                    # 没有数据的时候需等待，有数据时不需等待
                    time.sleep(0.004)
            except Exception as e:
                hx_logger_contact_debug.exception(e)
                logger_local_huaxin_l2_error.error(f"上传订单数据出错：{str(e)}")

 l2/cancel_buy_strategy.py

@@ -835,6 +835,8 @@

    __SecondCancelBigNumComputer = SecondCancelBigNumComputer()

    __last_l_up_compute_info = {}

    __instance = None

    def __new__(cls, *args, **kwargs):
@@ -1027,6 +1029,16 @@
        else:
            self.__compute_trade_progress_near_by_indexes(code, buy_single_index, index)

    # 重新计算L上
    def re_compute_l_up_watch_indexes(self, code):
        if code not in self.__last_trade_progress_dict:
            return
        if code not in self.__real_place_order_index_dict:
            return
        if code not in self.__last_l_up_compute_info or time.time() - self.__last_l_up_compute_info[code][0] >= 3:
            self.__compute_trade_progress_near_by_indexes(code, self.__last_trade_progress_dict.get(code) + 1,
                                                          self.__real_place_order_index_dict.get(code))

    # 计算范围内的成交位临近未撤大单
    def __compute_trade_progress_near_by_indexes(self, code, start_index, end_index):
        if start_index is None or end_index is None:
@@ -1055,9 +1067,16 @@
                watch_indexes.add(i)
                if len(watch_indexes) >= MAX_COUNT:
                    break

        changed = True
        if code in self.__last_l_up_compute_info:
            if self.__last_l_up_compute_info[code] == watch_indexes:
                changed = False
        # 保存数据
        l2_log.l_cancel_debug(code, f"设置成交位临近撤单监控范围：{watch_indexes} 计算范围：{start_index}-{end_index}")
        self.__set_near_by_trade_progress_indexes(code, watch_indexes)
        if changed:
            l2_log.l_cancel_debug(code, f"设置成交位临近撤单监控范围：{watch_indexes} 计算范围：{start_index}-{end_index}")
            self.__set_near_by_trade_progress_indexes(code, watch_indexes)
        self.__last_l_up_compute_info[code] = (time.time(), watch_indexes)

    # 设置成交位置,成交位置变化之后相应的监听数据也会发生变化
    def set_trade_progress(self, code, index, total_datas):

 l2/l2_transaction_data_manager.py

@@ -72,10 +72,9 @@
                    rate = round(total_deal_nums / (thresh_hold_money // (float(limit_up_price) * 100)), 2)
                    LCancelRateManager().set_big_num_deal_rate(code, rate)

                    # 获取执行位时间
            buy_single_index, buy_exec_index, compute_index, num, count, max_num_set, volume_rate = l2_data_manager.TradePointManager().get_buy_compute_start_data_cache(code)
            if buy_progress_index is not None:
                # 获取执行位时间
                buy_single_index, buy_exec_index, compute_index, num, count, max_num_set, volume_rate = l2_data_manager.TradePointManager().get_buy_compute_start_data_cache(
                    code)
                cls.__TradeBuyQueue.set_traded_index(code, buy_progress_index)
                async_log_util.info(logger_l2_trade_buy_queue, "获取成交位置成功： code-{} index-{}", code,
                                    buy_progress_index)
@@ -100,6 +99,10 @@
                    HourCancelBigNumComputer().set_transaction_index(code, buy_progress_index)
            else:
                pass
            if buy_exec_index and buy_exec_index > -1:
                # 触发L撤上重新计算
                LCancelBigNumComputer().re_compute_l_up_watch_indexes(code)

        except Exception as e:
            hx_logger_l2_transaction.exception(e)
        finally:

 l2/transaction_progress.py

@@ -5,6 +5,7 @@
# 买入队列
import itertools
import json
import time

import constant
from db import redis_manager_delegate as redis_manager
@@ -16,6 +17,9 @@

class TradeBuyQueue:
    buy_progress_index_cache = {}
    latest_buy_progress_index_cache = {}
    # 成交速率
    trade_speed_cache = {}

    __db = 0
    __redis_manager = redis_manager.RedisManager(0)
@@ -181,9 +185,22 @@
        index, is_default = self.__get_buy_progress_index_cache(code)
        return index, is_default

    def set_traded_index(self, code, index):
    # 设置交易进度
    def set_traded_index(self, code, index, total_datas=None):
        last_info = self.latest_buy_progress_index_cache.get(code)
        if not last_info or last_info[0] != index:
            if last_info:
                val = total_datas[last_info[0]]
                rate = round(val["num"] * float(val["price"]) * 100 / (time.time() - last_info[1]))
                # 成交速率
                self.trade_speed_cache[code] = rate
            self.latest_buy_progress_index_cache[code] = (index, time.time())
        self.__save_buy_progress_index(code, index, False)

    # 获取成交速率
    def get_trade_speed(self, code):
        return self.trade_speed_cache.get(code)


if __name__ == '__main__':
    a = [1, 2, 3, 4]

 outside_api_command_manager.py

@@ -54,6 +54,7 @@
API_TYPE_SYNC_L1_TARGET_CODES = "sync_l1_subscript_codes"  # 同步L1需要订阅的代码
API_TYPE_SYSTEM_LOG = "system_log"  # 系统日志
API_TYPE_GET_FROM_DATA_SERVER = "get_from_data_server"  # 从数据服务器拉取数据
API_TYPE_CODE_TRADE_INFO = "code_trade_info"

class ActionCallback(object):
    # 交易
@@ -94,6 +95,10 @@
        pass

    def OnGetFromDataServer(self, client_id, request_id, data):
        pass

    # 代码的交易信息
    def OnGetCodeTradeInfo(self, client_id, request_id, data):
        pass


@@ -195,6 +200,8 @@
                            cls.action_callback.OnSystemLog(client_id, request_id, data)
                        elif content_type == API_TYPE_GET_FROM_DATA_SERVER:
                            cls.action_callback.OnGetFromDataServer(client_id, request_id, data)
                        elif content_type == API_TYPE_CODE_TRADE_INFO:
                            cls.action_callback.OnGetCodeTradeInfo(client_id, request_id, data)
                    except Exception as e:
                        logging.exception(e)
                        pass

 third_data/data_server.py

@@ -8,7 +8,7 @@
import dask

from log_module.log import logger_system, logger_debug
from utils import global_util, tool
from utils import global_util, tool, data_export_util
from code_attribute import gpcode_manager
from log_module import log, log_analyse, log_export
from l2 import code_price_manager, l2_data_util, l2_data_manager_new, cancel_buy_strategy
@@ -299,20 +299,28 @@
            ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
            code = ps_dict['code']
            name = ps_dict.get('name')
            try:
                data = code_info_output.get_output_params(code, self.__jingxuan_cache_dict, self.__industry_cache_dict)
                if data["code_name"].find("None") > -1 and name:
                    data["code_name"] = f"{name} {code}"

            data = code_info_output.get_output_params(code, self.__jingxuan_cache_dict, self.__industry_cache_dict)
            if data["code_name"].find("None") > -1 and name:
                data["code_name"] = f"{name} {code}"
                self.__history_plates_dict[code] = (time.time(), data["kpl_code_info"]["code_records"])
                if "plate" in data["kpl_code_info"]:
                    self.__blocks_dict[code] = (time.time(), data["kpl_code_info"]["plate"])

            self.__history_plates_dict[code] = (time.time(), data["kpl_code_info"]["code_records"])
            if "plate" in data["kpl_code_info"]:
                self.__blocks_dict[code] = (time.time(), data["kpl_code_info"]["plate"])

            response_data = json.dumps({"code": 0, "data": data})
            print("get_score_info 耗时：", time.time() - start_time)
                response_data = json.dumps({"code": 0, "data": data})
                print("get_score_info 耗时：", time.time() - start_time)
            except Exception as e:
                logging.exception(e)

            # 获取评分信息
            pass
        elif url.path == "/get_l2_datas":
            # 获取L2的数据
            ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
            code = ps_dict['code']
            datas = data_export_util.get_l2_datas(code)
            response_data = json.dumps({"code": 0, "data": datas})
        elif url.path == "/kpl/get_limit_up_list":
            response_data = self.__get_limit_up_list()


 trade/huaxin/huaxin_trade_api.py

@@ -81,12 +81,15 @@
                                                          insertTime=insertTime, acceptTime=acceptTime,
                                                          insertDate=insertDate, direction=direction,
                                                          is_shadow_order=is_shadow_order)
                                TradeResultProcessor.process_order(order)

                                try:
                                    huaxin_trade_record_manager.DelegateRecordManager.add_one(data)
                                except:
                                    pass
                                    TradeResultProcessor.process_order(order)
                                finally:
                                    try:
                                        for j in range(2):
                                            # 加入2次，增大加入成功率
                                            huaxin_trade_record_manager.DelegateRecordManager.add_one(data)
                                    except Exception as e:
                                        hx_logger_trade_debug.exception(e)
                                if not is_shadow_order:
                                    # 订单相关回调
                                    # 重新请求委托列表与资金
@@ -370,7 +373,7 @@
        else:
            return {"order_ref": order_ref}
    finally:
        huaxin_trade_data_update.add_delegate_list("下单", delay=0.2)
        # huaxin_trade_data_update.add_delegate_list("下单", delay=0.2)
        huaxin_trade_data_update.add_money_list()


@@ -403,7 +406,7 @@
    try:
        return __read_response(request_id, blocking)
    finally:
        huaxin_trade_data_update.add_delegate_list("撤单")
        # huaxin_trade_data_update.add_delegate_list("撤单")
        huaxin_trade_data_update.add_money_list()



 trade/huaxin/huaxin_trade_server.py

@@ -31,7 +31,7 @@
from huaxin_client.client_network import SendResponseSkManager
from huaxin_client.trade_transform_protocol import TradeResponse
from l2 import l2_data_manager_new, l2_log, code_price_manager, l2_data_util, l2_data_manager, transaction_progress, \
    l2_data_log
    l2_data_log, l2_data_source_util
from l2.cancel_buy_strategy import HourCancelBigNumComputer, LCancelBigNumComputer, DCancelBigNumComputer, \
    GCancelBigNumComputer, SecondCancelBigNumComputer, LCancelRateManager, LatestCancelIndexManager
from l2.huaxin import huaxin_target_codes_manager
@@ -56,7 +56,7 @@
    huaxin_trade_record_manager
from trade.l2_trade_factor import L2PlaceOrderParamsManager
from trade.trade_manager import TradeTargetCodeModeManager
from utils import socket_util, data_export_util, middle_api_protocol, tool, huaxin_util
from utils import socket_util, data_export_util, middle_api_protocol, tool, huaxin_util, output_util

trade_data_request_queue = queue.Queue()

@@ -737,6 +737,81 @@
        except:
            self.send_response(json.dumps({"code": 1, "msg": "网络请求出错"}), client_id, request_id)

    # 代码的交易信息
    def OnGetCodeTradeInfo(self, client_id, request_id, data):
        try:
            code = data["code"]
            # 获取交易信息，
            # 获取正在成交的位置/获取下单位置/获取成交速率
            total_datas = l2_data_util.local_today_datas.get(code)
            trade_index = transaction_progress.TradeBuyQueue().get_traded_index(code)
            trade_speed = transaction_progress.TradeBuyQueue().get_trade_speed(code)
            # 下单位置
            place_order_index = SecondCancelBigNumComputer().get_real_place_order_index_cache(code)
            fdata = {}
            if trade_index and place_order_index:
                # 有成交进度位与下单位
                total_count = 0
                total_money = 0
                big_money_300_indexs = []
                big_money_200_indexs = []
                for i in range(trade_index, place_order_index):
                    data = total_datas[i]
                    val = data["val"]
                    if not L2DataUtil.is_limit_up_price_buy(val):
                        continue
                    # 是否已经撤单
                    left_count = l2_data_source_util.L2DataSourceUtils.get_limit_up_buy_no_canceled_count_v2(code, i,
                                                                                                             total_datas,
                                                                                                             l2_data_util.local_today_canceled_buyno_map.get(
                                                                                                                 code))
                    if left_count <= 0:
                        continue
                    total_count += left_count
                    money = val["num"] * int(val["price"] * 100)
                    total_money += money
                    if money >= 300 * 10000:
                        big_money_300_indexs.append(i)
                    elif money >= 200 * 10000:
                        big_money_200_indexs.append(i)
                fdata["waiting_for_trade"] = f"{total_count}笔&{output_util.money_desc(total_money)}"

                total_count = 0
                total_money = 0
                for i in big_money_300_indexs:
                    data = total_datas[i]
                    val = data["val"]
                    total_count += 1
                    money = val["num"] * int(val["price"] * 100)
                    total_money += money
                fdata["big_num_300"] = {"desc": f"{total_count}笔&{output_util.money_desc(total_money)}",
                                        "datas": [output_util.format_l2_data(total_datas[x]) for x in
                                                  big_money_300_indexs]}

                total_count = 0
                total_money = 0
                for i in big_money_200_indexs:
                    data = total_datas[i]
                    val = data["val"]
                    total_count += 1
                    money = val["num"] * int(val["price"] * 100)
                    total_money += money
                fdata["big_num_200"] = {"desc": f"{total_count}笔&{output_util.money_desc(total_money)}",
                                        "datas": [output_util.format_l2_data(total_datas[x]) for x in
                                                  big_money_200_indexs]}
                if trade_speed:
                    seconds = int(total_money / trade_speed)
                    h = seconds // 3600
                    m = seconds % 3600 // 60
                    s = seconds % 60
                    fdata["trade_use_time"] = "{:0>2d}:{:0>2d}:{:0>2d}".format(h, m, s)
                    fdata["trade_time"] = tool.trade_time_add_second(tool.get_now_time_str(), seconds)
            result = {"code": 0, "data": fdata}
            self.send_response(result, client_id, request_id)
        except Exception as e:
            logging.exception(e)
            self.send_response(json.dumps({"code": 1, "msg": f"数据处理出错：{e}"}), client_id, request_id)


class MyL2DataCallback(l2_data_transform_protocol.L2DataCallBack):
    def OnL2Order(self, code, datas, timestamp):

 utils/data_export_util.py

@@ -24,10 +24,21 @@
    # 获取L2处理位置信息
    process_indexs = log_export.get_l2_process_position(code, date)
    trade_indexs = log_export.get_l2_trade_position(code, date)
    export_l2_data(code, datas, process_indexs, trade_indexs)
    fdatas = export_l2_data(code, datas, process_indexs, trade_indexs)
    __save_l2_datas(code, fdatas)


def export_l2_data(code, datas, process_indexs, trade_indexs, dest_dir=f"{constant.get_path_prefix()}/export/l2"):
# 获取L2的数据
def get_l2_datas(code, date=None):
    local_today_datas = log_export.load_l2_from_log(date)
    datas = local_today_datas[code]
    process_indexs = log_export.get_l2_process_position(code, date)
    trade_indexs = log_export.get_l2_trade_position(code, date)
    fdatas = export_l2_data(code, datas, process_indexs, trade_indexs)
    return fdatas


def export_l2_data(code, datas, process_indexs, trade_indexs):
    def find_process_index(index):
        for i in range(0, len(process_indexs)):
            if process_indexs[i][0] <= index <= process_indexs[i][1]:
@@ -54,48 +65,23 @@
    #         num_dict[data["val"]["num"]] = []
    #     num_dict[data["val"]["num"]].append(data)

    local_time = time.strftime("%Y%m%dT%H%M%S", time.localtime(time.time()))
    file_name = "{}/{}_{}.xls".format(dest_dir, code, local_time)
    file_name_txt = "{}/{}_{}.txt".format(dest_dir, code, local_time)
    with open(file_name_txt, 'w') as openfile:
        for data in datas:
            openfile.write(json.dumps(data) + "\n")
    wb = xlwt.Workbook(encoding="utf-8")
    ws = wb.add_sheet('sheet1')
    ws.write(0, 0, '序号')
    ws.write(0, 1, '时间')
    ws.write(0, 2, '买撤间隔')
    ws.write(0, 3, '金额')
    ws.write(0, 4, '价格')
    ws.write(0, 5, '手数')
    ws.write(0, 6, '类型')
    ws.write(0, 7, '重复数量')
    ws.write(0, 8, '撤单时间')
    ws.write(0, 9, '订单号')

    index = 0
    fdatas = []
    for data in datas:
        index += 1
        trade_info = find_trade_index(data["index"])
        font = xlwt.Font()
        if trade_info:
            if trade_info[0] == 0:
                font.colour_index = 53
            elif trade_info[0] == 1:
                font.colour_index = 17
            elif trade_info[0] == 2:
                font.colour_index = 10
                ws.write(index, 8, trade_info[2])
        style = None
        style_int = None
        if find_process_index(data["index"]) % 2 == 0:
            style = xlwt.easyxf('pattern: pattern solid')
            style_int = 0
        else:
            style = xlwt.easyxf('pattern: pattern solid, fore_colour light_yellow')
        style.font = font
        cancel_style = xlwt.easyxf('pattern: pattern solid, fore_colour gray25')
            style_int = 1

        ws.write(index, 0, data["index"], style)
        ws.write(index, 1, data["val"]["time"] + (f".{data['val']['tms']}" if "tms" in data["val"] else ''), style)
        format_data = []
        # 索引
        format_data.append(data["index"])
        # 时间
        format_data.append(data["val"]["time"] + (f".{data['val']['tms']}" if "tms" in data["val"] else ''))

        cancel_time = data["val"]["cancelTime"]
        if cancel_time == '0':
            cancel_time = ''
@@ -108,55 +94,115 @@
                cancel_time += "m"
            elif int(data["val"]["cancelTimeUnit"]) == 2:
                cancel_time += "h"
        # 撤单间隔时间
        format_data.append(cancel_time)
        # 金额
        format_data.append("{}万".format(round(int(data["val"]["num"]) * float(data["val"]["price"]) / 100, 1)))
        # 单价
        format_data.append(data["val"]["price"])

        ws.write(index, 2, cancel_time, style)
        ws.write(index, 4, data["val"]["price"], style)
        if int(data["val"]["operateType"]) == 1 or int(data["val"]["operateType"]) == 2:
            ws.write(index, 5, 0 - int(data["val"]["num"]), style)
            format_data.append(0 - int(data["val"]["num"]))
        else:
            ws.write(index, 5, int(data["val"]["num"]), style)
            format_data.append(int(data["val"]["num"]))

        limit_price = ""
        if int(data["val"]["limitPrice"]) == 1:
            limit_price = "涨停"
        elif int(data["val"]["limitPrice"]) == 2:
            limit_price = "跌停"

        operateDesc = ""
        if int(data["val"]["operateType"]) == 0:
            if len(limit_price) > 0:
                ws.write(index, 6, '买 ({})'.format(limit_price), style)
                operateDesc = '买 ({})'.format(limit_price)
            else:
                ws.write(index, 6, '买', style)
                operateDesc = '买'
        elif int(data["val"]["operateType"]) == 1:
            if len(limit_price) > 0:
                ws.write(index, 6, '买撤 ({})'.format(limit_price), style)
                operateDesc = '买撤 ({})'.format(limit_price)

            else:
                ws.write(index, 6, '买撤', style)
                operateDesc = '买撤'

        elif int(data["val"]["operateType"]) == 2:
            if len(limit_price) > 0:
                ws.write(index, 6, '卖 ({})'.format(limit_price), style)
                operateDesc = '卖 ({})'.format(limit_price)
            else:
                ws.write(index, 6, '卖', style)
                operateDesc = '卖'

        elif int(data["val"]["operateType"]) == 3:
            if len(limit_price) > 0:
                ws.write(index, 6, '卖撤 ({})'.format(limit_price), style)
                operateDesc = '卖撤 ({})'.format(limit_price)
            else:
                ws.write(index, 6, '卖撤', style)
        ws.write(index, 7, data["re"], style)
        ws.write(index, 9, data["val"].get("orderNo"), style)
                operateDesc = '卖撤'
        format_data.append(operateDesc)
        format_data.append(data["re"])
        # 查询是否撤单
        cancel_info = None
        if int(data["val"]["operateType"]) == 0:
            cancel_data = l2.l2_data_util.local_today_canceled_buyno_map.get(code).get(str(data["val"]["orderNo"]))
            # 买
            if cancel_data:
                try:
                    ws.write(index, 8, "{}-{}".format(cancel_data["index"], cancel_data["val"]["time"]), cancel_style)
                    cancel_info = "{}-{}".format(cancel_data["index"], cancel_data["val"]["time"])
                except Exception as e:
                    logging.exception(e)
        format_data.append(cancel_info)
        cancel_order_info = None
        if trade_info:
            if trade_info[0] == 0:
                # font.colour_index = 53
                pass
            elif trade_info[0] == 1:
                # font.colour_index = 17
                pass
            elif trade_info[0] == 2:
                # font.colour_index = 10
                cancel_order_info = trade_info[2]
        format_data.append(cancel_order_info)
        format_data.append(data["val"].get("orderNo"))
        fdatas.append((style_int, format_data))
    return fdatas

        ws.write(index, 3, "{}万".format(round(int(data["val"]["num"]) * float(data["val"]["price"]) / 100, 2)), style)

def __save_l2_datas(code, fdatas, dest_dir=f"{constant.get_path_prefix()}/export/l2"):
    local_time = time.strftime("%Y%m%dT%H%M%S", time.localtime(time.time()))
    file_name = "{}/{}_{}.xls".format(dest_dir, code, local_time)
    wb = xlwt.Workbook(encoding="utf-8")
    ws = wb.add_sheet('sheet1')
    ws.write(0, 0, '序号')
    ws.write(0, 1, '时间')
    ws.write(0, 2, '买撤间隔')
    ws.write(0, 3, '金额')
    ws.write(0, 4, '价格')
    ws.write(0, 5, '手数')
    ws.write(0, 6, '类型')
    ws.write(0, 7, '重复数量')
    ws.write(0, 8, '撤单时间')
    ws.write(0, 9, '订单号')
    index = 0
    font = xlwt.Font()

    cancel_style = xlwt.easyxf('pattern: pattern solid, fore_colour gray25')
    for fdata in fdatas:
        index += 1
        style_int = fdata[0]
        data = fdatas[1]
        style = None
        if style_int == 0:
            style = xlwt.easyxf('pattern: pattern solid')
        else:
            style = xlwt.easyxf('pattern: pattern solid, fore_colour light_yellow')
        style.font = font

        for i in range(len(data)):
            if data[i] is None:
                continue
            if i == 8 and data[i]:
                ws.write(index, i, data[i], cancel_style)
            else:
                ws.write(index, i, data[i], style)
    wb.save(file_name)
    return file_name


def export_l2_data_origin(code, datas, key, dest_dir=f"{constant.get_path_prefix()}/export/l2_origin"):

 utils/output_util.py

New file
@@ -0,0 +1,8 @@
def money_desc(money):
    if abs(money) > 100000000:
        return f"{round(money / 100000000, 2)}亿"
    else:
        return f"{round(money / 10000, 2)}万"

def format_l2_data(item):
        return f"{item['val']['time']}#{item['val']['num']}手#{round(item['val']['num'] * float(item['val']['price']) * 100 / 10000, 1)}万"