L撤成交进度相关改进

Administrator

2024-06-13 cbe19ea6066a600cbd0b5110db5d43f8252d14a8

 third_data/data_server.py

@@ -8,12 +8,13 @@
import dask

from code_attribute.gpcode_manager import BlackListCodeManager, CodePrePriceManager
from l2.l2_transaction_data_manager import HuaXinBuyOrderManager
from log_module.log import logger_system, logger_debug, logger_kpl_limit_up
from utils import global_util, tool, data_export_util, init_data_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, transaction_progress
from l2.cancel_buy_strategy import HourCancelBigNumComputer
from cancel_strategy.s_l_h_cancel_strategy import HourCancelBigNumComputer, LCancelRateManager
from output.limit_up_data_filter import IgnoreCodeManager
from third_data import kpl_util, kpl_data_manager, kpl_api, block_info
from third_data.code_plate_key_manager import RealTimeKplMarketData, KPLPlateForbiddenManager, CodePlateKeyBuyManager
@@ -25,7 +26,7 @@
from urllib.parse import parse_qs
from output import code_info_output, limit_up_data_filter, output_util, kp_client_msg_manager

from trade import bidding_money_manager, trade_manager, l2_trade_util, trade_record_log_util
from trade import bidding_money_manager, trade_manager, l2_trade_util, trade_record_log_util, trade_constant
import concurrent.futures

# 禁用http.server的日志输出
@@ -306,7 +307,7 @@
            name = ps_dict.get('name')
            date = ps_dict.get('date')
            try:
                data = code_info_output.get_output_params(code, self.__jingxuan_cache_dict, self.__industry_cache_dict, date)
                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}"

@@ -320,7 +321,16 @@
                logger_debug.exception(e)
                logging.exception(e)

            # 获取评分信息
        elif url.path == "/get_trade_records":
            # 获取挂撤信息
            ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
            code = ps_dict['code']
            date = ps_dict.get('date')
            local_today_datas = log_export.load_l2_from_log(date)
            total_datas = local_today_datas.get(code)
            trade_info = code_info_output.load_trade_record(code, total_datas, date)
            response_data = json.dumps({"code": 0, "data": {"open_limit_up": trade_info[0], "records": trade_info[2]}})

        elif url.path == "/get_l2_cant_buy_reasons":
            # 获取L2没买的原因
            ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
@@ -346,10 +356,18 @@
                ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
                code = ps_dict['code']
                date = ps_dict.get('date')
                logger_debug.info(f"get_l2_datas:{code}  {date}")
                datas = data_export_util.get_l2_datas(code, l2_data_util.local_today_datas.get(code), date=date)
                time_str = ps_dict.get('time')
                total_datas = l2_data_util.local_today_datas.get(code)
                if date or time_str:
                    total_datas = None
                else:
                    date = tool.get_now_date_str()
                delegate_datas = data_export_util.get_l2_datas(code, total_datas, date=date)
                transaction_datas = data_export_util.get_l2_transaction_datas(code, date=date)
                code_name = gpcode_manager.get_code_name(code)
                response_data = json.dumps({"code": 0, "data": {"code": code, "code_name": code_name, "data": datas}})
                response_data = json.dumps({"code": 0, "data": {"code": code, "code_name": code_name,
                                                                "data": {"delegates": delegate_datas,
                                                                         "transactions": transaction_datas}}})
            except Exception as e:
                logger_debug.exception(e)
        elif url.path == "/get_trade_progress":
@@ -357,16 +375,26 @@
            ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
            code = ps_dict['code']
            trade_progress, is_default = transaction_progress.TradeBuyQueue().get_traded_index(code)
            # 获取正在成交, 计算成交进度
            dealing_info = HuaXinBuyOrderManager.get_dealing_order_info(code)
            percent = 100
            if dealing_info:
                total_datas = l2_data_util.local_today_datas.get(code)
                if str(total_datas[trade_progress]['val']["orderNo"]) == str(dealing_info[0]):
                    percent = int(dealing_info[1] / total_datas[trade_progress]['val']['num'])
            response_data = json.dumps(
                {"code": 0, "data": {"trade_progress": trade_progress, "is_default": is_default}})
                {"code": 0, "data": {"trade_progress": trade_progress, "is_default": is_default, "percent": percent}})
        elif url.path == "/get_l_cancel_datas":
            # 最新的l撤数据
            ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
            code = ps_dict['code']
            date = ps_dict.get('date')
            if not date:
                date = tool.get_now_date_str()
            buy_single_index = ps_dict.get('buy_single_index')
            if buy_single_index is not None:
                buy_single_index = int(buy_single_index)
            records = code_info_output.load_trade_record_cancel_watch_indexes(code)
            records = code_info_output.load_trade_record_cancel_watch_indexes(code, date=date)
            # 获取最新的L上与L下
            records.reverse()
            up_indexes = []
