"""本地：python scripts/unit_test_guard.py scripts/test_batch41_units.py。配置隔离、禁止联网。""" import unit_test_guard import os import sys import unittest from datetime import datetime, timedelta from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from test_wiring import install_fakes from app.repo import pms_repo, qmt_repo from app.services import executor, dispatcher, proposal_service, signal_service from app.core import signal_rules CODE = '002518.SZ' def ins(fake, iid, action, side, qty=600, done=0): fake.insert_instruction(instruction_id=iid, origin_type='proposal', origin_id=None, ts_code=CODE, action=action, side=side, qty=qty, status='DISPATCHED', progress={'children': [], 'deadline': '2026-10-21', 'is_command': False}) fake.instructions[iid]['exec_qty'] = done return fake.instructions[iid] class ConflictsTest(unittest.TestCase): def setUp(self): self.fake=install_fakes(params={'PMS_DISPATCH_MODE':'ws'}) self.fake.qmt.set_channel() def test_old_partial_buy_stopped_before_exit(self): from app.services import execution_conflicts as c buy=ins(self.fake,'old','OPEN','buy',600,400) sell=ins(self.fake,'exit','EXIT','sell',400,200) result=c.enforce_exit(sell) self.assertTrue(result['ok']) self.assertEqual(buy['status'],'CANCELLED') self.assertEqual((buy['qty'],buy['exec_qty']),(600,400)) self.assertEqual(sell['exec_qty'],200) n=len(self.fake.ledger) c.enforce_exit(sell) self.assertEqual(len(self.fake.ledger),n) def test_trim_pauses_without_cancelling_buy(self): from app.services import execution_conflicts as c buy=ins(self.fake,'old','OPEN','buy') sell=ins(self.fake,'trim','TRIM','sell') self.assertTrue(c.buy_blocker(CODE)) c.enforce_exit(sell) self.assertEqual(buy['status'],'DISPATCHED') sell['status']='CONFIRMED' self.assertFalse(c.buy_blocker(CODE)) def test_t0_pair_is_not_exit(self): from app.services import execution_conflicts as c ins(self.fake,'t0','T0_ROUND','sell') self.assertFalse(c.buy_blocker(CODE)) def test_dispatch_rejects_buy_with_active_sell(self): ins(self.fake,'exit','EXIT','sell',400) result=dispatcher.dispatch(instruction_id='buy_D01',parent_id='buy',ts_code=CODE, side='buy',qty=200,limit_price=28,valid_until=datetime.now()+timedelta(minutes=10)) self.assertFalse(result['ok']) self.assertEqual(self.fake.qmt.orders,{}) def test_lookup_failure_cannot_dispatch_buy(self): with patch.object(pms_repo,'list_instructions',side_effect=RuntimeError('source down')): r=dispatcher.dispatch(instruction_id='buy_D01',ts_code=CODE,side='buy',qty=200,limit_price=28) self.assertFalse(r['ok']) self.assertEqual(self.fake.qmt.orders,{}) def test_pending_sell_blocks_new_buy_proposals(self): self.fake.insert_proposal(proposal_id='sell',ts_code=CODE,action='EXIT',qty=200, hard_numbers={},expire_at=datetime.now()+timedelta(hours=1)) keys=proposal_service._inflight_keys() for action in ('OPEN','ADD','FILL','DCA'): self.assertIn((CODE,action),keys) def test_signal_reason_preserved(self): parsed=signal_rules.parse_intraday({'ts_code':CODE,'action':'SELL','confidence':'0.95', 'audit_reason':'支撑失守，资金流出','component_scores':'{"reason":"另一份理由"}'}) self.assertEqual(parsed['reason'],'支撑失守，资金流出') parsed=signal_rules.parse_intraday({'component_scores':'{"reason":"资金流出"}'}) self.assertEqual(parsed['reason'],'资金流出') def test_cancel_failure_keeps_stop_after_exit_completes(self): from app.services import execution_conflicts as c buy=ins(self.fake,'old','OPEN','buy',600,400) sell=ins(self.fake,'exit','EXIT','sell',400,200) with patch.object(dispatcher,'cancel',return_value={'ok':False,'error':'offline'}): result=c.enforce_exit(sell) self.assertFalse(result['ok']) self.assertEqual(buy['status'],'DISPATCHED') self.assertEqual(buy['progress']['stopped_by_exit'],'exit') sell['status']='CONFIRMED' r=dispatcher.dispatch(instruction_id='old_D03',parent_id='old',ts_code=CODE, side='buy',qty=200,limit_price=28) self.assertFalse(r['ok']) self.assertEqual(self.fake.qmt.orders,{}) def test_cleanup_cancels_gated_plans_and_strategy_buy_proposal(self): from app.services import execution_conflicts as c self.fake.plans=[{'plan_id':'p1','ts_code':CODE,'action':'ADD','status':'GATED','filled_qty':0}] self.fake.insert_proposal(proposal_id='t0buy',ts_code=CODE,action='T0_ROUND',qty=200, hard_numbers={'kind':'strategy','side':'buy'},expire_at=datetime.now()+timedelta(hours=1)) result=c.enforce_exit(ins(self.fake,'exit','EXIT','sell',400)) self.assertTrue(result['ok'],result) self.assertEqual(self.fake.plans[0]['status'],'CANCELLED') self.assertEqual(self.fake.proposals['t0buy']['status'],'DECLINED') self.assertTrue(any(x['arbiter']=='system' and x['ref_id']=='t0buy' for x in self.fake.ledger)) def test_all_exit_origins_share_executor_cleanup(self): from app.services import execution_conflicts as c, portfolio, exec_advisor for origin in ('system','proposal','plan','strategy'): with self.subTest(origin=origin): self.fake.instructions.clear() buy=ins(self.fake,'old','OPEN','buy',600,400) sell=ins(self.fake,'exit','EXIT','sell',400,200) sell['origin_type']=origin with patch.object(portfolio,'positions_view',return_value={'positions':[],'params':{}}), \ patch.object(executor.td,'trade_days_left',return_value=2), \ patch.object(exec_advisor,'decide',return_value={'action':'WAIT','reason':'等待可卖'}): result=executor.run_tick() self.assertEqual(buy['status'],'CANCELLED',result) self.assertFalse(result['fired']) self.assertEqual(sell['exec_qty'],200) def test_dry_run_does_not_cancel(self): from app.services import portfolio, exec_advisor buy=ins(self.fake,'old','OPEN','buy',600,400) ins(self.fake,'exit','EXIT','sell',400,200) with patch.object(portfolio,'positions_view',return_value={'positions':[],'params':{}}), \ patch.object(executor.td,'trade_days_left',return_value=2), \ patch.object(exec_advisor,'decide',return_value={'action':'WAIT','reason':'等待可卖'}): executor.run_tick(dry_run=True) self.assertEqual(buy['status'],'DISPATCHED') self.assertFalse(buy['progress'].get('stopped_by_exit')) self.assertFalse(self.fake.ledger) def test_queued_buy_rechecks_parent_and_exit(self): from app.services import execution_conflicts as c buy=ins(self.fake,'old','OPEN','buy') row={'parent_id':'old','ts_code':CODE,'side':'buy'} self.assertIsNone(c.queued_buy_blocker(row)) ins(self.fake,'exit','EXIT','sell') self.assertTrue(c.queued_buy_blocker(row)) self.fake.instructions['exit']['status']='CONFIRMED' buy['status']='CANCELLED' self.assertTrue(c.queued_buy_blocker(row)) self.assertIsNone(c.queued_buy_blocker(dict(row,side='sell'))) def test_missing_parent_keeps_queue_unverified(self): from app.services import execution_conflicts as c with self.assertRaises(RuntimeError): c.queued_buy_blocker({'parent_id':'missing','ts_code':CODE,'side':'buy'}) def test_finished_children_show_wait_instead_of_pending_order(self): from app.services import execution_conflicts as c row=ins(self.fake,'old','OPEN','buy',600,400) row['progress']['last_decision']={'reason':'当日配额已出完'} with patch.object(qmt_repo,'orders_for_parent',return_value=[{'status':'FILLED'}]): self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'等待下一批') row['progress']['last_decision']={'reason':'可卖量不足, 顺延'} with patch.object(qmt_repo,'orders_for_parent',return_value=[]): self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'等待可卖') def test_pending_cancel_never_shown_as_finished(self): from app.services import execution_conflicts as c row=ins(self.fake,'old','OPEN','buy');row['status']='CANCELLED' with patch.object(qmt_repo,'orders_for_parent',return_value=[{'status':'PARTIAL','cancel_state':'REQUESTED'}]): self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'撤单处理中') with patch.object(qmt_repo,'orders_for_parent',side_effect=RuntimeError('unavailable')): self.assertEqual(c.instruction_progress([row])[0]['execution_label'],'委托状态待核实') def test_other_stock_not_blocked(self): from app.services import execution_conflicts as c ins(self.fake,'exit','EXIT','sell') self.assertIsNone(c.buy_blocker('600000.SH')) def test_claim_rejects_cancel_requested_atomically(self): with patch.object(qmt_repo,'execute',return_value=0) as write: self.assertFalse(qmt_repo.claim_order('child')) self.assertIn("cancel_state = 'NONE'",write.call_args.args[0]) def test_outbox_drops_new_conflict_without_sending(self): import asyncio from unittest.mock import AsyncMock from app.ws.runner import WsRunner buy=ins(self.fake,'old','OPEN','buy') ins(self.fake,'exit','EXIT','sell') row={'instruction_id':'old_D03','parent_id':'old','ts_code':CODE,'side':'buy', 'qty':200,'limit_price':28,'valid_until':int((datetime.now()+timedelta(minutes=10)).timestamp()*1000), 'cancel_state':'NONE','status':'SENDING'} loop=asyncio.new_event_loop() asyncio.set_event_loop(loop) self.addCleanup(loop.close) runner=WsRunner() with patch.object(qmt_repo,'next_queued',return_value=[row]), \ patch.object(qmt_repo,'claim_order',return_value=True), \ patch.object(qmt_repo,'get_order',return_value=row), \ patch.object(qmt_repo,'abort_order') as abort, \ patch.object(qmt_repo,'mark_sent'), \ patch.object(runner,'_send',new_callable=AsyncMock) as send: asyncio.run(runner._drain_outbox('unused')) abort.assert_called_once() send.assert_not_called() def test_outbox_source_failure_preserves_queue(self): import asyncio from unittest.mock import AsyncMock from app.ws.runner import WsRunner row={'instruction_id':'old_D03','parent_id':'old','ts_code':CODE,'side':'buy', 'valid_until':int((datetime.now()+timedelta(minutes=10)).timestamp()*1000), 'cancel_state':'NONE'} loop=asyncio.new_event_loop() asyncio.set_event_loop(loop) self.addCleanup(loop.close) runner=WsRunner() with patch.object(qmt_repo,'next_queued',return_value=[row]), \ patch.object(qmt_repo,'claim_order',return_value=True), \ patch.object(qmt_repo,'get_order',return_value=row), \ patch.object(qmt_repo,'requeue_order') as retry, \ patch.object(qmt_repo,'mark_sent'), \ patch.object(runner,'_send',new_callable=AsyncMock) as send: asyncio.run(runner._drain_outbox('unused')) retry.assert_called_once() self.assertIs(retry.call_args.args[-1],False) send.assert_not_called() def test_stale_progress_cannot_restore_stopped_buy(self): from app.services import execution_conflicts as c buy=ins(self.fake,'old','OPEN','buy',600,400) sell=ins(self.fake,'exit','EXIT','sell',400,200) with patch.object(dispatcher,'cancel',return_value={'ok':False,'error':'offline'}): c.enforce_exit(sell) buy['progress']={'children':[]} # 模拟并发旧进度覆盖，停止事实仍在追加账本中。 sell['status']='CONFIRMED' result=dispatcher.dispatch(instruction_id='old_D03',parent_id='old',ts_code=CODE,side='buy',qty=200,limit_price=28) self.assertFalse(result['ok']) def test_terminal_parent_cannot_be_revived_by_old_dispatch_write(self): buy=ins(self.fake,'old','OPEN','buy');buy['status']='CANCELLED' self.assertEqual(pms_repo.update_instruction('old',status='DISPATCHED'),0) self.assertEqual(buy['status'],'CANCELLED') def test_human_accept_is_blocked_before_changing_proposal(self): from app.web import main ins(self.fake,'exit','EXIT','sell') self.fake.insert_proposal(proposal_id='p',ts_code=CODE,action='OPEN',qty=200, hard_numbers={},expire_at=datetime.now()+timedelta(hours=1)) with patch.object(main,'ok_logged',side_effect=lambda name,fn,**kw:fn()), \ patch.object(main,'_actor',return_value='test'): result=main.api_decide(None,'p',{'decision':'ACCEPTED','reason':'用户确认'}) self.assertFalse(result['ok']) self.assertEqual(self.fake.proposals['p']['status'],'WAIT_USER') self.assertEqual(len(self.fake.instructions),1) def test_human_exit_retains_t1_locked_quantity(self): from app.web import main self.fake.positions[CODE]={'ts_code':CODE,'total_qty':400,'avail_qty':0} self.fake.insert_proposal(proposal_id='exit_prop',ts_code=CODE,action='EXIT',qty=200, hard_numbers={},expire_at=datetime.now()+timedelta(hours=1)) with patch.object(main,'ok_logged',side_effect=lambda name,fn,**kw:fn()), \ patch.object(main,'_actor',return_value='test'), \ patch.object(main.td,'window_deadline',return_value='2026-10-21'): result=main.api_decide(None,'exit_prop',{'decision':'ACCEPTED','reason':'确认清仓'}) self.assertTrue(result['ok'],result) row=self.fake.instructions[result['instruction_id']] self.assertEqual(row['qty'],200) # 不扩成当前400股 self.assertEqual(row['action'],'EXIT') self.assertEqual(row['exec_qty'],0) def test_human_exit_after_position_closed_still_stops_old_buy(self): from app.web import main buy=ins(self.fake,'old','OPEN','buy',600,400) self.fake.positions[CODE]={'ts_code':CODE,'total_qty':0,'avail_qty':0} self.fake.insert_proposal(proposal_id='exit_prop',ts_code=CODE,action='EXIT',qty=200, hard_numbers={},expire_at=datetime.now()+timedelta(hours=1)) with patch.object(main,'ok_logged',side_effect=lambda name,fn,**kw:fn()), \ patch.object(main,'_actor',return_value='test'): result=main.api_decide(None,'exit_prop',{'decision':'ACCEPTED','reason':'确认清仓'}) self.assertTrue(result['ok'],result) self.assertIsNone(result['instruction_id']) self.assertEqual(buy['status'],'CANCELLED') self.assertEqual(buy['exec_qty'],400) if __name__=='__main__':unittest.main()