tradingSystem/scripts/probe_plan_api.py

# -*- coding: utf-8 -*-
"""
上游选股计划接口实机探活 (只读, 不写任何表)
============================================
    docker compose run --rm --no-deps pms-web python scripts/probe_plan_api.py
    docker compose run --rm --no-deps pms-web python scripts/probe_plan_api.py --date 2026-07-29
    docker compose run --rm --no-deps pms-web python scripts/probe_plan_api.py --top 30 --with-price

干什么: 拿真实应答验证四件事 —— ① 接口通不通、② 字段口径与单测 fixture 是否一致、
③ 按当前参数筛出来的候选池长什么样、④ (--with-price) 这些票行情里到底有没有价。
第 ④ 项最容易翻车: 计划不带价格, 价格取不到的票会在候选池里被静默剔除。

只读: 不会 upsert pms_industry_map (那由 /api/ops/plan-refresh 或盘前调度做), 也不下单。
"""
from __future__ import annotations

import argparse
import json
import os
import sys

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

from app.services import plan_feed as pf                                       # noqa: E402


def _fmt(v, n=4):
    return "-" if v is None else (f"{v:.{n}f}" if isinstance(v, float) else str(v))


def _pct(v):
    """upside 是相对现价的比例: 2.1203 → +212% (与上游 format=md 的「预期空间」同口径)。"""
    return "-" if v is None else f"{v * 100:+.0f}%"


def _default_gateway() -> str:
    """容器默认网关 = 宿主机在网桥上的地址。不依赖 ip/route 命令, 直接读 /proc。"""
    try:
        with open("/proc/net/route", encoding="utf-8") as f:
            for line in f.readlines()[1:]:
                p = line.split()
                if len(p) > 2 and p[1] == "00000000" and p[2] != "00000000":
                    h = p[2]
                    return ".".join(str(int(h[i:i + 2], 16)) for i in (6, 4, 2, 0))
    except Exception:
        pass
    return ""


