From c0bcfe746b97bc126636a658b1f01fc6a51f9f95 Mon Sep 17 00:00:00 2001
From: Administrator <admin@example.com>
Date: 星期一, 25 九月 2023 17:55:33 +0800
Subject: [PATCH] 将华鑫订单交易成功独立出来处理

---
 third_data/data_server.py |   47 ++++++++++++++++++++++++++---------------------
 1 files changed, 26 insertions(+), 21 deletions(-)

diff --git a/third_data/data_server.py b/third_data/data_server.py
index 26fba5b..1fa8dad 100644
--- a/third_data/data_server.py
+++ b/third_data/data_server.py
@@ -6,7 +6,7 @@
 from http.server import BaseHTTPRequestHandler
 import dask
 
-from log_module.log import logger_system
+from log_module.log import logger_system, logger_debug
 from utils import global_util, tool
 from code_attribute import gpcode_manager
 from log_module import log, log_analyse, log_export
@@ -25,6 +25,7 @@
 
 from trade import bidding_money_manager, trade_manager, l2_trade_util
 from trade.l2_trade_util import BlackListCodeManager
+import concurrent.futures
 
 
 class DataServer(BaseHTTPRequestHandler):
@@ -42,6 +43,7 @@
     __jingxuan_cache_dict = {}
     __industry_cache_dict = {}
     __latest_limit_up_codes_set = set()
+    __data_process_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=10)
 
     def __get_limit_up_list(self):
         # 缁熻鐩墠涓烘鐨勪唬鐮佹定鍋滄暟閲忥紙鍒嗘定鍋滃師鍥狅級
@@ -497,25 +499,28 @@
                     for code in add_codes:
                         # 鏍规嵁娑ㄥ仠鍘熷洜鍒ゆ柇鏄惁鍙互涔�
                         if code.find("00") == 0 or code.find("60") == 0:
-                            # 鍒ゆ柇鏄惁涓嬪崟
-                            trade_state = trade_manager.CodesTradeStateManager().get_trade_state(code)
-                            if trade_state == trade_manager.TRADE_STATE_BUY_PLACE_ORDER or trade_state == trade_manager.TRADE_STATE_BUY_DELEGATED:
-                                # 濮旀墭涓殑璁㈠崟锛屽垽鏂槸鍚﹂渶瑕佹挙鍗�
-                                if not gpcode_manager.WantBuyCodesManager().is_in_cache(code):
-                                    yesterday_codes = kpl_data_manager.get_yesterday_limit_up_codes()
-                                    current_limit_up_datas, limit_up_record_datas, yesterday_current_limit_up_codes, before_blocks_dict = kpl_data_manager.KPLLimitUpDataRecordManager.latest_origin_datas, kpl_data_manager.KPLLimitUpDataRecordManager.total_datas, yesterday_codes, block_info.get_before_blocks_dict()
-                                    if not current_limit_up_datas:
-                                        current_limit_up_datas = []
-                                    if not limit_up_record_datas:
-                                        limit_up_record_datas=[]
-
-
-                                    if CodePlateKeyBuyManager.is_need_cancel(code, limit_up_reasons.get(code),
-                                                                             current_limit_up_datas,
-                                                                             limit_up_record_datas,
-                                                                             yesterday_current_limit_up_codes,
-                                                                             before_blocks_dict):
-                                        l2_data_manager_new.L2TradeDataProcessor.cancel_buy(code, f"娑ㄥ仠鍘熷洜锛坽 limit_up_reasons.get(code)}锛変笉鏄�佸ぇ鎾ゅ崟", "鏉垮潡鎾�")
+                            try:
+                                # 鍒ゆ柇鏄惁涓嬪崟
+                                trade_state = trade_manager.CodesTradeStateManager().get_trade_state(code)
+                                if trade_state == trade_manager.TRADE_STATE_BUY_PLACE_ORDER or trade_state == trade_manager.TRADE_STATE_BUY_DELEGATED:
+                                    # 濮旀墭涓殑璁㈠崟锛屽垽鏂槸鍚﹂渶瑕佹挙鍗�
+                                    if not gpcode_manager.WantBuyCodesManager().is_in_cache(code):
+                                        yesterday_codes = kpl_data_manager.get_yesterday_limit_up_codes()
+                                        current_limit_up_datas, limit_up_record_datas, yesterday_current_limit_up_codes, before_blocks_dict = kpl_data_manager.KPLLimitUpDataRecordManager.latest_origin_datas, kpl_data_manager.KPLLimitUpDataRecordManager.total_datas, yesterday_codes, block_info.get_before_blocks_dict()
+                                        if not current_limit_up_datas:
+                                            current_limit_up_datas = []
+                                        if not limit_up_record_datas:
+                                            limit_up_record_datas=[]
+                                        if CodePlateKeyBuyManager.is_need_cancel(code, limit_up_reasons.get(code),
+                                                                                 current_limit_up_datas,
+                                                                                 limit_up_record_datas,
+                                                                                 yesterday_current_limit_up_codes,
+                                                                                 before_blocks_dict):
+                                            pass
+                                            # TODO 娴嬭瘯鏆傛椂娉ㄩ噴
+                                            # l2_data_manager_new.L2TradeDataProcessor.cancel_buy(code, f"娑ㄥ仠鍘熷洜锛坽 limit_up_reasons.get(code)}锛変笉鏄�佸ぇ鎾ゅ崟", "鏉垮潡鎾�")
+                            except Exception as e:
+                                logger_debug.exception(e)
                 kpl_data_manager.KPLLimitUpDataRecordManager.save_record(tool.get_now_date_str(), result_list_)
                 self.__kplDataManager.save_data(type_, result_list_)
 
@@ -536,7 +541,7 @@
                 self.__kplDataManager.save_data(type_, result_list)
         elif type_ == KPLDataType.LIMIT_UP.value:
             result_list = kpl_util.parseDaBanData(data["data"], kpl_util.DABAN_TYPE_LIMIT_UP)
-            threading.Thread(target=lambda: do_limit_up(result_list), daemon=True).start()
+            self.__data_process_thread_pool.submit(lambda: do_limit_up(result_list))
         elif type_ == KPLDataType.OPEN_LIMIT_UP.value:
             result_list = kpl_util.parseDaBanData(data["data"], kpl_util.DABAN_TYPE_OPEN_LIMIT_UP)
             if result_list:

--
Gitblit v1.8.0