逐笔交易数据API全解析：采集、清洗与量化分析实战

做股票数据API这个系列做到第三篇，前两篇分别写了基础实时行情和K线数据的获取方式，今天把"股票数据接口api"里最硬核的一类拿出来聊——股票当天逐笔交易数据。

逐笔交易数据是A股公开行情里颗粒度最细的一层，每一笔真实成交都被完整记录：成交时间、成交价格、成交量、主动买卖方向。它解决什么问题？一句话就能说清楚——普通分时图告诉你"这只股票今天从10块涨到10.5"，逐笔数据告诉你"9点31分07秒，10.01元，12000股主动买入，把价格从10.00顶到10.01"。这种细颗粒度数据是做盘口分析、短线复盘、量化策略回测和资金流分析绕不开的数据原料。

这篇内容适合两类人：第一类是已经会写点Python、想自己搭建数据管道的个人量化爱好者；第二类是天天看盘但觉得行情软件信息不够用、想拿原始数据做二次分析的朋友。文章不会只甩给你一个接口了事，我会把字段含义、分页逻辑、限流应对、去重落库、常见坑全部过一遍，你照着敲完基本就能把全天逐笔成交记录拉回本地。

1. 先搞清楚：逐笔成交和分时快照到底差在哪

1.1 快照是"跳着截图"，逐笔是"完整场记"

很多刚接触行情数据的朋友容易把"分时快照"和"逐笔成交"混为一谈，这两个概念差的不是一点点。

把行情比作拍电影：分时快照相当于每3秒钟截一张图，你看到的只是那一刻的最新价和买卖五档报价，两张图之间发生了什么你是完全不知道的。而逐笔成交相当于片场里的完整场记——无论这笔成交多大或多小，每一笔都会被记录下来。全天的成交过程用快照还原，中间有大段空白；用逐笔还原，每一帧都在。

具体到数据结构，逐笔成交记录的核心字段通常只有四类：

• 成交时间：精确到秒，付费L2接口可以到毫秒甚至更细
• 成交价格：这一笔实际撮合的价格
• 成交量：这一笔成交对应的股数（注意：不是手数，后面单独讲这个坑）
• 成交性质：主动买、主动卖、中性盘（集合竞价等）

这里的关键是"成交性质"。A股是撮合交易，一笔成交必然有买方和卖方，但"谁更着急"是通过成交价来体现的。如果这一笔是以卖一价甚至更高价成交的，说明买方主动扫单，标记为"主动买"（外盘）；如果是以买一价甚至更低价成交，说明卖方主动砸单，标记为"主动卖"（内盘）。这个方向信息是后续一切资金行为分析的基础。

1.2 用逐笔数据能还原哪些"看不见"的信息

没碰过逐笔数据的人可能觉得：不就是多了一点成交明细嘛，能有多大差别？我实际用下来，差别非常大。下面这几个场景是逐笔数据真正发光的地方：

第一，精确统计真实内外盘。 行情软件上显示的外盘内盘，很多是基于盘口快照或抽样估算的近似值，不是逐笔累加出来的。用逐笔数据可以一笔一笔累加，统计出当天真正主动买入了多少钱、主动卖出了多少钱，比软件上的数字更有参考价值。

第二，大单识别与异动预警。 把单笔成交金额超过一定阈值（比如100万元）的逐笔记录挑出来，按时间排列，你能清楚看到尾盘偷袭、盘中急拉急砸背后大资金的出手节奏。你还可以统计"大单活跃时段"，搞清楚这只股票的资金习惯是早盘干活还是尾盘干活。

第三，分钟级甚至秒级的放量早发现。 1分钟K线把60秒内的交易全部揉成一团，但一笔1万手的大单可能只花了几秒钟就打完了。逐笔数据可以让你看到30秒、15秒这种更细粒度的放量信号，对做T和抢反弹的人很有用。

第四，量化策略的信号口径修正。 很多日内策略（比如VWAP跟随、盘口异动策略）对撮合口径敏感，用分钟K线模拟出来的信号天然有偏差。逐笔数据能提供更接近真实撮合环境的回测原料。

