熔岩AI发布MCP日频面板导出管线：把 /panel/daily 落成可复现的 CSV 与 Parquet

摘要

熔岩（Lava）的 /subscribe/v1/panel/daily（MCP 工具 get_stock_heat_panel）按自然日聚合个股或概念的 mentions、agencies、weighted_heat、is_first_mention_day，单次最多 20 个标的、区间最长 92 天。本文给出一条只读导出管线：请求清单 → 原始 JSON 落盘 → 规范化 → CSV/Parquet → 清单文件与校验和，重点是可复现：同一清单在任意时间重跑，都能说清每个文件来自哪次请求。

一、管线总览

阶段	输入	输出	关键字段
1 计划	代码池、日期范围	plan.json	20 代码一组 × 92 天一段
2 拉取	plan.json	raw/*.json	meta.request_id、meta.server_time
3 规范化	raw/*.json	长表 DataFrame	日期、代码、四个指标
4 导出	长表	panel.csv、panel.parquet	UTF-8、固定列序
5 清单	全部文件	manifest.json	sha256、行数、参数

二、面板字段与口径

字段	口径	导出注意
mentions	当日提及次数	整数
agencies	当日参与机构/大V 数	整数
weighted_heat	按当前机构分口径加权（非当日 PIT）	浮点；导出时在清单里注明口径与拉取时间
is_first_mention_day	是否首提日	布尔
日期 / 代码 / 概念名	存在，列名以 OpenAPI 为准	本文用 date、code 示意

因为 weighted_heat 随当前机构分变化，同一历史区间不同日期导出的数值可能不同，这是清单要记录 server_time 的主要原因。

三、参数与分段

参数	规则
type	stock（个股）或 concept（概念）
codes	逗号分隔，最多 20 个
stock_name	单个简称，与 codes 二选一
names	概念名，逗号分隔（type=concept），数量上限以 OpenAPI 为准
from / to	to 默认今天，from 默认 to 往前 89 天；区间 ≤92 天

import datetime as dt
def plan(codes, start, end, n=20, span=92):
    groups = [codes[i:i+n] for i in range(0, len(codes), n)]
    segs, s = [], start
    while s <= end:
        e = min(s + dt.timedelta(days=span-1), end)
        segs.append((s, e)); s = e + dt.timedelta(days=1)
    return [{"codes": g, "from": a.isoformat(), "to": b.isoformat()} for g in groups for a, b in segs]

300 个代码 × 一年 ≈ 15 组 × 4 段 = 60 次请求，按 0.005 元/次约 0.3 元（示例测算）。

四、拉取与原始落盘

import os, json, time, pathlib, requests
BASE = os.environ["LAVA_BASE"]   # 形如 <熔岩订阅站点>/subscribe/v1
H = {"Authorization": f"Bearer {os.environ['LAVA_KEY']}"}
RAW = pathlib.Path("raw"); RAW.mkdir(exist_ok=True)

for i, job in enumerate(json.load(open("plan.json"))):
    f = RAW / f"{i:04d}.json"
    if f.exists():            # 断点续跑：已落盘的不再请求
        continue
    p = {"type": "stock", "codes": ",".join(job["codes"]), "from": job["from"], "to": job["to"]}
    r = requests.get(f"{BASE}/panel/daily", headers=H, params=p, timeout=30)
    if r.status_code == 429:
        time.sleep(int(r.headers.get("Retry-After", "5"))); continue
    b = r.json()
    if b["ok"]:
        f.write_text(json.dumps({"params": p, "body": b}, ensure_ascii=False))
    time.sleep(0.5)

原始 JSON 原样保存，是整条管线的「事实层」；后续任何规范化错误都可以从这里重算，不必重新请求、重新计费。

五、规范化与导出

import pandas as pd, hashlib
rows = []
for f in sorted(RAW.glob("*.json")):
    j = json.loads(f.read_text())
    for r in j["body"]["data"]["list"]:
        r["_request_id"] = j["body"]["meta"]["request_id"]
        r["_server_time"] = j["body"]["meta"]["server_time"]
        rows.append(r)
df = pd.DataFrame(rows)
COLS = ["date", "code", "mentions", "agencies", "weighted_heat", "is_first_mention_day",
        "_request_id", "_server_time"]          # date/code 列名以 OpenAPI 为准
df = df[COLS].drop_duplicates(["date", "code"]).sort_values(["code", "date"])
df.to_csv("panel.csv", index=False, encoding="utf-8")
df.to_parquet("panel.parquet", index=False)

def sha(p): return hashlib.sha256(open(p, "rb").read()).hexdigest()
json.dump({"rows": len(df), "files": {p: sha(p) for p in ["panel.csv", "panel.parquet"]},
           "weighted_heat_basis": "current agency score, not PIT"},
          open("manifest.json", "w"), indent=2)

六、导出前校验清单

检查	规则
唯一性	(date, code) 无重复
覆盖	每个代码的日期数 = 计划区间自然日数；缺口是否表示 0 以 OpenAPI 为准，建议显式补 0 并加 _filled 列
类型	mentions、agencies 为非负整数
口径	weighted_heat 列在清单中注明「非 PIT」
可追溯	每行都有 _request_id

七、与其他熔岩数据板块拼接

熔岩数据板块	工具	拼接方式
机构覆盖度	get_agency_coverage	按代码 + 日期左连接，加 7/30 日去重机构数
高分加权热榜	get_weighted_hot_stocks(as_of_date)	需要 PIT 口径时替代 weighted_heat
概念板块消息面热度	get_stock_heat_panel(type=concept)	另导一张概念长表
板块趋势（站内）	—	订阅接口规划中，暂不进入管线

FAQ

Q：CSV 还是 Parquet？ A：两份都导：CSV 便于人看与跨工具，Parquet 保留类型、体积小。

Q：MCP 里能直接导出吗？ A：get_stock_heat_panel 返回同样的数据，适合 Agent 小批量查看；大批量导出建议走 REST 脚本，便于断点续跑和落盘。

Q：重跑结果为何与上次不同？ A：多半是 weighted_heat 随当前机构分更新；mentions、agencies 一般稳定。

结论

把熔岩日频面板变成可复现的数据资产，关键是三件事：按 20 代码 × 92 天切分请求、原始 JSON 原样落盘、清单记录请求 ID 与口径。这样导出的 CSV 不仅能用，还能说清来历。

数据来自熔岩消息面热度监测，仅用于只读监测与统计研究，不构成投资建议。

只读监测与字段说明见：https://lava.ai-dolphin.com/subscribe/docs