3. 投研数据采集与清洗：数据源分类与实战流程

做量化投研，第一步就是搞数据。我见过太多人把精力全花在策略模型上，结果数据源一塌糊涂，回测跑得飞起，实盘直接翻车。说白了，数据质量决定了你的策略天花板。

今天咱们就聊聊数据采集与清洗。我会把数据源分类、Python爬虫与API采集、以及清洗标准化的流程，掰开揉碎了讲清楚。

3.1 数据源分类：结构化、半结构化、非结构化

数据源分三类，这个分类不是学术概念，而是直接决定了你怎么去拿数据、怎么存、怎么用。

类型	特点	典型例子	采集方式
结构化数据	行和列清晰，字段固定	MySQL表、CSV文件、Wind导出的日线数据	SQL查询、API接口、直接读取文件
半结构化数据	有标签或标记，但结构不固定	JSON、XML、HTML页面	解析器（json、BeautifulSoup）、API
非结构化数据	没有预定义的数据模型	PDF研报、新闻文本、音频、图片	OCR、NLP、爬虫+文本提取

我个人习惯，先把数据源按这个分类贴标签。为什么？因为清洗策略完全不同。比如结构化数据，你主要处理缺失值和异常值；而非结构化数据，你得先做实体识别和关系抽取。

3.2 Python爬虫与API采集实战

采集数据，我一般优先走API。API稳定、规范、有文档。但现实是，很多数据源没有API，或者API限制太严。这时候就得爬虫上场了。

3.2.1 API采集：优雅且高效

举个例子，获取A股日线数据。很多数据服务商都提供HTTP API。你发一个请求，它返回JSON。就这么简单。

import requests
import pandas as pd

# 假设这是某数据服务商的API
url = "https://api.example.com/stock/daily"
params = {
    "symbol": "000001.SZ",
    "start_date": "2024-01-01",
    "end_date": "2024-12-31",
    "api_key": "your_api_key_here"
}

response = requests.get(url, params=params)
data = response.json()

# 直接转成DataFrame
df = pd.DataFrame(data["records"])
print(df.head())

这里有个坑：API返回的JSON结构，不同服务商千奇百怪。我遇到过返回嵌套三层字典的，也遇到过字段名全是中文拼音缩写的。嗯，这时候别偷懒，写个解析函数，把字段名统一映射成你内部的标准命名。

3.2.2 爬虫采集：灵活但需谨慎

当API拿不到数据时，爬虫就是你的第二选择。比如抓取财经网站的新闻标题和发布时间。

import requests
from bs4 import BeautifulSoup

url = "https://finance.example.com/news"
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}

resp = requests.get(url, headers=headers)
soup = BeautifulSoup(resp.text, "html.parser")

# 假设每条新闻在 <div class="news-item"> 里
news_list = []
for item in soup.select("div.news-item"):
    title = item.find("h3").text.strip()
    time = item.find("span", class_="time").text.strip()
    news_list.append({"title": title, "time": time})

print(f"抓取到 {len(news_list)} 条新闻")

3.3 数据清洗与标准化流程

数据采集回来，只是万里长征第一步。清洗才是真正考验耐心的地方。我总结了一个三步走流程：去重、对齐、归一化。

3.3.1 去重：别让重复数据骗了你

重复数据在投研里特别隐蔽。比如同一只股票，不同数据源都报了当天的收盘价，但时间戳差了毫秒级。你如果不处理，回测时可能把同一笔交易算两次。

# 基于关键字段去重
df_clean = df.drop_duplicates(subset=["symbol", "trade_date", "close"])

# 如果时间戳有微小差异，先做四舍五入或截断
df["trade_date"] = pd.to_datetime(df["trade_date"]).dt.floor("D")
df_clean = df.drop_duplicates(subset=["symbol", "trade_date"])

为什么会这样？因为不同数据源的数据刷新频率不同。我建议你定一个规则：同一只股票、同一天，只保留一条记录。如果有多条，取最后一条或者取均值，看你的策略需求。

3.3.2 对齐：让不同数据源说同一种语言

对齐是清洗里最繁琐的一步。不同数据源，股票代码格式不一样。比如平安银行，在Wind里是"000001.SZ"，在Tushare里是"000001"，在Yahoo Finance里是"000001.SZ"。

我的做法是建一个映射表：

# 股票代码映射表示例
code_mapping = {
    "000001.SZ": {"wind": "000001.SZ", "tushare": "000001", "yahoo": "000001.SZ"},
    "600036.SH": {"wind": "600036.SH", "tushare": "600036", "yahoo": "600036.SS"},
    # ... 更多映射
}

def align_code(code, source, target):
    """将某个数据源的代码，映射到目标数据源的代码"""
    return code_mapping.get(code, {}).get(target)

3.3.3 归一化：把数据塞进同一个模子

归一化包括两个层面：量纲归一化和格式归一化。

• 量纲归一化： 比如有的数据源价格单位是元，有的是分。有的成交量单位是股，有的是手。统一成你内部的标准。
• 格式归一化： 日期格式统一成YYYY-MM-DD，浮点数统一保留两位小数，缺失值统一填充为NaN而不是空字符串。

# 量纲归一化示例
def normalize_price(price, source_unit="yuan"):
    if source_unit == "fen":
        return price / 100.0
    return price

# 格式归一化
df["trade_date"] = pd.to_datetime(df["trade_date"]).dt.strftime("%Y-%m-%d")
df["close"] = df["close"].astype(float).round(2)
df["volume"] = df["volume"].fillna(0).astype(int)

你想想看，如果这些细节没处理好，策略回测时可能因为一个字段类型错误，整个流程就崩了。我早期就吃过这个亏，一个字符串类型的价格字段，直接让计算收益率时报错，排查了整整一下午。

3.4 本章知识体系总览

下面这张图，是我梳理的数据采集与清洗的核心逻辑。你可以把它当作一个检查清单，每次做数据工程时对照着来。

这张图把整个流程串起来了。你从左边开始，先搞清楚你的数据源属于哪一类，然后选择合适的采集方式，最后进入清洗流水线。每一步都有对应的技术方案和常见坑点。