>

大宗交易数据采集实战：用 API 自动获取沪深大宗成交明细

Reading Time: < 1 minute

大宗交易数据采集实战：用 API 自动获取沪深大宗成交明细

免责声明：仅供参考，不构成投资建议。本文所有代码与数据仅用于技术演示与个人研究，请遵守数据源的使用条款与相关法律法规。

一、先说清楚问题：为什么大宗交易数据这么难拿

做量化研究、写复盘脚本或者只是想搭一个自选股的数据看板，很多同学早晚都会碰到「大宗交易」这个数据。它反映的是场外一次性成交、金额较大的那部分交易明细，通常包含成交日期、股票、成交价、成交量、买卖营业部等字段。

问题在于：这类数据没有一个官方开放的、稳定的 JSON 接口给你直接调。沪深交易所网站上确实有公开页面，但对开发者来说很不友好：

  • 页面是分页渲染的，翻页参数经常变；
  • 有的板块是 JavaScript 动态加载，直接 requests.get() 拿到的是空壳 HTML；
  • 偶尔还带风控，短时间多请求就会被限流；
  • 字段命名、日期格式在不同板块之间不统一，清洗起来很烦。

于是很多人退而求其次，每天手动打开网页、复制粘贴到 Excel，或者写一个脆弱的爬虫，隔三差五就因为页面改版而挂掉。

二、手动方式到底有多痛

我自己早期就是纯手动。流程大概是：

  1. 打开交易所大宗交易查询页；
  2. 选好日期、市场；
  3. 一页一页翻，Ctrl+C / Ctrl+V 到表格；
  4. 手动把「金额」列里的逗号、单位统一掉；
  5. 存成 CSV，再导入分析脚本。

一天一次还能忍，一旦你想回补最近三个月、或者想每天收盘后自动跑一遍，手动就完全不现实了。写自研爬虫呢？维护成本又高——页面一改版，选择器全废，你还得重新 F12 找 DOM。

对大多数人来说，真正需要的其实是一个稳定输入、干净输出的接口：我给你市场和金额门槛，你还我结构化的成交明细列表。

三、方案：用现成的 Actor 一行调用拿结果

这里用 Apify 上的 china-block-trades-tracker 这个 Actor。它把「翻页、渲染、清洗」都封装在服务端，你只需要传三个参数：

  • market：市场，取值为 all / shanghai / shenzhen / beijing；
  • minDealAmountYuan：成交金额下限（单位：元），用来过滤掉小额，只看大额成交；
  • maxResults：最多返回多少条。

Actor 页面：https://apify.com/nexgendata/china-block-trades-tracker?fpr=2ayu9b

3.1 最小可运行示例（run-sync 同步调用）

Apify 提供了 run-sync-get-dataset-items 端点，一次 HTTP 请求就能同步跑完 Actor 并直接拿到数据集里的条目，非常适合脚本化使用。下面是 Python 版本：

import requests
# 把占位符换成你自己的 Apify token
APIFY_TOKEN = "<你的-apify-token>"
ACTOR_SLUG = "nexgendata~china-block-trades-tracker"
ENDPOINT = (
    f"https://api.apify.com/v2/acts/{ACTOR_SLUG}"
    f"/run-sync-get-dataset-items?token={APIFY_TOKEN}"
)
# 字段名以 Actor 页面的输入说明为准
payload = {
    "market": "all",               # 目标市场：all / shanghai / shenzhen / beijing
    "minDealAmountYuan": 5000000,  # 只看成交金额 >= 500 万元的
    "maxResults": 100,
}
resp = requests.post(ENDPOINT, json=payload, timeout=180)
resp.raise_for_status()
items = resp.json()
print(f"共获取 {len(items)} 条大宗交易记录")
# 打印前几条看看结构
for row in items[:3]:
    print(row)

这段代码做的事很直接：POST 输入 JSON，服务端跑完 Actor，响应体就是数据集条目组成的数组。

3.2 诚实说明输出

需要坦白讲：这个 Actor 没有对外声明固定的输出字段 schema。所以正确的做法不是硬编码字段名，而是先跑一次、用代码探测实际返回的键，再据此清洗。这样即使数据源结构调整，你的脚本也更稳。

import csv
if not items:
    print("本次没有返回数据，检查一下日期或金额门槛是否过严")
else:
    # 动态收集所有出现过的字段名，避免假设字段
    all_keys = set()
    for row in items:
        all_keys.update(row.keys())
    fieldnames = sorted(all_keys)
    print("实际返回的字段：", fieldnames)
    # 写入 CSV，缺失字段留空
    with open("block_trades.csv", "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for row in items:
            writer.writerow(row)
    print("已保存到 block_trades.csv")

用 utf-8-sig 编码是为了让 Excel 直接双击打开时中文不乱码，这是个小但实用的细节。

四、定时与自动化：收盘后自动跑一遍

真正省事的地方是自动化。你可以在 Apify 平台上给这个 Actor 配置 Schedule，让它每个交易日固定时间自动运行；也可以在自己的服务器上用 cron 触发上面的脚本。

一个简单的本地定时思路（Linux crontab，工作日 16:30 执行）：

30 16 * * 1-5 /usr/bin/python3 /home/you/jobs/fetch_block_trades.py >> /home/you/logs/block.log 2>&1

再配合一点增量逻辑，比如把每天的结果按日期落到数据库、跑之前先查一下当天是否已抓取，就能积累出一份自己的历史大宗交易明细库，供后续做统计分析。

import datetime
today = datetime.date.today().isoformat()
outfile = f"block_trades_{today}.csv"
# ...沿用上面的抓取与写入逻辑，文件名带上日期即可实现按日归档

五、小结

把「大宗交易数据采集」这件事拆开看，痛点从来不是分析，而是稳定地拿到干净数据。用 china-block-trades-tracker 这类现成 Actor，好处是把翻页、渲染、限流这些脏活交给服务端，你只管传 market / minDealAmountYuan / maxResults 三个参数，拿回结构化条目，剩下的精力全花在你真正关心的研究上。

如果你想看更多面向 APAC 市场的数据采集 Actor 与实践，可以逛一逛 https://thenextgennexus.com 。按需取用即可，欢迎在评论区交流你踩过的坑。

再次提醒：仅供参考，不构成投资建议。