别再花钱买数据了！用Python+Baostock免费获取A股历史K线（附完整代码）

零成本构建A股量化数据库：Python+Baostock实战指南

在量化投资领域，数据获取一直是个人开发者和初创团队面临的首要门槛。传统金融数据接口动辄数万元的年费让许多有志于量化研究的开发者望而却步。本文将介绍如何利用完全免费的Baostock金融数据接口，配合Python生态中的强大工具链，构建一套完整的A股历史数据采集与分析系统。

1. 为什么选择Baostock作为量化数据源

对于个人量化爱好者而言，数据源的稳定性、完整性和易用性是三个核心考量维度。Baostock在这三方面都展现出了令人惊喜的表现：

• 完全免费：无需订阅费用，没有调用次数限制
• 数据全面：覆盖A股全部上市公司自1990年至今的日/周/月K线数据
• 接口友好：提供Python原生支持，返回标准Pandas DataFrame格式
• 更新及时：交易日数据通常在当日收盘后1小时内更新

与主流付费接口相比，Baostock在基础数据层面提供了极具竞争力的替代方案：

提示：对于刚入门的量化开发者，建议先用Baostock构建基础数据层，待策略验证有效后再考虑升级到付费接口获取更丰富的特征数据。

2. 环境配置与基础数据获取

2.1 安装与初始化

开始前，确保已安装Python 3.7+环境。通过pip安装Baostock仅需一条命令：

pip install baostock -U

验证安装成功的简单方法：

import baostock as bs lg = bs.login() print(lg.error_code) # 输出"0"表示成功 bs.logout()

2.2 获取全市场股票列表

构建量化数据库的第一步是获取标的证券清单。Baostock提供了便捷的接口查询全市场股票：

import baostock as bs import pandas as pd # 登录系统 bs.login() # 获取指定日期全市场股票 stock_df = bs.query_all_stock("2023-06-30").get_data() # 筛选沪深主板股票 main_board = stock_df[ (stock_df['code'].str.startswith('sh.60')) | (stock_df['code'].str.startswith('sz.00')) ] print(f"获取到{len(main_board)}只主板股票") bs.logout()

输出示例：

code tradeStatus code_name 0 sh.600000 1 浦发银行 1 sh.600004 1 白云机场 2 sh.600006 1 东风汽车 ... ... ... ... 1688 sz.003816 1 中国广核

3. 历史K线数据获取实战

3.1 日K线数据下载

获取个股历史行情是量化分析的基础。以下代码演示如何下载中国平安(601318)的复权历史数据：

def get_history_k_data(code, start_date, end_date, frequency='d', adjustflag='3'): """获取历史K线数据 Parameters: code: 股票代码，如'sh.601318' frequency: 数据类型，d-日k线，w-周，m-月 adjustflag: 复权类型，1-后复权，2-前复权，3-不复权 """ bs.login() # 定义需要获取的字段 fields = "date,open,high,low,close,volume,amount,turn,pctChg" rs = bs.query_history_k_data_plus( code, fields, start_date=start_date, end_date=end_date, frequency=frequency, adjustflag=adjustflag ) data_list = [] while (rs.error_code == '0') & rs.next(): data_list.append(rs.get_row_data()) df = pd.DataFrame(data_list, columns=rs.fields) # 类型转换 numeric_cols = ['open','high','low','close','volume','amount','turn','pctChg'] df[numeric_cols] = df[numeric_cols].apply(pd.to_numeric) df['date'] = pd.to_datetime(df['date']) bs.logout() return df.set_index('date') # 示例：获取中国平安2022年日K线(前复权) pingan = get_history_k_data('sh.601318', '2022-01-01', '2022-12-31', adjustflag='2') print(pingan.head())

3.2 批量下载全市场数据

对于量化回测，通常需要全市场历史数据。以下代码展示如何高效批量下载：

