夜雨聆风
置顶精品
一个视频
讲清楚
用法。
历史精品
分享
后记
给大家解决一个在下载专业财务数据的时候，股票数据包下载慢的问题，那就是单独写一个下载股票数据包的代码，设置如下
打开策略管理器
新建文件
勾选新增的策略，打开编辑
复制代码
import os
import requests
import zipfile
from datetime import datetime
# ==================== 配置区域 ====================
# 通达信安装目录，请根据自己的实际安装路径修改
TDX_DIR = r"D:\tdxV2026X64"
# 下载超时时间（秒）
TIMEOUT = 300
# 下载失败重试次数
MAX_RETRY = 3
# 下载完成后是否自动删除临时zip包
AUTO_DELETE_TEMP_ZIP = True
# ==================================================
# 仅保留股票数据包的下载配置
DOWNLOAD_CONFIG = {
"股票数据包": {
"url": "http://data.tdx.com.cn/vipdoc/tdxgp.zip",
"extract_dir": r"vipdoc\cw",
"type": "zip",
},
}
def download_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/',
}
for attempt in range(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
with open(save_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
downloaded += len(chunk)
if total_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")
return True
except Exception as e:
print(f"\n✗ 下载失败（第{attempt+1}次）: {e}")
if attempt < MAX_RETRY - 1:
print(" 正在重试...")
print(f"✗ {desc}下载失败")
return False
def unzip_file(zip_path, extract_dir):
"""解压zip文件"""
print(f"\n正在解压到: {extract_dir}")
if not zipfile.is_zipfile(zip_path):
print("✗ 不是有效的zip文件")
return False
try:
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
file_list = zip_ref.namelist()
print(f" 压缩包包含 {len(file_list)} 个文件/文件夹")
# 显示前10个文件
for f in file_list[:10]:
print(f" - {f}")
if len(file_list) > 10:
print(f" ... 还有 {len(file_list)-10} 个")
os.makedirs(extract_dir, exist_ok=True)
zip_ref.extractall(extract_dir)
print(f"✓ 解压完成！")
return True
except Exception as e:
print(f"✗ 解压失败: {e}")
return False
def download_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)
if not success:
return False
# 解压
success = unzip_file(zip_path, extract_dir)
# 解压成功后自动删除临时zip包
if success and AUTO_DELETE_TEMP_ZIP:
try:
os.remove(zip_path)
print(f"✓ 已自动清理临时压缩包: {zip_path}")
except Exception as e:
print(f"⚠ 清理临时文件失败: {e}")
return success
def main():
"""主函数"""
print("=" * 60)
print(" 通达信股票数据包自动下载工具（带自动清理）")
print(f" 通达信目录: {TDX_DIR}")
print(f" 当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 60)
# 检查目录
if not os.path.exists(TDX_DIR):
print(f"\n✗ 错误：通达信目录不存在: {TDX_DIR}")
return
results = {}
# 仅执行股票数据包下载
for name, config in DOWNLOAD_CONFIG.items():
success = download_zip_pack(name, config, TDX_DIR)
results[name] = success
# 总结
print("\n" + "=" * 60)
print(" 下载完成总结")
print("=" * 60)
for name, success in results.items():
status = "✓ 成功" if success else "✗ 失败"
print(f" {name}: {status}")
print("\n" + "=" * 60)
print(" 全部任务完成！")
print("=" * 60)
if __name__ == "__main__":
main()
到
完成代码编辑后运行
欢迎有空再
