| | |
|---|
| | | except Exception as e: |
|---|
| | | pass |
|---|
| | | |
|---|
| | | @unittest.skip("跳过此单元测试") |
|---|
| | | # @unittest.skip("跳过此单元测试") |
|---|
| | | def test_trade(self): |
|---|
| | | threading.Thread(target=async_log_util.run_sync, daemon=True).start() |
|---|
| | | code = "603178" |
|---|
| | | code = "002771" |
|---|
| | | clear_trade_data(code) |
|---|
| | | l2.l2_data_util.load_l2_data(code) |
|---|
| | | total_datas = deepcopy(l2.l2_data_util.local_today_datas[code]) |
|---|
| | | l2.l2_data_util.local_today_num_operate_map.get(code).clear() |
|---|
| | | l2.l2_data_util.local_today_buyno_map.get(code).clear() |
|---|
| | | l2.l2_data_util.local_today_canceled_buyno_map.get(code).clear() |
|---|
| | | |
|---|
| | | |
|---|
| | | if total_datas[0]["index"] > 0: |
|---|
| | | # 拼接数据 |
|---|
| | | for i in range(0, total_datas[0]["index"]): |
|---|
| | |
|---|
| | | l2.l2_data_manager_new.L2TradeDataProcessor.process_add_datas(code, total_datas[indexs[0]:indexs[1] + 1], 0, |
|---|
| | | 0) |
|---|
| | | |
|---|
| | | @unittest.skip("跳过此单元测试") |
|---|
| | | def test_place_order(self): |
|---|
| | | code = "002241" |
|---|
| | | l2.l2_data_util.load_l2_data(code) |
|---|
