受欢迎的博客标签

通达信数据 - 使用通达信量化TQ(tdx quant tqserver)(2) - 工作原理

Published
Created: 2026年9月7日星期一 18:56:05 Latest updated:2026年9月7日星期一 20:56:01 views(9)

tqserver 是一个基于 TdxAiData.dll 的Python数据接口封装，用于从通达信数据后台获取行情和证券相关数据。

直接通过调用 TdxAiData.dll 从服务器获取数据

通达信数据 - 使用通达信量化TQ(tdx quant)(2) - 工作原理

Table of Contents

1.无需启动支持 TQ 策略功能的通达信版本

2. 必须登录通达信账户- 登录后才能获取数据权限

3.通达信客户端需保持运行状态- TdxQuant 通过 API 与客户端通信

 

tdx quant tqserver目录结构

D:\Thirdprogram\newtdxtqv772\PYPlugins目录
├───\file
├───\sys
├───\user
│   ├───tqserver.py
│   ├───tqs_test.py
│   ├───
│   └───
│   TdxAiData.dll
│   TdxAiData.ini
└───
D:\Thirdprogram\newtdxtqv772\PYPlugins\TdxAiData.dll
D:\Thirdprogram\newtdxtqv772\PYPlugins\TdxAiData.ini

D:\Thirdprogram\newtdxtqv772\PYPlugins\user\tqserver.py

D:\Thirdprogram\newtdxtqv772\PYPlugins\user\tqs_test.py

采用 “Python 策略脚本 ←→（tqserver.py） ←→ TdxAiData.dll ←→ 通达信服务器” 的分层通信模式，实现高效的进程间数据交互。

 

 

tqserver.py - TDX数据访问类，提供了与通达信TdxAiData.dll交互的接口，封装了TdxAiData.dll中的方法

 

调用流程


Python 脚本运行 → tq.initialize(__file__) → 调用 TPythClient.dll的 InitConnect 函数，TPythClient.dll建立和TdxW.exe的连接。


tq.get_market_data(...) → tqcenter.py 封装参数 → 通过 ctypes 调用 TPythClient.dll 对应函数。


TdxAiData.dll → 通过共享内存/消息发送请求到 通达信服务器。


tdxw.exe 处理请求（查缓存或触发数据刷新）→ 将结果写入缓冲区 → DLL 读取并返回字符串。


tqcenter.py 解析 JSON → 返回 Pandas DataFrame 等给策略代码。


策略结束 → tq.close() 断开连接

 

 

tqserver.py

 

功能：tqserver.py（对TdxAiData.dll.dll接口封装层）

通过ctypes库加载TPythClient.dll；封装与DLL交互的核心逻辑；发起请求（如获取行情数据）并解析TPythClient.dll返回的结果；

D:\Thirdprogram\newtdxtqv772\PYPlugins\user\tqserver.py

'''
    Version: 1.0.2
    2026-09-04
'''


# 加载DLL


