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

Administrator

2023-10-24 60d02096d9dc4bc9cb54f8cf81becd10a9dd0c61

 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()