166 lines
9.2 KiB
Python
166 lines
9.2 KiB
Python
|
"""本地：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_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()
|