"""ZT 归属过滤：非本策略成交通知不得进入账本，也不得中断策略。""" import logging import unittest from sdk import DealItem from strategy.zt import boot from strategy.zt.ownership import OWNED_PREFIX, owned_deals, owns_local_order_id from strategy.zt.rounds import RoundStore, start_round from tests.zt_harness import Fixture TODAY = '2026-09-15' class OwnershipPredicateTests(unittest.TestCase): def test_only_local_order_ids_generated_by_zt_are_owned(self): owned = ['zt-base-8e9da97a42e957408489', 'zt-added-9239083181eb39712994', 'zt-entry-0b10c994b8242682983f', 'zt-exit-1'] foreign = ['', None, ' ', 'TREN-BUY-1', 'MORN-2', 'IPO-abc', 'DCM-3', 'zt', 'azt-base-1', 'ztbase-1'] for value in owned: with self.subTest(value=value): self.assertTrue(owns_local_order_id(value)) for value in foreign: with self.subTest(value=value): self.assertFalse(owns_local_order_id(value)) self.assertEqual(OWNED_PREFIX, 'zt-') def test_owned_deals_splits_and_counts(self): def deal(remark): return DealItem(stock_code='600000.SH', order_sys_id=remark or 'none', remark=remark) deals = [deal('zt-entry-a'), deal(''), deal('TREN-BUY-1'), deal('zt-exit-b')] owned, ignored = owned_deals(deals) self.assertEqual([d.get_local_order_id for d in owned], ['zt-entry-a', 'zt-exit-b']) self.assertEqual(ignored, 2) class ForeignDealIsolationTests(unittest.TestCase): """手工单与其他策略单既不进轮次，也不影响本策略的判断。""" def setUp(self): self.fx = Fixture() self.addCleanup(self.fx.cleanup) logging.disable(logging.CRITICAL) self.addCleanup(logging.disable, logging.NOTSET) self.fx.hold('600000.SH', volume=1000, price=10.0) self.fx.quote('600000.SH', 10.0) def run_tick(self): self.fx.tick() return self.fx.store.get('600000.SH') def test_manual_deal_without_remark_neither_raises_nor_blocks(self): self.fx.open_round() self.fx.deals = [DealItem(stock_code='600000.SH', order_sys_id='m1', remark='', offset_flag=48, volume=100, price=9.0, trade_amount=900.0, trade_date='20260915', trade_time='100000')] item = self.run_tick() # 旧实现在这里抛 IntegrityError self.assertEqual(item.entry_filled_qty, 0) self.assertEqual(item.outcome, 'aborted') # 开仓腿无成交且已不在途 self.assertEqual(self.fx.placed, []) def test_foreign_strategy_deal_cannot_touch_the_round(self): self.fx.open_round() self.fx.deals = [DealItem(stock_code='600000.SH', order_sys_id='t1', remark='TREN-BUY-9|trend', offset_flag=48, volume=500, price=20.0, trade_amount=10000.0, trade_date='20260915', trade_time='100000')] item = self.run_tick() # 旧实现把外部买入当补仓写进 added 桶 self.assertEqual(item.entry_filled_qty, 0) self.assertEqual(item.entry_amount, 0.0) self.assertEqual(item.base_qty, 1000) def test_owned_deals_are_still_counted(self): self.fx.open_round() self.fx.deals = [self.fx.deal('zt-entry-1', 300, 9.0)] item = self.run_tick() self.assertEqual(item.entry_filled_qty, 300) self.assertEqual(item.entry_avg_price, 9.0) # 现价 10.0 对买入均价 9.0 已超过一个网格步长，同一轮 tick 内即挂出卖单。 self.assertEqual(item.phase, 'CLOSING') self.assertEqual([p['stock_code'] for p in self.fx.placed], ['600000.SH']) self.assertTrue(self.fx.placed[0]['order_id'].startswith('zt-exit-')) def test_mixed_batch_keeps_only_owned_deals(self): self.fx.open_round() self.fx.deals = [ self.fx.deal('zt-entry-1', 100, 9.0, sys_id='own'), DealItem(stock_code='600000.SH', order_sys_id='manual', remark='', offset_flag=48, volume=100, price=9.0, trade_amount=900.0, trade_date='20260915', trade_time='100000'), DealItem(stock_code='600000.SH', order_sys_id='trend', remark='TREN-BUY-1', offset_flag=48, volume=100, price=9.0, trade_amount=900.0, trade_date='20260915', trade_time='100000'), ] item = self.run_tick() self.assertEqual(item.entry_filled_qty, 100) self.assertEqual(len(item.seen_deal_ids), 1) def test_repeated_ticks_never_double_count(self): self.fx.open_round() self.fx.deals = [self.fx.deal('zt-entry-1', 300, 9.0)] self.assertEqual(self.run_tick().entry_filled_qty, 300) self.assertEqual(self.run_tick().entry_filled_qty, 300) class RunOnceResilienceTests(unittest.TestCase): """任何单点失败都只能跳过本轮，不能打断唯一的交易定时线程。""" def setUp(self): self.fx = Fixture() self.addCleanup(self.fx.cleanup) logging.disable(logging.CRITICAL) self.addCleanup(logging.disable, logging.NOTSET) self.fx.hold('600000.SH', volume=1000, price=10.0) self.fx.quote('600000.SH', 10.0) def test_snapshot_failure_skips_the_round_quietly(self): from unittest.mock import patch with patch.object(boot, 'trading_time', return_value=True): self.fx.client.portfolio.side_effect = RuntimeError('api down') boot.RunOnce(self.fx.run, self.fx.store, []) self.assertEqual(self.fx.placed, []) def test_round_advance_failure_skips_trading(self): from unittest.mock import patch with patch.object(boot, 'trading_time', return_value=True), \ patch.object(boot, '_advance_rounds', side_effect=RuntimeError('broken')): boot.RunOnce(self.fx.run, self.fx.store, []) self.assertEqual(self.fx.placed, []) def test_market_data_failure_skips_trading(self): from unittest.mock import patch with patch.object(boot, 'trading_time', return_value=True): self.fx.client.full_tick.side_effect = RuntimeError('no ticks') boot.RunOnce(self.fx.run, self.fx.store, []) self.assertEqual(self.fx.placed, []) def test_startup_state_failure_closes_client_without_raising(self): from types import SimpleNamespace as NS from unittest.mock import patch client = self.fx.client with patch.object(boot.config, 'account_config', self.fx.account_cfg), \ patch.object(boot.config, 'global_config', NS(qmt_data_dir=str(self.fx.path), qmt_base_url='u', qmt_token='', api_host='u')), \ patch.object(boot, 'Client', return_value=client), \ patch.object(boot, '_open_store', side_effect=RuntimeError('disk')): boot.StartZT() # 不抛异常 client.close.assert_called_once() def test_corrupt_state_is_backed_up_and_rebuilt(self): from types import SimpleNamespace as NS from unittest.mock import patch path = self.fx.rounds_path path.write_text('{not json', encoding='utf-8') client = self.fx.client with patch.object(boot.config, 'account_config', self.fx.account_cfg), \ patch.object(boot.config, 'global_config', NS(qmt_data_dir=str(self.fx.path), qmt_base_url='u', qmt_token='', api_host='u')) , \ patch.object(boot, 'Client', return_value=client), \ patch.object(boot, 'init_signals', return_value=[]), \ patch.object(boot.time, 'localtime', return_value=NS(tm_hour=15, tm_min=0, tm_sec=0)): boot.StartZT() self.assertTrue(path.with_name(path.name + '.corrupt').is_file()) self.assertEqual(RoundStore(path).rounds, {}) client.close.assert_called_once() if __name__ == '__main__': unittest.main()