@@ -472,20 +500,20 @@
            for d in record_limit_up_datas:
                if kpl_util.filter_block(d[2]) != plate:
                    continue
                # 代码,名称,涨停时间,是否炸板,是否想买,是否已经下过单
                # 代码,名称,涨停时间,是否炸板,是否想买,是否已经下过单,涨停时间,自由流通市值,是否在黑名单里面
                codes_info.append(
                    [d[3], d[4], tool.to_time_str(int(d[5])), 1 if d[3] not in now_limit_up_codes else 0, 0, 0, d[12],
                     output_util.money_desc(d[13]), 1])
                     output_util.money_desc(d[13]), 1, 1 if l2_trade_util.is_in_forbidden_trade_codes(d[3]) else 0])

            for d in record_limit_up_datas:
                if kpl_util.filter_block(d[2]) == plate:
                    continue
                if plate not in [kpl_util.filter_block(k) for k in d[6].split("、")]:
                    continue
                # 代码,名称,涨停时间,是否炸板,是否想买,是否已经下过单
                # 代码,名称,涨停时间,是否炸板,是否想买,是否已经下过单,涨停时间,自由流通市值,是否在黑名单里面
                codes_info.append(
                    [d[3], d[4], tool.to_time_str(int(d[5])), 1 if d[3] not in now_limit_up_codes else 0, 0, 0, d[12],
                     output_util.money_desc(d[13]), 0])
                     output_util.money_desc(d[13]), 0, 1 if l2_trade_util.is_in_forbidden_trade_codes(d[3]) else 0])

            codes_info.sort(key=lambda x: x[2])
            # 查询是否为想买单
@@ -494,7 +522,7 @@
                code_info[4] = 1 if code_info[0] in want_codes else 0
                # 获取代码状态
                if trade_manager.CodesTradeStateManager().get_trade_state_cache(
                        code_info[0]) != trade_manager.TRADE_STATE_NOT_TRADE:
                        code_info[0]) != trade_constant.TRADE_STATE_NOT_TRADE:
                    code_info[5] = 1

            response_data = json.dumps({"code": 0, "data": codes_info})
@@ -506,7 +534,7 @@
            statistic = {}
            for result in results:
                for c in result[1]:
                    if not tool.is_shsz_code(c):
                    if not tool.is_can_buy_code(c):
                        continue
                    if code and code != c:
                        continue
@@ -517,8 +545,15 @@
            statistic_list = [(k, statistic[k]) for k in statistic]
            statistic_list.sort(key=lambda x: x[1], reverse=True)
            fresults = []
            limit_up_records = KPLLimitUpDataRecordManager.list_all_cache(tool.get_now_date_str())
            limit_up_count_dict = {}
            if limit_up_records:
                for d in limit_up_records:
                    limit_up_count_dict[d[3]] = d[12]

            for x in statistic_list:
                fresults.append((x[0], gpcode_manager.get_code_name(x[0]), x[1]))
                fresults.append((x[0], gpcode_manager.get_code_name(x[0]), x[1], limit_up_count_dict.get(x[0])))

            fresults = fresults[:30]
            response_data = json.dumps({"code": 0, "data": fresults})
        elif url.path == "/get_h_cancel_data":
@@ -531,7 +566,7 @@
                    total_datas = l2_data_util.local_today_datas.get(code)

                trade_state = trade_manager.CodesTradeStateManager().get_trade_state_cache(code)
                if trade_state == trade_manager.TRADE_STATE_BUY_PLACE_ORDER or trade_state == trade_manager.TRADE_STATE_BUY_DELEGATED or trade_state == trade_manager.TRADE_STATE_BUY_SUCCESS:
                if trade_state == trade_constant.TRADE_STATE_BUY_PLACE_ORDER or trade_state == trade_constant.TRADE_STATE_BUY_DELEGATED or trade_state == trade_constant.TRADE_STATE_BUY_SUCCESS:
                    hcancel_datas_dict, cancel_indexes_set = HourCancelBigNumComputer().get_watch_index_dict(code)
                    # 根据日志读取实时的计算数据
                    h_cancel_latest_compute_info = log_export.get_h_cancel_compute_info(code)
