3. 金融数据获取：使用 Tushare/Baostock 获取 A 股日线行情数据，数据清洗与存储

量化策略的回测质量高度依赖底层数据的准确性与完整性。本章将系统讲解如何通过国内主流的开源金融数据接口——Tushare 与 Baostock，获取 A 股日线行情数据，并完成数据清洗与本地化存储，为后续 LightGBM 特征工程提供干净、标准化的数据底座。

3.1 数据源对比与选型

Tushare 与 Baostock 是目前社区使用最广泛的两个免费 A 股数据源，但在数据覆盖、更新频率、接口限制等方面存在显著差异。下表从量化回测的核心维度进行对比：

3.2 数据获取流程总览

下图展示了从数据源到本地存储的完整数据流水线，包含获取、清洗、存储三个核心阶段：

3.3 Tushare Pro 日线数据获取

3.3.1 初始化与 Token 配置

使用 Tushare 前需在官网注册并获取个人 Token。建议将 Token 存储在环境变量或配置文件中，避免硬编码。

import tushare as ts
import os

# 方式一：从环境变量读取（推荐）
token = os.getenv("TUSHARE_TOKEN")
pro = ts.pro_api(token)

# 方式二：直接赋值（仅用于本地测试）
# pro = ts.pro_api('your_token_here')

3.3.2 获取日线行情与复权因子

Tushare 的 daily 接口返回不复权数据，需结合 adj_factor 接口自行计算复权价格。以下代码获取指定股票在指定时间范围内的日线数据：

def fetch_tushare_daily(pro, ts_code, start_date, end_date):
    """
    获取 Tushare 日线行情，包含复权因子
    返回 DataFrame，列：trade_date, open, high, low, close, vol, amount, adj_factor
    """
    # 获取日线行情
    df_daily = pro.daily(
        ts_code=ts_code,
        start_date=start_date,
        end_date=end_date,
        fields='trade_date,open,high,low,close,vol,amount'
    )
    if df_daily.empty:
        return df_daily

    # 获取复权因子
    df_adj = pro.adj_factor(
        ts_code=ts_code,
        start_date=start_date,
        end_date=end_date
    )

    # 合并数据
    df = df_daily.merge(df_adj[['trade_date', 'adj_factor']], on='trade_date', how='left')
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    df.sort_values('trade_date', inplace=True)
    df.reset_index(drop=True, inplace=True)
    return df

3.4 Baostock 数据获取与交叉验证

Baostock 无需注册，适合作为辅助数据源进行交叉验证。其日线接口直接返回前复权数据，使用方式如下：

import baostock as bs
import pandas as pd

def fetch_baostock_daily(stock_code, start_date, end_date):
    """
    获取 Baostock 日线行情（前复权）
    返回 DataFrame，列：date, open, high, low, close, volume, amount
    """
    # 登录
    lg = bs.login()
    if lg.error_code != '0':
        raise Exception(f"Baostock 登录失败: {lg.error_msg}")

    # 获取日线数据
    rs = bs.query_history_k_data_plus(
        stock_code,
        "date,open,high,low,close,volume,amount",
        start_date=start_date,
        end_date=end_date,
        frequency="d",
        adjustflag="2"  # 前复权
    )

    # 转换为 DataFrame
    rows = []
    while rs.next():
        rows.append(rs.get_row_data())
    df = pd.DataFrame(rows, columns=['trade_date', 'open', 'high', 'low', 'close', 'vol', 'amount'])

    # 数据类型转换
    numeric_cols = ['open', 'high', 'low', 'close', 'vol', 'amount']
    df[numeric_cols] = df[numeric_cols].apply(pd.to_numeric, errors='coerce')
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    df.sort_values('trade_date', inplace=True)
    df.reset_index(drop=True, inplace=True)

    bs.logout()
    return df

3.5 数据清洗核心步骤

原始数据通常包含缺失值、异常值、停牌导致的空值等问题。以下为标准化清洗流程：

1. 缺失值处理： 检查 open、high、low、close、vol 列是否存在 NaN。若某日停牌，价格列通常为 0 或 NaN，建议用前一个交易日数据填充或直接剔除。
2. 异常值过滤： 剔除涨跌幅超过 ±20% 的极端值（A 股正常涨跌幅限制为 ±10%，新股上市首日除外）。
3. 复权价格计算： 使用复权因子将收盘价转换为后复权价格，保证历史价格的可比性。
4. 数据类型统一： 将 trade_date 统一为 datetime64 类型，数值列统一为 float64。
5. 索引对齐： 按 trade_date 升序排列，重置索引。

def clean_daily_data(df):
    """
    标准化清洗日线数据
    """
    # 1. 剔除全空行
    df = df.dropna(subset=['open', 'high', 'low', 'close'], how='all')

    # 2. 填充停牌日（价格用前值填充，成交量填 0）
    df['close'] = df['close'].replace(0, np.nan).ffill()
    df[['open', 'high', 'low']] = df[['open', 'high', 'low']].replace(0, np.nan).ffill()
    df['vol'] = df['vol'].fillna(0)

    # 3. 计算涨跌幅，剔除异常值
    df['pct_chg'] = df['close'].pct_change()
    df = df[(df['pct_chg'] >= -0.20) & (df['pct_chg'] <= 0.20)]

    # 4. 计算后复权价格
    if 'adj_factor' in df.columns:
        base_adj = df['adj_factor'].iloc[-1]
        df['close_adj'] = df['close'] * (base_adj / df['adj_factor'])
        df['open_adj'] = df['open'] * (base_adj / df['adj_factor'])
        df['high_adj'] = df['high'] * (base_adj / df['adj_factor'])
        df['low_adj'] = df['low'] * (base_adj / df['adj_factor'])

    # 5. 排序与索引重置
    df = df.sort_values('trade_date').reset_index(drop=True)
    return df

3.6 数据存储方案

清洗后的数据建议采用 Parquet 格式存储，兼顾压缩率与读取速度。对于中小规模回测，也可使用 SQLite 数据库便于查询。

def save_to_parquet(df, file_path):
    """存储为 Parquet 格式"""
    df.to_parquet(file_path, index=False, compression='snappy')

def save_to_sqlite(df, db_path, table_name):
    """存储到 SQLite 数据库"""
    from sqlalchemy import create_engine
    engine = create_engine(f'sqlite:///{db_path}')
    df.to_sql(table_name, engine, if_exists='replace', index=False)

3.7 数据质量校验清单

完成存储后，建议执行以下校验，确保数据可用性：

• 时间连续性： 检查交易日是否连续，是否存在超过 5 个交易日的跳空（排除法定长假）。
• 价格合理性： 验证 high >= low、close 在 [low, high] 区间内。
• 成交量非负： 确保 vol >= 0，且停牌日成交量为 0。
• 双源交叉验证： 随机抽取 10 个交易日，对比 Tushare 与 Baostock 的收盘价，差异应小于 0.5%。

至此，我们已经完成了从数据源选择、API 调用、数据清洗到本地存储的全流程搭建。下一阶段将基于这些干净数据，构建 LightGBM 所需的特征矩阵与标签。