zlt
/
akg-factor-bridge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
akg-factor-bridge/test_events_pricing.py

201 lines
12 KiB
Python
Raw Normal View History

量价研判链 3.4：催化事件、事件日字段与定价状态、周报分组（只展示不进判决） sources：analyst_reports / analyst_events 从券商研报明细表算四类正向事件（深度覆盖前 365 天无覆盖且买入类、 同机构同预测期 180 天内上调五成、标题含超预期、同日合并为复合），窗口 60 天；price_history 分两段读前复权 行情（无事件 35 天、有事件 100 天）；event_day_fields 算事件前 5/20 日涨幅、跳空、日内、收盘位置、量比、涨停。 card：pricing_state 按四情形归类（价格发现、趋势延续、高位兑现、震荡消化），阈值一次定死；两类整句与短写法。 plan：候选卡带 events / pricing_state 与整句，候选单表格加两列并附口径说明。 plan_review：分组读数加"定价状态"与"催化事件"两类，另出定价状态多空差一段。 测试：新建 test_events_pricing.py；09-04 实机：档位表 1,210 只里 102 只有事件，定价状态分布 震荡消化 1,146、高位兑现 33、价格发现 16、趋势延续 15；计划装配 30 秒。 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-08 11:40:51 +08:00
"""催化事件、事件日字段与定价状态的离线单测（不连库）。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()