@@ -562,6 +614,13 @@ createApp({
const instructions = ref([]), ledger = ref([]), proposals = ref([]);
const report = ref({}), reportDrawer = ref(false), dm = ref({});
const opsDrawer = ref(false), opsResult = ref(''), opsLoading = ref(false);
+ // 上游选股计划: 单独取数, 不塞进 /api/overview —— overview 是轮询的, 上游不通时
+ // 会把 10 秒超时拖进每一次页面刷新。
+ const planDrawer = ref(false), planLoading = ref(false), planRaw = ref({}),
+ planResult = ref('');
+ const planStatus = computed(() => planRaw.value.status || {});
+ const planRows = computed(() => planRaw.value.rows || []);
+ const planCand = computed(() => planRaw.value.candidates_raw || null);
const issuing = ref(false);
const form = reactive({ cmd_type: '', params: {}, note: '' });
@@ -814,12 +873,26 @@ createApp({
const d = await call('get', '/api/ops/downstream-schema');
opsResult.value = JSON.stringify(d, null, 2); opsLoading.value = false;
}
+ async function loadPlan() {
+ planLoading.value = true;
+ const d = await call('get', '/api/upstream/plan?limit=50');
+ planRaw.value = (d && d.ok) ? d : { status: { ok: false, hint: (d||{}).error || '取数失败' } };
+ planLoading.value = false;
+ }
+ async function openPlan() { planDrawer.value = true; await loadPlan(); }
+ async function refreshPlan() {
+ planLoading.value = true;
+ const d = await call('post', '/api/ops/plan-refresh');
+ planResult.value = JSON.stringify(d, null, 2);
+ planLoading.value = false;
+ await Promise.all([loadPlan(), loadOverview()]);
+ }
async function openReport() {
const d = await call('get', '/api/report'); report.value = d.data || d || {};
reportDrawer.value = true;
}
- onMounted(loadAll);
+ onMounted(() => { loadAll(); loadPlan(); }); // 计划状态并行拉, 不阻塞首屏
return { tab, loading, err, health, ov, params, catalog, commands, plans, plansOf,
positions, lots, lotsOf, instructions, ledger, proposals, report, reportDrawer,
opsDrawer, opsResult, opsLoading, issuing, form, curSpec, dirtyCount, dm,
@@ -827,7 +900,9 @@ createApp({
scaleGap,
wsRaw, wsAuto, wsHidePong, wsc, wst, wsMode, wsOrders, wsInbox, wsWarnings, loadWs,
loadAll, loadParams, saveParams, loadPlans, loadLots, onCmdChange, issue,
- cancelCmd, replan, decide, ops, loadSchema, openReport, cancelIns };
+ cancelCmd, replan, decide, ops, loadSchema, openReport, cancelIns,
+ planDrawer, planLoading, planStatus, planRows, planCand, planResult,
+ loadPlan, openPlan, refreshPlan };
}
}).use(ElementPlus).mount('#app');
diff --git a/config/settings.py b/config/settings.py
index 768b0b0..b03f7a1 100644
--- a/config/settings.py
+++ b/config/settings.py
@@ -71,6 +71,22 @@ class Settings(BaseSettings):
PMS_WEAK_NEG_DAYS: int = 5 # 降仓「清弱票」判定: 安全垫连续为负 N 日 (设计 §3.2)
PMS_PROPOSAL_TTL_HOURS: int = 24 # 自主提议待确认有效期 (超时置 EXPIRED)
+ # --- 上游选股计划接口 (候选池的事实源; 口径与待确认项见 UPSTREAM_PLAN_API.md) ---
+ # 上游只回答「买什么、排第几」, 不给价格金额 —— 买多少/什么价是 PMS 自己算。
+ # 接口拿不到一律显式失败 (候选池为空 + ERROR), 绝不静默回退旧表。
+ PMS_CANDIDATE_SOURCE: str = "plan_api" # plan_api(默认) / buy_plan(旧表) / both(并集)
+ PMS_PLAN_API_BASE: str = "http://192.168.16.155:8300" # 空=停用, 此时候选池恒为空
+ PMS_PLAN_API_PATH: str = "/plan"
+ PMS_PLAN_TIMEOUT: int = 10 # 单次请求超时 (秒)
+ PMS_PLAN_CACHE_SEC: int = 300 # 计划缓存秒数 (上游日频产出, 没必要每跳都拉)
+ PMS_PLAN_TOP_N: int = 30 # 主榜按 score 降序取前 N 进池 (900+ 只是排序池不是清单)
+ PMS_PLAN_TIERS: str = "强传导" # 传导档白名单; 留空=不按档过滤
+ PMS_PLAN_INCLUDE_OBSERVE: bool = False # 观察档是否进候选池 (上游把它定位为备选, 无 upside)
+ PMS_PLAN_MIN_SCORE: float = 0.0 # score 下限, 0=不设
+ PMS_PLAN_MIN_SOURCES: int = 0 # evidence.n_sources 下限, 0=不设
+ PMS_PLAN_STALE_TDAYS: int = 1 # 计划日龄超此交易日数即判过期并拒用 (防上游停更)
+ PMS_PLAN_THEME_SYNC: bool = True # 刷新时把 evidence.theme 灌进 pms_industry_map
+
# --- 行业约束 (硬拦截; 数据源接口化) ---
PMS_SECTOR_SOURCE: str = "" # "" = 停用并页面提示 / custom_table / gp_stock_category
PMS_SECTOR_MAX_NAMES: int = 4
diff --git a/scripts/probe_plan_api.py b/scripts/probe_plan_api.py
new file mode 100644
index 0000000..1a806ab
--- /dev/null
+++ b/scripts/probe_plan_api.py
@@ -0,0 +1,119 @@
+# -*- 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 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("--json", action="store_true", help="原样打印解析后的结构")
+ args = ap.parse_args()
+
+ from app.services import param_store as ps
+ base = (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)
+ 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} {'upside':>8}"
+ 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} {_fmt(r['upside']):>8}")
+ 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())
diff --git a/scripts/run_tests.py b/scripts/run_tests.py
index 5e44b25..616fc11 100644
--- a/scripts/run_tests.py
+++ b/scripts/run_tests.py
@@ -5,13 +5,15 @@
运行: docker compose run --rm pms-web python scripts/run_tests.py
包含:
- test_core_units.py 仓位规划器 / 安全垫与成本账 (14 例)
- test_batch2_units.py 命令状态机 / 方案生成器 / 回放对账纯逻辑 (35 例)
- test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 (19 例)
- test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 (11 例)
- test_batch5_units.py 决策系统信号流解析与消化口径 (8 例)
- test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 (58 例)
- test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (32 例)
+ test_core_units.py 仓位规划器 / 安全垫与成本账 (14 例)
+ test_batch2_units.py 命令状态机 / 方案生成器 / 回放对账纯逻辑 (35 例)
+ test_batch3_units.py 规则闸 / 择时执行器实现B 纯逻辑 (21 例)
+ test_batch4_units.py 动作引擎 四类自主动作触发与数量口径 (11 例)
+ test_batch5_units.py 决策系统信号流解析与消化口径 (8 例)
+ test_batch6_units.py ws 通道: 测试向量/签名/公钥/水位/DDL/逐笔入账 (65 例)
+ test_batch7_units.py 上游选股计划: 解析/新鲜度/候选筛选/取数守卫 (23 例)
+ test_wiring.py 装配自检: 服务层→核心→落表 全链路 (内存桩) (51 例)
+ 共 228 例
任一子集失败即整体失败 (退出码 1)。
"""
import os
@@ -22,7 +24,7 @@ HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py",
"test_batch4_units.py", "test_batch5_units.py", "test_batch6_units.py",
- "test_wiring.py"]
+ "test_batch7_units.py", "test_wiring.py"]
def main():
diff --git a/scripts/test_batch7_units.py b/scripts/test_batch7_units.py
new file mode 100644
index 0000000..36f9938
--- /dev/null
+++ b/scripts/test_batch7_units.py
@@ -0,0 +1,400 @@
+# -*- coding: utf-8 -*-
+"""
+第七批模块单测 (零外部依赖, 不连库不触网)
+==========================================
+运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch7_units.py
+覆盖: 上游选股计划接口 (plan_feed) 的解析、新鲜度校验、候选筛选三段纯逻辑。
+
+fixture 用的是 2026-07-30 上游给的真实应答样例 (截取前若干条, 结构一字未改),
+坏数据用例另造 —— 「上游改了字段/漏了字段」这类事故必须由单测先炸, 而不是盘中静默少票。
+"""
+import os
+import sys
+import traceback
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from app.services import plan_feed as pf # noqa: E402
+
+RESULTS = []
+
+
+def case(name):
+ def deco(fn):
+ RESULTS.append((name, fn))
+ return fn
+ return deco
+
+
+def raises(exc, fn, *a, **kw):
+ try:
+ fn(*a, **kw)
+ except exc:
+ return True
+ except Exception as e:
+ raise AssertionError(f"期望 {exc.__name__}, 实际 {type(e).__name__}: {e}")
+ raise AssertionError(f"期望抛 {exc.__name__}, 但没抛")
+
+
+# ================================================================ fixture
+def _m(rank, code, name, score, theme, heat, upside, tier="强传导", n_sources=7, moved=0.0):
+ return {"rank": rank, "code": code, "name": name, "score": score,
+ "evidence": {"theme": theme, "n_sources": n_sources, "moved_ratio": moved},
+ "heat": heat, "upside": upside, "tier": tier}
+
+
+def _o(rank, code, name, score, theme, heat, n_sources=9, moved=0.075):
+ return {"rank": rank, "code": code, "name": name, "score": score,
+ "evidence": {"theme": theme, "n_sources": n_sources, "moved_ratio": moved},
+ "heat": heat, "upside": None}
+
+
+SAMPLE = {
+ "date": "2026-07-29",
+ "counts": {"main": 961, "observe": 107, "gate_covered": 2386},
+ "market_snapshot_days": ["2026-07-28"],
+ "heat_date": "2026-07-27",
+ "theme_cap": 5,
+ "main": [
+ _m(1, "SH600418", "江淮汽车", 242.24, "整车", 0.2249, 2.1203202525935945),
+ _m(2, "SH688717", "艾罗能源", 242.05, "储能", 0.4165, 1.5817543859649121),
+ _m(3, "SZ300952", "恒辉安防", 241.91, "传感器", 0.3391, 1.4061712846347607),
+ _m(4, "SH605598", "上海港湾", 241.72, "储能", 0.214, 1.292594091460947),
+ _m(6, "SH600875", "东方电气", 241.34, "整机", 0.4074, 1.2340787236824915,
+ n_sources=9, moved=0.075),
+ _m(10, "SH600104", "上汽集团", 241.18, "整车", 0.1239, 0.8065614360879017),
+ ],
+ "observe": [
+ _o(1, "SH600877", "电科芯片", 101.18, "集成电路设计", 0.4743, moved=0.0),
+ _o(2, "SH603611", "诺力股份", 101.12, "整机", 0.5868),
+ _o(3, "SZ000768", "中航西飞", 101.06, "整机", 0.121),
+ ],
+ "changes": None,
+ "encoding": "主榜分=200+传导档位×20+组内分（还没热、还便宜）；观察档分=100+0.6z(传导)+0.4z(−热度)",
+}
+
+
+# ================================================================ 解析
+@case("解析·真实样例: 计数/主题上限/热度日全部落位, 返回条数与 counts 分开记")
+def _():
+ p = pf.parse_plan(SAMPLE)
+ assert p["date"] == "2026-07-29" and p["heat_date"] == "2026-07-27"
+ assert p["theme_cap"] == 5
+ assert p["counts"] == {"main": 961, "observe": 107, "gate_covered": 2386}
+ # counts 是上游全量, returned 是这次应答里真的给了几条 —— 混用会看不出上游截断了
+ assert p["returned"] == {"main": 6, "observe": 3}
+ assert p["market_snapshot_days"] == ["2026-07-28"]
+ assert p["encoding"].startswith("主榜分=200")
+
+
+@case("解析·代码从前缀式归一为点式 (沪/深/科创/创业板四种都要对)")
+def _():
+ p = pf.parse_plan(SAMPLE)
+ codes = [r["ts_code"] for r in p["main"]]
+ assert codes[:4] == ["600418.SH", "688717.SH", "300952.SZ", "605598.SH"], codes
+ assert [r["ts_code"] for r in p["observe"]][2] == "000768.SZ"
+
+
+@case("解析·字段口径: score/heat/upside 转 float, evidence 拆平, bucket 打标")
+def _():
+ p = pf.parse_plan(SAMPLE)
+ r = p["main"][0]
+ assert r["name"] == "江淮汽车" and r["rank"] == 1 and r["tier"] == "强传导"
+ assert abs(r["score"] - 242.24) < 1e-9 and abs(r["heat"] - 0.2249) < 1e-9
+ assert abs(r["upside"] - 2.1203202525935945) < 1e-12
+ assert r["theme"] == "整车" and r["n_sources"] == 7 and r["moved_ratio"] == 0.0
+ assert r["bucket"] == pf.BUCKET_MAIN and p["observe"][0]["bucket"] == pf.BUCKET_OBSERVE
+ # 观察档没有 tier、upside 是 null —— 都得是 None 而不是 0
+ assert p["observe"][0]["tier"] is None and p["observe"][0]["upside"] is None
+
+
+@case("解析·themes 映射: 主榜优先于观察档, 同码不被观察档覆盖")
+def _():
+ d = dict(SAMPLE)
+ d["observe"] = list(SAMPLE["observe"]) + [_o(9, "SH600418", "江淮汽车", 100.1, "汽车零部件", 0.2)]
+ p = pf.parse_plan(d)
+ assert p["themes"]["600418.SH"] == "整车", p["themes"]["600418.SH"]
+ assert p["themes"]["600877.SH"] == "集成电路设计"
+
+
+@case("解析·变形应答一律抛 PlanFeedError (不是 JSON 对象 / 缺 date / 两档全空)")
+def _():
+ raises(pf.PlanFeedError, pf.parse_plan, ["不是对象"])
+ raises(pf.PlanFeedError, pf.parse_plan, None)
+ raises(pf.PlanFeedError, pf.parse_plan, {"main": SAMPLE["main"]}) # 缺 date
+ raises(pf.PlanFeedError, pf.parse_plan, {"date": "2026-07-29", "main": [], "observe": []})
+ raises(pf.PlanFeedError, pf.parse_plan, {"date": " ", "main": SAMPLE["main"]})
+
+
+@case("解析·脏行跳过而不整体失效: 非 dict / 缺 code / 同码重复 / evidence 非 dict")
+def _():
+ d = {"date": "2026-07-29", "main": [
+ "我是脏行", None, 42,
+ {"rank": 1, "code": "SH600418", "score": 1.0, "evidence": "不是对象"},
+ {"rank": 2, "code": "", "score": 2.0},
+ {"rank": 3, "code": "600418", "score": 3.0}, # 与第一条同码 (归一后相同)
+ {"code": "SZ300952", "score": 4.0}, # 缺 rank
+ ]}
+ p = pf.parse_plan(d)
+ codes = [r["ts_code"] for r in p["main"]]
+ assert codes == ["600418.SH", "300952.SZ"], codes
+ assert p["main"][0]["theme"] is None and p["main"][0]["n_sources"] is None
+ # rank 缺失用出现序号补 (第 7 个元素 → 7), 免得排序键出现 None
+ assert p["main"][1]["rank"] == 7, p["main"][1]["rank"]
+ assert p["counts"] == {"main": None, "observe": None, "gate_covered": None}
+
+
+@case("解析·数值容错: 字符串数字可用, 空串/None/非数字一律 None 而不是 0")
+def _():
+ d = {"date": "2026-07-29", "main": [
+ {"rank": "5", "code": "SH600418", "score": "242.24", "heat": "", "upside": None,
+ "evidence": {"theme": " 整车 ", "n_sources": "7", "moved_ratio": "abc"}}]}
+ r = pf.parse_plan(d)["main"][0]
+ assert r["rank"] == 5 and abs(r["score"] - 242.24) < 1e-9
+ assert r["heat"] is None and r["upside"] is None and r["moved_ratio"] is None
+ assert r["n_sources"] == 7 and r["theme"] == "整车" # theme 两头空白要去掉
+
+
+# ================================================================ 新鲜度
+@case("新鲜度·日龄: 当天 0 / 上一交易日 1 / 未来日期按 0 (为下一交易日出的计划)")
+def _():
+ assert pf.plan_age_tdays("2026-07-30", today="2026-07-30") == 0
+ assert pf.plan_age_tdays("2026-07-29", today="2026-07-30") == 1
+ assert pf.plan_age_tdays("2026-07-27", today="2026-07-30") == 3
+ assert pf.plan_age_tdays("2026-07-31", today="2026-07-30") == 0
+
+
+@case("新鲜度·跨周末只算交易日: 周五的计划到周一仍是 1 个交易日龄")
+def _():
+ # 2026-07-24 周五, 2026-07-27 周一 —— 中间两天不是交易日, 不该被算进日龄
+ assert pf.plan_age_tdays("2026-07-24", today="2026-07-27") == 1
+ assert pf.plan_age_tdays("2026-07-24", today="2026-07-28") == 2
+
+
+@case("新鲜度·超期抛错并说清日龄 (上游停更时拿旧榜当今天比没候选更危险)")
+def _():
+ p = pf.parse_plan(SAMPLE) # date = 2026-07-29
+ assert pf.assert_fresh(p, max_stale_tdays=1, today="2026-07-30") == 1
+ assert pf.assert_fresh(p, max_stale_tdays=0, today="2026-07-29") == 0
+ raises(pf.PlanFeedError, pf.assert_fresh, p, max_stale_tdays=0, today="2026-07-30")
+ try:
+ pf.assert_fresh(p, max_stale_tdays=1, today="2026-08-05")
+ except pf.PlanFeedError as e:
+ assert "2026-07-29" in str(e) and "交易日" in str(e), str(e)
+
+
+# ================================================================ 筛选
+@case("筛选·按 score 降序取前 N, 同分按上游 rank 稳定次序")
+def _():
+ p = pf.parse_plan(SAMPLE)
+ r = pf.select_candidates(p, top_n=3)
+ assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ"]
+ assert r["considered"] == 6 and r["eligible"] == 6 and r["dropped"]["capped"] == 3
+ # 同分: rank 小的在前
+ d = {"date": "2026-07-29", "main": [_m(9, "SH600001", "甲", 100.0, "整车", 0.1, 1.0),
+ _m(2, "SH600002", "乙", 100.0, "整车", 0.1, 1.0)]}
+ rr = pf.select_candidates(pf.parse_plan(d), top_n=2)
+ assert [x["ts_code"] for x in rr["items"]] == ["600002.SH", "600001.SH"]
+
+
+@case("筛选·top_n=0 视为不截断 (别把'不限'写成'一只都不要')")
+def _():
+ p = pf.parse_plan(SAMPLE)
+ assert len(pf.select_candidates(p, top_n=0)["items"]) == 6
+ assert len(pf.select_candidates(p, top_n=999)["items"]) == 6
+
+
+@case("筛选·已持有与黑名单剔除, 且两者分开计数 (输入接受前缀式)")
+def _():
+ p = pf.parse_plan(SAMPLE)
+ r = pf.select_candidates(p, held=["SH600418", "688717.SH"], black=["SZ300952"], top_n=10)
+ codes = [x["ts_code"] for x in r["items"]]
+ assert "600418.SH" not in codes and "688717.SH" not in codes and "300952.SZ" not in codes
+ assert r["dropped"]["held"] == 2 and r["dropped"]["black"] == 1
+ assert r["eligible"] == 3
+
+
+@case("筛选·tier 白名单只对带 tier 的行生效; 观察档的闸门是 include_observe")
+def _():
+ d = dict(SAMPLE)
+ d["main"] = list(SAMPLE["main"]) + [
+ _m(500, "SH600519", "弱票", 220.0, "白酒", 0.1, 0.5, tier="弱传导")]
+ p = pf.parse_plan(d)
+ r = pf.select_candidates(p, tiers=["强传导"], top_n=50)
+ assert "600519.SH" not in [x["ts_code"] for x in r["items"]]
+ assert r["dropped"]["tier"] == 1
+ # 观察档没有 tier —— tiers 非空时也不该被 tier 规则误杀, 但默认根本不进池
+ assert all(x["bucket"] == "main" for x in r["items"])
+ r2 = pf.select_candidates(p, tiers=["强传导"], include_observe=True, top_n=50)
+ obs = [x["ts_code"] for x in r2["items"] if x["bucket"] == "observe"]
+ assert obs == ["600877.SH", "603611.SH", "000768.SZ"], obs
+ assert r2["dropped"]["tier"] == 1
+
+
+@case("筛选·跨档同码只留一次, 且留主榜那份 (主榜分高, 排序天然优先)")
+def _():
+ d = dict(SAMPLE)
+ d["observe"] = list(SAMPLE["observe"]) + [_o(9, "SH600418", "江淮汽车", 100.1, "汽车零部件", 0.2)]
+ r = pf.select_candidates(pf.parse_plan(d), include_observe=True, top_n=50)
+ hit = [x for x in r["items"] if x["ts_code"] == "600418.SH"]
+ assert len(hit) == 1 and hit[0]["bucket"] == "main" and hit[0]["theme"] == "整车"
+ assert r["dropped"]["dup"] == 1
+
+
+@case("筛选·score 下限与 n_sources 下限各自独立计数")
+def _():
+ p = pf.parse_plan(SAMPLE)
+ r = pf.select_candidates(p, min_score=241.7, top_n=50)
+ assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ",
+ "605598.SH"]
+ assert r["dropped"]["score"] == 2
+ r2 = pf.select_candidates(p, min_sources=9, top_n=50)
+ assert [x["ts_code"] for x in r2["items"]] == ["600875.SH"], r2["items"]
+ assert r2["dropped"]["sources"] == 5
+
+
+@case("筛选·输出字段: sector 用 theme 灌 (planner 吃这个), score 缺失兜 0.0 不留 None")
+def _():
+ d = {"date": "2026-07-29", "main": [{"rank": 1, "code": "SH600418",
+ "evidence": {"theme": "整车"}}]}
+ x = pf.select_candidates(pf.parse_plan(d), top_n=5)["items"][0]
+ assert x["score"] == 0.0 and x["sector"] == "整车" and x["theme"] == "整车"
+ assert x["src"] == "plan_api" and x["ts_code"] == "600418.SH"
+ assert "price" not in x, "计划不带价格, 这里绝不能凭空造一个价出来"
+
+
+@case("筛选·计数自洽: considered = eligible + 各项丢弃; items = eligible − capped")
+def _():
+ d = dict(SAMPLE)
+ d["main"] = list(SAMPLE["main"]) + [
+ _m(500, "SH600519", "弱票", 220.0, "白酒", 0.1, 0.5, tier="弱传导")]
+ p = pf.parse_plan(d)
+ r = pf.select_candidates(p, held=["SH600418"], black=["SZ300952"], tiers=["强传导"],
+ min_sources=7, top_n=2, include_observe=True)
+ dr = r["dropped"]
+ assert r["considered"] == 10, r["considered"]
+ assert r["considered"] == r["eligible"] + dr["held"] + dr["black"] + dr["tier"] \
+ + dr["score"] + dr["sources"] + dr["dup"], (r, dr)
+ assert len(r["items"]) == r["eligible"] - dr["capped"] == 2
+
+
+@case("筛选·空计划不炸 (main/observe 缺键) 且 date 透传")
+def _():
+ r = pf.select_candidates({"date": "2026-07-29"}, top_n=5)
+ assert r["items"] == [] and r["considered"] == 0 and r["eligible"] == 0
+ assert r["date"] == "2026-07-29"
+
+
+# ================================================================ 取数守卫
+class _FakeResp:
+ def __init__(self, payload, status=200):
+ self._p, self.status_code = payload, status
+
+ def raise_for_status(self):
+ if self.status_code >= 400:
+ raise RuntimeError(f"HTTP {self.status_code}")
+
+ def json(self):
+ if isinstance(self._p, Exception):
+ raise self._p
+ return self._p
+
+
+def _with_fake_requests(handler):
+ """把 requests 换成假模块 (装在 sys.modules 上, plan_feed 是函数内 import)。"""
+ import types
+ calls = []
+ fake = types.ModuleType("requests")
+
+ def _get(url, params=None, timeout=None):
+ calls.append({"url": url, "params": params, "timeout": timeout})
+ return handler(url, params, timeout)
+ fake.get = _get
+ prev = sys.modules.get("requests")
+ sys.modules["requests"] = fake
+
+ def restore():
+ if prev is None:
+ sys.modules.pop("requests", None)
+ else:
+ sys.modules["requests"] = prev
+ return calls, restore
+
+
+@case("取数·base 为空立即抛 PlanFeedError, 一个 HTTP 请求都不发")
+def _():
+ calls, restore = _with_fake_requests(
+ lambda *a: (_ for _ in ()).throw(AssertionError("不该发请求")))
+ try:
+ raises(pf.PlanFeedError, pf.fetch, base="", path="/plan", timeout=1)
+ raises(pf.PlanFeedError, pf.fetch, base=" ", path="/plan", timeout=1)
+ finally:
+ restore()
+ assert calls == [], calls
+
+
+@case("取数·成功路径: URL 拼接/date 传参/超时透传, 并盖上 url 与 fetched_at")
+def _():
+ calls, restore = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
+ try:
+ p = pf.fetch(base="http://192.168.16.155:8300/", path="plan", timeout=7,
+ date="2026-07-29")
+ finally:
+ restore()
+ assert calls[0]["url"] == "http://192.168.16.155:8300/plan", calls
+ assert calls[0]["params"] == {"date": "2026-07-29"} and calls[0]["timeout"] == 7
+ assert p["date"] == "2026-07-29" and p["requested_date"] == "2026-07-29"
+ assert p["url"].endswith("/plan") and p["fetched_at"] > 0
+ # 不传 date 时不能带一个 date=None 的空参上去
+ calls2, restore2 = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
+ try:
+ pf.fetch(base="http://x:8300", path="/plan", timeout=3)
+ finally:
+ restore2()
+ assert calls2[0]["params"] is None, calls2
+
+
+@case("取数·HTTP 报错/JSON 坏掉一律包装成 PlanFeedError 并带上 URL")
+def _():
+ for handler in (lambda *a: _FakeResp(SAMPLE, status=500),
+ lambda *a: _FakeResp(ValueError("Expecting value")),
+ lambda *a: (_ for _ in ()).throw(OSError("Connection refused"))):
+ calls, restore = _with_fake_requests(handler)
+ try:
+ try:
+ pf.fetch(base="http://192.168.16.155:8300", path="/plan", timeout=1)
+ raise AssertionError("应当抛 PlanFeedError")
+ except pf.PlanFeedError as e:
+ assert "192.168.16.155:8300/plan" in str(e), str(e)
+ finally:
+ restore()
+
+
+@case("取数·来源常量与旧表来源都在: plan_api / buy_plan / both")
+def _():
+ assert (pf.SRC_PLAN_API, pf.SRC_BUY_PLAN, pf.SRC_BOTH) == ("plan_api", "buy_plan", "both")
+ assert pf.FAIL_CACHE_SEC > 0, "失败也要缓存一会儿, 否则每分钟的调度位会把超时叠成雪崩"
+
+
+def main():
+ import logging
+ logging.disable(logging.CRITICAL)
+ passed, failed = 0, 0
+ for name, fn in RESULTS:
+ try:
+ fn()
+ print(f" PASS {name}")
+ passed += 1
+ except Exception:
+ print(f" FAIL {name}")
+ traceback.print_exc()
+ failed += 1
+ print("-" * 60)
+ if failed:
+ print(f"FAILED: {failed} / {passed + failed}")
+ sys.exit(1)
+ print(f"ALL PASS ({passed} cases)")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/test_wiring.py b/scripts/test_wiring.py
index 5c0aca0..29b5111 100644
--- a/scripts/test_wiring.py
+++ b/scripts/test_wiring.py
@@ -469,6 +469,9 @@ def install_fakes(prices=None, positions=None, params=None, high5=None):
# 用 "0" 而不是随便一个字符串: next_cursor 只进不退, 且非数字 id 会退化成字典序比较,
# 占位值若比真实 order_id 大 (比如 "SEED"), 游标就永远推不动了。
fake.params.setdefault("PMS_REPLAY_CURSOR", "0")
+ # 上游选股计划接口在单测里一律停用 (base 为空 → plan_feed 立即抛 PlanFeedError,
+ # 不会发出任何 HTTP 请求)。要测候选池的用例自己 stub plan_feed.candidates。
+ fake.params.setdefault("PMS_PLAN_API_BASE", "")
param_store._cache.update({"at": 0.0, "data": {}, "error": None, "loaded": False})
portfolio.save_neg_streak({})
return fake
@@ -481,7 +484,8 @@ def _():
for m in ["app.core.sizer", "app.core.cushion", "app.core.command_spec", "app.core.planner",
"app.core.recon", "app.core.tradedays", "app.db.session", "app.repo.pms_repo",
"app.repo.downstream_repo", "app.services.param_store", "app.services.industry",
- "app.services.market", "app.services.portfolio", "app.services.command_service",
+ "app.services.market", "app.services.plan_feed",
+ "app.services.portfolio", "app.services.command_service",
"app.services.ledger_service", "app.web.main", "app.scheduler"]:
importlib.import_module(m)
@@ -496,7 +500,8 @@ def _():
"/api/positions/{ts_code}/lots", "/api/instructions", "/api/ledger", "/api/report",
"/api/proposals", "/api/proposals/{proposal_id}/decide", "/api/ops/replay",
"/api/ops/reconcile", "/api/ops/daily-settle", "/api/ops/report",
- "/api/ops/downstream-schema", "/api/industry"]
+ "/api/ops/downstream-schema", "/api/industry",
+ "/api/upstream/plan", "/api/ops/plan-refresh"]
missing = [p for p in need if p not in paths]
assert not missing, missing
import os as _os
@@ -508,11 +513,13 @@ def _():
def _():
from app import scheduler as sch
names = set(sch.celery_app.conf.beat_schedule)
- assert names == {"premarket", "command_poll", "replay_fills", "intraday_exec",
- "signal_digest", "t0_close", "daily_settle", "daily_report"}, names
+ assert names == {"plan_pull", "premarket", "command_poll", "replay_fills",
+ "intraday_exec", "signal_digest", "t0_close", "daily_settle",
+ "daily_report"}, names
tasks = set(sch.celery_app.tasks)
for t in ("pms.premarket", "pms.command_poll", "pms.replay_fills", "pms.daily_settle",
- "pms.daily_report", "pms.t0_close", "pms.intraday_exec", "pms.signal_digest"):
+ "pms.daily_report", "pms.t0_close", "pms.intraday_exec", "pms.signal_digest",
+ "pms.plan_pull"):
assert t in tasks, t
assert sch.in_session(datetime(2026, 7, 27, 10, 0)) is True
assert sch.in_session(datetime(2026, 7, 27, 12, 0)) is False
@@ -1664,6 +1671,91 @@ def _():
# ---------------------------------------------------------------- runner
+# ================================================================ 候选池 (上游 /plan)
+def _stub_sp(csvc, mapping):
+ """替换 effective_stock_params, 返回 restore 闭包。"""
+ orig = csvc.effective_stock_params
+ csvc.effective_stock_params = lambda: dict(mapping)
+ return lambda: setattr(csvc, "effective_stock_params", orig)
+
+
+@case("候选池·plan_api 独占: 价格现取 / 无价剔除 / 黑名单与持仓剔除 / 白名单压过计划票")
+def _():
+ from app.services import command_service as csvc, plan_feed
+ install_fakes(prices={"600418.SH": 12.5, "300952.SZ": 30.0, "600000.SH": 10.0,
+ "600104.SH": 18.0},
+ params={"PMS_TOTAL_SCALE": "980000", "PMS_CANDIDATE_SOURCE": "plan_api"})
+ restore_sp = _stub_sp(csvc, {"600000.SH": {"white": True}, "600104.SH": {"black": True}})
+ orig = plan_feed.candidates
+ try:
+ plan_feed.candidates = lambda **kw: {
+ "date": "2026-07-29", "considered": 4, "eligible": 4, "dropped": {},
+ "items": [{"ts_code": "600418.SH", "score": 242.24, "theme": "整车"},
+ {"ts_code": "688717.SH", "score": 242.05, "theme": "储能"},
+ {"ts_code": "300952.SZ", "score": 241.91, "theme": "传感器"},
+ {"ts_code": "600104.SH", "score": 241.18, "theme": "整车"}]}
+ pool = csvc._candidates({"held": [{"ts_code": "605598.SH"}]})
+ finally:
+ plan_feed.candidates = orig
+ restore_sp()
+ by = {c["ts_code"]: c for c in pool}
+ # 688717.SH 行情里没有价 → 必须剔除 (planner 对 price<=0 只是静默跳过, 那样少票看不出来)
+ assert "688717.SH" not in by, by
+ assert "600104.SH" not in by, "黑名单票不得进池"
+ assert by["600418.SH"]["price"] == 12.5 and by["600418.SH"]["src"] == "plan_api"
+ assert by["600418.SH"]["theme"] == "整车" and by["600418.SH"]["sector"] is None
+ # 白名单是用户点名的票, 分必须压过任何计划票 —— planner 只认 score 一把尺子
+ assert by["600000.SH"]["src"] == "whitelist"
+ plan_max = max(c["score"] for c in pool if c["src"] == "plan_api")
+ assert by["600000.SH"]["score"] > plan_max, (by["600000.SH"]["score"], plan_max)
+
+
+@case("候选池·上游计划不可用: 池为空, 绝不静默回退旧 trading_buy_plan 表")
+def _():
+ from app.repo import downstream_repo
+ from app.services import command_service as csvc, plan_feed
+ install_fakes(prices={"601111.SH": 9.0}, params={"PMS_CANDIDATE_SOURCE": "plan_api"})
+ restore_sp = _stub_sp(csvc, {})
+ orig_c, orig_bp = plan_feed.candidates, downstream_repo.fetch_buy_plans
+
+ def _boom(**kw):
+ raise plan_feed.PlanFeedError("接口超时")
+ try:
+ plan_feed.candidates = _boom
+ downstream_repo.fetch_buy_plans = lambda **kw: [
+ {"ts_code": "601111.SH", "price": 9.0, "score": 0.8}]
+ pool = csvc._candidates({"held": []})
+ finally:
+ plan_feed.candidates, downstream_repo.fetch_buy_plans = orig_c, orig_bp
+ restore_sp()
+ assert pool == [], f"上游拿不到时必须让候选池为空, 实际拿到 {pool}"
+
+
+@case("候选池·both: 计划票与旧表票取并集, 同码只留计划那份")
+def _():
+ from app.repo import downstream_repo
+ from app.services import command_service as csvc, plan_feed
+ install_fakes(prices={"600418.SH": 12.5, "601111.SH": 9.0},
+ params={"PMS_CANDIDATE_SOURCE": "both"})
+ restore_sp = _stub_sp(csvc, {})
+ orig_c, orig_bp = plan_feed.candidates, downstream_repo.fetch_buy_plans
+ try:
+ plan_feed.candidates = lambda **kw: {
+ "date": "2026-07-29", "considered": 1, "eligible": 1, "dropped": {},
+ "items": [{"ts_code": "600418.SH", "score": 242.24, "theme": "整车"}]}
+ downstream_repo.fetch_buy_plans = lambda **kw: [
+ {"ts_code": "600418.SH", "price": 11.0, "score": 0.9}, # 同码, 计划优先
+ {"ts_code": "601111.SH", "price": 9.0, "score": 0.8}]
+ pool = csvc._candidates({"held": []})
+ finally:
+ plan_feed.candidates, downstream_repo.fetch_buy_plans = orig_c, orig_bp
+ restore_sp()
+ by = {c["ts_code"]: c for c in pool}
+ assert set(by) == {"600418.SH", "601111.SH"}, by
+ assert by["600418.SH"]["src"] == "plan_api" and by["600418.SH"]["price"] == 12.5
+ assert by["601111.SH"]["src"] == "buy_plan" and by["601111.SH"]["price"] == 9.0
+
+
def main():
# 静音日志: 本套里有好几条用例**故意**触发异常与告警来验证「守成」行为
# (调度守卫吞异常、外部成交告警、连续对账升级 ERROR、窗口耗尽告警),