tradingSystem/scripts/probe_strategy_signals.py

# -*- coding: utf-8 -*-
"""
策略自动挂载 · 步骤一只读探测 (STRATEGY_AUTO_ATTACH_PLAN.md 第九节)
====================================================================
只读, 不写任何表, 随时可跑。回答五个问题:

  1. 吸筹定性的最新日分布 (与数据底座前端清单对读, 口径同源自证)
  2. 每只持仓票: 定性 / 结论日龄 / 热度分 / 按 V1 规则的预判去向
  3. 候选池合格票与「明确吸筹」的重合率 (拍板②的实测依据)
  4. 热度分的分位数与超阈值持仓票 (热度阈值 PMS_AUTO_HEAT_TH 的标定依据)
  5. 两个信号源的新鲜度

运行 (桥机 factorevaluation, 新文件要先重建镜像):
    docker compose run --rm --no-deps pms-web python scripts/probe_strategy_signals.py

数据源 (都经 153 代理, 严格单表):
    strategy_daily_results.raw_logic_json 的 fund_flow{score,state,pos_tag}  —— 吸筹定性
    stock_fund_heat_scores(trade_date, batch_no, stock_code, score)          —— 热度分
"""
import json
import os
import sys
from datetime import datetime, timedelta

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.db.session import fetch_all, fetch_one          # noqa: E402
from app.repo import pms_repo                             # noqa: E402
from app.repo.downstream_repo import to_dot, to_prefix    # noqa: E402
from app.services import param_store                      # noqa: E402

# 决策系统定性词表 (契约; 见方案「口径同源声明」)。不在表里的词一律当无标志。
# v1.1 (2026-08-25): 归类从前缀匹配改成**子串包含** —— 首跑 62 只「词表外」的根因是
# state 带前后缀修饰 (如 "→ 高位派发风险"), 数据底座 feed.py 的 _ACCUM_KEEP 用的
# 正是 any(k in state)。次序即优先级, 派发最先 (两词同现按保守方向), 与 strategy_advisor
# 的 classify_accum 完全同法 —— 探测读数和线上判定必须是同一双眼睛。
FIVE_STATES = ("明确吸筹", "潜在吸筹", "无吸筹迹象", "高位派发", "信号不明")
_CLS_ORDER = (("派发", "高位派发"), ("明确吸筹", "明确吸筹"), ("潜在吸筹", "潜在吸筹"),
              ("无吸筹迹象", "无吸筹迹象"), ("不明", "信号不明"))
ACCUM_WINDOW_DAYS = 45          # 每票取近 45 自然日内最新一条 (方案口径是 30 天, 取宽探测)


def _ymd_int(v):
    """trade_date 可能是 int / str / date, 统一成 int YYYYMMDD; 解析不了返回 None。"""
    if v is None:
        return None
    if hasattr(v, "strftime"):
        return int(v.strftime("%Y%m%d"))
    s = str(v).strip().replace("-", "")[:8]
    return int(s) if s.isdigit() and len(s) == 8 else None


def _age_days(ymd):
    if not ymd:
        return None
    try:
        d = datetime.strptime(str(ymd), "%Y%m%d").date()
        return (datetime.now().date() - d).days
    except ValueError:
        return None


def _fund_flow(raw):
    """raw_logic_json → fund_flow dict 或 None。脏 JSON 不炸。"""
    try:
        d = raw if isinstance(raw, dict) else json.loads(raw or "{}")
        ff = d.get("fund_flow")
        return ff if isinstance(ff, dict) and ff.get("state") else None
    except (ValueError, TypeError):
        return None


def _classify(state):
    s = str(state or "").strip()
    if not s:
        return "无字段"
    for key, cls in _CLS_ORDER:
        if key in s:
            return cls
    return "词表外"


def _variants(dot_codes):
    """{每种写法: 点式} 的反查表 (前缀式 / 点式 / 纯数字)。"""
    rev = {}
    for c in dot_codes:
        d = to_dot(c)
        if not d:
            continue
        rev[d] = d
        rev[to_prefix(d)] = d
        rev[d.split(".")[0]] = d
    return rev


def _in_clause(values, prefix, params):
    keys = []
    for i, v in enumerate(values):
        keys.append(f":{prefix}{i}")
        params[f"{prefix}{i}"] = v
    return ", ".join(keys)


# ================================================================ 取数
def latest_accum_day():
    r = fetch_one("SELECT MAX(trade_date) AS td FROM strategy_daily_results")
    return (r or {}).get("td")


def accum_distribution(td):
    """最新一日的定性分布 (全池)。返回 (总数, 分布, 词表外原文样本)。"""
    rows = fetch_all("SELECT stock_code, raw_logic_json FROM strategy_daily_results "
                     "WHERE trade_date = :td", {"td": td})
    dist, total, oov = {}, 0, {}
    for r in rows:
        total += 1
        ff = _fund_flow(r.get("raw_logic_json"))
        state = (ff or {}).get("state")
        cls = _classify(state)
        dist[cls] = dist.get(cls, 0) + 1
        if cls == "词表外":
            oov[str(state)] = oov.get(str(state), 0) + 1
    samples = sorted(oov.items(), key=lambda kv: -kv[1])[:10]
    return total, dist, samples


