3、数据源接入与预处理：对接Tushare/Wind/本地CSV数据，数据清洗、对齐与标准化

做量化回测，第一步就是搞数据。

数据搞不定，后面全是白搭。我见过太多人，策略写得漂漂亮亮，一跑回测就崩，最后发现是数据没对齐。嗯，这章咱们就把数据源接入和预处理这件事聊透。

3.1 数据源的选择：Tushare、Wind、本地CSV

我个人习惯把数据源分成三类：免费开源、商业付费、本地文件。各有各的适用场景。

数据源	特点	适用场景
Tushare Pro	免费但需积分，覆盖A股/期货/基金	个人研究、小规模回测
Wind	收费，数据质量高，接口稳定	机构级、生产环境
本地CSV	灵活，可离线，格式自定	历史数据回放、自定义数据

你想想看，如果只是自己写写策略玩，Tushare完全够用。但要是跑实盘或者做高频分析，Wind的稳定性和数据精度就很重要了。我曾经在项目中遇到过Tushare某天接口超时，导致整个回测任务中断的情况——从那以后，我就在代码里加了重试机制和本地缓存。

3.2 数据清洗：脏数据是回测的隐形杀手

数据拿到手，第一件事不是分析，是清洗。说白了，就是要把那些「不对劲」的数据揪出来。

常见的脏数据有这几类：

• 缺失值：某天停牌，价格字段为空
• 异常值：价格突然跳涨100倍，明显是数据录入错误
• 重复数据：同一时间戳出现多条记录
• 未来数据：时间戳错乱，把未来的数据混进来了

我一般这样处理：

import pandas as pd

def clean_market_data(df):
    # 1. 去重
    df = df.drop_duplicates(subset=['trade_date', 'ts_code'])
    
    # 2. 处理缺失值：前向填充
    df = df.sort_values('trade_date')
    df['close'] = df['close'].fillna(method='ffill')
    
    # 3. 剔除异常值：超过5个标准差
    mean = df['close'].mean()
    std = df['close'].std()
    df = df[(df['close'] > mean - 5*std) & (df['close'] < mean + 5*std)]
    
    # 4. 检查时间顺序
    df = df[df['trade_date'] <= pd.Timestamp.today()]
    
    return df

3.3 数据对齐：不同频率、不同品种的「时间配准」

做多因子回测时，最头疼的就是数据对齐。你想想看，股票A是日频数据，股票B是分钟频数据，指数又是另一个频率——怎么把它们拉到同一个时间轴上？

我常用的对齐策略：

• 降频对齐：把高频数据聚合到低频（比如分钟转日线）
• 重采样对齐：用resample方法统一时间戳
• 交叉对齐：取所有数据源的时间戳并集，缺失的用NaN填充

def align_multi_assets(data_dict, freq='D'):
    """
    data_dict: {'000001.SZ': df1, '600000.SH': df2, ...}
    freq: 'D' 日频, 'W' 周频, 'M' 月频
    """
    aligned = []
    for code, df in data_dict.items():
        df = df.set_index('trade_date')
        # 重采样到统一频率
        df = df.resample(freq).last()
        df['ts_code'] = code
        aligned.append(df)
    
    # 合并所有资产
    result = pd.concat(aligned).reset_index()
    return result.pivot(index='trade_date', columns='ts_code', values='close')

这里有个坑：不同交易所的交易日历不一样。A股有春节、国庆，港股有圣诞，美股有感恩节。我曾经因为没处理交易日历差异，导致回测结果里出现了「在A股休市日交易」的荒谬情况。嗯，后来我专门维护了一份交易日历表。

3.4 数据标准化：让不同量纲的数据「可比较」

标准化不是为了好看，是为了让模型能正确学习。你想想看，如果价格是几百块，成交量是几百万手，这两个数值直接丢进模型，价格那列基本就被忽略了。

常用的标准化方法：

• Z-score标准化：(x - mean) / std，适合正态分布数据
• Min-Max归一化：(x - min) / (max - min)，把数据压到[0,1]区间
• 对数变换：log(x)，适合右偏分布（比如收益率）

def standardize_features(df, method='zscore'):
    if method == 'zscore':
        return (df - df.mean()) / df.std()
    elif method == 'minmax':
        return (df - df.min()) / (df.max() - df.min())
    elif method == 'log':
        return np.log(df - df.min() + 1)

3.5 数据流水线：从原始数据到可用特征

在实际项目中，我不会手动一步步操作。我会搭一个数据流水线（Pipeline），把清洗、对齐、标准化串起来。

下面是我常用的流程：

class DataPipeline:
    def __init__(self, sources):
        self.sources = sources  # ['tushare', 'wind', 'csv']
        self.raw_data = {}
        self.clean_data = {}
    
    def fetch(self, codes, start, end):
        for source in self.sources:
            if source == 'tushare':
                self.raw_data.update(self._fetch_tushare(codes, start, end))
            elif source == 'wind':
                self.raw_data.update(self._fetch_wind(codes, start, end))
            elif source == 'csv':
                self.raw_data.update(self._fetch_csv(codes, start, end))
    
    def process(self):
        for code, df in self.raw_data.items():
            df = clean_market_data(df)
            df = align_multi_assets({code: df})
            df = standardize_features(df)
            self.clean_data[code] = df
        return self.clean_data

这样做的好处是：哪天想换数据源，或者加一个新的预处理步骤，只需要改Pipeline里的一个方法就行。我在项目中用过这个模式，维护成本低很多。

3.6 避坑指南：数据预处理中的常见陷阱

做数据预处理这些年，我踩过的坑不少。挑几个典型的说说：

• 幸存者偏差：只用了当前还在交易的股票，退市的股票数据全丢了。回测结果会虚高。解决办法：用历史快照数据，或者包含退市股票。
• 前视偏差：用未来的数据来填充过去的缺失值。比如用明天的收盘价来填充今天的NaN。这会导致回测结果「未卜先知」。
• 除权除息未处理：股票分红送股后，价格会跳空。如果不做复权处理，技术指标全乱套。我一般用后复权数据做回测。

3.7 本章知识体系

下面这张图，是我对数据预处理流程的总结。你可以把它当作一个检查清单：

数据预处理这件事，说白了就是「垃圾进，垃圾出」。你花80%的时间在数据上，剩下20%的时间写策略，这才是正确的比例。别想着跳过这一步，我试过，结果就是回测曲线漂亮得像假的一样——嗯，它确实是假的。

---