138 lines
8.2 KiB
Python
138 lines
8.2 KiB
Python
|
# -*- 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 apply_judge(adv: Optional[dict], judge: Optional[dict], conf_min: int = 60) -> Optional[dict]:
|
|||
|
|
"""提议落账前补研判一档: 把握度低于 conf_min 或结论不可用 → 试探仓, 只在人工确认后下。不改数量, 数量交人。"""
|
|||
|
|
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 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
|