3. 数据采集与接入：数据源类型与实时流式接入

数据采集这块，说白了就是给异常检测系统「喂饭」。饭的质量好不好，直接决定了系统能不能干活。我见过太多项目，算法模型写得天花乱坠，结果数据源没搞好，上线第一天就崩了。

今天咱们重点聊三个核心数据源：订单簿、成交记录、账户信息。以及怎么把它们实时接入系统。

3.1 数据源类型：三大核心数据

异常交易检测，离不开这三类数据。我习惯把它们比作「交易的三要素」：谁在交易、交易了什么、交易结果如何。

3.1.1 订单簿（Order Book）

订单簿就是市场的「实时供需快照」。它记录了当前所有未成交的买单和卖单。

核心字段：

字段名	说明	示例值
symbol	交易对	BTC/USDT
bids	买单列表（价格, 数量）	[[50000, 1.5], [49990, 2.0]]
asks	卖单列表（价格, 数量）	[[50010, 0.8], [50020, 1.2]]
timestamp	快照时间戳	1710000000000

我在项目中遇到过一个问题：订单簿数据量极大，每秒可能更新上千次。如果全量推送，带宽和存储都扛不住。所以实际中常用「增量更新」——只推送变化的部分。

3.1.2 成交记录（Trade / Fill）

成交记录是「已经发生」的交易事实。它比订单簿更可靠，因为订单簿可能被撤单操纵，但成交记录是板上钉钉的。

核心字段：

字段名	说明	示例值
trade_id	成交唯一ID	123456789
price	成交价格	50005.5
quantity	成交数量	0.5
side	主动成交方向	buy / sell
timestamp	成交时间	1710000001000

你想想看，如果一个人频繁用「对倒」手法制造虚假成交量，成交记录里就会留下明显痕迹——同一价格、同一数量、买卖交替出现。这就是异常检测的典型特征。

3.1.3 账户信息（Account Info）

账户信息是「谁在交易」的画像数据。包括账户余额、持仓、历史交易行为等。

核心字段：

字段名	说明	示例值
account_id	账户ID	acc_001
balance	账户余额	100000.00
positions	持仓列表	[{"symbol":"BTC","qty":2.5}]
risk_level	风控等级	high

我个人习惯把账户信息作为「辅助特征」来用。比如，一个账户突然从低风险变成高风险，同时订单簿上出现大量异常挂单——这基本就是预警信号了。

3.2 实时流式接入：Kafka 实战

实时性要求高的场景，比如秒级检测，必须用流式接入。Kafka 是目前业界最主流的选择。

为什么选 Kafka？说白了，它有三个核心优势：

• 高吞吐： 单机每秒能处理几十万条消息
• 持久化： 消息落盘，挂了也能恢复
• 解耦： 数据生产者和消费者互不依赖

典型接入架构：

Kafka 生产者代码示例（Python）：

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# 发送订单簿增量数据
orderbook_update = {
    "symbol": "BTC/USDT",
    "bids": [[50000, 1.5]],  # 新增买单
    "asks": [],               # 无变化
    "timestamp": 1710000000000
}

producer.send('orderbook', value=orderbook_update)
producer.flush()

3.3 批量数据导入：历史数据回测

实时流式接入搞定了，但别忘了历史数据。做模型训练和回测时，我们需要批量导入过去几周甚至几个月的交易数据。

我常用的批量导入方式：

• CSV/Parquet 文件： 适合离线分析，用 Pandas 直接读取
• 数据库导出： 从 MySQL/PostgreSQL 导出到分析集群
• 对象存储： 从 S3/MinIO 拉取历史快照

批量导入示例（Python）：

import pandas as pd
from kafka import KafkaProducer

# 读取历史成交记录
df = pd.read_parquet('historical_trades.parquet')

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: v.encode('utf-8')
)

# 逐行发送到Kafka（模拟实时流）
for _, row in df.iterrows():
    message = row.to_json()
    producer.send('trades_historical', value=message)
    
producer.flush()
print(f"已导入 {len(df)} 条历史成交记录")

3.4 数据质量检查：别让脏数据毁了模型

数据接进来了，但质量怎么样？我见过最离谱的情况：某交易所的订单簿里，买单价格比卖单还高，明显是数据错乱。如果直接喂给模型，检测结果全是错的。

我常用的数据质量检查清单：

检查项	检查方法	处理方式
时间戳单调递增	比较前后两条记录的时间戳	丢弃乱序数据或重新排序
价格非负且合理	检查价格是否在 [0, 1e6] 范围内	标记异常或丢弃
买卖单不交叉	检查 bids 最高价 < asks 最低价	丢弃该快照并告警
数据完整性	检查必填字段是否缺失	补默认值或丢弃

嗯，这里要注意：数据质量检查最好放在 Kafka 消费者端做，而不是生产者端。因为生产者可能来自不同交易所，格式不统一，统一在消费者端清洗更可控。

好了，数据采集和接入这块就聊到这儿。记住一句话：数据质量决定了检测系统的天花板。把数据源管好了，后面的模型和策略才能发挥价值。

---