# 设置DLL函数的返回类型
 def _setup_functions(self):
        """绑定 TdxAiData.h 中需要使用的函数签名。"""
        dll = self._dll
        dll.start.argtypes = [c_char_p]
        dll.start.restype = c_int
        dll.stop.argtypes = []
        dll.stop.restype = c_int
        dll.FreeResult.argtypes = [c_char_p]
        dll.FreeResult.restype = None

        dll.GetStockInfo_Sub.argtypes = [c_char_p, AsyncReceiveDataCallBack]
        dll.GetStockInfo_Sub.restype = c_int
        dll.GetStockInfo_UnSub.argtypes = []
        dll.GetStockInfo_UnSub.restype = c_int
        dll.GetRealDATsInStr.argtypes = [c_char_p, POINTER(c_char_p)]
        dll.GetRealDATsInStr.restype = c_int
        dll.GetHISDATsInStr.argtypes = [c_char_p, c_char_p, c_char_p, c_char_p, c_char_p, c_int, POINTER(c_char_p)]
        dll.GetHISDATsInStr.restype = c_int
        dll.GetTickDATsInStr.argtypes = [c_char_p, c_char_p, c_int, c_int, POINTER(c_char_p)]
        dll.GetTickDATsInStr.restype = c_int
        dll.GetStockListInStr.argtypes = [c_char_p, c_int, POINTER(c_char_p)]
        dll.GetStockListInStr.restype = c_int
        dll.GetBlockStocksInStr.argtypes = [c_char_p, c_int, POINTER(c_char_p)]
        dll.GetBlockStocksInStr.restype = c_int
        dll.GetGPBlockInStr.argtypes = [c_char_p, POINTER(c_char_p)]
        dll.GetGPBlockInStr.restype = c_int
        dll.GetSTOCKInStr.argtypes = [c_char_p, POINTER(c_char_p)]
        dll.GetSTOCKInStr.restype = c_int
        dll.GetMoreInfoInStr.argtypes = [c_char_p, POINTER(c_char_p)]
        dll.GetMoreInfoInStr.restype = c_int
        dll.GetTradeCalendarInStr.argtypes = [c_char_p, c_char_p, c_char_p, c_int, POINTER(c_char_p)]
        dll.GetTradeCalendarInStr.restype = c_int
        dll.GetProDataInStr.argtypes = [c_char_p, POINTER(c_char_p)]
        dll.GetProDataInStr.restype = c_int
        dll.GetCWDATAInStr.argtypes = [c_char_p, c_char_p, c_char_p, POINTER(c_char_p)]
        dll.GetCWDATAInStr.restype = c_int
        dll.GetTrackZsETFInfoInStr.argtypes = [c_char_p, POINTER(c_char_p)]
        dll.GetTrackZsETFInfoInStr.restype = c_int
        dll.GetCBINFOInStr.argtypes = [c_char_p, POINTER(c_char_p)]
        dll.GetCBINFOInStr.restype = c_int
        dll.GetZDTDataInStr.argtypes = [c_char_p, POINTER(c_char_p)]
        dll.GetZDTDataInStr.restype = c_int
        dll.GetGbInfoInStr.argtypes = [c_char_p, c_int, c_int, c_char_p, c_char_p, c_char_p, POINTER(c_char_p)]
        dll.GetGbInfoInStr.restype = c_int
        dll.GetSpecZSGPInStr.argtypes = [c_char_p, c_int, POINTER(c_char_p)]
        dll.GetSpecZSGPInStr.restype = c_int
        dll.GetMinuteDataInStr.argtypes = [c_char_p, c_char_p, POINTER(c_char_p)]
        dll.GetMinuteDataInStr.restype = c_int
        dll.GetTodayMinuteDataInStr.argtypes = [c_char_p, c_int, POINTER(c_char_p)]
        dll.GetTodayMinuteDataInStr.restype = c_int
        dll.GetEXDAYL2InStr.argtypes = [c_char_p, c_int, POINTER(c_char_p)]
        dll.GetEXDAYL2InStr.restype = c_int
        dll.GetTdxDataStr.argtypes = [c_char_p, POINTER(c_char_p)]
        dll.GetTdxDataStr.restype = c_int

