4. 数据获取与清洗：从API/CSV获取历史Tick数据

做市回测这件事，数据就是你的弹药库。弹药不行，枪法再好也白搭。我见过太多人把精力全花在策略上，结果数据一塌糊涂，回测跑出来全是幻觉。今天咱们就把数据获取和清洗这块硬骨头啃下来。

4.1 数据来源：API vs CSV，怎么选？

先说结论：没有完美的数据源，只有适合你的数据源。我个人习惯是，做快速原型用CSV，做生产级回测用API。

对比维度	API实时获取	CSV本地文件
获取速度	慢（受网络限制）	快（本地读取）
数据完整性	可能断流	可校验完整性
历史数据深度	有限（通常1-2年）	可积累多年
成本	按调用量收费	一次性存储成本

你想想看，做市策略需要的是毫秒级的Tick数据，API如果中间断个几秒，你的回测就废了。所以我建议：先用CSV做本地缓存，API只做增量更新。

4.2 从CSV读取Tick数据

这是最基础的操作。但别小看它，我见过有人用pandas直接read_csv，结果内存爆了——几千万行Tick数据可不是闹着玩的。

import pandas as pd
import dask.dataframe as dd

# 小数据量（少于100万行）
df = pd.read_csv('eurusd_tick_2024.csv', 
                 parse_dates=['timestamp'],
                 dtype={'bid': 'float32', 'ask': 'float32'})

# 大数据量（推荐用Dask分块读取）
ddf = dd.read_csv('eurusd_tick_2024.csv',
                  blocksize='64MB',
                  parse_dates=['timestamp'])
df = ddf.compute()  # 只在需要时加载到内存

4.3 从API获取实时Tick

API获取的核心问题是断流。我曾经在回测中发现某天下午3点到3点05分的数据全没了，查了半天才发现是API在那段时间做了维护。

import requests
import time
from datetime import datetime

def fetch_tick_data(symbol, start_date, end_date, api_key):
    """
    带重试机制的API获取
    """
    url = f"https://api.forexdata.com/v1/ticks/{symbol}"
    params = {
        'start': start_date,
        'end': end_date,
        'api_key': api_key
    }
    
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            data = response.json()
            return pd.DataFrame(data['ticks'])
        except Exception as e:
            print(f"第{attempt+1}次尝试失败: {e}")
            time.sleep(2 ** attempt)  # 指数退避
    
    raise Exception("API获取失败，请检查网络或API配额")

4.4 数据清洗：去重

Tick数据重复的原因很多：API重发、网络抖动、交易所撮合引擎的bug。去重是第一步，也是最简单的一步。

# 基于时间戳和价格去重
df_clean = df.drop_duplicates(subset=['timestamp', 'bid', 'ask'])

# 检查重复率
dup_rate = (len(df) - len(df_clean)) / len(df) * 100
print(f"重复率: {dup_rate:.2f}%")

# 如果重复率超过5%，建议检查数据源
if dup_rate > 5:
    print("警告：数据重复率异常，请检查API或CSV文件")

嗯，这里要注意：不要只按时间戳去重。同一毫秒内可能有多个不同的Tick，这是正常的市场行为。必须结合价格一起判断。

4.5 数据清洗：补全

补全是最头疼的部分。Tick数据缺失的原因五花八门：节假日、流动性枯竭、交易所故障。我的原则是：能补则补，不能补就标记。

# 检测缺失的时间段
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.set_index('timestamp')
full_range = pd.date_range(start=df.index.min(), 
                           end=df.index.max(), 
                           freq='1ms')

# 找出缺失的时间点
missing_times = full_range.difference(df.index)
print(f"缺失时间点数量: {len(missing_times)}")

# 前向填充（适用于短时间缺失，如<100ms）
df_filled = df.reindex(full_range, method='ffill', limit=100)

# 长时间缺失（如>1秒）标记为NaN
df_filled = df.reindex(full_range)
df_filled.loc[df_filled.isna().any(axis=1), 'is_missing'] = True

4.6 数据清洗：异常值处理

异常值是最坑人的。我记得有一次回测跑出来年化收益300%，兴奋得不行。结果一查，原来是某天有个Tick的bid价格写成了1.2345，实际应该是1.1234，差了1000个点。这种异常值会让你的策略「看起来」很赚钱，实盘直接亏到姥姥家。