def accum_of(dot_codes):
    """{点式: {state, cls, score, ymd, age}} —— 窗口内每票最新一条。"""
    if not dot_codes:
        return {}
    rev = _variants(dot_codes)
    since = int((datetime.now().date() - timedelta(days=ACCUM_WINDOW_DAYS)).strftime("%Y%m%d"))
    p = {"since": since}
    sql = ("SELECT stock_code, trade_date, raw_logic_json FROM strategy_daily_results "
           f"WHERE trade_date >= :since AND stock_code IN ({_in_clause(list(rev), 'c', p)})")
    best = {}
    for r in fetch_all(sql, p):
        dot = rev.get(str(r.get("stock_code") or "").strip())
        ymd = _ymd_int(r.get("trade_date"))
        if not dot or not ymd:
            continue
        if dot in best and best[dot]["ymd"] >= ymd:
            continue
        ff = _fund_flow(r.get("raw_logic_json"))
        best[dot] = {"ymd": ymd, "age": _age_days(ymd),
                     "state": (ff or {}).get("state"), "cls": _classify((ff or {}).get("state")),
                     "score": (ff or {}).get("score"), "pos_tag": (ff or {}).get("pos_tag")}
    return best


def heat_snapshot():
    """(交易日, 批次, {点式或原码: score}, 全市场分数列表)。"""
    r = fetch_one("SELECT MAX(trade_date) AS td FROM stock_fund_heat_scores")
    td = (r or {}).get("td")
    if td is None:
        return None, None, {}, []
    r2 = fetch_one("SELECT MAX(batch_no) AS b FROM stock_fund_heat_scores "
                   "WHERE trade_date = :td", {"td": td})
    b = (r2 or {}).get("b")
    rows = fetch_all("SELECT stock_code, score FROM stock_fund_heat_scores "
                     "WHERE trade_date = :td AND batch_no = :b LIMIT 20000",
                     {"td": td, "b": b})
    scores, all_scores = {}, []
    for x in rows:
        try:
            v = float(x.get("score"))
        except (TypeError, ValueError):
            continue
        scores[str(x.get("stock_code") or "").strip()] = v
        all_scores.append(v)
    return td, b, scores, sorted(all_scores)


def _pct(sorted_vals, q):
    if not sorted_vals:
        return None
    i = min(len(sorted_vals) - 1, max(0, int(round(q * (len(sorted_vals) - 1)))))
    return sorted_vals[i]


