271 lines
17 KiB
Python
271 lines
17 KiB
Python
"""催化事件、事件日字段与定价状态的离线单测（不连库）。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 test_industry_catalyst():
|
||
"""行业催化（台账 051）：按环节取、按票汇总、量级排序、缺席不断产。"""
|
||
rows = [
|
||
{"segment_name": "端侧AI芯片", "event_date": "2026-09-05", "first_seen": "2026-09-06", "title": "2B 模型媲美百亿模型",
|
||
"direction": "利好", "magnitude": "大", "horizon": "长期", "confidence": "中", "mechanism": "端侧推理可行性大增"},
|
||
{"segment_name": "端侧AI芯片", "event_date": None, "first_seen": "2026-09-01", "title": "某厂扩产",
|
||
"direction": "利好", "magnitude": "小", "horizon": "短期", "confidence": "低", "mechanism": None},
|
||
{"segment_name": "存储", "event_date": "2026-09-07", "first_seen": "2026-09-07", "title": "2B 模型媲美百亿模型",
|
||
"direction": "利好", "magnitude": "中", "horizon": "中期", "confidence": "高", "mechanism": "端侧内存需求上移"},
|
||
]
|
||
calls = []
|
||
def rp(sql, params=None):
|
||
calls.append(params); return _DF(rows)
|
||
by_seg = sources.segment_catalysts(["端侧AI芯片", "存储", "无材料环节"], "2026-09-08", read_pg=rp)
|
||
t("按环节名与 180 天窗口查", calls[0][0] == ["存储", "无材料环节", "端侧AI芯片"] and calls[0][1] == "2026-03-12")
|
||
t("量级大的排前、没事件日的用首见日", by_seg["端侧AI芯片"][0]["magnitude"] == "大" and by_seg["端侧AI芯片"][1]["event_date"] == "2026-09-01")
|
||
t("没材料的环节不在字典里", "无材料环节" not in by_seg)
|
||
per = sources.catalysts_for_codes({"SH600000": ["端侧AI芯片", "存储"], "SZ000001": ["无材料环节"], "SZ000002": []}, by_seg)
|
||
t("按票汇总同标题去重、量级大→小", [x["title"] for x in per["SH600000"]] == ["2B 模型媲美百亿模型", "某厂扩产"])
|
||
t("所在环节没有催化的票不在字典里", "SZ000001" not in per and "SZ000002" not in per)
|
||
v = card.industry_catalyst_view(per["SH600000"])
|
||
t("整句带量级持续期方向与环节名", v.startswith("行业催化（环节级") and "[大·长期·利好·把握中] 端侧AI芯片" in v)
|
||
t("缺席整句与短句", "暂无" in card.industry_catalyst_view(None) and card.industry_catalyst_short([]) == "—")
|
||
def boom(*a, **k): raise RuntimeError("no table")
|
||
t("表读不到整体缺席不断产", sources.segment_catalysts(["端侧AI芯片"], "2026-09-08", read_pg=boom) == {})
|
||
|
||
|
||
def main():
|
||
test_events()
|
||
test_event_day_fields()
|
||
test_pricing_state()
|
||
test_news()
|
||
test_industry_catalyst()
|
||
print("ALL OK — 四类事件 / 事件日字段 / 定价状态四情形 / 卡上文字 / 相关快讯 全部通过")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|