3. 数据获取：使用 tushare/akshare 获取 A 股日线行情数据、复权因子处理、数据清洗与对齐

数据是量化回测的基石。本章将系统讲解如何从 tushare 和 akshare 两个主流开源库获取 A 股日线行情数据，并完成复权因子处理、数据清洗与多标的对齐操作。掌握这些技能后，你就能为后续的因子计算和分层回测构建干净、可靠的数据集。

3.1 数据源选择与对比

tushare 和 akshare 各有特点，选择哪个取决于你的使用场景和权限等级。下表从关键维度进行了对比：

3.2 使用 tushare 获取日线行情与复权因子

tushare 的日线行情接口 daily() 返回未复权数据，需要配合 adj_factor() 复权因子表进行后处理。以下是标准获取流程：

import tushare as ts
import pandas as pd

# 初始化 API（需在 tushare.pro 注册获取 token）
pro = ts.pro_api('your_token_here')

# 1. 获取股票列表（以沪深 300 成分股为例）
stock_list = pro.index_weight(index_code='000300.SH', 
                              start_date='20200101', 
                              end_date='20231231')
stocks = stock_list['con_code'].tolist()

# 2. 批量获取日线行情（循环请求，注意频率限制）
def fetch_daily_batch(stock_list, start_date, end_date):
    df_list = []
    for code in stock_list:
        df = pro.daily(ts_code=code, start_date=start_date, 
                       end_date=end_date)
        if not df.empty:
            df_list.append(df)
    return pd.concat(df_list, ignore_index=True)

daily_raw = fetch_daily_batch(stocks[:50], '20200101', '20231231')

# 3. 获取复权因子
adj = pro.adj_factor(ts_code='', trade_date='')
# 实际使用中需按股票代码和日期范围筛选
adj_filtered = adj[adj['ts_code'].isin(stocks[:50])]

print(f"日线数据量: {len(daily_raw)} 行")
print(f"复权因子数据量: {len(adj_filtered)} 行")

3.3 复权因子处理：前复权与后复权

复权因子的核心作用是消除分红、送股、配股等事件对股价连续性的影响。tushare 的 adj_factor 表包含以下关键字段：

复权计算的核心公式如下：

• 前复权价格 = 未复权价格 × 当日前复权因子 / 最新交易日的前复权因子
• 后复权价格 = 未复权价格 × 当日后复权因子 / 基准日（上市首日）的后复权因子

以下代码演示如何将未复权数据转换为前复权数据：

def apply_forward_adjustment(daily_df, adj_df):
    """
    对日线数据应用前复权处理
    :param daily_df: 包含 open, high, low, close, vol 的 DataFrame
    :param adj_df: 复权因子 DataFrame
    :return: 前复权后的 DataFrame
    """
    # 合并日线与复权因子
    merged = pd.merge(daily_df, adj_df[['ts_code', 'trade_date', 'adj_factor']],
                      on=['ts_code', 'trade_date'], how='left')
    
    # 获取每只股票最新交易日的复权因子作为基准
    latest_adj = merged.groupby('ts_code')['adj_factor'].transform('last')
    
    # 计算前复权乘数
    merged['adj_ratio'] = merged['adj_factor'] / latest_adj
    
    # 对价格列应用复权
    for col in ['open', 'high', 'low', 'close']:
        merged[col + '_adj'] = merged[col] * merged['adj_ratio']
    
    # 成交量不需要复权，但需调整（送股后成交量会放大）
    merged['vol_adj'] = merged['vol'] / merged['adj_ratio']
    
    return merged

daily_adj = apply_forward_adjustment(daily_raw, adj_filtered)
print(daily_adj[['ts_code', 'trade_date', 'close', 'close_adj']].head())

3.4 使用 akshare 获取复权数据（替代方案）

如果你没有 tushare token，akshare 提供了更简洁的接口，直接返回前复权数据：

import akshare as ak

# 获取单只股票的前复权日线数据
stock_zh_a_hist_df = ak.stock_zh_a_hist(symbol="000001", 
                                         period="daily", 
                                         start_date="20200101", 
                                         end_date="20231231", 
                                         adjust="qfq")  # qfq=前复权, hfq=后复权

print(stock_zh_a_hist_df.head())
# 返回字段：日期, 开盘, 收盘, 最高, 最低, 成交量, 成交额, 振幅, 涨跌幅, 涨跌额, 换手率

akshare 的优点是无需 token、代码简洁，适合快速验证。但需要注意：

• akshare 返回的复权数据是静态的，即基于当前最新交易日计算的复权价格。如果后续有新的分红事件，历史数据不会自动更新。
• 批量获取多只股票时，akshare 没有内置的批量接口，需要自行循环请求，速度较慢。
• 数据源来自东方财富，偶尔会出现字段缺失或类型不一致的情况，需要额外清洗。

3.5 数据清洗与对齐

获取原始数据后，必须进行清洗和对齐，才能用于因子计算。以下是标准清洗流程：

1. 缺失值处理： 检查并处理停牌导致的 NaN 值（通常用前向填充或直接删除）。
2. 异常值检测： 剔除涨跌幅超过 ±20%（A 股正常涨跌幅限制为 ±10%，新股首日除外）的异常记录。
3. 日期对齐： 将所有股票的交易日期对齐到统一的日历，确保每个截面都有相同的时间戳。
4. 索引设置： 将 trade_date 和 ts_code 设置为 MultiIndex，便于后续面板数据处理。

def clean_and_align(daily_adj):
    """
    数据清洗与对齐
    """
    df = daily_adj.copy()
    
    # 1. 删除缺失值过多的股票（缺失率 > 20%）
    missing_ratio = df.groupby('ts_code')['close_adj'].apply(
        lambda x: x.isnull().mean())
    valid_stocks = missing_ratio[missing_ratio < 0.2].index
    df = df[df['ts_code'].isin(valid_stocks)]
    
    # 2. 异常值处理：涨跌幅超过 ±20% 视为异常
    df['pct_chg'] = df.groupby('ts_code')['close_adj'].pct_change()
    df = df[(df['pct_chg'] >= -0.2) & (df['pct_chg'] <= 0.2)]
    
    # 3. 日期对齐：生成完整的交易日历
    trade_cal = df['trade_date'].drop_duplicates().sort_values()
    full_index = pd.MultiIndex.from_product(
        [trade_cal, df['ts_code'].unique()],
        names=['trade_date', 'ts_code'])
    
    # 4. 重索引并前向填充停牌日数据
    df_aligned = df.set_index(['trade_date', 'ts_code']).reindex(full_index)
    df_aligned = df_aligned.groupby('ts_code').fillna(method='ffill')
    
    # 5. 删除仍然缺失的数据（上市前或退市后的日期）
    df_aligned = df_aligned.dropna(subset=['close_adj'])
    
    return df_aligned.reset_index()

daily_clean = clean_and_align(daily_adj)
print(f"清洗后数据量: {len(daily_clean)} 行")
print(f"覆盖股票数: {daily_clean['ts_code'].nunique()}")
print(f"交易日数: {daily_clean['trade_date'].nunique()}")

3.6 数据获取与清洗流程总览

下图展示了从数据获取到清洗对齐的完整流水线，帮助你建立全局视角：

3.7 数据质量检查清单

在进入因子计算之前，建议使用以下清单对清洗后的数据进行最终验证：