1.3 免费API的边界：能做什么，不能做什么

先把丑话说在前面：免费的"逐笔成交"和券商Level-2的"逐笔成交流"不是同一个东西。免费接口拿到的是一天结束后（或略有延迟）的成交明细，记录粒度和更新频率都有限制，拿不到完整的委托队列、撤单记录和十档盘口。如果你要做的是真正的日内高频交易，请直接去开正规的L2行情，免费接口满足不了那种低延迟需求。

但如果你做的是收盘后复盘、日频策略回测、资金流向分析，免费逐笔接口完全够用，而且已经足够你跑出很多有意思的东西。我的建议是：先用免费接口把整套采集、清洗、落库、分析的流程跑通，等策略验证确实有稳定价值，再考虑升级付费数据源。这样既省钱，踩坑也踩得明白。

2. 数据源选型与整体方案设计

2.1 免费逐笔数据源横向对比

做逐笔数据，市面上能拿来直接用、不需要付费的渠道其实不多。我把主流方案都实测过一轮，结果整理成一张表：

这张表是我个人实测的体感结论，不是精确的官方文档，大家做选型时还要结合自己的场景。我最终选了新浪的逐笔成交接口作为主力，后面会说为什么。

2.2 为什么我最终选了这套方案

先说结论：我用的主接口是新浪的成交明细接口，选它的核心原因有四个。

第一，零门槛。不需要注册账号、不需要申请Token、不需要配置复杂的签名逻辑，直接构造HTTP请求就能拿数据。对个人开发者的"快速验证"场景来说，这太友好了。第二，字段完整。返回的JSON里直接给出了成交性质（主动买/主动卖/中性盘），省掉了我自己根据成交价和盘口去推断方向的麻烦，误差也小。第三，支持页码参数。做全量拉取、翻页抓取非常方便。第四，返回结构规整，解析成本低。

当然它也有明显短板：盘中数据有延迟，不是真·逐笔实时推送；高峰期请求频率高容易触发限流。所以我的定位很明确——它是"收盘后全量复盘"的好工具，不是"盘中抢单"的工具。想通了这一点，方案选型就不纠结了。

2.3 整体架构与更新策略

整个项目我做成了一条单向流水线，没有引入消息队列、微服务这些复杂东西，单机跑完全够用：

code复制定时任务触发 → 按股票分页拉取逐笔数据 → 解析清洗 → 标准化 DataFrame → 写入 SQLite → 生成分析结果

为什么落SQLite而不是CSV？因为一只股票一天就有几千上万条逐笔记录，一个月下来几十万条，CSV读起来越来越卡，而且增量更新、按股票和时间范围查询都很麻烦。SQLite单文件方案在个人项目里几乎是完美的：零部署、支持SQL、支持唯一约束去重、查询速度快。

更新策略上，我保守得多：每个交易日15:30跑一次全量抓取。为什么是15:30而不是15:00？因为15:00收盘后交易所的行情还要经过数据商清算和发布，太早去拉容易拿到不完整或仍在变动中的数据，等到15:30基本就稳定了。免费接口的定位是复盘工具，不是实时系统，所以没必要做分钟级轮询——既扛不住，也容易被封。

3. 核心代码全流程：抓取、清洗、落库、调度

3.1 读懂接口参数和返回字段

逐笔成交接口的请求地址和参数，基于我长期使用验证过的格式整理如下（各数据平台随时可能调整，拿到代码后先确认接口仍然有效）：

code复制https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketDataService.getTransactionData

关键参数有三个：

• symbol：股票代码，必须带市场前缀，比如 sz000001（深市）、sh600000（沪市）
• page：页码，从1开始翻
• num：每页条数，建议100~200，不要贪大

返回的JSON是一个数组，每条记录包含的核心字段如下表：

两个容易踩的细节，先给你提个醒：volume单位是"股"，不是行情软件上常见的"手"（1手=100股），后面统计时必须统一换算；kind字段在不同数据源里可能是中文（买盘/卖盘）也可能是英文（B/S/M），代码里建议做一层映射处理。

