L2总卖实现

Administrator

2023-10-26 97cc7f2d7428ea890c0a0ada76e5bffafd2463e4

 test/l2_trade_test.py

@@ -13,6 +13,7 @@
from code_attribute import big_money_num_manager, gpcode_manager
from db.redis_manager_delegate import RedisUtils
from l2.huaxin import huaxin_delegate_postion_manager
from l2.l2_sell_manager import L2MarketSellManager
from log_module import log, log_export, async_log_util
from trade.huaxin import huaxin_trade_api
from utils import tool
@@ -84,10 +85,10 @@
                except Exception as e:
                    pass

    @unittest.skip("跳过此单元测试")

    def test_trade(self):
        threading.Thread(target=async_log_util.run_sync, daemon=True).start()
        code = "002771"
        code = "600203"
        clear_trade_data(code)
        l2.l2_data_util.load_l2_data(code)
        total_datas = deepcopy(l2.l2_data_util.local_today_datas[code])
@@ -101,6 +102,11 @@
                data = total_datas[0].copy()
                data["index"] = i
                total_datas.insert(i, data)
        l2_market_datas = log_export.load_l2_market_data()
        l2_market_datas = l2_market_datas.get(code)
        if l2_market_datas:
            l2_market_datas.reverse()


        pos_list = log_export.get_l2_process_position(code)
        # pos_list.insert(108,(375,448))
@@ -170,8 +176,16 @@
                    break

            print("----------------处理位置", indexs)
            if indexs[0] >= 661:
                print("进入调试")
            for l2_m in l2_market_datas:
                if l2_m["dataTimeStamp"] < int(total_datas[indexs[0]]["val"]["time"].replace(":","")):
                    time_str = f"{l2_m['dataTimeStamp']}"
                    if time_str.startswith("9"):
                        time_str = "0" + time_str
                    time_str = time_str[:6]
                    time_str = f"{time_str[0:2]}:{time_str[2:4]}:{time_str[4:6]}"
                    L2MarketSellManager().set_current_total_sell_data(code,time_str,l2_m["totalAskVolume"]*l2_m["avgAskPrice"])
                    break

            l2.l2_data_manager_new.L2TradeDataProcessor.process_add_datas(code, total_datas[indexs[0]:indexs[1] + 1], 0,
                                                                          0)

@@ -192,6 +206,7 @@
        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 = "000628"
        KPLCodeJXBlockManager().load_jx_blocks(code, 17.96, 17.96)