高股息建仓：独立候选复用后台准备与原子采纳（台账 082）

一、补齐研究至建仓的缺失入口。未知资料不放行，固定依据变化交后台处理。

二、本地桥9项新增与PMS24项专项通过，PMS48批隔离禁网回归通过。部署另行判收。

三、保留原每日榜单、自主档位、仓位和执行逻辑。新来源人工发起，不合并进原自主扫描，不制造交易验收。

DEVLOG.md

@@ -2113,3 +2113,10 @@ QMT在线、对账无差异。午休任务按原时段跳过，不制造交易
 桥逐票0.208—0.326秒；真实提议预检1.085—1.227秒。自然网格买入后仍保留原500股确认内容。
 10:22:38对账无差异、QMT在线。实际用户确认未代验，没有制造交易。
 详细备份、回退、历史边界见四仓交接10:25节。
+
+
+## 2026-10-22 高股息独立候选与建仓准备（开发验证）
+
+桥提供独立资格契约，PMS通过dividend_feed消费。人工请求先保存准备状态，正常交易时段完成计算与研判，原采纳事务负责最终确认。固定研究、消费凭据与来源跟随提议、审计和指令。原自动扫描候选及所有参数保持。
+
+新增24项专项通过。全量48批配置隔离、禁止联网回归通过，含三页面守卫。桥新增9项通过。生产部署与自然采纳尚未判收。详见四仓《高股息候选接入PMS开发规格_2026-10-22.md》及同日交接。

app/core/action_engine.py

@@ -860,6 +860,7 @@ def eval_open(c: dict, params: dict, caps: dict, room_amt: float):
         "plan_src": c.get("src"), "sector": c.get("sector"),
         **{k: c.get(k) for k in ('decision', 'consumption_id', 'consumption_error',
             'plan_basis', 'selection_rank', 'research_rank', 'research', 'review_ref')},
+        **{k: c[k] for k in ('candidate_source','dividend_ref','dividend_policy_version','dividend_research') if k in c},
         # 决策系统今天盘中判过这只票转多没有。这是**定性材料**不是仓位数字, 所以送研判时
         # 也带过去 (见 judge.OPEN_JUDGE_KEYS) —— 等于当面告诉它「你自己今天判过这只票转多」,
         # 它可以拿自己的结论去对照; 没有信号时这两项是 None, 不误导。
@@ -903,6 +904,9 @@ def eval_open(c: dict, params: dict, caps: dict, room_amt: float):
               + ("; **决策系统今天盘中判过该股转多**" if c.get("sig_buy") else "")
               + f", 现价 {price}, 目标仓位 {target_pct:.0%} 约 {full_amt:,.0f} 元, "
                 f"先建底仓 {qty} 股 (约 {qty * price:,.0f} 元)")
+    if c.get('candidate_source') == 'dividend':
+        reason = (f"高股息建仓：研究核验通过，现价 {price}，目标仓位 {target_pct:.0%}，"
+                  f"先建底仓 {qty} 股（约 {qty * price:,.0f} 元）")
     cand = _cand(c, A_OPEN, BUY, qty, reason, hard)
     # 这两项是 OPEN 独有的, 供 proposal_service 用:
     #   price  —— 新票在账本里没有行, _pos_of 拿不到现价, 取它会是 0 而被规则闸判 PRICE_MISSING

app/core/plan_revision.py

@@ -10,9 +10,15 @@ def fingerprint(value):
 
 def basis(row):
     # 排名与审计时间不是资格/风险/数量依据。完整评析引用变化需要重新研判。
