数据模块（Data Feed）：数据源接入、清洗与OHLCV设计
做量化回测，第一道坎就是数据。我见过太多人策略写得漂亮，结果数据一进来就崩了。说白了，数据模块是整个回测引擎的根基——地基没打牢，上面盖多高的楼都得塌。
今天咱们就聊聊这个数据模块怎么设计。我会从数据源接入讲起，再到清洗对齐，最后给出OHLCV的数据结构。嗯，这些都是我踩过坑之后总结出来的经验。
数据源接入：CSV、数据库、API
数据源接入，说白了就是怎么把数据弄进来。常见的就三种：CSV文件、数据库、API接口。每种都有它的适用场景。
CSV文件接入
CSV是最简单的，也是我最常用的调试方式。写策略的时候，从网上下载一份历史数据，存成CSV就能跑起来。但要注意，CSV的坑也不少。
import pandas as pd
def load_csv(filepath, date_column='date', date_format='%Y-%m-%d'):
"""
加载CSV数据，带基本的容错处理
"""
try:
df = pd.read_csv(filepath)
# 统一日期格式
df[date_column] = pd.to_datetime(df[date_column], format=date_format, errors='coerce')
# 丢弃解析失败的行
df = df.dropna(subset=[date_column])
return df
except Exception as e:
print(f"加载CSV失败: {e}")
return None数据库接入
数据量大了之后，CSV就不够用了。我习惯用SQLite做本地存储，MySQL或PostgreSQL做服务端存储。数据库的好处是查询灵活，能按日期范围、股票代码快速筛选。
import sqlite3
import pandas as pd
class DatabaseFeed:
def __init__(self, db_path):
self.conn = sqlite3.connect(db_path)
def load_data(self, symbol, start_date, end_date):
query = """
SELECT date, open, high, low, close, volume
FROM daily_data
WHERE symbol = ? AND date BETWEEN ? AND ?
ORDER BY date
"""
df = pd.read_sql_query(query, self.conn, params=(symbol, start_date, end_date))
return dfAPI接入
做实盘或者需要实时数据的时候，就得用API了。我建议把API封装成一个统一的接口，这样切换数据源的时候不用改业务代码。
class APIFeed:
def __init__(self, api_key, base_url):
self.api_key = api_key
self.base_url = base_url
def fetch_ohlcv(self, symbol, interval='1d', limit=100):
# 伪代码，实际API调用需要处理认证、限流等
url = f"{self.base_url}/klines?symbol={symbol}&interval={interval}&limit={limit}"
response = requests.get(url, headers={"X-API-Key": self.api_key})
return self._parse_response(response.json())数据清洗与对齐
数据拿进来之后，千万别直接用。我敢说，90%的回测错误都跟数据质量有关。清洗这一步，省不得。
常见的数据问题
- 缺失值：某天没有交易数据，或者数据源漏了
- 异常值：价格突然跳空，或者成交量异常放大
- 重复数据：同一时间戳出现多条记录
- 时间戳不对齐：不同数据源的时间精度不一样
def clean_ohlcv(df):
"""
数据清洗流程
"""
# 1. 去重
df = df.drop_duplicates(subset=['date'])
# 2. 排序
df = df.sort_values('date')
# 3. 处理缺失值：用前一天的收盘价填充
df = df.fillna(method='ffill')
# 4. 异常值检测：价格不能为负，成交量不能为0
df = df[(df['close'] > 0) & (df['volume'] > 0)]
# 5. 重置索引
df = df.reset_index(drop=True)
return df数据对齐
做多品种回测时，数据对齐是个大问题。比如A股和港股的开盘时间不一样，期货还有夜盘。你想想看，如果不对齐，策略信号就全乱了。
我常用的方法是：统一转换成UTC时间，然后按分钟或日线做重采样。这样不管原始数据是什么频率，最终都能对齐到同一个时间轴上。
def align_data(data_dict, freq='1d'):
"""
多品种数据对齐
data_dict: {symbol: DataFrame}
"""
aligned = {}
for symbol, df in data_dict.items():
# 设置时间索引
df = df.set_index('date')
# 重采样到统一频率
df = df.resample(freq).agg({
'open': 'first',
'high': 'max',
'low': 'min',
'close': 'last',
'volume': 'sum'
})
aligned[symbol] = df
return alignedOHLCV数据结构设计
OHLCV是回测引擎的核心数据结构。设计得好，后面所有模块都顺畅；设计得不好，处处都得打补丁。
基础结构
我建议用Pandas的DataFrame作为底层存储，但对外暴露统一的接口。这样既利用了Pandas的高效计算，又保持了接口的灵活性。
class OHLCVData:
def __init__(self, df):
# 确保数据格式正确
required_cols = ['open', 'high', 'low', 'close', 'volume']
for col in required_cols:
if col not in df.columns:
raise ValueError(f"缺少必要列: {col}")
self._data = df.copy()
self._validate()
def _validate(self):
"""数据校验"""
assert (self._data['high'] >= self._data['low']).all(), "最高价不能低于最低价"
assert (self._data['close'] >= 0).all(), "收盘价不能为负"
@property
def latest(self):
"""获取最新一条数据"""
return self._data.iloc[-1]
def slice(self, start, end):
"""按时间范围切片"""
mask = (self._data['date'] >= start) & (self._data['date'] <= end)
return OHLCVData(self._data[mask])扩展字段
除了标准的OHLCV，我还会加几个常用字段：
| 字段名 | 类型 | 说明 |
|---|---|---|
| date | datetime | 时间戳，统一为UTC |
| open | float | 开盘价 |
| high | float | 最高价 |
| low | float | 最低价 |
| close | float | 收盘价 |
| volume | float | 成交量 |
| vwap | float | 成交量加权平均价（可选） |
| returns | float | 收益率（计算字段） |
数据模块的整体架构
说了这么多，咱们来看看数据模块的整体架构。我画了一张图，把数据流的走向展示清楚。
从图上能看出来，数据从源头进来，经过清洗、对齐，最终变成标准的OHLCV格式，供策略使用。每个环节都有它存在的意义，缺一不可。
好了，数据模块就聊到这儿。记住一句话：数据质量决定回测质量。花时间把数据模块做好，后面会省心很多。