连接交易通道的基础设施
量化交易实现自动化的起点是获取稳定的API接口。股票市场常用券商提供的HTTP或WebSocket接口，期货市场则多采用CTP（综合交易平台）接口。这些接口允许程序直接发送委托、查询持仓与资金、订阅行情。构建自动化系统前，需要向经纪商申请交易权限，获取API密钥、账户ID和服务器地址。
一个健壮的连接层需要处理网络断开、会话过期和频率限制。股票API通常限制每秒请求次数，期货CTP接口有流控和重连机制。使用Python的requests或websocket-client库时，应封装重试逻辑与心跳检测。
import requests
import time
class BrokerAPI:
def __init__(self, base_url, api_key):
self.base_url = base_url
self.headers = {'Authorization': f'Bearer {api_key}'}
self.session = requests.Session()
def place_order(self, symbol, side, quantity, price=None, order_type='LIMIT'):
payload = {
'symbol': symbol,
'side': side,
'quantity': quantity,
'price': price,
'type': order_type
}
for attempt in range(3):
try:
resp = self.session.post(f'{self.base_url}/orders',
json=payload, headers=self.headers, timeout=5)
if resp.status_code == 200:
return resp.json()
except requests.exceptions.RequestException:
time.sleep(1)
raise ConnectionError('下单失败，重试三次')
策略信号生成与订单执行
自动化交易的核心是把策略信号转化为实际委托。股票量化常用双均线、布林带或因子模型，期货量化侧重趋势跟踪与套利。信号生成模块输出目标仓位，订单执行模块计算与当前持仓的差额，再拆分成小单以降低市场冲击。
执行环节要区分限价单与市价单。股票市场流动性好，可用限价单挂单；期货市场夜盘波动大，趋势策略常使用市价单或对手价快速成交。无论哪种，都需要记录每笔订单的本地状态，防止重复下单。
class OrderExecutor:
def __init__(self, api, max_position):
self.api = api
self.max_position = max_position
self.current_position = 0
def execute_signal(self, symbol, target_position, price):
delta = target_position - self.current_position
if abs(delta) < 1e-6:
return
if abs(target_position) > self.max_position:
target_position = self.max_position * (1 if delta > 0 else -1)
side = 'BUY' if delta > 0 else 'SELL'
quantity = abs(delta)
order = self.api.place_order(symbol, side, quantity, price)
self.current_position += delta if side == 'BUY' else -delta
return order
风险控制与资金管理
自动化交易缺少人工干预，风险控制必须内嵌到代码中。股票交易要设置单票持仓上限、行业暴露限制和日内亏损阈值。期货交易需动态计算保证金占用，防止强平。每个订单发出前，检查可用资金、持仓限额和涨跌停价格。
资金管理采用固定比例或凯利公式。固定比例更稳健，每笔交易风险不超过总资金的1%~2%。期货杠杆高，应使用ATR（平均真实波幅）计算止损距离，并据此确定手数。
def calculate_position_size(capital, risk_per_trade, entry_price, stop_price, multiplier=1):
risk_per_unit = abs(entry_price - stop_price) * multiplier
if risk_per_unit == 0:
return 0
risk_amount = capital * risk_per_trade
size = risk_amount / risk_per_unit
return int(size)
止损单由程序监控，触发后立即发送平仓指令。股票T+1制度下，当日买入无法卖出，自动化策略需避免日内频繁交易。期货T+0可随时平仓，但需注意平今仓手续费差异。
状态同步与异常处理
API接口返回的订单状态可能延迟或丢失。自动化系统要定时查询持仓和成交回报，与本地记录对账。股票接口通常提供查询当日委托和成交的端点，期货CTP有ReqQryInvestorPosition和ReqQryTrade。发现不一致时，以柜台数据为准，并暂停策略直到修复。
异常处理覆盖网络超时、部分成交、拒绝委托和会话失效。部分成交后，剩余数量可继续挂单或撤单。拒绝委托常见原因包括资金不足、价格超出涨跌停、标的不存在。记录日志并触发告警，人工介入前停止开新仓。
def reconcile_positions(api, local_positions):
remote_positions = api.get_positions()
for symbol, remote_qty in remote_positions.items():
local_qty = local_positions.get(symbol, 0)
if abs(remote_qty - local_qty) > 1e-6:
print(f'持仓不一致 {symbol}: 远程{remote_qty}, 本地{local_qty}')
local_positions[symbol] = remote_qty
return local_positions
事件驱动与低延迟优化
高频或日内策略需要事件驱动架构。行情推送到达后，立即计算信号并下单，避免轮询延迟。股票Level-2行情和期货Tick数据通过WebSocket或CTP的OnRtnDepthMarketData回调获得。
低延迟优化包括使用C++扩展、共享内存和内核旁路。Python层面可用asyncio配合uvloop，将下单逻辑放在独立线程。网络方面，选择离交易所近的云服务器，期货CTP前置机通常有专线接入。
import asyncio
async def on_tick(tick, strategy, executor):
signal = strategy.generate(tick)
if signal:
await executor.execute_signal(signal['symbol'],
signal['target'],
tick['last_price'])
async def main():
tick_queue = asyncio.Queue()
async def market_feed():
while True:
tick = await get_tick_from_websocket()
await tick_queue.put(tick)
async def trading_loop():
while True:
tick = await tick_queue.get()
await on_tick(tick, strategy, executor)
await asyncio.gather(market_feed(), trading_loop())
回测与实盘的一致性
自动化交易的最终考验是回测与实盘表现一致。滑点、手续费和成交延迟在回测中必须模拟。股票回测使用历史逐笔或分钟数据，期货回测考虑保证金和合约乘数。实盘小资金验证后再逐步放大。
每天收盘后，程序自动生成交易报告，对比预期信号与实际成交。偏差超过阈值时，检查API版本、行情源和策略参数。持续监控夏普比率和最大回撤，确保系统在可控范围内运行。