-    return {k: row.get(k) for k in ('score', 'tier', 'upside', 'heat', 'theme',
+    result = {k: row.get(k) for k in ('score', 'tier', 'upside', 'heat', 'theme',
         'verdict', 'risk', 'missing', 'reasons', 'logic_state', 'company_gate',
         'company_review', 'company_review_text', 'events', 'pricing_state', 'news')}
+    if row.get('candidate_source') == 'dividend':
+        result.update({k: row.get(k) for k in ('candidate_source', 'dividend_ref', 'dividend_policy_version')})
+        # 资料年龄不是研究变化；固定报告引用和研究内容才是。
+        result['company_review'] = {k:v for k,v in (row.get('company_review') or {}).items() if k != 'age_days'}
+        result['review_ref'] = row.get('review_ref')
+    return result
 
 
 def portfolio_context(view, commitments, stock=None):

app/services/dividend_feed.py

@@ -0,0 +1,147 @@
+"""高股息独立候选来源。人工发起，复用后台准备与原子采纳，不参与原主榜扫描。"""
+from copy import deepcopy
+from datetime import datetime, timezone, timedelta
+import logging
+import re
+import requests
+from app.core.plan_revision import fingerprint, basis
+from app.core import tradedays as td
+from app.services.plan_feed import PlanFeedError
+
+SOURCE = 'dividend'
+BJ = timezone(timedelta(hours=8))
+logger = logging.getLogger('pms.dividend')
+
+
+def read(*, code=None):
+    from app.services import param_store
+    if code and not re.fullmatch(r'\d{6}\.(SH|SZ|BJ)', code):
+        raise ValueError('股票代码格式无效')
+    base = (param_store.get('PMS_PLAN_API_BASE', '') or '').rstrip('/')
+    if not base:
+        raise PlanFeedError('高股息研究地址尚未配置')
+    try:
+        res = requests.get(base+'/research/dividend/candidates', params={'code':code} if code else {}, timeout=(1,3))
+        res.raise_for_status()
+        data = res.json()
+        now = datetime.now(BJ)
+        if td.calendar_degraded():
+            raise ValueError('交易日历暂不可核对')
+        expected = str(td.prev_trade_day(now.date()))
+        if data['schema_version'] != 1 or data['source'] != SOURCE or data['data_date'] != expected:
+            raise ValueError('高股息资料尚未更新到上一交易日')
+        decision = data['decision']
+        if (decision.get('source') != SOURCE or decision.get('policy_version') != data.get('policy_version')
+                or not re.fullmatch(r'dividend_[0-9a-f]{64}', decision['id'])):
+            raise ValueError('高股息研究来源不完整')
+        for value in (decision['cutoff_at'], data['checked_at']):
+            at = datetime.fromisoformat(value.replace('Z','+00:00'))
+            if at.tzinfo is None or at > now:
+                raise ValueError('研究检查时间无法核对')
+        codes = set()
+        for row in data['rows']:
+            c = row['candidate']
+            if (not re.fullmatch(r'\d{6}\.(SH|SZ|BJ)', row['ts_code']) or row['ts_code'] in codes
+                    or (code and row['ts_code'] != code) or c['ts_code'] != row['ts_code']
+                    or c.get('candidate_source') != SOURCE or c.get('decision') != decision
+                    or not isinstance(row.get('eligible'), bool)):
+                raise ValueError('高股息候选内容无法核对')
+            codes.add(row['ts_code'])
+            if row['eligible'] and (c.get('verdict') != '候选' or row.get('reasons')
+                    or not re.fullmatch('[0-9a-f]{64}', str((c.get('dividend_research') or {}).get('ref', '')))):
+                raise ValueError('高股息候选缺少准入依据')
+        return data
+    except PlanFeedError:
+        raise
+    except (requests.RequestException, KeyError, TypeError, ValueError) as e:
+        logger.warning('高股息来源暂不可用: %s', type(e).__name__)
+        message = str(e) if isinstance(e, ValueError) and not isinstance(e, requests.RequestException) else '高股息研究暂不可用，恢复后可继续'
+        raise PlanFeedError(message) from e
+
+
+def candidates(*, held=(), black=(), recheck_code=None, record=True, route_by_verdict=True):
+    data = read(code=recheck_code)
+    items, excluded = [], []
+    for row in data['rows']:
+        reason = ('已持仓' if row['ts_code'] in held else '在黑名单中' if row['ts_code'] in black
+                  else '；'.join(row['reasons']) if not row['eligible'] else '')
+        if reason:
+            excluded.append({'ts_code':row['ts_code'], 'why':reason})
+        else:
+            c = deepcopy(row['candidate'])
+            c['dividend_ref'] = row['research_ref']
+            c['dividend_policy_version'] = data['policy_version']
+            items.append(c)
+    consumption = {'kind':'rechecked' if recheck_code else 'consumed', 'source':SOURCE,
+        'decision':data['decision'], 'params':{'policy_version':data['policy_version']},
+        'context':{'held':sorted(held),'black':sorted(black),'recheck_code':recheck_code},
+        'rank_mode':'code', 'filtered':excluded, 'order':[r['ts_code'] for r in items]}
+    cid = fingerprint(consumption)
+    for row in items:
+        row.update(consumption_id=cid, plan_basis=basis(row))
+    out = {'items':items, 'decision':data['decision'], 'date':data['data_date'],
+           'display_only':[], 'filtered':excluded, 'rank_mode':'code', 'consumption':dict(consumption,id=cid)}
+    if record:
+        from app.repo import pms_repo
+        try:
+            pms_repo.insert_plan_snapshot(plan_date=data['data_date'],digest=cid[:40],roster=items,
+                meta=out['consumption'],n_main=len(items))
+        except Exception as e:
+            logger.exception('高股息消费凭据保存失败')
+            out['consumption']['error'] = type(e).__name__
+            for row in items:
+                row['consumption_error'] = type(e).__name__
+    return out
+
+
+def request_preparation(code, research_ref, *, actor):
+    """只排入准备，不计算数量或调用模型。确定性编号与唯一键拦重复请求。"""
+    from app.services import portfolio, command_service, proposal_service as ps, param_store
+    from app.services.execution_conflicts import buy_blocker
+    from app.repo import pms_repo
+    now = datetime.now()
+    if not td.is_trade_day(now.date()) or now.hour >= 15 or td.calendar_degraded():
+        raise ValueError('请在交易日收盘前发起，建议当日有效')
+    if (param_store.get_bool('PMS_GLOBAL_EXEC_HALT',False)
+            or param_store.get('PMS_AUTONOMY','propose_only') == 'off'
+            or param_store.get('PMS_OPEN_AUTONOMY','full') == 'off'):
+        raise ValueError('建仓评估已暂停')
+    data = read(code=code)
+    row = next((r for r in data['rows'] if r['ts_code'] == code), None)
+    if not row or not row['eligible']:
+        raise ValueError('暂不具备建仓评估条件：'+('；'.join(row['reasons']) if row else '未进入有效研究名单'))
+    if not research_ref or row['research_ref'] != research_ref:
+        raise ValueError('研究内容已更新，请刷新后查看')
+    pid = f"PRP_{td.ymd(now)}_{code.replace('.','')}_OPEN"
+    old = pms_repo.get_proposal(pid)
+    if old:
+        return _existing(old)
+    stock = command_service.effective_stock_params()
+    view = portfolio.positions_view()
+    if any(r['ts_code'] == code for r in view['held']) or (stock.get(code) or {}).get('black'):
+        raise ValueError('该票已持仓或在黑名单中，不能新建仓')
+    blocked = buy_blocker(code)
+    if blocked:
+        raise ValueError(blocked)
+    if code in ps._inflight_open_commitments(view, strict=True)['names']:
+        raise ValueError('该票已有建仓提议或在途买单，请查看现有进度')
+    hard = {'candidate_source':SOURCE,'dividend_ref':research_ref,'dividend_policy_version':data['policy_version'],
+            'decision':data['decision'],'requested_by':actor,'requested_at':now.astimezone().isoformat(),
+            'name':row.get('name'), 'preparation':{'status':'pending','stage':'requested',
+                'reason':'等待交易时段完成建仓评估'}}
+    try:
+        pms_repo.insert_proposal(proposal_id=pid,ts_code=code,action='OPEN',qty=0,hard_numbers=hard,
+            expire_at=ps.proposal_expire_at(ttl_hours=24),judge_verdict=None,judge_reason='高股息建仓评估待完成')
+    except Exception:
+        old = pms_repo.get_proposal(pid)
+        if old:
+            return _existing(old)
+        raise
+    return {'ok':True,'proposal_id':pid,'preparing':True,'message':'已排入建仓评估，准备完成后可确认'}
+
+
+def _existing(old):
+    if old['status'] in ('WAIT_USER','ACCEPTED'):
+        return {'ok':True,'proposal_id':old['proposal_id'],'already_requested':True,
+                'message':'已有该票建议，请查看待确认提议或指令进度'}
+    raise ValueError('该票今日建议已经结束，请先核对原处理结果')

app/services/plan_recheck.py

@@ -13,8 +13,13 @@ def current_basis(hard, code, proposal_id=None, *, record=True):
     view = portfolio.positions_view()
     stock = command_service.effective_stock_params()
     params = ps._scan_params(view)
+    feed = plan_feed
+    if hard.get('candidate_source') == 'dividend':
+        from app.services import dividend_feed as feed
+        options = {'record': record}
+    else:
         options = {'force': True} if record else {'record': False, 'plan_override': plan_feed.read_current_plan(code=code)}
-    selected = plan_feed.candidates(held=[x['ts_code'] for x in view['held']],
+    selected = feed.candidates(held=[x['ts_code'] for x in view['held']],
         black=[c for c, v in stock.items() if v.get('black')],
         route_by_verdict=bool(params.get('open_route_by_verdict')), recheck_code=code, **options)
     current = next((r for r in selected['items'] if r['ts_code'] == code), None)
@@ -31,7 +36,10 @@ def current_basis(hard, code, proposal_id=None, *, record=True):
     if result == 'unavailable':
         return {'ok': False, 'error': '当前研究资料无法核对，恢复后自动更新建议', 'stage': 'source'}
     ph = fingerprint({k: v for k, v in params.items() if not k.startswith('_')})
-    commitments = ps._inflight_open_commitments(view, exclude_proposal=proposal_id)
+    commitments = ps._inflight_open_commitments(view, exclude_proposal=proposal_id,
+        **({'strict':True} if hard.get('candidate_source') == 'dividend' else {}))
+    if hard.get('candidate_source') == 'dividend' and code in commitments['names']:
+        return {'ok':False, 'withdraw':True, 'material':True, 'error':'该票已有其他建仓承诺，当前建议已撤下'}
     context = portfolio_context(view, commitments, stock.get(code))
     changed = {'research_changed': result == 'material',
                'params_changed': hard.get('scan_params_hash') != ph,
@@ -181,8 +189,12 @@ def refresh_pending(*, now=None, deadline=None):
     from app.services import proposal_service as ps
     from app.repo import pms_repo
     results = []
+    from app.services import param_store
+    legacy_on = param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False)
     rows = pms_repo.list_proposals(statuses=('WAIT_USER',), limit=200)
     for p in rows:
+        if not legacy_on and (p.get('hard_numbers') or {}).get('candidate_source') != 'dividend':
+            continue
         if p.get('action') != 'OPEN' or (p.get('hard_numbers') or {}).get('kind') == 'strategy' or ps.is_expired(p, now):
             continue
         try:
@@ -246,7 +258,8 @@ def refresh_one(p, *, now=None, deadline=None):
     mkt = {}
     params['_mkt'] = mkt
     fresh = ps._scan_open(check['view'], params, check['stock'], {}, mkt, out, now=now,
-                           review_code=p['ts_code'], exclude_proposal=p['proposal_id'])
+                           review_code=p['ts_code'], exclude_proposal=p['proposal_id'],
+                           **({'candidate_source':'dividend'} if hard.get('candidate_source') == 'dividend' else {}))
     if not fresh:
         return _failed_recheck(p, out)
     c = fresh[0]
@@ -258,6 +271,7 @@ def refresh_one(p, *, now=None, deadline=None):
     if not result:
         return _failed_recheck(p, out)
     hn = c['hard_numbers']
+    hn.update({k:hard[k] for k in ('requested_by','requested_at') if k in hard})
     latest = current_basis(hn, p['ts_code'], proposal_id=p['proposal_id'])
     if not latest.get('ok'):
         return _pending_response(p, latest.get('error') or '建议依据再次变化，下一轮继续更新', latest.get('stage', 'changed'))

app/services/proposal_service.py

@@ -235,7 +235,7 @@ def scan_and_route(*, now=None, dry_run: bool = False) -> dict:
         0, param_store.get_int("PMS_JUDGE_TICK_BUDGET_SEC", 150))
 
     for group_index, group in enumerate((list(scanned['candidates']), open_cands)):
-        if group_index == 1 and not dry_run and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False):
+        if group_index == 1 and not dry_run:
             from app.services.plan_recheck import refresh_pending
             out['prepared_proposals'] = refresh_pending(now=now, deadline=deadline)
         for c in group:
