如何用Python或R持续监听DDE数据并记录所有变更？

核心思路：从轮询改为事件驱动监听 #

你之前用的tk2dde.request（R）和Python的dde库调用都是主动轮询模式，间隔内的数据更新会被跳过。要捕获所有变更，必须切换为被动监听DDE消息通知，让第三方应用主动推送数据变更，而非主动拉取。

R（tcltk2）实现事件驱动监听 #

tcltk2的DDE支持绑定DDEAdvise回调，数据变更时自动触发记录：

library(tcltk2)

# 初始化DDE客户端
dde_client <- tk2dde.client(service = "目标应用服务名", topic = "目标主题")

# 定义数据变更回调函数，自动接收新数据并写入日志
on_data_change <- function(data) {
  # 可写入文件、数据库或内存队列
  cat(paste(Sys.time(), data, "\n"), file = "dde_data_log.txt", append = TRUE)
}

# 注册监听：指定项变更时触发回调
tk2dde.advise(dde_client, item = "目标数据项", callback = on_data_change)

# 保持会话运行，避免脚本退出
tk2dde.loop()

• 替换service、topic、item为实际参数
• tk2dde.advise会让DDE服务端主动推送所有变更，无遗漏

Python（win32dde）实现事件驱动监听 #

Python的win32dde库支持Advise模式，通过回调捕获所有数据更新：

import win32dde
import time
from datetime import datetime

# 定义回调类处理数据变更
class DDECallback(win32dde.DDECallback):
    def OnAdvise(self, topic, item, data):
        # 记录数据到日志或存储
        with open("dde_data_log.txt", "a", encoding="utf-8") as f:
            f.write(f"{datetime.now()}\t{item}\t{data}\n")

# 建立DDE连接并注册监听
client = win32dde.DDEClient("目标应用服务名", "目标主题")
client.SetCallback(DDECallback())
client.Advise("目标数据项")

# 保持客户端运行
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.Disconnect()

• 先安装依赖：pip install pywin32
• Advise模式会订阅数据项的所有变更，服务端有更新就推送

备选方案：本地DDE中转服务 #

如果第三方应用不支持主动推送，可搭建本地DDE中转服务：

• 用Python或C#编写DDE服务端，以极短间隔（10ms级）轮询第三方应用DDE数据，将所有快照存入队列
• 业务程序从中转服务的队列读取数据，避免直接轮询丢数据
• 中转服务需做批量写入优化，避免IO瓶颈

注意事项 #

• 确认第三方应用的DDE服务支持Advise通知模式，多数专业软件（如Excel、行情软件）均支持
• 高频写入场景建议用内存队列批量落地，降低IO压力

内容的提问来源于stack exchange，提问作者gaut