batch-inference-analysis

> ⚙️ 本技能遵循公共运行环境契约（最高优先级，先于本文其余内容执行）： > 详见 [_shared/env-contract.md](../_shared/env-contract.md)，执行前先读它。 # 批量推理结果分析技能 基于 QuantMind 选股策略方法论，分析批量推理产出的每日信号，做出入场/选股/买卖/做空决策。 ## 数据基础 批量推理产出： - **batch**：多个交易日的推理集合（`/models/inference/batch/{batch_id}`） - `member_runs`：每个交易日的 run 记录（run_id/trade_date/signals_count） - **单日 run**：`/models/inference/runs/{run_id}` - `items`：5377+ 只股票信号，每只含 `fusion_score`（分数）/`board`（板块）/`industry`（行业）/`market_cap_tier`（市值分档）/`trend`（趋势）/`prev_score`/`prev2_score`/`next_score` ## 认证 ```bash BASE=http://127.0.0.1:8000 TOKEN=$(curl -s -X POST $BASE/api/v1/auth/login -H "Content-Type: application/json" \ -d '{"username":"admin","password":"admin123","tenant_id":"default"}' \ | python3 -c "import sys,json; print(json.load(sys.stdin).get('access_token',''))") AUTH="Authorization: Bearer $TOKEN" ``` ## 1. 拉取批量推理数据 ```bash # 1. 批量推理历史 curl -s -H "$AUTH" "$BASE/api/v1/models/inference/batches?page=1&page_size=10" # 2. 某批次详情（含 member_runs 每日记录） curl -s -H "$AUTH" "$BASE/api/v1/models/inference/batch/{batch_id}" # 3. 单日 run 信号（分析数据源） curl -s -H "$AUTH" "$BASE/api/v1/models/inference/runs/{run_id}" ``` **分析入口**：用 batch 的 `member_runs` 拿到每日 run_id，逐个拉取信号分析，或用最近一个 run 分析今日。 ### 批量聚合（推荐，一次拿全） `/models/inference/batch/{batch_id}/aggregate` 一次性返回整批的聚合分析（免去逐日拉取）： ```bash curl -s -H "$AUTH" "$BASE/api/v1/models/inference/batch/{batch_id}/aggregate" # 返回: {per_symbol, groups, movers, daily, meta} # per_symbol: 每只股票跨日分数/排名/IC # groups: 行业/板块分组聚合 # movers: 分数变动最大的标的 # daily: 每日 TopN/分布 # meta: 批次数/区间/IC/共识带（Mann-Kendall 趋势） ``` ## 2. 市场状态判断（三层过滤） ### 第1层：行业信号强度 — 决定是否入场 用 Python 计算行业 avg Top1： ```python import json, urllib.request def fetch_run(run_id): req = urllib.request.Request(f"{BASE}/api/v1/models/inference/runs/{run_id}", headers={"Authorization": f"Bearer {TOKEN}"}) return json.load(urllib.request.urlopen(req)) # 顶层直接含 items/total def market_signal(items, top_n=20): # 取 Top20 按分数 top = sorted(items, key=lambda x: -(x.get("fusion_score") or 0))[:top_n] # 按行业分组，取每行业最高分 ind_top = {} for it in top: ind = it.get("industry", "") or "未知" s = it.get("fusion_score") or 0 if ind not in ind_top or s > ind_top[ind]: ind_top[ind] = s ind_avg_top1 = sum(ind_top.values()) / max(1, len(ind_top)) strong = sum(1 for s in ind_top.values() if s >= 0.10) return {"ind_avg_top1": ind_avg_top1, "strong_industries": strong, "top_industries": dict(sorted(ind_top.items(), key=lambda x:-x[1])[:5])} ``` **入场阈值**（参数扫描优化）： | 策略 | 入场线 | 空仓线 | 强行业数 | |------|--------|--------|---------| | 保守 | ≥0.10 | <0.10 | ≥5 | | **平衡（推荐）** | **≥0.09** | **<0.06** | **≥2** | | 激进 | ≥0.07 | <0.06 | ≥1 | **判断**： - `ind_avg_top1 < 0.06` → 绝对空仓 - `0.06-0.09` → 观望/极轻仓 - `≥0.09` → 正常选股 ### 第2层：大盘均线过滤 — 防崩盘 模型在暴跌时给高分是反向信号。用指数 K 线判断： ```bash # 上证指数 K 线（QuantDB index_daily） curl -s -H "$AUTH" "$BASE/api/v1/market/kline?symbol=000001.SH&market=A&period=daily&days=30" ``` **规则**：上证指数 < MA20 → 强制空仓（避开 2024 年 2 月微盘崩盘）。 ### 第3层：强行业数仓位管理 | 强行业数 | 市场状态 | 仓位 | |---------|---------|------| | ≥5 | 强势 | 100% 满仓 | | 3-5 | 震荡偏强 | 50% 半仓 | | 2-3 | 震荡 | 30% 轻仓 | | <1.5 | 弱势 | 空仓 | ## 3. 个股选股（分数区间） ### 核心分数区间 | 个股分数 | 操作 | 理由 | |---------|------|------| | **0.10-0.12** | **首选（黄金区间）** | 胜率64.8%，均收+1.19%，最大亏损-10.4% | | 0.12-0.15 | 可选（主板优先） | 胜率79%但样本少，警惕追高 | | 0.15-0.20 | 谨慎 | 仅强市有效 | | ≥0.20 | 极谨慎 | 趋势加速，样本少 | | <0.10 | 不买 | 信号太弱 | **注意**：0.12-0.14 是"追高陷阱"（动量特征强，易买山顶）；0.10-0.11 假信号区必须配合行业确认。 **重要**：以上绝对阈值基于特定模型分布，换模型后先看 `score_distribution`（第7节）用分位数映射。 ```python def pick_stocks(items, market_sig): picks = [] for it in items: s = it.get("fusion_score") or 0 if not (0.10 <= s <= 0.12): continue if market_sig["ind_avg_top1"] < 0.09 and s < 0.11: continue # 假信号区 # 主板优先 board = it.get("board", "") if "创业" in board or "科创" in board: continue # 排除 ST name = it.get("stock_name", "") if "ST" in name or "退" in name: continue picks.append(it) picks.sort(key=lambda x: -(x.get("fusion_score") or 0)) return picks[:5] # 每天选3-5只 ``` ### 3天趋势（决定买点） 用 `prev_score`/`next_score` 判断： | 趋势 | 模式 | 操作 | |------|------|------| | **先升后降** | 低→高→降 | **最佳买点**（胜率78%） | | 连续上升 | 低→高→更高 | 过热不追 | | 连续下降 | 高→降→降 | 信号衰退不买 | **反直觉**：分数下降比上升好——高位回落说明绝对分仍高。 ## 4. 负分参考（做空/回避） ### 精确做空分数线 | 条件 | 下跌概率 | 做空均收 | |------|---------|---------| | **微盘(<30亿) + 分数≤-0.20** | 72.0% | +5.14% | | 微盘 + 分数≤-0.15 | 68.6% | +3.17% | | 小盘 + 分数≤-0.20 | 65.5% | +2.62% | | 中盘 + 分数≤-0.17 | 60.4% | +1.16% | | 大盘/超大盘 + 任何负分 | ≤52% | 不做空 | ### 负分也会涨（错杀） - **超大盘负分 -0.13~-0.14**：下跌概率仅 42%，均收 +3%——错杀最佳例证 - 大盘负分 -0.11：上涨概率 51% - **科创板负分 -0.06~-0.15**：均收全为正，做空价值最低 - 警戒线 -0.22：即使大盘股低于此分也会崩 ### 一句话落地 **做空只做微盘/小盘 + 分数≤-0.15；大盘/超大盘/科创板的负分是错杀，-0.22 以下才真危险；轻负分(>-0.06)无信息。** ```python def short_candidates(items): shorts = [] for it in items: s = it.get("fusion_score") or 0 if s > -0.15: continue tier = it.get("market_cap_tier", "") if tier in ("大盘", "超大盘"): continue # 错杀 if "科创" in it.get("board", ""): continue # 抗跌 shorts.append(it) shorts.sort(key=lambda x: x.get("fusion_score") or 0) return shorts[:10] ``` ## 5. 行业轮动分析 ```python def sector_rotation(batch_runs): # 跨多个交易日统计强行业出现频次 from collections import Counter ind_days = Counter() for run in batch_runs[:30]: # 最近30个交易日 items = fetch_run(run["run_id"]).get("items", []) top = sorted(items, key=lambda x: -(x.get("fusion_score") or 0))[:20] for it in top: if (it.get("fusion_score") or 0) >= 0.10: ind_days[it.get("industry", "")] += 1 return ind_days.most_common(10) ``` **判断**： - 行业信号持续 ≥3 天 = 真轮动（非一日游） - 新行业出现 Top1 ≥0.10 = 资金切换方向 - 强行业数 ≥3/天 = 有行情；≤1.5/天 = 空仓 ## 6. 完整分析流程 当用户要求"分析批量推理结果"时： 1. **确认批次**：`/models/inference/batches` 找最近 completed 的 batch 2. **拿 member_runs**：`/models/inference/batch/{id}` 得到每日 run_id 3. **拉今日信号**：取最新 run 的 items 4. **市场状态**：行业 avg Top1 + 强行业数 + 大盘均线 5. **选股**：分数 0.10-0.12 + 主板 + 3天趋势"先升后降" 6. **负分参考**：微盘/小盘极端负分列出做空候选 7. **行业轮动**：跨日统计强行业 8. **输出决策**：入场/仓位/个股清单/做空清单/规避 ## 7. 不同模型分数范围的处理（关键） **每个模型训练后分数范围不同**（实测：这个融合模型 fusion_score 范围 -0.996 ~ 0.965，而方法论假设 0-0.2 是另一套模型）。**绝对阈值不能直接套用**，必须先了解当前模型的分数分布。 ```python def score_distribution(items): scores = sorted(x.get("fusion_score") or 0 for x in items) n = len(scores) return { "min": round(scores[0], 3), "max": round(scores[-1], 3), "p10": round(scores[int(n*0.1)], 3), "p25": round(scores[int(n*0.25)], 3), "p50": round(scores[int(n*0.5)], 3), "p75": round(scores[int(n*0.75)], 3), "p90": round(scores[int(n*0.9)], 3), } ``` ### 模型分数范围 → 方法论阈值映射 方法论阈值（0.10-0.12 黄金区间等）基于特定模型的分布。使用时： 1. **先跑 `score_distribution`** 了解当前模型分布 2. **用分位数替代绝对分数**判断高低： - 黄金区间 ≈ 高分位带（如 p75-p90，需结合回测验证） - 追高区 ≈ 最高分位（p90+） - 负分做空 ≈ 最低分位（p10 以下） 3. **跨模型对比时**，统一用百分位排名，不用原始分数 ### 模型选择（多模型择优） 不同模型分数分布和预测能力不同，分析时可对比： ```bash # 各模型推理历史分数（单股） curl -s -H "$AUTH" "$BASE/api/v1/models/inference/stock/{symbol}/history" # 模型列表（含元数据） curl -s -H "$AUTH" "$BASE/api/v1/models" ``` **择优标准**： - 分数分布合理（不过度集中、有区分度） - 该模型回测胜率/IC 高（见 [[backtest-center]]） - 黄金区间候选数量适中（太多=无区分度，太少=过严） ### 融合模型特殊处理 融合模型已按百分位加权合成，`fusion_score` 可直接用，但分数范围仍因源模型而异——同样先看分布再定阈值。 ## 8. 模型分数校准回测（核心新增） **每个模型训练后分数范围不同**，用历史信号自动回测找出该模型最适合的分数区间。 ### 8.1 调用校准接口（异步任务 + 进度） 校准是**后台任务**，提交立即返回 `task_id`，轮询查进度（避免阻塞引擎）： ```bash # 1. 提交校准任务（秒回 task_id） curl -s -X POST -H "$AUTH" "$BASE/api/v1/selection/score-calibration?days=180&horizons=1,3,5,10&top_n=50" # 返回: {"status":"submitted","task_id":"calib_xxx", ...} # 2. 轮询任务进度（progress: 5→100，message 显示阶段） curl -s -H "$AUTH" "$BASE/api/v1/selection/score-calibration/{task_id}" # 返回: {"status":"running/completed","progress":85,"message":"回测中... 45/60 天","result":{...}} ``` **参数**：