通达信条件预警是非常强大的“半量化”功能，如果让其自动下单，就等同于完整的可视化量化策略。

今天就来仔细讲讲，如何通过量化软件实现通达信条件预警自动下单！

所需软件环境

1. 通达信 - 条件预警信号源

2. QMT（mini版） - 量化交易执行端

3. PyCharm - Python开发环境

以上软件如何获取在之前的文章已有详细介绍，可以爬楼看看！

python量化交易环境搭建初学,含绘图代码

废话少说，开讲！

通达信条件预警设置详解

如果你还不知道通达信有条件预警功能，那么今天的内容将为你打开新世界的大门！跟随图一起操作。

步骤1：开启条件预警

• 打开通达信软件，点击主页面左下角的小铃铛图标

步骤2：条件预警设置

• 点击进入条件预警设置

步骤3：添加监控标的

• 添加你想要监控的品种，可以是：

  • 自选股列表

  • 特定ETF成分股

  • 某个行业的所有股票

这相当于量化策略的第一步：确定监控标的池

步骤4：设置选股策略

• 点击预警公式设置

• 选择你参考的技术指标公式：

  • MACD、KDJ等经典指标

  • 自定义编写的个性化公式

实现真正的个性化策略配置

步骤5：关键设置 - 导出选股路径

• 点击操作 → 将内容导为文档文件

• 选择保存路径（请记住此路径，后续量化软件需要读取）

• 保存为易于记忆的英文名称（示例：TDX123）

步骤6：验证信号输出

• 开始选股后，实时查看文本状态

• 确认文档内容随着监控标的的变化而持续更新

文档随着监控的标的变化不断更新，那么就没问题了。

量化系统对接实战

打开QMT（mini）和PyCharm，将以下代码复制到PyCharm中进行测试。

重要提示：请根据你的实际环境修改代码中的关键参数。（不懂可以问我）

