4、实盘交易接口：券商API对比与CTP入门

做量化策略，写代码回测只是第一步。真正要拿真金白银去跑，你得面对一个现实问题——怎么跟交易所打交道？说白了，就是选哪个接口去下单。

我见过不少朋友，回测曲线画得漂漂亮亮，一到实盘就卡在API对接上。嗯，今天咱们就把这事掰扯清楚。

4.1 券商API横向对比

国内主流的量化交易接口，我接触下来，华泰xtquant、中泰xtquant、国信TradeX这三家用得最多。先看个表格，心里有个底。

对比项	华泰xtquant	中泰xtquant	国信TradeX
底层架构	基于xtquant框架	基于xtquant框架	自研C++交易引擎
行情速度	约50ms延迟	约50ms延迟	约10ms延迟
交易品种	A股、ETF、可转债	A股、ETF、可转债	A股、ETF、期货
Python支持	原生支持	原生支持	需封装
稳定性	中等	中等	较高
费用	免费	免费	按量收费

我个人习惯，如果只是做A股普通策略，华泰xtquant够用了。但如果你要跑高频或者期货，国信TradeX更靠谱。

华泰xtquant

这个接口我最早接触。它的优点是上手快，Python直接调，文档也全。我在项目中遇到过一个问题——它的行情推送偶尔会断，需要自己写心跳重连。嗯，这里要注意，别指望它一直稳如老狗。

中泰xtquant

跟华泰的底层其实一样，都是xtquant框架。但中泰在账户风控上做得更细。比如单笔下单量限制、日内撤单次数，这些都有默认阈值。我建议，如果你资金量比较大，中泰的接口更安全。

国信TradeX

这个接口我是在做期货策略时用的。它的延迟确实低，10ms以内。但代价是什么？你得自己封装Python接口。说白了，它给的是C++的SDK，你得用ctypes或者pybind11去调。我曾经为了封装一个下单函数，折腾了一整天。

4.2 CTP接口入门

CTP（综合交易平台）是上期技术开发的，国内期货市场的事实标准。你想想看，几乎所有期货公司都接CTP，你不学它，做期货就是空谈。

CTP的核心概念其实不多：

• 交易前置：你连的服务器地址，负责收发订单
• 行情前置：专门推送行情数据的服务器
• 投资者ID：你的资金账号
• 合约代码：比如rb2401，螺纹钢2401合约

我刚开始学CTP时，被它的回调机制搞晕了。它不像普通API那样调用就返回结果，而是异步的。你发一个下单请求，它通过回调函数告诉你结果。为什么会这样？因为交易系统不能阻塞，必须异步处理。

看个最简单的CTP登录代码：

# 伪代码，展示CTP登录流程
from ctp import CTPTrader

trader = CTPTrader()
trader.OnFrontConnected = lambda: print("连接成功")
trader.OnRspUserLogin = lambda data: print(f"登录成功，账号: {data['InvestorID']}")

trader.connect("tcp://180.168.146.187:10000")
trader.login("your_account", "your_password", "your_broker_id")

注意，CTP的IP和端口每个期货公司都不一样。我曾经因为配错了端口，折腾了三个小时才发现是文档写错了。嗯，拿到接口文档后，先ping一下服务器，确认网络通不通。

4.3 WebSocket行情订阅

行情数据怎么拿？传统方式是轮询，每隔几秒拉一次。但这样延迟高，还浪费带宽。WebSocket是更好的选择——它建立长连接，服务端主动推送数据给你。

我常用的WebSocket行情库是websocket-client。看个订阅行情的例子：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    print(f"最新价: {data['price']}, 时间: {data['time']}")

def on_error(ws, error):
    print(f"连接出错: {error}")

def on_close(ws, close_status_code, close_msg):
    print("连接关闭")

def on_open(ws):
    # 订阅深证成指
    subscribe_msg = {
        "action": "subscribe",
        "symbol": "399001"
    }
    ws.send(json.dumps(subscribe_msg))

ws = websocket.WebSocketApp("wss://market.example.com/ws",
                            on_open=on_open,
                            on_message=on_message,
                            on_error=on_error,
                            on_close=on_close)
ws.run_forever()

这里有个细节——WebSocket连接会断。网络波动、服务器重启，都可能断开。我建议在on_close里加自动重连逻辑，不然半夜断了，第二天起来一看，策略没跑。

4.4 交易指令封装

直接调用券商API下单，代码会变得很乱。我习惯把交易指令封装成统一的接口。这样换券商时，只需要改底层实现，上层策略代码不用动。

看一个简单的封装示例：

class OrderManager:
    def __init__(self, broker_type="huatai"):
        if broker_type == "huatai":
            self.api = HuataiAPI()
        elif broker_type == "guoxin":
            self.api = GuoxinAPI()
        else:
            raise ValueError("不支持的券商类型")

    def buy(self, symbol, price, volume):
        """买入指令"""
        order_id = self.api.place_order(
            symbol=symbol,
            direction="buy",
            price=price,
            volume=volume
        )
        return order_id

    def sell(self, symbol, price, volume):
        """卖出指令"""
        order_id = self.api.place_order(
            symbol=symbol,
            direction="sell",
            price=price,
            volume=volume
        )
        return order_id

    def cancel(self, order_id):
        """撤单"""
        self.api.cancel_order(order_id)

封装的好处是什么？你想想看，如果有一天华泰接口升级了，你只需要改HuataiAPI这个类，上层策略代码一行都不用动。我在项目中吃过亏，一开始没封装，后来换券商，改了三天代码。从那以后，我再也不偷懒了。

知识体系总览

下面这张图，把本章的核心逻辑串起来了。你可以看到，从行情获取到交易执行，整个链路是怎么走的。

这张图把整个流程串起来了。行情数据通过WebSocket或CTP进来，经过你的策略逻辑，生成交易信号，然后通过封装好的指令层，调用券商API去执行。每一层各司其职，出了问题也好排查。

好了，这一章的内容就到这。接口选型、CTP入门、WebSocket订阅、指令封装，这四个点你吃透了，实盘交易的第一步就算站稳了。

---