3.2 分页全量抓取：把当天所有成交翻完

写一个fetch_all_ticks函数，循环翻页直到取完。判断结束的条件我用了双保险：一是返回数组长度小于请求的page_size，说明已经到了最后一页；二是捕获到空数组或解析异常就自动退出。为什么这么谨慎？因为免费接口偶尔会出现"这一页空了但下一页还有数据"的边界情况，只看返回长度容易漏数据。

python复制import requests
import time
import json

BASE_URL = "https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/CN_MarketDataService.getTransactionData"

def fetch_all_ticks(symbol: str, base_url: str = BASE_URL,
                    page_size: int = 100, max_pages: int = 200,
                    sleep_sec: float = 0.3):
    """
    分页抓取某只股票当天的逐笔成交数据。
    symbol: 形如 'sz000001' / 'sh600000'
    返回: list[dict]
    """
    all_ticks = []
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": "https://finance.sina.com.cn/"
    }

    for page in range(1, max_pages + 1):
        params = {
            "symbol": symbol,
            "page": page,
            "num": page_size,
        }
        try:
            resp = requests.get(base_url, params=params, headers=headers, timeout=10)
            resp.raise_for_status()
            # 某些旧版接口返回带函数名包裹或前后括号的脏JSON，先剥壳
            text = resp.text.strip()
            if text.startswith("(") and text.endswith(")"):
                text = text[1:-1]
            data = json.loads(text)
        except Exception as e:
            print(f"[第{page}页] 请求异常: {e}")
            time.sleep(1.0)  # 容错：异常后稍等再继续下一页，不让整段流程挂掉
            continue

        if not isinstance(data, list) or len(data) == 0:
            break

        all_ticks.extend(data)
        print(f"[第{page}页] 获取 {len(data)} 条, 累计 {len(all_ticks)} 条")

        if len(data) < page_size:
            break
        time.sleep(sleep_sec)

    return all_ticks

if __name__ == "__main__":
    ticks = fetch_all_ticks("sz000001")
    print(f"合计获取 {len(ticks)} 条逐笔成交")
    if ticks:
        print(json.dumps(ticks[:3], ensure_ascii=False, indent=2))

几个细节动作解释一下。请求头里必须带Referer，这是我实战中遇到最多、也最容易被忽略的问题，不带它就等着收403吧。返回内容偶尔不是标准JSON，可能是JSONP风格或者前后多了括号，代码里做了简单的剥壳处理，防患于未然。每页之间的sleep间隔设在0.3秒，一天下来一只股票一般也就几百页，整体耗时可控，既不会因为太快被限流，也不会慢到让人打瞌睡。

3.3 数据清洗：类型、单位、顺序一个都不能错

原始接口返回的字段几乎全是字符串，直接拿去算均值、求和会出各种莫名其妙的问题。我的清洗流程固定四步：只留需要的字段、把数值转成float/int、补全symbol和date两列、按时间排序。

python复制import pandas as pd

def clean_ticks(raw_ticks: list, symbol: str, trade_date: str) -> pd.DataFrame:
    if not raw_ticks:
        return pd.DataFrame(columns=["symbol", "date", "time", "price", "volume", "amount", "kind"])

    df = pd.DataFrame(raw_ticks)

    # 补全股票代码和交易日期，方便多只股票合并查询
    df["symbol"] = symbol
    df["date"] = trade_date

    # 字符串 -> 数值
    df["price"] = pd.to_numeric(df["price"], errors="coerce")
    df["volume"] = pd.to_numeric(df["volume"], errors="coerce")
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

    # 丢弃关键字段缺失的脏数据
    df = df.dropna(subset=["price", "volume", "amount"])

    # 统一成交方向字段：兼容中文和英文写法
    kind_map = {"买盘": "buy", "卖盘": "sell", "中性盘": "neutral", "B": "buy", "S": "sell", "M": "neutral"}
    df["kind"] = df["kind"].map(kind_map).fillna("unknown")

    # 按时间排序，这是后续一切时间序列分析的前提
    df = df.sort_values("time").reset_index(drop=True)

    return df[["symbol", "date", "time", "price", "volume", "amount", "kind"]]

