夜雨聆风下载A股历史交易数据
基于现有的 Tushare + MySQL + Python 技术栈，下载历史数据其实已经成功跑通过（中国铝业那38条记录就是例子）。现在要做的就是从“单只股票测试”升级到“全市场批量下载”。下面我分两个数据源分别说明，并给出可直接运行的脚本。
一、用 Tushare 下载全历史数据（推荐主力）
Tushare 作为你的主力数据源，优势是稳定、规范、有复权因子。以下是完整的批量下载方案：
1. 准备工作
确保已安装所需库：
bash
pip install tushare pandas sqlalchemy pymysql tqdm
2. 完整脚本：Tushare 全市场历史数据下载
创建一个新文件 tushare_download_all.py：
python
import tushare as ts
import pandas as pd
from sqlalchemy import create_engine
from urllib.parse import quote_plus
import time
from tqdm import tqdm
import datetime
print("="*60)
print("Tushare 全市场历史数据下载")
print("="*60)
# 1. 配置Tusharets.set_token('你的Tushare Token')# 替换成你的token
pro = ts.pro_api()
# 2. 连接MySQLpassword ='你的MySQL密码'# 替换成你的密码
encoded_password = quote_plus(password)
engine = create_engine(f'mysql+pymysql://su:{encoded_password}@localhost:3306/stock')
# 3. 获取所有A股股票列表print("\n[1/4] 获取股票列表...")
stock_list = pro.stock_basic(exchange='', list_status='L', fields='ts_code,symbol,name,market,list_date')
print(f"✅ 获取到 {len(stock_list)} 只股票")
print(stock_list.head())
# 4. 设置下载时间范围start_date ='20050101'# 从2005年开始（可根据需要调整）
end_date = datetime.datetime.now().strftime('%Y%m%d')# 到今天
print(f"\n[2/4] 下载时间范围: {start_date} 至 {end_date}")
# 5. 批量下载print("\n[3/4] 开始批量下载...")
success_count =0
fail_count =0
for idx, row in tqdm(stock_list.iterrows(), total=len(stock_list), desc="下载进度"):
ts_code = row['ts_code']
try:
# 获取日线数据df = pro.daily(ts_code=ts_code,
start_date=start_date, end_date=end_date)
if df isnotNoneandlen(df)>0:
# 写入MySQL（追加模式）df.to_sql('daily_a', engine, if_exists='append', index=False)
success_count +=1
else:
fail_count +=1
# 礼貌停顿，避免触发频率限制time.sleep(0.3)
# 每100只股票打印一次进度if(idx +1)%100==0:
print(f"\n已处理 {idx+1} 只，成功: {success_count}，失败: {fail_count}")
except Exception as e:
fail_count +=1
print(f"\n❌ {ts_code} 失败: {e}")
time.sleep(1)# 出错后多等一会儿
continueprint(f"\n✅ 下载完成！成功: {success_count} 只，失败: {fail_count} 只")
# 6. 数据去重（防止重复运行导致重复数据）print("\n[4/4] 正在去重...")
with engine.connect()as conn:
# 删除重复数据，保留最早的一条conn.execute("""
DELETE t1 FROM daily_a t1 INNER JOIN daily_a t2 WHERE t1.id > t2.id AND t1.ts_code = t2.ts_code AND t1.trade_date = t2.trade_date""")
conn.execute("ALTER TABLE daily_a ADD UNIQUE INDEX idx_unique (ts_code, trade_date)")
print("✅ 去重完成，唯一约束已添加")
print("\n"+"="*60)
print("🎉 所有数据下载完成！")
3. 注意事项
Tushare 积分限制：免费用户有调用频率限制（约200次/分钟），脚本中的
time.sleep(0.3)就是为了避免触发限制 -4数据量预估：5000只股票 × 20年 × 250交易日 ≈ 2500万条，MySQL完全能处理
断点续传：脚本没有内置断点续传，如果中途中断，需要手动清理重复数据（脚本最后有去重）
二、用 AKshare 下载全历史数据（备用/补充）
AKshare 的优势是完全免费、无需积分，适合作为 Tushare 的补充。但需要注意它的反爬限制。
1. 准备工作
bash
pip install akshare pandas tqdm
2. 完整脚本：AKshare 全市场历史数据下载
创建一个新文件 akshare_download_all.py：
python
import akshare as ak
import pandas as pd
import time
import os
from tqdm import tqdm
import random
from sqlalchemy import create_engine
from urllib.parse import quote_plus
print("=" * 60)
print("AKshare 全市场历史数据下载")
print("=" * 60)
# 1. 连接MySQL
password = '你的MySQL密码'
encoded_password = quote_plus(password)
engine = create_engine(f'mysql+pymysql://su:{encoded_password}@localhost:3306/stock')
# 2. 获取所有A股股票列表
print("\n[1/4] 获取股票列表...")
try:
stock_df = ak.stock_info_a_code_name()
print(f"✅ 获取到 {len(stock_df)} 只股票")
print(stock_df.head())
except Exception as e:
print(f"❌ 获取股票列表失败: {e}")
exit()
# 3. 设置下载参数
start_date = "20050101"
end_date = "20251231"
success_count = 0
fail_count = 0
# 4. 批量下载
print("\n[2/4] 开始批量下载...")
for idx, row in tqdm(stock_df.iterrows(), total=len(stock_df), desc="下载进度"):
symbol = row['code'] # AKshare使用纯数字代码，如 '600519'
name = row['name']
try:
# 随机等待 2-5 秒，避免被封
wait_time = random.uniform(2, 5)
time.sleep(wait_time)
# 获取历史数据
df = ak.stock_zh_a_hist(
symbol=symbol,
period="daily",
start_date=start_date,
end_date=end_date,
adjust="qfq" # 前复权
)
if df is not None and len(df) > 0:
# 添加股票代码列
df['ts_code'] = f"{symbol}.SH" if symbol.startswith('6') else f"{symbol}.SZ"
# 重命名字段以匹配 daily_a 表
df.rename(columns={
'日期': 'trade_date',
'开盘': 'open',
'收盘': 'close',
'最高': 'high',
'最低': 'low',
'成交量': 'vol',
'成交额': 'amount',
'振幅': 'amplitude',
'涨跌幅': 'pct_chg',
'涨跌额': 'change',
'换手率': 'turnover_rate'
}, inplace=True)
# 转换日期格式
df['trade_date'] = pd.to_datetime(df['trade_date']).dt.strftime('%Y%m%d')
# 只保留需要的列
df = df[['ts_code', 'trade_date', 'open', 'high', 'low', 'close',
'pre_close', 'change', 'pct_chg', 'vol', 'amount']]
# 写入MySQL
df.to_sql('daily_a', engine, if_exists='append', index=False)
success_count += 1
else:
fail_count += 1
except Exception as e:
fail_count += 1
print(f"\n❌ {symbol} {name} 失败: {e}")
time.sleep(5) # 出错后多等一会儿
continue
# 每20只股票打印一次进度
if (idx + 1) % 20 == 0:
print(f"\n已处理 {idx+1} 只，成功: {success_count}，失败: {fail_count}")
print("休息30秒避免被封...")
time.sleep(30)
print(f"\n✅ 下载完成！成功: {success_count} 只，失败: {fail_count} 只")
# 5. 数据去重
print("\n[3/4] 正在去重...")
with engine.connect() as conn:
conn.execute("""
DELETE t1 FROM daily_a t1
INNER JOIN daily_a t2
WHERE t1.id > t2.id
AND t1.ts_code = t2.ts_code
AND t1.trade_date = t2.trade_date
""")
print("✅ 去重完成")
print("\n" + "=" * 60)
print("🎉 所有数据下载完成！")
3. AKshare 注意事项
AKshare 的核心挑战是反爬虫机制，需要特别注意：
pip install --upgrade akshare | |
三、断点续传增强版（推荐）
如果担心下载中断，可以先保存为CSV文件，最后统一导入MySQL：
python
import os
import pandas as pd
import akshare as ak
from tqdm import tqdm
import time
# 创建保存目录
os.makedirs('stock_history', exist_ok=True)
# 获取股票列表
stock_df = ak.stock_info_a_code_name()
# 记录已下载的股票
downloaded = set([f.replace('.csv', '') for f in os.listdir('stock_history') if f.endswith('.csv')])
for idx, row in tqdm(stock_df.iterrows(), total=len(stock_df)):
symbol = row['code']
# 跳过已下载的
if symbol in downloaded:
continue
# 下载数据...
df = ak.stock_zh_a_hist(symbol=symbol, ...)
# 保存为CSV
df.to_csv(f'stock_history/{symbol}.csv', index=False)
time.sleep(3) # 礼貌等待
四、总结
| Tushare | ||
| AKshare |
建议顺序：
先用 Tushare 脚本下载（2005年至今）
如果 Tushare 积分不够，用 AKshare 补充
每天盘后运行增量更新脚本（只获取当天数据）
现在你的 daily_a 表已经准备好，可以开始下载全历史数据了。
如果遇到任何问题，可以把反馈截图求助AI，太给力了！
关于检查是否有重复项，创建唯一索引在下载的时候就避免重复项，查找缺漏项，以及每天收盘后下载当日数据等操作，可以求助Kimi或豆包，本篇从略。
