接口修改/扫入板块修改

Administrator

2024-10-14 46789f04b9d1525c38bc64e8ecd244e16661b71b

 test/l2_trade_test.py

@@ -18,13 +18,15 @@
from l2.l2_transaction_data_manager import HuaXinSellOrderStatisticManager
from log_module import log, log_export, async_log_util
from trade.huaxin import huaxin_trade_api
from trade.radical_buy_data_manager import RedicalBuyDataManager
from utils import tool, init_data_util
from db import redis_manager_delegate as redis_manager
from l2 import l2_log, l2_data_manager, transaction_progress, l2_data_manager_new, l2_transaction_data_processor, \
    cancel_buy_strategy
from l2.transaction_progress import TradeBuyQueue
from third_data import kpl_util, kpl_data_manager, block_info
from third_data.code_plate_key_manager import LimitUpCodesPlateKeyManager, CodePlateKeyBuyManager
from third_data.code_plate_key_manager import LimitUpCodesPlateKeyManager, CodePlateKeyBuyManager, \
    RadicalBuyBlockManager
from third_data.kpl_data_manager import KPLDataManager
from trade import trade_data_manager, current_price_process_manager, l2_trade_util, trade_manager
import l2.l2_data_manager_new, l2.l2_data_manager, l2.l2_data_util, l2.cancel_buy_strategy
@@ -235,36 +237,52 @@
        l2.l2_data_util.local_today_datas[code] = total_datas
        l2.l2_data_util.load_num_operate_map(l2.l2_data_util.local_today_num_operate_map, code, total_datas, True)




    # @unittest.skip("跳过此单元测试")
    def test_block(self):
        code = "001376"
        # KPLCodeJXBlockManager().load_jx_blocks(code, 23.52,23.62,
        #                                        kpl_data_manager.KPLLimitUpDataRecordManager.get_current_reasons())
        #
        block_info.init_code(code)
        codes_str = "603778"
        codes = codes_str.split(",")  # ["002889", "300337", "001298", "002771"]
        kpl_data_manager.KPLLimitUpDataRecordManager.load_total_datas()
        latest_current_limit_up_records = kpl_data_manager.get_latest_current_limit_up_records()
        kpl_data_manager.KPLLimitUpDataRecordManager.save_record(tool.get_now_date_str(),
                                                                 kpl_data_manager.KPLDataManager.get_data(
                                                                     kpl_util.KPLDataType.LIMIT_UP))
        limit_up_data = kpl_data_manager.KPLLimitUpDataRecordManager.record_code_dict.get(code)
        if limit_up_data:
            limit_up_time = tool.to_time_str(limit_up_data[2])
        CodePlateKeyBuyManager.update_can_buy_blocks(code,
                                                     kpl_data_manager.KPLLimitUpDataRecordManager.latest_origin_datas,
                                                     kpl_data_manager.KPLLimitUpDataRecordManager.total_datas,
                                                     latest_current_limit_up_records,
                                                     block_info.get_before_blocks_dict(),
                                                     kpl_data_manager.KPLLimitUpDataRecordManager.get_current_limit_up_reason_codes_dict())
        can_buy_result = CodePlateKeyBuyManager.can_buy(code)
        print(can_buy_result)
        if can_buy_result:
            if can_buy_result[0]:
                blocks = ",".join([f"{x[0]}-{x[1] + 1}({x[2]}&{x[3] - x[2]})" for x in can_buy_result[0]])
                print(blocks)
        for code in codes:
            # KPLCodeJXBlockManager().load_jx_blocks(code, 23.52,23.62,
            #                                        kpl_data_manager.KPLLimitUpDataRecordManager.get_current_reasons())
            #
            block_info.init_code(code)
            # latest_current_limit_up_records = kpl_data_manager.get_latest_current_limit_up_records()

            init_data_util.re_set_price_pre(code, True)

            limit_up_data = kpl_data_manager.KPLLimitUpDataRecordManager.record_code_dict.get(code)
            if limit_up_data:
                limit_up_time = tool.to_time_str(limit_up_data[2])
            RadicalBuyBlockManager.set_current_limit_up_datas(
                kpl_data_manager.KPLLimitUpDataRecordManager.latest_origin_datas)
            # CodePlateKeyBuyManager.update_can_buy_blocks(code,
            #                                              kpl_data_manager.KPLLimitUpDataRecordManager.latest_origin_datas,
            #                                              kpl_data_manager.KPLLimitUpDataRecordManager.total_datas,
            #                                              latest_current_limit_up_records,
            #                                              block_info.get_before_blocks_dict(),
            #                                              kpl_data_manager.KPLLimitUpDataRecordManager.get_current_limit_up_reason_codes_dict())
            # can_buy_result = CodePlateKeyBuyManager.can_buy(code)
            # print(can_buy_result)
            # if can_buy_result:
            #     if can_buy_result[0]:
            #         blocks = ",".join([f"{x[0]}-{x[1] + 1}({x[2]}&{x[3] - x[2]})" for x in can_buy_result[0]])
            #         print(blocks)

            yesterday_codes = kpl_data_manager.get_yesterday_limit_up_codes()
            if yesterday_codes is None:
                yesterday_codes = set()
            result = RadicalBuyBlockManager.is_radical_buy(code, yesterday_codes)
            print(code, result)
            if result[0]:
                can_buy_result = RedicalBuyDataManager.can_buy(code)
                if can_buy_result[0]:
                    print("可以买", code, result)
                else:
                    print("不可以买", code, can_buy_result)

        # l2.l2_data_manager_new.L2TradeDataProcessor.can_buy_first(code, None)