"""安全边际三情景的离线单测（不连库）。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()