diff --git a/config.py b/config.py index 0b83039..7ad56bc 100644 --- a/config.py +++ b/config.py @@ -240,6 +240,17 @@ MARKET_MYSQL_SOURCE = os.environ.get("MARKET_MYSQL_SOURCE", "price").strip().low # 只展示、不作门槛、不进判决；设 0 表示不读该视图（视图未建时也可用它关掉那一行告警）。 LOGIC_CLAIMS_PER_STOCK = int(os.environ.get("LOGIC_CLAIMS_PER_STOCK", "3")) +# 四态甲路吃的论断上限（2026-09-07）。它与上面那个"卡上展示几条"是两回事：展示三条够看， +# 但判跨期翻转与同期分歧要看全量——三条里"之前的论断"最多两条，翻转两头都会判错。 +# 200 与覆盖率读数脚本同口径（实测单票最多 44 条，200 是护栏不是预期）。 +LOGIC_CLAIMS_FULL = int(os.environ.get("LOGIC_CLAIMS_FULL", "200")) + +# 乙路硬触发的维持窗口（计划日数，2026-09-07）。采信倾向从偏多迁移到偏空那一天记 +# migrated=1，次日归零；不带历史的话硬触发只维持一个计划日，而倾向仍偏空并未回稳。 +# 规则是：近 N 个计划日内出现过这次迁移且此后一直偏空，硬触发就仍成立。N 取 5， +# 与四态退出存疑要求"连续三个计划日不再存疑"同一量级、再留两天余量。一次定死。 +JUDGEMENT_HOLD_DAYS = int(os.environ.get("JUDGEMENT_HOLD_DAYS", "5")) + # 论断陈旧线（自然日）。超过它就在候选卡的缺失项里写明"研报论断已过 N 天"，只提示不作门槛。 # 取 90 天的理由是与吸筹评分的日龄上限可比：那一条是三十个交易日，约合六周多的自然日， # 而研报的更新节奏本来就比资金结构评分慢得多（行业深度按季度、跟踪报告按事件）， diff --git a/judgement.py b/judgement.py index e069e50..f7cc119 100644 --- a/judgement.py +++ b/judgement.py @@ -186,6 +186,36 @@ def by_segment_name(snaps: dict) -> dict: return out +def recent_rows(day: str, days: int = 5, read_mysql=None) -> dict: + """每个簇最近 days 个计划日的快照行，按簇键索引、行按计划日升序，只取严格早于 day 的。 + + 给四态乙路判"硬触发要不要维持"用（2026-09-07）：迁移只在指纹变化那天记 migrated=1， + 次日归零，光看当天那一行硬触发只能维持一个计划日。一次查询取回全部簇的近几行， + 不逐簇查——计划装配一轮要看几百只票，逐簇查是几百次往返。 + 读不到返回空字典，那样乙路退回"只看当天一行"，不断产。 + """ + table = config.JUDGEMENT_SNAPSHOT_TABLE + reader = read_mysql or db.read_mysql + try: + start = (dt.date.fromisoformat(day) - dt.timedelta(days=int(days) * 2 + 7)).isoformat() + except ValueError: + return {} + try: + rows = sources._records(reader( # noqa: SLF001 —— 同仓自用 + "factor", + f"SELECT * FROM {table} WHERE plan_date >= %s AND plan_date < %s " + f"ORDER BY cluster_key, plan_date", (start, day))) + except Exception as e: # noqa: BLE001 + print(f" （{table} 读不到近日行，乙路只看当天一行: {e!r}）") + return {} + out: dict = {} + for r in rows: + key = str(r.get("cluster_key") or "").strip() + if key: + out.setdefault(key, []).append({k: _blank_to_none(v) for k, v in dict(r).items()}) + return {k: v[-int(days):] for k, v in out.items()} + + def load_previous(day: str, read_mysql=None) -> dict: """本计划日之前、回看窗口之内，每个簇最近一次快照的行，按簇键索引。 @@ -239,15 +269,29 @@ def save(day: str, rows: list, conn_factory=None) -> None: conn.commit() -def snapshot(day: str | None = None, fetch=None, load_prev=None, write=None) -> dict: +def snapshot(day: str | None = None, fetch=None, load_prev=None, write=None, + calendar=None) -> dict: """一个计划日的完整一步：取最新一版结论、配上一版比对、幂等写当日行、打印读数。 day 不传取今天。补写一个早于今天的计划日要当心：视图给的永远是此刻的最新一版， 把它记在过去某一天名下，等于把今天的观点写成那天的观察结果，版本史会失真—— 所以这种情况下会多打印一行提醒，但仍照写（重跑与补跑都幂等）。 - 三个函数都可注入（离线单测）：取数、读上一版、落库。""" - day = day or dt.date.today().isoformat() - if day != dt.date.today().isoformat(): + 三个函数都可注入（离线单测）：取数、读上一版、落库。calendar 注入交易日历（升序 ISO 串）。 + + 非交易日不落行（2026-09-07 审查修）。快照按工作日 20:40 投递、plan_date 取当天， + 法定假日（工作日）那行 plan_date 大于节前的数据日、小于节后首个数据日；计划读的是 + plan_date 不晚于数据日的最新一行，节后计划读到的是与假日行比对得到的 migrated=0， + 指纹迁移若首次被假日快照记录，任何计划都读不到那个 migrated=1，乙路的硬触发就丢了。 + 国庆那一周连续五个工作日假日，几乎必然命中。所以：今天不是交易日就不写，直接返回。""" + today = dt.date.today().isoformat() + if day is None: + cal = calendar if calendar is not None else sources.trading_days(today, back_days=21) + if cal and today not in cal: + print(f" 行业观点快照：{today} 不是交易日，不落行（最近交易日 {cal[-1]}）") + return {"date": today, "rows": 0, "skipped": "非交易日", "last_trading_day": cal[-1], + "table": config.JUDGEMENT_SNAPSHOT_TABLE} + day = today + if day != today: print(f" ⚠️ 计划日 {day} 不是今天：抄下来的是此刻的最新一版结论，" f"记在过去的日期名下会让版本史失真，确认是有意为之再用。") rows_in = (fetch or sources.judgement_rows)() diff --git a/logic_state.py b/logic_state.py index 74ad243..1d9d31c 100644 --- a/logic_state.py +++ b/logic_state.py @@ -188,7 +188,8 @@ def _refs(rows, limit: int = 3) -> list: # 乙路：产业研判 # ============================================================================ -def from_judgement(cur, prev=None, *, stale_days: int = 60) -> dict: +def from_judgement(cur, prev=None, *, stale_days: int = 60, + history=None, hold_days: int = 5) -> dict: """把一个主题的产业研判快照归一成一路信号。cur 与 prev 是 judgement.build_rows 的行。 判的是采信倾向这个受控枚举的迁移，不是它当前的取值。理由是强化说的是"变好了"： @@ -198,6 +199,12 @@ def from_judgement(cur, prev=None, *, stale_days: int = 60) -> dict: 只有材料指纹变过才算一次迁移。指纹没变而日子变老，只累加陈旧天数；超过 stale_days 按缺失处理，因为产业研判是人工点按钮触发的，长期不变说明没人重新看过它， 不是"看过了没变"。 + + history 是这个簇最近若干个计划日的快照行（按日升序，不含 cur），2026-09-07 加的。 + 没有它，硬触发只维持一个计划日：迁移只在指纹变化那天记 migrated=1，次日就归零， + 而采信倾向仍是偏空、并未回稳，存疑却自动解除。方案定的是"进入即刻成立，退出要 + 连续若干日同向"。这里的规则：近 hold_days 个计划日内出现过偏多到偏空的迁移， + 且此后一直偏空，硬触发就仍然成立；一旦倾向离开偏空，立刻不再算。 """ if not cur: return signal(PATH_JUDGE, SIG_NONE, why="这个环节对不上任何产业研判主题") @@ -231,12 +238,35 @@ def from_judgement(cur, prev=None, *, stale_days: int = 60) -> dict: return signal(PATH_JUDGE, SIG_FLAT, as_of=as_of, coverage=cov, refs=refs, why=f"产业研判的采信倾向从{prev_lean}迁移到{lean}，不是多空方向的变化") if lean == LEAN_BEAR: + # 近几个计划日内翻过空、此后一直偏空 → 硬触发仍然成立（2026-09-07）。 + flip_day = _recent_bear_flip(history, hold_days) + if flip_day: + return signal(PATH_JUDGE, SIG_DOWN, as_of=as_of, coverage=cov, hard=True, refs=refs, + why=f"产业研判 {flip_day} 从偏多迁移到偏空，此后一直偏空") return signal(PATH_JUDGE, SIG_DOWN, as_of=as_of, coverage=cov, refs=refs, why="产业研判的采信倾向是偏空") return signal(PATH_JUDGE, SIG_FLAT, as_of=as_of, coverage=cov, refs=refs, why=f"产业研判的采信倾向是{lean}，与上一版相同") +def _recent_bear_flip(history, hold_days: int): + """近 hold_days 行里最后一次"偏多→偏空"的迁移日；迁移之后若倾向离开过偏空，返回 None。 + + history 按日升序、不含当天。只看末尾 hold_days 行：更早的翻转已经超出方案定的 + 退出窗口，不该再撑着硬触发。 + """ + rows = [r for r in (history or []) if isinstance(r, dict)][-int(hold_days):] + flip = None + for r in rows: + lean = _s(r.get("leaning")) + if r.get("migrated") == 1 and _s(r.get("leaning_prev")) == LEAN_BULL and lean == LEAN_BEAR: + flip = _s(r.get("plan_date")) or _s(r.get("review_date")) or "近日" + continue + if flip and lean != LEAN_BEAR: + flip = None # 翻空之后又离开了偏空，硬触发作废 + return flip + + # ============================================================================ # 丙路：券商行动 # ============================================================================ diff --git a/plan.py b/plan.py index 1dce725..23c2cd3 100644 --- a/plan.py +++ b/plan.py @@ -180,16 +180,20 @@ def _segments_of(ds: str) -> dict: def _logic_state_of(k: str, evd: dict, seg_of: dict, seg_view: dict, - broker: dict, ds: str) -> dict: + broker: dict, ds: str, full_logic=None, seg_hist=None) -> dict: """一只票的逻辑状态四态。四路各自归一，再按合成规则合成。 乙路的取法：一只票可能挂在多个被指向的环节上，取第一个有行业观点的那个。 取不到就是缺失，卡上会写明缺的是哪一路——缺失既不算负面也不算正面证据， 但必须让人看得见系统缺的是什么，不能让人以为系统判过了。 """ - a = logic_state.from_claims(evd.get("logic"), ds, stale_days=config.LOGIC_STALE_DAYS) + # 甲路吃全量论断（2026-09-07）；没传全量时退回卡上那几条，只为让旧调用与单测不断。 + claims = full_logic if full_logic is not None else evd.get("logic") + a = logic_state.from_claims(claims, ds, stale_days=config.LOGIC_STALE_DAYS) row = next((seg_view[t] for t in seg_of.get(k, []) if t in seg_view), None) - b = logic_state.from_judgement(row) + # 乙路带上这个簇近几个计划日的历史，让"偏多到偏空"的硬触发能维持住（2026-09-07）。 + hist = (seg_hist or {}).get(str(row.get("cluster_key") or "")) if row else None + b = logic_state.from_judgement(row, history=hist, hold_days=config.JUDGEMENT_HOLD_DAYS) c = broker.get(k) or logic_state.signal( logic_state.PATH_BROKER, logic_state.SIG_NONE, why="两个等长窗口里算不出可比的每股收益预测") @@ -211,7 +215,12 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series, daily = sources.stock_daily(ds) night = sources.night_conclusions(codes, ds) # 因果论断（2026-09-03）：数据基座抽取的论断挂在卡上作证据线，只展示不进判决；视图未建时为空。 - logic = sources.logic_claims(codes, ds) + # 2026-09-07 审查修：四态的甲路要吃**全量**论断，卡上展示只取最近几条。原先两者共用 + # 一份截断到三条的列表，跨期翻转与同期分歧都只在三条上算，翻转两头都会判错—— + # 历史八空两好加最新一空，全量判无法判断、三条判逻辑存疑；历史五好近三空则反过来。 + # 全量取一次，展示从里面切前几条，取数层在截断前算的质量画像照旧挂在每条上。 + logic_full = sources.logic_claims(codes, ds, per_stock=config.LOGIC_CLAIMS_FULL) + logic = {k: v[:config.LOGIC_CLAIMS_PER_STOCK] for k, v in logic_full.items()} # 逻辑状态四态的三路输入（丁路公司事件无数据源，logic_state 那边恒出缺失）。 # 这三路都是"研究证据还在不在"的跟踪，与候选卡的三门槛判决是正交的两维： # 判决回答今天要不要买，四态回答支撑它的研究证据还在不在。收敛规则在 @@ -223,6 +232,8 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series, # 调度中心一直在按时跑）。load_previous(D+1) 取的正是 plan_date <= D 的每簇最新一行， # 顺带也扛得住某个晚上没跑：会退回上一版，陈旧天数照常累加。 seg_view = judgement.by_segment_name(judgement.load_previous(_next_day(ds))) + # 每簇近几个计划日的快照，给乙路判"硬触发要不要维持"（2026-09-07 审查修）。 + seg_hist = judgement.recent_rows(_next_day(ds), days=config.JUDGEMENT_HOLD_DAYS) broker = sources.broker_actions(codes, ds) seg_of = _segments_of(ds) if risk is None: # collect 会传入读过一次的名单；单独调用时自己读 @@ -245,7 +256,8 @@ def _assemble_cards(ds: str, codes: list, ev: dict, upside: pd.Series, "accum_state": n.get("accum_state"), "accum_score": n.get("accum_score"), "accum_age": n.get("accum_age"), "y_signal": n.get("signal"), "stale_snapshot": stale, "logic": logic.get(k) or []} - state = _logic_state_of(k, evd, seg_of, seg_view, broker, ds) + state = _logic_state_of(k, evd, seg_of, seg_view, broker, ds, + full_logic=logic_full.get(k) or [], seg_hist=seg_hist) j = card.judge(evd, start_pct=config.CARD_START_PCT, accum_max_age=config.CARD_ACCUM_MAX_AGE, neg_tol=config.UPSIDE_NEG_TOLERANCE, diff --git a/test_judgement_snapshot.py b/test_judgement_snapshot.py index 78632ea..539223c 100644 --- a/test_judgement_snapshot.py +++ b/test_judgement_snapshot.py @@ -273,8 +273,45 @@ def test_migration_on_fingerprint_change(): t("空版本史不报错", judgement.migrations([]) == [] and judgement.migrations(None) == []) +def test_recent_rows_and_holiday_guard(): + print("近日行读取与非交易日守卫（审查 2026-09-07 第 9、10 条）") + rows = [ + {"plan_date": "2026-09-01", "cluster_key": "Segment::甲", "leaning": "偏多", "migrated": 0}, + {"plan_date": "2026-09-02", "cluster_key": "Segment::甲", "leaning": "偏空", "migrated": 1, + "leaning_prev": "偏多"}, + {"plan_date": "2026-09-03", "cluster_key": "Segment::甲", "leaning": "偏空", "migrated": 0}, + {"plan_date": "2026-09-03", "cluster_key": "Segment::乙", "leaning": "偏多", "migrated": None, + "n_bull": float("nan")}, + ] + got = judgement.recent_rows("2026-09-04", days=2, read_mysql=lambda src, sql, params: rows) + t("按簇索引、只留最近 days 行、按日升序", + [r["plan_date"] for r in got["Segment::甲"]] == ["2026-09-02", "2026-09-03"]) + t("NaN 归一成 None", got["Segment::乙"][0]["n_bull"] is None) + t("读失败返回空字典不抛错", judgement.recent_rows("2026-09-04", read_mysql=_boom) == {}) + t("日期认不出返回空字典", judgement.recent_rows("不是日期", read_mysql=lambda *a, **k: rows) == {}) + + wrote = [] + r = judgement.snapshot(None, fetch=lambda: [], load_prev=lambda d: {}, + write=lambda d, rs: wrote.append((d, rs)), calendar=["2000-01-03"]) + t("今天不在交易日历里 -> 不落行、返回跳过", r.get("skipped") == "非交易日" and wrote == []) + import datetime as _dt + today = _dt.date.today().isoformat() + r = judgement.snapshot(None, fetch=lambda: [], load_prev=lambda d: {}, + write=lambda d, rs: wrote.append((d, rs)), calendar=["2000-01-03", today]) + t("今天是交易日 -> 照常落行", r.get("rows") == 0 and wrote and wrote[0][0] == today) + wrote.clear() + r = judgement.snapshot(None, fetch=lambda: [], load_prev=lambda d: {}, + write=lambda d, rs: wrote.append((d, rs)), calendar=[]) + t("日历读不到 -> 不拦，按今天落行（不让守卫本身成为断产点）", wrote and wrote[0][0] == today) + wrote.clear() + r = judgement.snapshot("2026-09-01", fetch=lambda: [], load_prev=lambda d: {}, + write=lambda d, rs: wrote.append((d, rs)), calendar=["2000-01-03"]) + t("显式给了计划日 -> 不查日历，照写", wrote and wrote[0][0] == "2026-09-01") + + def main(): test_fetch() + test_recent_rows_and_holiday_guard() test_fetch_failure() test_prev_read_failure() test_same_day_idempotent() diff --git a/test_logic_state.py b/test_logic_state.py index 858a38f..40b344d 100644 --- a/test_logic_state.py +++ b/test_logic_state.py @@ -98,6 +98,28 @@ def test_judgement(): t("迁到中性 -> 平稳（不是多空方向的变化）", r["signal"] == ls.SIG_FLAT) +def test_judgement_history(): + print("乙路 硬触发的维持窗口（审查 2026-09-07 第 9 条）") + # 迁移只在指纹变化那天记 migrated=1，次日归零；不带历史时硬触发只维持一个计划日。 + flip = jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=1, plan_date="2026-09-02") + after1 = jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BEAR, migrated=0, plan_date="2026-09-03") + today = jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BEAR, migrated=0, plan_date="2026-09-04") + r = ls.from_judgement(today) + t("不带历史：迁移次日之后只是转弱、不硬", r["signal"] == ls.SIG_DOWN and not r["hard"]) + r = ls.from_judgement(today, history=[flip, after1]) + t("带历史：近几日翻过空且此后一直偏空 -> 硬触发维持", r["hard"] and "此后一直偏空" in r["why"]) + back = jrow(leaning=ls.LEAN_BULL, prev=ls.LEAN_BEAR, migrated=1, plan_date="2026-09-03") + r = ls.from_judgement(today, history=[flip, back]) + t("翻空之后又回到偏多过一次 -> 硬触发作废", not r["hard"]) + old = [jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=1, plan_date="2026-08-20")] \ + + [jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BEAR, migrated=0, plan_date=f"2026-08-2{i}") + for i in range(1, 7)] + r = ls.from_judgement(today, history=old, hold_days=5) + t("翻空早于维持窗口 -> 不再撑着硬触发", not r["hard"]) + r = ls.from_judgement(today, history=[None, "坏行", flip]) + t("历史里的坏行被跳过、不抛错", r["hard"]) + + def test_broker(): print("丙路 券商行动") r = ls.from_broker({"eps": 2.3, "firms": 1}, {"eps": 2.4, "firms": 5}, as_of=DS) @@ -241,11 +263,12 @@ def test_apply(): def main(): test_claims() test_judgement() + test_judgement_history() test_broker() test_compose() test_settle() test_apply() - print("ALL OK — 四路归一 / 合成四条规则 / 四态入口 / 负向优先 / 进出不对称 / " + print("ALL OK — 四路归一 / 乙路硬触发维持窗口 / 合成四条规则 / 四态入口 / 负向优先 / 进出不对称 / " "作用于候选卡的单调性 全部通过") diff --git a/test_plan_logic_state.py b/test_plan_logic_state.py index 8e6c4b3..dbf5aec 100644 --- a/test_plan_logic_state.py +++ b/test_plan_logic_state.py @@ -85,7 +85,24 @@ def main(): t("次日推算正确", plan._next_day("2026-09-03") == "2026-09-04") # noqa: SLF001 t("认不出的日期原样返回，不抛异常", plan._next_day("不是日期") == "不是日期") # noqa: SLF001 - print("ALL OK — 四态随卡产出 / 缺席不断产 / 下游形状 / 判决不被改动 / 计划日推算 全部通过") + print("乙路历史传到计划装配（审查 2026-09-07 第 9 条）") + cur = jrow(leaning="偏空") + cur["cluster_key"] = "Segment::固态电解质" + hist = {"Segment::固态电解质": [ + {"plan_date": "2026-09-01", "leaning": "偏空", "leaning_prev": "偏多", "migrated": 1}, + {"plan_date": "2026-09-02", "leaning": "偏空", "leaning_prev": "偏空", "migrated": 0}]} + st = plan._logic_state_of("SH600000", {"logic": []}, {"SH600000": ["固态电解质"]}, # noqa: SLF001 + {"固态电解质": cur}, {}, DS, seg_hist=hist) + t("近日翻空且一直偏空 -> 硬触发维持 -> 逻辑存疑", st["state"] == ls.STATE_DOUBT) + st = plan._logic_state_of("SH600000", {"logic": []}, {"SH600000": ["固态电解质"]}, # noqa: SLF001 + {"固态电解质": cur}, {}, DS) + t("不传历史时退回只看当天一行（不断产）", st["state"] != ls.STATE_DOUBT) + full = [claim("2026-07-10"), claim("2026-07-20"), claim("2026-08-28", "利空")] + st = plan._logic_state_of("SH600000", {"logic": full[-1:]}, {}, {}, {}, DS, full_logic=full) # noqa: SLF001 + t("甲路吃全量论断：三条里判得出跨期翻转", any(p["path"] == ls.PATH_CLAIM and p["signal"] == ls.SIG_DOWN + for p in st["paths"]) and st["state"] == ls.STATE_DOUBT) + + print("ALL OK — 四态随卡产出 / 缺席不断产 / 下游形状 / 判决不被改动 / 计划日推算 / 乙路历史与甲路全量 全部通过") if __name__ == "__main__":