分享一个量化交易项目的数据处理经验

最近在做一个量化交易项目，前期花了不少时间在策略本身，后来发现真正影响回测结果的，反而是行情数据。
刚开始拿到 Tick 数据时，我觉得只要按照时间排序、转成 K 线就差不多了。实际处理下来才发现，实时行情里会遇到重复 Tick、时间戳异常、乱序、短暂缺失等情况。如果这些问题不处理，最后策略收益看起来可能很好，但很难判断到底是策略有效，还是数据有问题。

先解决行情获取
我的做法是把实时行情和历史数据分开处理。实时部分使用 WebSocket 持续接收 Tick，落地后再进行清洗；历史数据则用于回测和参数验证。
import json
import websocket

API_KEY = "your_api_key"
WS_URL = f"wss://quote.alltick.co/quote-stock-b-ws-api?token={API_KEY}"

def on_open(ws):
    subscribe_msg = {
        "cmd_id": 22004,
        "seq_id": 1,
        "trace": "sub-us-stock",
        "data": {
            "symbol_list": [
                {"code": "AAPL.US"},
                {"code": "TSLA.US"}
            ]
        }
    }
    ws.send(json.dumps(subscribe_msg))

def on_message(ws, message):
    data = json.loads(message)
    print("收到行情：", data)

def on_error(ws, error):
    print("连接出错：", error)

def on_close(ws, close_status_code, close_msg):
    print("连接关闭，准备重连")

if __name__ == "__main__":
    ws = websocket.WebSocketApp(
        WS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.run_forever()
这里有个细节我觉得比较重要：不要把收到的数据直接当成干净的行情数据使用。
实时数据是一个持续到来的数据流，网络延迟、连接重连以及不同数据源的时间戳，都可能影响数据顺序。

重视Tick清洗
我后来给 Tick 数据增加了一层比较简单的数据清洗。
首先检查时间戳。比如同一个交易品种，如果后一条 Tick 的时间明显早于上一条，就不能简单认为它就是正常数据。对于轻微乱序，可以先缓存几条数据，再按照时间戳重新排序。
其次是重复数据。有时候相邻 Tick 的价格、成交量和时间都完全一样，如果直接聚合成 K 线，可能导致成交量被重复计算。
我现在通常会先做：
df = df.sort_values("timestamp")

df = df.drop_duplicates(
    subset=["symbol", "timestamp", "price", "volume"]
)
然后再检查时间间隔。
例如正常情况下 Tick 每隔几十毫秒或者几百毫秒出现，但中间突然出现几分钟没有数据，这时候我不会直接用前一个价格填充，而是把这段区间标记出来。
因为对于量化回测来说，“没有数据”和“价格没有变化”是两件完全不同的事情。

Tick 转 K 线不能太随意
完成清洗之后，才进入 K 线聚合。
比如做 1 分钟策略，可以根据 Tick 的时间戳分组，计算：

• Open = 第一笔价格
• High = 最高价格
• Low = 最低价格
• Close = 最后一笔价格
• Volume = 成交量汇总

这里我遇到的一个问题是跨分钟 Tick。
假设一笔 Tick 恰好发生在 10:01:00.001，它应该属于 10:01 这一分钟，而不是上一分钟。时间边界处理不一致，会导致不同程序生成的 K 线出现细微差异。
所以我会先统一时区和时间戳格式，再进行 resample，而不是拿字符串直接切时间。

回测阶段重点检查“未来数据”
数据清洗完成之后才进入策略回测。
我觉得回测里最容易忽略的是 look‑ahead bias，也就是不小心使用了当时还不知道的数据。
比如使用一根 1 分钟 K 线的收盘价产生交易信号，然后又假设自己能够在这根 K 线的收盘价成交，这在很多实际场景下是不成立的。
所以我的处理方式是让信号和成交至少错开一个时间点。例如：
df["signal"] = (
    df["close"] > df["ma20"]
).astype(int)

df["position"] = df["signal"].shift(1)
这样至少可以避免策略直接使用当前 K 线的结果进行当前价格成交。

最后才看策略收益
以前我做回测时比较关注收益率和 Sharpe Ratio。
现在会先看数据质量，再看策略结果。
如果发现策略突然出现非常漂亮的收益曲线，我反而会先检查 Tick 是否重复、时间是否错乱、K 线是否存在未来数据，以及手续费和滑点有没有计算。
尤其是高频或者短周期策略，几个 Tick 的差异就可能改变进出场价格。回测收益如果没有经过这些检查，我觉得参考价值其实比较有限。
这次项目给我的一个比较直接的经验就是：量化项目里，行情获取只是第一步，真正需要花时间的是把原始行情变成可以信任的数据。
策略代码可能只有几百行，但从行情接入、Tick 清洗、K 线聚合到回测验证，这一整套数据处理链路如果没有做好，后面的策略优化其实都是建立在不稳定的数据基础上。

干活，赞一个，学习学习

做数据清洗， 直接让AI做然后回测应该就可以了吧？ 为什么需要手搓呢