轻松下载所有A股5000多只股票历史数据(附Python完整代码)

今天给大家分享，如何完整地下载所有A股5000多只股票历史行情数据，并将历史行情数据存储至自己本地电脑，方便后续直接测试各种技术指标策略，以及开发属于自己的交易策略，及时给出交易信号，实现量化交易。

1. 选择数据源

我们这里使用了两个接口，分别是东方财富网-沪深京 A 股-实时行情数据，东方财富-沪深京 A 股日频率数据。第一个接口可以获取当前A股所有的股票代码及名称。第二个接口可以获取所有A 股上市公司、指定周期和指定日期间的历史行情日频率数据。

import pandas as pdimport akshare as ak
# 东财实时行情数据接口获取所有股票代码接口df = ak.stock_zh_a_spot_em()  # 沪深京 A 股print(df.shape)

2. 设置参数

我们这里设定的是从2023-01-01直至今天的所有A股5426只股票后复权的日频率数据。

3. 编写Python代码

我们通过HTTP请求访问东方财富网的历史行情数据接口，获取数据并保存到本地文件中。借助Python中的requests库可以用来发送HTTP请求，而pandas库则可以方便地处理和保存数据。

import osimport timeimport randomimport pandas as pdimport akshare as akfrom multiprocessing import Poolfrom datetime import datetimeimport warningswarnings.filterwarnings("ignore")pd.set_option('display.max_rows',None)     #显示所有行pd.set_option('display.max_columns',None)  #显示所有列pd.set_option('display.unicode.ambiguous_as_wide', True)  #设置列名对齐pd.set_option('display.unicode.east_asian_width', True)   #设置列名对齐pd.set_option('display.float_format',  '{:,.2f}'.format)  #设置浮点精度# 创建文件存储路径def create_path(ak_code):    global path    # date_str = str(pd.to_datetime(start_date).date()) # 日期转换为字符串    path = os.path.join(".", "all_stock_candle", "stock")    # 保存数据    if not os.path.exists(path):        os.makedirs(path)  # 创建多级文件夹    file_name = ak_code + '.csv'    return os.path.join(path, file_name)# 获取所有股票的历史数据def do_load(ak_code, ak_name, period, start_date, end_date, adj, timeout):    print(ak_code, ak_name)    for i in range(3):        try:            # 历史行情数据-后复权            df = ak.stock_zh_a_hist(symbol=ak_code, period=period, start_date=start_date, end_date=end_date, adjust=adj,                                    timeout=timeout)            if df.empty:                continue            if ak_code.startswith('6') == True:                df['股票代码'] = ak_code + '.SH'            elif ak_code.startswith('8') == True:                df['股票代码'] = ak_code + '.BJ'            else:                df['股票代码'] = ak_code + '.SZ'            df['股票名称'] = ak_name            df.sort_values(by=['日期'], ascending=True, inplace=True)            df.reset_index(drop=True, inplace=True)            path = create_path(ak_code)            df.to_csv(path, index=False, mode='w', encoding='gbk')            # 休眠指定的时间            time.sleep(random.uniform(300, 1000) / 100)            break  # 成功运行后跳出循环        except Exception as e:            # 处理其他未特定捕获的异常            print("发生了其他异常！", e)