并行框架修改

Administrator

2025-03-19 499837510adcbba27665fd48b3eb3ac135d6c48f

并行框架修改

 l2/l2_transaction_data_processor.py

@@ -104,11 +104,17 @@
            return temp_data

        limit_up_price = gpcode_manager.get_limit_up_price_as_num(code)
        # 并行处理买单与卖单
        f1 = statistic_big_buy_data()
        f2 = statistic_big_sell_data()
        dask_result = statistic_big_data(f1, f2)
        buy_datas, sell_datas = dask_result.compute()

        if len(fdatas) > 100:
            # 并行处理买单与卖单
            # 超过100条数据才需要并行处理
            f1 = dask.delayed(statistic_big_buy_data)()
            f2 = dask.delayed(statistic_big_sell_data)()
            dask_result = dask.delayed(statistic_big_data)(f1, f2)
            buy_datas, sell_datas = dask_result.compute()
        else:
            buy_datas = statistic_big_buy_data()
            sell_datas = statistic_big_sell_data()
        # L撤的比例与买卖大单无直接关系了
        # if buy_datas or sell_datas:
        #     buy_money = BigOrderDealManager().get_total_buy_money(code)
@@ -344,7 +350,7 @@
        # =====格式化数据=====
        # 整形数据，格式：[(数据本身, 是否主动买, 是否涨停, 总成交额, 不含ms时间，含ms时间)]
        use_time_list = []
        __start_time = int(time.time()*1000)
        __start_time = int(time.time() * 1000)
        fdatas = [
            [d, d[6] > d[7], limit_up_price == d[1], d[1] * d[2], '', '']
            for d in o_datas]
@@ -356,7 +362,7 @@
            d[4] = d[5][:8]
        temp_time_dict.clear()
        _start_time = int(time.time() * 1000)
        use_time_list.append((_start_time - __start_time , "数据整形"))
        use_time_list.append((_start_time - __start_time, "数据整形"))

        try:

@@ -423,4 +429,4 @@
            _start_time = int(time.time() * 1000)
            if _start_time - __start_time > 5:
                l2_log.info(code, hx_logger_l2_upload,
                            f"{code}处理成交用时：{_start_time - __start_time} 数据数量：{len(fdatas)}  详情:{use_time_list}")
                            f"{code}处理成交用时：{_start_time - __start_time} 数据数量：{len(fdatas)}  详情:{use_time_list}")