# Python 量化数据处理技巧：复权、对齐、缺失值与换手率计算（附实战代码）

> 量化研究中，拿到数据只是第一步。如果不理解复权、不处理缺失值、不会做多股票数据对齐，算出来的指标可能全是错的。本文系统讲解量化数据处理中最常见的几个坑，以及对应的解决方案。

---

## 一、这些问题你一定遇到过

- 策略回测结果特别好，一查发现是没复权，除权日价格跳空导致假信号
- 多只股票做横截面分析，日期对不齐，merge 后一堆 NaN
- 计算换手率但不知道流通股本怎么来
- 停牌股的空数据把整个分析搞乱

这些都是量化数据处理中的典型问题，下面逐一解决。

---

## 二、环境准备

```bash
pip install "tickflow[all]" --upgrade
```

```python
from tickflow import TickFlow

# 免费服务（日 K 线够用）
tf_free = TickFlow.free()

# 完整服务（需要实时行情、财务数据等）
tf = TickFlow(api_key="your-api-key")
```

> TickFlow 文档：[https://docs.tickflow.org](https://docs.tickflow.org)

---

## 三、复权处理

### 为什么必须复权？

上市公司会进行分红派息、送股转增等操作，导致股价在除权日出现跳空。如果不复权，技术指标会在除权日产生错误信号。

举个例子：某股票原价 20 元，10 送 10 后变成 10 元。如果不复权，均线系统会认为股价"暴跌"了 50%。

### TickFlow 支持的复权方式

```python
from tickflow import TickFlow

tf = TickFlow.free()

symbol = "600519.SH"

# 比例前复权（默认，推荐用于量化回测）
df_forward = tf.klines.get(symbol, period="1d", count=2000, adjust="forward", as_dataframe=True)

# 差值前复权（与东方财富/同花顺价格一致）
df_additive = tf.klines.get(symbol, period="1d", count=2000, adjust="forward_additive", as_dataframe=True)

# 不复权（原始成交价格）
df_none = tf.klines.get(symbol, period="1d", count=2000, adjust="none", as_dataframe=True)

# 比例后复权
df_backward = tf.klines.get(symbol, period="1d", count=2000, adjust="backward", as_dataframe=True)

# 差值后复权
df_backward_add = tf.klines.get(symbol, period="1d", count=2000, adjust="backward_additive", as_dataframe=True)
```

### 不同复权方式的对比

```python
import pandas as pd

compare = pd.DataFrame({
    "trade_date": df_none["trade_date"],
    "不复权": df_none["close"],
    "比例前复权": df_forward["close"],
    "差值前复权": df_additive["close"],
    "比例后复权": df_backward["close"],
})

print(compare.tail(5))
print(f"\n不复权最新价: {df_none['close'].iloc[-1]}")
print(f"前复权最新价: {df_forward['close'].iloc[-1]}")
print(f"后复权最新价: {df_backward['close'].iloc[-1]}")
```

### 选哪种复权？

| 场景 | 推荐复权方式 | 原因 |
|------|------------|------|
| 收益率计算、量化回测 | `forward`（比例前复权） | 收益率连续，不失真 |
| 与东方财富/同花顺价格核对 | `forward_additive`（差值前复权） | 价格数值一致 |
| 长期投资收益对比 | `backward`（后复权） | 能直观看到真实涨幅 |
| 原始价格分析 | `none`（不复权） | 看真实成交价格 |

关键原则：**做回测和计算收益率一定要用前复权。**

### 查看除权因子

```python
tf = TickFlow(api_key="your-api-key")

factors = tf.klines.ex_factors(["600519.SH", "000001.SZ"], as_dataframe=True)

# 按标的分组查看
for symbol in ["600519.SH", "000001.SZ"]:
    sym_factors = factors[factors["symbol"] == symbol]
    print(f"\n{symbol} 最近 5 次除权:")
    print(sym_factors[["trade_date", "ex_factor"]].tail(5).to_string(index=False))
```

---

## 四、多股票日期对齐

### 问题：日期不一致

不同股票可能因为停牌、上市时间不同等原因，K 线日期不完全对齐。直接用 merge 会产生大量 NaN。

### 方案一：基于交易日历对齐

```python
import pandas as pd
from tickflow import TickFlow

tf = TickFlow.free()

symbols = ["600519.SH", "000858.SZ", "601318.SH"]
dfs = tf.klines.batch(symbols, period="1d", count=500, adjust="forward", as_dataframe=True)

# 提取所有交易日的并集
all_dates = set()
for df in dfs.values():
    all_dates.update(df["trade_date"].tolist())
all_dates = sorted(all_dates)

# 对齐到统一日期索引
aligned = pd.DataFrame({"trade_date": all_dates})
for symbol, df in dfs.items():
    close = df[["trade_date", "close"]].rename(columns={"close": symbol})
    aligned = aligned.merge(close, on="trade_date", how="left")

print(f"对齐后形状: {aligned.shape}")
print(aligned.tail())
```

### 方案二：只保留共有日期

```python
# 取交集（所有股票都有数据的日期）
common_dates = None
for df in dfs.values():
    dates = set(df["trade_date"].tolist())
    common_dates = dates if common_dates is None else common_dates & dates

common_dates = sorted(common_dates)

aligned_common = pd.DataFrame({"trade_date": common_dates})
for symbol, df in dfs.items():
    close = df[["trade_date", "close"]].rename(columns={"close": symbol})
    aligned_common = aligned_common.merge(close, on="trade_date", how="inner")

print(f"共有交易日: {len(aligned_common)}")
print(aligned_common.tail())
```

### 方案三：前向填充停牌日

停牌期间用最后一个有效价格填充：

```python
# 在方案一的基础上，前向填充
aligned_filled = aligned.copy()
for symbol in symbols:
    aligned_filled[symbol] = aligned_filled[symbol].ffill()

# 检查填充效果
null_before = aligned[symbols].isnull().sum()
null_after = aligned_filled[symbols].isnull().sum()
print("填充前缺失值:")
print(null_before)
print("\n填充后缺失值:")
print(null_after)
```

---

## 五、缺失值处理

### 常见缺失值来源

1. **停牌**：股票停牌期间没有交易数据
2. **上市时间晚**：某只股票的历史数据比其他股票短
3. **退市**：股票退市后没有新数据
4. **数据异常**：极少数情况下数据源缺失

### 检查缺失值

```python
import pandas as pd
from tickflow import TickFlow

tf = TickFlow.free()

symbols = ["600519.SH", "000858.SZ", "601318.SH", "000001.SZ", "600036.SH"]
dfs = tf.klines.batch(symbols, period="1d", count=1000, adjust="forward", as_dataframe=True)

print("各标的数据量:")
for symbol, df in dfs.items():
    start = df["trade_date"].iloc[0]
    end = df["trade_date"].iloc[-1]
    print(f"  {symbol}: {len(df)} 根 K 线 ({start} ~ {end})")
```

### 处理策略

```python
# 将多只股票收盘价合并为一个 DataFrame
close_df = pd.DataFrame()
for symbol, df in dfs.items():
    s = df.set_index("trade_date")["close"]
    close_df[symbol] = s

# 方法 1：前向填充（适合停牌）
close_filled = close_df.ffill()

# 方法 2：删除含有任何缺失值的行（保守但干净）
close_clean = close_df.dropna()
print(f"原始行数: {len(close_df)}, 清洗后: {len(close_clean)}")

# 方法 3：仅删除缺失值过多的行（某天超过半数股票无数据才删）
threshold = len(symbols) // 2
close_filtered = close_df.dropna(thresh=threshold)
print(f"过滤后: {len(close_filtered)}")
```

---

## 六、换手率计算

换手率是衡量股票活跃度的重要指标，但很多数据源不直接提供换手率，需要用成交量和流通股本自己算。

### 公式

```
换手率 = 成交量(手) × 100 / 流通股本(股)
```

A 股成交量单位为"手"（1 手 = 100 股），所以需要乘以 100 换算为股。

### 实现

```python
import pandas as pd
from tickflow import TickFlow

tf = TickFlow(api_key="your-api-key")

symbol = "600000.SH"

# 获取日 K 线
kdf = tf.klines.get(symbol, period="1d", count=2000, as_dataframe=True)

# 获取股本数据
sdf = tf.financials.shares([symbol], as_dataframe=True)

# 转为日期类型
kdf["date"] = pd.to_datetime(kdf["trade_date"])
sdf["date"] = pd.to_datetime(sdf["period_end"])

# 使用 merge_asof 为每根 K 线匹配当时有效的流通股本
merged = pd.merge_asof(
    kdf.sort_values("date"),
    sdf[["date", "float_shares"]].sort_values("date"),
    on="date",
    direction="backward",
)

# 计算换手率
merged["turnover_rate"] = (merged["volume"] * 100 / merged["float_shares"]).round(4)
merged["turnover_pct"] = merged["turnover_rate"] * 100

print(merged[["trade_date", "close", "volume", "float_shares", "turnover_pct"]].tail(10))
```

输出示例：

```text
   trade_date  close  volume  float_shares  turnover_pct
   2026-04-22   9.61  685905  3.330584e+10          0.21
   2026-04-23   9.54  806247  3.330584e+10          0.24
   2026-04-24   9.45  848590  3.330584e+10          0.25
   2026-04-27   9.36  872815  3.330584e+10          0.26
   2026-04-28   9.37  594550  3.330584e+10          0.18
```

### 为什么用 merge_asof？

流通股本不是每天都变。公司只在特定时间点（增发、回购、限售股解禁等）更新股本数据。`merge_asof` 会为每个交易日匹配最近一次公布的流通股本，这是正确的做法。

### 批量计算换手率

```python
symbols = ["600519.SH", "000858.SZ", "601318.SH"]

# 批量获取 K 线
kline_dfs = tf.klines.batch(symbols, period="1d", count=60, as_dataframe=True)

# 批量获取股本
shares_df = tf.financials.shares(symbols, as_dataframe=True)

for symbol in symbols:
    kdf = kline_dfs.get(symbol)
    sdf = shares_df[shares_df["symbol"] == symbol].copy()

    if kdf is None or len(sdf) == 0:
        continue

    kdf["date"] = pd.to_datetime(kdf["trade_date"])
    sdf["date"] = pd.to_datetime(sdf["period_end"])

    merged = pd.merge_asof(
        kdf.sort_values("date"),
        sdf[["date", "float_shares"]].sort_values("date"),
        on="date",
        direction="backward",
    )

    merged["turnover_pct"] = (merged["volume"] * 100 / merged["float_shares"] * 100).round(2)
    avg_turnover = merged["turnover_pct"].mean()
    print(f"{symbol}: 近 60 日平均换手率 {avg_turnover:.2f}%")
```

---

## 七、收益率计算的正确姿势

### 简单收益率 vs 对数收益率

```python
import numpy as np
from tickflow import TickFlow

tf = TickFlow.free()

df = tf.klines.get("600519.SH", period="1d", count=250, adjust="forward", as_dataframe=True)

# 简单收益率（常用）
df["simple_return"] = df["close"].pct_change()

# 对数收益率（适合统计分析，可加性更好）
df["log_return"] = np.log(df["close"] / df["close"].shift(1))

print(df[["trade_date", "close", "simple_return", "log_return"]].tail(10))
```

### 什么时候用哪种？

| 场景 | 推荐 |
|------|------|
| 策略收益计算 | 简单收益率 |
| 统计分析（相关性、因子收益） | 对数收益率 |
| 多期累计收益 | 简单收益率用乘法，对数收益率用加法 |

### 累计收益率

```python
# 简单收益率的累计
df["cum_return"] = (1 + df["simple_return"]).cumprod() - 1

# 或用对数收益率
df["cum_return_log"] = np.exp(df["log_return"].cumsum()) - 1

print(f"期间总收益: {df['cum_return'].iloc[-1] * 100:.2f}%")
```

---

## 八、数据质量检查

拿到数据后，建议做一些基础的数据质量检查：

```python
from tickflow import TickFlow

tf = TickFlow.free()

df = tf.klines.get("600000.SH", period="1d", count=5000, adjust="forward", as_dataframe=True)

# 1. 检查是否有重复日期
dup = df[df["trade_date"].duplicated()]
print(f"重复日期: {len(dup)} 个")

# 2. 检查价格异常（收盘价为 0 或负数）
bad_price = df[df["close"] <= 0]
print(f"异常价格: {len(bad_price)} 个")

# 3. 检查 OHLC 逻辑（high >= low, high >= open, high >= close）
ohlc_err = df[(df["high"] < df["low"]) | (df["high"] < df["open"]) | (df["high"] < df["close"])]
print(f"OHLC 逻辑异常: {len(ohlc_err)} 个")

# 4. 检查日期连续性（是否有异常缺失）
df["date"] = pd.to_datetime(df["trade_date"])
df["gap"] = df["date"].diff().dt.days
big_gaps = df[df["gap"] > 5]  # 超过 5 天的间隔（排除周末和短假）
print(f"超过 5 天的间隔: {len(big_gaps)} 个")
if len(big_gaps) > 0:
    print(big_gaps[["trade_date", "gap"]].head())
```

---

## 九、数据缓存与增量更新

每次运行策略都重新下载全量数据会浪费时间。可以做简单的本地缓存：

```python
import os
import pandas as pd
from tickflow import TickFlow

CACHE_DIR = "./kline_cache"
os.makedirs(CACHE_DIR, exist_ok=True)


def get_klines_cached(tf, symbol, period="1d", count=10000):
    """带本地缓存的 K 线获取"""
    cache_file = os.path.join(CACHE_DIR, f"{symbol}_{period}.csv")

    if os.path.exists(cache_file):
        cached = pd.read_csv(cache_file)
        last_date = cached["trade_date"].iloc[-1]
        print(f"{symbol}: 缓存到 {last_date}，增量更新...")

        # 获取缓存之后的新数据
        import datetime
        last_ts = int(datetime.datetime.strptime(last_date, "%Y-%m-%d").timestamp() * 1000)
        new_data = tf.klines.get(
            symbol, period=period, start_time=last_ts + 86400000,
            count=count, as_dataframe=True,
        )

        if len(new_data) > 0:
            # 过滤掉可能重复的最后一天
            new_data = new_data[new_data["trade_date"] > last_date]
            combined = pd.concat([cached, new_data], ignore_index=True)
            combined.to_csv(cache_file, index=False)
            print(f"  新增 {len(new_data)} 根 K 线")
            return combined
        return cached
    else:
        print(f"{symbol}: 首次获取...")
        df = tf.klines.get(symbol, period=period, count=count, as_dataframe=True)
        df.to_csv(cache_file, index=False)
        return df


tf = TickFlow.free()
df = get_klines_cached(tf, "600519.SH")
print(f"总计 {len(df)} 根 K 线")
```

---

## 十、总结

量化数据处理中最容易踩坑的几个地方：

| 问题 | 正确做法 |
|------|---------|
| 不复权就算指标 | 回测必须用前复权 |
| 多股票日期不对齐 | merge + ffill 或取交集 |
| 缺失值导致计算出错 | 先检查再决定填充或删除 |
| 换手率不知道怎么算 | volume * 100 / float_shares，用 merge_asof |
| 收益率用错方法 | 策略用简单收益率，统计分析用对数收益率 |
| 每次都全量下载 | 做本地缓存 + 增量更新 |

这些技巧看似基础，但每一个处理不当都可能导致回测失真或策略结论错误。

TickFlow 在数据层面已经做了很多：多种复权方式开箱即用、批量接口高效稳定、股本数据可直接获取。在此基础上做好数据处理，你的量化研究才有可靠的地基。

---

## 相关链接

- 官网：[https://tickflow.org](https://tickflow.org?utm_source=tencent_dev)
- 文档：[https://docs.tickflow.org](https://docs.tickflow.org)
- Github：[https://github.com/tickflow-org/tickflow](https://github.com/tickflow-org/tickflow)

---

> 数据处理不性感，但它决定了你的回测结果是真是假。


量化研究中，拿到数据只是第一步。如果不理解复权、不处理缺失值、不会做多股票数据对齐，算出来的指标可能全是错的。本文系统讲解量化数据处理中最常见的几个坑，以及对应的解决方案。

Python 量化数据处理技巧：复权、对齐、缺失值与换手率计算（附实战代码）

编程语言

大数据

后端

本文详细讲解Python量化数据处理中的关键技巧，包括复权处理、多股票日期对齐、缺失值填充和换手率计算。通过实战代码演示如何使用TickFlow库获取前复权数据，避免因除权导致的假信号。文章还介绍了日期对齐的三种方案、缺失值处理策略以及换手率的正确计算方法，帮助量化研究者构建可靠的数据基础，确保回测结果准确无误。

批量计算

数据缓存

Python

2026采购季 | AI焕新·智启新局

batch

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Python 量化数据处理技巧：复权、对齐、缺失值与换手率计算（附实战代码）-腾讯云开发者社区-腾讯云

Python 量化数据处理技巧：复权、对齐、缺失值与换手率计算（附实战代码）

Python 量化数据处理技巧：复权、对齐、缺失值与换手率计算（附实战代码）

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