量价研判链 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>

card.py

@@ -372,3 +372,89 @@ def valuation_short(scn) -> str:
         return "算不出"
     odds = f"赔率 {scn['odds']:.2f}" if scn.get("odds") is not None else "赔率不成立"
     return f"{scn['down']:+.0%}/{scn['up_neut']:+.0%}，{odds}" + ("（分歧极大）" if scn.get("wide") else "")
+
+
+# ============================================================================
+# 定价状态与催化事件的文字（2026-09-08《量价研判链吸收方案》3.4）：只展示，不进判决
+# ============================================================================
+#
+# 定价状态回答"这只票现在是刚启动还是尾声"，是启动线（当日涨幅 3%，已关）的替代品，先只展示
+# 加复盘分组。四情形照研报，规则一次定死（台账 046）：
+#   事件日确认   = 量比不低于 1.5 且收盘位置不低于 0.6 且当日上涨
+#   冲高回落     = 量比不低于 1.5 且收盘位置不高于 0.4 且（跳空高开超过 2% 或日内收益为负）
+#   价格发现     = 事件前 20 日涨幅低于 5% 且事件日确认
+#   趋势延续     = 事件前 20 日涨幅不低于 5% 且事件日确认
+#   高位兑现     = 事件前 20 日涨幅不低于 10% 且冲高回落
+#   震荡消化     = 其余
+PRICING_DISCOVERY, PRICING_CONTINUE = "价格发现", "趋势延续"
+PRICING_CASHOUT, PRICING_DIGEST = "高位兑现", "震荡消化"
+PRICING_PRE_RUN = 0.05        # 事件前 20 日涨幅，低于它算"没抢跑"
+PRICING_PRE_HIGH = 0.10       # 事件前 20 日涨幅，不低于它才谈"高位"
+PRICING_VR = 1.5              # 量比线
+PRICING_CLOSE_HI = 0.6        # 收盘位置：收在高位
+PRICING_CLOSE_LO = 0.4        # 收盘位置：收在低位
+PRICING_GAP = 0.02            # 跳空高开线
+
+
+def pricing_state(f) -> dict | None:
+    """把事件日字段归成四情形之一。字段不够（没有 20 日历史或没有量比）时不归类，写明缺什么。"""
+    if not isinstance(f, dict) or not f.get("event_date"):
+        return None
+    pre20, vr, pos = f.get("pre20"), f.get("vol_ratio"), f.get("close_pos")
+    gap, intra, pct = f.get("gap"), f.get("intraday"), f.get("day_pct")
+    base = {**f, "state": None, "why": None}
+    missing = [n for n, v in (("事件前 20 日涨幅", pre20), ("量比", vr), ("收盘位置", pos)) if v is None]
+    if missing:
+        return {**base, "why": "行情不够，算不出" + "、".join(missing)}
+    confirmed = vr >= PRICING_VR and pos >= PRICING_CLOSE_HI and (pct or 0) > 0
+    faded = vr >= PRICING_VR and pos <= PRICING_CLOSE_LO and ((gap or 0) > PRICING_GAP or (intra or 0) < 0)
+    if confirmed and pre20 < PRICING_PRE_RUN:
+        state, why = PRICING_DISCOVERY, f"事件前 20 日涨幅 {pre20:+.1%}，没有抢跑；事件日量比 {vr:.1f} 收在高位"
+    elif confirmed:
+        state, why = PRICING_CONTINUE, f"事件前 20 日已涨 {pre20:+.1%}，事件日量比 {vr:.1f} 仍收在高位"
+    elif faded and pre20 >= PRICING_PRE_HIGH:
+        state, why = PRICING_CASHOUT, f"事件前 20 日已涨 {pre20:+.1%}，事件日放量冲高回落，收盘位置 {pos:.2f}"
+    else:
+        state, why = PRICING_DIGEST, f"事件前 20 日 {pre20:+.1%}，事件日量比 {vr:.1f}，收盘位置 {pos:.2f}，没有明确的确认或兑现"
+    return {**base, "state": state, "why": why}
+
+
+def pricing_view(p) -> str:
+    """定价状态的整句。"""
+    if not isinstance(p, dict):
+        return "定价状态：行情取不到，算不出"
+    head = f"定价状态（{'事件日' if p.get('has_event') else '无事件，按数据日'} {p.get('event_date')}）"
+    if not p.get("state"):
+        return f"{head}：{p.get('why') or '算不出'}"
+    nums = (f"事件前 5 日 {_pct(p.get('pre5'))}、20 日 {_pct(p.get('pre20'))}；事件日跳空 {_pct(p.get('gap'))}、"
+            f"日内 {_pct(p.get('intraday'))}、收盘位置 {p.get('close_pos'):.2f}、量比 {p.get('vol_ratio'):.1f}"
+            f"{'、涨停' if p.get('limit_up') else ''}")
+    return f"{head}：{p['state']}。{p['why']}。{nums}"
+
+
+def pricing_short(p) -> str:
+    if not isinstance(p, dict):
+        return "—"
+    return p.get("state") or "算不出"
+
+
+def events_view(ev) -> str:
+    """催化事件的整句：最新在前，最多五条。"""
+    if not isinstance(ev, dict) or not ev.get("events"):
+        return "催化事件：近 60 天没有券商正向事件（深度覆盖、上调盈利预测、业绩超预期）"
+    parts = []
+    for e in ev["events"]:
+        orgs = "、".join(e.get("orgs") or [])[:40]
+        parts.append(f"{e['date']} {'与'.join(e.get('types') or [])}（{orgs}{'，复合' if e.get('compound') else ''}）")
+    return f"催化事件（近 60 天 {ev.get('count')} 天有事件）：" + "；".join(parts)
+
+
+def events_short(ev) -> str:
+    if not isinstance(ev, dict) or not ev.get("events"):
+        return "—"
+    e = ev["events"][0]
+    return f"{e['date'][5:]} {'与'.join(e.get('types') or [])}" + ("（复合）" if e.get("compound") else "")
+
+
+def _pct(v) -> str:
+    return "—" if v is None else f"{v:+.1%}"