• 
  from xtquant import xttrader, xtconstant, xtdata
  from xtquant.xttype import StockAccount
  from xtquant.xttrader import XtQuantTraderCallback
  import logging
  import time
  import pandas as pd
  import re
  import os
  import hashlib
  
  # 精简日志配置 - 只显示重要信息
  logging.basicConfig(
      level=logging.INFO,
      format='%(asctime)s - %(message)s'
  )
  
  # 参数配置
  account_id = '你的模拟账户'
  mini_qmt_path = r'修改为你的QMT文件路径'
  file_path = r'你保存的导出文件路径'
  interval = 3
  buy_signs = ['选股关键词', '例如：MACD买入点条件选股', '例如连涨数天']
  sell_signs = ['例如：KDJ卖出条件选股', '例如：MACD卖出点条件选股']
  session_id = int(time.time() * 1000)#里面的1000代表购买数量，可自行修改
  
  # 全局状态跟踪
  pending_orders = {}
  held_positions = {}
  xt_trader = None
  account = None
  processed_signals = set()
  
  # 指数代码过滤列表
  INDEX_CODES = {
      '999999', '999998', '880005', '000001', '399001', '399006',
      '000300', '399005', '399007', '000688', '000016', '000905'
  }
  
  
  class OrderCallback(XtQuantTraderCallback):
      def on_stock_order(self, order):
          stock_code = order.stock_code.split('.')[0]
          if order.order_status in [1, 4]:
              pending_orders[stock_code] = time.time()
          elif order.order_status in [2, 3, 5]:
              if stock_code in pending_orders:
                  del pending_orders[stock_code]
  
          # 只记录委托状态变化
          status_desc = self._get_status_desc(order.order_status)
          if status_desc in ["已成交", "废单", "已撤"]:
              logging.info(f"委托{status_desc}: {stock_code}")
  
      def on_stock_trade(self, trade):
          stock_code = trade.stock_code.split('.')[0]
          if trade.trade_type == 0:
              held_positions[stock_code] = held_positions.get(stock_code, 0) + trade.traded_volume
          elif trade.trade_type == 1:
              held_positions[stock_code] = held_positions.get(stock_code, 0) - trade.traded_volume
              if held_positions.get(stock_code, 0) <= 0 and stock_code in held_positions:
                  del held_positions[stock_code]
          if stock_code in pending_orders:
              del pending_orders[stock_code]
  
          # 只记录成交回报
          action = "买入" if trade.trade_type == 0 else "卖出"
          logging.info(f"✅ 成交: {action} {stock_code} {trade.traded_volume}股 @ {trade.traded_price}")
  
      def on_order_error(self, error):
          msg = error.error_msg if hasattr(error, 'error_msg') else str(error)
          logging.info(f"❌ 委托失败: {msg}")
  
      def _get_status_desc(self, status_code):
          status_map = {0: "未报", 1: "已报", 2: "已成交", 3: "已撤", 4: "部成", 5: "废单"}
          return status_map.get(status_code, f"未知状态({status_code})")
  
  
  def init_positions():
      global held_positions, xt_trader, account
      try:
          if not xt_trader or not account:
              return
  
          positions = xt_trader.query_stock_positions(account)
          if positions is None:
              return
  
          for pos in positions:
              code = pos.stock_code.split('.')[0]
              held_positions[code] = pos.volume
  
          logging.info(f"📊 初始化持仓: {len(held_positions)}只股票")
      except Exception as e:
          logging.info(f"❌ 持仓初始化异常: {e}")
  
  
  def init_trader():
      global xt_trader, account
      try:
          local_trader = xttrader.XtQuantTrader(mini_qmt_path, session_id)
          local_account = StockAccount(account_id)
          callback = OrderCallback()
  
          local_trader.register_callback(callback)
          local_trader.start()
  
          if local_trader.connect() != 0:
              logging.info("❌ 交易服务器连接失败")
              return None, None
  
          if local_trader.subscribe(local_account) != 0:
              logging.info("❌ 账户订阅失败")
              return None, None
  
          logging.info("✅ 交易系统就绪")
          xt_trader = local_trader
          account = local_account
          init_positions()
          return xt_trader, account
  
      except Exception as e:
          logging.info(f"❌ 初始化异常: {e}")
          return None, None
  
  
  def format_stock_code(code):
      code = str(code).strip()
      if code.startswith(('6', '9', '688')):
          return f"{code}.SH"
      elif code.startswith(('0', '3', '001')):
          return f"{code}.SZ"
      else:
          return None
  
  
  def is_buy_signal(signal_text):
      return any(buy_sign in signal_text for buy_sign in buy_signs)
  
  
  def is_sell_signal(signal_text):
      return any(sell_sign in signal_text for sell_sign in sell_signs)
  
  
  def parse_signal_line(line):
      try:
          line = line.strip()
          if not line or line.startswith('使用') or line.startswith('【错误行】'):
              return None
  
          # 使用制表符分割字段
          parts = re.split(r'\t+', line)
  
          # 清理每个字段的空格
          parts = [part.strip() for part in parts if part.strip()]
  
          if len(parts) >= 7:
              stock_code = parts[0]  # 股票代码
              stock_name = parts[1]  # 股票名称
              datetime_str = parts[2]  # 日期时间
              price_str = parts[3]  # 价格
              change = parts[4]  # 涨跌幅
              signal = parts[6]  # 信号类型
  
              # 过滤指数代码
              if stock_code in INDEX_CODES:
                  return None
  
              # 确保价格是数字
              try:
                  price = float(price_str)
              except ValueError:
                  return None
  
              stock_info = {
                  'name': stock_name,
                  'code': stock_code,
                  'time': datetime_str,
                  'price': price,
                  'change': change,
                  'signal': signal
              }
  
              return stock_info
          else:
              return None
  
      except Exception as e:
          return None
  
  
  def generate_signal_id(stock_info):
      content = f"{stock_info['code']}_{stock_info['signal']}_{stock_info['time']}"
      return hashlib.md5(content.encode()).hexdigest()
  
  
  def buy_event(stock_info):
      global xt_trader, account
      try:
          if not xt_trader or not account:
              logging.info("❌ 交易未就绪")
              return None
  
          stock_code = stock_info.get('code')
          stock_name = stock_info.get('name', '未知')
          signal_type = stock_info.get('signal', '')
  
          if not stock_code:
              return None
  
  
  
         完整代码找我获取！文末+                 
             

             

 实际运行效果展示

信号生成端（通达信）

下图展示了通达信导出的已满足条件的标的：

信号转换端

下图展示了将导出文档中的股票信号发送给QMT的过程：

交易执行端（QMT）

下图展示了QMT的委托成交明细

好了，整个链路并不复杂，其实当你了解了量化以后，会发现其实是一件很有趣的事情。能够真正的增加你选股的效率。并且在满足条件后自动下单也可以完全把握住转瞬即逝的机会。比手工快了不止一个维度。

如果你在公式里面加上“打板公式”，这就是一个打板的量化策略。关键是全链路自己可控，可修改，就算你不懂量化也可以玩儿。不像之前看到python代码一头包，都不知道从何入手。

还有一个关键点，比如你有通达信的L2数据，是不是可以通过通达信公式直接完成买卖点，而量化软件只是一个自动交易端口。通达信的L2可比量化数据的L2便宜多了。

我们下期再见，或者你们想要知道什么都可以在评论区告诉我哦

免责声明：本公众号部分资料、素材、文字、图片等来源于互联网，所有转载都已经注明来源出处。如果您发现有侵犯您的知识产权以及个人合法权益的作品，请与我们取得联系，我们会及时修改或删除，文章中涉及个股不作为投资建议,仅供参考。