# 基于统计学的异常检测
def detect_outliers(df, column='spread', threshold=3):
    """
    使用Z-score检测异常值
    """
    mean = df[column].mean()
    std = df[column].std()
    z_scores = (df[column] - mean) / std
    
    outliers = df[abs(z_scores) > threshold]
    return outliers

# 基于业务规则的异常检测
def business_rule_outliers(df):
    """
    检测明显不合逻辑的数据
    """
    # 1. 买卖价差为负
    negative_spread = df[df['ask'] <= df['bid']]
    
    # 2. 价格跳变超过100点（1个标准跳变是0.0001）
    df['price_change'] = df['bid'].diff().abs()
    large_jump = df[df['price_change'] > 0.01]  # 100点
    
    # 3. 价格超出合理范围
    price_range = df[(df['bid'] < 1.0) | (df['bid'] > 2.0)]
    
    return negative_spread, large_jump, price_range

# 处理异常值：直接删除或标记
df_clean = df[~df.index.isin(outliers.index)]
# 或者标记但不删除（保留审计痕迹）
df['is_outlier'] = False
df.loc[outliers.index, 'is_outlier'] = True

4.7 数据清洗流程总览

下面这张图是我在实际项目中总结的清洗流程，你可以直接拿来用。

4.8 实战：完整的清洗函数

最后，给你一个可以直接用的清洗函数。这是我项目里一直在用的，经过了几十亿条Tick数据的考验。

def clean_tick_data(df, config=None):
    """
    完整的Tick数据清洗函数
    
    参数:
        df: DataFrame，包含timestamp, bid, ask列
        config: 配置字典，可自定义阈值
    
    返回:
        df_clean: 清洗后的DataFrame
        report: 清洗报告
    """
    if config is None:
        config = {
            'max_spread': 0.005,      # 最大价差（50点）
            'max_price_jump': 0.01,   # 最大价格跳变（100点）
            'max_missing_ms': 100,    # 最大可填充缺失时间（毫秒）
            'z_score_threshold': 3    # Z-score阈值
        }
    
    report = {'total_rows': len(df)}
    
    # 1. 去重
    df = df.drop_duplicates(subset=['timestamp', 'bid', 'ask'])
    report['after_dedup'] = len(df)
    report['dup_removed'] = report['total_rows'] - report['after_dedup']
    
    # 2. 排序
    df = df.sort_values('timestamp').reset_index(drop=True)
    
    # 3. 检测并标记异常
    df['is_outlier'] = False
    
    # 3.1 负价差
    df.loc[df['ask'] <= df['bid'], 'is_outlier'] = True
    
    # 3.2 价格跳变
    df['price_change'] = df['bid'].diff().abs()
    df.loc[df['price_change'] > config['max_price_jump'], 'is_outlier'] = True
    
    # 3.3 价差过大
    df['spread'] = df['ask'] - df['bid']
    df.loc[df['spread'] > config['max_spread'], 'is_outlier'] = True
    
    # 3.4 Z-score异常
    z_scores = (df['spread'] - df['spread'].mean()) / df['spread'].std()
    df.loc[abs(z_scores) > config['z_score_threshold'], 'is_outlier'] = True
    
    report['outliers'] = df['is_outlier'].sum()
    
    # 4. 处理缺失（前向填充短时间缺失）
    df = df.set_index('timestamp')
    full_range = pd.date_range(start=df.index.min(), 
                               end=df.index.max(), 
                               freq='1ms')
    df = df.reindex(full_range)
    df = df.ffill(limit=config['max_missing_ms'])
    
    report['missing_filled'] = df['bid'].isna().sum()
    
    # 5. 删除无法填充的缺失
    df = df.dropna(subset=['bid', 'ask'])
    
    report['final_rows'] = len(df)
    report['quality_score'] = (1 - report['outliers'] / report['final_rows']) * 100
    
    return df.reset_index(), report

# 使用示例
df_clean, report = clean_tick_data(df_raw)
print(f"数据质量评分: {report['quality_score']:.2f}%")

数据获取和清洗，说白了就是「垃圾进，垃圾出」的反面。你花80%的时间把数据搞干净，后面20%的策略开发才能安心。别嫌麻烦，这是做市回测的基石。