Batch Inference Analysis

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. 批量推理历史
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}"

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 趋势）

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])}

# 上证指数 K 线（QuantDB index_daily）
curl -s -H "$AUTH" "$BASE/api/v1/market/kline?symbol=000001.SH&market=A&period=daily&days=30"

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只

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]

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)

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),
    }