将华鑫订单交易成功独立出来处理

Administrator

2023-09-25 c0bcfe746b97bc126636a658b1f01fc6a51f9f95

 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,23 +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_)

@@ -534,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: