# -*- coding: utf-8 -*- """建议建仓方案 (2026-09-09 接入方案第六之二节, 台账 053)。纯函数, 不读库不调模型。 输入六样: 公司质地 (好/中/差, 来自数据基座代码合成)、估值标签 (贵/中/便宜)、赔率 (安全边际三情景)、 逻辑状态 (逻辑成立/强化/存疑/无法判断)、研判结论与把握度 (提议落账前才有)、定价状态。 输出一份「建议方案」: 仓位档、分批、附加条件、持有期限标签、一句话。矩阵一次定死, 只按复盘读数换档位: 质地好 + 估值中或便宜 + 逻辑成立或强化 → 标准仓, 5/2.5/2.5 成, 第二批要求二十日头不看空 质地好 + 估值贵 → 减半仓, 两批各半, 第二批等下一期财报失效条件未触发 质地中 → 标准仓, 第一批 3 成、后两批各 3.5 成, 后两批要求二十日头看多且非高位兑现 研判把握度低或不可用、赔率低于 1 → 试探仓 (1%), 一批, 只在人工确认后下 定价状态高位兑现 → 档位不变, 首批推迟到回撤至支撑 质地差、逻辑存疑 → 不建 (判决与分流已经拦住; 这里只写原因) 没有质地读数 (证据不足 / 没报告) → 沿用机械方案, 只标「质地未知」 开关 PMS_PLAN_BY_ADVICE (模拟侧默认开): 开着时建议方案直接作为提议的仓位档与分批; 关着时只并排显示。 止损定性与风控卖出不动; 择时决策系统照旧决定每一批的下单时点。 """ from __future__ import annotations from typing import Any, Optional TIER_STD, TIER_HALF, TIER_TRIAL, TIER_NONE = "标准仓", "减半仓", "试探仓", "不建" DEFAULT_SPLIT = (0.5, 0.25, 0.25) def _f(v, d=None): try: return float(v) except (TypeError, ValueError): return d def odds_of(valuation: Optional[dict]) -> Optional[float]: """赔率 = 中性上行 ÷ 悲观下行 (选股系统安全边际三情景的口径)。算不出返回 None。""" if not isinstance(valuation, dict): return None neut, pess, price = _f(valuation.get("neut")), _f(valuation.get("pess")), _f(valuation.get("price")) if not neut or not pess or not price or price <= 0: return None up, down = neut / price - 1.0, pess / price - 1.0 if down >= 0: return 9.99 # 悲观情景都在现价之上: 赔率极好, 封顶 return round(up / abs(down), 2) if up > 0 else 0.0 def advise(c: dict, params: dict) -> dict[str, Any]: """候选产出前的建议 (没有研判结论)。返回的 tier / factor / splits 由 params_for 变成评估参数。""" cr = c.get("company_review") if isinstance(c.get("company_review"), dict) else {} overall = cr.get("overall") valuation_label = cr.get("valuation") ls = (c.get("logic_state") or {}).get("state") if isinstance(c.get("logic_state"), dict) else None ps = (c.get("pricing_state") or {}).get("state") if isinstance(c.get("pricing_state"), dict) else None odds = odds_of(c.get("valuation")) base_pct = _f(params.get("stock_target_default"), 0.06) base_split = tuple(params.get("batch_split") or DEFAULT_SPLIT) adv: dict[str, Any] = {"tier": TIER_STD, "factor": 1.0, "splits": base_split, "conditions": [], "why": [], "hold_type": "待定", "delay_first": False, "quality": overall, "valuation_label": valuation_label, "logic_state": ls, "pricing_state": ps, "odds": odds, "mechanical": {"target_pct": base_pct, "splits": base_split}, "version": "1"} if overall == "差" or ls == "逻辑存疑": adv.update(tier=TIER_NONE, factor=0.0, splits=(), hold_type="不适用") adv["why"].append("质地差或逻辑存疑：判决与分流已经拦住，不建") elif overall == "好" and valuation_label in ("中", "便宜") and ls in ("逻辑成立", "逻辑强化", None): adv.update(tier=TIER_STD, factor=1.0, splits=(0.5, 0.25, 0.25), hold_type="长期持有类") adv["conditions"].append("第二批要求二十日头不看空") adv["why"].append(f"质地好、估值{valuation_label}、逻辑{ls or '未读到'}") elif overall == "好" and valuation_label == "贵": adv.update(tier=TIER_HALF, factor=0.5, splits=(0.5, 0.5), hold_type="长期持有类") adv["conditions"].append("第二批等下一期财报失效条件未触发再加") adv["why"].append("质地好但估值贵") elif overall == "中": adv.update(tier=TIER_STD, factor=1.0, splits=(0.3, 0.35, 0.35), hold_type="事件驱动类") adv["conditions"].append("后两批要求二十日头看多且定价状态不是高位兑现") adv["why"].append(f"质地中、估值{valuation_label or '无'}") elif overall == "好": adv.update(tier=TIER_STD, factor=1.0, splits=(0.5, 0.25, 0.25), hold_type="长期持有类") adv["conditions"].append("第二批要求二十日头不看空") adv["why"].append(f"质地好、估值{valuation_label or '无'}、逻辑{ls or '未读到'}") else: adv["why"].append("没有质地读数（证据不足或没报告），沿用机械方案") if odds is not None and odds < 1.0 and adv["tier"] not in (TIER_NONE,): adv.update(tier=TIER_TRIAL, factor=1.0 / 6.0, splits=(1.0,)) adv["conditions"].append("只在人工确认后下") adv["why"].append(f"赔率 {odds:.2f} 低于 1") if ps == "高位兑现" and adv["tier"] != TIER_NONE: adv["delay_first"] = True adv["conditions"].append("定价状态高位兑现：首批推迟，等回撤到支撑再下") adv["target_pct"] = round(base_pct * adv["factor"], 4) adv["text"] = text(adv) return adv def advise_v2(c: dict, params: dict, *, fund: dict, tech: dict) -> dict: """基本面 × 技术面仓位矩阵 (2026-09-11 方案附录乙, 台账 006-011)。旧 advise 原样保留, 这是 叠了技术面维度的新函数。fund 是 fund_rules 结果块 (stance/priced_in/valuation), tech 是技术面 立场块 (stance/phase)。返回 adv, 多带 route (放行/观察/交人/跳过) 与 tight_trail (弱基本面 试探仓要配紧止盈)。保留既有调整: 逻辑存疑不建、定价高位兑现首批推迟、赔率低降试探仓; 研判把握度低那一档由 apply_judge 在落账时再降。""" fs = fund.get("stance") ts = (tech or {}).get("stance") or "无读数" base_pct = _f(params.get("stock_target_default"), 0.06) ls = (c.get("logic_state") or {}).get("state") if isinstance(c.get("logic_state"), dict) else None ps = (c.get("pricing_state") or {}).get("state") if isinstance(c.get("pricing_state"), dict) else None odds = odds_of(c.get("valuation")) adv: dict[str, Any] = {"tier": TIER_STD, "factor": 1.0, "splits": DEFAULT_SPLIT, "conditions": [], "why": [], "hold_type": "待定", "delay_first": False, "route": "放行", "tight_trail": False, "fund_stance": fs, "tech_stance": ts, "priced_in": bool(fund.get("priced_in")), "mechanical": {"target_pct": base_pct, "splits": DEFAULT_SPLIT}, "version": "2"} def finish(why_extra=None): if why_extra: adv["why"].append(why_extra) adv["target_pct"] = round(base_pct * adv["factor"], 4) adv["text"] = text(adv) return adv # 既有硬拦: 逻辑存疑 → 不建 if ls == "逻辑存疑": adv.update(tier=TIER_NONE, factor=0.0, splits=(), route="跳过") return finish("逻辑存疑：判决与分流已拦，不建") # 基本面无读数 → 不建 (没有买方评析) if fs == "无读数": adv.update(tier=TIER_NONE, factor=0.0, splits=(), route="跳过") return finish("没有买方评析（质地无读数或超期）") # 基本面看空 if fs == "看空": if ts == "看多": # 试探仓一批, 交人, 配紧止盈与 SAR 线 adv.update(tier=TIER_TRIAL, factor=1.0 / 6.0, splits=(1.0,), route="交人", tight_trail=True, hold_type="试探仓") adv["conditions"] += ["基本面看空、技术面看多：只给试探仓一批，人工确认后下", "配回撤 3%、硬目标 8% 的紧止盈与 SAR 线"] adv["target_pct"] = 0.01 adv["why"].append("基本面看空但技术面看多") adv["text"] = text(adv) return adv adv.update(tier=TIER_NONE, factor=0.0, splits=(), route="跳过") return finish("基本面看空、技术面未看多") # 基本面看多 / 中性 × 技术面看空 → 观察 (等技术面转向; 已持仓只停增持) if ts == "看空": adv.update(route="观察", hold_type="观察") adv["conditions"].append("技术面看空：进观察，转向后再分批建；已持仓只停增持") return finish("基本面偏多但技术面看空，等技术面转向") # 基本面看多 / 中性 × 技术面看多 / 中性 → 按矩阵定档 if fs == "看多": adv.update(splits=(0.5, 0.25, 0.25), hold_type="长期持有类") adv["conditions"].append("第二批要求技术面不看空") if "贵" in str(fund.get("valuation") or "") or fund.get("priced_in"): adv.update(tier=TIER_HALF, factor=0.5, splits=(0.5, 0.5)) if fund.get("priced_in"): adv["delay_first"] = True adv["conditions"].append("共识已定价：减半仓且首批推迟") else: adv["conditions"].append("估值贵：减半仓") adv["why"].append(f"基本面看多、技术面{ts}") else: # 基本面中性 adv.update(splits=(0.3, 0.35, 0.35), hold_type="事件驱动类") adv["conditions"].append("后两批要求技术面看多且非高位兑现") adv["why"].append(f"基本面中性、技术面{ts}") # 既有叠加: 赔率低 → 试探; 高位兑现 → 首批推迟 if odds is not None and odds < 1.0: adv.update(tier=TIER_TRIAL, factor=1.0 / 6.0, splits=(1.0,)) adv["conditions"].append("只在人工确认后下") adv["why"].append(f"赔率 {odds:.2f} 低于 1") if ps == "高位兑现": adv["delay_first"] = True adv["conditions"].append("定价状态高位兑现：首批推迟") return finish() def apply_judge(adv: Optional[dict], judge: Optional[dict], conf_min: int = 60) -> Optional[dict]: """提议落账前补研判一档: 把握度低于 conf_min 或结论不可用 → 试探仓, 只在人工确认后下。 这个函数本身只改档位、不改数量 —— 数量由调用方在拿到新档位之后调 resize_to_advice 重算。2026-09-10 之前这里的说明写的是「不改数量, 数量交人」, 而页面上采纳按钮 根本没有填数量的地方, 那个「交人」无处兑现, 结果是卡上写试探仓 1%、实下 3%。 """ if not isinstance(adv, dict) or not isinstance(judge, dict): return adv verdict = str(judge.get("verdict") or "").upper() conf = _f(judge.get("confidence")) low = (verdict in ("UNAVAILABLE", "不可用")) or (conf is not None and conf < conf_min) if low and adv.get("tier") not in (TIER_NONE, TIER_TRIAL): adv = dict(adv) adv.update(tier=TIER_TRIAL, factor=1.0 / 6.0, splits=(1.0,)) adv["conditions"] = list(adv.get("conditions") or []) + ["只在人工确认后下（研判把握度低或不可用）"] adv["why"] = list(adv.get("why") or []) + [f"研判 {verdict or '无结论'}、把握度 {conf if conf is not None else '无'}"] adv["target_pct"] = round(_f((adv.get("mechanical") or {}).get("target_pct"), 0.06) * adv["factor"], 4) adv["needs_user_confirm"] = True adv["text"] = text(adv) return adv def resize_to_advice(adv: Optional[dict], hn: Optional[dict], ts_code: str) -> tuple: """按建议档位把这一批的股数重算一遍。返回 (改动字典, 说明)；不该改或算不出时返回 (None, 原因)。 ## 为什么需要它 候选产出时 advise() 的档位是**进过**数量的（params_for 把 target_pct 塞进 stock_target_default，action_engine 据此拆批）。但研判回来之后 apply_judge 还会 再降一档 —— 把握度低或结论不可用就降成试探仓 —— 而那一步只改档位文字，数量已经 算完了，于是卡上写的和真下的对不上。 2026-09-10 实测：155 上 8 张待确认提议，7 张写着「试探仓（目标 1.0%）」， 而每一张实际都会买 3.00%（标准仓 6% 的第一批），正好三倍。 apply_judge 的函数说明原文是「不改数量, 数量交人」，但页面上采纳按钮根本没有 填数量的地方，这个「交人」无处兑现；采纳时后端直接抄提议里的股数原样落单。 ## 总规模从哪来 不另外去读参数表，直接从硬数字里已有的两个数反推：target_amount 是整只目标金额、 target_pct 是它占的比例，相除就是总规模。这样不会出现「重算用的规模」与 「当初算数量用的规模」不是同一个数的情况。 ## 买不足一手怎么办 抬到一手，但一手的金额不许超过机械档的目标金额。试探仓的本意是小仓位试水， 100 股 264 元的票是 1.32%（仍远小于 6%），抬上去是合理的；而如果一手就已经 超过机械档，说明这只票对当前规模本来就太贵，那就不改数量、交回调用方处置。 """ from app.core.sizer import lot_of, split_batches if not isinstance(adv, dict) or not isinstance(hn, dict): return None, "没有建议方案或硬数字" new_pct = _f(adv.get("target_pct")) old_pct = _f(hn.get("target_pct")) price = _f(hn.get("price")) old_amt = _f(hn.get("target_amount")) if new_pct <= 0 or old_pct <= 0 or price <= 0 or old_amt <= 0: return None, "缺少重算所需的读数（现价、目标金额或目标仓位）" if abs(new_pct - old_pct) < 1e-9: return None, "档位没变，不用重算" scale = old_amt / old_pct want = scale * new_pct lot = lot_of(ts_code) splits = tuple(adv.get("splits") or (1.0,)) # merge=False: 试探仓本来就是一次买满 (splits=(1.0,)), 不要让降档梯子把它改成别的分批。 sp = split_batches(want, price, splits=splits, merge=False, min_lot=lot) if sp.get("ok"): qty = int(sp["batches"][0]["qty"]) full = sum(b["qty"] * price for b in sp["batches"]) return {"qty": qty, "target_pct": round(new_pct, 4), "target_amount": round(full, 2), "base_amount": round(qty * price, 2), "batch_scheme": ",".join(str(x) for x in splits), "note": ""}, "" mech_pct = _f((adv.get("mechanical") or {}).get("target_pct"), old_pct) mech_amt = scale * mech_pct one_lot_amt = lot * price if one_lot_amt > mech_amt: return None, (f"一手 {lot} 股就要 {one_lot_amt:,.0f} 元，" f"超过机械档的 {mech_amt:,.0f} 元（{mech_pct:.0%}），这只票对当前规模太贵") return {"qty": lot, "target_pct": round(one_lot_amt / scale, 4), "target_amount": round(one_lot_amt, 2), "base_amount": round(one_lot_amt, 2), "batch_scheme": "1.0", "note": f"目标 {want:,.0f} 元买不足一手，抬到一手 {lot} 股（{one_lot_amt / scale:.2%}）"}, "" def params_for(adv: dict, params: dict) -> dict: """把建议方案变成评估参数 (开关开着时用): 目标仓位乘档位系数, 分批按建议。不建的返回原参数 (判决已拦)。""" if not isinstance(adv, dict) or adv.get("tier") in (None, TIER_NONE): return params out = dict(params) out["stock_target_default"] = adv.get("target_pct") or params.get("stock_target_default") if adv.get("splits"): out["batch_split"] = tuple(adv["splits"]) return out def text(adv: dict) -> str: if not isinstance(adv, dict): return "" if adv.get("tier") == TIER_NONE: return "建议方案：不建（" + "；".join(adv.get("why") or []) + "）" sp = "/".join(f"{x:.0%}" for x in (adv.get("splits") or ())) mech = adv.get("mechanical") or {} s = (f"建议方案：{adv.get('tier')}（目标 {adv.get('target_pct', 0):.1%}，分批 {sp or '一批'}）" f"；机械方案 {_f(mech.get('target_pct'), 0):.1%} 分批 {'/'.join(f'{x:.0%}' for x in (mech.get('splits') or ()))}") if adv.get("conditions"): s += "；条件：" + "；".join(adv["conditions"]) if adv.get("delay_first"): s += "；首批推迟" s += f"；{adv.get('hold_type')}；依据：" + "；".join(adv.get("why") or []) return s