Tushare 量化实战 06：复权数据处理与技术指标计算

前一篇完成了数据库存储，这篇解决量化交易中的两个核心问题：复权数据处理和技术指标计算。这是新手最容易踩坑的地方。

复权数据的重要性

为什么需要复权？

问题场景：

贵州茅台 2023-07-20：10 派 259.96 元（含税）
除息前收盘价：1780 元
除息后开盘价：1520 元 # 看起来暴跌了 14.6%！

实际涨跌幅：0%（只是分红除息）

不复权的问题：

• K 线图出现巨大跳空
• 技术指标计算失真
• 回测收益率严重偏差
• 策略信号错误

复权类型对比

推荐：回测用前复权，实盘用不复权

复权数据获取

方案 1：直接获取复权数据

from data_client import TushareClient

client = TushareClient()

# 获取前复权数据
df_qfq = client.pro.bar(
ts_code=\’600519.SH\’,
start_date=\’20230101\’,
end_date=\’20231231\’,
adj=\’qfq\’ # 前复权
)

# 获取后复权数据
df_hfq = client.pro.bar(
ts_code=\’600519.SH\’,
start_date=\’20230101\’,
end_date=\’20231231\’,
adj=\’hfq\’ # 后复权
)

# 获取不复权数据
df = client.pro.bar(
ts_code=\’600519.SH\’,
start_date=\’20230101\’,
end_date=\’20231231\’,
adj=\’\’ # 不复权
)

方案 2：使用复权因子手动计算

获取复权因子：

def get_adj_factor(ts_code: str, start_date: str, end_date: str) –> pd.DataFrame:
\”\”\”获取复权因子\”\”\”
df = client.pro.adj_factor(
ts_code=ts_code,
start_date=start_date,
end_date=end_date
)
return df

# 使用示例
adj_df = get_adj_factor(\’600519.SH\’, \’20230101\’, \’20231231\’)
print(adj_df.head())

# 输出：
# ts_code trade_date adj_factor
# 0 600519.SH 20230103 1.2345
# 1 600519.SH 20230104 1.2345
# 2 600519.SH 20230105 1.2340 # 复权因子变化（分红除息）

手动计算复权价格：

def calculate_adjusted_price(df: pd.DataFrame, adj_factor: pd.DataFrame) –> pd.DataFrame:
\”\”\”
手动计算复权价格

公式：
前复权价格 = 原始价格 × 当前复权因子 / 基准复权因子
\”\”\”
# 合并数据
merged = pd.merge(df, adj_factor, on=[\’ts_code\’, \’trade_date\’])

# 获取基准复权因子（最后一个交易日）
base_factor = merged[\’adj_factor\’].iloc[–1]

# 计算前复权价格
merged[\’close_qfq\’] = merged[\’close\’] * merged[\’adj_factor\’] / base_factor
merged[\’open_qfq\’] = merged[\’open\’] * merged[\’adj_factor\’] / base_factor
merged[\’high_qfq\’] = merged[\’high\’] * merged[\’adj_factor\’] / base_factor
merged[\’low_qfq\’] = merged[\’low\’] * merged[\’adj_factor\’] / base_factor

return merged[[\’ts_code\’, \’trade_date\’, \’open_qfq\’, \’high_qfq\’, \’low_qfq\’, \’close_qfq\’]]

# 使用示例
adjusted_df = calculate_adjusted_price(df, adj_df)
print(adjusted_df.head())

方案 3：封装复权数据获取

创建 src/data/adjusted_data.py：

\”\”\”复权数据获取\”\”\”
from data_client import TushareClient
from typing import Optional

class AdjustedDataFetcher:
\”\”\”复权数据获取器\”\”\”

def __init__(self, token: str = None):
self.client = TushareClient(token)

def get_adjusted_daily(
self,
ts_code: str,
start_date: str,
end_date: str,
adj_type: str = \’qfq\’
) –> Optional[pd.DataFrame]:
\”\”\”
获取复权日线数据

Args:
ts_code: 股票代码
start_date: 开始日期
end_date: 结束日期
adj_type: 复权类型（qfq: 前复权，hfq: 后复权，: 不复权）

Returns:
DataFrame: 复权后的日线数据
\”\”\”
try:
# 直接使用 pro.bar 接口
df = self.client.pro.bar(
ts_code=ts_code,
start_date=start_date,
end_date=end_date,
adj=adj_type
)

if df is None or df.empty:
return None

# 数据清洗
df[\’trade_date\’] = pd.to_datetime(df[\’trade_date\’])
df = df.sort_values(\’trade_date\’).reset_index(drop=True)

return df

except Exception as e:
print(f\”获取复权数据失败：{

e}\”)
return None

def compare_adjustment(self, ts_code: str, date: str) –> pd.DataFrame:
\”\”\”
对比不同复权方式的价格差异

Args:
ts_code: 股票代码
date: 对比日期

Returns:
DataFrame: 三种复权方式的价格对比
\”\”\”
# 获取前后复权因子
adj_df = self.client.pro.adj_factor(ts_code=ts_code, trade_date=date)
if adj_df is None or adj_df.empty:
return pd.DataFrame()

adj_factor = adj_df.iloc[0][\’adj_factor\’]

# 获取不复权数据
df_raw = self.client.pro.daily(ts_code=ts_code, trade_date=date)

# 获取前复权数据
df_qfq = self.client.pro.bar(ts_code=ts_code, trade_date=date, adj=\’qfq\’)

# 获取后复权数据
df_hfq = self.client.pro.bar(ts_code=ts_code, trade_date=date, adj=\’hfq\’)

# 对比
comparison = pd.DataFrame({

\’复权类型\’: [\’不复权\’, \’前复权\’, \’后复权\’],
\’收盘价\’: [
df_raw.iloc[0][\’close\’] if df_raw is not None else None,
df_qfq.iloc[0][\’close\’] if df_qfq is not None