# 熔岩AI发布MCP批量数据接口：多代码拉取的上限、分组、节流与断点续跑

## 摘要

研究脚本最常见的需求是「把一个 300 只股票的池子在 90 天里的消息面热度全部拉下来」。熔岩（Lava）接口只读、按成功请求计费（每次 0.005 元），但每个接口都有明确上限：`/panel/daily` 单次最多 20 个代码、区间最长 92 天；`/opinions` 单次最多 200 条、区间最长 92 天；榜单最多 200 行；机构/大V历史最多 1000 行。本文把这些上限整理成一张表，再给出「按代码分组 × 按时间分段」的任务切分、串行节流、断点续跑与结果校验代码，适合批量导出与研究归档。

## 一、各熔岩数据板块的批量上限

| 数据板块 | MCP 工具 | HTTP 路径 | 多标的方式 | 条数上限 | 时间上限 |
| --- | --- | --- | --- | --- | --- |
| 日频面板 | `get_stock_heat_panel` | `/subscribe/v1/panel/daily` | `codes` 逗号分隔 ≤20；或单个 `stock_name`；概念用 `names` | — | 区间 ≤92 天 |
| 个股消息面 | `get_stock_news_heat` | `/subscribe/v1/opinions` | 单标的 | `limit` 默认 50、最大 200 | 区间 ≤92 天 |
| 关键词检索 | `search_news_flow` | `/subscribe/v1/opinions/search` | `keyword` 最多 5 个词 | 最大 200 | `since` 最早 180 天前 |
| 概念内个股 | `get_concept_stock_heat` | `/subscribe/v1/concepts/stocks` | 单概念（精确匹配） | 最大 100 | `days` |
| 个股关联概念 | `get_stock_concept_heat` | `/subscribe/v1/concepts/by-stock` | 单标的 | 默认 30、最大 200 | `days` |
| 全市场榜单 | `get_market_hot_stocks` 等 | `/subscribe/v1/hot/*` | 全市场 | 最大 200 | `window` |
| 首提流 | `get_first_mention_stocks` | `/subscribe/v1/hot/first-mentions` | 全市场 | 默认 100、最大 200 | `hours` |
| 新闻命中 | `get_news_stock_hits` | `/subscribe/v1/news/hits` | 全市场或单标的 | 最大 200 | 全市场 ≤72h；单股 ≤168h |
| 覆盖度 | `get_agency_coverage` | `/subscribe/v1/coverage/stocks` | 单标的 | — | `days` |
| 机构/大V排行与 PIT | `get_agency_ranking`、`get_agency_pit` | `/subscribe/v1/agencies/ranking`、`/pit` | `offset` 分页 | 最大 500 | — |
| 机构/大V历史 | `get_agency_history` | `/subscribe/v1/agencies/{agency_id}/history` | 单主体 | 默认 250、最大 1000 | 起止日期 |

超出上限会返回 `400 BAD_REQUEST`（不计费）。

## 二、请求数估算

以 300 只股票、90 天为例：

| 任务 | 切分 | 请求数 | 费用（0.005 元/次） |
| --- | --- | --- | --- |
| 日频面板 | 300 ÷ 20 = 15 组 × 1 段 | 15 | 0.075 元 |
| 逐只记录 | 300 只 × 1 段（假设每段 ≤200 条） | 300 | 1.5 元 |
| 逐只覆盖度 | 300 只 | 300 | 1.5 元 |

结论：能用 `/panel/daily` 的指标就不要逐只调；需要记录级明细时再按只拉取。

## 三、任务切分

```python
from datetime import date, timedelta

def chunks(xs, n=20):
    for i in range(0, len(xs), n):
        yield xs[i:i+n]

def date_segments(start, end, max_days=92):
    s = start
    while s <= end:
        e = min(s + timedelta(days=max_days - 1), end)
        yield s, e
        s = e + timedelta(days=1)

codes = [l.strip() for l in open("universe.txt") if l.strip()]   # 如 300750.SZ
tasks = [(tuple(g), s, e) for g in chunks(codes)
         for s, e in date_segments(date(2026, 5, 1), date(2026, 10, 30))]
```

