3、金融数据采集Agent：数据源分类与实战
金融数据采集，说白了就是给Agent喂饭。饭好不好吃，新不新鲜，直接决定了后面所有策略的成败。我见过太多团队，模型调得再漂亮，数据源一塌糊涂，最后全白搭。
今天咱们就聊聊，怎么把这顿饭做好。
3.1 数据源分类：行情、财报、新闻
金融数据源，我习惯分成三大类。每一类的脾气秉性都不一样，采集策略也完全不同。
| 数据类别 | 典型数据 | 更新频率 | 采集难点 |
|---|---|---|---|
| 行情数据 | 股票价格、成交量、盘口深度 | 毫秒级～秒级 | 高并发、低延迟、断线重连 |
| 财报数据 | 营收、净利润、资产负债表 | 季度/年度 | 格式不统一、PDF解析困难 |
| 新闻舆情 | 公告、研报、社交媒体 | 分钟级～小时级 | 噪音大、时效性要求高 |
行情数据，讲究一个「快」。我记得有一次做量化回测，用的延迟了3秒的行情数据，结果策略在模拟盘上赚得飞起，实盘一跑就亏。后来才发现，3秒的延迟在T+0市场里，足够让滑点吃掉所有利润。
财报数据，讲究一个「准」。你想想看，一个数字错了，整个估值模型就偏了。我踩过最大的坑，是某家公司的财报里「净利润」和「归母净利润」混用，直接导致因子计算错误。嗯，这里要注意，一定要看清楚数据字典。
新闻舆情，讲究一个「全」。但全不代表好。我曾经抓了某平台所有财经新闻，结果80%都是垃圾广告。后来加了关键词过滤和来源白名单，才把信噪比提上来。
核心原则：行情要快，财报要准，新闻要全。三者不可偏废。
3.2 爬虫与API设计
采集数据，无非两条路：要么用别人提供的API，要么自己写爬虫。
API优先，爬虫兜底，这是我的原则。为什么？API稳定、规范、有文档。爬虫呢？今天网站改个版，明天加个反爬，后天IP被封，运维成本极高。
但现实是，很多数据根本没有API。比如某些交易所的深度数据，或者小众公司的财报。这时候，就得自己动手了。
我分享一个我常用的爬虫框架设计思路：
# 伪代码：金融数据爬虫基类
class FinancialCrawler:
def __init__(self, source_name):
self.source = source_name
self.session = self._init_session()
self.retry_count = 3
def fetch(self, url, params=None):
for i in range(self.retry_count):
try:
resp = self.session.get(url, params=params, timeout=10)
if resp.status_code == 200:
return self._parse(resp.text)
elif resp.status_code == 429:
time.sleep(2 ** i) # 指数退避
else:
self._log_error(f"HTTP {resp.status_code}")
except Exception as e:
self._log_error(str(e))
return None
def _parse(self, html):
# 子类实现具体解析逻辑
raise NotImplementedError
这个框架的核心就两点：重试机制和错误日志。我曾经因为没加重试，一个爬虫跑了三天，结果第三天网站挂了5分钟，数据就断了。后来加了重试和告警，再也没出过类似问题。
小技巧：对于高频行情API，建议使用WebSocket而非HTTP轮询。WebSocket是长连接，延迟低，而且节省带宽。我实测过，同样采集100只股票的tick数据，WebSocket的延迟比HTTP轮询低了一个数量级。
3.3 数据清洗与标准化
数据采回来，只是第一步。真正的噩梦，从清洗开始。
我见过最离谱的情况：同一只股票，在A平台叫「贵州茅台」，在B平台叫「贵州茅台酒股份有限公司」，在C平台叫「600519.SH」。三个名字，你让Agent怎么识别？
所以，标准化是必须的。我一般做三件事：
- 字段统一：所有日期统一成YYYY-MM-DD格式，所有价格统一成浮点数，所有股票代码统一成「市场+代码」格式（如SH600519）。
- 缺失值处理：行情数据缺失，用前值填充；财报数据缺失，标记为NaN，不填充。为什么？财报数据缺失往往意味着信息不对称，填充反而会引入偏差。
- 异常值检测：比如某股票突然涨了1000%，大概率是数据错误。我会用3-sigma法则或者IQR方法过滤掉这些异常点。
# 数据清洗示例
import pandas as pd
def clean_market_data(df):
# 1. 去重
df = df.drop_duplicates(subset=['code', 'timestamp'])
# 2. 排序
df = df.sort_values(['code', 'timestamp'])
# 3. 缺失值填充（前向填充）
df['close'] = df.groupby('code')['close'].ffill()
# 4. 异常值过滤（涨跌幅超过20%视为异常）
df['pct_change'] = df.groupby('code')['close'].pct_change()
df = df[df['pct_change'].abs() < 0.2]
return df
避坑指南：我曾经在清洗财报数据时，直接删除了所有包含NaN的行。结果发现，那些被删除的行里，有一半是因为「研发费用」字段为空——而很多科技公司根本不披露研发费用。这一删，样本量直接少了一半，模型也废了。所以，清洗前一定要先分析缺失模式。
3.4 定时任务调度
数据采集不是一次性的，而是持续性的。定时任务调度，就是让Agent按点干活。
我常用的调度策略分三种：
| 数据类别 | 调度频率 | 推荐工具 |
|---|---|---|
| 行情数据 | 交易时段持续采集 | APScheduler + WebSocket |
| 财报数据 | 每天凌晨检查一次 | Crontab + 增量爬虫 |
| 新闻舆情 | 每10分钟轮询一次 | Celery + Redis |
我个人习惯用 APScheduler，因为它轻量、灵活，而且支持多种触发器。比如行情数据用cron触发器，只在交易时段运行；财报数据用interval触发器，每天跑一次。
# APScheduler 调度示例
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
# 行情采集：交易时段每5秒一次
@scheduler.scheduled_job('cron', day_of_week='mon-fri', hour='9-15', minute='*', second='*/5')
def fetch_market_data():
print("采集行情数据...")
# 实际调用采集逻辑
# 财报采集：每天凌晨2点
@scheduler.scheduled_job('cron', hour=2, minute=0)
def fetch_financial_reports():
print("采集财报数据...")
# 实际调用采集逻辑
scheduler.start()
这里有个坑要注意：任务超时。我曾经有个财报采集任务，因为网络波动跑了2个小时，结果和下一个任务冲突了，导致数据重复。后来我加了超时控制，每个任务最多跑30分钟，超时就强制终止。
总结一下：数据采集Agent的核心，就是「快、准、全」三个字。快靠WebSocket和合理调度，准靠清洗和标准化，全靠多源融合和增量更新。把这三点做好，你的Agent就成功了一半。
这张图把整个流程串起来了。从数据源分类，到采集方式选择，再到清洗标准化，最后通过定时调度持续输出。每一步都有坑，但每一步也都有解法。
好了，数据采集这块就聊到这儿。记住一句话：数据质量决定了Agent的天花板。后面咱们再聊怎么用这些数据做策略分析。