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