这里我特别强调按时间排序这一步。逐笔接口的返回顺序在跨页的时候不一定严格递增，偶尔会出现下一页的数据时间比上一页更早，不做排序，后面算累积量、画曲线全乱套。另外，方向字段做了中英文归一化，避免不同数据源返回格式不一致时程序直接崩掉。

3.4 落库SQLite：用唯一索引天然去重

表结构设计得比较简单，核心是给(symbol, date, time, price, volume)建唯一约束。为什么用这五列做唯一？因为同一个交易日同一只股票，理论上不可能出现完全相同的两笔成交（时间、价格、成交量都相同还重复的话，必然是重复请求造成的脏数据）。

sql复制CREATE TABLE IF NOT EXISTS tick_data (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    symbol TEXT NOT NULL,
    date TEXT NOT NULL,
    time TEXT NOT NULL,
    price REAL,
    volume INTEGER,
    amount REAL,
    kind TEXT,
    UNIQUE(symbol, date, time, price, volume)
);

写入时使用INSERT OR IGNORE语义，靠唯一约束把重复数据挡在门外。这个设计对"重复跑同一天的数据"这个场景特别友好——你不需要先查一遍哪些已经存在，直接跑抓取任务，重复记录自动被忽略。

python复制import sqlite3

DB_PATH = "stock_ticks.db"

def save_ticks_to_sqlite(df: pd.DataFrame, db_path: str = DB_PATH):
    if df is None or df.empty:
        return
    conn = sqlite3.connect(db_path)
    try:
        df.to_sql("tick_data", conn, if_exists="append", index=False)
        print(f"写入 {len(df)} 条到 SQLite")
    finally:
        conn.close()

查询的时候也方便，比如查某只股票某一天的逐笔数据，一行SQL就搞定：

python复制def load_ticks(symbol: str, trade_date: str, db_path: str = DB_PATH) -> pd.DataFrame:
    conn = sqlite3.connect(db_path)
    try:
        df = pd.read_sql_query(
            "SELECT * FROM tick_data WHERE symbol=? AND date=? ORDER BY time",
            conn, params=(symbol, trade_date)
        )
        return df
    finally:
        conn.close()

3.5 多股票批量抓取与定时调度

单只股票搞定了，多只股票批量抓取和每日自动调度其实是同一件事。我的方案很简单：用schedule库，周一至周五15:30依次遍历股票列表，逐只抓取。重点在于"依次"——不要开多线程并发去刷免费接口，把自己IP刷进黑名单就得不偿失了。

python复制import schedule
import time as time_module
from datetime import date

SYMBOLS = ["sz000001", "sh600000", "sz300750", "sh601318"]

def job():
    print(f"{date.today().isoformat()} 开始拉取当日逐笔数据...")
    for symbol in SYMBOLS:
        try:
            raw = fetch_all_ticks(symbol, page_size=200)
            df = clean_ticks(raw, symbol, date.today().isoformat())
            save_ticks_to_sqlite(df)
            # 每只股票之间主动间隔几秒，降低被限流概率
            time_module.sleep(2)
        except Exception as e:
            print(f"{symbol} 抓取失败: {e}")
    print("当日任务完成")

# 周一至周五 15:30 触发
schedule.every().monday.at("15:30").do(job)
schedule.every().tuesday.at("15:30").do(job)
schedule.every().wednesday.at("15:30").do(job)
schedule.every().thursday.at("15:30").do(job)
schedule.every().friday.at("15:30").do(job)

while True:
    schedule.run_pending()
    time_module.sleep(30)

如果你股票池比较大（比如几百只），建议把任务拆成多段：午盘14:50跑一部分，收盘后15:30跑一部分，既分散请求压力，也留出容错重试的时间窗口。

4. 拿到数据后能干什么：两个实战场景

4.1 场景一：统计真实内外盘，还原资金主动意愿

行情软件上的外盘内盘经常有人吐槽"对不上"，因为很多是基于盘口快照的估算。现在有了逐笔主动买卖方向，直接一笔一笔累加就行了，逻辑干净利落：

