178 lines
8.1 KiB
Python
178 lines
8.1 KiB
Python
"""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()
|