@@ -315,12 +315,13 @@ def _save_tech_exit_done(codes) -> None:
 
 # ================================================================ 新建仓的取数与筛选
 def _scan_open(view, params, stock_params, skip, mkt, out, now=None,
-               review_code=None, exclude_proposal=None, record_consumption=True) -> list:
+               review_code=None, exclude_proposal=None, record_consumption=True, candidate_source='plan') -> list:
     """候选池 → 新建仓候选。产出的候选数天生不超过剩余名额 (名额与金额在纯逻辑里边走边扣)。
 
     与命令驱动那条路 (command_service._candidates) 有意不同的两点:
       1. **只认上游计划接口这一个来源。** buy_plan 那张旧表在目标架构下没有明确的写入方,
          白名单是给人下命令用的 —— 无人值守地建仓, 事实源必须单一。
+         人工发起的高股息准备显式传来源，只从原来源重核，不加入此自主扫描。
       2. **价格只认实时价。** 那边用 market.plan_price, 拿不到实时价会回落昨收; 它自己的
          注释就写着「拿昨收当现价去做不追高这类判断会出错, 只给规划期定量用」。
          这条路是要真下单的, 所以走 market.get_prices, 取不到就整只跳过并留痕。
@@ -336,7 +337,7 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None,
 
     src = (param_store.get("PMS_CANDIDATE_SOURCE", plan_feed.SRC_PLAN_API)
            or plan_feed.SRC_PLAN_API).strip()
-    if src == plan_feed.SRC_BUY_PLAN:
+    if src == plan_feed.SRC_BUY_PLAN and candidate_source == 'plan':
         out["skipped"].append({"action": ae.A_OPEN,
                                "why": f"候选池来源是 {src}, 自主新建仓只认上游计划接口"})
         return []
@@ -346,7 +347,10 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None,
     try:
         # 按判决分流开着时, 上游判为「仅展示」的票在候选阶段就剔掉 (理由见
         # plan_feed.select_candidates 的说明: 不剔的话它们会白占 top_n 的名额)。
-        sel = plan_feed.candidates(held=held, black=black,
+        feed = plan_feed
+        if candidate_source == 'dividend':
+            from app.services import dividend_feed as feed
+        sel = feed.candidates(held=held, black=black,
                                    route_by_verdict=bool(params.get("open_route_by_verdict")),
                                    **({'recheck_code': review_code} if review_code else {}),
                                    **({'record': False} if not record_consumption else {}))
@@ -393,6 +397,10 @@ def _scan_open(view, params, stock_params, skip, mkt, out, now=None,
     # 名额与金额都要扣掉**在途的建仓承诺** (2026-08-28 审查修): names_count 只数已成交
     # 持仓, 在途 OPEN 指令与等拍板的 OPEN 提议两头都不占 —— 候选没进买入区间时指令滞留
     # 在途, 下一跳名额照旧又放两只, 几分钟就把持仓数上限击穿。同票去重挡不住换一只票。
+    if candidate_source == 'dividend':
+        inflight = _inflight_open_commitments(view, exclude_proposal=exclude_proposal, strict=True)
+        cands = [c for c in cands if c['ts_code'] not in inflight['names']]
+    else:
         inflight = (_inflight_open_commitments(view, exclude_proposal=exclude_proposal)
                     if exclude_proposal else _inflight_open_commitments(view))
     slots = int(p["max_names"] or 0) - int(t["names_count"] or 0) - len(inflight["names"])
@@ -602,7 +610,7 @@ def _no_candidate_why(sel) -> str:
     return head + "：" + "，".join(parts) + "。"
 
 
-def _inflight_open_commitments(view, exclude_proposal=None) -> dict:
+def _inflight_open_commitments(view, exclude_proposal=None, strict=False) -> dict:
     """在途的新建仓承诺: names 为代码集合，amount 为预估占用，basis 为稳定核验依据。
 
     口径: 在途 (LIVE) 的 OPEN 指令 + 等拍板 (WAIT_USER) 的 OPEN 提议, 只算**尚未持有**
