通达信 + miniQMT 期权期货量化交易 技术方案

1. 背景与目标

1.1 问题域

在期权期货交易中，交易者面临以下核心痛点：

• 通达信提供强大的技术指标和公式系统，但无实盘交易接口，信号无法自动执行
• 期权期货盯盘强度高、决策时效短，人工执行难以保证纪律性和响应速度
• 多品种、多合约的跨市场套利需要程序化协调，手工无法胜任

1.2 方案目标

• 信号自动化 — 通达信技术指标信号实时捕获，零人工干预完成交易执行
• 低延迟 — 全链路端到端延迟控制在 500ms 以内（含网络传输）
• 全品种覆盖 — 股指期权、商品期货、商品期权统一接入
• 工程化风控 — 资金管理、仓位控制、异常熔断三防体系
• 可观测 — 全链路日志、实时监控、操作审计

1.3 适用读者

• 熟悉通达信指标公式、希望将其转化为自动化策略的交易者
• 具备 Python 基础、希望搭建个人量化交易系统的开发者
• 需要为团队搭建期权期货交易基础设施的 IT 人员

2. 系统架构

2.1 分层架构总览

2.2 组件职责

2.3 进程拓扑

3. 数据流设计

3.1 正常交易流程

3.2 延迟预算分解

端到端延迟目标：< 500ms（从通达信信号触发到委托到达柜台）

3.3 数据一致性保障

4. 接口定义

4.1 通达信 → 信号文件契约

通达信条件预警输出格式定义为 JSON Lines（每行一个完整 JSON 对象）：

# 文件路径: D:\signals\signals.jsonl
# 编码: UTF-8 without BOM

{"signal_id": "20260515_143021_001", "ts": 1897657821, "symbol": "IO2506-C-4000", "direction": "buy_open", "price": 125.0, "qty": 1, "reason": "MACD金叉+成交量放大", "indicator": {"macd": 2.15, "vol_ratio": 1.8}}

字段定义

4.2 miniQMT / xtquant API 核心接口

以下为方案中使用的核心 API：

passorder 参数详解

# 方法签名
def passorder(
    order_type: int,        # 委托类型: 23=限价, 24=市价, 25=条件单
    order_mode: int,        # 买卖方向: 0=买入/买开, 1=卖出/卖平, 2=卖开, 3=买平
    account_id: str,        # 资金账号
    code: str,              # 合约代码: "IO2506-C-4000"
    price: float,           # 委托价格
    qty: int,               # 委托数量
    strategy_name: str,     # 策略名称（用于标识）
    order_remark: str       # 备注（将出现在成交记录中）
)

5. 信号输出配置（通达信端）

5.1 通达信条件预警设置

1. 菜单：功能 → 条件预警 → 条件预警设置
2. 勾选需要监控的指标公式和股票/合约
3. 输出设置："预警输出到文件"，路径 D:\signals\
4. 格式选择"自定义格式"，配置如下：

# 通达信预警自定义格式模板
# 导出字段映射（通过通头信输出设置界面配置）
# {DATE} {TIME} {CODE} {NAME} {PRICE} {VOL} {SIGNAL}

# 推荐配置：
"D:\signals\signals.jsonl" ← 输出文件路径

# 输出格式（建议用通达信的DLL输出接口做JSON序列化）：
# 如果通头信不支持直接输出JSON，可用中间转换脚本

5.2 中间转换方案（当通达信不支持直接输出JSON时）

大部分通达信版本不支持直接输出 JSON 格式，此时采用二级方案：

# 通达信输出示例（CSV格式）
IO2506-C-4000,买入开仓,125.0,1,MACD金叉,20260515,143021

6. 监控引擎实现

6.1 核心代码框架

# quant_engine.py — 监控引擎核心
import json
import os
import time
import logging
from pathlib import Path
from threading import Thread, Lock
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from xtquant import xtdata, xttrader

