Python自动化：构建通达信数据定时抓取与本地化存储系统

1. 为什么需要自动化抓取通达信数据

做量化分析的朋友都知道，数据就是我们的生命线。以前我每天下午收盘后，都要手动打开通达信软件，一个个点击菜单下载收盘数据和财务数据。这种重复劳动不仅浪费时间，还经常因为手抖点错位置导致下载失败。最痛苦的是出差或者休假的时候，还得找人帮忙操作，实在是不方便。

后来我发现用Python实现自动化操作简直太香了。只需要写一次脚本，设置好定时任务，电脑就会在指定时间自动完成所有操作。即使人不在电脑前，数据也能准时下载保存。这样既保证了数据的及时性，又解放了双手，可以把更多时间用在数据分析上。

这个自动化系统特别适合以下几类人：

• 量化交易初学者，需要稳定获取历史数据做回测
• 个人投资者，想建立自己的股票数据库
• 金融相关专业学生，需要大量数据做研究分析
• 任何厌倦了重复点击操作的程序员朋友

2. 系统架构设计

2.1 整体工作流程

我们的自动化系统主要包含四个核心模块：

1. 启动模块：负责自动启动通达信软件
2. 操作模块：模拟鼠标点击完成数据下载
3. 存储模块：将下载的数据整理保存到指定位置
4. 调度模块：通过任务计划实现定时执行

整个系统的运行流程是这样的：

• 每天收盘后（比如下午3:30），系统自动启动
• 先打开通达信软件并登录
• 依次点击菜单下载收盘数据
• 再切换到财务数据页面下载财务数据
• 所有数据下载完成后自动关闭软件
• 最后将数据按日期分类存储到数据库或CSV文件

2.2 关键技术选型

实现这个系统主要用到以下几个Python库：

• pyautogui：用于模拟鼠标移动和点击操作
• subprocess：启动外部程序（通达信）
• pandas：数据处理和存储
• schedule：定时任务调度（可选方案）

对于屏幕分辨率的适配，我们采用字典存储不同分辨率下的点击坐标。目前支持三种常见分辨率：

• 1440×900
• 1920×1080
• 1366×768

3. 详细实现步骤

3.1 环境准备与安装

首先需要安装必要的Python库：

pip install pyautogui pandas schedule

通达信软件可以从官网下载免费版，安装时建议选择默认路径，这样后面代码配置会更简单。如果安装在其他路径，需要修改代码中的软件路径。

3.2 核心代码解析

下面是自动下载收盘数据的关键代码：

import subprocess import pyautogui from time import sleep # 设置操作间隔和安全措施 pyautogui.PAUSE = 1 pyautogui.FAILSAFE = True # 获取当前屏幕分辨率 screen_size = f"{pyautogui.size().width}*{pyautogui.size().height}" print(f"检测到屏幕分辨率: {screen_size}") # 定义不同分辨率下的点击坐标 click_positions = { '1440*900': { '免费确认': (858, 488), '系统菜单': (38, 10), '盘后数据': (90, 260), # 其他坐标... }, '1920*1080': { '免费确认': (1100, 568), '系统菜单': (38, 10), '盘后数据': (90, 242), # 其他坐标... }, '1366*768': { '免费确认': (820, 420), '系统菜单': (38, 10), '盘后数据': (85, 240), # 其他坐标... } } # 启动通达信 try: subprocess.Popen(r'C:\new_tdx\TdxW.exe') # 修改为你的安装路径 except Exception as e: print(f"启动通达信失败: {e}") exit() sleep(5) # 等待软件启动 # 开始自动操作 try: # 点击免费确认 pyautogui.click(click_positions[screen_size]['免费确认']) sleep(1) # 进入系统菜单 pyautogui.click(click_positions[screen_size]['系统菜单']) sleep(0.5) # 选择盘后数据下载 pyautogui.click(click_positions[screen_size]['盘后数据']) sleep(0.5) # 后续操作... except Exception as e: print(f"自动操作出错: {e}")

3.3 财务数据下载实现

财务数据的下载流程与收盘数据类似，但需要注意几个关键点：

1. 财务数据下载通常需要更多时间，等待间隔要适当延长
2. 有些数据包需要额外确认，要增加相应的点击操作
3. 财务数据量较大，建议分批下载

# 下载财务数据 pyautogui.click(click_positions[screen_size]['系统菜单']) sleep(0.5) pyautogui.click(click_positions[screen_size]['专业数据']) sleep(8) # 需要较长时间加载 # 等待下载按钮可用 while True: if not pyautogui.pixelMatchesColor(x, y, (128,128,128)): break sleep(3) pyautogui.click(click_positions[screen_size]['财务数据包']) sleep(0.5)

