| | |
|---|
| | | from l2 import l2_log, l2_data_manager, transaction_progress, safe_count_manager |
|---|
| | | from l2.safe_count_manager import BuyL2SafeCountManager |
|---|
| | | from l2.transaction_progress import TradeBuyQueue |
|---|
| | | from third_data import kpl_util |
|---|
| | | from third_data.code_plate_key_manager import RealTimeKplMarketData, LimitUpCodesPlateKeyManager |
|---|
| | | from third_data.kpl_data_manager import KPLDataManager |
|---|
| | | from trade import trade_data_manager, l2_trade_factor |
|---|
| | | from trade.trade_queue_manager import THSBuy1VolumnManager |
|---|
| | | import l2.l2_data_manager_new, l2.l2_data_manager, l2.l2_data_util, l2.cancel_buy_strategy |
|---|
| | |
|---|
| | | decimal.Decimal("0.00")) |
|---|
| | | # 获取执行位时间 |
|---|
| | | exec_time = None |
|---|
| | | buy_single_index, buy_exec_index, compute_index, num, count, max_num_set,volume_rate = l2_data_manager.TradePointManager.get_buy_compute_start_data( |
|---|
| | | buy_single_index, buy_exec_index, compute_index, num, count, max_num_set, volume_rate = l2_data_manager.TradePointManager.get_buy_compute_start_data( |
|---|
| | | code) |
|---|
| | | if buy_exec_index: |
|---|
| | | try: |
|---|
| | |
|---|
| | | |
|---|
| | | # @unittest.skip("跳过此单元测试") |
|---|
| | | def test_trade(self): |
|---|
| | | code = "603819" |
|---|
| | | code = "002989" |
|---|
| | | clear_trade_data(code) |
|---|
| | | l2.l2_data_util.load_l2_data(code) |
|---|
| | | total_datas = deepcopy(l2.l2_data_util.local_today_datas[code]) |
|---|
| | |
|---|
| | | |
|---|
| | | # 获取交易进度 |
|---|
| | | trade_progress_list, buy_queues = log.get_trade_progress(code) |
|---|
| | | jingxuan_ranks = KPLDataManager().get_from_file(kpl_util.KPLDataType.JINGXUAN_RANK, tool.get_now_date_str()) |
|---|
| | | industry_ranks = KPLDataManager().get_from_file(kpl_util.KPLDataType.INDUSTRY_RANK, tool.get_now_date_str()) |
|---|
| | | RealTimeKplMarketData().set_top_5_reasons(jingxuan_ranks) |
|---|
| | | RealTimeKplMarketData().set_top_5_industry(industry_ranks) |
|---|
| | | |
|---|
| | | LimitUpCodesPlateKeyManager().set_today_limit_up(KPLDataManager().get_from_file(kpl_util.KPLDataType.LIMIT_UP, tool.get_now_date_str())) |
|---|
| | | |
|---|
| | | for indexs in pos_list: |
|---|
| | | l2_log.threadIds[code] = mock.Mock( |
|---|
| | |
|---|
| | | |
|---|
| | | @unittest.skip("跳过此单元测试") |
|---|
| | | def test_h_cancel(self): |
|---|
| | | code = "002870" |
|---|
| | | code = "600540" |
|---|
| | | l2.l2_data_util.load_l2_data(code) |
|---|
| | | total_datas = l2.l2_data_util.local_today_datas.get(code) |
|---|
| | | total_datas = total_datas[:899] |
|---|
