From e4f87b449be61087c0dfe9f357cb3d8a78632d6a Mon Sep 17 00:00:00 2001 From: zlt Date: Mon, 17 Aug 2026 15:27:45 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=84=E7=90=86=E9=80=89=E8=82=A1=E6=96=B9?= =?UTF-8?q?=E6=A1=88=EF=BC=8C=E7=9B=AE=E7=9A=84=E6=98=AF=E5=92=8C=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=95=E5=BA=A7=E4=B8=80=E8=87=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config.py | 19 ++++- factors.py | 10 ++- plan_reconcile.py | 7 +- pool.py | 14 +++- score_lab.py | 203 ++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 240 insertions(+), 13 deletions(-) create mode 100644 score_lab.py diff --git a/config.py b/config.py index 4dbe9bb..3d39547 100644 --- a/config.py +++ b/config.py @@ -147,6 +147,15 @@ ENABLE_TRACK_GATE = os.environ.get("ENABLE_TRACK_GATE", "0") == "1" # 因此不进主榜、观察档、计划与升降档。研究口径想看全貌时置 0 关闭。 EXCLUDE_RISK_NAMES = os.environ.get("EXCLUDE_RISK_NAMES", "1") == "1" +# --- 预期空间否决的负容忍（2026-08-17 拍板：默认否决，参数控制）---------------- +# 主榜判据原文是「upside >= 0，贵了不买是绝对下限」。默认 0.0 = 与拍板逐字一致。 +# 背景：upside 是"报喜不报忧"的噪音口径（下游文档明写"当相对排序用"），传导强而 +# upside 轻微为负的票会整只消失（实例：太极实业传导分 1.89、两源指向，因 -52% 被否 +# ——这个量级该否；但 -3% 也同样消失）。把阈值放开到如 0.10，即 upside >= -10% 仍进 +# 主榜，组内分的 0.4·z(upside) 自然把它压到组内低位（只降权不否决）。是否放开、放多少， +# 等 score_lab 对比器跑出数据再定；改这个值属于口径变更，改前先看对比器读数。 +UPSIDE_NEG_TOLERANCE = float(os.environ.get("UPSIDE_NEG_TOLERANCE", "0")) + # --- 统一任务调度平台触发（08-03；键名与决策系统保持一致，便于平台侧统一配置）---- # 空串 = /api/v1/xxl/* 整组端点禁用（安全默认）。回调契约见 xxl.py 模块说明。 XXL_TRIGGER_KEY = os.environ.get("XXL_TRIGGER_KEY", "") @@ -155,16 +164,18 @@ XXL_CALLBACK_MAX_RETRIES = int(os.environ.get("XXL_CALLBACK_MAX_RETRIES", "3")) # --- 选股计划入池（08-03 定稿；规则与流程见 docs/选股计划入池_对接说明.md）------ # 写 Mongo stock_groups 的独立分组，决策系统每晚扫描按分组并集覆盖 → 候选票自动 -# 获得夜间推理。入池范围与 PMS 候选同口径：强传导主榜前 POOL_TOP 只 + 当前持仓。 -# 注意：POOL_TOP / POOL_TIERS 若调整，记得与 PMS 页面的 PMS_PLAN_TOP_N / -# PMS_PLAN_TIERS 保持一致，两边看到的候选才是同一批。 +# 获得夜间推理。入池范围与 PMS 候选**次序口径**一致：先筛强传导、再按绝对得分截断 +# （2026-08-17 拍板：分散不由选层做，主题限额默认关，组合分散归 PMS 的行业闸）。 +# 池深允许比 PMS 候选浅——POOL_TOP 直接决定决策系统每晚推理量（每只两三分钟）， +# 想跟 PMS_PLAN_TOP_N 拉平就同步调大 POOL_MAX 并接受夜扫时长变长。 POOL_GROUP_CODE = os.environ.get("POOL_GROUP_CODE", "AKG_PLAN") POOL_GROUP_NAME = os.environ.get("POOL_GROUP_NAME", "AKG每日选股计划池") POOL_ORG_ID = os.environ.get("POOL_ORG_ID", "489281497140") POOL_COLLECTION = os.environ.get("POOL_COLLECTION", "stock_groups") POOL_RECYCLE_COLLECTION = os.environ.get("POOL_RECYCLE_COLLECTION", "stock_recycle_bin") POOL_TOP = int(os.environ.get("POOL_TOP", "20")) -POOL_THEME_CAP = int(os.environ.get("POOL_THEME_CAP", "5")) +# 入池主题限额。默认 0 = 不限（2026-08-17：分散不由选层体现）；配非零值即应急回退旧行为。 +POOL_THEME_CAP = int(os.environ.get("POOL_THEME_CAP", "0")) # 档位白名单（逗号分隔；空串=主榜全部）。默认只收强传导——候选宁缺毋滥。 POOL_TIERS = {s.strip() for s in os.environ.get("POOL_TIERS", "强传导").split(",") if s.strip()} # 池子上限：计划+持仓+留池观察合计超过它时，从留池观察里清最久没上榜的。 diff --git a/factors.py b/factors.py index 45ff2dc..c913c23 100644 --- a/factors.py +++ b/factors.py @@ -394,17 +394,19 @@ def _track_set(): def _gate_of(p: pd.DataFrame, track_set, risk_set=None) -> pd.Series: """三档（07-30 拍板）： - 2 = 主榜：有业绩锚（券商覆盖且 upside>=0）；C 闸开启后再交赛道成员； + 2 = 主榜：有业绩锚（券商覆盖且 upside 不低于负容忍线）；C 闸开启后再交赛道成员； 1 = 观察档：无业绩锚但有产业链锚（当前=在传导链上；C 转正后并入赛道成员）； - 0 = 不采纳：两锚皆无，有覆盖但 upside<0（贵了不买是绝对下限）， - 或命中重大风险闸（ST/退市族，07-31 拍板默认杜绝）。""" + 0 = 不采纳：两锚皆无，有覆盖但 upside 低于负容忍线（贵了不买是绝对下限）， + 或命中重大风险闸（ST/退市族，07-31 拍板默认杜绝）。 + 负容忍线 = -config.UPSIDE_NEG_TOLERANCE（2026-08-17 加，默认 0 = 原判据逐字不变； + 放开后轻微为负的传导票进主榜、由组内分自然降权，见 config 里那段说明）。""" covered = p["upside"].notna() chain = p["transmission"].fillna(0.0) > 0 if track_set: in_track = pd.Series(p.index.isin(track_set), index=p.index) chain = chain | in_track g = pd.Series(0.0, index=p.index) - main = covered & (p["upside"] >= 0.0) + main = covered & (p["upside"] >= -float(config.UPSIDE_NEG_TOLERANCE or 0.0)) if track_set: main = main & in_track g[main] = 2.0 diff --git a/plan_reconcile.py b/plan_reconcile.py index 633a118..6c0e0d6 100644 --- a/plan_reconcile.py +++ b/plan_reconcile.py @@ -170,10 +170,13 @@ def explain(k: str, L: dict) -> None: print(f" ▶ 判决：观察档第 {L['rank_obs'].get(k)}/{len(L['obs'])} 名（score={sc:.1f}，" f"无券商覆盖、低置信）") elif g == 0.0: + import config as _cfg + _tol = float(getattr(_cfg, "UPSIDE_NEG_TOLERANCE", 0.0) or 0.0) if risk: reason = "重大风险闸（ST/退市族），07-31 拍板默认杜绝" - elif up is not None and up < 0: - reason = f"有覆盖但 upside={up:+.1%} < 0——贵了不买是绝对下限" + elif up is not None and up < -_tol: + reason = (f"有覆盖但 upside={up:+.1%} < {-_tol:+.0%}——贵了不买是绝对下限" + + ("" if _tol == 0 else f"（负容忍 {_tol:.0%} 已启用仍不够）")) elif up is not None and hits: reason = "覆盖、upside、赛道三者都过——按机制不该是 0，把本行发我核（疑似口径错位）" elif up is not None: diff --git a/pool.py b/pool.py index 7f6483b..a3d97af 100644 --- a/pool.py +++ b/pool.py @@ -232,11 +232,19 @@ def _kick_bionic_scan() -> str: # ============================================================================ def push(date: str | None = None, top: int | None = None, dry_run: bool = False, kick: bool = True) -> dict: - # 1. 当日计划（与 /plan 同一段装配代码，口径天然一致），按档位白名单过滤 + # 1. 当日计划（与 /plan 同一段装配代码）→ 先筛档位、再按绝对得分截断。 + # 次序要紧（2026-08-17 修）：原来是"全档位取前 top、再筛强传导"——排进 top 的 + # 弱/无传导票把 top 之外的强传导票挤掉了，与 PMS 候选的"先筛强传导再截断"分叉， + # 两边看到的候选不是同一批。所以这里向 collect 要一个宽池子（top 的十倍、至少 + # 二百，全在打分池规模之内），先过档位白名单，最后取前 top。 + # 主题限额默认 0（config.POOL_THEME_CAP，2026-08-17 拍板：分散不由选层做， + # 组合分散归 PMS 行业闸；配非零值即应急回退）。 top = top or config.POOL_TOP - data = plan.collect(date, top=top, obs_top=0, theme_cap=config.POOL_THEME_CAP) + data = plan.collect(date, top=max(top * 10, 200), obs_top=0, + theme_cap=config.POOL_THEME_CAP) tiers = config.POOL_TIERS - plan_rows = [r for r in data["main"] if not tiers or r.get("tier") in tiers] + plan_rows = [r for r in data["main"] + if not tiers or r.get("tier") in tiers][:top] plan_codes = [r["code"] for r in plan_rows] ds = data["date"] diff --git a/score_lab.py b/score_lab.py new file mode 100644 index 0000000..8eb4a98 --- /dev/null +++ b/score_lab.py @@ -0,0 +1,203 @@ +"""得分口径对比器（score_lab，纯只读）——三套排序口径对未来收益，用数据收口权重之争。 + +背景（2026-08-17 用户拍板）：选层撤掉主题限额之后，排序完全由总分决定， +"哪种加权更合理"不再靠感觉定。本工具对每个历史档位日，用三套口径各生成 +一份主榜前 N 名单，对齐之后 5、10、20 个交易日的真实收益，输出逐日明细与汇总。 + +三套口径： + A 现行词典序 t_factor_akg_score 降序（先档后分已编码在分值里，强弱档按当日中位切） + B 绝对档界 同样先档后分，但强弱传导改为绝对判据：传导分 >= strong_min 记强档， + 0 < 传导分 < strong_min 记弱档（组内分沿用 A 里已编码的那份，只重排档） + C 连续加权 不分档：w1·z(log1p 传导) + w2·z(upside) + w3·z(−热度)，仅主榜票参赛 +基准：主榜（gate=2）全体等权——选层不做任何排序时的底线，跑不赢它的口径直接出局。 + +读数怎么看：mean 是名单等权买入持有 k 个交易日的平均收益，hit 是上涨占比， +excess 是相对基准的超额。样本告知：传导史 2026-07 起、每天只多一个样本点， +头一两个月只看方向、不下死结论。 + +跑法【桥机 factorevaluation · ~/akg-factor-bridge】： + docker compose exec -T akg-factor-bridge python score_lab.py + docker compose exec -T akg-factor-bridge python score_lab.py --top 30 --horizons 5,10,20 \ + --strong-min 1.0 --weights 0.5,0.3,0.2 +产出：data/score_lab/对比_<起>_<止>.csv（逐日×口径×期限一行）+ 终端汇总表。 +纯只读：因子表与行情表全部 SELECT，不写任何库；产物只落容器内 data/ 目录。 +""" +from __future__ import annotations + +import argparse +import os + +import numpy as np +import pandas as pd + +import common +import db +import factors + + +def _factor_map(table: str, ds: str) -> pd.Series: + df = db.read_mysql("factor", f"SELECT stock_code, factor_value FROM {table} " + f"WHERE trade_date=%s", (ds,)) + if df.empty: + return pd.Series(dtype=float) + return df.set_index("stock_code")["factor_value"].astype(float) + + +def _score_dates(start=None, end=None) -> list[str]: + df = db.read_mysql("factor", "SELECT DISTINCT trade_date FROM t_factor_akg_score " + "ORDER BY trade_date") + out = [pd.Timestamp(x).date().isoformat() for x in df["trade_date"]] + if start: + out = [d for d in out if d >= start] + if end: + out = [d for d in out if d <= end] + return out + + +def _price_panel(start: str, end_plus: str): + """(交易日历, {(date, code): close})。行情取到 end 之后一段，好算前瞻收益。""" + px = factors._read_gp_price(start, end_plus) # noqa: SLF001 —— 同仓自用, 口径不分叉 + if px.empty: + raise SystemExit("gp_day_data 在该区间没有行情——先确认行情库连通。") + px = px.dropna(subset=["close"]) + px["k"] = px["ts_code"].map(lambda s: common.to_prefix(str(s).strip().upper())) + px["d"] = px["trade_date"].map(lambda x: pd.Timestamp(x).date().isoformat()) + cal = sorted(px["d"].unique()) + close = {(r.d, r.k): float(r.close) for r in px.itertuples()} + return cal, close + + +def _tier_from_score(sc: float) -> float: + """A 口径分值里编码的档位（200+档×20+组内分, 组内分夹 ±9.9）。""" + return float(max(0, min(2, int((sc - 190.0) // 20)))) + + +def _rank_a(score: pd.Series) -> list[str]: + main = score[score >= 150.0].sort_values(ascending=False) + return list(main.index) + + +def _rank_b(score: pd.Series, trans: pd.Series, strong_min: float) -> list[str]: + """绝对档界：组内分沿用 A（从分值反解），档位按传导分绝对阈值重记。""" + main = score[score >= 150.0] + if main.empty: + return [] + rows = [] + for k, sc in main.items(): + inner = float(sc) - (200.0 + _tier_from_score(float(sc)) * 20.0) + st = float(trans.get(k) or 0.0) + tier = 2.0 if st >= strong_min else (1.0 if st > 0 else 0.0) + rows.append((k, 200.0 + tier * 20.0 + inner)) + rows.sort(key=lambda x: -x[1]) + return [k for k, _ in rows] + + +def _rank_c(score: pd.Series, trans: pd.Series, upside: pd.Series, heat: pd.Series, + w: tuple) -> list[str]: + """连续加权：仅主榜票参赛；缺失处置同 build_score（传导缺=0，热度缺=中位数）。""" + main = score[score >= 150.0] + if main.empty: + return [] + idx = main.index + t = np.log1p(pd.Series({k: float(trans.get(k) or 0.0) for k in idx})) + u = pd.Series({k: upside.get(k) for k in idx}, dtype=float) + h = pd.Series({k: heat.get(k) for k in idx}, dtype=float) + h = h.fillna(h.median()) + u = u.fillna(0.0) + z = factors._robust_z # noqa: SLF001 —— 与生产打分同一把尺子 + comp = w[0] * z(t) + w[1] * z(u) + w[2] * (-z(h)) + return list(comp.sort_values(ascending=False).index) + + +def _fwd(cal: list, close: dict, d: str, codes: list, k: int): + """名单在 d 买入、持有 k 个交易日的逐票收益（两头都要有价，缺价的票跳过）。""" + try: + i = cal.index(d) + except ValueError: + return [] + if i + k >= len(cal): + return [] + dt = cal[i + k] + out = [] + for c in codes: + p0, p1 = close.get((d, c)), close.get((dt, c)) + if p0 and p1 and p0 > 0: + out.append(p1 / p0 - 1.0) + return out + + +def main() -> int: + ap = argparse.ArgumentParser(description="得分口径对比器（只读）") + ap.add_argument("--top", type=int, default=30, help="每套口径取主榜前几只（默认 30）") + ap.add_argument("--horizons", default="5,10,20", help="前瞻交易日数，逗号分隔") + ap.add_argument("--strong-min", type=float, default=1.0, + help="B 口径的强传导绝对阈值（传导分 = 源数×(1−已动比例)）") + ap.add_argument("--weights", default="0.5,0.3,0.2", + help="C 口径权重 传导,upside,−热度") + ap.add_argument("--start") + ap.add_argument("--end") + a = ap.parse_args() + horizons = sorted({int(x) for x in a.horizons.split(",") if x.strip()}) + w = tuple(float(x) for x in a.weights.split(",")) + if len(w) != 3: + raise SystemExit("--weights 需要三个数，如 0.5,0.3,0.2") + + dates = _score_dates(a.start, a.end) + if not dates: + raise SystemExit("t_factor_akg_score 在该区间没有数据。") + end_plus = (pd.Timestamp(dates[-1]) + pd.Timedelta(days=int(max(horizons) * 2.2 + 14)) + ).date().isoformat() + print(f"档位日 {dates[0]} ~ {dates[-1]} 共 {len(dates)} 天；" + f"前瞻 {horizons} 交易日；行情取到 {end_plus}") + cal, close = _price_panel(dates[0], end_plus) + + rows = [] + for ds in dates: + score = _factor_map("t_factor_akg_score", ds) + if score.empty: + continue + trans = _factor_map("t_factor_akg_transmission", ds) + upside = _factor_map("t_factor_akg_upside", ds) + hd = factors._heat_day_for(ds) # noqa: SLF001 + heat = _factor_map("t_factor_akg_heat", hd) if hd else pd.Series(dtype=float) + + base_codes = list(score[score >= 150.0].index) + lists = {"A现行": _rank_a(score)[:a.top], + "B绝对档界": _rank_b(score, trans, a.strong_min)[:a.top], + "C连续加权": _rank_c(score, trans, upside, heat, w)[:a.top], + "基准主榜等权": base_codes} + for name, codes in lists.items(): + for k in horizons: + rets = _fwd(cal, close, ds, codes, k) + if not rets: + continue + rows.append({"date": ds, "scheme": name, "horizon": k, + "n": len(rets), "mean": float(np.mean(rets)), + "hit": float(np.mean([1.0 if r > 0 else 0.0 for r in rets])), + "median": float(np.median(rets))}) + + if not rows: + raise SystemExit("没有任何可评样本——多半是前瞻天数超出了已有行情（等几天再跑）。") + df = pd.DataFrame(rows) + os.makedirs("data/score_lab", exist_ok=True) + out = f"data/score_lab/对比_{dates[0]}_{dates[-1]}.csv" + df.to_csv(out, index=False, encoding="utf-8-sig") + + print(f"\n—— 汇总（逐日等权平均；excess = 相对基准主榜等权）——") + print(f"{'口径':<10}{'期限':>4}{'样本日':>5}{'均值':>9}{'胜率':>7}{'超额':>9}") + agg = df.groupby(["scheme", "horizon"]).agg(days=("date", "nunique"), + mean=("mean", "mean"), + hit=("hit", "mean")).reset_index() + base = {(r.horizon): r.mean for r in agg[agg["scheme"] == "基准主榜等权"].itertuples()} + for r in agg.itertuples(): + ex = r.mean - base.get(r.horizon, 0.0) + print(f"{r.scheme:<10}{r.horizon:>4}{r.days:>5}{r.mean:>9.2%}{r.hit:>7.1%}" + + (f"{ex:>9.2%}" if r.scheme != "基准主榜等权" else f"{'—':>9}")) + n_days = df["date"].nunique() + print(f"\n已写 {out}（{len(df)} 行）。样本 {n_days} 天" + + ("——样本很小, 只看方向、别下死结论。" if n_days < 40 else "。")) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())