数据源接入与格式统一
做行情数据清洗，第一步就是跟各种数据源打交道。说实话，这步要是没做好，后面所有工作都是白搭。我见过太多团队，清洗逻辑写得天花乱坠，结果源头数据格式就乱了，最后全得返工。
今天咱们就聊聊，怎么把那些乱七八糟的数据源，统一成一套好用的标准格式。
常见行情数据源
行情数据从哪来？无非就三个路子：交易所API、第三方数据商、还有各种CSV文件。每种我都踩过坑，一个一个说。
交易所API
交易所API是最直接的来源。像Binance、Coinbase、OKX这些大所，都提供REST和WebSocket接口。
我个人习惯，实时行情用WebSocket，历史数据用REST。为什么？WebSocket是长连接，数据推得快，适合做实时监控。REST是一次性请求，适合拉取历史K线。
关键点：不同交易所的API返回格式完全不同。比如Binance的K线数据字段名是openTime、open、high，而OKX用的是ts、o、h。你想想看，要是直接存，后面查询时得多痛苦。
我的经验：建议在接入层就做一次格式转换。我曾在项目里写了个统一的API适配器，每个交易所实现一个适配器类，输出统一的数据结构。这样上游再怎么变，下游都不受影响。
第三方数据商
第三方数据商像彭博、路透、Wind这些，数据质量高，但贵。而且他们的数据格式往往更复杂。
我记得有一次，接一个第三方数据商的FIX协议数据。那家伙，字段名全是数字编号，比如55代表股票代码，44代表价格。没有文档根本看不懂。
注意：第三方数据商的数据通常有延迟。别指望他们能给你毫秒级的数据。做回测还好，做高频交易就得慎重了。
CSV文件
CSV文件是最古老也最通用的格式。很多量化团队的历史数据，都是CSV格式存的。
但CSV有个大问题——没有schema。你拿到一个CSV，根本不知道里面有哪些列，每列是什么类型。我见过最离谱的，同一个CSV文件里，日期列有时候是2024-01-01，有时候是01/01/2024，还有时候是20240101。
我的建议：处理CSV时，一定要先做一次schema推断。用Pandas的read_csv时，指定dtype参数，别让它自动推断。自动推断有时候会翻车，比如把股票代码当成数字，前面的0就丢了。
数据格式标准化
数据源接进来了，接下来就是格式统一。常用的格式有三种：JSON、CSV、Parquet。各有各的适用场景。
| 格式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| JSON | 可读性好，结构灵活 | 体积大，解析慢 | API交互、小数据量 |
| CSV | 通用性强，工具支持多 | 无schema，类型信息丢失 | 数据交换、历史数据存储 |
| Parquet | 压缩率高，查询快 | 可读性差，工具支持有限 | 大数据分析、列式存储 |
我个人建议，实时数据用JSON，历史数据用Parquet。CSV作为中间交换格式，但别把它当主力存储。
为什么？JSON是API的标准格式，解析方便。Parquet是列式存储，做聚合查询时快得飞起。CSV嘛，说白了就是兼容性好，但性能不行。
JSON标准化
JSON格式的标准化，主要是统一字段名和数据类型。比如所有时间戳都用Unix毫秒，所有价格都用字符串（避免浮点精度问题）。
// 标准化前的JSON
{
"symbol": "BTCUSDT",
"price": 50000.12,
"time": "2024-01-01T00:00:00Z"
}
// 标准化后的JSON
{
"symbol": "BTCUSDT",
"price": "50000.12",
"timestamp": 1704067200000
}注意：价格用字符串存，不是矫情。浮点数在计算机里是近似值，0.1+0.2不等于0.3。做金融计算，一分钱都不能差。
CSV标准化
CSV标准化，核心是定义好列名和列顺序。我习惯用一套固定的列名模板，所有CSV都按这个模板输出。
# 标准CSV格式
symbol,timestamp,open,high,low,close,volume
BTCUSDT,1704067200000,50000.00,50100.00,49900.00,50050.00,1234.56我的经验：CSV文件最好加上表头。没有表头的CSV，处理起来就是噩梦。我曾经接手过一个项目，CSV没表头，全靠猜列含义，最后发现有两列顺序搞反了，回测结果全错。
Parquet标准化
Parquet是二进制格式，没法直接看。但它的schema是自描述的，存的时候是什么类型，读出来就是什么类型。
# 用PyArrow定义Parquet schema
import pyarrow as pa
schema = pa.schema([
pa.field("symbol", pa.string()),
pa.field("timestamp", pa.int64()),
pa.field("open", pa.decimal128(18, 8)),
pa.field("high", pa.decimal128(18, 8)),
pa.field("low", pa.decimal128(18, 8)),
pa.field("close", pa.decimal128(18, 8)),
pa.field("volume", pa.decimal128(18, 8))
])注意：Parquet的decimal类型，精度和标度一定要统一。我见过有人用decimal(38, 0)存价格，结果小数点后面全丢了。嗯，这坑我踩过。
字段映射与对齐
字段映射，说白了就是把不同数据源的字段名，映射到统一的字段名上。比如Binance的openTime，映射成timestamp。
字段对齐，是确保不同数据源的数据，在时间上是对齐的。比如两个交易所的1分钟K线，起始时间必须一致。
字段映射表
我习惯维护一张字段映射表，用YAML或者JSON存。这样新增数据源时，改配置就行，不用改代码。
# 字段映射配置
field_mapping:
binance:
openTime: timestamp
open: open
high: high
low: low
close: close
volume: volume
okx:
ts: timestamp
o: open
h: high
l: low
c: close
vol: volume我的建议：字段映射表最好版本控制。每次修改都记录原因和时间。这样出了问题，能快速回滚。
时间对齐
时间对齐是行情数据清洗里最头疼的问题。不同交易所的K线起始时间可能不一样，有的从整点开始，有的从整分开始。
我常用的方法是，先统一时间基准。比如所有数据都按UTC时间，然后做重采样。
# 用Pandas做时间对齐
import pandas as pd
# 假设df是原始数据，时间戳列是timestamp
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
# 重采样到1分钟K线
df_resampled = df.resample('1min').agg({
'open': 'first',
'high': 'max',
'low': 'min',
'close': 'last',
'volume': 'sum'
})关键点：重采样时要注意缺失值。如果某个时间窗口没有数据，是填前值还是填空？这取决于你的业务场景。做回测时，我一般填前值；做实时监控时，我填空。
知识体系总览
下面这张图，把数据源接入与格式统一的整体流程画出来了。你可以对照着看，心里有个谱。
你看，整个流程分四层：数据源层、格式标准化层、字段映射与对齐层、输出层。每一层解决一个核心问题。数据源层解决「从哪来」，格式标准化层解决「怎么存」，字段映射层解决「怎么对应」，输出层解决「怎么用」。
做这步时，别想着一步到位。先接一个数据源，跑通全流程，再慢慢加其他数据源。我见过太多人，一开始就想把所有数据源都接进来，结果搞了三个月还在对接。
最后说一句：数据源接入这块，慢就是快。把基础打牢了，后面清洗、分析、建模，都会顺畅很多。别问我怎么知道的，都是血泪教训。
公众号:蓝海资料掘金营，微信deep3321