3. 行情数据接入：CTP API 简介、行情订阅与推送机制、Tick 数据与 K 线数据、数据清洗与对齐

行情数据接入，是整个量化系统的“眼睛”。

没有数据，策略就是瞎猜。我见过不少团队，策略逻辑写得天花乱坠，结果一上实盘，行情延迟几百毫秒，直接亏穿。所以，这块咱们得扎扎实实讲清楚。

3.1 CTP API 简介

CTP，全称是“综合交易平台”，上期技术开发的。国内期货市场，90%以上的交易接口都基于它。

说白了，CTP 就是期货公司和交易所之间的“翻译官”。你的策略要下单、要拿行情，都得通过它。

CTP 提供两套 API：

• 交易 API（Trader API）：负责下单、撤单、查持仓、查资金。
• 行情 API（MdSpi / MdApi）：负责订阅行情、接收实时推送。

我个人习惯，把行情 API 单独封装成一个服务。交易和行情分离，万一行情模块崩了，不影响下单逻辑。

3.2 行情订阅与推送机制

CTP 的行情推送，不是“你问它答”的模式。

它是订阅-推送模式。你先告诉 CTP：“我要看螺纹钢主力合约的行情。” 然后 CTP 每收到一笔新的交易所数据，就主动推给你。

流程大致是这样：

1. 初始化 MdApi 实例
2. 注册回调函数（OnFrontConnected、OnRtnDepthMarketData 等）
3. 调用 SubscribeMarketData 订阅合约
4. CTP 收到新数据后，触发 OnRtnDepthMarketData

嗯，这里有个坑。我曾经遇到过：订阅了多个合约，结果回调里只处理了第一个合约的数据。为什么？因为回调是单线程的，处理慢了，后面的数据就丢了。

代码示例（伪代码，展示核心逻辑）：

class MyMdSpi : public CThostFtdcMdSpi {
public:
    void OnFrontConnected() {
        // 连接成功，订阅合约
        char* instruments[] = {"rb2401", "rb2405"};
        m_pMdApi->SubscribeMarketData(instruments, 2);
    }

    void OnRtnDepthMarketData(CThostFtdcDepthMarketDataField* pData) {
        // 收到行情，推入队列
        queue.push(*pData);
    }
};

3.3 Tick 数据与 K 线数据

行情数据分两种粒度：Tick 和 K 线。

Tick 数据，就是每一笔成交的快照。包含最新价、成交量、持仓量、买卖盘口等。CTP 推送的 OnRtnDepthMarketData，就是 Tick 数据。

Tick 数据的特点是：高频、海量。一个活跃合约，一天能产生几十万笔 Tick。我做过统计，螺纹钢主力合约，高峰期每秒能推送 50-100 笔 Tick。

K 线数据，是对 Tick 数据的聚合。比如 1 分钟 K 线，就是把 1 分钟内的所有 Tick，按时间切分，计算开高低收、成交量。

K 线生成逻辑：

• 收到 Tick 后，判断它属于哪个时间窗口（比如 09:30:00 - 09:30:59）
• 更新该窗口的 Open（第一笔）、High（最高）、Low（最低）、Close（最后一笔）
• 窗口结束时，输出这根 K 线，重置状态

下面这张图，展示了 Tick 到 K 线的转换过程：

3.4 数据清洗与对齐

行情数据不是完美的。你想想看，网络抖动、交易所重启、CTP 服务器切换，都会导致数据异常。

我总结了几类常见问题：

• 重复数据：同一笔 Tick 被推送两次。检查时间戳和成交量的唯一性。
• 缺失数据：某几秒没有 Tick。可能是网络丢包。需要做插值或标记。
• 乱序数据：后推送的 Tick 时间戳反而更早。需要按时间戳排序。
• 异常值：价格突然跳空 10%，明显是错误数据。需要设置阈值过滤。

数据对齐，指的是把不同来源的数据，对齐到同一个时间轴上。

比如，你同时订阅了螺纹钢和铁矿石的行情。它们的 Tick 推送时间不一样。你要做策略，需要知道“同一时刻”两个品种的价格。

对齐方法：

1. 时间戳对齐：以交易所时间为准，把不同品种的 Tick 按时间戳排序。
2. 插值对齐：如果某个品种在某个时刻没有 Tick，用前一笔 Tick 的价格填充。
3. 快照对齐：每隔固定时间（比如 1 秒），取所有品种的最新价格，生成一个快照。

数据清洗的代码示例（Python 风格）：

def clean_tick_data(tick_list):
    # 1. 去重
    seen = set()
    unique_ticks = []
    for tick in tick_list:
        key = (tick['UpdateTime'], tick['LastPrice'], tick['Volume'])
        if key not in seen:
            seen.add(key)
            unique_ticks.append(tick)

    # 2. 按时间排序
    unique_ticks.sort(key=lambda x: x['UpdateTime'])

    # 3. 过滤异常值
    filtered = []
    for i, tick in enumerate(unique_ticks):
        if i == 0:
            filtered.append(tick)
            continue
        prev = unique_ticks[i-1]
        # 价格变化超过 5% 视为异常
        if abs(tick['LastPrice'] - prev['LastPrice']) / prev['LastPrice'] > 0.05:
            continue
        filtered.append(tick)

    return filtered

好了，行情数据接入这块，核心就是这些。CTP 的 API 虽然有点老，但稳定。Tick 到 K 线的转换，逻辑不复杂，但细节多。数据清洗，说白了就是“去伪存真”。

下一章，咱们聊聊交易执行模块。嗯，那个更刺激。

---