# 熔岩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 天 |

```python
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 元（示例测算）。

## 四、拉取与原始落盘

```python
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 原样保存，是整条管线的「事实层」；后续任何规范化错误都可以从这里重算，不必重新请求、重新计费。

## 五、规范化与导出

```python
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