def _try_bases(ps, args) -> int:
    """逐个试候选地址。容器 → 宿主机这条路最容易断, 这里一次把可能的走法全试掉。

    ConnectTimeout  = 包被丢了 (宿主机 INPUT 链拦了网桥进来的流量), 不是没人监听
    ConnectionError = 端口没人听 (地址对了但服务不在这台/没起)
    """
    port = "8300"
    cfg = (ps.get("PMS_PLAN_API_BASE", "") or "").strip().rstrip("/")
    gw = _default_gateway()
    cands = []
    for b in ([args.base] if args.base else []) + [
            cfg, f"http://host.docker.internal:{port}",
            (f"http://{gw}:{port}" if gw else ""), f"http://172.17.0.1:{port}",
            f"http://192.168.16.155:{port}", f"http://127.0.0.1:{port}"]:
        b = (b or "").strip().rstrip("/")
        if b and b not in cands:
            cands.append(b)
    print(f"[探路] 容器默认网关 = {gw or '(读不到)'}; 当前配置 = {cfg or '(空)'}")
    print(f"       共 {len(cands)} 个候选, 每个 3 秒超时")
    ok = []
    for b in cands:
        tag = "  当前配置" if b == cfg else ""
        try:
            plan = pf.fetch(base=b, path=ps.get("PMS_PLAN_API_PATH", "/plan"), timeout=3)
            print(f"  通    {b}{tag}   → 计划 {plan['date']}, 主榜 {plan['returned']['main']} 条")
            ok.append(b)
        except pf.PlanFeedError as e:
            msg = str(e).split(": ", 1)[-1]
            print(f"  不通  {b}{tag}   → {msg[:150]}")
    print("-" * 62)
    if not ok:
        print("全都不通。按下面顺序排:")
        print("  1. 宿主机上确认在听:  ss -ltnp | grep 8300")
        print("  2. 若上面显示 127.0.0.1:8300 —— 服务只绑了 loopback, 容器永远进不来,")
        print("     要么让它绑 0.0.0.0, 要么把 PMS 容器改 network_mode: host (不推荐)。")
        print("  3. 若显示 0.0.0.0:8300 而这里仍超时 —— 宿主机 INPUT 链拦了网桥流量, 放行:")
        print("       sudo firewall-cmd --permanent --zone=trusted --add-source=172.16.0.0/12")
        print("       sudo firewall-cmd --reload")
        print("     (或 iptables: sudo iptables -I INPUT -s 172.16.0.0/12 -p tcp --dport 8300 -j ACCEPT)")
        print("  4. 若计划服务本身也在 docker 里 —— 最干净的是把它和 pms-* 放同一个 network,")
        print("     base 直接填 http://<那个容器名>:8300, 不绕宿主机。")
        return 1
    print(f"能通的地址: {ok[0]}")
    print(f"→ 页面「参数设置」把 PMS_PLAN_API_BASE 改成它, 或先跑:")
    print(f"   curl -s -X POST http://127.0.0.1:38100/api/params -H 'Content-Type: application/json' \\")
    print(f"     -d '{{\"key\":\"PMS_PLAN_API_BASE\",\"value\":\"{ok[0]}\"}}'")
    return 0


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--date", default=None, help="指定计划日期 YYYY-MM-DD (缺省=上游最新)")
    ap.add_argument("--top", type=int, default=20, help="打印主榜前 N 行")
    ap.add_argument("--observe", action="store_true", help="连观察档一起打印")
    ap.add_argument("--with-price", action="store_true",
                    help="逐只查行情价 (需 Redis 通; 用来看候选池会不会因无价掉票)")
    ap.add_argument("--base", default=None,
                    help="临时覆盖 PMS_PLAN_API_BASE (只影响本次, 不落库)")
    ap.add_argument("--try-bases", action="store_true",
                    help="逐个试候选地址, 报告哪个能通 —— 容器连不上宿主机时先跑这个")
    ap.add_argument("--json", action="store_true", help="原样打印解析后的结构")
    args = ap.parse_args()

    from app.services import param_store as ps
    if args.try_bases:
        return _try_bases(ps, args)
    base = (args.base or ps.get("PMS_PLAN_API_BASE", "") or "").strip()
    print(f"[1] 接口配置  base={base or '(空 —— 候选池将恒为空)'} "
          f"path={ps.get('PMS_PLAN_API_PATH', '/plan')} "
          f"timeout={ps.get_int('PMS_PLAN_TIMEOUT', 10)}s")
    if not base:
        print("    → PMS_PLAN_API_BASE 为空。页面「参数设置」填上再跑。")
        return 2

    try:
        plan = pf.fetch(date=args.date, base=base)
    except pf.PlanFeedError as e:
        print(f"[2] 取数失败: {e}")
        return 1
    print(f"[2] 取数成功  {plan['url']}")
    print(f"    date={plan['date']}  heat_date={plan['heat_date']}  "
          f"snapshot={plan['market_snapshot_days']}  theme_cap={plan['theme_cap']}")
    print(f"    counts(上游全量)={plan['counts']}   returned(本次应答)={plan['returned']}")
    if plan["counts"]["main"] is not None and plan["returned"]["main"] != plan["counts"]["main"]:
        print(f"    ! 主榜 counts={plan['counts']['main']} 但只回了 "
              f"{plan['returned']['main']} 条 —— 上游做了截断, 确认是否分页")

    age = pf.plan_age_tdays(plan["date"])
    limit = ps.get_int("PMS_PLAN_STALE_TDAYS", 1)
    verdict = "新鲜" if age <= limit else f"**过期** (上限 {limit}, 候选池会拒用)"
    print(f"[3] 新鲜度    日龄 {age} 个交易日 → {verdict}")

    print(f"[4] 主榜前 {args.top} (score 降序即上游 rank 序)")
    hdr = f"    {'rank':>4} {'代码':<11} {'名称':<8} {'score':>8} {'档位':<6} {'主题':<10} " \
          f"{'heat':>7} {'预期空间':>9}"
    print(hdr + ("   " + "行情价" if args.with_price else ""))
    rows = plan["main"][:max(1, args.top)]
    if args.observe:
        rows += plan["observe"][:max(1, args.top)]
    prices = {}
    if args.with_price:
        from app.services import market
        for r in rows:
            try:
                prices[r["ts_code"]] = market.get_price(r["ts_code"])
            except Exception as e:
                prices[r["ts_code"]] = f"ERR {type(e).__name__}"
    for r in rows:
        line = (f"    {str(r['rank']):>4} {r['ts_code']:<11} {(r['name'] or '-'):<8} "
                f"{_fmt(r['score'], 2):>8} {(r['tier'] or '观察'):<6} "
                f"{(r['theme'] or '-'):<10} {_fmt(r['heat']):>7} {_pct(r['upside']):>9}")
        if args.with_price:
            p = prices.get(r["ts_code"])
            line += f"   {p if p else '**无价(会被剔除)**'}"
        print(line)

    themes = {}
    for r in plan["main"]:
        themes[r["theme"] or "(无)"] = themes.get(r["theme"] or "(无)", 0) + 1
    top_themes = sorted(themes.items(), key=lambda x: -x[1])[:12]
    print(f"[5] 主榜主题分布 (共 {len(themes)} 个主题): " +
          ", ".join(f"{k}×{v}" for k, v in top_themes))

    sel = pf.select_candidates(
        plan, top_n=ps.get_int("PMS_PLAN_TOP_N", 30), tiers=ps.get_list("PMS_PLAN_TIERS", []),
        include_observe=ps.get_bool("PMS_PLAN_INCLUDE_OBSERVE", False),
        min_score=(ps.get_float("PMS_PLAN_MIN_SCORE", 0.0) or None),
        min_sources=ps.get_int("PMS_PLAN_MIN_SOURCES", 0))
    print(f"[6] 按当前参数筛选 (top_n={ps.get_int('PMS_PLAN_TOP_N', 30)} "
          f"tiers={ps.get_list('PMS_PLAN_TIERS', [])} "
          f"observe={ps.get_bool('PMS_PLAN_INCLUDE_OBSERVE', False)})")
    print(f"    排序池 {sel['considered']} → 合格 {sel['eligible']} → 取 {len(sel['items'])} 只")
    print(f"    丢弃明细 {sel['dropped']}   (此处未扣持仓/黑名单, 下命令时还会再扣)")
    print("    " + ", ".join(x["ts_code"] for x in sel["items"]))

    if args.json:
        print("[7] 解析结构")
        print(json.dumps({k: v for k, v in plan.items() if k not in ("main", "observe")},
                         ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())