@@ -620,6 +628,8 @@ def _inflight_open_commitments(view, exclude_proposal=None) -> dict:
                 basis.append(('instruction', str(i.get('instruction_id') or ''),
                               i['ts_code'], int(i.get('qty') or 0), int(i.get('exec_qty') or 0)))
     except Exception as e:
+        if strict:
+            raise
         logger.warning("读在途 OPEN 指令失败 (按无在途): %s", e)
     try:
         for pr in pms_repo.list_proposals(statuses=("WAIT_USER",), limit=200):
@@ -632,6 +642,8 @@ def _inflight_open_commitments(view, exclude_proposal=None) -> dict:
                 basis.append(('proposal', str(pr.get('proposal_id') or ''),
                               pr['ts_code'], int(pr.get('qty') or 0), 0))
     except Exception as e:
+        if strict:
+            raise
         logger.warning("读在途 OPEN 提议失败 (按无在途): %s", e)
     if not rows:
         return {"names": names, "amount": 0.0, "basis": []}
@@ -1468,6 +1480,11 @@ def _make_proposal(c, price, verdict) -> str:
             judge_verdict=verdict.get("verdict"),
             judge_reason=(verdict.get("reason") or "")[:500])
     except Exception:
+        existing = pms_repo.get_proposal(pid)
+        if existing and existing.get('status') in ('WAIT_USER', 'ACCEPTED'):
+            return pid
+        if existing is None:
+            raise
         # 确定性编号被当日已终态 (过期/驳回) 的旧提议占着 —— 补时间后缀重试一次,
         # 不让唯一键冲突把整轮扫描刷成 error (2026-08-28 审查修; 用户驳回的当日
         # 不重提另有 _declined_today_keys 挡在扫描入口, 这里兜的是过期重生成)

app/web/main.py

@@ -476,9 +476,10 @@ def api_proposals(status: str = Query("WAIT_USER"), limit: int = 100,
             tech_service.attach(rows)
         except Exception as e:  # noqa: BLE001
             logger.warning("[提议] 技术面立场挂不上: %s", e)
-        if param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False):
+        legacy_on = param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False)
         for row in rows:
             h = row.get('hard_numbers') or {}
+            if (legacy_on or h.get('candidate_source') == 'dividend'):
                 if row.get('action') == 'OPEN' and h.get('kind') != 'strategy':
                     preparation = h.get('preparation') or {}
                     row['accept_ready'] = preparation.get('status') == 'ready' and bool(h.get('consumption_id'))
@@ -487,6 +488,23 @@ def api_proposals(status: str = Query("WAIT_USER"), limit: int = 100,
     return ok(_list)
 
 
+@app.get("/api/dividend/candidates")
+def api_dividend_candidates():
+    from app.services import dividend_feed
+    return ok(dividend_feed.read)
+
+
+@app.post("/api/dividend/proposals")
+def api_dividend_proposal(request: Request, payload: dict = Body(default={})):
+    from app.services import dividend_feed
+    actor = _actor(request, payload)
+    result = ok(dividend_feed.request_preparation, str(payload.get('ts_code') or ''),
+                payload.get('research_ref'), actor=actor)
+    _oplog('dividend_prepare', ok_flag=result.get('ok',False), reason=result.get('error') or result.get('message'),
+           ts_code=payload.get('ts_code'), ref=result.get('proposal_id'), by=actor)
+    return result
+
+
 @app.get("/api/proposals/decide-reasons")
 def api_decide_reasons():
     """裁决预设理由的两张码表 (采纳 / 驳回), 页面据此铺按钮。台账 015。"""
@@ -549,7 +567,8 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default=
                     return {'ok': False, 'error': blocked, 'proposal_id': proposal_id}
         expected_hard = None
         if (decision == 'ACCEPTED' and p['action'] == ae.A_OPEN
-                and param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False)):
+                and (param_store.get_bool('PMS_PLAN_VERSION_RECHECK', False)
+                     or (p.get('hard_numbers') or {}).get('candidate_source') == 'dividend')):
             from app.services.plan_recheck import before_accept
             checked = before_accept(p, payload.get('proposal_revision'))
             if not checked.get('ok'):
@@ -567,7 +586,8 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default=
             instruction = {'id': iid, 'window': window, 'progress': {
                 'from_proposal': proposal_id, 'is_command': False, 'children': [],
                 'plan': {k: hn.get(k) for k in ('decision', 'consumption_id', 'review_ref',
-                                               'execution_rank', 'priority_reason', 'version_check')},
+                                               'execution_rank', 'priority_reason', 'version_check',
+                                               'candidate_source','dividend_ref','dividend_policy_version')},
                 'deadline': str(td.window_deadline(None, window))}}
             if pms_repo.accept_open_proposal(p, hn, instruction, reason):
                 return {'ok': True, 'decision': decision, 'instruction_id': iid}
@@ -632,7 +652,8 @@ def api_decide(request: Request, proposal_id: str, payload: dict = Body(default=
                 status="PROPOSED",
                 progress={"from_proposal": proposal_id, "is_command": False, "children": [],
                           'plan': {k: hn.get(k) for k in ('decision', 'consumption_id', 'review_ref',
-                                                         'execution_rank', 'priority_reason', 'version_check')},
+                                                         'execution_rank', 'priority_reason', 'version_check',
+                                               'candidate_source','dividend_ref','dividend_policy_version')},
                           "deadline": str(td.window_deadline(None, _win))})
             # 与自主执行同一口径: 采纳即算「做过一次」, 计数器要跟着走
             # (否则页面采纳的那条动作绕开了 §6 的一次性约束)

app/web/static/index.html

@@ -1250,7 +1250,7 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
                        公司深度那句最长 —— 现在卡面只留结论标签, 整句进弹窗。判定用的键一个没动。 -->
                   <div class="prop-src">
                     <div class="prop-src-row" v-if="propPlanTag(p)" @click="openPropDetail(p,'plan')">
-                      <span class="lbl">选股系统</span>
+                      <span class="lbl">{{ (p.hard_numbers||{}).candidate_source==='dividend' ? '高股息研究' : '选股系统' }}</span>
                       <span class="ptag" :class="propPlanTag(p).cls">{{ propPlanTag(p).text }}</span>
                       <span class="sub">{{ propPlanSub(p) }}</span><i class="car">›</i>
                     </div>
@@ -1404,6 +1404,26 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
                 <button class="research-link" style="margin-top:8px" @click="openPlan">查看完整选股计划</button>
               </div>
             </div>
