第四件：安全边际三情景，只展示不进判决
sources：券商研报原始行取一次（broker_reports，带市盈率列），券商行动改用共用分箱 （行为不变，单测钉住）；新增 scenarios（悲观=最低每股收益×最低市盈率、中性=两项中位数、 乐观=两项最高，隐含市盈率、赔率、机构分歧标注）、valuation_scenarios（优先年度预测期、 同机构只留最近）、close_prices（与预期空间同源的前复权收盘价，左闭右开区间）。 card：valuation_view 整句与 valuation_short 短写法，四种不适用与两种赔率不成立各一句人话。 plan：候选卡带 valuation 原值与整句，候选单表格加一列并附口径说明。 测试：新建 test_valuation.py，恩捷手算对表（悲观 42.64、中性 56.75、乐观 84.96、隐含 21.4、赔率 0.87）。 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
zlt
parent
54607015f9
commit
e00b38e36b
46
card.py
46
card.py
|
|
@ -326,3 +326,49 @@ def upside_text(upside, neg_tol: float = 0.0) -> str:
|
|||
if float(neg_tol) > 0:
|
||||
return f"券商给的目标价比现价低 {gap:.0%}，最多只接受低 {float(neg_tol):.0%}"
|
||||
return f"券商给的目标价比现价低 {gap:.0%}，目标价低于现价的不买"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 安全边际三情景的那一行（2026-09-07 下一阶段方案第四件）：只展示，不进判决
|
||||
# ============================================================================
|
||||
|
||||
def _period_cn(q) -> str:
|
||||
"""预测期写成人话：2026Q4 是 2026 年度，2026Q2 是 2026 年中期，认不出的原样。"""
|
||||
s = str(q or "").strip().upper()
|
||||
if len(s) == 6 and s[:4].isdigit() and s[4] == "Q":
|
||||
return f"{s[:4]} 年度" if s[5] == "4" else f"{s[:4]} 年{'一季' if s[5] == '1' else '中期' if s[5] == '2' else '三季'}"
|
||||
return s or "未知预测期"
|
||||
|
||||
|
||||
def valuation_view(scn) -> str:
|
||||
"""把三情景估值写成一句完整的话。数据缺席、四种不适用、算得出三种情形各有各的写法。
|
||||
|
||||
赔率是中性上行对悲观下行，写成"0.87 比 1"——不到一比一就是说这个位置的赔率不吸引人。
|
||||
这一行回答的是"最坏情况下现价还有多少下跌空间"，是事前的安全边际；与建仓后的浮盈无关。
|
||||
"""
|
||||
if not isinstance(scn, dict):
|
||||
return "安全边际：近三个月没有券商的盈利预测，算不出"
|
||||
if scn.get("na"):
|
||||
return f"安全边际算不出：{scn['na']}"
|
||||
head = f"安全边际（{_period_cn(scn.get('quarter'))}预测，{scn.get('firms')} 家）"
|
||||
body = (f"悲观 {scn['pess']:.2f} 元（{scn['down']:+.1%}）、中性 {scn['neut']:.2f} 元（{scn['up_neut']:+.1%}）、"
|
||||
f"乐观 {scn['opt']:.2f} 元（{scn['up_opt']:+.1%}）；隐含市盈率 {scn['implied_pe']:.1f} 倍")
|
||||
if scn.get("odds") is not None:
|
||||
tail = f"赔率 {scn['odds']:.2f} 比 1（中性上行对悲观下行）"
|
||||
else:
|
||||
tail = scn.get("note") or "赔率不成立"
|
||||
if scn.get("wide"):
|
||||
sp = scn.get("spread") or {}
|
||||
tail += (f"。机构分歧极大（每股收益最高是最低的 {sp.get('eps', 0):.1f} 倍、"
|
||||
f"市盈率 {sp.get('pe', 0):.1f} 倍），两头的数字只当参考")
|
||||
return f"{head}：{body}；{tail}"
|
||||
|
||||
|
||||
def valuation_short(scn) -> str:
|
||||
"""表格里放得下的短写法：悲观下行 / 中性上行，赔率。"""
|
||||
if not isinstance(scn, dict):
|
||||
return "—"
|
||||
if scn.get("na"):
|
||||
return "算不出"
|
||||
odds = f"赔率 {scn['odds']:.2f}" if scn.get("odds") is not None else "赔率不成立"
|
||||
return f"{scn['down']:+.0%}/{scn['up_neut']:+.0%}，{odds}" + ("（分歧极大）" if scn.get("wide") else "")
|
||||
|
|
|
|||
24
plan.py
24
plan.py
|
|
@ -228,11 +228,14 @@ def _logic_inputs(codes: list, ds: str) -> dict:
|
|||
hist 是逐票日频表里每只票的近日行，给抗抖动用（2026-09-07 第三件桥侧前置）；表没建或
|
||||
读不到为空字典，那时落定态等于原始态，计划照出。这张表只在早上 generate 里写，这里只读。
|
||||
"""
|
||||
# 券商研报原始行取一次，券商行动（丙路）与安全边际三情景（第四件）共用——两路看同一批研报。
|
||||
broker_rows = sources.broker_reports(codes, ds)
|
||||
return {
|
||||
"logic_full": sources.logic_claims(codes, ds, per_stock=config.LOGIC_CLAIMS_FULL),
|
||||
"seg_view": judgement.by_segment_name(judgement.load_previous(_next_day(ds))),
|
||||
"seg_hist": judgement.recent_rows(_next_day(ds), days=config.JUDGEMENT_HOLD_DAYS),
|
||||
"broker": sources.broker_actions(codes, ds),
|
||||
"broker_rows": broker_rows,
|
||||
"broker": sources.broker_actions(codes, ds, rows=broker_rows),
|
||||
"seg_of": _segments_of(ds),
|
||||
"hist": logic_state_daily.history(ds, codes=None if len(codes) > 50 else codes),
|
||||
}
|
||||
|
|
@ -278,6 +281,11 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
|
|||
logic_full, seg_view, seg_hist = inp["logic_full"], inp["seg_view"], inp["seg_hist"]
|
||||
broker, seg_of, hist = inp["broker"], inp["seg_of"], inp["hist"]
|
||||
logic = {k: v[:config.LOGIC_CLAIMS_PER_STOCK] for k, v in logic_full.items()}
|
||||
# 安全边际三情景（2026-09-07 第四件）：按同一预测期的每股收益与市盈率预测算悲观、中性、乐观
|
||||
# 三个估值与现价的差，只展示不进判决。现价与预期空间同源（前复权行情表）；两处任一读不到，
|
||||
# 卡上那一行写"算不出"并说明原因，不拦票、不断产。
|
||||
prices = sources.close_prices(ds)
|
||||
valuation = sources.valuation_scenarios(codes, ds, prices, rows=inp["broker_rows"])
|
||||
if risk is None: # collect 会传入读过一次的名单；单独调用时自己读
|
||||
try:
|
||||
risk = factors._risk_set() or set() # noqa: SLF001 —— 同仓自用
|
||||
|
|
@ -311,6 +319,7 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
|
|||
require_started=config.CARD_REQUIRE_STARTED)
|
||||
cards[k] = {
|
||||
**j, "logic_state": state,
|
||||
"valuation": valuation.get(k), "valuation_text": card.valuation_view(valuation.get(k)),
|
||||
"theme": theme, "n_sources": n_sources, "chain_fit": evd["chain_fit"],
|
||||
"started_source": "moved_view" if mv else None,
|
||||
"logic_claims": evd["logic"],
|
||||
|
|
@ -437,6 +446,8 @@ def collect(date: str | None = None, top: int = 20, obs_top: int = 10,
|
|||
# 截止日，不发权重也不发判决改动——四态怎么作用于建仓通道是 PMS
|
||||
# 那边的事，这里只提供状态与出处。
|
||||
logic_state=_state_out(c.get("logic_state")),
|
||||
# 安全边际三情景（2026-09-07 第四件）：原值给程序，整句给人；只展示不进判决。
|
||||
valuation=c.get("valuation"), valuation_text=c.get("valuation_text"),
|
||||
card={"pct0": c.get("pct0"), "net_z": c.get("net_z"),
|
||||
"heat_chg": c.get("heat_chg"), "accum": c.get("accum"),
|
||||
"night": c.get("night"), "gates": c.get("gates"),
|
||||
|
|
@ -675,17 +686,22 @@ def render_md(d: dict) -> str:
|
|||
if not cands:
|
||||
L.append("（今日无候选——候选为空不是故障：环节没被指向、成员没启动或没有明确吸筹，都会为空。）")
|
||||
else:
|
||||
L.append("| # | 代码 | 名称 | 环节 | 源数 | 当日涨幅 | 吸筹 | 预期空间 | 理由 | 因果论断（出处） |")
|
||||
L.append("|---|------|------|------|------|----------|------|----------|------|------------------|")
|
||||
L.append("| # | 代码 | 名称 | 环节 | 源数 | 当日涨幅 | 吸筹 | 预期空间 | 安全边际（悲观下行/中性上行，赔率） "
|
||||
"| 理由 | 因果论断（出处） |")
|
||||
L.append("|---|------|------|------|------|----------|------|----------|----------|------|------------------|")
|
||||
for r in cands:
|
||||
c = r.get("card") or {}
|
||||
ac = c.get("accum") or {}
|
||||
ev_ = r.get("evidence") or {}
|
||||
L.append(f"| {r['rank']} | {r['code']} | {r['name'] or '—'} | {ev_.get('theme') or '—'} "
|
||||
f"| {ev_.get('n_sources') or '—'} | {_fmt_pct0(c.get('pct0'))} "
|
||||
f"| {_fmt_accum(ac)} | {_fmt_pct(r.get('upside'))} "
|
||||
f"| {_fmt_accum(ac)} | {_fmt_pct(r.get('upside'))} | {card.valuation_short(r.get('valuation'))} "
|
||||
f"| {'；'.join(r.get('reasons') or [])} | {_fmt_logic(r.get('logic'))} |")
|
||||
L.append("")
|
||||
L.append("安全边际按同一预测期的每股收益与市盈率预测算：悲观是最低每股收益乘最低市盈率，中性是两项中位数，"
|
||||
"乐观是两项最高；赔率是中性上行对悲观下行，不到一比一就是这个位置的赔率不吸引人。"
|
||||
"它只展示、不进判决，与上面用目标价平均算的预期空间是两个口径。")
|
||||
L.append("")
|
||||
segs = d.get("segments_pointed") or []
|
||||
L.append(f"## 关注环节（今日被传导指向的 {len(segs)} 个环节：定位对不对看这里，挑票看候选单）")
|
||||
L.append("")
|
||||
|
|
|
|||
235
sources.py
235
sources.py
|
|
@ -570,8 +570,55 @@ def _latest_row(rmy, src: str, table: str) -> tuple[dict | None, str | None]:
|
|||
BROKER_WINDOW_DAYS = 45
|
||||
|
||||
|
||||
def broker_reports(codes, ds: str, *, days: int = BROKER_WINDOW_DAYS * 2, read_mysql=None) -> list:
|
||||
"""券商研报明细表里这些票近 days 个自然日的每股收益与市盈率预测——原始行，按机构分箱之前。
|
||||
|
||||
券商行动（两个等长窗口）与安全边际三情景（整段窗口）共用这一次取数，两路看的是同一批研报。
|
||||
每行归一成 {k 前缀码, date 报告日, quarter 预测期, org 机构, eps, pe}；没有每股收益或
|
||||
预测期的行不要。读失败返回空列表并打印原因，两路都按缺席处理，计划不断产。
|
||||
"""
|
||||
reader = read_mysql or db.read_mysql
|
||||
end = dt.date.fromisoformat(ds)
|
||||
start = end - dt.timedelta(days=int(days))
|
||||
dotted = sorted({_to_dot(c) for c in codes if c})
|
||||
if not dotted:
|
||||
return []
|
||||
marks = ",".join(["%s"] * len(dotted))
|
||||
try:
|
||||
df = reader(
|
||||
"factor",
|
||||
f"SELECT ts_code, report_date, quarter, org_name, eps, pe FROM gp_report_rc "
|
||||
f"WHERE ts_code IN ({marks}) AND report_date > %s AND report_date <= %s "
|
||||
f"AND eps IS NOT NULL AND quarter IS NOT NULL",
|
||||
tuple(dotted) + (start.isoformat(), end.isoformat()))
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f" （券商研报明细表读取失败，券商行动与安全边际两路整体缺席: {e!r}）")
|
||||
return []
|
||||
out = []
|
||||
for r in _records(df):
|
||||
d = _ymd(r.get("report_date"))
|
||||
q = str(r.get("quarter") or "").strip()
|
||||
eps = _f(r.get("eps"))
|
||||
if not d or not q or eps is None:
|
||||
continue
|
||||
out.append({"k": common.to_prefix(str(r.get("ts_code") or "").strip()), "date": d,
|
||||
"quarter": q, "org": str(r.get("org_name") or "").strip() or "未署名",
|
||||
"eps": eps, "pe": _f(r.get("pe"))})
|
||||
return out
|
||||
|
||||
|
||||
def _latest_by_org(rows) -> dict:
|
||||
"""同一家机构在窗口里可能发多篇，只留最近一篇：{预测期: {机构: 行}}。"""
|
||||
box: dict = defaultdict(dict)
|
||||
for r in rows:
|
||||
slot = box[r["quarter"]]
|
||||
if r["org"] not in slot or r["date"] > slot[r["org"]]["date"]:
|
||||
slot[r["org"]] = r
|
||||
return box
|
||||
|
||||
|
||||
def broker_actions(codes, ds: str, *, window_days: int = BROKER_WINDOW_DAYS,
|
||||
read_mysql=None) -> dict:
|
||||
read_mysql=None, rows=None) -> dict:
|
||||
"""券商用行动说话这一路：同一财年同一预测期的每股收益预测中位数与覆盖机构数，
|
||||
比较最近两个等长窗口。返回前缀码到 logic_state.signal 的字典（算不出的票不进字典）。
|
||||
|
||||
|
|
@ -583,53 +630,167 @@ def broker_actions(codes, ds: str, *, window_days: int = BROKER_WINDOW_DAYS,
|
|||
|
||||
看的是券商的行动不是言辞——券商极少明说不看好某个行业，所以等不到它开口，
|
||||
只能看预测在不在下修、覆盖在不在收缩。
|
||||
|
||||
rows 可传 broker_reports 的返回（计划装配一次取数两路共用）；不传就自己取。
|
||||
"""
|
||||
import logic_state as ls
|
||||
|
||||
reader = read_mysql or db.read_mysql
|
||||
if rows is None:
|
||||
rows = broker_reports(codes, ds, days=int(window_days) * 2, read_mysql=read_mysql)
|
||||
end = dt.date.fromisoformat(ds)
|
||||
mid = end - dt.timedelta(days=int(window_days))
|
||||
start = end - dt.timedelta(days=int(window_days) * 2)
|
||||
dotted = sorted({_to_dot(c) for c in codes if c})
|
||||
if not dotted:
|
||||
return {}
|
||||
marks = ",".join(["%s"] * len(dotted))
|
||||
try:
|
||||
df = reader(
|
||||
"factor",
|
||||
f"SELECT ts_code, report_date, quarter, org_name, eps FROM gp_report_rc "
|
||||
f"WHERE ts_code IN ({marks}) AND report_date > %s AND report_date <= %s "
|
||||
f"AND eps IS NOT NULL AND quarter IS NOT NULL",
|
||||
tuple(dotted) + (start.isoformat(), end.isoformat()))
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f" （券商研报明细表读取失败，券商行动这一路整体缺席: {e!r}）")
|
||||
return {}
|
||||
|
||||
rows = df.itertuples() if hasattr(df, "itertuples") else []
|
||||
box: dict = defaultdict(lambda: defaultdict(lambda: {"now": {}, "prev": {}}))
|
||||
mid = (end - dt.timedelta(days=int(window_days))).isoformat()
|
||||
start = (end - dt.timedelta(days=int(window_days) * 2)).isoformat()
|
||||
by_code: dict = defaultdict(list)
|
||||
for r in rows:
|
||||
d = _ymd(r.report_date)
|
||||
if not d:
|
||||
continue
|
||||
win = "now" if d > mid.isoformat() else "prev"
|
||||
k = common.to_prefix(str(r.ts_code).strip())
|
||||
org = str(r.org_name or "").strip() or "未署名"
|
||||
slot = box[k][str(r.quarter).strip()][win]
|
||||
if org not in slot or d > slot[org][0]: # 同机构多篇只留最近一篇
|
||||
slot[org] = (d, float(r.eps))
|
||||
if start < r["date"] <= ds: # rows 可能来自更长的窗口，这里再裁一次
|
||||
by_code[r["k"]].append(r)
|
||||
|
||||
out = {}
|
||||
for k, by_q in box.items():
|
||||
usable = [(q, v) for q, v in by_q.items() if v["now"] and v["prev"]]
|
||||
for k, rs in by_code.items():
|
||||
now_box = _latest_by_org([r for r in rs if r["date"] > mid])
|
||||
prev_box = _latest_by_org([r for r in rs if r["date"] <= mid])
|
||||
usable = [(q, now_box[q], prev_box[q]) for q in now_box if q in prev_box]
|
||||
if not usable:
|
||||
continue
|
||||
q, v = max(usable, key=lambda kv: len(kv[1]["now"]) + len(kv[1]["prev"]))
|
||||
now = {"eps": statistics.median([x[1] for x in v["now"].values()]),
|
||||
"firms": len(v["now"])}
|
||||
prev = {"eps": statistics.median([x[1] for x in v["prev"].values()]),
|
||||
"firms": len(v["prev"])}
|
||||
q, n, p = max(usable, key=lambda t: len(t[1]) + len(t[2]))
|
||||
now = {"eps": statistics.median([r["eps"] for r in n.values()]), "firms": len(n)}
|
||||
prev = {"eps": statistics.median([r["eps"] for r in p.values()]), "firms": len(p)}
|
||||
sig = ls.from_broker(now, prev, as_of=ds)
|
||||
if sig["refs"]:
|
||||
sig["refs"][0]["quarter"] = q
|
||||
out[k] = sig
|
||||
return out
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 安全边际三情景（2026-09-07 下一阶段方案第四件）：只展示，不进判决
|
||||
# ============================================================================
|
||||
#
|
||||
# 回答的是价值投资的经典问题：买入价相对内在价值的折扣有多少，最坏情况下现价还有多少
|
||||
# 下跌空间。这是事前概念，与 PMS 里那个也叫"垫"的字段（建仓后的浮盈）不是一回事。
|
||||
#
|
||||
# 口径：同一预测期的每股收益预测与市盈率预测，按机构去重后各取最小、中位、最大——
|
||||
# 悲观 = 最低每股收益 × 最低市盈率，中性 = 两项中位数，乐观 = 两项最高。
|
||||
# 隐含市盈率 = 现价 ÷ 中位每股收益；赔率 = 中性上行 ÷ 悲观下行。
|
||||
# 用盈利预测不用目标价：目标价字段覆盖不到三成且不去重不加权，出现过 +181% 的读数；
|
||||
# 每股收益预测覆盖 98%、市盈率预测 91%。恩捷股份 09-02 手算（方案四之三）：悲观 42.64、
|
||||
# 中性 56.75、乐观 84.96，赔率 0.87 比 1——单测用它对表。
|
||||
|
||||
# 机构分歧"极大"的线：每股收益或市盈率预测的最高对最低超过这个倍数就标注。取 3 倍——正常的
|
||||
# 分歧在一倍多到两倍之间（恩捷 1.3 与 1.5 倍），三倍以上多半是有机构的口径不同或数据有误。
|
||||
# 一次定死，只影响标注，不影响算法。
|
||||
WIDE_SPREAD = 3.0
|
||||
|
||||
|
||||
def _pick_period(box: dict):
|
||||
"""挑哪个预测期：优先年度（预测期以 Q4 结尾），其中覆盖机构最多的；同数取更近的年份。"""
|
||||
cands = [(q, len(orgs)) for q, orgs in box.items() if orgs]
|
||||
if not cands:
|
||||
return None
|
||||
annual = [c for c in cands if c[0].upper().endswith("Q4")]
|
||||
pool = annual or cands
|
||||
return sorted(pool, key=lambda c: (-c[1], c[0]))[0][0]
|
||||
|
||||
|
||||
def scenarios(eps_vals, pe_vals, price, *, quarter=None, as_of=None, min_firms: int = 2) -> dict:
|
||||
"""纯函数：三情景估值。算不出时 na 写明原因（四种不适用各一句人话），算得出时 na 为 None。"""
|
||||
eps_vals = [float(x) for x in (eps_vals or []) if x is not None]
|
||||
pe_vals = [float(x) for x in (pe_vals or []) if x is not None and float(x) > 0]
|
||||
base = {"quarter": quarter, "firms": len(eps_vals), "as_of": as_of,
|
||||
"price": None if price is None else float(price)}
|
||||
if len(eps_vals) < int(min_firms):
|
||||
return {**base, "na": f"覆盖机构只有 {len(eps_vals)} 家，不足 {min_firms} 家"}
|
||||
if price is None or float(price) <= 0:
|
||||
return {**base, "na": "现价取不到"}
|
||||
if min(eps_vals) <= 0:
|
||||
return {**base, "na": "每股收益预测有负值或零，市盈率口径不适用"}
|
||||
if len(pe_vals) < int(min_firms):
|
||||
return {**base, "na": f"市盈率预测只有 {len(pe_vals)} 家给了，不足 {min_firms} 家"}
|
||||
px = float(price)
|
||||
e = {"min": min(eps_vals), "med": statistics.median(eps_vals), "max": max(eps_vals)}
|
||||
p = {"min": min(pe_vals), "med": statistics.median(pe_vals), "max": max(pe_vals)}
|
||||
pess, neut, opt = e["min"] * p["min"], e["med"] * p["med"], e["max"] * p["max"]
|
||||
down, up_n, up_o = pess / px - 1, neut / px - 1, opt / px - 1
|
||||
odds = None
|
||||
note = None
|
||||
if down >= 0:
|
||||
note = "悲观情景仍高于现价，没有下行空间可比"
|
||||
elif up_n <= 0:
|
||||
note = "中性情景低于现价，赔率不成立"
|
||||
else:
|
||||
odds = up_n / (-down)
|
||||
# 机构分歧的量：最高对最低的倍数。悲观取最低乘最低、乐观取最高乘最高，分歧一大两头就会被
|
||||
# 放大到离谱（实测有票悲观 -89%、乐观 +1054%）。超过阈值只标注"分歧极大"，不改算法——
|
||||
# 这本身就是一条信息：券商对这家公司的盈利路径没有共识。
|
||||
spread = {"eps": round(e["max"] / e["min"], 2), "pe": round(p["max"] / p["min"], 2)}
|
||||
wide = spread["eps"] > WIDE_SPREAD or spread["pe"] > WIDE_SPREAD
|
||||
return {**base, "na": None,
|
||||
"eps": {k: round(v, 4) for k, v in e.items()},
|
||||
"pe": {k: round(v, 2) for k, v in p.items()},
|
||||
"pess": round(pess, 2), "neut": round(neut, 2), "opt": round(opt, 2),
|
||||
"down": round(down, 4), "up_neut": round(up_n, 4), "up_opt": round(up_o, 4),
|
||||
"implied_pe": round(px / e["med"], 1),
|
||||
"odds": None if odds is None else round(odds, 2), "note": note,
|
||||
"spread": spread, "wide": wide}
|
||||
|
||||
|
||||
def valuation_scenarios(codes, ds: str, prices: dict, *, rows=None,
|
||||
window_days: int = BROKER_WINDOW_DAYS * 2, min_firms: int = 2,
|
||||
read_mysql=None) -> dict:
|
||||
"""每只票的三情景估值，按前缀码索引；没有任何研报行的票给 None（卡上写"没有券商预测"）。"""
|
||||
if rows is None:
|
||||
rows = broker_reports(codes, ds, days=int(window_days), read_mysql=read_mysql)
|
||||
by_code: dict = defaultdict(list)
|
||||
for r in rows:
|
||||
by_code[r["k"]].append(r)
|
||||
out = {}
|
||||
for k in {str(c) for c in (codes or []) if c}:
|
||||
rs = by_code.get(k) or []
|
||||
if not rs:
|
||||
out[k] = None
|
||||
continue
|
||||
box = _latest_by_org(rs)
|
||||
q = _pick_period(box)
|
||||
firms = box.get(q, {}) if q else {}
|
||||
out[k] = scenarios([r["eps"] for r in firms.values()],
|
||||
[r["pe"] for r in firms.values()],
|
||||
(prices or {}).get(k), quarter=q,
|
||||
as_of=max((r["date"] for r in firms.values()), default=None),
|
||||
min_firms=min_firms)
|
||||
return out
|
||||
|
||||
|
||||
def _prefix_any(s: str) -> str:
|
||||
"""行情表的代码列可能是 600000.SH，也可能是裸 6 位码；统一成前缀式。"""
|
||||
s = str(s or "").strip().upper()
|
||||
if "." in s:
|
||||
return common.to_prefix(s)
|
||||
if len(s) == 6 and s.isdigit():
|
||||
return ("SH" if s[0] == "6" else ("BJ" if s[0] in "48" else "SZ")) + s
|
||||
return s
|
||||
|
||||
|
||||
def close_prices(ds: str, read_mysql=None, code_col: str | None = None) -> dict:
|
||||
"""数据日的收盘价（前复权行情表，与预期空间同一来源），按前缀码索引；读不到返回空字典。
|
||||
|
||||
区间写成 [ds, ds+1) 而不是 = ds：时间列若带时分秒，等号会一行都对不上，而这样写两种
|
||||
形态都对、也走得了索引。code_col 可注入（离线单测），不传就沿用预期空间那一路的探列。
|
||||
"""
|
||||
reader = read_mysql or db.read_mysql
|
||||
try:
|
||||
if code_col is None:
|
||||
import factors
|
||||
code_col = factors._price_code_col() # noqa: SLF001 —— 同仓自用
|
||||
nxt = (dt.date.fromisoformat(ds) + dt.timedelta(days=1)).isoformat()
|
||||
df = reader("price", f"SELECT `{code_col}` AS ts_code, close FROM gp_day_data "
|
||||
f"WHERE `timestamp` >= %s AND `timestamp` < %s", (ds, nxt))
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f" （收盘价读取失败，安全边际这一行整体缺席: {e!r}）")
|
||||
return {}
|
||||
out = {}
|
||||
for r in _records(df):
|
||||
k = _prefix_any(r.get("ts_code"))
|
||||
px = _f(r.get("close"))
|
||||
if k and px and px > 0:
|
||||
out[k] = px
|
||||
return out
|
||||
|
|
|
|||
|
|
@ -0,0 +1,166 @@
|
|||
"""安全边际三情景的离线单测（不连库）。2026-09-07 下一阶段方案第四件。
|
||||
|
||||
钉住五件事：
|
||||
一，三情景的算法与方案四之三的恩捷股份手算逐项一致（悲观 42.64、中性 56.75、乐观 84.96，
|
||||
隐含市盈率 21.4 倍，赔率 0.87 比 1）。
|
||||
二，四种不适用各有一句人话：每股收益为负、机构不足两家、市盈率缺、现价缺；另两种赔率不成立的情形也说得清。
|
||||
三，预测期怎么挑：优先年度（Q4）里覆盖最多的，同数取更近的年份；同一机构只留最近一篇。
|
||||
四，取数层共用一次：券商行动（两个等长窗口）改用共用分箱后行为不变；收盘价的代码形态归一与日期区间。
|
||||
五，卡上的整句与表格短写法。
|
||||
|
||||
开发机没有 pandas 与数据库驱动时只给缺席的模块装最小桩（与 test_judgement_snapshot.py 同一约定）。
|
||||
跑法：python3 test_valuation.py 或 pytest test_valuation.py
|
||||
"""
|
||||
import sys
|
||||
import types
|
||||
|
||||
_STUBS = ("pandas", "psycopg", "pymysql", "dotenv")
|
||||
for _n in _STUBS:
|
||||
if _n not in sys.modules:
|
||||
try:
|
||||
__import__(_n)
|
||||
except ImportError:
|
||||
_m = types.ModuleType(_n)
|
||||
if _n == "pandas":
|
||||
_m.DataFrame = type("DataFrame", (), {})
|
||||
_m.Series = type("Series", (), {})
|
||||
sys.modules[_n] = _m
|
||||
|
||||
import card # noqa: E402
|
||||
import logic_state as ls # noqa: E402
|
||||
import sources # noqa: E402
|
||||
|
||||
|
||||
def t(name, cond):
|
||||
assert cond, name
|
||||
print(" ok", name)
|
||||
|
||||
|
||||
DS = "2026-09-02"
|
||||
PRICE = 50.17
|
||||
# 九家机构对 2026 年度的预测，最小、中位、最大与方案四之三的手算对得上。
|
||||
EPS = [2.08, 2.20, 2.30, 2.32, 2.34, 2.37, 2.42, 2.50, 2.68]
|
||||
PE = [20.5, 22.3, 23.6, 24.0, 24.253, 26.9, 27.5, 30.0, 31.7]
|
||||
|
||||
|
||||
def rows_for(k="SZ002812", quarter="2026Q4", eps=EPS, pe=PE, date="2026-08-25"):
|
||||
return [{"k": k, "date": date, "quarter": quarter, "org": f"机构{i}", "eps": e, "pe": p}
|
||||
for i, (e, p) in enumerate(zip(eps, pe))]
|
||||
|
||||
|
||||
def test_scenarios():
|
||||
print("三情景与恩捷手算对表")
|
||||
s = sources.scenarios(EPS, PE, PRICE, quarter="2026Q4", as_of="2026-08-25")
|
||||
t("算得出，na 为空", s["na"] is None and s["firms"] == 9)
|
||||
t("悲观 42.64 元、下行 15.0%", s["pess"] == 42.64 and round(s["down"], 3) == -0.150)
|
||||
t("中性 56.75 元、上行 13.1%", s["neut"] == 56.75 and round(s["up_neut"], 3) == 0.131)
|
||||
t("乐观 84.96 元、上行 69.3%", s["opt"] == 84.96 and round(s["up_opt"], 3) == 0.693)
|
||||
t("隐含市盈率 21.4 倍", s["implied_pe"] == 21.4)
|
||||
t("赔率 0.87 比 1", s["odds"] == 0.87 and s["note"] is None)
|
||||
t("每股收益与市盈率的三个数都带出来", s["eps"]["med"] == 2.34 and s["pe"]["med"] == 24.25)
|
||||
|
||||
print("四种不适用与两种赔率不成立")
|
||||
t("每股收益为负", "负值" in sources.scenarios([-0.5, 1.2], [20, 30], PRICE)["na"])
|
||||
t("机构不足两家", "不足 2 家" in sources.scenarios([2.0], [20], PRICE)["na"])
|
||||
t("市盈率缺（只有一家给了）", "市盈率预测只有 1 家" in sources.scenarios([2.0, 2.2], [20, None], PRICE)["na"])
|
||||
t("市盈率为负的不算数", "市盈率预测只有 0 家" in sources.scenarios([2.0, 2.2], [-3, 0], PRICE)["na"])
|
||||
t("现价缺", sources.scenarios([2.0, 2.2], [20, 30], None)["na"] == "现价取不到")
|
||||
lo = sources.scenarios([2.0, 2.2], [30, 40], 30.0) # 悲观 60 > 现价 30
|
||||
t("悲观仍高于现价：赔率不成立并说明", lo["odds"] is None and "悲观情景仍高于现价" in lo["note"])
|
||||
hi = sources.scenarios([2.0, 2.2], [20, 21], 60.0) # 中性 44.1 < 现价 60
|
||||
t("中性低于现价：赔率不成立并说明", hi["odds"] is None and "中性情景低于现价" in hi["note"])
|
||||
|
||||
|
||||
def test_period_and_org():
|
||||
print("预测期怎么挑、同机构只留最近")
|
||||
rows = (rows_for(quarter="2026Q4", eps=EPS[:3], pe=PE[:3]) +
|
||||
rows_for(quarter="2027Q4", eps=EPS[:5], pe=PE[:5]) +
|
||||
rows_for(quarter="2026Q2", eps=EPS, pe=PE))
|
||||
box = sources._latest_by_org(rows) # noqa: SLF001
|
||||
t("优先年度：2026Q2 覆盖最多也不选，选 Q4 里覆盖最多的 2027Q4", sources._pick_period(box) == "2027Q4") # noqa: SLF001
|
||||
box2 = sources._latest_by_org(rows_for(quarter="2026Q4", eps=EPS[:3], pe=PE[:3]) + # noqa: SLF001
|
||||
rows_for(quarter="2027Q4", eps=EPS[3:6], pe=PE[3:6]))
|
||||
t("同数取更近的年份", sources._pick_period(box2) == "2026Q4") # noqa: SLF001
|
||||
t("没有年度预测时退回覆盖最多的", sources._pick_period(sources._latest_by_org( # noqa: SLF001
|
||||
rows_for(quarter="2026Q2", eps=EPS[:4], pe=PE[:4]) + rows_for(quarter="2026Q3", eps=EPS[:2], pe=PE[:2]))) == "2026Q2")
|
||||
dup = [{"k": "SZ002812", "date": "2026-07-01", "quarter": "2026Q4", "org": "机构A", "eps": 1.0, "pe": 10.0},
|
||||
{"k": "SZ002812", "date": "2026-08-20", "quarter": "2026Q4", "org": "机构A", "eps": 2.0, "pe": 20.0},
|
||||
{"k": "SZ002812", "date": "2026-08-10", "quarter": "2026Q4", "org": "机构B", "eps": 3.0, "pe": 30.0}]
|
||||
b = sources._latest_by_org(dup) # noqa: SLF001
|
||||
t("同机构多篇只留最近一篇", b["2026Q4"]["机构A"]["eps"] == 2.0 and len(b["2026Q4"]) == 2)
|
||||
v = sources.valuation_scenarios(["SZ002812", "SH600000"], DS, {"SZ002812": PRICE, "SH600000": 10.0},
|
||||
rows=rows_for() + dup)
|
||||
t("按票给结果；没有研报行的票为 None", v["SH600000"] is None and v["SZ002812"]["na"] is None)
|
||||
t("用的是 2026 年度、机构数按去重后算", v["SZ002812"]["quarter"] == "2026Q4" and v["SZ002812"]["firms"] == 11)
|
||||
t("截止日取所用行里最近的报告日", v["SZ002812"]["as_of"] == "2026-08-25")
|
||||
t("没有现价的票写现价取不到",
|
||||
sources.valuation_scenarios(["SZ002812"], DS, {}, rows=rows_for())["SZ002812"]["na"] == "现价取不到")
|
||||
|
||||
|
||||
def test_fetch_shared():
|
||||
print("共用取数与券商行动不变")
|
||||
seen = {}
|
||||
|
||||
def _reader(which, sql, params):
|
||||
seen["sql"], seen["params"] = " ".join(sql.split()), params
|
||||
return [
|
||||
{"ts_code": "002812.SZ", "report_date": "2026-08-25", "quarter": "2026Q4", "org_name": "甲", "eps": "2.4", "pe": "22"},
|
||||
{"ts_code": "002812.SZ", "report_date": "2026-08-20", "quarter": "2026Q4", "org_name": "乙", "eps": 2.2, "pe": None},
|
||||
{"ts_code": "002812.SZ", "report_date": "2026-07-10", "quarter": "2026Q4", "org_name": "甲", "eps": 2.6, "pe": 25},
|
||||
{"ts_code": "002812.SZ", "report_date": "2026-07-05", "quarter": "2026Q4", "org_name": "乙", "eps": 2.7, "pe": 26},
|
||||
{"ts_code": "002812.SZ", "report_date": "2026-07-05", "quarter": "", "org_name": "丙", "eps": 2.7, "pe": 26},
|
||||
{"ts_code": "002812.SZ", "report_date": None, "quarter": "2026Q4", "org_name": "丁", "eps": 2.7, "pe": 26},
|
||||
]
|
||||
rows = sources.broker_reports(["SZ002812"], DS, read_mysql=_reader)
|
||||
t("查的是 90 个自然日、带市盈率列、按点分形态传代码",
|
||||
"pe FROM gp_report_rc" in seen["sql"] and seen["params"] == ("002812.SZ", "2026-06-04", DS))
|
||||
t("没有预测期或报告日的行不要；字符串数字归一", len(rows) == 4 and rows[0]["eps"] == 2.4 and rows[0]["pe"] == 22.0)
|
||||
t("市盈率缺就是 None，不当成零", rows[1]["pe"] is None)
|
||||
sig = sources.broker_actions(["SZ002812"], DS, rows=rows)["SZ002812"]
|
||||
t("券商行动：近 45 天（07-19 之后）两家中位 2.3 对前 45 天两家中位 2.65，下修 13% 转弱",
|
||||
sig["path"] == ls.PATH_BROKER and sig["signal"] == ls.SIG_DOWN and "下修 13%" in sig["why"])
|
||||
t("覆盖没收缩，不到硬触发", not sig.get("hard") and sig["refs"][0]["quarter"] == "2026Q4")
|
||||
t("不传 rows 时自己取，结果一样", sources.broker_actions(["SZ002812"], DS, read_mysql=_reader)["SZ002812"]["why"] == sig["why"])
|
||||
t("读失败两路都是空", sources.broker_reports(["SZ002812"], DS, read_mysql=lambda *a: (_ for _ in ()).throw(OSError("x"))) == [])
|
||||
|
||||
def _price_reader(which, sql, params):
|
||||
seen["psql"], seen["pparams"] = " ".join(sql.split()), params
|
||||
return [{"ts_code": "002812.SZ", "close": "50.17"}, {"ts_code": "600000", "close": 10.5},
|
||||
{"ts_code": "430047", "close": 3.0}, {"ts_code": "300750.SZ", "close": None}]
|
||||
px = sources.close_prices(DS, read_mysql=_price_reader, code_col="symbol")
|
||||
t("收盘价按前缀码索引，两种代码形态都认", px == {"SZ002812": 50.17, "SH600000": 10.5, "BJ430047": 3.0})
|
||||
t("日期写成左闭右开区间", "`timestamp` >= %s AND `timestamp` < %s" in seen["psql"] and seen["pparams"] == (DS, "2026-09-03"))
|
||||
t("收盘价读失败返回空字典", sources.close_prices(DS, read_mysql=lambda *a: (_ for _ in ()).throw(OSError("x")), code_col="symbol") == {})
|
||||
|
||||
|
||||
def test_card_text():
|
||||
print("卡上的文字")
|
||||
s = sources.scenarios(EPS, PE, PRICE, quarter="2026Q4")
|
||||
line = card.valuation_view(s)
|
||||
t("整句：预测期、机构数、三个价与幅度、隐含市盈率、赔率",
|
||||
line == "安全边际（2026 年度预测，9 家）：悲观 42.64 元（-15.0%）、中性 56.75 元（+13.1%）、"
|
||||
"乐观 84.96 元（+69.3%）；隐含市盈率 21.4 倍；赔率 0.87 比 1（中性上行对悲观下行）")
|
||||
t("短写法", card.valuation_short(s) == "-15%/+13%，赔率 0.87")
|
||||
t("没有数据", card.valuation_view(None) == "安全边际：近三个月没有券商的盈利预测，算不出" and card.valuation_short(None) == "—")
|
||||
na = sources.scenarios([2.0], [20], PRICE)
|
||||
t("不适用写原因", card.valuation_view(na) == "安全边际算不出：覆盖机构只有 1 家，不足 2 家" and card.valuation_short(na) == "算不出")
|
||||
lo = sources.scenarios([2.0, 2.2], [30, 40], 30.0)
|
||||
t("赔率不成立时整句写说明、短写法写不成立", "悲观情景仍高于现价" in card.valuation_view(lo) and card.valuation_short(lo).endswith("赔率不成立"))
|
||||
t("预测期写成人话", card._period_cn("2026Q2") == "2026 年中期" and card._period_cn("x") == "X") # noqa: SLF001
|
||||
t("恩捷的分歧不算大，不标注", not s["wide"] and s["spread"] == {"eps": 1.29, "pe": 1.55} and "分歧" not in line)
|
||||
w = sources.scenarios([0.03, 1.0, 2.0], [20, 30, 40], 10.0) # 每股收益最高是最低的 67 倍
|
||||
t("分歧极大：整句标注倍数、短写法带括号",
|
||||
w["wide"] and "机构分歧极大（每股收益最高是最低的 66.7 倍" in card.valuation_view(w)
|
||||
and card.valuation_short(w).endswith("（分歧极大）"))
|
||||
|
||||
|
||||
def main():
|
||||
test_scenarios()
|
||||
test_period_and_org()
|
||||
test_fetch_shared()
|
||||
test_card_text()
|
||||
print("ALL OK — 三情景对表 / 不适用四种 / 预测期与机构去重 / 共用取数与券商行动不变 / 收盘价 / 卡上文字 全部通过")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Reference in New Issue