4. 数据存储与管理

4.1 本地文件存储方案

最简单的存储方式是将数据保存为CSV文件。我们可以按日期创建文件夹，方便后续查询：

import os from datetime import datetime # 创建存储目录 today = datetime.now().strftime("%Y%m%d") data_dir = f"D:/tdx_data/{today}" os.makedirs(data_dir, exist_ok=True) # 假设已经获取到数据df df.to_csv(f"{data_dir}/close_price.csv", index=False) df.to_csv(f"{data_dir}/financial.csv", index=False)

4.2 数据库存储方案

对于更专业的需求，建议使用SQLite或MySQL数据库：

import sqlite3 from sqlalchemy import create_engine # SQLite方案 conn = sqlite3.connect('tdx_data.db') df.to_sql('close_price', conn, if_exists='append', index=False) # MySQL方案 engine = create_engine('mysql+pymysql://user:password@localhost/tdx_data') df.to_sql('financial_data', engine, if_exists='append', index=False)

4.3 数据清洗与整理

下载的原始数据通常需要做一些处理：

1. 统一日期格式
2. 处理缺失值
3. 标准化字段名称
4. 类型转换

def clean_data(df): # 统一日期格式 df['trade_date'] = pd.to_datetime(df['trade_date']) # 处理缺失值 df.fillna(method='ffill', inplace=True) # 标准化列名 df.columns = df.columns.str.lower() # 类型转换 df['volume'] = df['volume'].astype(int) return df

5. 异常处理与日志记录

5.1 常见问题及解决方案

在实际运行中可能会遇到这些问题：

1. 通达信启动失败：检查安装路径是否正确
2. 点击位置偏移：确认屏幕分辨率是否匹配
3. 下载超时：适当增加等待时间
4. 数据不完整：添加校验逻辑

5.2 完善的日志系统

好的日志能帮助我们快速定位问题：

import logging logging.basicConfig( filename='tdx_auto.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) try: # 业务代码... logging.info("开始下载收盘数据") except Exception as e: logging.error(f"下载失败: {e}", exc_info=True)

日志应该记录这些关键信息：

• 操作开始和结束时间
• 下载的数据类型
• 遇到的错误详情
• 系统状态变化

6. 定时任务设置

6.1 Windows任务计划程序

最稳定的定时方案是使用Windows自带的任务计划：

1. 打开"任务计划程序"
2. 创建基本任务
3. 设置触发器为"每天"，时间15:30
4. 操作为"启动程序"，选择Python解释器
5. 参数填写脚本路径

6.2 Python定时方案

如果不想用系统任务，也可以用Python实现：

import schedule import time def job(): print("开始执行下载任务...") # 调用下载函�� # 每天15:30执行 schedule.every().day.at("15:30").do(job) while True: schedule.run_pending() time.sleep(60)

不过这种方案需要脚本一直运行，不如系统任务稳定。

7. 系统优化建议

7.1 性能优化技巧

1. 减少不必要的等待：精确控制sleep时间
2. 并行下载：多线程下载不同数据
3. 增量更新：只下载新数据
4. 错误重试：对失败操作自动重试

7.2 扩展功能思路

这个基础系统可以进一步扩展：

1. 邮件通知：下载完成后发送结果报告
2. 自动备份：定期压缩归档历史数据
3. 数据校验：检查数据完整性
4. 可视化监控：展示数据更新状态

# 邮件通知示例 import smtplib from email.mime.text import MIMEText def send_email(subject, content): msg = MIMEText(content) msg['Subject'] = subject msg['From'] = 'sender@example.com' msg['To'] = 'receiver@example.com' with smtplib.SMTP('smtp.example.com') as server: server.login('user', 'password') server.send_message(msg)

8. 实际使用经验分享

在开发这个系统的过程中，我踩过不少坑。最开始没有考虑屏幕分辨率的差异，导致在笔记本上写的脚本放到台式机上完全点不准位置。后来改用分辨率自适应的方案才解决这个问题。

另一个常见问题是通达信软件更新后界面变化，导致点击坐标失效。我的解决办法是定期检查脚本是否还能正常工作，并在代码中预留足够的容错空间。

数据存储方面，最初我直接覆盖旧文件，后来发现这样无法追溯历史变化。现在改为按日期存储，并保留至少一年的数据备份。对于特别重要的数据，还会上传到云存储做双重备份。