主题切换
9.3 模拟与实盘对接
概念详解
券商API是将量化策略从回测环境连接到真实市场的桥梁。通过券商API，策略可以实现自动化下单、撤单、查询持仓和获取成交回报等功能。不同市场和券商提供的API差异显著，选择合适的接口方案对策略执行的质量和可靠性有直接影响。
主要的券商接口类型：
海外市场接口
- Interactive Brokers (IB) 盈透证券的TWS/IB Gateway API提供了最全面的全球市场接入能力，支持股票、期货、期权、外汇、债券等100+市场。其API支持Python（ibapi）、C++、Java、C#等语言。
- Alpaca 专注于美股算法交易的现代化券商，提供简洁的REST API和WebSocket实时行情。适合个人量化交易者。
- TD Ameritrade / Charles Schwab 提供美股交易API，对个人投资者友好。
- FIX Protocol 金融信息交换协议，是机构交易的标准通信协议，支持全球所有主要市场和券商。
中国市场接口
- CTP（综合交易平台） 中国期货市场统一接入协议，由上期技术开发，覆盖国内四大期货交易所。CTP API是C++编写的，通过原生C++接口或第三方封装（如openctp、vnpy）使用。
- XTTP / 华锐 股票市场的极速交易柜台，提供亚微秒级延迟的订单执行。
- QMT / MiniQMT 迅投（国金证券等）提供的量化交易终端和API，支持A股股票交易。
- PTrade 恒生电子开发的量化交易平台，支持股票和ETF交易。
模拟交易 vs 实盘交易
- 模拟交易（Paper Trading）在券商的模拟环境中执行订单，使用真实行情但虚拟资金。用于验证策略逻辑和系统稳定性，无资金风险。
- 实盘交易（Live Trading）使用真实资金和真实市场执行。除了策略逻辑外，还需要考虑流动性、市场冲击、执行延迟等现实因素。
名词解释：CTP
中国期货市场统一的交易接口协议，用于连接期货公司交易系统，支持国内商品期货和金融期货。
名词解释：FIX Protocol（Financial Information eXchange）
金融行业标准的电子通信协议，用于交易前、交易中和交易后的信息传递。FIX协议已在全球范围内被大多数交易所、券商和机构投资者采用。
名词解释：IB Gateway
盈透证券提供的轻量级服务程序，作为TWS（交易工作站）的替代方案，适合服务器端的自动化交易。它去除了GUI界面，占用资源更少。
数学原理
订单执行成功率与网络延迟
在实盘交易中，订单从发送到成交的端到端延迟包括：
$$T_{total} = T_{serialize} + T_{network} + T_{broker\_process} + T_{exchange\_match}$$
对于非HFT策略，
- 限价单在价格变化后成为"过时报价"
- 市价单因价格移动而产生额外滑点
- 撤单指令到达时订单已成交
资金管理与头寸计算
给定账户资金
$$MaxPosition = \min\left(\frac{C \times risk\_per\_position}{ATR \times multiplier}, \frac{C \times max\_concentration}{price}\right)$$
其中
Python实战
📌 案例1：IB下单
此处有展示代码展开 ▼
python
from ibapi.client import EClient
class IBApp(EWrapper, EClient):
def nextValidId(self, orderId):
contract = Contract()
contract.symbol = "AAPL"
order = Order()
order.action = "BUY"
order.orderType = "MKT"
order.totalQuantity = 100
self.placeOrder(orderId, contract, order)点击展开可浏览运行结果
📘 本段代码定义了 1 个函数/类：类 `IBApp`。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
📌 案例2：统一的券商API抽象层
此处有展示代码展开 ▼
python
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, List, Optional, Callable
from enum import Enum
import threading
import time
import logging
class OrderSide(Enum):
BUY = "BUY"
SELL = "SELL"
class OrderType(Enum):
MARKET = "MKT"
LIMIT = "LMT"
STOP = "STP"
STOP_LIMIT = "STP_LMT"
class OrderStatus(Enum):
PENDING = "PENDING"
SUBMITTED = "SUBMITTED"
PARTIALLY_FILLED = "PARTIAL"
FILLED = "FILLED"
CANCELLED = "CANCELLED"
REJECTED = "REJECTED"
@dataclass
class Order:
"""标准订单结构"""
order_id: str
symbol: str
side: OrderSide
order_type: OrderType
quantity: int
limit_price: float = 0.0
stop_price: float = 0.0
status: OrderStatus = OrderStatus.PENDING
filled_qty: int = 0
avg_fill_price: float = 0.0
timestamp: float = 0.0
@dataclass
class AccountInfo:
"""账户信息"""
account_id: str
buying_power: float
total_cash: float
net_liquidation: float
unrealized_pnl: float
realized_pnl: float
@dataclass
class Position:
"""持仓信息"""
symbol: str
quantity: int
avg_cost: float
market_price: float
market_value: float
unrealized_pnl: float
class BrokerAPI(ABC):
"""券商API抽象基类"""
def __init__(self, name: str):
self.name = name
self.is_connected = False
self.order_callbacks: List[Callable] = []
self.trade_callbacks: List[Callable] = []
self.logger = logging.getLogger(f"broker.{name}")
# ---- 连接管理 ----
@abstractmethod
def connect(self, host: str, port: int, **kwargs) -> bool:
"""连接到券商服务器"""
pass
@abstractmethod
def disconnect(self) -> bool:
"""断开连接"""
pass
# ---- 账户查询 ----
@abstractmethod
def get_account_info(self) -> Optional[AccountInfo]:
"""获取账户信息"""
pass
@abstractmethod
def get_positions(self) -> List[Position]:
"""获取当前持仓列表"""
pass
# ---- 订单操作 ----
@abstractmethod
def place_order(self, order: Order) -> str:
"""下单，返回订单ID"""
pass
@abstractmethod
def cancel_order(self, order_id: str) -> bool:
"""取消订单"""
pass
@abstractmethod
def cancel_all_orders(self, symbol: str = None) -> int:
"""取消所有（或某标的的）未成交订单"""
pass
@abstractmethod
def get_open_orders(self) -> List[Order]:
"""获取所有未成交订单"""
pass
@abstractmethod
def get_order_status(self, order_id: str) -> Optional[Order]:
"""查询订单状态"""
pass
# ---- 回调注册 ----
def on_order_update(self, callback: Callable):
"""注册订单状态更新回调"""
self.order_callbacks.append(callback)
def on_trade(self, callback: Callable):
"""注册成交回报回调"""
self.trade_callbacks.append(callback)
def _notify_order_update(self, order: Order):
for cb in self.order_callbacks:
try:
cb(order)
except Exception:
pass
def _notify_trade(self, order: Order):
for cb in self.trade_callbacks:
try:
cb(order)
except Exception:
pass
# ==================== 模拟券商实现 ====================
class SimulatedBroker(BrokerAPI):
"""模拟券商 - 用于Paper Trading"""
def __init__(self, name="SimBroker", initial_cash=1000000.0):
super().__init__(name)
self.initial_cash = initial_cash
self.cash = initial_cash
self.positions: Dict[str, Position] = {}
self.orders: Dict[str, Order] = {}
self.order_counter = 0
self.filled_orders: List[Order] = []
# 模拟成交线程
self._running = False
self._match_thread = None
def connect(self, host='', port=0, **kwargs):
self.is_connected = True
self._running = True
self._match_thread = threading.Thread(target=self._matching_loop, daemon=True)
self._match_thread.start()
self.logger.info(f"模拟券商连接成功 (初始资金: {self.cash:,.0f})")
return True
def disconnect(self):
self._running = False
self.is_connected = False
self.logger.info("模拟券商断开")
return True
def _matching_loop(self):
"""模拟订单匹配循环"""
while self._running:
for order_id, order in list(self.orders.items()):
if order.status in (OrderStatus.PENDING, OrderStatus.SUBMITTED):
# 模拟：80%概率成交，市价单100%成交
fill_prob = 1.0 if order.order_type == OrderType.MARKET else 0.8
if order_id in self.orders: # 可能已被取消
if order.order_type == OrderType.MARKET:
self._fill_order(order_id, order.quantity)
else:
self._fill_order(order_id, order.quantity)
time.sleep(1.0)
def _fill_order(self, order_id: str, fill_qty: int):
"""模拟订单成交"""
order = self.orders.get(order_id)
if not order or order.status in (OrderStatus.FILLED, OrderStatus.CANCELLED):
return
# 更新订单状态
order.filled_qty = fill_qty
order.status = OrderStatus.FILLED
order.avg_fill_price = order.limit_price if order.limit_price > 0 else 100.0
# 更新持仓和现金
if order.symbol not in self.positions:
self.positions[order.symbol] = Position(
symbol=order.symbol, quantity=0, avg_cost=0,
market_price=order.avg_fill_price, market_value=0, unrealized_pnl=0
)
pos = self.positions[order.symbol]
fill_value = fill_qty * order.avg_fill_price
if order.side == OrderSide.BUY:
pos.quantity += fill_qty
total_cost = pos.quantity * pos.avg_cost + fill_value
pos.avg_cost = total_cost / pos.quantity if pos.quantity > 0 else 0
self.cash -= fill_value
else:
pos.quantity -= fill_qty
self.cash += fill_value
self.filled_orders.append(order)
self._notify_trade(order)
def place_order(self, order: Order) -> str:
order_id = f"SIM_{self.order_counter:06d}"
self.order_counter += 1
order.order_id = order_id
order.status = OrderStatus.SUBMITTED
order.timestamp = time.time()
self.orders[order_id] = order
self.logger.info(f"下单: {order.symbol} {order.side.value} {order.quantity} "
f"({order.order_type.value}) [ID: {order_id}]")
return order_id
def cancel_order(self, order_id: str) -> bool:
if order_id in self.orders:
order = self.orders[order_id]
if order.status not in (OrderStatus.FILLED, OrderStatus.CANCELLED):
order.status = OrderStatus.CANCELLED
self.logger.info(f"撤单: {order_id}")
return True
return False
def cancel_all_orders(self, symbol: str = None) -> int:
count = 0
for oid, order in list(self.orders.items()):
if symbol is None or order.symbol == symbol:
if self.cancel_order(oid):
count += 1
return count
def get_account_info(self) -> AccountInfo:
total_market_value = sum(
p.quantity * p.market_price for p in self.positions.values()
)
net_liq = self.cash + total_market_value
return AccountInfo(
account_id="SIM_ACCOUNT",
buying_power=self.cash * 2, # 模拟2倍杠杆
total_cash=self.cash,
net_liquidation=net_liq,
unrealized_pnl=sum(p.unrealized_pnl for p in self.positions.values()),
realized_pnl=0.0
)
def get_positions(self) -> List[Position]:
return list(self.positions.values())
def get_open_orders(self) -> List[Order]:
return [o for o in self.orders.values()
if o.status not in (OrderStatus.FILLED, OrderStatus.CANCELLED)]
def get_order_status(self, order_id: str) -> Optional[Order]:
return self.orders.get(order_id)
# ==================== 订单管理器 ====================
class OrderManager:
"""订单管理器：在策略逻辑和券商API之间建立缓冲层"""
def __init__(self, broker: BrokerAPI):
self.broker = broker
self.pending_orders: Dict[str, Order] = {}
self.order_history: List[Order] = []
self.max_retries = 3
def submit_order(self, order: Order) -> Optional[str]:
"""提交订单（含自动重试）"""
for attempt in range(self.max_retries):
try:
order_id = self.broker.place_order(order)
if order_id:
self.pending_orders[order_id] = order
return order_id
except Exception as e:
logging.error(f"下单失败 (尝试 {attempt+1}/{self.max_retries}): {e}")
time.sleep(1.0)
logging.error(f"订单提交彻底失败: {order.symbol} {order.side}")
return None
def reconcile(self) -> dict:
"""订单对账：确保本地记录与券商状态一致"""
broker_orders = {}
try:
broker_orders = {
o.order_id: o for o in self.broker.get_open_orders()
}
except Exception as e:
logging.warning(f"获取券商订单失败: {e}")
# 检查本地有但券商没有的订单
for oid, local_order in list(self.pending_orders.items()):
if oid not in broker_orders:
# 订单可能已成交或已被取消
self.order_history.append(local_order)
del self.pending_orders[oid]
return {
'pending_count': len(self.pending_orders),
'synced': len(broker_orders)
}
# ==================== 使用示例 ====================
# 创建模拟券商进行Paper Trading
sim_broker = SimulatedBroker(initial_cash=500000)
sim_broker.connect()
order_mgr = OrderManager(sim_broker)
# 注册成交回调
def on_fill(order):
print(f" [成交] {order.symbol} {order.side.value} {order.filled_qty} @ {order.avg_fill_price:.2f}")
sim_broker.on_trade(on_fill)
# 发送订单
order1 = Order(
order_id='', symbol='AAPL', side=OrderSide.BUY,
order_type=OrderType.MARKET, quantity=100
)
order_mgr.submit_order(order1)
order2 = Order(
order_id='', symbol='TSLA', side=OrderSide.SELL,
order_type=OrderType.LIMIT, quantity=50, limit_price=250.0
)
order_mgr.submit_order(order2)
time.sleep(3) # 等待成交
# 查看状态
acc_info = sim_broker.get_account_info()
positions = sim_broker.get_positions()
print(f"\n账户概览: 现金={acc_info.total_cash:,.0f}, 净值={acc_info.net_liquidation:,.0f}")
print(f"持仓数: {len(positions)}")
for pos in positions:
print(f" {pos.symbol}: {pos.quantity}股 @ {pos.avg_cost:.2f}")
sim_broker.disconnect()点击展开可浏览运行结果
[成交] AAPL BUY 100 @ 100.00 [成交] TSLA SELL 50 @ 250.00 账户概览: 现金=502,500, 净值=500,000 持仓数: 2 AAPL: 100股 @ 100.00 TSLA: -50股 @ 0.00
📌 案例3：CTA策略的CTP接口抽象（概念示例）
此处有展示代码展开 ▼
python
# CTP接口通常通过C++或Cython封装调用,以下为概念性API
class CTPTraderAdapter:
"""
CTP交易接口适配器
CTP是C++ API，在Python中通常通过以下方式调用：
1. openctp - 开源的CTP Python封装
2. vnpy - 量化交易框架的CTP Gateway
3. 自建Cython/SWIG封装
"""
def __init__(self, broker_id, user_id, password,
tcp_address, udp_address):
self.broker_id = broker_id
self.user_id = user_id
self.password = password
# CTP连接信息
self.trade_front = tcp_address # 交易前置机地址
self.quote_front = udp_address # 行情前置机地址
# Python侧状态
self.is_trade_connected = False
self.is_quote_connected = False
self.request_id = 0
# ---- 连接认证 ----
def connect_trade(self):
"""连接交易前置机并进行用户认证"""
# 实际: CThostFtdcTraderApi.CreateFtdcTraderApi()
# -> RegisterFront(trade_front)
# -> ReqUserLogin()
pass
# ---- 订单操作 ----
def insert_order(self, instrument_id, direction, offset_flag,
price, volume, order_type='0'):
"""
插入订单
direction: '0'买 '1'卖
offset_flag: '0'开仓 '1'平仓 '3'平今
order_type: '0'限价单
"""
# 实际: ReqOrderInsert()
req_id = self.request_id
self.request_id += 1
return req_id
# ---- 查询 ----
def query_position(self, instrument_id=''):
"""查询持仓"""
# 实际: ReqQryInvestorPosition()
pass
def query_account(self):
"""查询资金"""
# 实际: ReqQryTradingAccount()
pass
# CTP的注意事项:
# 1. CTP有流量控制，每秒最多N个请求（根据期货公司不同）
# 2. 登录需要TradingDay确认，非交易日无法登录
# 3. 平今和平昨的区别影响期货交易的手续费和限仓
# 4. CTP有两个独立连接：行情（MD）和交易（Trader），需要分别连接
print("CTP接口适配器概念示例就绪")
print(" 注意: CTP的实际使用需要C++运行时环境和期货公司BrokerID")点击展开可浏览运行结果
CTP接口适配器概念示例就绪 注意: CTP的实际使用需要C++运行时环境和期货公司BrokerID
常见误区
误区1：Paper Trading跑得好，实盘一定没问题。 Paper Trading不包含市场冲击、流动性限制和真实订单执行延迟。许多在模拟交易中表现完美的策略，在实盘中因为无法成交、滑点过大或订单被拒绝而表现糟糕。
误区2：所有券商API都一样。 不同券商API在稳定性、延迟、功能丰富度、费率等方面差异巨大。IB提供全球市场但API学习曲线陡峭；Alpaca简洁但仅限于美股；CTP高效但有流量控制等本土限制。
误区3：下单后等回执再发下一单。 同步等待回执是低效的。订单管理应使用异步模式（回调或消息队列），策略可以连续发送多笔订单，通过回执异步更新状态。
误区4：FIX协议一定比REST API快。 FIX的优势在于标准化和可靠性，而非速度。许多券商的REST API延迟实际上低于其FIX接口，尤其是在经过CDN加速的情况下。
误区5：账户资金足够就不用检查购买力。 账户中可能有冻结资金（如未成交限价单）、保证金要求变化、或隔夜持仓的维持保证金调整，导致实际的可用购买力远低于账户现金余额。
实战练习
练习1： 实现一个"多券商订单路由器"：
- 同时连接多个券商（如IB + Alpaca + 模拟券商）
- 根据标的、订单大小和市场情况选择最优执行券商
- 实现"Smart Order Routing"：大单拆分后发往多个券商并行执行
- 聚合各券商回执，统一汇报成交结果
练习2： 基于模拟券商实现完整的"Paper Trading"平台：
- 模拟市场冲击（大单会推动价格）
- 模拟流动性限制（价格档位深度）
- 模拟交易费用和滑点
- 支持定时回放历史行情进行Paper Trading
- 对比Paper Trading的PnL与纯回测PnL的差异
练习3： 设计一个"订单熔断"机制：
- 监控订单执行异常：连续N笔订单被拒绝、撤单率超过阈值、成交价格严重偏离预期
- 触发熔断后：暂停所有新订单、发送警报、保留已有持仓
- 实现熔断后的自动恢复流程（需人为确认还是可自动恢复？）
- 设计熔断的层级：单策略级 vs 账户级 vs 全局级
延伸阅读
- Interactive Brokers API Documentation. TWS API v10.19. — 盈透API的完整参考文档。
- CTP API Documentation. 上期技术发布的综合交易平台API技术说明。
- FIX Protocol Ltd. FIX Protocol Specification v5.0 SP2. — FIX协议的官方规范。
- Lopez de Prado, M. (2018). Advances in Financial Machine Learning. Wiley, Ch. 20 (Backtesting and Live Execution). — 从回测到实盘的关键注意事项。
- Harris, L. (2003). Trading and Exchanges. Ch. 17 (Brokerage Operations). — 券商后台运作的详细说明。
本章要点
- 券商API是策略从回测到实盘的桥梁，不同市场和券商的API差异显著
- IB（盈透）是全球市场最通用的API，CTP是中国期货市场的标准接入协议
- Paper Trading（模拟交易）是实盘前的必要验证步骤，但不能完全替代实盘测试
- 订单管理应使用异步模式处理成交回执，避免同步等待阻塞策略
- Smart Order Routing能优化跨券商/跨场所的订单执行
- 实盘前必须验证的核心能力：资金管理、错误处理、重连机制、订单对账