python复制def calculate_inout(df: pd.DataFrame) -> dict:
    buy = df[df["kind"] == "buy"]
    sell = df[df["kind"] == "sell"]
    buy_amount = buy["amount"].sum()
    sell_amount = sell["amount"].sum()
    total = df["amount"].sum()
    return {
        "主动买金额": round(buy_amount, 2),
        "主动卖金额": round(sell_amount, 2),
        "总成交金额": round(total, 2),
        "主动买占比": round(buy_amount / total * 100, 2) if total else 0,
        "主动卖占比": round(sell_amount / total * 100, 2) if total else 0,
    }

df = load_ticks("sz000001", "2026-01-15")
print(calculate_inout(df))

这里必须提醒一个容易误判的点：主动买占比高并不等于股价一定涨。如果主动买集中在股价已经大幅拉升后的高位，反而可能是散户在接盘；主动卖占比高也不等于一定跌，有时是主力在低位故意砸盘吸筹。逐笔数据是"显微镜"，能把动作看得很清楚，但判断方向背后还需要成交量位置、大盘环境、板块联动综合来看。

4.2 场景二：大单识别与异动时段定位

把单笔成交金额超过阈值的记录筛出来，按时间窗口聚合，就能快速定位盘中大单出现的时段。阈值怎么定才合理？固定绝对值（比如100万）会失真——小盘股50万就算大单，大盘股500万可能平平无奇。我一般用动态阈值：单笔金额超过当天平均单笔成交额的5倍，就算异常大单。

python复制def detect_large_trades(df: pd.DataFrame, multiplier: float = 5.0) -> pd.DataFrame:
    avg_amount = df["amount"].mean()
    threshold = avg_amount * multiplier
    large = df[df["amount"] >= threshold].copy()
    print(f"平均单笔成交额: {avg_amount:.2f} 元")
    print(f"动态阈值: {threshold:.2f} 元")
    print(f"识别大单 {len(large)} 笔, 合计金额 {large['amount'].sum() / 10000:.2f} 万元")
    return large.sort_values("amount", ascending=False)

large_trades = detect_large_trades(df)
print(large_trades.head(10))

大单识别的核心价值是与时间配合，看"谁在什么时刻干活"。我复盘时重点关注三类：早盘竞价前后突然出现的大单（资金抢筹或出逃）、尾盘最后10分钟连续放量的大单（隔夜筹码交换）、同一价格区间短时间内反复出现的大单（有人在偷偷护盘或压盘）。

但要注意：逐笔成交不包含委托人的任何身份信息，大单可能是机构调仓、量化拆单、也可能是多个散户同时买入的巧合，不能一上来就扣"主力操作"的帽子。它只是帮你发现值得关注的信号，不是结论本身。

4.3 复盘模式：自定义分时聚合，看到软件看不到的细节

默认的1分钟K线会把60秒内的所有交易揉成一团，但很多关键异动其实发生在更短的时间窗内。有了逐笔数据，可以按任意粒度聚合自定义分时，我一般习惯看30秒：

python复制# 拼完整时间戳并用它做索引
df["dt"] = pd.to_datetime(df["date"] + " " + df["time"])
df.set_index("dt", inplace=True)

# 按30秒聚合：价格取最后一笔，成交量成交额求和
agg = df.resample("30S").agg(
    price=("price", "last"),
    volume=("volume", "sum"),
    amount=("amount", "sum")
)
print(agg.head(10))

这种自定义分时对做T（日内回转）的朋友特别有用：30秒级别的放量信号比1分钟级别更前置，复盘时能更准确地定位到自己当时的买卖点踩在什么位置。你也可以把聚合结果结合大单识别，做一张"30秒成交量+大单叠加图"，盘中异动几乎一目了然。

5. 实测中的坑与排查技巧实录

5.1 请求被拒绝或返回403，多半是请求头的问题

