Python自动化:构建通达信数据定时抓取与本地化存储系统
1. 为什么需要自动化抓取通达信数据
做量化分析的朋友都知道,数据就是我们的生命线。以前我每天下午收盘后,都要手动打开通达信软件,一个个点击菜单下载收盘数据和财务数据。这种重复劳动不仅浪费时间,还经常因为手抖点错位置导致下载失败。最痛苦的是出差或者休假的时候,还得找人帮忙操作,实在是不方便。
后来我发现用Python实现自动化操作简直太香了。只需要写一次脚本,设置好定时任务,电脑就会在指定时间自动完成所有操作。即使人不在电脑前,数据也能准时下载保存。这样既保证了数据的及时性,又解放了双手,可以把更多时间用在数据分析上。
这个自动化系统特别适合以下几类人:
- 量化交易初学者,需要稳定获取历史数据做回测
- 个人投资者,想建立自己的股票数据库
- 金融相关专业学生,需要大量数据做研究分析
- 任何厌倦了重复点击操作的程序员朋友
2. 系统架构设计
2.1 整体工作流程
我们的自动化系统主要包含四个核心模块:
- 启动模块:负责自动启动通达信软件
- 操作模块:模拟鼠标点击完成数据下载
- 存储模块:将下载的数据整理保存到指定位置
- 调度模块:通过任务计划实现定时执行
整个系统的运行流程是这样的:
- 每天收盘后(比如下午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 财务数据下载实现
财务数据的下载流程与收盘数据类似,但需要注意几个关键点:
- 财务数据下载通常需要更多时间,等待间隔要适当延长
- 有些数据包需要额外确认,要增加相应的点击操作
- 财务数据量较大,建议分批下载
# 下载财务数据
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 数据清洗与整理
下载的原始数据通常需要做一些处理:
- 统一日期格式
- 处理缺失值
- 标准化字段名称
- 类型转换
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 常见问题及解决方案
在实际运行中可能会遇到这些问题:
- 通达信启动失败:检查安装路径是否正确
- 点击位置偏移:确认屏幕分辨率是否匹配
- 下载超时:适当增加等待时间
- 数据不完整:添加校验逻辑
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自带的任务计划:
- 打开"任务计划程序"
- 创建基本任务
- 设置触发器为"每天",时间15:30
- 操作为"启动程序",选择Python解释器
- 参数填写脚本路径
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 性能优化技巧
- 减少不必要的等待:精确控制sleep时间
- 并行下载:多线程下载不同数据
- 增量更新:只下载新数据
- 错误重试:对失败操作自动重试
7.2 扩展功能思路
这个基础系统可以进一步扩展:
- 邮件通知:下载完成后发送结果报告
- 自动备份:定期压缩归档历史数据
- 数据校验:检查数据完整性
- 可视化监控:展示数据更新状态
# 邮件通知示例
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. 实际使用经验分享
在开发这个系统的过程中,我踩过不少坑。最开始没有考虑屏幕分辨率的差异,导致在笔记本上写的脚本放到台式机上完全点不准位置。后来改用分辨率自适应的方案才解决这个问题。
另一个常见问题是通达信软件更新后界面变化,导致点击坐标失效。我的解决办法是定期检查脚本是否还能正常工作,并在代码中预留足够的容错空间。
数据存储方面,最初我直接覆盖旧文件,后来发现这样无法追溯历史变化。现在改为按日期存储,并保留至少一年的数据备份。对于特别重要的数据,还会上传到云存储做双重备份。
更多推荐
所有评论(0)