"""本地：python scripts/unit_test_guard.py scripts/test_batch38_units.py。禁止联网。""" import unit_test_guard # noqa: F401 import copy import os import sys import unittest from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.services import plan_feed as pf from app.core import plan_revision as pr from app.core import action_engine as ae from test_batch20_units import params, caps def selected(): plan = pf.parse_plan({'date': '2026-10-14', 'rank_axis': 'quality', 'decision': {'id': 'v1'}, 'main': [{'code': 'SH600001', 'name': '甲', 'rank': 1, 'score': 210, 'verdict': '候选'}, {'code': 'SH600002', 'name': '乙', 'rank': 2, 'score': 250, 'verdict': '候选'}]}) return pf.select_candidates(plan, rank_by_upstream=True, theme_cap=1)['items'] class DailyPlanTest(unittest.TestCase): def test_judge_cache_requires_same_material_basis(self): old = {'plan_basis': {'risk': []}, 'scan_params_hash': 'p', 'portfolio_context_hash': 'c'} ranked = dict(old, decision={'id': 'new'}, research_rank=99) self.assertFalse(pr.needs_new_judgement(ranked, old)) self.assertTrue(pr.needs_new_judgement(dict(old, plan_basis={'risk': ['变化']}), old)) self.assertTrue(pr.needs_new_judgement(old, None)) def test_portfolio_quantity_basis_detects_new_commitments(self): view = {'held': [{'ts_code': '600001.SH', 'total_qty': 100, 'sector': '电子', 'price': 10}], 'totals': {'cash_source': 'ws', 'cash_avail': 10000}, 'sector_ready': True} commitments = {'names': [], 'amount': 0} original = pr.portfolio_context(view, commitments) view['held'][0]['price'] = 11 self.assertEqual(original, pr.portfolio_context(view, commitments)) commitments.update(names=['600002.SH'], amount=1000) self.assertNotEqual(original, pr.portfolio_context(view, commitments)) def test_commitment_quotes_do_not_change_quantity_basis(self): view = {'held': [], 'totals': {'cash_source': 'ws', 'cash_avail': 10000}} old = {'names': {'600002.SH'}, 'amount': 1000, 'basis': [('instruction', 'i1', '600002.SH', 100, 0)]} changed = dict(old, amount=1100) self.assertEqual(pr.portfolio_context(view, old), pr.portfolio_context(view, changed)) for basis in [[('instruction', 'i2', '600002.SH', 100, 0)], [('instruction', 'i1', '600002.SH', 200, 0)], [('instruction', 'i1', '600002.SH', 100, 50)], []]: self.assertNotEqual(pr.portfolio_context(view, old), pr.portfolio_context(view, dict(old, basis=basis))) double = dict(old, basis=old['basis'] + [('proposal', 'p1', '600003.SH', 100, 0)]) self.assertEqual(pr.portfolio_context(view, double), pr.portfolio_context(view, dict(double, basis=list(reversed(double['basis']))))) self.assertNotEqual(pr.portfolio_context(view, old), pr.portfolio_context(view, old, {'black': True})) changed_view = copy.deepcopy(view) changed_view['totals']['cash_avail'] = 9000 self.assertNotEqual(pr.portfolio_context(view, old), pr.portfolio_context(changed_view, old)) legacy = {k: v for k, v in old.items() if k != 'basis'} self.assertNotEqual(pr.portfolio_context(view, legacy), pr.portfolio_context(view, dict(legacy, amount=1100))) def test_inflight_basis_is_stable_and_preserves_amount_rules(self): from app.services import proposal_service as ps from app.repo import pms_repo ins = [{'instruction_id': 'i1', 'action': 'OPEN', 'ts_code': '600002.SH', 'qty': 100, 'exec_qty': 0}, {'instruction_id': 'held', 'action': 'OPEN', 'ts_code': '600003.SH', 'qty': 100}] proposals = [{'proposal_id': 'own', 'action': 'OPEN', 'ts_code': '600001.SH', 'qty': 100}, {'proposal_id': 'p2', 'action': 'OPEN', 'ts_code': '600002.SH', 'qty': 200, 'hard_numbers': {'price': 9}}] view = {'held': [{'ts_code': '600003.SH'}]} with patch.object(pms_repo, 'list_instructions', return_value=ins), \ patch.object(pms_repo, 'list_proposals', return_value=proposals), \ patch.object(ps.market, 'get_prices', side_effect=[{'600002.SH': 10}, {'600002.SH': 11}, {}]): old = ps._inflight_open_commitments(view, exclude_proposal='own') new = ps._inflight_open_commitments(view, exclude_proposal='own') missing_price = ps._inflight_open_commitments(view, exclude_proposal='own') self.assertEqual(old['names'], {'600002.SH'}) self.assertEqual(old['amount'], 1000) self.assertEqual(new['amount'], 1100) self.assertEqual(missing_price['amount'], 0) # 原首条指令优先去重口径保持。 self.assertEqual(old['basis'], new['basis']) self.assertEqual(old['basis'], [('instruction', 'i1', '600002.SH', 100, 0), ('proposal', 'p2', '600002.SH', 200, 0)]) with patch.object(pms_repo, 'list_instructions', return_value=[]), \ patch.object(pms_repo, 'list_proposals', return_value=[]), \ patch.object(ps.market, 'get_prices', side_effect=AssertionError('无在途不读行情')): self.assertEqual(ps._inflight_open_commitments(view)['basis'], []) def test_quote_only_recheck_keeps_confirmation_without_model(self): from app.services import plan_recheck as rc, proposal_service as ps, portfolio, command_service row = dict(selected()[0], consumption_id='test-consumption') view = {'held': [], 'totals': {'cash_source': 'ws', 'cash_avail': 10000}} old = {'names': {'600002.SH'}, 'amount': 1000, 'basis': [('instruction', 'i1', '600002.SH', 100, 0)]} hard = {'plan_basis': pr.basis(row), 'scan_params_hash': pr.fingerprint({}), 'portfolio_context_hash': pr.portfolio_context(view, old), 'proposal_revision': 'same'} p = {'proposal_id': 'p1', 'action': 'OPEN', 'ts_code': row['ts_code'], 'hard_numbers': hard} with patch.object(portfolio, 'positions_view', return_value=view), \ patch.object(command_service, 'effective_stock_params', return_value={}), \ patch.object(ps, '_scan_params', return_value={}), \ patch.object(pf, 'candidates', return_value={'items': [row], 'decision': row['decision']}), \ patch.object(ps, '_inflight_open_commitments', return_value=dict(old, amount=1100)), \ patch.object(ps, '_scan_open', side_effect=AssertionError('报价变化不得重算')) as scan, \ patch.object(ps, '_route_one', side_effect=AssertionError('报价变化不得调用研判')) as route: out = rc.before_accept(p, 'same') self.assertTrue(out['ok'], out) self.assertEqual(out['proposal']['hard_numbers']['version_check']['status'], 'passed') scan.assert_not_called() route.assert_not_called() def test_recheck_ignores_rank_cutoffs_but_keeps_qualification(self): raw = {'date': '2026-10-14', 'rank_axis': 'quality', 'decision': {'id': 'v2'}, 'main': [{'code': 'SH600001', 'rank': 1, 'score': 230, 'tier': '强传导', 'evidence': {'theme': '电子'}}, {'code': 'SH600002', 'rank': 2, 'score': 210, 'tier': '强传导', 'evidence': {'theme': '电子'}}]} p = dict(top_n=1, tiers=['强传导'], include_observe=False, min_score=0, min_sources=0, min_upside=0, theme_cap_local=1, exclude_st=False, rank_by_quality=True, rank_by_upstream=True, observe_if_fund_bull=False) with patch.object(pf, '_params', return_value=p), patch.object(pf, 'get_plan', return_value=pf.parse_plan(raw)): self.assertEqual([x['ts_code'] for x in pf.candidates(record=False)['items']], ['600001.SH']) fresh = pf.candidates(record=False, recheck_code='600002.SH') self.assertEqual([x['ts_code'] for x in fresh['items']], ['600002.SH']) self.assertEqual(fresh['consumption']['kind'], 'rechecked') self.assertEqual(pf.candidates(record=False, recheck_code='600002.SH', black=['600002.SH'])['items'], []) def test_snapshot_retention_never_deletes_by_row_count(self): from app.repo import pms_repo with patch.object(pms_repo, 'execute', side_effect=AssertionError('不得按行数删除历史')): self.assertEqual(pms_repo.prune_plan_snapshots(2), 0) def test_material_change_replaces_qty_and_never_accepts_old_confirmation(self): from app.services import plan_recheck as rc, proposal_service as ps, param_store from app.repo import pms_repo p = {'proposal_id': 'p1', 'action': 'OPEN', 'ts_code': '600001.SH', 'qty': 300, 'hard_numbers_json': '{}', 'hard_numbers': {'proposal_revision': 'old'}} c = {'qty': 100, 'ts_code': p['ts_code'], 'hard_numbers': {'price': 10, 'plan_basis': {}}} check = {'material': True, 'view': {'held': []}, 'params': {}, 'stock': {}} with patch.object(rc, 'current_basis', return_value=check), \ patch.object(ps, '_scan_open', return_value=[c]) as scan, \ patch.object(ps, '_route_one', return_value={'verdict': {'verdict': 'PASS'}}) as route, \ patch.object(param_store, 'get_int', return_value=0), \ patch.object(pms_repo, 'revise_pending_proposal', return_value=1) as revise, \ patch.object(pms_repo, 'insert_ledger', return_value=1): out = rc.before_accept(p, 'old') self.assertFalse(out['ok']) self.assertEqual(out['proposal']['qty'], 100) self.assertNotEqual(out['proposal']['hard_numbers']['proposal_revision'], 'old') self.assertEqual(scan.call_args.kwargs['exclude_proposal'], 'p1') self.assertTrue(route.call_args.kwargs['review_only']) self.assertTrue(c['_plan_recheck']) def test_version_unavailable_and_qualification_loss_preserve_proposal(self): from app.services import plan_recheck as rc p = {'action': 'OPEN', 'ts_code': '600001.SH', 'hard_numbers': {}} for check in ({'ok': False, 'error': '版本不可用'}, {'ok': False, 'material': True, 'error': '资格失效'}): with patch.object(rc, 'current_basis', return_value=check): self.assertFalse(rc.before_accept(p, None)['ok']) def test_compare_and_set_rejects_stale_and_duplicate_acceptance(self): import sqlite3 import threading from concurrent.futures import ThreadPoolExecutor from app.repo import pms_repo db = sqlite3.connect(':memory:', check_same_thread=False) self.addCleanup(db.close) db.execute('CREATE TABLE pms_proposal(proposal_id TEXT, status TEXT, decided_at TEXT, hard_numbers_json TEXT)') db.execute("INSERT INTO pms_proposal VALUES ('p1', 'WAIT_USER', NULL, 'new')") lock = threading.Lock() def execute(sql, params): with lock: n = db.execute(sql, params).rowcount db.commit() return n with patch.object(pms_repo, 'execute', execute): self.assertEqual(pms_repo.decide_proposal('p1', 'ACCEPTED', expected_hard='old'), 0) with ThreadPoolExecutor(2) as pool: results = list(pool.map(lambda _: pms_repo.decide_proposal('p1', 'ACCEPTED', expected_hard='new'), range(2))) self.assertEqual(sum(results), 1) def test_rank_survives_to_single_slot(self): rows = [dict(r, price=10) for r in selected()] out = ae.scan_open(candidates=rows, params=params(open_signal_priority=False), caps=caps(), room_amt=1_000_000, slots=1) self.assertEqual(out['candidates'][0]['ts_code'], '600001.SH') self.assertEqual(out['candidates'][0]['hard_numbers']['execution_rank'], 1) def test_signal_priority_has_reason(self): rows = [dict(r, price=10) for r in selected()] rows[1]['sig_buy'] = {'reason': '盘中转多'} out = ae.scan_open(candidates=rows, params=params(open_signal_priority=True), caps=caps(), room_amt=1_000_000, slots=1) self.assertEqual(out['candidates'][0]['ts_code'], '600002.SH') self.assertEqual(out['candidates'][0]['hard_numbers']['priority_reason'], '盘中转多信号优先') def test_unknown_themes_do_not_form_sector(self): self.assertEqual(len(selected()), 2) def test_rank_only_change_keeps_proposal(self): row = selected()[0] old = pr.basis(row) row.update(selection_rank=20, research_rank=100, decision={'id': 'v2'}) self.assertEqual(pr.compare(old, row, row['ts_code']), 'unchanged') row['risk'] = ['材料存在矛盾'] self.assertEqual(pr.compare(old, row, row['ts_code']), 'material') def test_pending_or_unverifiable_does_not_pass(self): row = selected()[0] old = pr.basis(row) row['refresh_status'] = {'pending_codes': [row['ts_code']]} self.assertEqual(pr.compare(old, row, row['ts_code']), 'unavailable') row.pop('refresh_status') row['decision'] = None self.assertEqual(pr.compare(old, row, row['ts_code']), 'unavailable') def test_stale_human_confirmation_rejected(self): from app.services import plan_recheck as rc p = {'action': 'OPEN', 'ts_code': '600001.SH', 'hard_numbers': {'proposal_revision': 'new'}} with patch.object(rc, 'current_basis', return_value={'ok': True, 'provenance': {}}): self.assertFalse(rc.before_accept(p, 'old')['ok']) self.assertTrue(rc.before_accept(p, 'new')['ok']) def test_received_and_consumed_fingerprint_include_context(self): old = {'decision': {'id': 'v1'}, 'params': {'top_n': 2}, 'order': ['A', 'B']} new = copy.deepcopy(old) new['params']['top_n'] = 1 self.assertNotEqual(pr.fingerprint(old), pr.fingerprint(new)) self.assertEqual(old['params']['top_n'], 2) def test_pending_prefix_codes_normalize(self): p = pf.parse_plan({'date': '2026-10-14', 'main': [{'code': 'SH600001', 'score': 1}], 'refresh_status': {'pending_codes': ['SH600001']}}) self.assertEqual(p['refresh_status']['pending_codes'], ['600001.SH']) if __name__ == '__main__': unittest.main()