+            <div class="sect" :class="{closed:sectClosed.dividend}">
+              <div class="secth" @click="toggleSect('dividend')">高股息研究
+                <span class="n">· {{ dividend.eligible_count || 0 }} 只可评估</span><span class="car">{{ sectClosed.dividend?'▸':'▾' }}</span></div>
+              <div class="sectb">
+                <div class="rail-note" v-if="dividend.data_date">股息数据截至 {{ dividend.data_date }} · 人工发起建仓评估</div>
+                <div class="muted" v-if="dividend.error">{{ dividend.error }}</div>
+                <div class="muted" v-else-if="!dividend.eligible_count">暂无公司通过全部核验。</div>
+                <div class="mini" v-for="r in dividendShown" :key="r.ts_code">
+                  <div class="top"><b>{{ r.name || r.ts_code }}</b><span class="spacer"></span><span class="muted">滚动股息 {{ Number(r.yield_ttm_pct).toFixed(2) }}%</span></div>
+                  <div class="muted">{{ r.eligible ? '可评估建仓' : r.conclusion }}</div>
+                  <details style="margin-top:5px"><summary class="research-link">依据与待核验事项</summary>
+                    <p v-if="r.thesis">{{ r.thesis }}</p><p v-if="r.antithesis">风险：{{ r.antithesis }}</p>
+                    <div v-for="reason in [...new Set([...(r.needs||[]),...(r.reasons||[])])]" :key="reason" class="muted">{{ reason }}</div>
+                  </details>
+                  <el-button v-if="r.eligible" size="small" style="margin-top:6px" :loading="dividendBusy===r.ts_code"
+                             :disabled="!!dividendBusy" @click="prepareDividend(r)">准备建仓建议</el-button>
+                </div>
+                <button class="research-link" v-if="(dividend.rows||[]).length>3" @click="dividendMore=!dividendMore">{{ dividendMore?'收起':'查看全部研究公司' }}</button>
+              </div>
+            </div>
 
 
           </div>
@@ -2154,7 +2174,14 @@ body.dock-r:not(.r-fold) .side-r .strip{display:none;}
               <div class="h">为什么选它</div>
               <div class="st" v-for="(x,ri) in propReasons(propDetailP)" :key="ri">· {{ x }}</div>
             </div>
-            <div class="cs-sect"><div class="h">榜单</div><div class="st muted">{{ propWhy(propDetailP) }}</div></div>
+            <template v-if="(propDetailP.hard_numbers||{}).candidate_source==='dividend'">
+              <div class="cs-sect" v-for="item in Object.values((propDetailP.hard_numbers.dividend_research||{}).assessments||{})" :key="item.reason">
+                <div class="st">{{ item.reason }}</div>
+                <details class="research-source"><summary>查看依据原文</summary><p v-for="e in item.evidence||[]" :key="e.ref">{{ e.quote }}</p></details>
+              </div>
+              <div class="cs-sect"><div class="h">反方与风险</div><p v-for="item in (propDetailP.hard_numbers.dividend_research||{}).risks||[]" :key="item.reason">{{ item.reason }}</p></div>
+            </template>
+            <div class="cs-sect" v-else><div class="h">榜单</div><div class="st muted">{{ propWhy(propDetailP) }}</div></div>
           </template>
 
           <template v-else-if="propDetailKind==='size'">
