1-股票数据下载-baostock
·
股票数据下载：
程序使用 baostock API 下载中国 A 股市场的股票数据，支持多种时间周期（日、周、月、5/15/30/60 分钟线）和复权方式（前复权、后复权、不复权）。主要功能包括：
- 自动获取股票列表：根据指定日期获取所有 A 股股票代码
- 数据下载与存储：将不同周期的股票数据保存为 CSV 文件
- 错误处理与重试：处理网络异常和 API 调用失败的情况
- 数据预处理：删除停牌数据、转换时间格式等
程序通过模块化设计，将股票列表获取和数据下载分为独立功能，方便维护和扩展。用户可以根据需要调整参数，如下载时间范围、数据字段和复权类型等。
# -*- coding: utf-8 -*-
"""
Created on Wed Jun 4 11:26:27 2025
@author: Administrator
"""
import baostock as bs
import pandas as pd
import datetime
import time
'''
日线指标参数包括：'date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,pbMRQ,psTTM,pcfNcfTTM,isST'
周、月线指标参数包括：'date,code,open,high,low,close,volume,amount,adjustflag,turn,pctChg'
分钟指标参数包括：'date,time,code,open,high,low,close,volume,amount,adjustflag'
adjustflag：复权类型，默认不复权：3；1：后复权；2：前复权。已支持分钟线、日线、周线、月线前后复权。
'''
# 是否删除停盘数据
DROP_SUSPENSION = True
def update_stk_list(date = None):
# 如果没有提供日期，使用最近的交易日
if date is None:
# 获取最近的交易日
today = datetime.date.today()
# 尝试使用今天作为日期
date = today
# 如果今天是周末，往前找最近的工作日
if today.weekday() >= 5: # 5是周六，6是周日
date = today - datetime.timedelta(days=(today.weekday() - 4))
print(f"尝试获取 {date} 的股票列表")
# 获取指定日期的指数、股票数据
max_retries = 3
for attempt in range(max_retries):
try:
stock_rs = bs.query_all_stock(date.strftime('%Y-%m-%d'))
stock_df = stock_rs.get_data()
# 检查API返回状态
if stock_rs.error_code != '0':
print(f"API调用错误: {stock_rs.error_code} - {stock_rs.error_msg}")
if attempt < max_retries - 1:
print(f"尝试重试 ({attempt+1}/{max_retries})...")
time.sleep(2) # 等待2秒后重试
continue
else:
print("达到最大重试次数，退出")
return []
# 检查是否成功获取数据
if stock_df.empty:
print(f"未获取到 {date} 的股票列表数据")
# 如果今天没数据，尝试昨天
if date == today:
print("尝试获取昨天的数据...")
return update_stk_list(today - datetime.timedelta(days=1))
else:
print("请检查日期或网络连接")
return []
# 打印列名以确认实际列名
print(f"获取到的股票列表列名: {list(stock_df.columns)}")
# 保存完整股票列表
stock_df.to_csv('./stk_data/all_list.csv', encoding = 'gbk', index = False)
# 筛选股票代码范围，使用实际列名
try:
# 尝试使用stock_df['code']而不是stock_df.code
stock_df.drop(stock_df[stock_df['code'] < 'sh.600000'].index, inplace = True)
stock_df.drop(stock_df[stock_df['code'] > 'sz.399000'].index, inplace = True)
stock_df = stock_df[['code']] # 确保只保留code列
stock_df.to_csv('./stk_data/stk_list.csv', encoding = 'gbk', index = False)
print(f"成功获取 {len(stock_df)} 支股票")
return stock_df['code'].tolist()
except KeyError as e:
print(f"列名错误: {e}")
print("请检查baostock返回的数据结构和列名")
return []
break # 如果成功，跳出重试循环
except Exception as e:
print(f"获取股票列表时发生异常: {e}")
if attempt < max_retries - 1:
print(f"尝试重试 ({attempt+1}/{max_retries})...")
time.sleep(2) # 等待2秒后重试
else:
print("达到最大重试次数，退出")
return []
return []
def load_stk_list():
try:
df = pd.read_csv('./stk_data/stk_list.csv')
return df['code'].tolist()
except FileNotFoundError:
print("股票列表文件不存在，请先运行update_stk_list函数")
return []
def convert_time(t):
H = t[8:10]
M = t[10:12]
S = t[12:14]
return H + ':' + M + ':' + S
def download_data(stk_list = [], fromdate = '1990-12-19', todate = datetime.date.today(),
datas = 'date,open,high,low,close,volume,amount,turn,pctChg',
frequency = 'd', adjustflag = '2'):
# 确保stk_list不为空
if not stk_list:
print("股票列表为空，无法下载数据")
return
# 创建目录（如果不存在）
import os
os.makedirs(f'./stk_data/{frequency}', exist_ok=True)
# 统计成功和失败的股票数量
success_count = 0
fail_count = 0
for code in stk_list:
print(f"Downloading ({success_count+fail_count+1}/{len(stk_list)}): {code}")
try:
k_rs = bs.query_history_k_data_plus(code, datas, start_date = fromdate, end_date = todate.strftime('%Y-%m-%d'),
frequency = frequency, adjustflag = adjustflag)
datapath = f'./stk_data/{frequency}/{code}.csv'
out_df = k_rs.get_data()
# 检查API返回状态
if k_rs.error_code != '0':
print(f"API调用错误: {k_rs.error_code} - {k_rs.error_msg}")
fail_count += 1
continue
# 检查是否成功获取数据
if out_df.empty:
print(f"未获取到{code}的数据")
fail_count += 1
continue
if DROP_SUSPENSION and 'volume' in list(out_df):
out_df.drop(out_df[out_df.volume == '0'].index, inplace = True)
# 做time转换
if frequency in ['5', '15', '30', '60'] and 'time' in list(out_df):
out_df['time'] = out_df['time'].apply(convert_time)
out_df.to_csv(datapath, encoding = 'gbk', index = False)
success_count += 1
except Exception as e:
print(f"下载{code}数据时出错: {e}")
fail_count += 1
print(f"数据下载完成: 成功 {success_count}/{len(stk_list)}, 失败 {fail_count}/{len(stk_list)}")
if __name__ == '__main__':
# 创建数据目录（如果不存在）
import os
os.makedirs('./stk_data', exist_ok=True)
os.makedirs('./stk_data/d', exist_ok=True)
os.makedirs('./stk_data/w', exist_ok=True)
os.makedirs('./stk_data/m', exist_ok=True)
os.makedirs('./stk_data/5', exist_ok=True)
os.makedirs('./stk_data/15', exist_ok=True)
os.makedirs('./stk_data/30', exist_ok=True)
os.makedirs('./stk_data/60', exist_ok=True)
# 登录
lg = bs.login()
if lg.error_code != '0':
print(f"登录失败: {lg.error_code} - {lg.error_msg}")
exit(1)
print("登录成功")
try:
# 首次运行
stk_list = update_stk_list()
# 非首次运行
#stk_list = load_stk_list()
if stk_list:
# 下载日线
download_data(stk_list)
# 下载周线
download_data(stk_list, frequency = 'w')
# 下载月线
download_data(stk_list, frequency = 'm')
# 下载5分钟线
download_data(stk_list, fromdate = '2020-6-1', frequency = '5', datas = 'date,time,open,high,low,close,volume,amount,adjustflag')
# 下载15分钟线
download_data(stk_list, fromdate = '2020-6-1', frequency = '15', datas = 'date,time,open,high,low,close,volume,amount,adjustflag')
# 下载30分钟线
download_data(stk_list, fromdate = '2020-6-1', frequency = '30', datas = 'date,time,open,high,low,close,volume,amount,adjustflag')
# 下载60分钟线
download_data(stk_list, fromdate = '2020-6-1', frequency = '60', datas = 'date,time,open,high,low,close,volume,amount,adjustflag')
else:
print("没有获取到股票列表，无法继续下载数据")
finally:
# 登出
bs.logout()
print("登出成功")
代码详细注释
# -*- coding: utf-8 -*-
"""
Created on Wed Jun 4 11:26:27 2025
@author: Administrator
"""
import baostock as bs # 导入baostock金融数据API库
import pandas as pd # 导入pandas数据处理库
import datetime # 导入datetime日期处理库
import time # 导入time时间处理库
'''
日线指标参数包括：'date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,pbMRQ,psTTM,pcfNcfTTM,isST'
周、月线指标参数包括：'date,code,open,high,low,close,volume,amount,adjustflag,turn,pctChg'
分钟指标参数包括：'date,time,code,open,high,low,close,volume,amount,adjustflag'
adjustflag：复权类型，默认不复权：3；1：后复权；2：前复权。已支持分钟线、日线、周线、月线前后复权。
'''
# 是否删除停盘数据
DROP_SUSPENSION = True
def update_stk_list(date = None): # 定义更新股票列表的函数，参数date为可选的日期
# 如果没有提供日期，使用最近的交易日
if date is None:
# 获取最近的交易日
today = datetime.date.today() # 获取当前日期
# 尝试使用今天作为日期
date = today
# 如果今天是周末，往前找最近的工作日
if today.weekday() >= 5: # 5是周六，6是周日
date = today - datetime.timedelta(days=(today.weekday() - 4)) # 计算最近的工作日
print(f"尝试获取 {date} 的股票列表") # 打印正在获取的日期
# 获取指定日期的指数、股票数据
max_retries = 3 # 设置最大重试次数
for attempt in range(max_retries): # 循环尝试获取数据
try:
stock_rs = bs.query_all_stock(date.strftime('%Y-%m-%d')) # 调用baostock API获取股票列表
stock_df = stock_rs.get_data() # 获取API返回的数据并转换为DataFrame
# 检查API返回状态
if stock_rs.error_code != '0': # 如果API调用出错
print(f"API调用错误: {stock_rs.error_code} - {stock_rs.error_msg}") # 打印错误信息
if attempt < max_retries - 1: # 如果还没达到最大重试次数
print(f"尝试重试 ({attempt+1}/{max_retries})...") # 打印重试信息
time.sleep(2) # 等待2秒后重试
continue # 继续下一次循环
else:
print("达到最大重试次数，退出") # 打印达到最大重试次数
return [] # 返回空列表
# 检查是否成功获取数据
if stock_df.empty: # 如果获取的数据为空
print(f"未获取到 {date} 的股票列表数据") # 打印未获取到数据
# 如果今天没数据，尝试昨天
if date == today: # 如果尝试的是今天
print("尝试获取昨天的数据...") # 打印尝试获取昨天数据
return update_stk_list(today - datetime.timedelta(days=1)) # 递归调用函数获取昨天数据
else:
print("请检查日期或网络连接") # 打印检查提示
return [] # 返回空列表
# 打印列名以确认实际列名
print(f"获取到的股票列表列名: {list(stock_df.columns)}") # 打印获取到的列名
# 保存完整股票列表
stock_df.to_csv('./stk_data/all_list.csv', encoding = 'gbk', index = False) # 保存完整股票列表到CSV文件
# 筛选股票代码范围，使用实际列名
try:
# 尝试使用stock_df['code']而不是stock_df.code
stock_df.drop(stock_df[stock_df['code'] < 'sh.600000'].index, inplace = True) # 删除上海证券交易所股票代码小于sh.600000的股票
stock_df.drop(stock_df[stock_df['code'] > 'sz.399000'].index, inplace = True) # 删除深圳证券交易所股票代码大于sz.399000的股票
stock_df = stock_df[['code']] # 确保只保留code列
stock_df.to_csv('./stk_data/stk_list.csv', encoding = 'gbk', index = False) # 保存筛选后的股票列表到CSV文件
print(f"成功获取 {len(stock_df)} 支股票") # 打印获取到的股票数量
return stock_df['code'].tolist() # 返回股票代码列表
except KeyError as e: # 捕获列名错误异常
print(f"列名错误: {e}") # 打印列名错误信息
print("请检查baostock返回的数据结构和列名") # 打印检查提示
return [] # 返回空列表
break # 如果成功，跳出重试循环
except Exception as e: # 捕获其他异常
print(f"获取股票列表时发生异常: {e}") # 打印异常信息
if attempt < max_retries - 1: # 如果还没达到最大重试次数
print(f"尝试重试 ({attempt+1}/{max_retries})...") # 打印重试信息
time.sleep(2) # 等待2秒后重试
else:
print("达到最大重试次数，退出") # 打印达到最大重试次数
return [] # 返回空列表
return [] # 返回空列表
def load_stk_list(): # 定义加载股票列表的函数
try:
df = pd.read_csv('./stk_data/stk_list.csv') # 从CSV文件读取股票列表
return df['code'].tolist() # 返回股票代码列表
except FileNotFoundError: # 捕获文件不存在异常
print("股票列表文件不存在，请先运行update_stk_list函数") # 打印提示信息
return [] # 返回空列表
def convert_time(t): # 定义转换时间格式的函数
H = t[8:10] # 提取小时部分
M = t[10:12] # 提取分钟部分
S = t[12:14] # 提取秒部分
return H + ':' + M + ':' + S # 返回格式化后的时间字符串
def download_data(stk_list = [], fromdate = '1990-12-19', todate = datetime.date.today(), # 定义下载数据的函数，参数包括股票列表、起始日期、结束日期
datas = 'date,open,high,low,close,volume,amount,turn,pctChg', # 数据字段
frequency = 'd', adjustflag = '2'): # 数据频率和复权类型
# 确保stk_list不为空
if not stk_list: # 如果股票列表为空
print("股票列表为空，无法下载数据") # 打印提示信息
return # 返回
# 创建目录（如果不存在）
import os # 导入os模块
os.makedirs(f'./stk_data/{frequency}', exist_ok=True) # 创建保存数据的目录
# 统计成功和失败的股票数量
success_count = 0 # 成功计数器
fail_count = 0 # 失败计数器
for code in stk_list: # 遍历股票列表
print(f"Downloading ({success_count+fail_count+1}/{len(stk_list)}): {code}") # 打印当前下载进度和股票代码
try:
k_rs = bs.query_history_k_data_plus(code, datas, start_date = fromdate, end_date = todate.strftime('%Y-%m-%d'), # 调用baostock API下载历史K线数据
frequency = frequency, adjustflag = adjustflag) # 设置数据频率和复权类型
datapath = f'./stk_data/{frequency}/{code}.csv' # 设置保存文件路径
out_df = k_rs.get_data() # 获取API返回的数据并转换为DataFrame
# 检查API返回状态
if k_rs.error_code != '0': # 如果API调用出错
print(f"API调用错误: {k_rs.error_code} - {k_rs.error_msg}") # 打印错误信息
fail_count += 1 # 失败计数器加1
continue # 继续下一次循环
# 检查是否成功获取数据
if out_df.empty: # 如果获取的数据为空
print(f"未获取到{code}的数据") # 打印未获取到数据
fail_count += 1 # 失败计数器加1
continue # 继续下一次循环
if DROP_SUSPENSION and 'volume' in list(out_df): # 如果需要删除停盘数据且数据中包含volume列
out_df.drop(out_df[out_df.volume == '0'].index, inplace = True) # 删除成交量为0的记录
# 做time转换
if frequency in ['5', '15', '30', '60'] and 'time' in list(out_df): # 如果是分钟级数据且包含time列
out_df['time'] = out_df['time'].apply(convert_time) # 应用时间格式转换函数
out_df.to_csv(datapath, encoding = 'gbk', index = False) # 保存数据到CSV文件
success_count += 1 # 成功计数器加1
except Exception as e: # 捕获其他异常
print(f"下载{code}数据时出错: {e}") # 打印异常信息
fail_count += 1 # 失败计数器加1
print(f"数据下载完成: 成功 {success_count}/{len(stk_list)}, 失败 {fail_count}/{len(stk_list)}") # 打印下载结果统计
if __name__ == '__main__': # 程序入口点
# 创建数据目录（如果不存在）
import os # 导入os模块
os.makedirs('./stk_data', exist_ok=True) # 创建主数据目录
os.makedirs('./stk_data/d', exist_ok=True) # 创建日线数据目录
os.makedirs('./stk_data/w', exist_ok=True) # 创建周线数据目录
os.makedirs('./stk_data/m', exist_ok=True) # 创建月线数据目录
os.makedirs('./stk_data/5', exist_ok=True) # 创建5分钟线数据目录
os.makedirs('./stk_data/15', exist_ok=True) # 创建15分钟线数据目录
os.makedirs('./stk_data/30', exist_ok=True) # 创建30分钟线数据目录
os.makedirs('./stk_data/60', exist_ok=True) # 创建60分钟线数据目录
# 登录
lg = bs.login() # 登录baostock API
if lg.error_code != '0': # 如果登录失败
print(f"登录失败: {lg.error_code} - {lg.error_msg}") # 打印错误信息
exit(1) # 退出程序
print("登录成功") # 打印登录成功
try:
# 首次运行
stk_list = update_stk_list() # 获取股票列表
# 非首次运行
#stk_list = load_stk_list() # 从文件加载股票列表
if stk_list: # 如果成功获取股票列表
# 下载日线
download_data(stk_list) # 下载日线数据
# 下载周线
download_data(stk_list, frequency = 'w') # 下载周线数据
# 下载月线
download_data(stk_list, frequency = 'm') # 下载月线数据
# 下载5分钟线
download_data(stk_list, fromdate = '2020-6-1', frequency = '5', datas = 'date,time,open,high,low,close,volume,amount,adjustflag') # 下载5分钟线数据
# 下载15分钟线
download_data(stk_list, fromdate = '2020-6-1', frequency = '15', datas = 'date,time,open,high,low,close,volume,amount,adjustflag') # 下载15分钟线数据
# 下载30分钟线
download_data(stk_list, fromdate = '2020-6-1', frequency = '30', datas = 'date,time,open,high,low,close,volume,amount,adjustflag') # 下载30分钟线数据
# 下载60分钟线
download_data(stk_list, fromdate = '2020-6-1', frequency = '60', datas = 'date,time,open,high,low,close,volume,amount,adjustflag') # 下载60分钟线数据
else:
print("没有获取到股票列表，无法继续下载数据") # 打印未获取到股票列表
finally:
# 登出
bs.logout() # 登出baostock API
print("登出成功") # 打印登出成功
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
4
0- 0
已为社区贡献1条内容
所有评论(0)