class SignalHandler(FileSystemEventHandler):
    """通达信信号文件监控处理器"""

    def __init__(self, signal_path: str, engine):
        self.signal_path = Path(signal_path)
        self.engine = engine
        self.processed_ids = set()  # 去重集合
        self._lock = Lock()
        self._load_processed_ids()

    def on_modified(self, event):
        if not event.src_path.endswith('.jsonl'):
            return
        # 检测锁文件
        lock_file = self.signal_path.with_suffix('.lock')
        if lock_file.exists():
            # 通达信仍在写入，跳过本轮
            return
        self._process_file(event.src_path)

    def _process_file(self, filepath: str):
        """读取并处理信号文件"""
        with open(filepath, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    signal = json.loads(line)
                    self._handle_signal(signal)
                except json.JSONDecodeError as e:
                    logging.error(f"JSON解析失败: {e}, line: {line}")

    def _handle_signal(self, signal: dict):
        """处理单条信号：去重→风控→执行"""
        signal_id = signal.get('signal_id')
        if not signal_id or signal_id in self.processed_ids:
            return
        with self._lock:
            # 风控前置校验
            risk_result = self.engine.risk_controller.check(signal)
            if not risk_result.passed:
                logging.warning(f"风控拦截: {signal_id}, 原因: {risk_result.reason}")
                return
            # 执行交易
            self.engine.executor.execute(signal)
            self.processed_ids.add(signal_id)

    def _load_processed_ids(self):
        """重启时加载已处理ID，避免重复"""
        record_file = Path('data/processed_ids.parquet')
        if record_file.exists():
            # 加载历史 ID 到布隆过滤器
            pass

6.2 执行器实现

class OrderExecutor:
    """交易执行器 - 封装 xttrader.passorder()"""

    def __init__(self, account_id: str):
        self.account_id = account_id
        self.session = xttrader.TradeSession()
        self.session.connect()

    def execute(self, signal: dict) -> dict:
        """将信号转换为委托并发送"""
        direction = signal['direction']
        order_mode = self._direction_to_mode(direction)

        # 获取实时价格（如果 signal 中 price 为 0，则用市价单）
        price = signal.get('price', 0)
        if price == 0:
            price = self._get_market_price(signal['symbol'])

        # 调用 passorder
        order_id = xttrader.passorder(
            order_type=23,              # 限价单
            order_mode=order_mode,
            account_id=self.account_id,
            code=signal['symbol'],
            price=price,
            qty=signal['qty'],
            strategy_name="TDX_QUANT",
            order_remark=signal.get('signal_id', '')
        )

        logging.info(f"委托已发送: signal_id={signal['signal_id']}, order_id={order_id}")
        return {'order_id': order_id, 'status': 'sent'}

    def _direction_to_mode(self, direction: str) -> int:
        """将通达信方向转换为 xtquant order_mode"""
        mapping = {
            'buy_open': 0,    # 买入开仓
            'sell_close': 1,  # 卖出平仓
            'sell_open': 2,   # 卖出开仓
            'buy_close': 3,   # 买入平仓
        }
        mode = mapping.get(direction)
        if mode is None:
            raise ValueError(f"未知方向: {direction}")
        return mode

    def _get_market_price(self, symbol: str) -> float:
        # 获取最新行情价格
        tick = xtdata.get_full_tick([symbol])
        return tick[symbol]['lastPrice']

7. 策略框架

7.1 在通达信中编写指标公式

以下为示例：一个基于 MACD 金叉 + 成交量确认的股指期权买入开仓信号：

{通达信公式 - MACD金叉+量能确认}
// 适用: 沪深300股指期权 IO 合约

DIFF := EMA(CLOSE,12) - EMA(CLOSE,26);
DEA  := EMA(DIFF,9);
MACD := 2*(DIFF-DEA);

// 金叉条件: DIFF 上穿 DEA
CROSS_DIFF_DEA := CROSS(DIFF, DEA);

// 成交量确认: 当前成交量 > 5日均量的1.5倍
VOL_CONFIRM := VOL > MA(VOL,5) * 1.5;

// 信号输出
BUY_SIGNAL := CROSS_DIFF_DEA AND VOL_CONFIRM;

// 将信号输出到预警（在条件预警设置中绑定此公式）
DRAWTEXT(BUY_SIGNAL, LOW, 'BUY_IO');

7.2 策略层设计（Python 端）

7.3 Python 侧策略编排

class StrategyRouter:
    """策略路由：根据信号来源分派到具体策略"""

    def __init__(self):
        self.strategies = {
            'trend_breakout': TrendBreakoutStrategy(),
            'option_buyer': OptionBuyerStrategy(),
            'basis_arbitrage': BasisArbitrageStrategy(),
        }

    def route(self, signal: dict) -> Optional[OrderSignal]:
        """路由决策：返回 OrderSignal 或 None（不执行）"""
        symbol = signal['symbol']
        strategy_type = self._classify(symbol)

        strategy = self.strategies.get(strategy_type)
        if not strategy:
            logging.warning(f"无匹配策略: {symbol}")
            return None

        return strategy.process(signal)

    def _classify(self, symbol: str) -> str:
        """根据合约代码判断策略类型"""
        if symbol.startswith('IO') or symbol.startswith('HO'):
            return 'option_buyer'
        if symbol.endswith('-C-') or symbol.endswith('-P-'):
            return 'option_buyer'
        if any(f in symbol for f in ['IF', 'IC', 'IH', 'RB', 'MA']):
            return 'trend_breakout'
        return 'trend_breakout'

8. 风控体系

8.1 风控架构

风控分为 三层防线，层层拦截：

8.2 风控参数配置

# risk_config.yaml — 风控参数
risk:
  # 资金类
  max_single_order_cost: 100000           # 单笔最大委托金额 (元)
  max_total_position_pct: 0.7             # 总仓位上限 (总资产的70%)
  max_single_position_pct: 0.3            # 单一品种仓位上限 (30%)

  # 频率类
  min_order_interval_ms: 1000             # 最小下单间隔 (1秒)
  max_orders_per_minute: 30               # 每分钟最大委托数
  max_cancel_per_minute: 10               # 每分钟最大撤单数

  # 熔断类
  daily_loss_limit_pct: -0.05             # 日内亏损熔断 (-5%)
  daily_loss_limit_abs: -50000            # 日内亏损熔断 (-5万)
  circuit_breaker_cooldown_min: 30        # 熔断后冷却时间 (30分钟)

  # 价格类
  price_deviation_pct: 0.03              # 价格偏离现价上限 (3%)
  min_price: 0.001                        # 最低委托价格
  max_price: 999999                       # 最高委托价格

  # 白名单
  whitelist_symbols: [
    "IO2506-C-4000", "IO2506-P-4000",
    "IF2506", "IC2506", "IH2506"
  ]

8.3 风控控制器实现

class RiskController:
    """风控前置校验引擎"""

    def __init__(self, config: dict):
        self.config = config
        self.daily_pnl = 0.0           # 当日盈亏
        self.order_count_minute = 0   # 本分钟委托计数
        self.last_order_ts = 0        # 上次委托时间戳
        self.circuit_broken = False    # 熔断状态
        self.circuit_break_ts = 0     # 熔断触发时间

    def check(self, signal: dict) -> RiskResult:
        """风控前置校验，返回校验结果"""
        now = time.time()

        # 1. 熔断检查
        if self.circuit_broken:
            remaining = self.circuit_break_ts + \
                self.config['circuit_breaker_cooldown_min'] * 60 - now
            if remaining > 0:
                return RiskResult(False, f"熔断中，剩余 {int(remaining)}s")
            self.circuit_broken = False

        # 2. 白名单检查
        if signal['symbol'] not in self.config['whitelist_symbols']:
            return RiskResult(False, "合约不在白名单中")

        # 3. 频率检查
        if now - self.last_order_ts < float(
                self.config['min_order_interval_ms']) / 1000:
            return RiskResult(False, "下单间隔过短")

        # 4. 价格偏离检查
        current_price = self._get_current_price(signal['symbol'])
        deviation = abs(signal['price'] - current_price) / \
            (current_price or 1)
        if deviation > self.config['price_deviation_pct']:
            return RiskResult(False,
                f"价格偏离 {deviation:.2%} 超限")

        # 5. 日内亏损熔断
        if self.daily_pnl < float(self.config['daily_loss_limit_pct']) * \
                self._get_total_asset():
            self.circuit_broken = True
            self.circuit_break_ts = now
            return RiskResult(False, "日内亏损熔断触发")

        return RiskResult(True, "")

9. 部署方案

9.1 硬件要求

9.2 目录结构

D:\quant\
├── tdx\                    # 通达信安装目录
│   ├── TdxW.exe
│   └── ...
├── signals\                # 信号文件交互目录
│   ├── signals.jsonl
│   ├── signals.jsonl.lock  # 写锁文件
│   └── archive\            # 历史信号归档
├── engine\                 # 策略引擎
│   ├── main.py
│   ├── config.yaml
│   ├── risk_controller.py
│   ├── strategy_router.py
│   ├── order_executor.py
│   └── requirements.txt
├── logs\                   # 日志目录
│   ├── engine.log
│   ├── trades.log          # 交易明细审计日志
│   └── errors.log
├── data\                   # 持久化数据
│   ├── processed_ids.parquet
│   └── daily_pnl.json
└── scripts\                # 辅助脚本
    ├── start.bat
    ├── stop.bat
    └── health_check.bat

9.3 启动脚本

@REM start.bat — 一键启动策略引擎
@echo off
set PYTHONPATH=D:\quant\engine
cd /d D:\quant\engine

REM 检查通达信是否运行
tasklist | findstr /i "TdxW.exe" >nul
if %errorlevel% neq 0 (
    echo [WARN] 通达信未运行，请先启动通达信
)

REM 启动策略引擎
python main.py --config config.yaml >> ..\logs\engine.log 2>&1

if %errorlevel% equ 0 (
    echo [OK] 策略引擎已启动
) else (
    echo [ERR] 启动失败，请检查日志
)

10. 性能指标与容量规划

10.1 性能目标

10.2 容量约束

11. 监控与告警

11.1 监控指标

11.2 告警通道

# 告警通知示例 — 企业微信 Webhook
import requests

def send_alert(message: str, level: str = "P0"):
    webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": f"【{level}】{message}"
        }
    }
    requests.post(webhook_url, json=payload, timeout=5)

11.3 交易日检查清单

12. 容灾与恢复

12.1 故障场景应对

12.2 看门狗实现

# watchdog.py — 独立的进程看门狗，由 Windows Task Scheduler 拉起
import psutil
import subprocess
import time

class ProcessWatchdog:
    """监控关键进程，崩溃时自动重启"""

    PROCESSES = {
        'tdx': {'name': 'TdxW.exe', 'path': r'D:\quant\tdx\TdxW.exe'},
        'engine': {'name': 'python.exe', 'path': r'D:\quant\engine\main.py',
                    'args': ['--config', 'config.yaml']},
    }

    def check(self):
        for name, proc in self.PROCESSES.items():
            running = any(p.info['name'] == proc['name']
                         for p in psutil.process_iter(['name']))
            if not running:
                logging.warning(f"进程 {name} 未运行，尝试重启...")
                self._restart(name, proc)

    def _restart(self, name, proc):
        if name == 'engine':
            subprocess.Popen(['python', proc['path']] + proc.get('args', []))
        elif name == 'tdx':
            subprocess.Popen([proc['path']])

13. 实施检查清单

14. 术语表