测试激进买1手

Administrator

2024-04-07 0d5e54227474721a5c25704a9e285e8a6fa1ac8b

测试激进买1手

 constant.py

@@ -3,6 +3,9 @@

from huaxin_client import constant

# 所有的票激进买
ALL_ACTIVE_BUY = True

TEST = False
# 是否允许交易
TRADE_ENABLE = True
@@ -164,5 +167,5 @@
    TRADE_ENABLE = True

# 最大的代码价格
MAX_CODE_PRICE = 30
MAX_SUBSCRIPT_CODE_PRICE = 30
MAX_CODE_PRICE = 15 if ALL_ACTIVE_BUY else 30
MAX_SUBSCRIPT_CODE_PRICE = 15 if ALL_ACTIVE_BUY else 30

 l2/l2_data_manager_new.py

@@ -1056,6 +1056,9 @@
        if zyltgb_as_yi < zylt_threshold_as_yi[0]:
            return False, True, f"{zylt_threshold_as_yi[0]}亿以下的都不买（{zyltgb_as_yi}）"

        if constant.ALL_ACTIVE_BUY:
            return True, False, "买所有"

        # 最优市值
        is_best_zylt = True if zylt_threshold_as_yi[4] <= zyltgb_as_yi <= zylt_threshold_as_yi[5] else False

@@ -1644,7 +1647,7 @@
        # 获取板块是否可以激进买
        # 可买的板块, 是否独苗, 消息, 可买的强势板块, 关键词, 激进买的板块
        can_buy_result = CodePlateKeyBuyManager.can_buy(code)
        if can_buy_result and can_buy_result[0] and can_buy_result[5]:
        if (can_buy_result and can_buy_result[0] and can_buy_result[5]) or constant.ALL_ACTIVE_BUY:
            # 有可买板块，有激进买板块
            # 第一步： 计算总卖额
            threshold_money = refer_sell_data[1]

 test/l2_trade_test.py

@@ -10,6 +10,7 @@
from copy import deepcopy
from unittest import mock

import constant
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
@@ -86,11 +87,12 @@
                except Exception as e:
                    pass

    @unittest.skip("跳过此单元测试")
    # @unittest.skip("跳过此单元测试")
    def test_trade(self):
        constant.TEST=True
        trade_manager.TradeStateManager().open_buy()
        threading.Thread(target=async_log_util.run_sync, daemon=True).start()
        code = "603135"
        code = "000691"
        clear_trade_data(code)
        l2.l2_data_util.load_l2_data(code)
        total_datas = deepcopy(l2.l2_data_util.local_today_datas[code])
@@ -138,8 +140,8 @@
        # pos_list.insert(84, (516, 532))

        # hook
        tool.get_now_time_str = mock.Mock(return_value="09:35:00")
        CodePlateKeyBuyManager.can_buy = mock.Mock(return_value=(["测试"], False, "", []))
        # tool.get_now_time_str = mock.Mock(return_value="09:35:00")


        # 获取交易进度
        trade_progress_list, buy_queues = log_export.get_trade_progress(code)
@@ -233,7 +235,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("跳过此单元测试")
    @unittest.skip("跳过此单元测试")
    def test_block(self):
        code = "002801"
        # KPLCodeJXBlockManager().load_jx_blocks(code, 23.52,23.62,

 test/test_sell.py

@@ -17,14 +17,17 @@


def test_sell_order_info():
    code = "003001"
    code = "002734"
    l2_data_util.load_l2_data(code, force=True)
    date = tool.get_now_date_str()
    data_map = log_export.load_huaxin_transaction_map(date=date)
    datas = data_map.get(code)
    for i in range(len(datas)):
        d = datas[i]
        big_sell_order_info = HuaXinSellOrderStatisticManager.add_transaction_datas(code, d, buy_exec_index=57)
        big_sell_order_info = HuaXinSellOrderStatisticManager.add_transaction_datas(code, d)
        if d[-1][7] >= 3911165:
            sell_infos = HuaXinSellOrderStatisticManager.get_latest_transaction_datas(code, 3910694,"09:35:27")
            print( sum([x[1]*x[2] for x in sell_infos]))
        if big_sell_order_info[0] < 50 * 10000:
            continue
        print(i, sum([x[1] * x[2] for x in big_sell_order_info[1]]), big_sell_order_info)