import os from tqdm import tqdm def batch_download(stock_list, start_date, end_date, save_dir='data'): """批量下载股票历史数据""" if not os.path.exists(save_dir): os.makedirs(save_dir) bs.login() for code in tqdm(stock_list['code']): try: df = get_history_k_data(code, start_date, end_date) df.to_csv(f"{save_dir}/{code.replace('.','_')}.csv") except Exception as e: print(f"下载{code}失败: {str(e)}") bs.logout() # 下载2022年沪深300成分股数据 hs300_codes = ['sh.601318', 'sh.600519', 'sz.000333', ...] # 示例代码 batch_download(hs300_codes, '2018-01-01', '2022-12-31')

注意：实际批量下载时建议添加适当的延时(如time.sleep(0.5))，避免对服务器造成过大压力。

4. 数据清洗与增强

原始获取的数据通常需要经过清洗才能用于量化分析。以下是几个常见处理步骤：

4.1 处理异常值和缺失值

def clean_stock_data(df): """数据清洗流程""" # 去除交易状态异常的记录 if 'tradestatus' in df.columns: df = df[df['tradestatus'] == '1'] # 处理涨跌幅异常值 if 'pctChg' in df.columns: df = df[(df['pctChg'] > -20) & (df['pctChg'] < 20)] # 前向填充缺失值 df.fillna(method='ffill', inplace=True) return df # 应用清洗函数 pingan_clean = clean_stock_data(pingan)

4.2 添加技术指标

使用TA-Lib或Pandas计算常用技术指标：

def add_technical_indicators(df): """添加技术指标""" # 移动平均线 df['MA5'] = df['close'].rolling(5).mean() df['MA20'] = df['close'].rolling(20).mean() # 布林带 std = df['close'].rolling(20).std() df['upper_band'] = df['MA20'] + 2*std df['lower_band'] = df['MA20'] - 2*std # RSI delta = df['close'].diff() gain = delta.where(delta > 0, 0) loss = -delta.where(delta < 0, 0) avg_gain = gain.rolling(14).mean() avg_loss = loss.rolling(14).mean() rs = avg_gain / avg_loss df['RSI'] = 100 - (100 / (1 + rs)) return df pingan_with_indicators = add_technical_indicators(pingan_clean)

5. 数据存储与可视化

5.1 使用SQLite构建本地数据库

对于长期积累的数据，建议使用数据库管理：

import sqlite3 from sqlalchemy import create_engine def save_to_sqlite(df, db_path, table_name): """保存到SQLite数据库""" engine = create_engine(f'sqlite:///{db_path}') df.to_sql(table_name, engine, if_exists='replace', index=True) # 示例 save_to_sqlite(pingan_with_indicators, 'stock_data.db', 'pingan_daily')

5.2 使用Matplotlib可视化

import matplotlib.pyplot as plt import mplfinance as mpf def plot_kline(df, title): """绘制K线图""" df = df.copy() df.index.name = 'Date' mpf.plot(df[-100:], # 显示最近100个交易日 type='candle', style='charles', title=title, ylabel='Price (CNY)', volume=True, mav=(5,20), figratio=(12,6)) # 示例 plot_kline(pingan_with_indicators, '中国平安(601318) 2022年K线')

6. 构建自动化数据管道

要实现数据的定期自动更新，可以结合Python的定时任务：

import schedule import time def daily_update(): """每日数据更新任务""" today = datetime.now().strftime('%Y-%m-%d') print(f"开始执行{today}数据更新...") # 获取最新交易日数据 df = get_history_k_data('sh.601318', today, today) if not df.empty: # 更新数据库 conn = sqlite3.connect('stock_data.db') df.to_sql('pingan_daily', conn, if_exists='append') print("数据更新成功") else: print("今日无交易数据") # 设置每天17:00执行 schedule.every().day.at("17:00").do(daily_update) while True: schedule.run_pending() time.sleep(60)

在实际项目中，我曾用这套方法构建了包含3000+只股票、10年历史的本地量化数据库，总存储空间不到5GB，完全满足了中低频策略的研发需求。最关键的是，整个过程没有产生任何数据采购成本，这对于个人开发者来说意义重大。