333 lines
18 KiB
Python
333 lines
18 KiB
Python
# -*- 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 或盘前调度做), 也不下单。
|
||
第 ⑤ 项默认也是只读 (拿手上这份跟库里最新快照比, 不写库); **唯一的写操作**是显式加
|
||
`--snapshot` 时落一行 pms_plan_snapshot。
|
||
"""
|
||
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
|
||
|
||
|
||
LIMIT_KEYS = ["limit", "top", "topn", "top_n", "n", "size", "page_size", "per_page",
|
||
"count", "max", "rows", "num", "full"]
|
||
|
||
|
||
def _try_limit(ps, args) -> int:
|
||
"""探"取全量"的参数名。
|
||
|
||
上游默认只回主榜 20 条, 而 counts.main=961 —— 候选池实际只在这 20 条里选, 且那 20 条
|
||
已经被上游按自己的「每主题限额 5」裁过。参数名探出来填进 PMS_PLAN_QUERY_EXTRA 即可,
|
||
不用改代码。全都不认就只能等上游给分页方式 (UPSTREAM_PLAN_API.md Q3)。
|
||
"""
|
||
base = (args.base or ps.get("PMS_PLAN_API_BASE", "") or "").strip().rstrip("/")
|
||
path = ps.get("PMS_PLAN_API_PATH", "/plan")
|
||
if not base:
|
||
print("PMS_PLAN_API_BASE 为空, 先把地址配上 (或用 --base)")
|
||
return 2
|
||
try:
|
||
base_plan = pf.fetch(base=base, path=path, timeout=15, date=args.date)
|
||
except pf.PlanFeedError as e:
|
||
print(f"基准请求就失败了: {e}")
|
||
return 1
|
||
n0, c0 = base_plan["returned"]["main"], base_plan["counts"]["main"]
|
||
print(f"[基准] 不带任何参数: 主榜回 {n0} 条 (上游 counts={c0})")
|
||
if not base_plan["truncated"]["main"]:
|
||
print(" 没有截断 —— 已经是全量, 不需要分页参数。")
|
||
return 0
|
||
print(f"[探参] 依次试 {len(LIMIT_KEYS)} 个常见参数名, 取值 {args.limit_value}")
|
||
hits = []
|
||
for k in LIMIT_KEYS:
|
||
try:
|
||
p = pf.fetch(base=base, path=path, timeout=20, date=args.date,
|
||
extra_params={k: args.limit_value})
|
||
n = p["returned"]["main"]
|
||
flag = " ← 生效" if n > n0 else ""
|
||
print(f" {k:<10}= {args.limit_value:<6} → 主榜 {n} 条{flag}")
|
||
if n > n0:
|
||
hits.append((k, n))
|
||
except pf.PlanFeedError as e:
|
||
print(f" {k:<10}= {args.limit_value:<6} → 报错: {str(e).split(': ', 1)[-1][:90]}")
|
||
print("-" * 62)
|
||
if not hits:
|
||
print(f"没有一个参数名生效, 主榜始终 {n0} 条。")
|
||
print("→ 这条只能问上游: 取全量 961 条的正确方式是什么 (分页? 另一个端点?)")
|
||
print(" 见 UPSTREAM_PLAN_API.md Q3。在此之前候选池就是从这 20 条里选,")
|
||
print(" 并且要知道: 这 20 条已被上游按「每主题限额 5」裁过, PMS 侧的行业约束")
|
||
print(" 是在一个已经被裁过的池子上再裁一次。")
|
||
return 1
|
||
hits.sort(key=lambda x: -x[1])
|
||
k, n = hits[0]
|
||
print(f"生效: {k}={args.limit_value} → 主榜 {n} 条 (原 {n0})")
|
||
print("→ 页面「参数设置」把 PMS_PLAN_QUERY_EXTRA 填成下面这串, 或直接:")
|
||
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_QUERY_EXTRA\",\"value\":\"{k}={args.limit_value}\"}}'")
|
||
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("--try-limit", action="store_true",
|
||
help="探取全量的参数名 (上游默认只回主榜 20 条, counts 却是 961)")
|
||
ap.add_argument("--limit-value", default="1000", help="--try-limit 用的取值 (默认 1000)")
|
||
ap.add_argument("--snapshot", action="store_true",
|
||
help="把这份名册落进 pms_plan_snapshot (**本脚本唯一的写操作**, 默认不写)")
|
||
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)
|
||
if args.try_limit:
|
||
return _try_limit(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
|
||
|
||
q = pf.effective_query()
|
||
print(f" 发出的查询参数 {q or '(无 —— 将吃上游默认 top=20/obs_top=10/theme_cap=5)'}")
|
||
try:
|
||
plan = pf.fetch(date=args.date, base=base, extra_params=q)
|
||
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" 请求参数 {plan['requested']}")
|
||
f = plan["funnel"]
|
||
print(f" 漏斗: 打分池 主榜 {f['scored_main']} / 观察 {f['scored_observe']}"
|
||
f" →(券商预期 + 目标价≥现价 + 主题限额 + top)→"
|
||
f" 实收 主榜 {f['returned_main']} / 观察 {f['returned_observe']}")
|
||
if plan["truncated"]["main"]:
|
||
print(f" · 主榜正好吃满 top={plan['requested']['top']} —— 要不要紧看 [6] 的"
|
||
f"「白名单档位取全」那一行")
|
||
else:
|
||
print(f" 主榜没吃满 top={plan['requested']['top']}, 说明这就是上游能给的全部"
|
||
f" (受主题限额 {plan['theme_cap']} 与价格筛限制)")
|
||
|
||
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:
|
||
pp = market.plan_price(r["ts_code"])
|
||
prices[r["ts_code"]] = ((f"{pp['price']}" if pp["source"] == "realtime"
|
||
else f"{pp['price']} (昨收)") if pp["price"]
|
||
else "**无价(会被剔除)**")
|
||
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:
|
||
line += f" {prices.get(r['ts_code']) or '-'}"
|
||
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),
|
||
min_upside=(ps.get_float("PMS_PLAN_MIN_UPSIDE", 0.0) or None),
|
||
theme_cap=ps.get_int("PMS_PLAN_THEME_CAP_LOCAL", 5),
|
||
exclude_st=ps.get_bool("PMS_PLAN_EXCLUDE_ST", True))
|
||
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)} "
|
||
f"PMS侧主题限额={ps.get_int('PMS_PLAN_THEME_CAP_LOCAL', 5)} "
|
||
f"剔ST={ps.get_bool('PMS_PLAN_EXCLUDE_ST', True)})")
|
||
tc = sel.get("tier_complete")
|
||
tcs = ("是 —— 已看到白名单之外的档位, 后面再多也不影响候选池" if tc is True else
|
||
("**否** —— 返回的全是白名单档位, 说明被 top 卡住了, 调大 PMS_PLAN_TOP" if tc is False
|
||
else "不适用 (没设档位白名单)"))
|
||
print(f" 白名单档位取全: {tcs}")
|
||
print(f" 排序池 {sel['considered']} → 合格 {sel['eligible']} → 取 {len(sel['items'])} 只")
|
||
if sel.get("st_unknown"):
|
||
print(f" ! {len(sel['st_unknown'])} 只没有名称, ST 判不了已放行: {sel['st_unknown'][:8]}")
|
||
print(f" 丢弃明细 {sel['dropped']} (此处未扣持仓/黑名单, 下命令时还会再扣)")
|
||
st = {}
|
||
for x in sel["items"]:
|
||
st[x.get("theme") or "(无)"] = st.get(x.get("theme") or "(无)", 0) + 1
|
||
print(" 入池主题分布: " + ", ".join(f"{k}×{v}" for k, v in
|
||
sorted(st.items(), key=lambda y: -y[1])))
|
||
print(" " + ", ".join(x["ts_code"] for x in sel["items"]))
|
||
|
||
# --- [7] 榜单变化 (PMS 自算; 上游 changes 恒为 null, 见 UPSTREAM_PLAN_API §9)
|
||
# 默认**只读**: 拿刚取到的这份跟库里最新快照比, 一个字都不写 —— 本脚本的契约
|
||
# 是"只读探活", 破了它比少个功能糟得多。要落库请显式加 --snapshot。
|
||
print("[7] 榜单变化" + (" (落库)" if args.snapshot else " (只读, 不写库)"))
|
||
try:
|
||
# **必须先比后写。** 反过来的话 preview 会拿这份计划跟"刚写进去的它自己"比,
|
||
# 于是永远报"完全相同、无变化" —— 加个 --snapshot 就把要看的东西看没了。
|
||
ch = pf.preview_changes(plan)
|
||
if args.snapshot:
|
||
snap = pf.snapshot(plan, keep=ps.get_int("PMS_PLAN_SNAPSHOT_KEEP", 200))
|
||
print(f" 落快照: "
|
||
f"{'新版本已入库' if snap.get('stored') else '与库里最新那份相同, 未新增'} "
|
||
f"(digest={snap.get('digest', '')[:12]} {snap.get('rows')} 只)")
|
||
d = ch.get("diff")
|
||
if not ch.get("ok"):
|
||
print(f" {ch.get('hint') or '算不了'}")
|
||
else:
|
||
if d.get("prev_broken"):
|
||
print(" ! 上一版名册**读不出来** (快照损坏) —— 这次比不了, "
|
||
"下面的空白不代表没有变化")
|
||
elif ch.get("same_as_stored"):
|
||
print(" 与库里最新那份**完全相同** (指纹一致) —— 上游没重算")
|
||
kind = {"first": "库里还没有快照, 无可比对的上一版 (跑一次 --snapshot 或页面强刷即有)",
|
||
"cross_day": "跨交易日的正常升降档",
|
||
"same_date_revision": "**同一计划日的新版本 —— 上游重算过** (§7.3 已知行为)"}
|
||
print(f" {kind.get(d['kind'], d['kind'])}")
|
||
print(f" {d['note']}")
|
||
h = d.get("held") or {}
|
||
for x in (h.get("exited") or []):
|
||
print(f" ! 持仓票掉榜: {x['ts_code']} {x['name']} "
|
||
f"(上版 rank {x['rank']} / {x['tier']})")
|
||
for x in (h.get("tier_down") or []):
|
||
print(f" ! 持仓票降档: {x['ts_code']} {x['name']} "
|
||
f"{x['tier_from']} → {x['tier_to']}")
|
||
for x in (h.get("coverage_lost") or []):
|
||
print(f" ! 持仓票丢了券商覆盖 (主榜→观察档): {x['ts_code']} {x['name']}")
|
||
if not h.get("n_watch") and d["kind"] != "first" and not d.get("prev_broken"):
|
||
print(" 持仓票没被摘、没降档、也没丢券商覆盖")
|
||
log = pf.snapshot_log(limit=30)
|
||
if log.get("revised"):
|
||
print(" 同一计划日出现多版: "
|
||
+ ", ".join(f"{k}×{v}" for k, v in log["revised"].items()))
|
||
except Exception as e:
|
||
print(f" ! 算榜单变化失败: {type(e).__name__}: {e}")
|
||
|
||
if args.json:
|
||
print("[8] 解析结构")
|
||
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())
|