# ================================================================ 主流程
def main():
    heat_th = param_store.get_float("PMS_AUTO_HEAT_TH", 0.80)
    print("=" * 66)
    print("策略自动挂载 · 只读探测   %s" % datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    print("(只读, 不写任何表; 各段独立, 一段失败不影响其余)")
    print("=" * 66)

    # ---- 一、吸筹定性最新日分布 ----
    print("\n【一】吸筹定性最新日分布 (与数据底座前端清单对读)")
    accum_td = None
    try:
        accum_td = latest_accum_day()
        if accum_td is None:
            print("  结论表里一行都没有 —— 先确认决策系统夜间链在跑")
        else:
            total, dist, oov_samples = accum_distribution(accum_td)
            age = _age_days(_ymd_int(accum_td))
            print(f"  最新结论日 {accum_td} (距今 {age} 个自然日), 当日共 {total} 只")
            for k in list(FIVE_STATES) + ["词表外", "无字段"]:
                if dist.get(k):
                    print(f"    {k:<6} {dist[k]} 只")
            if dist.get("词表外"):
                print("  ⚠ 仍有子串也认不出的定性 —— 五档契约可能真漂了, 拿下面原文与决策系统核对:")
                for s, n in oov_samples:
                    print(f"      「{s}」 × {n}")
            print("  对读口径: 本段是**单日**分布; 数据底座前端清单是 30 天窗口每票最新一条,"
                  " 总数天然更大, 两边比的是档位结构而不是绝对数")
    except Exception as e:
        print(f"  ✗ 本段失败: {type(e).__name__}: {e}")

    # ---- 二、热度分布 ----
    print(f"\n【二】热度分布 (阈值初值 {heat_th}, 用本段读数校准)")
    heat_map, heat_rev = {}, {}
    try:
        htd, hb, heat_map, all_scores = heat_snapshot()
        if htd is None:
            print("  热度表里一行都没有")
        else:
            print(f"  最新交易日 {htd} 批次 {hb}, 覆盖 {len(all_scores)} 只")
            print(f"  分位数: 一半位 {_pct(all_scores, 0.50)} · 八成位 {_pct(all_scores, 0.80)}"
                  f" · 九成位 {_pct(all_scores, 0.90)} · 最高 {all_scores[-1] if all_scores else '—'}")
            n_over = sum(1 for v in all_scores if v >= heat_th)
            print(f"  全市场不低于 {heat_th} 的共 {n_over} 只")
    except Exception as e:
        print(f"  ✗ 本段失败: {type(e).__name__}: {e}")

    # ---- 三、持仓票逐只 ----
    print("\n【三】持仓票逐只 (定性 / 日龄 / 热度 / 按 V1 规则的预判去向)")
    held_dots = []
    try:
        positions = pms_repo.list_positions(only_open=True)
        held_dots = [p["ts_code"] for p in positions]
        if not positions:
            print("  当前无持仓 —— 边一边二暂无对象, 只剩建仓插队那半边")
        else:
            acc = accum_of(held_dots)
            heat_rev = _variants(held_dots)
            held_heat = {}
            for raw, v in heat_map.items():
                dot = heat_rev.get(raw)
                if dot:
                    held_heat[dot] = v
            try:
                strat_codes = {s.get("ts_code") for s in
                               pms_repo.list_strategies(statuses=["ACTIVE", "PAUSED"], limit=500)}
            except Exception:
                strat_codes = set()
            stale = param_store.get_int("PMS_AUTO_ACCUM_STALE_TDAYS", 3)
            print(f"  {'代码':<10} {'定性':<8} {'结论日':<9} {'热度':<6} {'垫':<7} 预判")
            for p in positions:
                c = p["ts_code"]
                a = acc.get(c) or {}
                cls = a.get("cls") or "无结论"
                ymd = a.get("ymd") or "—"
                hv = held_heat.get(c)
                cushion = p.get("cushion_pct")
                cu = ("%.1f%%" % (float(cushion) * 100)) if cushion is not None else "未知"
                fresh = a.get("age") is not None and a["age"] <= stale * 2
                hot = hv is not None and hv >= heat_th
                pos_ok = cushion is not None and float(cushion) > 0
                if c in strat_codes:
                    verdict = "已挂策略, 自动挂载不碰它"
                elif hot and pos_ok:
                    verdict = ("双命中取止盈" if (cls == "明确吸筹" and fresh) else "挂跟踪止盈")
                elif cls == "明确吸筹" and fresh:
                    verdict = "挂网格" + ("" if (p.get("frozen_reason") or "NONE") == "NONE"
                                          else " (但该股被冻结, 实扫会跳过)")
                elif cls == "高位派发":
                    verdict = "若挂着网格则停买入腿"
                elif hot and not pos_ok:
                    verdict = "热度够但垫不正, 不挂止盈 (护补仓评估)"
                elif cls == "明确吸筹":
                    verdict = "吸筹结论超日龄, 视为无标志"
                else:
                    verdict = "不动"
                print(f"  {c:<10} {cls:<8} {str(ymd):<9} "
                      f"{('%.3f' % hv) if hv is not None else '—':<6} {cu:<7} {verdict}")
            print("  注: 预判未含黑名单与在途指令两道排除, 以实盘扫描为准")
    except Exception as e:
        print(f"  ✗ 本段失败: {type(e).__name__}: {e}")

    # ---- 四、候选池与明确吸筹的重合 ----
    print("\n【四】候选池合格票与吸筹的重合 (拍板②「资格不放宽」的实测依据)")
    try:
        from app.services import command_service, plan_feed
        try:
            black = command_service.blacklist()
        except Exception:
            black = set()
        sel = plan_feed.candidates(held=set(held_dots), black=black)
        cands = [x["ts_code"] for x in (sel.get("items") or [])]
        print(f"  候选池合格票 {len(cands)} 只 (计划日 {sel.get('date')}, "
              f"考察 {sel.get('considered')} 只)")
        if cands:
            acc_c = accum_of(cands)
            clear = [(c, acc_c[c]) for c in cands if acc_c.get(c, {}).get("cls") == "明确吸筹"]
            maybe = [c for c in cands if acc_c.get(c, {}).get("cls") == "潜在吸筹"]
            none_ = [c for c in cands if c not in acc_c]
            print(f"  其中 明确吸筹 {len(clear)} 只 · 潜在吸筹 {len(maybe)} 只 · "
                  f"无结论 {len(none_)} 只")
            for c, a in clear:
                print(f"    明确: {c}  结论日 {a.get('ymd')}  评分 {a.get('score')}")
            if not clear:
                print("  ⚠ 重合为零 —— 吸筹建仓插队这半边今天空转, 只剩持仓票挂网格那半边。"
                      "多观察几天, 持续为零再议资格")
    except Exception as e:
        print(f"  ✗ 本段失败 (计划接口没起或已过期都会走到这里): {type(e).__name__}: {e}")

    # ---- 五、新鲜度小结 ----
    print("\n【五】信号源新鲜度")
    try:
        a_age = _age_days(_ymd_int(accum_td)) if accum_td is not None else None
        print(f"  吸筹结论: 最新 {accum_td} "
              f"({'距今 %s 天' % a_age if a_age is not None else '取不到'})"
              f"{' ⚠ 超过 6 个自然日, 按方案口径会被判过期' if (a_age or 0) > 6 else ''}")
    except Exception as e:
        print(f"  ✗ {e}")
    print("\n探测结束。请把整段输出发回, 用【二】定热度阈值、用【四】看资格结论。")


if __name__ == "__main__":
    main()