akg-factor-bridge/test_events_pricing.py

"""催化事件、事件日字段与定价状态的离线单测（不连库）。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))


class _DF:
    def __init__(self, rows): self._rows = rows
    def to_dict(self, _kind): return self._rows


def test_news():
    """相关快讯（台账 050）：按财联社 StockID 标签落到前缀码，最新在前、最多五条、只当上下文。"""
    import datetime as dt
    import json
    rows = [
        {"cls_id": 3, "level": "C", "publish_time": dt.datetime(2026, 9, 8, 12, 39), "article_url": "u3",
         "title": "燕东微：上半年8英寸SiN产线产能达3000片/月", "content": "",
         "stock_list_json": json.dumps([{"StockID": "sh688172", "name": "燕东微"}])},
        {"cls_id": 2, "level": "B", "publish_time": dt.datetime(2026, 9, 8, 11, 32), "article_url": "u2",
         "title": "财联社9月8日电，农业板块再度拉升 金健米业17天9板", "content": "",
         "stock_list_json": json.dumps([{"StockID": "sz000998"}, {"StockID": "sh688172"}, {"StockID": "bad"}])},
        {"cls_id": 1, "level": "C", "publish_time": dt.datetime(2026, 9, 7, 9, 0), "article_url": "u1",
         "title": None, "content": "财联社9月7日电，华鲁恒升：尿素价格未来走势存在较大不确定性",
         "stock_list_json": "not json"},
    ]
    calls = []
    def rm(which, sql, params=None):
        calls.append((which, params)); return _DF(rows)
    out = sources.news_flashes(["SH688172", "SZ000998", "SH600000"], "2026-09-08", read_mysql=rm)
    t("用行情同一连接、窗口按自然日往前 3 天", calls[0][0] == "factor" and calls[0][1] == ("2026-09-05",))
    t("燕东微两条、最新在前", out["SH688172"]["count"] == 2 and out["SH688172"]["items"][0]["cls_id"] == 3)
    t("标题去掉「财联社X月X日电」前缀", out["SZ000998"]["items"][0]["title"].startswith("农业板块"))
    t("没被点名的票不在字典里、坏 JSON 跳过", "SH600000" not in out and all(i["cls_id"] != 1 for i in out["SH688172"]["items"]))
    t("时间落成 MM-DD HH:MM", out["SH688172"]["latest"] == "09-08 12:39")
    big = [dict(rows[0], cls_id=10 + i, publish_time=dt.datetime(2026, 9, 8, 13, i)) for i in range(7)]
    out2 = sources.news_flashes(["SH688172"], "2026-09-08", read_mysql=lambda *a, **k: _DF(big))
    t("最多五条但计数照实", len(out2["SH688172"]["items"]) == 5 and out2["SH688172"]["count"] == 7)
    def boom(*a, **k): raise RuntimeError("no table")
    t("表读不到整体缺席不断产", sources.news_flashes(["SH688172"], "2026-09-08", read_mysql=boom) == {})
    t("日期坏了也不断产", sources.news_flashes(["SH688172"], "bad", read_mysql=rm) == {})
    v = card.news_view(out["SH688172"])
    t("整句带条数与不判正负的说明", v.startswith("相关快讯（近 3 天 2 条") and "不判利好利空" in v and "[C]" in v)
    t("没有快讯的整句", "没有点名" in card.news_view(None) and card.news_short(None) == "—")
    t("短句是最新一条", card.news_short(out["SH688172"]).startswith("09-08 12:39 燕东微"))


def main():
    test_events()
    test_event_day_fields()
    test_pricing_state()
    test_news()
    print("ALL OK — 四类事件 / 事件日字段 / 定价状态四情形 / 卡上文字 / 相关快讯 全部通过")


if __name__ == "__main__":
    main()