From e00b38e36bc2133c2b55882fa726130c567d5583 Mon Sep 17 00:00:00 2001 From: zlt Date: Mon, 7 Sep 2026 15:48:27 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AC=AC=E5=9B=9B=E4=BB=B6=EF=BC=9A=E5=AE=89?= =?UTF-8?q?=E5=85=A8=E8=BE=B9=E9=99=85=E4=B8=89=E6=83=85=E6=99=AF=EF=BC=8C?= =?UTF-8?q?=E5=8F=AA=E5=B1=95=E7=A4=BA=E4=B8=8D=E8=BF=9B=E5=88=A4=E5=86=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- card.py | 46 +++++++++ plan.py | 24 ++++- sources.py | 235 ++++++++++++++++++++++++++++++++++++++-------- test_valuation.py | 166 ++++++++++++++++++++++++++++++++ 4 files changed, 430 insertions(+), 41 deletions(-) create mode 100644 test_valuation.py diff --git a/card.py b/card.py index c80bb07..0e7a2a5 100644 --- a/card.py +++ b/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 "") diff --git a/plan.py b/plan.py index 5b68f1b..6ca6e1f 100644 --- a/plan.py +++ b/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,16 +686,21 @@ 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)} 个环节：定位对不对看这里，挑票看候选单）") diff --git a/sources.py b/sources.py index 9b8ceb5..cffe1ce 100644 --- a/sources.py +++ b/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 diff --git a/test_valuation.py b/test_valuation.py new file mode 100644 index 0000000..c4b7f72 --- /dev/null +++ b/test_valuation.py @@ -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()