plan.py

@@ -298,6 +298,10 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
     # 卡上那一行写"算不出"并说明原因，不拦票、不断产。
     prices = sources.close_prices(ds)
     valuation = sources.valuation_scenarios(codes, ds, prices, rows=inp["broker_rows"])
+    # 催化事件与定价状态（2026-09-08《量价研判链吸收方案》3.4）：四类券商正向事件从研报明细表算，
+    # 事件日字段从前复权行情表算，归成四情形。只展示加复盘分组，不进判决；任一读不到整体缺席不断产。
+    events = sources.analyst_events(codes, ds)
+    ev_fields = sources.event_day_fields(codes, ds, events)
     if risk is None:                              # collect 会传入读过一次的名单；单独调用时自己读
         try:
             risk = factors._risk_set() or set()  # noqa: SLF001 —— 同仓自用
@@ -332,6 +336,8 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series,
         cards[k] = {
             **j, "logic_state": state,
             "valuation": valuation.get(k), "valuation_text": card.valuation_view(valuation.get(k)),
+            "events": events.get(k), "events_text": card.events_view(events.get(k)),
+            "pricing_state": card.pricing_state(ev_fields.get(k)),
             "theme": theme, "n_sources": n_sources, "chain_fit": evd["chain_fit"],
             "started_source": "moved_view" if mv else None,
             "logic_claims": evd["logic"],
@@ -460,6 +466,10 @@ def collect(date: str | None = None, top: int = 20, obs_top: int = 10,
                      logic_state=_state_out(c.get("logic_state")),
                      # 安全边际三情景（2026-09-07 第四件）：原值给程序，整句给人；只展示不进判决。
                      valuation=c.get("valuation"), valuation_text=c.get("valuation_text"),
+                     # 催化事件与定价状态（2026-09-08）：原值给程序，整句给人；只展示不进判决。
+                     events=c.get("events"), events_text=c.get("events_text"),
+                     pricing_state=c.get("pricing_state"),
+                     pricing_text=card.pricing_view(c.get("pricing_state")),
                      card={"pct0": c.get("pct0"), "net_z": c.get("net_z"),
                            "heat_chg": c.get("heat_chg"), "accum": c.get("accum"),
                            "night": c.get("night"), "gates": c.get("gates"),
@@ -699,8 +709,8 @@ def render_md(d: dict) -> str:
         L.append("（今日无候选——候选为空不是故障：环节没被指向、成员没启动或没有明确吸筹，都会为空。）")
     else:
         L.append("| # | 代码 | 名称 | 环节 | 源数 | 当日涨幅 | 吸筹 | 预期空间 | 安全边际（悲观下行/中性上行，赔率） "
-                 "| 理由 | 因果论断（出处） |")
+                 "| 催化事件 | 定价状态 | 理由 | 因果论断（出处） |")
-        L.append("|---|------|------|------|------|----------|------|----------|----------|------|------------------|")
+        L.append("|---|------|------|------|------|----------|------|----------|----------|----------|----------|------|------------------|")
         for r in cands:
             c = r.get("card") or {}
             ac = c.get("accum") or {}
@@ -708,11 +718,16 @@ def render_md(d: dict) -> str:
             L.append(f"| {r['rank']} | {r['code']} | {r['name'] or '—'} | {ev_.get('theme') or '—'} "
                      f"| {ev_.get('n_sources') or '—'} | {_fmt_pct0(c.get('pct0'))} "
                      f"| {_fmt_accum(ac)} | {_fmt_pct(r.get('upside'))} | {card.valuation_short(r.get('valuation'))} "
+                     f"| {card.events_short(r.get('events'))} | {card.pricing_short(r.get('pricing_state'))} "
                      f"| {'；'.join(r.get('reasons') or [])} | {_fmt_logic(r.get('logic'))} |")
         L.append("")
         L.append("安全边际按同一预测期的每股收益与市盈率预测算：悲观是最低每股收益乘最低市盈率，中性是两项中位数，"
                  "乐观是两项最高；赔率是中性上行对悲观下行，不到一比一就是这个位置的赔率不吸引人。"
                  "它只展示、不进判决，与上面用目标价平均算的预期空间是两个口径。")
+        L.append("催化事件是近 60 天券商的正向事件（深度覆盖、上调盈利预测、业绩超预期），从研报明细表算，最新一条列在表里。"
+                 "定价状态按最近一次事件日的事件前涨幅、事件日跳空、收盘位置与量比归成四情形："
+                 "价格发现（事件前没涨、事件日放量收高）、趋势延续（事件前已涨、事件日仍放量收高）、"
+                 "高位兑现（事件前大涨、事件日放量冲高回落）、震荡消化（其余）。两者都只展示、不进判决。")
     L.append("")
     segs = d.get("segments_pointed") or []
     L.append(f"## 关注环节（今日被传导指向的 {len(segs)} 个环节：定位对不对看这里，挑票看候选单）")

plan_review.py

@@ -310,6 +310,31 @@ def summarize(ret: pd.Series, codes: list[str], base_all: float, base_main: floa
 # ============================================================================
 # 主流程
 # ============================================================================
+def _pricing_spread_md(groups_tbl: pd.DataFrame) -> str:
+    """定价状态的"多空差"式对照（2026-09-08）：候选单里价格发现与趋势延续两组的超额均值，减去高位兑现组，
+    按期限各一行。研报用它检验方向区分能力；我们只作名单级观察，样本不够只看方向。"""
+    if groups_tbl is None or groups_tbl.empty or "group" not in groups_tbl.columns:
+        return ""
+    g = groups_tbl[(groups_tbl["list"] == "候选单") & groups_tbl["group"].astype(str).str.startswith("定价状态=")]
+    if g.empty:
+        return "（定价状态分组本期无样本，多空差不算。）"
+    lines = ["定价状态的多空差（候选单，价格发现与趋势延续两组的「比全池多涨几个点」均值，减去高位兑现组）：", ""]
+    for h, gh in g.groupby("h"):
+        val = {str(r["group"]).replace("定价状态=", ""): r for _, r in gh.iterrows()}
+        bull = [val[k] for k in ("价格发现", "趋势延续") if k in val]
+        bear = val.get("高位兑现")
+        if not bull or bear is None:
+            lines.append(f"- {h} 日：两端不齐（多头端 {len(bull)} 组，高位兑现组{'有' if bear is not None else '无'}），不算。")
+            continue
+        bull_ex = sum(float(r["excess_all"]) for r in bull) / len(bull)
+        n_bull = sum(int(r["n"]) for r in bull)
+        diff = bull_ex - float(bear["excess_all"])
+        lines.append(f"- {h} 日：多头端 {n_bull} 只样本、比全池多涨 {bull_ex:+.2f} 个点；高位兑现 {int(bear['n'])} 只、"
+                     f"{float(bear['excess_all']):+.2f} 个点；差 {diff:+.2f} 个点，样本"
+                     f"{'够' if n_bull >= 100 and int(bear['n']) >= 30 else '不够，只看方向'}。")
+    return "\n".join(lines)
+
+
 def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str,
         with_cards: bool = True) -> dict:
     days_plan = plan_dates(since, until)
@@ -407,6 +432,23 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str
                     if s:
                         rows.append({"date": day, "h": h, "list": "候选单", "group": f"市值={cap}",
                                      "regime_post": regime_post, "regime_pre": regime_pre, **s})
+            # 定价状态四情形与有无催化事件（2026-09-08《量价研判链吸收方案》3.4 第三件）：
+            # 候选单与档位表各分一次，只作观察，决定"要不要当门槛"的依据在这里攒。
+            pricing_of = {r["code"]: (r.get("pricing_state") or {}).get("state") for r in main_rows + obs_rows}
+            event_of = {r["code"]: bool((r.get("events") or {}).get("events")) for r in main_rows + obs_rows}
+            for lst_name, lst_codes in (("候选单", cands), ("档位表", list(pricing_of))):
+                for st_name in ("价格发现", "趋势延续", "高位兑现", "震荡消化", "算不出"):
+                    codes = [c for c in lst_codes if (pricing_of.get(c) or "算不出") == st_name]
+                    s = summarize(ret, codes, base_all, base_main, caps, cap_base)
+                    if s:
+                        rows.append({"date": day, "h": h, "list": lst_name, "group": f"定价状态={st_name}",
+                                     "regime_post": regime_post, "regime_pre": regime_pre, **s})
+                for flag, label in ((True, "有"), (False, "无")):
+                    codes = [c for c in lst_codes if event_of.get(c, False) is flag]
+                    s = summarize(ret, codes, base_all, base_main, caps, cap_base)
+                    if s:
+                        rows.append({"date": day, "h": h, "list": lst_name, "group": f"催化事件={label}",
+                                     "regime_post": regime_post, "regime_pre": regime_pre, **s})
         notes.append(f"{day}: 主榜 {len(main_codes)} 观察 {len(obs_rows)} 候选 {len(cands)} "
                      f"关注 {len(watch)} 生产 {len(prod)}（{prod_note}）；"
                      f"账本 机器通过 {len(ledger['机器通过名单'])} 人批 {len(ledger['人批名单'])} "
@@ -468,6 +510,10 @@ def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str
           "带「股票多的日子算得重」前缀的列是另一种算法，只用来和手工读数对账。", "",
           _md(lists_tbl), "",
           "## 二、分组读数", "", _md(groups_tbl), "",
+          "分组里的「定价状态」与「催化事件」两类是 2026-09-08 起加的：定价状态按最近一次券商正向事件日的"
+          "事件前涨幅、事件日跳空、收盘位置与量比归成四情形，催化事件是近 60 天有没有深度覆盖、上调盈利预测、"
+          "业绩超预期。两者都只展示不进判决，这两组读数是将来决定要不要当门槛的依据。", "",
+          _pricing_spread_md(groups_tbl), "",
           f"## 二之二、候选单逐日（期限 {h0} 日；第一节按日等权的读数就是这张表的平均，看集中度）", "",
           _md(daily_c), "",
           "## 三、按事后环境分组（未来 h 日全池涨跌，只作解释，不作交易前置）", "",

sources.py

@@ -794,3 +794,240 @@ def close_prices(ds: str, read_mysql=None, code_col: str | None = None) -> dict:
         if k and px and px > 0:
             out[k] = px
     return out
+
+
+# ============================================================================
+# 催化事件与事件日字段（2026-09-08《量价研判链吸收方案》3.4）：只展示，不进判决
+# ============================================================================
+#
+# 研报的起点是四类券商正向事件："间隔一年后深度覆盖推荐买入"、"主动上调盈利预测"、"研报标题含
+# 业绩超预期"、"两者兼有"。这四类全部能从券商研报明细表 gp_report_rc 复现（报告类型、标题、评级、
+# 每股收益历史都在），不用动数据基座。阈值一次定死（台账 045）：
+#   深度覆盖   报告类型是"深度"，评级是买入类，且该票在这篇之前 365 天内没有任何研报
+#   上调预测   同一家机构对同一预测期，180 天内上一篇每股收益为正且这一篇高出五成以上
+#   超预期     标题含"超预期"
+#   事件窗口   数据日往前 60 个自然日；同一天多篇合并成一个事件（研报的"同日复合"）
+# 它是论点卡"催化剂"一栏的第一个数据源（此前标无数据源），也是送研判的事件上下文。
+EVENT_WINDOW_DAYS = 60
+EVENT_COVER_GAP_DAYS = 365
+EVENT_UPGRADE_LOOKBACK_DAYS = 180
+EVENT_UPGRADE_RATIO = 1.5
+EVENT_KEEP = 5                     # 每票最多带几条事件到卡上
+BUY_RATINGS = {"买入", "增持", "推荐", "强烈推荐", "强推", "跑赢行业", "优于大市", "买入-A", "买入-B",
+               "推荐-A", "增持-A", "审慎增持", "谨慎增持", "优于大市评级"}
+EV_DEEP, EV_UPGRADE, EV_BEAT = "深度覆盖", "上调盈利预测", "业绩超预期"
+
+
+def analyst_reports(codes, ds: str, *, days: int = EVENT_WINDOW_DAYS + EVENT_COVER_GAP_DAYS,
+                    read_mysql=None) -> list:
+    """券商研报明细表的原始行：报告日、类型、标题、评级、机构、预测期、每股收益。窗口要盖住
+    事件窗口加"前 365 天有没有覆盖"的回看，所以默认取 425 天。读失败返回空列表并打印原因。"""
+    reader = read_mysql or db.read_mysql
+    end = dt.date.fromisoformat(ds)
+    start = end - dt.timedelta(days=int(days))
+    dotted = sorted({_to_dot(c) for c in codes if c})
+    if not dotted:
+        return []
+    marks = ",".join(["%s"] * len(dotted))
+    try:
+        df = reader(
+            "factor",
+            f"SELECT ts_code, report_date, report_type, report_title, rating, org_name, quarter, eps "
+            f"FROM gp_report_rc WHERE ts_code IN ({marks}) AND report_date > %s AND report_date <= %s",
+            tuple(dotted) + (start.isoformat(), end.isoformat()))
+    except Exception as e:  # noqa: BLE001
+        print(f"  （券商研报明细表读取失败，催化事件这一行整体缺席: {e!r}）")
+        return []
+    out = []
+    for r in _records(df):
+        d = _ymd(r.get("report_date"))
+        if not d:
+            continue
+        out.append({"k": common.to_prefix(str(r.get("ts_code") or "").strip()), "date": d,
+                    "type": str(r.get("report_type") or "").strip(),
+                    "title": str(r.get("report_title") or "").strip(),
+                    "rating": str(r.get("rating") or "").strip(),
+                    "org": str(r.get("org_name") or "").strip() or "未署名",
+                    "quarter": str(r.get("quarter") or "").strip(), "eps": _f(r.get("eps"))})
+    return out
+
+
+def analyst_events(codes, ds: str, *, rows=None, window_days: int = EVENT_WINDOW_DAYS,
+                   read_mysql=None) -> dict:
+    """每只票近 window_days 天的券商正向事件，按前缀码索引；没有事件的票不在字典里。
+
+    返回 {k: {"latest": 最近事件日, "count": 事件天数, "events": [{date, types, orgs, title, n_reports,
+    compound}, ...] 最新在前}}。types 是这一天命中的事件类型列表；compound 为真表示同一篇研报
+    同时命中上调预测与超预期（研报的第四类），或同一天多篇研报命中不同类型。"""
+    if rows is None:
+        rows = analyst_reports(codes, ds, read_mysql=read_mysql)
+    by_code: dict = defaultdict(list)
+    for r in rows:
+        by_code[r["k"]].append(r)
+    try:
+        cut = (dt.date.fromisoformat(ds) - dt.timedelta(days=int(window_days))).isoformat()
+    except ValueError:
+        return {}
+    out = {}
+    for k, rs in by_code.items():
+        rs = sorted(rs, key=lambda r: r["date"])
+        dates = [r["date"] for r in rs]
+        by_org_q: dict = defaultdict(list)
+        for r in rs:
+            if r["eps"] is not None:
+                by_org_q[(r["org"], r["quarter"])].append(r)
+        days_hit: dict = {}
+        for r in rs:
+            if r["date"] <= cut or r["date"] > ds:
+                continue
+            types = []
+            if "超预期" in r["title"]:
+                types.append(EV_BEAT)
+            if r["type"] == "深度" and r["rating"] in BUY_RATINGS:
+                gap_start = (dt.date.fromisoformat(r["date"])
+                             - dt.timedelta(days=EVENT_COVER_GAP_DAYS)).isoformat()
+                # 这篇之前 365 天内有没有任何研报（不含同一天）
+                if not any(gap_start < d < r["date"] for d in dates):
+                    types.append(EV_DEEP)
+            if r["eps"] is not None and r["eps"] > 0:
+                look = (dt.date.fromisoformat(r["date"])
+                        - dt.timedelta(days=EVENT_UPGRADE_LOOKBACK_DAYS)).isoformat()
+                prev = [p for p in by_org_q[(r["org"], r["quarter"])]
+                        if look < p["date"] < r["date"] and p["eps"] and p["eps"] > 0]
+                if prev and r["eps"] >= EVENT_UPGRADE_RATIO * prev[-1]["eps"]:
+                    types.append(EV_UPGRADE)
+            if not types:
+                continue
+            ev = days_hit.setdefault(r["date"], {"date": r["date"], "types": [], "orgs": [],
+                                                 "title": r["title"][:60], "n_reports": 0,
+                                                 "compound": False})
+            for t in types:
+                if t not in ev["types"]:
+                    ev["types"].append(t)
+            if r["org"] not in ev["orgs"]:
+                ev["orgs"].append(r["org"])
+            ev["n_reports"] += 1
+            if EV_UPGRADE in types and EV_BEAT in types:
+                ev["compound"] = True
+        if not days_hit:
+            continue
+        events = sorted(days_hit.values(), key=lambda e: e["date"], reverse=True)
+        for e in events:
+            if len(e["types"]) > 1:
+                e["compound"] = True
+        out[k] = {"latest": events[0]["date"], "count": len(events), "events": events[:EVENT_KEEP]}
+    return out
+
+
+# 事件日字段：研报"输入事实"的第三块。事件前 5 日与 20 日涨幅判有没有抢跑，事件日跳空、日内收益、
+# 收盘位置、量比、涨停判市场有没有确认。行情表 gp_day_data 是前复权（用户 09-08 确认），跳空与
+# 涨幅直接算。没有事件的票按数据日算同一组数（那时它回答的是"今天这根 K 线的样子"）。
+PRICE_HISTORY_DAYS = 100           # 有事件的票取多少天行情：事件最远 60 天，前面还要 20 个交易日算量与涨幅
+PRICE_HISTORY_DAYS_NO_EVENT = 35   # 没有事件的票按数据日算，只要 21 个交易日的历史
+
+
+def _limit_up_threshold(k: str) -> float:
+    """涨停线（百分数）：创业板与科创板两成，北交所三成，其余一成。用 9.8 而不是 10 是给四舍五入留余地。"""
+    s = str(k or "")
+    num = s[2:] if len(s) == 8 else s
+    if num.startswith(("30", "68")):
+        return 19.8
+    if num.startswith(("4", "8")):
+        return 29.8
+    return 9.8
+
+
+def price_history(codes, ds: str, *, days: int = PRICE_HISTORY_DAYS, read_mysql=None,
+                  code_col: str | None = None) -> dict:
+    """每只票近 days 个自然日的前复权日线，按前缀码索引、按日升序。行情表的代码列是前缀式。"""
+    reader = read_mysql or db.read_mysql
+    codes = sorted({_prefix_any(c) for c in codes if c})
+    if not codes:
+        return {}
+    try:
+        if code_col is None:
+            import factors
+            code_col = factors._price_code_col()  # noqa: SLF001 —— 同仓自用
+        end = dt.date.fromisoformat(ds)
+        start = (end - dt.timedelta(days=int(days))).isoformat()
+        nxt = (end + dt.timedelta(days=1)).isoformat()
+        marks = ",".join(["%s"] * len(codes))
+        df = reader("price",
+                    f"SELECT `{code_col}` AS ts_code, `timestamp` AS d, open, high, low, close, "
+                    f"pre_close, percent, volume FROM gp_day_data "
+                    f"WHERE `timestamp` >= %s AND `timestamp` < %s AND `{code_col}` IN ({marks})",
+                    (start, nxt) + tuple(codes))
+    except Exception as e:  # noqa: BLE001
+        print(f"  （行情表读取失败，事件日字段与定价状态整体缺席: {e!r}）")
+        return {}
+    out: dict = defaultdict(list)
+    for r in _records(df):
+        k = _prefix_any(r.get("ts_code"))
+        d = _ymd(r.get("d"))
+        if not k or not d:
+            continue
+        out[k].append({"date": d, "open": _f(r.get("open")), "high": _f(r.get("high")),
+                       "low": _f(r.get("low")), "close": _f(r.get("close")),
+                       "pre_close": _f(r.get("pre_close")), "pct": _f(r.get("percent")),
+                       "volume": _f(r.get("volume"))})
+    return {k: sorted(v, key=lambda r: r["date"]) for k, v in out.items()}
+
+
+def event_day_fields(codes, ds: str, events: dict, *, hist=None, read_mysql=None,
+                     code_col: str | None = None) -> dict:
+    """每只票的事件日字段，按前缀码索引。events 是 analyst_events 的返回；没有事件的票按数据日算。
+
+    六个数：事件前 5 日与 20 日涨幅、事件日跳空幅度（开盘对前收）、日内收益（收盘对开盘）、
+    收盘位置（收盘在当日高低区间里的位置，0 到 1）、事件日量比（对此前 20 个交易日均量），
+    外加当日涨幅与是否涨停。行情不够时相应字段为 None，不硬算。"""
+    if hist is None:
+        # 两段取：没有事件的票按数据日算，只要 21 个交易日的历史（35 个自然日够）；有事件的票
+        # 事件最远 60 天，前面再要 20 个交易日，取 100 天。一段取 100 天要拉十二万行、十秒多，
+        # 计划的实时重算撑不起（PMS 拉计划的超时是 60 秒），分两段行数少六成。
+        hist = price_history(codes, ds, days=PRICE_HISTORY_DAYS_NO_EVENT, read_mysql=read_mysql,
+                             code_col=code_col)
+        with_event = [k for k in (events or {}) if k in {str(c).strip() for c in (codes or []) if c}]
+        if with_event:
+            hist.update(price_history(with_event, ds, days=PRICE_HISTORY_DAYS, read_mysql=read_mysql,
+                                      code_col=code_col))
+    out = {}
+    for k in {str(c).strip() for c in (codes or []) if c}:
+        rows = hist.get(k) or []
+        if not rows:
+            continue
+        ev = (events or {}).get(k)
+        target = ev["latest"] if ev else ds
+        idx = None
+        for i, r in enumerate(rows):
+            if r["date"] <= target:
+                idx = i
+        if idx is None:
+            continue
+        cur = rows[idx]
+        closes = [r["close"] for r in rows]
+
+        def _ret(back: int):
+            if idx - back < 0 or not closes[idx - 1] or not closes[idx - back - 1]:
+                return None
+            return closes[idx - 1] / closes[idx - back - 1] - 1
+
+        prev_close = cur["pre_close"] or (closes[idx - 1] if idx >= 1 else None)
+        gap = (cur["open"] / prev_close - 1) if cur["open"] and prev_close else None
+        intraday = (cur["close"] / cur["open"] - 1) if cur["close"] and cur["open"] else None
+        rng = (cur["high"] - cur["low"]) if cur["high"] is not None and cur["low"] is not None else None
+        close_pos = ((cur["close"] - cur["low"]) / rng) if rng and cur["close"] is not None else None
+        vols = [r["volume"] for r in rows[max(0, idx - 20):idx] if r["volume"]]
+        vol_ratio = (cur["volume"] / (sum(vols) / len(vols))) if cur["volume"] and len(vols) >= 5 else None
+        pct = cur["pct"]
+        out[k] = {"event_date": cur["date"], "has_event": bool(ev),
+                  "pre5": _r4(_ret(5)), "pre20": _r4(_ret(20)),
+                  "gap": _r4(gap), "intraday": _r4(intraday), "close_pos": _r4(close_pos),
+                  "vol_ratio": None if vol_ratio is None else round(vol_ratio, 2),
+                  "day_pct": None if pct is None else round(pct / 100.0, 4),
+                  "limit_up": None if pct is None else bool(pct >= _limit_up_threshold(k)),
+                  "history_days": idx}
+    return out
+
+
+def _r4(v):
+    return None if v is None else round(float(v), 4)

test_events_pricing.py

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