从零搭建Python期货行情监控系统：基于CtpPlus的实战指南

引言

在量化交易的世界里，实时获取期货行情数据是构建任何策略的第一步。对于刚接触这个领域的Python开发者来说，CTP接口常常像一堵高墙——官方只提供C++版本，文档晦涩难懂，入门曲线陡峭。这就是为什么CtpPlus这样的Python封装库如此重要：它让开发者能用熟悉的Python语法直接调用CTP接口，把学习成本从"理解整个期货交易系统"降低到"写几行Python代码"。

本文将带你用CtpPlus搭建一个监控螺纹钢期货(rb2410)的实时行情系统。不同于复杂的量化框架，我们聚焦于一个具体目标： 在控制台打印实时行情数据 。这个看似简单的功能，却是许多量化策略的起点。你会学到如何配置SimNow模拟账户、理解CTP的回调机制，并最终实现一个持续运行的行情监控脚本。

1. 环境准备与CtpPlus安装

1.1 Python环境配置

建议使用Python 3.7-3.9版本，这是目前CtpPlus测试最充分的版本范围。使用conda创建独立环境：

conda create -n ctp_env python=3.8
conda activate ctp_env

1.2 安装CtpPlus

CtpPlus可以通过pip直接安装，它已经打包了CTP的底层C++库，无需单独下载：

pip install CtpPlus

注意：在Windows上可能需要安装Visual C++ Redistributable，Linux则需要glibc 2.17以上版本。

1.3 获取SimNow测试账号

SimNow是中国金融期货交易所提供的官方模拟环境，注册流程：

1. 访问 SimNow官网
2. 完成手机号验证和基本信息填写
3. 系统会自动分配一个6位数的investor_id和密码

重要提示 ：首次登录后，建议修改默认密码，并记录以下关键信息：

• BrokerID: 9999 (SimNow固定值)
• AppID: simnow_client_test
• AuthCode: 0000000000000000

2. 理解CTP行情接口的工作机制

2.1 CTP的异步回调模型

与传统API不同，CTP采用事件驱动架构。当行情到达时，系统会主动调用你预先定义的回调函数。这种设计虽然高效，但对初学者可能反直觉。主要回调函数包括：

回调函数	触发时机	典型用途
OnRspUserLogin	登录完成时	检查登录状态
OnRtnDepthMarketData	新行情到达时	处理tick数据
OnRspSubMarketData	订阅合约后	确认订阅成功

2.2 CtpPlus的简化封装

CtpPlus通过继承机制简化了回调处理。你只需要重写关心的方法，比如：

class MyMdHandler(MdApiBase):
    def OnRtnDepthMarketData(self, pDepthMarketData):
        # 在这里处理实时行情
        print(f"时间: {pDepthMarketData.UpdateTime}, 最新价: {pDepthMarketData.LastPrice}")

这种设计避免了复杂的回调注册流程，让代码更符合Pythonic风格。

3. 构建行情监控脚本

3.1 账户配置

创建 config.py 存放账户信息（切勿将真实账户信息提交到版本控制）：

SIMNOW_ACCOUNT = {
    "investor_id": "你的SimNow账号",
    "password": "你的密码",
    "broker_id": "9999",
    "md_server": "180.168.146.187:10211",  # 电信1行情服务器
    "app_id": "simnow_client_test",
    "auth_code": "0000000000000000"
}

3.2 主程序实现

完整的 monitor.py 脚本：

from CtpPlus.CTP.MdApiBase import MdApiBase
from CtpPlus.CTP.FutureAccount import FutureAccount
import config

class MarketMonitor(MdApiBase):
    def __init__(self, account):
        super().__init__()
        self.account = account
        
    def OnRspUserLogin(self, pRspUserLogin, pRspInfo, nRequestID, bIsLast):
        if pRspInfo.ErrorID == 0:
            print("登录成功，开始订阅行情...")
            self.SubscribeMarketData([b'rb2410'])  # 订阅螺纹钢2410合约
        else:
            print(f"登录失败: {pRspInfo.ErrorMsg}")
    
    def OnRtnDepthMarketData(self, pDepthMarketData):
        instrument = pDepthMarketData.InstrumentID.decode('gbk')
        print(f"""
合约: {instrument}
时间: {pDepthMarketData.UpdateTime}.{pDepthMarketData.UpdateMillisec}
最新价: {pDepthMarketData.LastPrice}
买一价: {pDepthMarketData.BidPrice1} 量: {pDepthMarketData.BidVolume1}
卖一价: {pDepthMarketData.AskPrice1} 量: {pDepthMarketData.AskVolume1}
成交量: {pDepthMarketData.Volume}  持仓量: {pDepthMarketData.OpenInterest}
        """)

if __name__ == "__main__":
    account = FutureAccount(
        broker_id=config.SIMNOW_ACCOUNT["broker_id"],
        server_dict={"MDServer": config.SIMNOW_ACCOUNT["md_server"]},
        investor_id=config.SIMNOW_ACCOUNT["investor_id"],
        password=config.SIMNOW_ACCOUNT["password"],
        app_id=config.SIMNOW_ACCOUNT["app_id"],
        auth_code=config.SIMNOW_ACCOUNT["auth_code"],
        subscribe_list=[]
    )
    
    monitor = MarketMonitor(account)
    monitor.Create()
    monitor.Join()

3.3 关键点解析

1. 编码问题 ：CTP接口返回的数据是GBK编码，需要使用 decode('gbk') 转换
2. 订阅合约 ：合约代码需要转为bytes类型，如 b'rb2410'
3. 连接流程 ：
   • Create() : 初始化连接
   • Join() : 进入事件循环，保持程序运行

4. 进阶功能与调试技巧

4.1 多合约监控

修改订阅列表即可监控多个合约：

self.SubscribeMarketData([b'rb2410', b'cu2409', b'au2412'])

4.2 数据存储

将行情数据保存到CSV：

import csv

class MarketMonitor(MdApiBase):
    def __init__(self, account):
        self.writer = csv.writer(open('market_data.csv', 'a'))
        
    def OnRtnDepthMarketData(self, pDepthMarketData):
        row = [
            pDepthMarketData.InstrumentID.decode('gbk'),
            f"{pDepthMarketData.UpdateTime}.{pDepthMarketData.UpdateMillisec}",
            pDepthMarketData.LastPrice,
            pDepthMarketData.Volume
        ]
        self.writer.writerow(row)

4.3 常见问题排查

问题现象	可能原因	解决方案
登录失败	账号密码错误	检查SimNow账号状态
无行情返回	合约未订阅成功	确认合约代码正确
连接断开	网络问题	尝试切换服务器(电信1/电信2)

5. 从监控到交易：下一步学习路径

完成行情监控后，你可能会想：

• 如何基于这些数据生成交易信号？
• 怎样将监控脚本升级为自动化交易系统？

这需要了解CTP的交易接口(TraderApi)。与行情接口类似，但增加了以下关键功能：

from CtpPlus.CTP.TraderApiBase import TraderApiBase

class MyTrader(TraderApiBase):
    def OnRspOrderInsert(self, pInputOrder, pRspInfo, nRequestID, bIsLast):
        # 处理报单回报
        pass
        
    def buy_open(self, exchange_id, instrument_id, price, volume):
        # 实现买入开仓逻辑
        pass

建议的学习路线：

1. 先熟悉行情接口（本文内容）
2. 学习查询账户资金和持仓
3. 尝试限价单的发送
4. 最后实现完整的策略逻辑