
置顶精品
一个视频
讲清楚
用法。
历史精品
分享
通达信为广大用户提供了一个下载专业财务数据的下载窗口

但是,每天手动去下载解压总是略显麻烦,通过和豆包的不断交流,她用python写了一个角本,每天需要时运行一下角本就可以下载下来了,看样子非常方便
关键点在安装目录一定要和你自己的一致




"""
通达信专业财务数据自动下载工具(完整版v2)
支持:财务数据包 + 股票数据包 + 系统自定义数据
"""
importos
importrequests
importzipfile
fromdatetimeimportdatetime
# ==================== 配置区域 ====================
# 通达信安装目录
TDX_DIR=r"E:\new_tdx64"
# 下载超时时间(秒)
TIMEOUT=300
# 下载失败重试次数
MAX_RETRY=3
# 是否下载系统自定义数据
DOWNLOAD_EXTERN=True
# ==================================================
# 下载地址配置
DOWNLOAD_CONFIG= {
"财务数据包": {
"url": "http://data.tdx.com.cn/vipdoc/tdxfin.zip",
"extract_dir": r"vipdoc\cw",
"type": "zip",
},
"股票数据包": {
"url": "http://data.tdx.com.cn/vipdoc/tdxgp.zip",
"extract_dir": r"vipdoc\cw",
"type": "zip",
},
}
defdownload_file(url, save_path, desc="文件"):
"""带进度显示的文件下载"""
os.makedirs(os.path.dirname(save_path), exist_ok=True)
headers= {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Referer': 'https://www.tdx.com.cn/',
}
forattemptinrange(MAX_RETRY):
try:
print(f"\n开始下载 {desc}...")
response=requests.get(url, headers=headers, stream=True, timeout=TIMEOUT)
response.raise_for_status()
total_size=int(response.headers.get('content-length', 0))
print(f"文件大小: {total_size/1024/1024:.1f} MB")
downloaded=0
withopen(save_path, 'wb') asf:
forchunkinresponse.iter_content(chunk_size=8192):
ifchunk:
f.write(chunk)
downloaded+=len(chunk)
iftotal_size>0:
percent=downloaded/total_size*100
print(f"\r 进度: {percent:.1f}% ({downloaded/1024/1024:.1f}MB / {total_size/1024/1024:.1f}MB)", end='')
actual_size=os.path.getsize(save_path)
print(f"\n✓ 下载完成!实际大小: {actual_size/1024/1024:.1f} MB")
returnTrue
exceptExceptionase:
print(f"\n✗ 下载失败(第{attempt+1}次): {e}")
ifattempt<MAX_RETRY-1:
print(" 正在重试...")
print(f"✗ {desc}下载失败")
returnFalse
defunzip_file(zip_path, extract_dir):
"""解压zip文件"""
print(f"\n正在解压到: {extract_dir}")
ifnotzipfile.is_zipfile(zip_path):
print("✗ 不是有效的zip文件")
returnFalse
try:
withzipfile.ZipFile(zip_path, 'r') aszip_ref:
file_list=zip_ref.namelist()
print(f" 压缩包包含 {len(file_list)} 个文件/文件夹")
# 显示前10个文件
forfinfile_list[:10]:
print(f" - {f}")
iflen(file_list) >10:
print(f" ... 还有 {len(file_list)-10} 个")
os.makedirs(extract_dir, exist_ok=True)
zip_ref.extractall(extract_dir)
print(f"✓ 解压完成!")
returnTrue
exceptExceptionase:
print(f"✗ 解压失败: {e}")
returnFalse
defget_extern_file_list():
"""获取系统自定义数据文件列表"""
print("\n正在获取系统自定义数据文件列表...")
list_url="http://data.tdx.com.cn/tdxext/gpextern.txt"
try:
headers= {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
response=requests.get(list_url, headers=headers, timeout=30)
response.raise_for_status()
lines=response.text.strip().split('\n')
files= []
forlineinlines:
parts=line.strip().split(',')
iflen(parts) >=1andparts[0]:
files.append({
'filename': parts[0],
'md5': parts[1] iflen(parts) >1else'',
'size': int(parts[2]) iflen(parts) >2andparts[2].isdigit() else0
})
print(f" 找到 {len(files)} 个外部数据文件")
forfinfiles:
print(f" - {f['filename']} ({f['size']/1024:.1f} KB)")
returnfiles
exceptExceptionase:
print(f" 获取文件列表失败: {e}")
return []
defdownload_extern_data(files, tdx_dir):
"""下载并解压系统自定义数据"""
ifnotfiles:
print("没有可下载的外部数据文件")
returnFalse
# 系统自定义数据应该放在 T0002/signals 目录
signals_dir=os.path.join(tdx_dir, "T0002", "signals")
os.makedirs(signals_dir, exist_ok=True)
# 临时下载目录
temp_dir=os.path.join(tdx_dir, "temp_download", "extern")
os.makedirs(temp_dir, exist_ok=True)
print(f"\n下载系统自定义数据...")
print(f"解压目标目录: {signals_dir}")
base_url="http://data.tdx.com.cn/tdxext/"
success_count=0
fori, file_infoinenumerate(files):
filename=file_info['filename']
file_url=base_url+filename
zip_path=os.path.join(temp_dir, filename)
print(f"\n[{i+1}/{len(files)}] 处理 {filename}...")
# 下载
try:
headers= {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
response=requests.get(file_url, headers=headers, timeout=60)
response.raise_for_status()
withopen(zip_path, 'wb') asf:
f.write(response.content)
print(f" ✓ 下载完成")
# 解压到 signals 目录
ifzipfile.is_zipfile(zip_path):
unzip_file(zip_path, signals_dir)
success_count+=1
else:
# 如果不是zip,直接复制到signals目录
importshutil
dest_path=os.path.join(signals_dir, filename)
shutil.copy2(zip_path, dest_path)
print(f" ✓ 已复制到: {dest_path}")
success_count+=1
exceptExceptionase:
print(f" ✗ 处理失败: {e}")
print(f"\n✓ 系统自定义数据处理完成!成功: {success_count}/{len(files)}")
print(f"\n提示:")
print(f" 数据已解压到 {signals_dir}")
print(f" 打开通达信后应该就能使用系统自定义数据了")
print(f" 可以在 公式->自定义数据管理器 中查看")
returnTrue
defdownload_zip_pack(name, config, tdx_dir):
"""下载一个zip数据包"""
print("\n"+"="*60)
print(f"【{name}】")
print("="*60)
url=config['url']
extract_dir=os.path.join(tdx_dir, config['extract_dir'])
# 临时保存路径
temp_dir=os.path.join(tdx_dir, "temp_download")
zip_filename=os.path.basename(url)
zip_path=os.path.join(temp_dir, zip_filename)
# 下载
success=download_file(url, zip_path, name)
ifnotsuccess:
returnFalse
# 解压
success=unzip_file(zip_path, extract_dir)
returnsuccess
defmain():
"""主函数"""
print("="*60)
print(" 通达信专业数据自动下载工具(完整版v2)")
print(f" 通达信目录: {TDX_DIR}")
print(f" 当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("="*60)
# 检查目录
ifnotos.path.exists(TDX_DIR):
print(f"\n✗ 错误:通达信目录不存在: {TDX_DIR}")
return
results= {}
# 1. 下载财务数据包和股票数据包(zip包)
forname, configinDOWNLOAD_CONFIG.items():
success=download_zip_pack(name, config, TDX_DIR)
results[name] =success
# 2. 下载系统自定义数据
ifDOWNLOAD_EXTERN:
print("\n"+"="*60)
print("【系统自定义数据(外部数据)】")
print("="*60)
print("用于 EXTERNSTR(TYPE=1)、EXTERNVALUE(TYPE=1) 等函数")
extern_files=get_extern_file_list()
ifextern_files:
success=download_extern_data(extern_files, TDX_DIR)
results["系统自定义数据"] =success
else:
results["系统自定义数据"] =False
print(" 未获取到外部数据文件列表")
# 总结
print("\n"+"="*60)
print(" 下载完成总结")
print("="*60)
forname, successinresults.items():
status="✓ 成功"ifsuccesselse"✗ 失败"
print(f" {name}: {status}")
print("\n"+"="*60)
print(" 全部任务完成!")
print("="*60)
if__name__=="__main__":
main()






后记
欢迎有空再