# 提供的函数

    def get_his_dats(self, code, start, end, period, dividend, count):
        return self._call_with_free(
            self._dll.GetHISDATsInStr,
            self._encode(code), self._encode(start), self._encode(end),
            self._encode(period), self._encode(dividend), int(count),
        )

    def get_cw_data(self, code, start, end):
        return self._call_with_free(self._dll.GetCWDATAInStr, self._encode(code), self._encode(start), self._encode(end))

    def get_stock_info(self, code):
        return self._call_with_free(self._dll.GetSTOCKInStr, self._encode(code))

    def get_real_dats(self, code):
        return self._call_with_free(self._dll.GetRealDATsInStr, self._encode(code))

    def get_more_info(self, code):
        return self._call_with_free(self._dll.GetMoreInfoInStr, self._encode(code))

    def get_stock_list(self, market, list_type):
        return self._call_with_free(self._dll.GetStockListInStr, self._encode(market), int(list_type))

    def get_block_from_stock(self, block, list_type):
        return self._call_with_free(self._dll.GetBlockStocksInStr, self._encode(block), int(list_type))

    def get_trade_calendar(self, start, end, market, count):
        return self._call_with_free(self._dll.GetTradeCalendarInStr, self._encode(start), self._encode(end), self._encode(market), int(count))

    def get_pro_data(self, request):
        return self._call_with_free(self._dll.GetProDataInStr, self._encode(request))

    def get_relation(self, code):
        return self._call_with_free(self._dll.GetGPBlockInStr, self._encode(code))

    def get_kzz_info(self, code):
        return self._call_with_free(self._dll.GetCBINFOInStr, self._encode(code))

    def get_trackzs_etf_info(self, code):
        return self._call_with_free(self._dll.GetTrackZsETFInfoInStr, self._encode(code))

    def get_zdt_data(self, codes):
        """通过原生 GetZDTDataInStr 获取涨跌停数据。"""
        return self._call_with_free(self._dll.GetZDTDataInStr, self._encode(codes))

    def get_gb_info(self, code, gb_type, count, date_list, start_date, end_date):
        """通过原生 GetGbInfoInStr 获取股本数据。"""
        return self._call_with_free(
            self._dll.GetGbInfoInStr,
            self._encode(code), int(gb_type), int(count), self._encode(date_list),
            self._encode(start_date), self._encode(end_date),
        )

    def get_zzgz_stocklist(self, index_code, list_type):
        """通过原生 GetSpecZSGPInStr 获取指数成分股。"""
        return self._call_with_free(
            self._dll.GetSpecZSGPInStr,
            self._encode(index_code), int(list_type),
        )

    def get_minute_data(self, codestr, date):
        """通过原生 GetMinuteDataInStr 获取指定日期分时数据。"""
        return self._call_with_free(
            self._dll.GetMinuteDataInStr,
            self._encode(codestr), self._encode(date),
        )

    def get_today_minute_data(self, codestr, minute):
        """通过原生 GetTodayMinuteDataInStr 获取当日分时数据。"""
        return self._call_with_free(
            self._dll.GetTodayMinuteDataInStr,
            self._encode(codestr), int(minute),
        )

    def get_exday_data(self, codestr, count):
        """通过原生 GetEXDAYL2InStr 获取日线统计数据。"""
        return self._call_with_free(
            self._dll.GetEXDAYL2InStr,
            self._encode(codestr), int(count),
        )
    def get_tick_data(self, code, date, startxh, wantnum):
        return self._call_with_free(self._dll.GetTickDATsInStr, self._encode(code), self._encode(date), int(startxh), int(wantnum))

    def subscribe(self, codes: str, callback: Callable[[str, int, int], int]):
        self._ensure_started()

        def wrapped(data, datanum, datatype):
            try:
                text = data.decode("gb2312", errors="ignore") if data else ""
                return int(callback(text, datanum, datatype) or 0)
            except Exception:
                return 0

        self._callback = AsyncReceiveDataCallBack(wrapped)
        self._check_result(self._dll.GetStockInfo_Sub(self._encode(codes), self._callback))

    def unsubscribe(self):
        self._ensure_started()
        self._check_result(self._dll.GetStockInfo_UnSub())
        self._callback = None

 

 

TdxAiData.dll

TdxAiData.dll 是 TdxQuant（TQ）框架的进程间通信桥接库。

进程间通信：通过tdxw.exe共享内存/消息队列：实现DLL与tdxw.exe的高效数据交换。

D:\Thirdprogram\newtdxtqv772\PYPlugins\TdxAiData.dll

功能：

处理 TPythClient.dll 与 tdxw.exe 的进程间通信：发送请求（InitConnect、GetHISDATsInStr、GetStockListInStr 等）、接收结果（通常以字符串/JSON 格式返回）

 tdxrpcx64.dll：RPC 通信辅助库

 

TdxAiData.dll提供的函数

 

DLL通常返回字符串格式的JSON数据，Python再将其解析为字典

data_dict = json.loads(ptr)

 

查看TPythClient.dll的依赖

 

通达信航情服务器和数据服务器

D:\Thirdprogram\newtdxtqv772\PYPlugins\TdxAiData.ini

[TcHqHost]
HostNum=1
PrimaryHost=0
HostName01=aihs
IPAddress01=aihs.tdx.com.cn
Port01=7709

[TcDsHost]
HostNum=1
HostName01=aids
IPAddress01=aids.tdx.com.cn
Port01=7727

[Token]
token=
;token请登录通达信个人版商城的会员中心->积分和Key管理->创建数据服务Key

Note

通达信 TDX AI 数据服务 / 数据服务器的配置文件。

它主要配置两类服务器：

TcHqHost：行情（HQ = 行情）服务器
TcDsHost：数据服务（DS = Data Service）服务器
Token：数据服务 Key，用于身份认证

 

 

 

 

useful links