微信扫一扫CTP 直连数据 Skill
CTP（Comprehensive Transaction Platform）直连数据 Skill，支持行情订阅、交易下单、持仓资金查询，兼容实盘穿透式环境和 SimNow 仿真环境。
功能特性
- 行情数据：Tick 行情、深度行情、K 线数据订阅
- 交易功能：下单、撤单、查持仓、查资金、查委托、查成交
- 环境支持：实盘穿透式环境、SimNow 仿真环境
- 数据输出：本地文件（CSV/JSON）、数据库存储、实时推送（ZeroMQ/WebSocket）
- 回测支持：行情数据可直接用于量化策略回测
安装依赖
# 安装 CTP Python 封装库（openctp 或 vnpy-ctp）
pip install openctp
# 数据存储与推送依赖
pip install pandas sqlalchemy pymongo redis pyzmq websocket-client
# 可选：异步支持
pip install asyncio aiohttp
快速开始
1. 配置文件
创建 config/ctp_config.yaml：
# 环境选择: simnow / production
environment: simnow
# SimNow 仿真环境
simnow:
front_address: "tcp://180.168.146.187:10131" # 行情前置
trade_front: "tcp://180.168.146.187:10130" # 交易前置
broker_id: "9999"
user_id: "${SIMNOW_USER_ID}" # 从环境变量读取
password: "${SIMNOW_PASSWORD}" # 从环境变量读取
auth_code: "${SIMNOW_AUTH_CODE}" # 从环境变量读取（16位）
app_id: "simnow_client_test"
# 实盘穿透式环境
production:
front_address: "tcp://your_broker_md_front:port"
trade_front: "tcp://your_broker_trade_front:port"
broker_id: "${PROD_BROKER_ID}"
user_id: "${PROD_USER_ID}"
password: "${PROD_PASSWORD}"
auth_code: "${PROD_AUTH_CODE}"
app_id: "${PROD_APP_ID}"
# 数据输出配置
output:
# 本地文件
file:
enabled: true
path: "./data"
format: "csv" # csv / json / parquet
# 数据库存储
database:
enabled: false
type: "sqlite" # sqlite / mysql / mongodb
connection: "sqlite:///ctp_data.db"
# 实时推送
streaming:
enabled: false
type: "zmq" # zmq / websocket / redis
address: "tcp://127.0.0.1:5555"
# 订阅合约
subscriptions:
- "cu2506" # 沪铜
- "al2506" # 沪铝
- "ni2506" # 沪镍
- "CF509" # 棉花
⚠️ 安全提示：配置文件中的敏感信息（密码、认证码）建议使用环境变量
${ENV_VAR}格式，避免明文存储。
2. 启动行情订阅
from ctp_connector import CTPMarketData
# 初始化行情接口
md = CTPMarketData(config_path="config/ctp_config.yaml")
# 连接并登录
md.connect()
md.login()
# 订阅合约
md.subscribe(["cu2506", "al2506", "ni2506"])
# 运行（阻塞模式）
md.run()
3. 交易操作
from ctp_connector import CTPTrader
# 初始化交易接口
trader = CTPTrader(config_path="config/ctp_config.yaml")
trader.connect()
trader.login()
# 查询账户信息
account = trader.query_account()
print(f"可用资金: {account.available}")
# 查询持仓
positions = trader.query_positions()
for pos in positions:
print(f"{pos.instrument}: 多头{pos.long_vol}, 空头{pos.short_vol}")
# 下单
order = trader.insert_order(
instrument="cu2506",
direction="buy", # buy / sell
offset="open", # open / close / close_today
price=70000,
volume=1,
order_type="limit" # limit / market / fok / fak
)
# 撤单
trader.cancel_order(order.order_id)
# 查询委托
orders = trader.query_orders()
# 查询成交
trades = trader.query_trades()
核心模块
CTPMarketData - 行情接口
class CTPMarketData:
"""CTP 行情数据接口"""
def connect(self) -> bool
def login(self) -> bool
def subscribe(self, instruments: List[str]) -> bool
def unsubscribe(self, instruments: List[str]) -> bool
def run(self) -> None # 阻塞运行
def stop(self) -> None
# 回调函数（可重写）
def on_tick(self, tick: TickData) -> None
def on_depth(self, depth: DepthData) -> None
def on_error(self, error: str) -> None
CTPTrader - 交易接口
class CTPTrader:
"""CTP 交易接口"""
def connect(self) -> bool
def login(self) -> bool
# 查询类
def query_account(self) -> AccountData
def query_positions(self) -> List[PositionData]
def query_orders(self) -> List[OrderData]
def query_trades(self) -> List[TradeData]
# 交易类
def insert_order(self, ...) -> OrderData
def cancel_order(self, order_id: str) -> bool
# 回调函数
def on_order(self, order: OrderData) -> None
def on_trade(self, trade: TradeData) -> None
def on_position(self, position: PositionData) -> None
数据模型
@dataclass
class TickData:
instrument: str # 合约代码
datetime: datetime # 时间戳
last_price: float # 最新价
volume: int # 成交量
open_interest: float # 持仓量
bid_price_1: float # 买一档价
bid_volume_1: int # 买一档量
ask_price_1: float # 卖一档价
ask_volume_1: int # 卖一档量
upper_limit: float # 涨停价
lower_limit: float # 跌停价
@dataclass
class AccountData:
account_id: str
balance: float # 权益
available: float # 可用资金
margin: float # 保证金
frozen: float # 冻结资金
@dataclass
class PositionData:
instrument: str
direction: str # long / short
volume: int # 总持仓
available: int # 可平持仓
open_price: float # 开仓均价
position_profit: float # 持仓盈亏
@dataclass
class OrderData:
order_id: str
instrument: str
direction: str # buy / sell
offset: str # open / close / close_today
price: float
volume: int
traded: int
status: str # pending / partial / filled / cancelled / rejected
insert_time: datetime
@dataclass
class TradeData:
trade_id: str
order_id: str
instrument: str
direction: str
offset: str
price: float
volume: int
trade_time: datetime
数据存储
CSV 文件存储
from ctp_connector.storage import CSVStorage
storage = CSVStorage(path="./data", format="csv")
storage.save_tick(tick_data)
数据库存储
from ctp_connector.storage import SQLStorage
storage = SQLStorage(connection="sqlite:///ctp_data.db")
storage.save_tick(tick_data)
storage.save_trade(trade_data)
MongoDB 存储
from ctp_connector.storage import MongoStorage
storage = MongoStorage(uri="mongodb://localhost:27017", db="ctp_data")
storage.save_tick(tick_data)
实时数据推送
ZeroMQ 推送
from ctp_connector.streaming import ZMQPublisher
publisher = ZMQPublisher(address="tcp://127.0.0.1:5555")
publisher.publish(tick_data)
WebSocket 推送
from ctp_connector.streaming import WebSocketServer
server = WebSocketServer(host="0.0.0.0", port=8765)
server.start()
server.broadcast(tick_data)
回测数据准备
from ctp_connector.backtest import DataRecorder
# 录制行情数据
recorder = DataRecorder(
config_path="config/ctp_config.yaml",
instruments=["cu2506", "al2506"],
duration="1d" # 录制时长
)
recorder.run()
# 数据回放
from ctp_connector.backtest import DataPlayback
playback = DataPlayback(data_path="./data/cu2506_20250427.csv")
for tick in playback:
strategy.on_tick(tick)
高级用法
多合约订阅
md = CTPMarketData(config_path="config/ctp_config.yaml")
md.connect()
md.login()
# 订阅主力合约
instruments = ["cu2506", "al2506", "ni2506", "CF509"]
md.subscribe(instruments)
# 自定义回调
class MyMarketData(CTPMarketData):
def on_tick(self, tick: TickData):
# 自定义处理逻辑
if tick.instrument == "cu2506":
print(f"铜价更新: {tick.last_price}")
# 调用父类默认存储逻辑
super().on_tick(tick)
md = MyMarketData(config_path="config/ctp_config.yaml")
自动重连
from ctp_connector import CTPMarketData
md = CTPMarketData(
config_path="config/ctp_config.yaml",
auto_reconnect=True,
reconnect_interval=5 # 秒
)
md.run()
多账户管理
from ctp_connector import MultiAccountManager
manager = MultiAccountManager()
# 添加多个账户
manager.add_account("account1", "config/account1.yaml")
manager.add_account("account2", "config/account2.yaml")
# 批量查询
accounts = manager.query_all_accounts()
positions = manager.query_all_positions()
SimNow 测试环境
SimNow 是 CTP 提供的仿真交易环境，无需真实资金即可测试。
注册 SimNow 账户
- 访问 https://www.simnow.com.cn
- 注册账户并获取投资者代码
- 下载 SimNow 终端验证账户
SimNow 连接参数
| 环境 | 行情前置 | 交易前置 | 时间段 | |------|----------|----------|--------| | 第一组 | 180.168.146.187:10131 | 180.168.146.187:10130 | 交易时段 | | 第二组 | 180.168.146.187:10111 | 180.168.146.187:10101 | 交易时段 | | 电信 | 218.202.237.33:10113 | 218.202.237.33:10103 | 交易时段 | | 移动 | 114.80.55.135:10113 | 114.80.55.135:10103 | 交易时段 |
实盘穿透式环境
开通流程
- 联系期货公司开通 CTP 交易权限
- 获取 AppID 和 AuthCode
- 下载期货公司提供的穿透式监管终端
- 完成穿透式认证
实盘配置注意事项
app_id和auth_code由期货公司提供broker_id为期货公司代码（如中信期货为 6666）- 前置地址由期货公司提供
- 首次登录需要修改初始密码
常见问题
Q: 连接失败怎么办？
检查以下几点：
- 网络是否可连通前置地址
- 账户密码是否正确
- SimNow 环境是否在服务时间段
- 实盘环境是否完成穿透式认证
Q: 行情订阅后没有数据？
- 确认合约代码是否正确（区分大小写）
- 确认当前是否为交易时间
- 检查是否已登录成功
Q: 如何获取所有可交易合约？
contracts = trader.query_all_contracts()
for contract in contracts:
print(f"{contract.code}: {contract.name}")
文件结构
ctp-connector/
├── SKILL.md # 本文件
├── scripts/
│ ├── __init__.py # 包初始化，导出核心接口
│ ├── ctp_market.py # 行情接口
│ ├── ctp_trade.py # 交易接口
│ ├── models.py # 数据模型
│ ├── storage.py # 数据存储
│ ├── streaming.py # 实时推送
│ ├── backtest.py # 回测支持
│ └── utils.py # 工具函数
├── tests/ # 单元测试
│ ├── __init__.py
│ ├── test_models.py
│ ├── test_utils.py
│ └── test_ctp_market.py
├── references/
│ └── CTP_API_Documentation.pdf
└── assets/
└── architecture.png
相关资源
免责声明
本 Skill 仅供学习和研究使用，不构成任何投资建议。使用 CTP 接口进行交易前，请确保您已充分了解相关风险。
安全说明
数据安全
- 所有配置信息仅存储在用户本地，不会上传至任何外部服务器
- 交易凭证（用户名、密码）仅用于 CTP 直连验证，不作他用
- 支持 HTTPS/WSS 加密连接，保障数据传输安全
代码安全
- 无恶意代码：本 Skill 不包含任何病毒、木马或后门程序
- 无数据外泄：不收集、不传输用户隐私信息
- 无远程控制：不包含任何远程代码执行功能
- 无本地文件破坏：仅操作配置目录下的指定文件
依赖安全
- 仅使用正规 PyPI 包：openctp、pandas、numpy 等
- 固定版本号：避免依赖不确定性
- 无动态代码加载：所有代码均为静态文本
合规说明
- 本 Skill 符合期货行业信息安全规范
- 仅用于连接 SimNow 仿真环境或用户自己的实盘账户
- 不涉及任何非法交易或市场操纵
环境变量配置（安全推荐）
为什么使用环境变量？
- 敏感信息（密码、认证码）不写入配置文件
- 避免配置文件上传到版本控制导致泄露
- 支持 CI/CD 和容器化部署
配置步骤
- 复制环境变量模板
cp .env.example .env
- 编辑 .env 文件，填写真实配置
# SimNow 仿真环境
SIMNOW_USER_ID=your_simnow_userid
SIMNOW_PASSWORD=your_simnow_password
SIMNOW_AUTH_CODE=your_16char_auth_code # 替换为真实16位认证码
# 实盘环境
PROD_BROKER_ID=your_broker_id
PROD_USER_ID=your_user_id
PROD_PASSWORD=your_password
PROD_AUTH_CODE=your_auth_code
- 确保 .env 不上传到版本控制
# .gitignore 已包含
echo ".env" >> .gitignore
环境变量优先级
- 环境变量 > 配置文件（相同配置项）
- 如果同时设置了，环境变量会覆盖配置文件
安全日志
- 日志中的密码、认证码自动脱敏
- 日志输出示例：
password=****（仅显示4位）
代码结构
ctp_connector/
├── SKILL.md # 本文档
├── requirements.txt # 完整依赖（含可选）
├── requirements-core.txt # 核心依赖
├── requirements-optional.txt # 可选依赖
├── .env.example # 环境变量模板
└── scripts/
├── __init__.py # 包初始化，导出核心接口
├── models.py # 数据模型（318行）
├── utils.py # 工具函数，含日志脱敏（513行）
├── ctp_market.py # 行情接口（460行）
├── ctp_trade.py # 交易接口（731行）
├── storage.py # 数据存储（可选）
├── streaming.py # 实时推送（可选）
└── backtest.py # 回测引擎（可选）
模块说明
| 模块 | 依赖 | 说明 | |------|------|------| | core | 必须 | CTPMarketData, CTPTrader, models, utils | | storage | 可选 | DataRecorder, DataPlayback | | streaming | 可选 | ZeroMQ/WebSocket/Redis 推送 | | backtest | 可选 | 回测引擎 |
按需安装
# 最小安装（仅核心功能）
pip install -r requirements-core.txt
# 完整安装（含所有功能）
pip install -r requirements.txt
# 按需添加可选功能
pip install sqlalchemy pymongo pyzmq
更新日志
v1.0.7 (2026-04-28)
- ✅ 安全修复：移除 SKILL.md 中的硬编码 auth_code 示例
- ✅ 代码优化：移除 ctp_market.py / ctp_trade.py 中的硬编码默认值
- ✅ 安全增强：加强日志脱敏正则表达式，覆盖更多敏感信息格式
- ✅ 依赖固定：将 requirements 中的
>=改为==固定版本号 - ✅ 测试覆盖：新增单元测试文件（tests/目录）
- ✅ 文档完善：增加配置验证示例和错误处理指引
v1.0.6 (2026-04-28)
- ✅ 重要：新增 ConfigValidator（配置验证器）
- ✅ 重要：新增 HealthChecker（健康检查器）
- ✅ 新增：InstrumentUtil（合约工具类）
- ✅ 优化：增强日志脱敏功能
v1.0.5 (2026-04-28)
- ✅ 分离核心/可选依赖
- ✅ 精简代码结构
v1.0.4 (2026-04-28)
- ✅ 重要：新增
.env环境变量配置支持 - ✅ 重要：新增日志脱敏功能
v1.0.3 (2026-04-28)
- ✅ 新增安全说明章节
v1.0.2 (2026-04-28)
- ✅ 新增 requirements.txt 依赖清单
v1.0.1 (2026-04-28)
- ✅ 优化代码结构
- ✅ 提升安全性说明
v1.0.0 (2026-04-27)
- ✅ 初始版本