第四章：暗池数据源与获取
做暗池交易，最头疼的问题是什么？
不是策略，不是算法，而是——数据。
我刚开始接触暗池交易时，以为跟普通市场一样，拉个API就能拿到所有数据。结果呢？折腾了两周，发现连最基本的ATS交易量都拿不全。嗯，这里面的坑，我今天带你一个个踩一遍。
4.1 公开数据源：FINRA ATS数据
FINRA（美国金融业监管局）要求所有ATS（Alternative Trading System）每周披露交易数据。这是目前最权威的公开暗池数据源。
核心数据字段：
- ATS名称与ID
- 周交易量（股数）
- 周交易额（美元）
- 交易笔数
- 证券类型（股票、ETF等）
我个人习惯每周一早上拉取上一周的ATS数据。为什么？因为FINRA通常在周五收盘后更新，周末正好可以用来做数据清洗。
小技巧：FINRA官网提供CSV和XML两种格式。我建议用CSV，解析起来更顺手。XML虽然结构清晰，但处理速度慢不少。
4.2 SEC 605/606报告
SEC 605报告披露的是券商订单执行质量，606报告披露的是订单路由信息。这两份报告，说白了就是告诉你——你的订单到底去了哪里，执行得怎么样。
我曾经用606报告追踪过某大型券商的订单路由情况。发现它把大量小单送去了暗池，大单却留在交易所。为什么？因为暗池对小单的滑点影响小，大单进去容易被狙击。
| 报告类型 | 披露内容 | 更新频率 | 获取方式 |
|---|---|---|---|
| SEC 605 | 订单执行质量（价差、滑点、成交速度） | 季度 | SEC EDGAR系统 |
| SEC 606 | 订单路由去向（交易所、暗池、做市商） | 季度 | SEC EDGAR系统 |
注意：SEC报告有3个月的滞后性。你拿到的数据，反映的是3个月前的情况。做实时交易策略时，只能作为参考，不能作为决策依据。
4.3 商业数据源：Bloomberg与Reuters
公开数据虽然免费，但颗粒度太粗。你想看某个暗池的逐笔交易？对不起，FINRA不提供。这时候就得靠商业数据源了。
Bloomberg的ATS数据，我用了三年。说实话，贵是真贵，但值也是真值。它提供的是T+1的逐笔交易数据，包括每笔交易的时间戳、价格、数量、买卖方向。
Reuters（现属Refinitiv）的数据，我个人觉得在暗池覆盖面上不如Bloomberg全。但它有个好处——历史数据保存得更久，适合做回测。
商业数据源对比：
- Bloomberg：实时性强，覆盖约40家主要暗池，年费约2万美元起
- Reuters：历史数据全，覆盖约30家暗池，年费约1.5万美元起
- 其他选择：QuantHouse、BMLL等，各有侧重
4.4 数据API接入实战
理论说完了，咱们来点实际的。下面是我写的一个FINRA ATS数据获取脚本，用Python实现。
import requests
import pandas as pd
from datetime import datetime, timedelta
class FINRAATSDataFetcher:
def __init__(self):
self.base_url = "https://atsdata.finra.org/api/v1"
def fetch_weekly_data(self, week_end_date):
"""
获取指定周结束日期的ATS数据
"""
url = f"{self.base_url}/weekly/{week_end_date}"
params = {
"format": "csv",
"limit": 10000
}
try:
response = requests.get(url, params=params)
response.raise_for_status()
# 解析CSV数据
df = pd.read_csv(pd.StringIO(response.text))
return df
except requests.exceptions.RequestException as e:
print(f"数据获取失败: {e}")
return None
def fetch_multiple_weeks(self, start_date, end_date):
"""
批量获取多周数据
"""
all_data = []
current_date = start_date
while current_date <= end_date:
print(f"正在获取 {current_date} 的数据...")
df = self.fetch_weekly_data(current_date)
if df is not None:
all_data.append(df)
current_date += timedelta(days=7)
return pd.concat(all_data, ignore_index=True) if all_data else None
# 使用示例
fetcher = FINRAATSDataFetcher()
last_week = datetime.now() - timedelta(days=7)
data = fetcher.fetch_weekly_data(last_week.strftime("%Y-%m-%d"))
print(data.head())避坑指南：我曾经遇到过FINRA API限流的问题。同一个IP在1分钟内请求超过60次，就会被封24小时。解决办法？加个time.sleep(1)就好。
4.5 数据清洗与预处理
数据拿到手，别急着用。暗池数据有个特点——脏。为什么？因为不同数据源的格式、字段名、时间戳格式都不一样。
我总结了一套清洗流程，你直接拿去用：
- 去重：同一笔交易可能被多个数据源重复记录
- 时间对齐：把不同时区的时间统一到UTC
- 异常值处理：交易量为0或负数的，直接删除
- 缺失值填充：价格缺失的，用前一笔交易价格填充
- 格式统一：所有数值字段转为float，日期字段转为datetime
def clean_ats_data(df):
"""
清洗ATS数据
"""
# 1. 去重
df = df.drop_duplicates(subset=['trade_id'])
# 2. 时间对齐（假设原始数据是美东时间）
df['trade_time_utc'] = pd.to_datetime(df['trade_time_est']) + pd.Timedelta(hours=4)
# 3. 异常值处理
df = df[df['volume'] > 0]
df = df[df['price'] > 0]
# 4. 缺失值填充
df['price'] = df['price'].fillna(method='ffill')
# 5. 格式统一
df['volume'] = df['volume'].astype(float)
df['price'] = df['price'].astype(float)
return df
# 执行清洗
clean_data = clean_ats_data(raw_data)
print(f"清洗前数据量: {len(raw_data)}")
print(f"清洗后数据量: {len(clean_data)}")重要提醒：数据清洗时，千万别用全局填充。比如价格缺失，用整个数据集的均值填充，那你的策略就废了。一定要用时间序列的局部填充方法。
4.6 知识体系总览
下面这张图，是我做暗池数据获取时的完整流程。你照着这个框架走，基本不会漏掉关键环节。
这张图里，我特别想强调数据清洗层。很多人觉得数据清洗是体力活，随便搞搞就行。但我在实际项目中吃过亏——有一次因为时间戳没对齐，导致策略信号全部偏移了3秒，回测结果看起来很美，实盘一跑就亏钱。
所以，数据清洗这步，宁可慢一点，也要做扎实。