"""催化事件、事件日字段与定价状态的离线单测（不连库）。2026-09-08《量价研判链吸收方案》3.4。 钉住四件事： 一，四类券商正向事件的定义各自成立：深度覆盖看前 365 天有无覆盖与评级；上调预测看同机构同预测期 180 天内的上一篇；超预期看标题；同一天多篇合并成一条并标复合；窗口之外的不算。 二，事件日字段：事件前涨幅、跳空、日内收益、收盘位置、量比、涨停各自算对；行情不够时留空不硬算； 没有事件的票按数据日算。 三，定价状态四情形的规则一次定死（台账 046），四种各有样例，缺字段时写明缺什么。 四，卡上的文字与表格短写法。 开发机没有 pandas 与数据库驱动时只给缺席的模块装最小桩（与 test_valuation.py 同一约定）。 跑法：python3 test_events_pricing.py 或 pytest test_events_pricing.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 sources # noqa: E402 def t(name, cond): assert cond, name print(" ok", name) DS = "2026-09-04" K = "SZ002812" def rep(date, *, typ="点评", title="跟踪点评", rating="买入", org="甲", quarter="2026Q4", eps=2.0, k=K): return {"k": k, "date": date, "type": typ, "title": title, "rating": rating, "org": org, "quarter": quarter, "eps": eps} def test_events(): print("四类事件") rows = [rep("2026-09-03", typ="深度", title="深度报告：迎来拐点", rating="买入", org="乙")] ev = sources.analyst_events([K], DS, rows=rows)[K] t("前 365 天无覆盖的深度买入 -> 深度覆盖", ev["events"][0]["types"] == [sources.EV_DEEP]) rows2 = rows + [rep("2026-01-15", org="丙")] t("前 365 天有覆盖就不算深度覆盖", K not in sources.analyst_events([K], DS, rows=rows2)) rows3 = [rep("2026-09-03", typ="深度", rating="中性", org="乙")] t("深度但评级不是买入类不算", K not in sources.analyst_events([K], DS, rows=rows3)) rows = [rep("2026-06-20", org="甲", eps=2.0), rep("2026-09-01", org="甲", eps=3.1)] ev = sources.analyst_events([K], DS, rows=rows)[K] t("同机构同预测期 180 天内上调五成以上 -> 上调盈利预测", ev["events"][0]["date"] == "2026-09-01" and ev["events"][0]["types"] == [sources.EV_UPGRADE]) rows = [rep("2026-06-20", org="甲", eps=2.0), rep("2026-09-01", org="甲", eps=2.5)] t("只上调两成五不算", K not in sources.analyst_events([K], DS, rows=rows)) rows = [rep("2026-06-20", org="甲", eps=2.0, quarter="2027Q4"), rep("2026-09-01", org="甲", eps=3.1)] t("预测期不同不算", K not in sources.analyst_events([K], DS, rows=rows)) rows = [rep("2025-12-01", org="甲", eps=2.0), rep("2026-09-01", org="甲", eps=3.1)] t("上一篇超过 180 天不算", K not in sources.analyst_events([K], DS, rows=rows)) rows = [rep("2026-06-20", org="甲", eps=-0.5), rep("2026-09-01", org="甲", eps=1.0)] t("上一篇为负不算比例", K not in sources.analyst_events([K], DS, rows=rows)) rows = [rep("2026-08-28", title="2026 中报点评：业绩超预期，产能释放")] ev = sources.analyst_events([K], DS, rows=rows)[K] t("标题含超预期", ev["events"][0]["types"] == [sources.EV_BEAT] and not ev["events"][0]["compound"]) rows = [rep("2026-06-20", org="甲", eps=2.0), rep("2026-09-01", org="甲", eps=3.1, title="业绩超预期"), rep("2026-09-01", org="丁", typ="深度", title="深度：业绩超预期")] ev = sources.analyst_events([K], DS, rows=rows)[K] e0 = ev["events"][0] t("同一篇同时上调与超预期 -> 复合；同一天多篇合并成一条、机构合在一起", e0["compound"] and set(e0["types"]) == {sources.EV_BEAT, sources.EV_UPGRADE} and e0["n_reports"] == 2 and e0["orgs"] == ["甲", "丁"]) t("深度那篇因为同期有覆盖不算深度覆盖", sources.EV_DEEP not in e0["types"]) rows = [rep("2026-06-20", org="甲", eps=2.0), rep("2026-09-01", org="甲", eps=3.1, title="业绩超预期"), rep("2026-09-01", org="丁", typ="深度")] e0 = sources.analyst_events([K], DS, rows=rows)[K]["events"][0] t("同一天另一篇没有命中任何事件的研报不计入机构与篇数", e0["n_reports"] == 1 and e0["orgs"] == ["甲"]) rows = [rep("2026-06-30", title="业绩超预期"), rep("2026-09-02", title="业绩超预期"), rep("2026-09-05", title="业绩超预期")] ev = sources.analyst_events([K], DS, rows=rows)[K] t("窗口：60 天之前与数据日之后的都不算，最新在前", [e["date"] for e in ev["events"]] == ["2026-09-02"] and ev["latest"] == "2026-09-02") t("没有研报行的票不在结果里", "SH600000" not in sources.analyst_events([K, "SH600000"], DS, rows=rows)) t("数据日不合法返回空", sources.analyst_events([K], "不是日期", rows=rows) == {}) def bar(date, o, h, l, c, pre=None, pct=None, vol=1000.0): return {"date": date, "open": o, "high": h, "low": l, "close": c, "pre_close": pre, "pct": pct, "volume": vol} def hist_rows(n=30, base=10.0, step=0.0, last=None): """n 根平淡的 K 线，最后一根可替换。""" rows = [] for i in range(n): px = base + step * i rows.append(bar(f"2026-08-{i + 1:02d}" if i < 31 else f"2026-09-{i - 30:02d}", px, px * 1.01, px * 0.99, px, pre=px, pct=0.0, vol=1000.0)) if last: rows[-1] = last return rows def test_event_day_fields(): print("事件日字段") # 事件日：跳空 3% 高开、日内再涨 2%、收在区间高位、量三倍、涨幅 5.06% last = bar("2026-08-30", 10.3, 10.6, 10.25, 10.506, pre=10.0, pct=5.06, vol=3000.0) hist = {K: hist_rows(30, last=last)} f = sources.event_day_fields([K], DS, {K: {"latest": "2026-08-30"}}, hist=hist)[K] t("事件日取事件当天那根", f["event_date"] == "2026-08-30" and f["has_event"]) t("跳空 3%、日内 2%、当日涨幅 5.06%", f["gap"] == 0.03 and f["intraday"] == 0.02 and f["day_pct"] == 0.0506) t("收盘位置 0.73、量比 3.0、不涨停", round(f["close_pos"], 2) == 0.73 and f["vol_ratio"] == 3.0 and f["limit_up"] is False) t("事件前 5 日与 20 日涨幅（平的行情）为零", f["pre5"] == 0.0 and f["pre20"] == 0.0) hist = {K: hist_rows(30, base=10.0, step=0.1, last=bar("2026-08-30", 13.0, 13.2, 12.9, 13.1, pre=12.8, pct=2.34, vol=1000.0))} f = sources.event_day_fields([K], DS, {K: {"latest": "2026-08-30"}}, hist=hist)[K] t("事件前 20 日涨幅按事件前一日对二十一日前算", round(f["pre20"], 3) == round(12.8 / 10.8 - 1, 3)) t("创业板涨停线 19.8：科创板代码 5% 不算涨停", sources.event_day_fields(["SH688001"], DS, {}, hist={"SH688001": hist_rows(30, last=bar("2026-08-30", 10, 10.6, 10, 10.5, pre=10, pct=5.0))})["SH688001"]["limit_up"] is False) t("主板 9.9% 算涨停", sources.event_day_fields([K], DS, {}, hist={K: hist_rows(30, last=bar("2026-08-30", 10, 11, 10, 10.99, pre=10, pct=9.9))})[K]["limit_up"] is True) f = sources.event_day_fields([K], DS, {}, hist={K: hist_rows(3)})[K] t("没有事件按数据日（最后一根），历史不够时 20 日涨幅与量比为空", not f["has_event"] and f["pre20"] is None and f["vol_ratio"] is None and f["gap"] == 0.0) t("事件日晚于行情最后一根时取不晚于事件日的最后一根", sources.event_day_fields([K], DS, {K: {"latest": "2026-09-30"}}, hist={K: hist_rows(30)})[K]["event_date"] == "2026-08-30") t("没有行情的票不在结果里", "SH600000" not in sources.event_day_fields([K, "SH600000"], DS, {}, hist={K: hist_rows(30)})) t("行情表读失败返回空字典", sources.price_history([K], DS, read_mysql=lambda *a: (_ for _ in ()).throw(OSError("x")), code_col="symbol") == {}) seen = {} def _reader(which, sql, params): seen["sql"], seen["params"] = " ".join(sql.split()), params return [{"ts_code": "SZ002812", "d": "2026-09-04", "open": "10", "high": "11", "low": "9", "close": "10.5", "pre_close": 10, "percent": 5.0, "volume": 100}] ph = sources.price_history(["002812.SZ", K], DS, read_mysql=_reader, code_col="symbol") t("行情表按前缀式代码查、区间左闭右开、去重代码", seen["params"] == ("2026-05-27", "2026-09-05", "SZ002812") and "`symbol` IN (%s)" in seen["sql"] and ph[K][0]["close"] == 10.5) def fields(**kw): base = {"event_date": "2026-08-30", "has_event": True, "pre5": 0.01, "pre20": 0.02, "gap": 0.0, "intraday": 0.01, "close_pos": 0.8, "vol_ratio": 2.0, "day_pct": 0.03, "limit_up": False} base.update(kw) return base def test_pricing_state(): print("定价状态四情形") p = card.pricing_state(fields()) t("事件前没涨、事件日放量收高 -> 价格发现", p["state"] == card.PRICING_DISCOVERY) p = card.pricing_state(fields(pre20=0.08)) t("事件前已涨 8%、事件日仍放量收高 -> 趋势延续", p["state"] == card.PRICING_CONTINUE) p = card.pricing_state(fields(pre20=0.15, gap=0.03, intraday=-0.02, close_pos=0.2, day_pct=0.01)) t("事件前大涨、事件日放量跳空冲高回落 -> 高位兑现", p["state"] == card.PRICING_CASHOUT) p = card.pricing_state(fields(pre20=0.06, gap=0.0, intraday=-0.02, close_pos=0.2, day_pct=-0.01)) t("事件前涨 6% 且冲高回落但不到 10% -> 震荡消化", p["state"] == card.PRICING_DIGEST) p = card.pricing_state(fields(vol_ratio=1.0)) t("量比不够 -> 震荡消化", p["state"] == card.PRICING_DIGEST) p = card.pricing_state(fields(day_pct=-0.01)) t("收在高位但当日下跌 -> 不算确认，震荡消化", p["state"] == card.PRICING_DIGEST) p = card.pricing_state(fields(pre20=None)) t("缺 20 日涨幅 -> 不归类并写明", p["state"] is None and "事件前 20 日涨幅" in p["why"]) t("没有字段 -> None", card.pricing_state(None) is None and card.pricing_state({}) is None) print("卡上的文字") p = card.pricing_state(fields()) line = card.pricing_view(p) t("整句带情形、依据与六个数", line.startswith("定价状态（事件日 2026-08-30）：价格发现。") and "量比 2.0" in line and "收盘位置 0.80" in line) t("短写法", card.pricing_short(p) == "价格发现" and card.pricing_short(None) == "—" and card.pricing_short(card.pricing_state(fields(pre20=None))) == "算不出") t("无事件按数据日的整句写明", "无事件，按数据日" in card.pricing_view(card.pricing_state(fields(has_event=False)))) ev = {"latest": "2026-09-01", "count": 2, "events": [ {"date": "2026-09-01", "types": ["上调盈利预测", "业绩超预期"], "orgs": ["甲", "丁"], "title": "x", "n_reports": 2, "compound": True}, {"date": "2026-08-20", "types": ["深度覆盖"], "orgs": ["乙"], "title": "y", "n_reports": 1, "compound": False}]} t("催化事件整句", card.events_view(ev) == "催化事件（近 60 天 2 天有事件）：2026-09-01 上调盈利预测与业绩超预期（甲、丁，复合）；2026-08-20 深度覆盖（乙）") t("催化事件短写法", card.events_short(ev) == "09-01 上调盈利预测与业绩超预期（复合）" and card.events_short(None) == "—") t("没有事件的整句", "没有券商正向事件" in card.events_view(None)) def main(): test_events() test_event_day_fields() test_pricing_state() print("ALL OK — 四类事件 / 事件日字段 / 定价状态四情形 / 卡上文字 全部通过") if __name__ == "__main__": main()