一、为什么需要“联动”

在量化交易、盘面复盘、策略告警等场景里，我们往往先用 Python 完成计算，再把结果“丢”到行情软件里人工确认。
主流行情软件（通达信、同花顺、东方财富）均未开放官方 API，于是“如何优雅地把股票代码塞进去”就成了刚需。
本文给出一种零侵入、零依赖、零封号的 Windows 消息联动方案，并开源完整代码，5 分钟可集成到任意 Python 投研框架。

二、技术选型：Windows 消息机制

通达信在 2008 年之后一直支持一条未公开的消息：

复制

RegisterWindowMessage('stock')  
wParam = 市场前缀 + 股票代码  
lParam = 0

只要向 HWND_BROADCAST 广播这条消息，通达信主窗口便会自动跳转到对应分时图。
优点：

• 不注入 DLL，不读取内存，安全合规

• 无需管理员权限

• 兼容 32/64 位所有版本（含最新 7.63 Build 2025.08）

三、架构设计

复制

+----------------+      +------------------+      +---------------+
| Python 策略/脚本 | ---> | TdxIntegration | ---> | Windows 消息 |
+----------------+      +------------------+      +---------------+

核心类 TdxIntegration 对外暴露 4 个 API：

表格

复制

完整代码：
 

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
通达信联动模块
用于将股票代码发送到通达信软件进行查看
"""

import time
import win32api
import win32gui
import win32con
from logger import get_logger


class TdxIntegration:
    """通达信联动类"""
    
    def __init__(self):
        """初始化"""
        self.logger = get_logger()
    
    def send_to_tdx(self, stock_code):
        """
        发送股票代码到通达信
        
        Args:
            stock_code (str): 股票代码，支持带后缀(.SZ/.SH)或不带后缀的格式
            
        Returns:
            bool: 发送是否成功
        """
        try:
            # 处理股票代码格式
            clean_code = self._clean_stock_code(stock_code)
            if not clean_code:
                self.logger.warning(f"无效的股票代码格式: {stock_code}")
                return False
            
            # 根据股票代码前缀确定通达信内部代码
            codestr = self._convert_to_tdx_code(clean_code)
            if not codestr:
                self.logger.warning(f"无法转换股票代码: {clean_code}")
                return False
            
            self.logger.info(f"发送股票代码到通达信: {stock_code} -> {codestr}")
            
            # 注册窗口消息并发送
            uwm_stock = win32api.RegisterWindowMessage('stock')
            win32gui.PostMessage(win32con.HWND_BROADCAST, uwm_stock, int(codestr), 0)
            
            self.logger.info(f"成功发送股票代码 {stock_code} 到通达信")
            return True
            
        except Exception as e:
            self.logger.error(f"发送股票代码到通达信失败: {stock_code}, 错误: {str(e)}")
            return False
    
    def _clean_stock_code(self, stock_code):
        """
        清理股票代码格式
        
        Args:
            stock_code (str): 原始股票代码
            
        Returns:
            str: 清理后的6位数字代码，如果格式无效则返回None
        """
        if not stock_code:
            return None
        
        # 转换为字符串并去除空格
        code_str = str(stock_code).strip()
        
        # 如果包含后缀，去除后缀
        if '.' in code_str:
            code_str = code_str.split('.')[0]
        
        # 检查是否为6位数字
        if len(code_str) == 6 and code_str.isdigit():
            return code_str
        
        return None
    
    def _convert_to_tdx_code(self, clean_code):
        """
        将股票代码转换为通达信内部代码
        
        Args:
            clean_code (str): 6位数字股票代码
            
        Returns:
            str: 通达信内部代码，如果无法识别则返回None
        """
        if not clean_code or len(clean_code) != 6:
            return None
        
        # 根据股票代码前缀确定市场
        if clean_code.startswith('6'):  # 上海主板
            return '7' + clean_code
        elif clean_code.startswith('0') or clean_code.startswith('3'):  # 深圳主板/创业板
            return '6' + clean_code
        elif clean_code.startswith('688'):  # 科创板
            return '7' + clean_code  # 科创板也属于上海市场
        elif clean_code.startswith('4'):  # 新三板
            return '4' + clean_code
        else:
            # 其他代码，尝试默认处理
            self.logger.warning(f"未识别的股票代码前缀: {clean_code}")
            return '6' + clean_code  # 默认按深圳市场处理
    
    def is_tdx_available(self):
        """
        检查通达信是否可用（简单检查，实际可能需要更复杂的逻辑）
        
        Returns:
            bool: 通达信是否可能可用
        """
        try:
            # 尝试注册窗口消息，如果失败说明系统不支持
            win32api.RegisterWindowMessage('stock')
            return True
        except Exception as e:
            self.logger.warning(f"通达信可能不可用: {str(e)}")
            return False
    
    def send_multiple_codes(self, stock_codes, delay=1.0):
        """
        批量发送多个股票代码到通达信
        
        Args:
            stock_codes (list): 股票代码列表
            delay (float): 每次发送间的延时（秒）
            
        Returns:
            dict: 发送结果统计
        """
        results = {'success': 0, 'failed': 0, 'total': len(stock_codes)}
        
        for i, code in enumerate(stock_codes):
            if self.send_to_tdx(code):
                results['success'] += 1
            else:
                results['failed'] += 1
            
            # 如果不是最后一个，添加延时
            if i < len(stock_codes) - 1 and delay > 0:
                time.sleep(delay)
        
        self.logger.info(f"批量发送完成: 成功 {results['success']}, 失败 {results['failed']}, 总计 {results['total']}")
        return results


# 全局实例
_tdx_instance = None

def get_tdx_integration():
    """
    获取通达信联动实例（单例模式）
    
    Returns:
        TdxIntegration: 通达信联动实例
    """
    global _tdx_instance
    if _tdx_instance is None:
        _tdx_instance = TdxIntegration()
    return _tdx_instance


# 便捷函数
def send_stock_to_tdx(stock_code):
    """
    便捷函数：发送股票代码到通达信
    
    Args:
        stock_code (str): 股票代码
        
    Returns:
        bool: 发送是否成功
    """
    return get_tdx_integration().send_to_tdx(stock_code)


if __name__ == '__main__':
    # 测试代码
    tdx = TdxIntegration()
    
    print("测试通达信联动功能...")
    
    # 测试各种股票代码格式
    test_codes = [
        '600000',      # 上证股票
        '000001.SZ',   # 深证股票（带后缀）
        '301585',      # 创业板股票
        '688365',      # 科创板股票
        '430139',      # 新三板股票
        '002277'       # 深证股票
    ]
    
    for code in test_codes:
        print(f"发送股票代码: {code}")
        success = tdx.send_to_tdx(code)
        print(f"结果: {'成功' if success else '失败'}")
        time.sleep(2)  # 延时2秒
        print("-" * 30)