Baostock学习系列2：批量下载股市日交易明细 配套源代码

import baostock as bs import datetime import time from dateutil.relativedelta import relativedelta import os import pandas as pd def check_is_trading_day(input_date): # 输入日期格式 yyyy-mm-dd lg = bs.login() # 登陆系统 if lg.error_code != '0' : print('login respond error_code:'+lg.error_code) # 显示登陆返回信息 print('login respond error_msg:'+lg.error_msg) # 显示登陆返回信息 else: rs = bs.query_trade_dates(start_date=input_date, end_date=input_date) # 调用 baostock 的 query_trade_dates 函数 if rs.error_code != '0': print('query_trade_dates respond error_code:'+rs.error_code) # 返回错误编码 print('query_trade_dates respond error_msg:'+rs.error_msg) # 返回错误信息 else: try: is_trading_day = int(rs.get_row_data()[1]) # 返回结果为str 此处进行强制转换为整型 1 和 0 用于表示是否交易日 except: is_trading_day = 0 bs.login() else: bs.logout() return is_trading_day def time_offset_cal(input_time, offset, time_type): # input_time 类型为： datetime.datetime offset 为 int 整数 if time_type == 'year': re_dt = input_time + relativedelta(years=offset) return re_dt elif time_type == 'week': re_dt = input_time + datetime.timedelta(weeks=offset) return re_dt elif time_type == 'month': re_dt = input_time + relativedelta(months=offset) return re_dt elif time_type == 'day': re_dt = input_time + datetime.timedelta(days=offset) return re_dt elif time_type == 'hour': re_dt = input_time + datetime.timedelta(hours=offset) return re_dt elif time_type == 'minute': re_dt = input_time + datetime.timedelta(minutes=offset) return re_dt elif time_type == 'second': re_dt = input_time + datetime.timedelta(seconds=offset) return re_dt elif time_type == 'microsecond': re_dt = input_time + datetime.timedelta(microseconds=offset) return re_dt elif time_type == 'millisecond': re_dt = input_time + datetime.timedelta(milliseconds=offset) return re_dt else: print("error") def downloadDayData(date, path): starttime = datetime.datetime.now() #设置 开始时间 bs.login() # 获取指定日期的指数、股票数据 stock_rs = bs.query_all_stock(date) stock_df = stock_rs.get_data() data_df = pd.DataFrame() for code in stock_df["code"]: # print("Downloading :" + code) k_rs = bs.query_history_k_data_plus(code, "date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,psTTM,pcfNcfTTM,pbMRQ,isST", date, date) data_df = pd.concat([data_df,k_rs.get_data()]) # data_df = data_df.append(k_rs.get_data()) # append 在最新的python 3.9 不适用了 bs.logout() data_df.to_csv(path, encoding="utf-8", index=False) endtime = datetime.datetime.now() #设置 结束时间 print("运行时间：" + str((endtime - starttime).seconds) + "秒===========>>>" + "已完成"+ date+ "数据下载") if __name__ == '__main__': config_start_day = '2022-04-01' # 'YYYY-MM-DD' config_end_day = '2022-04-04' # 'YYYY-MM-DD' start_day = datetime.datetime.strptime(config_start_day, "%Y-%m-%d") end_day = datetime.datetime.strptime(config_end_day, "%Y-%m-%d") # print(check_is_trading_day(start_day.strftime('%Y-%m-%d'))) while (int( time.mktime(start_day.timetuple())) <= int( time.mktime(end_day.timetuple()))) : if check_is_trading_day(start_day.strftime('%Y-%m-%d')) == 1 & (not os.path.exists(os.path.join("./download", start_day.strftime('%Y-%m-%d')+".csv"))): downloadDayData(start_day.strftime('%Y-%m-%d'), os.path.join("./download", start_day.strftime('%Y-%m-%d')+".csv")) start_day = time_offset_cal(start_day, 1, 'day') print("success==>"+str(start_day)) print(datetime.datetime.now()) else: start_day = time_offset_cal(start_day, 1, 'day') print("error==>"+str(start_day)) print(datetime.datetime.now())