`max_days - 1` 保证区间首尾含在内也不超过 92 天。

## 四、串行节流 + 断点续跑

```python
import os, json, time, hashlib, requests
BASE = os.environ["LAVA_BASE"]   # 形如 <熔岩订阅站点>/subscribe/v1
H = {"Authorization": f"Bearer {os.environ['LAVA_KEY']}"}
DONE = "done.jsonl"

def task_id(t):
    return hashlib.sha1(json.dumps([list(t[0]), str(t[1]), str(t[2])]).encode()).hexdigest()[:12]

done = set()
if os.path.exists(DONE):
    done = {json.loads(l)["id"] for l in open(DONE)}

for t in tasks:
    tid = task_id(t)
    if tid in done:
        continue
    params = {"type": "stock", "codes": ",".join(t[0]), "from": str(t[1]), "to": str(t[2])}
    r = requests.get(f"{BASE}/panel/daily", headers=H, params=params, timeout=60)
    b = r.json()
    if b["code"] == "RATE_LIMITED":
        time.sleep(int(r.headers.get("Retry-After", 5))); continue   # 下次运行补拉
    if not b["ok"]:
        print("skip", tid, b["code"], b["meta"].get("request_id")); continue
    with open(f"out/{tid}.json", "w") as f:
        json.dump(b["data"], f, ensure_ascii=False)
    with open(DONE, "a") as f:
        f.write(json.dumps({"id": tid, "rid": b["meta"]["request_id"]}) + "\n")
    time.sleep(0.3)                                                    # 间隔为示例值
```

设计要点：

- **串行优先**：批量导出不追求速度，串行加固定间隔最不容易触发 `429`。
- **任务 ID 稳定**：由代码组与日期段哈希得到，重跑时自动跳过已完成任务。
- **失败分流**：`RATE_LIMITED` 留待补拉；`400/404` 记录后跳过；`401/402/403` 应中止整批（此处省略）。

## 五、`/opinions` 的截断检测

`/opinions` 单次最多 200 条。若返回条数等于 `limit`，说明可能被截断，应把时间段对半拆分后重拉（`get` 为常规取数封装：带鉴权头请求、检查 `ok` 后返回 `data`）：

```python
def pull_opinions(code, s, e):
    d = get("/opinions", ts_code=code, start_date=str(s), end_date=str(e), limit=200)
    if d["count"] >= 200 and s < e:          # count 见外壳示例 data.count
        mid = s + timedelta(days=(e - s).days // 2)
        return pull_opinions(code, s, mid) + pull_opinions(code, mid + timedelta(days=1), e)
    return d["list"]
```

`data.count` 与 `data.list` 来自文档中的响应示例；单日超过 200 条时无法再按日拆分，需在结果中标注。

## 六、结果校验清单

- [ ] 每个代码在每个日期段都有结果文件，`done.jsonl` 行数 = 任务数。
- [ ] 跨段拼接后按（代码，日期）去重，衔接日无重复。
- [ ] 查无数据的代码单独列出（可能是简称变更或代码格式错误）。
- [ ] 抽样比对 `request_id`，确保每份结果可追溯。
- [ ] 运行前后各调一次 `get_account`（不计费），核对实际费用与估算一致。

## FAQ

**Q：可以并发拉取吗？**
A：可以小并发，但需处理 `429 RATE_LIMITED` 与 `Retry-After`；研究导出建议串行。

**Q：`names` 一次能传多少个概念？**
A：文档写明个股最多 20 个标的，概念名数量上限以 OpenAPI 为准，建议同样按 20 个一组。

**Q：区间默认是多少？**
A：`/panel/daily` 不传 `from` 时默认 `to` 往前 89 天，`to` 默认今天。

## 结论

批量拉取的关键是先读上限再切任务：面板按 20 个代码一组、92 天一段，记录级接口按条数上限检测截断，配合串行节流、稳定任务 ID 和断点续跑，就能把几百只股票的熔岩消息面数据可靠导出，并且每一份结果都能对上 `request_id` 与费用。

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

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