我调试过程中遇到最多的就是403。最常见原因有两个：缺Referer头、请求频率太快。解决方案也简单：请求头里老老实实带上User-Agent和Referer，每页请求间隔保持在0.3秒以上。如果已经被临时封了，别硬刚，停个三五分钟再继续。

另外提醒一下，免费接口不是生产环境，别在上面跑大规模并发采集。我的原则是：自用、低频、克制，这是保证接口长期可用的默契。

5.2 返回数据有缺口或重复，完整性校验不能省

免费接口偶尔会丢页、跳号，这是数据源的特性，不是你的代码有问题。我的标准做法有三步：

• 入库时靠UNIQUE约束自动去重；
• 抓取完成后跑一个"分钟覆盖检查"，检查当天交易时段每个分钟是否都有成交记录，没有的分钟记录下来；
• 发现某只股票有大量分钟缺口，就直接重跑一次全量抓取，INSERT OR IGNORE会自动补齐缺失部分。

python复制def check_minute_coverage(df: pd.DataFrame) -> list:
    minutes = set(df["time"].str[:5])
    missing = []
    # 早盘 09:30 - 11:30
    for m in range(9 * 60 + 30, 11 * 60 + 30):
        hh, mm = divmod(m, 60)
        key = f"{hh:02d}:{mm:02d}"
        if key not in minutes:
            missing.append(key)
    # 午盘 13:00 - 15:00
    for m in range(13 * 60, 15 * 60 + 1):
        hh, mm = divmod(m, 60)
        key = f"{hh:02d}:{mm:02d}"
        if key not in minutes:
            missing.append(key)
    return missing

missing_minutes = check_minute_coverage(df)
if missing_minutes:
    print(f"存在 {len(missing_minutes)} 个分钟缺口，建议重跑该股票")

这个检查只是粗筛，逐笔数据的秒级连续性用免费接口几乎不可能做到完美，能保证分钟级别的覆盖已经算及格。太吹毛求疵就得上付费L2了。

5.3 单位陷阱：股、手、金额必须统一

这个坑必须单独拿出来说。新浪逐笔接口返回的volume单位是"股"，但如果你拿它去和某些显示"手"的行情软件对账，会发现差100倍。我第一次统计全天空方总量时被这个坑带偏过，排查了半天才发现是单位问题。解决方案：入库统一用"股"，展示成"手"就除以100。表结构注释里把单位写死，别指望着靠"我记得"来换算。

5.4 分页边界：最后一页的判断别太死板

做翻页抓取时，判断"是否最后一页"看似简单，实际容易出错。有的接口返回空数组表示没数据了，有的返回长度小于page_size，还有的会返回跟上一页完全重复的数据。我实测遇到的情况是：偶尔某一页返回空，但下一页又有内容。所以代码里我除了判断len(data) < page_size之外，还加了max_pages兜底上限，防止极端情况下无限循环。如果抓到重复数据也不用慌，落库时的唯一约束会兜底。

5.5 数据使用边界与合规提示

最后聊几句关于数据使用的个人建议。逐笔成交数据属于公开行情数据，个人用于学习、研究和复盘完全没问题，但有三点要守住：第一，别把采集程序包装成商业数据服务对外销售，数据源的规则和使用边界要看清；第二，采集频率务必克制，别把免费接口当生产环境用，这也是我把更新频率卡在每天一次的原因；第三，任何基于这些数据的判断都只能是参考，逐笔数据帮你把交易过程看清楚，但它不负责预测未来，投资决策还得靠自己的分析和判断，风险自负。

做这个项目我最大的体会是：逐笔数据的价值不在数据本身，而在你愿意花多少心思去清洗和挖掘。免费接口拿回来的原始JSON字段乱、单位混、偶有缺失，但一旦把整条链路跑通，后面无论换什么数据源，都只是换一个API地址的问题，架构完全可以复用。我当初从逐笔数据切入做盘口分析，在字段单位、分页边界、请求头拦截上各踩了一次坑，每解决一个，对行情数据的理解就深一层。想玩转股票数据API，别急着追L2和高大上的量价模型，先从全天的逐笔成交开始，把最底层的数据链路跑通，后面所有分析都会顺很多。