@@ -673,61 +708,63 @@

    def __process_kpl_data(self, data_origin):
        def do_limit_up(result_list_):
            if result_list_:
                # 保存涨停时间
                codes_set = set()
                limit_up_reasons = {}
                for d in result_list_:
                    code = d[0]
                    limit_up_reasons[code] = d[5]
                    codes_set.add(code)
                    if tool.is_shsz_code(code):
                        limit_up_time = time.strftime("%H:%M:%S", time.localtime(d[2]))
                        code_price_manager.Buy1PriceManager().set_limit_up_time(code, limit_up_time)
                add_codes = codes_set - self.__latest_limit_up_codes_set
                self.__latest_limit_up_codes_set = codes_set
            try:
                if result_list_:
                    # 保存涨停时间
                    codes_set = set()
                    limit_up_reasons = {}
                    for d in result_list_:
                        code = d[0]
                        limit_up_reasons[code] = d[5]
                        codes_set.add(code)
                        if tool.is_can_buy_code(code):
                            limit_up_time = time.strftime("%H:%M:%S", time.localtime(d[2]))
                            code_price_manager.Buy1PriceManager().set_limit_up_time(code, limit_up_time)
                    add_codes = codes_set - self.__latest_limit_up_codes_set
                    self.__latest_limit_up_codes_set = codes_set

                if limit_up_reasons:
                    # 统计涨停原因的票的个数
                    limit_up_reason_code_dict = {}
                    for code in limit_up_reasons:
                        b = limit_up_reasons[code]
                        if b not in limit_up_reason_code_dict:
                            limit_up_reason_code_dict[b] = set()
                        limit_up_reason_code_dict[b].add(code)
                    cancel_buy_strategy.LCancelRateManager.set_block_limit_up_count(limit_up_reason_code_dict)
                    if limit_up_reasons:
                        # 统计涨停原因的票的个数
                        limit_up_reason_code_dict = {}
                        for code in limit_up_reasons:
                            b = limit_up_reasons[code]
                            if b not in limit_up_reason_code_dict:
                                limit_up_reason_code_dict[b] = set()
                            limit_up_reason_code_dict[b].add(code)
                        LCancelRateManager.set_block_limit_up_count(limit_up_reason_code_dict)

                if add_codes:
                    for code in add_codes:
                        # 根据涨停原因判断是否可以买
                        if tool.is_shsz_code(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):
                                        #     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_)

                    if add_codes:
                        for code in add_codes:
                            # 根据涨停原因判断是否可以买
                            if tool.is_can_buy_code(code):
                                try:
                                    # 判断是否下单
                                    trade_state = trade_manager.CodesTradeStateManager().get_trade_state(code)
                                    if trade_state == trade_constant.TRADE_STATE_BUY_PLACE_ORDER or trade_state == trade_constant.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)}）不是老大撤单",
                                            #                                                         "板块撤")
                                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_)
            except Exception as e:
                logger_debug.exception(e)
        # 将"概念"二字替换掉
        data = data_origin
        type_ = data["type"]
@@ -815,7 +852,7 @@
def run(addr, port):
    # 运行看盘消息采集
    # kp_client_msg_manager.run_capture()
    kpl_data_manager.run_pull_task()
    kpl_data_manager.PullTask.run_pull_task()

    handler = DataServer
    # httpd = socketserver.TCPServer((addr, port), handler)
@@ -829,25 +866,10 @@


if __name__ == "__main__":
    code = "002676"
    buy_single_index = 716
    records = code_info_output.load_trade_record_cancel_watch_indexes(code)
    # 获取最新的L上与L下
    records.reverse()
    up_indexes = []
    down_indexes = []
    for r in records:
        if buy_single_index and buy_single_index != r[1]:
            continue
        if r[0] == trade_record_log_util.CancelWatchIndexesInfo.CANCEL_TYPE_L_UP:
            up_indexes = r[2]
            break
    for r in records:
        if buy_single_index and buy_single_index != r[1]:
            continue
        if r[0] == trade_record_log_util.CancelWatchIndexesInfo.CANCEL_TYPE_L_DOWN:
            down_indexes = r[2]
            break
    code = "603359"
    records = code_info_output.load_trade_record_cancel_watch_indexes(code,
                                                                      trade_record_log_util.CancelWatchIndexesInfo.CANCEL_TYPE_H)
    print(records)

    response_data = json.dumps(
        {"code": 0, "data": {"up": up_indexes, "down": down_indexes}})
    # data = code_info_output.get_output_params(code, self.__jingxuan_cache_dict, self.__industry_cache_dict,
    #                                           trade_record_date=date)