@@ -2720,6 +2747,29 @@ createApp({
     const _lvLoad = () => { try { const v = JSON.parse(sessionStorage.getItem(LV_STORE_KEY)); if (v && typeof v === 'object') return { ..._lvDefault, ...v }; } catch (e) {} return { ..._lvDefault }; };
     const levelShow = reactive(_lvLoad());
     const openScan = ref({ by_code: {}, notes: [] });
+    const dividend = ref({ rows: [] }), dividendBusy = ref(''), dividendMore = ref(false);
+    const dividendShown = computed(() => {
+      const rows = [...(dividend.value.rows || [])].sort((a,b) => Number(b.eligible)-Number(a.eligible)
+        || Number(b.quality==='好')-Number(a.quality==='好') || a.ts_code.localeCompare(b.ts_code));
+      return dividendMore.value ? rows : rows.slice(0,3);
+    });
+    let dividendLoadedAt = 0;
+    async function loadDividend(force=false) {
+      if (!force && Date.now()-dividendLoadedAt < 60000) return;
+      dividendLoadedAt = Date.now();
+      try {
+        const r = await axios.get('/api/dividend/candidates');
+        dividend.value = r.data.ok ? r.data.data : {rows:[],error:'研究暂不可用，稍后自动重试'};
+      } catch(e) { dividend.value = {rows:[],error:'研究暂不可用，稍后自动重试'}; }
+    }
+    async function prepareDividend(row) {
+      if (dividendBusy.value) return;
+      dividendBusy.value = row.ts_code;
+      try {
+        const result = await call('post','/api/dividend/proposals',{ts_code:row.ts_code,research_ref:row.research_ref});
+        if (result.ok) { ElementPlus.ElMessage.success(result.message); await loadProposals(); }
+      } finally { dividendBusy.value = ''; }
+    }
     const insTab = ref('live');
     const stratDlg = reactive({ visible:false, busy:false, reasons:[], ts_code:'', price:0,
       total_qty:0, avail_qty:0, cushion_pct:null, type:'T0', autonomy:'auto',
@@ -4300,7 +4350,7 @@ createApp({
         // ws-channel 是管理员专属接口: 交易员不调它, 免得每次刷新都弹一条「无权限」
         // (通道告警对交易员改由 dispatch-mode 里的 channel 兜底, 见 wsc 的注释)
         (isAdmin.value ? loadWs() : Promise.resolve()),
-        loadStrategies(), loadOpLog(), loadOpenScan(), loadUpstream(),
+        loadStrategies(), loadOpLog(), loadOpenScan(), loadDividend(), loadUpstream(),
         loadMacro(), loadTech()]);
       await loadNames();
       lastRefresh.value = new Date().toTimeString().slice(0, 8);
@@ -4672,7 +4722,7 @@ createApp({
     async function refreshLive() {
       // allSettled: 单个 loader 失败/超时(axios 60s 兜底)不拖累其余与时间戳更新。
       await Promise.allSettled([loadOverview(), loadPositions(), loadInstructions(), loadLedger(),
-        loadProposals(), loadStrategies(), loadOpLog(), loadOpenScan(), loadUpstream(), loadMacro()]);
+        loadProposals(), loadStrategies(), loadOpLog(), loadOpenScan(), loadDividend(), loadUpstream(), loadMacro()]);
       try { await loadNames(); } catch (e) {}
       // 用客户端钟, 保证"自动·HH:MM:SS"稳定跳动, 不依赖 /health 的 now。
       lastRefresh.value = new Date().toTimeString().slice(0, 8);
@@ -4830,7 +4880,7 @@ createApp({
     const _rf = _railLoad();
     const lFold = ref(!!_rf.l), rFold = ref(!!_rf.r);
     const railOpen = ref(null);            // null / 'l' / 'r'
-    const sectClosed = reactive({ prop:false, strat:false, pool:false });
+    const sectClosed = reactive({ prop:false, strat:false, pool:false, dividend:false });
     // 提议卡上「展开证据」的状态, 按提议号记。用整体替换而不是原地增删,
     // 因为 Vue 对 Set 的方法调用不作响应, 替换引用才会重画。
     const propMoreOpen = ref(new Set());
@@ -4948,6 +4998,7 @@ createApp({
       autoScanRes, autoScanBusy, autoScanPreview, stageTagType, autoInfo,
       openStrategy, validateStrategy, attachStrategy, setStrategyStatus, stratStateText, resumeBuy,
       showLedger, onPosExpand,
+      dividend, dividendShown, dividendBusy, dividendMore, prepareDividend, loadDividend,
       openScan, insTab, todayYmd, insLive, insDone, insEnd, dispOf, conOf, loadOpenScan, denyToday,
       commandsShown, strategiesShown, instrLive, showAllCommands, showAllStrategies, showAllInstr,
       cmdArchivable, stratArchivable, archiveRow,

scripts/run_tests.py

@@ -118,6 +118,7 @@
                         开关、四类交人、逻辑存疑放弃、深亏补仓三分支、目标价直接采纳、每日上限、record/decline
                         降级)/mark_decide 优先级/eval_dca 深档打 decide_kind/三参数登记与范围/两个 repo 查询
                         + 评审补: 真通过只认带应答体的 PASS、弱表态按强弱打标记 (24 例)
+  test_batch44_units.py 高股息独立来源、后台准备、快速采纳及失败守成 (24例)
   test_batch35_units.py 单测隔离：观察读数内存仓储、整轮扫描与覆盖重建、
                         异常吞掉仍判联网失败、配置隔离与独立入口护栏 (8 例)
   test_batch36_units.py 命令按方案额结算：原分支回归、未解锁批次收尾与故障重试 (13 例)
@@ -194,7 +195,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
           "test_batch35_units.py",
           "test_batch36_units.py",
           "test_batch37_units.py",
-          "test_batch38_units.py", "test_batch39_units.py", "test_batch40_units.py", "test_batch41_units.py", "test_batch42_units.py", "test_batch43_units.py",
+          "test_batch38_units.py", "test_batch39_units.py", "test_batch40_units.py", "test_batch41_units.py", "test_batch42_units.py", "test_batch43_units.py", "test_batch44_units.py",
           "test_page_copy_guard.py",
           "test_page_enum_guard.py",
           "test_page_wiring_guard.py",

scripts/test_batch44_units.py

@@ -0,0 +1,216 @@
+"""开发机：python scripts/unit_test_guard.py scripts/test_batch44_units.py。高股息来源，隔离禁网。"""
+import unit_test_guard
+import os,sys,copy,json
+from datetime import datetime,timedelta
+from types import SimpleNamespace
+from unittest.mock import patch
+from concurrent.futures import ThreadPoolExecutor
+import unittest
+sys.path.insert(0,os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from sqlalchemy import text
+import test_batch43_units as base43
+from app.services import dividend_feed as df,plan_recheck as rc,proposal_service as ps,portfolio,command_service,param_store
+from app.core import plan_revision as pr,action_engine as ae,tradedays as td
+from app.repo import pms_repo
+from app.web import main,auth
+
+CODE='300693.SZ'
+REF='a'*64
+
+def source():
+    now=datetime.now().astimezone()
+    decision={'id':'dividend_'+'b'*64,'source':'dividend','policy_version':'2026.10-1','cutoff_at':now.isoformat()}
+    c={'ts_code':CODE,'name':'合成公司','candidate_source':'dividend','src':'dividend','bucket':'dividend','verdict':'候选',
+       'decision':decision,'company_review':{'overall':'好','age_days':1},'company_review_text':'合成研究依据',
+       'review_ref':{'ref':'original-report'},'dividend_research':{'ref':REF,'conclusion':'有支持'},'rank':1,'selection_rank':1,
+       'risk':['需求可能下降'],'reasons':['高股息研究核验通过']}
+    return {'schema_version':1,'source':'dividend','data_date':str(td.prev_trade_day(now.date())),
+            'policy_version':'2026.10-1','checked_at':now.isoformat(),'decision':decision,'eligible_count':1,
+            'rows':[{'ts_code':CODE,'name':'合成公司','eligible':True,'reasons':[],'research_ref':REF,'candidate':c}]}
+
+class DividendPathTest(unittest.TestCase):
+    decide=base43.AcceptancePathTest.decide
+    counts=base43.AcceptancePathTest.counts
+    def setUp(self):
+        base43.AcceptancePathTest.setUp(self)
+        self.data=source()
+        self.patch('app.services.param_store.get',side_effect=lambda k,d=None:'http://bridge' if k=='PMS_PLAN_API_BASE' else d)
+        self.http.side_effect=lambda url,**kw:self.response(url,kw)
+        self.patch('app.core.tradedays.calendar_degraded',return_value=False)
+        self.patch('app.core.tradedays.is_trade_day',return_value=True)
+        # Existing switch remains off: new source cannot bypass preparation/acceptance checks.
+        self.patch('app.services.param_store.get_bool',side_effect=lambda k,d=False:False if k=='PMS_PLAN_VERSION_RECHECK' else d)
+        c=df.candidates(record=False,recheck_code=CODE)['items'][0]
+        self.hard.update(candidate_source='dividend',dividend_ref=REF,dividend_policy_version='2026.10-1',plan_basis=c['plan_basis'])
+        self.save(self.hard)
+    def patch(self,target,**kw):
+        p=patch(target,**kw);r=p.start();self.addCleanup(p.stop);return r
+    def response(self,url,kw):
+        self.assertEqual(url,'http://bridge/research/dividend/candidates')
+        self.assertEqual(kw['timeout'],(1,3))
+        return SimpleNamespace(raise_for_status=lambda:None,json=lambda:copy.deepcopy(self.data))
+    def save(self,hard):
+        with self.engine.begin() as c:c.execute(text('UPDATE pms_proposal SET hard_numbers_json=:h WHERE proposal_id=:p'),{'h':pms_repo._dumps(hard),'p':self.pid})
+    def test_ready_accepts_without_research_and_retains_source(self):
+        self.assertTrue(self.decide()['ok']);self.assertEqual(self.counts(),(1,1))
+        with self.engine.connect() as c: row=c.execute(text('SELECT progress_json FROM pms_instruction')).scalar()
+        self.assertEqual(json.loads(row)['plan']['candidate_source'],'dividend')
+        self.assertEqual(json.loads(row)['plan']['dividend_ref'],REF)
+    def test_only_rank_and_age_change_do_not_reask(self):
+        self.data['rows'][0]['candidate'].update(rank=900,selection_rank=900)
+        self.data['rows'][0]['candidate']['company_review']['age_days']=9
+        self.assertTrue(self.decide()['ok'])
+    def test_company_research_change_closes_confirmation(self):
+        self.data['rows'][0]['research_ref']='c'*64
+        self.data['rows'][0]['candidate']['dividend_research']['ref']='c'*64
+        r=self.decide();self.assertTrue(r['pending']);self.assertEqual(self.counts(),(0,0))
+    def test_loss_of_eligibility_withdraws(self):
+        self.data['rows'][0].update(eligible=False,reasons=['资料不完整'])
+        self.data['rows'][0]['candidate']['verdict']='仅展示'
+        r=self.decide();self.assertTrue(r['withdrawn']);self.assertEqual(self.counts()[0],0)
+        self.assertEqual(pms_repo.get_proposal(self.pid)['status'],'EXPIRED')
+    def test_other_source_open_commitment_withdraws(self):
+        self.commitments['names'].add(CODE)
+        self.assertTrue(self.decide()['withdrawn']);self.assertEqual(self.counts()[0],0)
+    def test_unavailable_keeps_existing_prepared_suggestion(self):
+        import requests
+        self.http.side_effect=requests.Timeout('network')
+        self.assertTrue(self.decide()['retryable']);self.assertEqual(self.counts(),(0,0))
+        self.assertEqual(pms_repo.get_proposal(self.pid)['hard_numbers']['preparation']['status'],'ready')
+    def test_pending_not_accepted_when_legacy_switch_off(self):
+        self.hard['preparation']={'status':'pending','reason':'等待准备'};self.save(self.hard)
+        self.assertTrue(self.decide()['pending']);self.assertEqual(self.counts(),(0,0))
+        self.patch('app.services.tech_service.attach')
+        r=main.api_proposals(status='WAIT_USER',limit=100,include_archived=False)
+        self.assertFalse(r['data'][0]['accept_ready'])
+    def test_concurrent_accepts_use_one_transaction(self):
+        with ThreadPoolExecutor(2) as pool: rows=list(pool.map(lambda _:self.decide(),range(2)))
+        self.assertTrue(all(r['ok'] for r in rows),rows);self.assertEqual(self.counts(),(1,1))
+    def test_bad_contract_future_date_calendar_or_reference_refuses(self):
+        for change in (lambda d:d.update(data_date='2000-01-01'),lambda d:d['decision'].update(cutoff_at='2099-01-01T00:00:00+08:00'),lambda d:d['rows'][0]['candidate'].update(decision={}),lambda d:d['rows'][0]['candidate']['dividend_research'].update(ref='bad')):
+            self.data=source();change(self.data)
+            with self.assertRaises(df.PlanFeedError):df.read(code=CODE)
+        self.data=source()
+        with patch.object(td,'calendar_degraded',return_value=True):
+            with self.assertRaises(df.PlanFeedError):df.read(code=CODE)
+    def test_readonly_is_not_consumption_and_failed_record_blocks(self):
+        d=df.candidates(record=False,recheck_code=CODE);self.assertNotIn('error',d['consumption'])
+        with patch.object(pms_repo,'insert_plan_snapshot',side_effect=RuntimeError('save failed')):
+            d=df.candidates(record=True,recheck_code=CODE)
+        self.assertEqual(d['items'][0]['consumption_error'],'RuntimeError')
+    def test_snapshot_uses_source_and_compatible_consumption_kind(self):
+        with patch.object(pms_repo,'insert_plan_snapshot',return_value=1) as write:
+            d=df.candidates(record=True,recheck_code=CODE)
+        kw=write.call_args.kwargs;self.assertEqual(kw['meta']['source'],'dividend');self.assertEqual(kw['meta']['kind'],'rechecked')
+        self.assertEqual(kw['meta']['order'],[CODE]);self.assertEqual(kw['roster'][0]['plan_basis'],d['items'][0]['plan_basis'])
+    def test_first_background_prepare_dispatches_original_source(self):
+        h=copy.deepcopy(self.hard);h.update(preparation={'status':'pending','stage':'requested'},plan_basis=None)
+        self.save(h)
+        current=df.candidates(record=False,recheck_code=CODE)['items'][0]
+        fresh={'ts_code':CODE,'action':'OPEN','side':'buy','qty':500,'reason':'高股息建仓','price':34.18,
+               'hard_numbers':dict(self.hard,plan_basis=current['plan_basis'],consumption_id=current['consumption_id'])}
+        with patch.object(ps,'_scan_open',return_value=[fresh]) as scan, patch.object(ps,'_route_one',return_value={'verdict':{'verdict':'PASS','reason':'合成核验','confidence':0.8}}) as route, patch.object(pms_repo,'insert_plan_snapshot',return_value=1):
+            r=rc.refresh_one(pms_repo.get_proposal(self.pid))
+        self.assertTrue(r['ok'],r);self.assertEqual(scan.call_args.kwargs['candidate_source'],'dividend')
+        self.assertTrue(route.call_args.kwargs['review_only']);self.assertEqual(self.counts()[0],0)
+        self.assertEqual(pms_repo.get_proposal(self.pid)['hard_numbers']['preparation']['status'],'ready')
+    def test_legacy_off_still_prepares_only_dividend(self):
+        p=pms_repo.get_proposal(self.pid);other=copy.deepcopy(p);other['hard_numbers'].pop('candidate_source');other['proposal_id']='old'
+        with patch.object(pms_repo,'list_proposals',return_value=[p,other]),patch.object(rc,'refresh_one',return_value={'ok':True}) as call:
+            result=rc.refresh_pending()
+        self.assertEqual(len(result),1);self.assertEqual(call.call_args.args[0]['proposal_id'],self.pid)
+
+class RequestTest(unittest.TestCase):
+    def setUp(self):
+        self.data=source();self.rows={}
+        self.mock('app.services.dividend_feed.read',side_effect=lambda **kw:copy.deepcopy(self.data))
+        self.mock('app.core.tradedays.is_trade_day',return_value=True)
+        self.mock('app.core.tradedays.calendar_degraded',return_value=False)
+        fixed=datetime.now().replace(hour=10,minute=0)
+        d=self.mock('app.services.dividend_feed.datetime');d.now.return_value=fixed
+        self.mock('app.services.param_store.get',side_effect=lambda k,d=None:d)
+        self.mock('app.services.param_store.get_bool',side_effect=lambda k,d=False:d)
+        self.mock('app.repo.pms_repo.get_proposal',side_effect=lambda pid:self.rows.get(pid))
+        self.insert=self.mock('app.repo.pms_repo.insert_proposal',side_effect=self.add)
+        self.mock('app.services.command_service.effective_stock_params',return_value={})
+        self.mock('app.services.portfolio.positions_view',return_value={'held':[]})
+        self.mock('app.services.execution_conflicts.buy_blocker',return_value=None)
+        self.mock('app.services.proposal_service._inflight_open_commitments',return_value={'names':set(),'amount':0,'basis':[]})
+        self.mock('app.services.proposal_service.proposal_expire_at',return_value=fixed.replace(hour=15))
+        self.mock('app.repo.pms_repo.insert_instruction',side_effect=AssertionError('请求不得生成指令'))
+        self.mock('app.services.proposal_service._route_one',side_effect=AssertionError('请求不得调用研判'))
+    def mock(self,key,**kw):
+        p=patch(key,**kw);v=p.start();self.addCleanup(p.stop);return v
+    def add(self,**kw):
+        if kw['proposal_id'] in self.rows:raise RuntimeError('duplicate')
+        self.rows[kw['proposal_id']]=dict(kw,status='WAIT_USER');return 1
+    def request(self):return df.request_preparation(CODE,REF,actor='test')
+    def test_request_has_zero_quantity_and_no_approval(self):
+        r=self.request();p=self.rows[r['proposal_id']]
+        self.assertEqual(p['qty'],0);self.assertEqual(p['hard_numbers']['preparation']['stage'],'requested')
+        self.assertEqual(p['hard_numbers']['requested_by'],'test')
+    def test_repeated_request_returns_existing_without_overwrite(self):
+        a=self.request();b=self.request();self.assertEqual(a['proposal_id'],b['proposal_id']);self.assertEqual(self.insert.call_count,1)
+    def test_terminal_proposal_not_revived(self):
+        p=self.request();self.rows[p['proposal_id']]['status']='DECLINED'
+        with self.assertRaises(ValueError):self.request()
+    def test_missing_support_or_changed_reference_does_not_queue(self):
+        self.data['rows'][0]['eligible']=False
+        with self.assertRaises(ValueError):self.request()
+        self.data['rows'][0]['eligible']=True
+        with self.assertRaises(ValueError):df.request_preparation(CODE,'old',actor='test')
+        self.assertEqual(self.insert.call_count,0)
+    def test_held_black_selling_inflight_or_failure_block(self):
+        cases=[('app.services.portfolio.positions_view',{'held':[{'ts_code':CODE}]}),
+               ('app.services.command_service.effective_stock_params',{CODE:{'black':True}}),
+               ('app.services.execution_conflicts.buy_blocker','正在清仓'),
+               ('app.services.proposal_service._inflight_open_commitments',{'names':{CODE}})]
+        for target,value in cases:
+            with patch(target,return_value=value):
+                with self.assertRaises(ValueError):self.request()
+        with patch.object(ps,'_inflight_open_commitments',side_effect=RuntimeError('db failure')):
+            with self.assertRaises(RuntimeError):self.request()
+        self.assertEqual(self.insert.call_count,0)
+    def test_duplicate_insert_race_reuses_winner(self):
+        def raced(**kw):self.add(**kw);raise RuntimeError('duplicate')
+        self.insert.side_effect=raced
+        r=self.request();self.assertTrue(r['already_requested']);self.assertEqual(len(self.rows),1)
+    def test_api_authorization_requires_login_not_admin(self):
+        self.assertFalse(auth.is_public_path('/api/dividend/proposals'))
+        self.assertEqual(auth.authz_decision('POST','/api/dividend/proposals'),'any')
+    def test_original_basis_unchanged(self):
+        r={'score':1,'decision':{'id':'x'}}
+        self.assertNotIn('candidate_source',pr.basis(r));self.assertNotIn('dividend_ref',pr.basis(r))
+
+
+class AssemblyTest(unittest.TestCase):
+    def test_real_scan_uses_existing_sizer_and_preserves_dividend_basis(self):
+        from contextlib import ExitStack
+        from test_batch20_units import params,caps
+        data=source()
+        view={'held':[],'params':{'max_names':20,'scale':2000000,'portfolio_cap':0.7},
+              'totals':{'names_count':0,'portfolio_mv':0,'cash_source':portfolio.CASH_WS,'cash_avail':500000}}
+        p=params(consensus_route=False,plan_by_advice=False)
+        settings=[(df,'read',data),(ps,'_macro_gate',{}),(param_store,'get','plan_api'),
+                  (portfolio,'caps_ctx',caps()),(ps,'_inflight_open_commitments',{'names':set(),'amount':0,'basis':[]}),
+                  (ps,'_buy_signals_today',{}),(ps,'_scan_params',p),(ps.market,'get_prices',{CODE:10}),
+                  (ps.market,'get_ma5',10),(ps.market,'day_snapshot',{}),(ps.industry,'get_many',{CODE:'电子'}),
+                  (pms_repo,'insert_plan_snapshot',1)]
+        with ExitStack() as stack:
+            for module,name,value in settings:stack.enter_context(patch.object(module,name,return_value=value))
+            out={'skipped':[]};rows=ps._scan_open(view,p,{}, {},{},out,review_code=CODE,exclude_proposal='own',candidate_source='dividend')
+        self.assertEqual(len(rows),1,out);c=rows[0]
+        expected,_=ae.eval_open(dict(data['rows'][0]['candidate'],price=10,sector='电子'),p,caps(),500000)
+        self.assertEqual(c['qty'],expected['qty']);self.assertEqual(c['hard_numbers']['dividend_ref'],REF)
+        self.assertEqual(c['hard_numbers']['plan_basis']['candidate_source'],'dividend');self.assertIn('高股息建仓',c['reason'])
+    def test_strict_inflight_read_cannot_hide_db_failure(self):
+        with patch.object(pms_repo,'list_instructions',side_effect=RuntimeError('db')):
+            with self.assertRaises(RuntimeError):ps._inflight_open_commitments({'held':[]},strict=True)
+    def test_normal_source_duplicate_does_not_overwrite_new_request(self):
+        c={'ts_code':CODE,'action':'OPEN','qty':500,'reason':'normal','hard_numbers':{}}
+        with patch.object(param_store,'get_int',return_value=24),patch.object(param_store,'get_bool',return_value=False),patch.object(ps,'_apply_advice_tier'),patch.object(pms_repo,'insert_proposal',side_effect=RuntimeError('duplicate')) as put,patch.object(pms_repo,'get_proposal',return_value={'status':'WAIT_USER'}):
+            pid=ps._make_proposal(c,10,{'verdict':'PASS'})
+        self.assertEqual(put.call_count,1);self.assertTrue(pid.endswith('_OPEN'))
+
+if __name__=='__main__':unittest.main()
+

scripts/test_wiring.py

@@ -750,6 +750,7 @@ def _():
             "/api/commands/{command_id}/cancel", "/api/plans", "/api/positions",
             "/api/positions/{ts_code}/lots", "/api/instructions", "/api/ledger", "/api/report",
             "/api/proposals", "/api/proposals/{proposal_id}/decide",
+            "/api/dividend/candidates", "/api/dividend/proposals",
             "/api/proposals/decide-reasons",   # 2026-09-14 系统自决包: 裁决预设理由码表
             "/api/ops/replay",
             "/api/ops/reconcile", "/api/ops/daily-settle", "/api/ops/report",