From 1fc874b3136baebf93c9124d2efcfc8179e4cd95 Mon Sep 17 00:00:00 2001 From: zlt Date: Tue, 8 Sep 2026 11:40:51 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8F=E4=BB=B7=E7=A0=94=E5=88=A4=E9=93=BE?= =?UTF-8?q?=203.4=EF=BC=9A=E5=82=AC=E5=8C=96=E4=BA=8B=E4=BB=B6=E3=80=81?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E6=97=A5=E5=AD=97=E6=AE=B5=E4=B8=8E=E5=AE=9A?= =?UTF-8?q?=E4=BB=B7=E7=8A=B6=E6=80=81=E3=80=81=E5=91=A8=E6=8A=A5=E5=88=86?= =?UTF-8?q?=E7=BB=84=EF=BC=88=E5=8F=AA=E5=B1=95=E7=A4=BA=E4=B8=8D=E8=BF=9B?= =?UTF-8?q?=E5=88=A4=E5=86=B3=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- card.py | 86 +++++++++++++++ plan.py | 19 +++- plan_review.py | 46 ++++++++ sources.py | 237 +++++++++++++++++++++++++++++++++++++++++ test_events_pricing.py | 200 ++++++++++++++++++++++++++++++++++ 5 files changed, 586 insertions(+), 2 deletions(-) create mode 100644 test_events_pricing.py diff --git a/card.py b/card.py index 0e7a2a5..2426a09 100644 --- a/card.py +++ b/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%}" diff --git a/plan.py b/plan.py index b5ae734..0760d90 100644 --- a/plan.py +++ b/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)} 个环节：定位对不对看这里，挑票看候选单）") diff --git a/plan_review.py b/plan_review.py index b490cd0..aabc59e 100644 --- a/plan_review.py +++ b/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 日全池涨跌，只作解释，不作交易前置）", "", diff --git a/sources.py b/sources.py index cffe1ce..baaed6a 100644 --- a/sources.py +++ b/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) diff --git a/test_events_pricing.py b/test_events_pricing.py new file mode 100644 index 0000000..a19d7c0 --- /dev/null +++ b/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()