詳解Python中量化數(shù)據(jù)的處理技巧(附實戰(zhàn)代碼)
更新時間：2026年05月21日 08:26:07 作者：用戶941614693365
量化研究中,拿到數(shù)據(jù)只是第一步,本文系統(tǒng)講解量化數(shù)據(jù)處理(復(fù)權(quán)、對齊、缺失值與換手率計算)中最常見的幾個坑,以及對應(yīng)的解決方案,有需要的小伙伴可以
量化研究中，拿到數(shù)據(jù)只是第一步。如果不理解復(fù)權(quán)、不處理缺失值、不會做多股票數(shù)據(jù)對齊，算出來的指標(biāo)可能全是錯的。本文系統(tǒng)講解量化數(shù)據(jù)處理中最常見的幾個坑，以及對應(yīng)的解決方案。
一、這些問題你一定遇到過
- 策略回測結(jié)果特別好，一查發(fā)現(xiàn)是沒復(fù)權(quán)，除權(quán)日價格跳空導(dǎo)致假信號
- 多只股票做橫截面分析，日期對不齊，merge 后一堆 NaN
- 計算換手率但不知道流通股本怎么來
- 停牌股的空數(shù)據(jù)把整個分析搞亂
這些都是量化數(shù)據(jù)處理中的典型問題，下面逐一解決。
二、環(huán)境準(zhǔn)備
pip install "tickflow[all]" --upgrade
from tickflow import TickFlow # 免費服務(wù)（日 K 線夠用） tf_free = TickFlow.free() # 完整服務(wù)（需要實時行情、財務(wù)數(shù)據(jù)等） tf = TickFlow(api_key="your-api-key")
TickFlow 文檔：docs.tickflow.org
三、復(fù)權(quán)處理
為什么必須復(fù)權(quán)
上市公司會進(jìn)行分紅派息、送股轉(zhuǎn)增等操作，導(dǎo)致股價在除權(quán)日出現(xiàn)跳空。如果不復(fù)權(quán)，技術(shù)指標(biāo)會在除權(quán)日產(chǎn)生錯誤信號。
舉個例子：某股票原價 20 元，10 送 10 后變成 10 元。如果不復(fù)權(quán)，均線系統(tǒng)會認(rèn)為股價"暴跌"了 50%。
TickFlow 支持的復(fù)權(quán)方式
from tickflow import TickFlow tf = TickFlow.free() symbol = "600519.SH" # 比例前復(fù)權(quán)（默認(rèn)，推薦用于量化回測） df_forward = tf.klines.get(symbol, period="1d", count=2000, adjust="forward", as_dataframe=True) # 差值前復(fù)權(quán)（與東方財富/同花順價格一致） df_additive = tf.klines.get(symbol, period="1d", count=2000, adjust="forward_additive", as_dataframe=True) # 不復(fù)權(quán)（原始成交價格） df_none = tf.klines.get(symbol, period="1d", count=2000, adjust="none", as_dataframe=True) # 比例后復(fù)權(quán) df_backward = tf.klines.get(symbol, period="1d", count=2000, adjust="backward", as_dataframe=True) # 差值后復(fù)權(quán) df_backward_add = tf.klines.get(symbol, period="1d", count=2000, adjust="backward_additive", as_dataframe=True)
不同復(fù)權(quán)方式的對比
import pandas as pd
compare = pd.DataFrame({
"trade_date": df_none["trade_date"],
"不復(fù)權(quán)": df_none["close"],
"比例前復(fù)權(quán)": df_forward["close"],
"差值前復(fù)權(quán)": df_additive["close"],
"比例后復(fù)權(quán)": df_backward["close"],
})
print(compare.tail(5))
print(f"\n不復(fù)權(quán)最新價: {df_none['close'].iloc[-1]}")
print(f"前復(fù)權(quán)最新價: {df_forward['close'].iloc[-1]}")
print(f"后復(fù)權(quán)最新價: {df_backward['close'].iloc[-1]}")
選哪種復(fù)權(quán)
| 場景 | 推薦復(fù)權(quán)方式 | 原因 |
|---|---|---|
| 收益率計算、量化回測 | forward（比例前復(fù)權(quán)） | 收益率連續(xù)，不失真 |
| 與東方財富/同花順價格核對 | forward_additive（差值前復(fù)權(quán)） | 價格數(shù)值一致 |
| 長期投資收益對比 | backward（后復(fù)權(quán)） | 能直觀看到真實漲幅 |
| 原始價格分析 | none（不復(fù)權(quán)） | 看真實成交價格 |
關(guān)鍵原則：做回測和計算收益率一定要用前復(fù)權(quán)。
查看除權(quán)因子
tf = TickFlow(api_key="your-api-key")
factors = tf.klines.ex_factors(["600519.SH", "000001.SZ"], as_dataframe=True)
# 按標(biāo)的分組查看
for symbol in ["600519.SH", "000001.SZ"]:
sym_factors = factors[factors["symbol"] == symbol]
print(f"\n{symbol} 最近 5 次除權(quán):")
print(sym_factors[["trade_date", "ex_factor"]].tail(5).to_string(index=False))
四、多股票日期對齊
問題：日期不一致
不同股票可能因為停牌、上市時間不同等原因，K 線日期不完全對齊。直接用 merge 會產(chǎn)生大量 NaN。
方案一：基于交易日歷對齊
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)
# 對齊到統(tǒng)一日期索引
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())
方案二：只保留共有日期
# 取交集（所有股票都有數(shù)據(jù)的日期）
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())
方案三：前向填充停牌日
停牌期間用最后一個有效價格填充：
# 在方案一的基礎(chǔ)上，前向填充
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)
五、缺失值處理
常見缺失值來源
- 停牌：股票停牌期間沒有交易數(shù)據(jù)
- 上市時間晚：某只股票的歷史數(shù)據(jù)比其他股票短
- 退市：股票退市后沒有新數(shù)據(jù)
- 數(shù)據(jù)異常：極少數(shù)情況下數(shù)據(jù)源缺失
檢查缺失值
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("各標(biāo)的數(shù)據(jù)量:")
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})")
處理策略
# 將多只股票收盤價合并為一個 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"原始行數(shù): {len(close_df)}, 清洗后: {len(close_clean)}")
# 方法 3：僅刪除缺失值過多的行（某天超過半數(shù)股票無數(shù)據(jù)才刪）
threshold = len(symbols) // 2
close_filtered = close_df.dropna(thresh=threshold)
print(f"過濾后: {len(close_filtered)}")
六、換手率計算
換手率是衡量股票活躍度的重要指標(biāo)，但很多數(shù)據(jù)源不直接提供換手率，需要用成交量和流通股本自己算。
公式
換手率 = 成交量(手) × 100 / 流通股本(股)
A 股成交量單位為"手"（1 手 = 100 股），所以需要乘以 100 換算為股。
實現(xiàn)
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)
# 獲取股本數(shù)據(jù)
sdf = tf.financials.shares([symbol], as_dataframe=True)
# 轉(zhuǎn)為日期類型
kdf["date"] = pd.to_datetime(kdf["trade_date"])
sdf["date"] = pd.to_datetime(sdf["period_end"])
# 使用 merge_asof 為每根 K 線匹配當(dāng)時有效的流通股本
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))
輸出示例：
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
流通股本不是每天都變。公司只在特定時間點（增發(fā)、回購、限售股解禁等）更新股本數(shù)據(jù)。merge_asof 會為每個交易日匹配最近一次公布的流通股本，這是正確的做法。
批量計算換手率
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 對數(shù)收益率
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()
# 對數(shù)收益率（適合統(tǒng)計分析，可加性更好）
df["log_return"] = np.log(df["close"] / df["close"].shift(1))
print(df[["trade_date", "close", "simple_return", "log_return"]].tail(10))
什么時候用哪種
| 場景 | 推薦 |
|---|---|
| 策略收益計算 | 簡單收益率 |
| 統(tǒng)計分析（相關(guān)性、因子收益） | 對數(shù)收益率 |
| 多期累計收益 | 簡單收益率用乘法，對數(shù)收益率用加法 |
累計收益率
# 簡單收益率的累計
df["cum_return"] = (1 + df["simple_return"]).cumprod() - 1
# 或用對數(shù)收益率
df["cum_return_log"] = np.exp(df["log_return"].cumsum()) - 1
print(f"期間總收益: {df['cum_return'].iloc[-1] * 100:.2f}%")
八、數(shù)據(jù)質(zhì)量檢查
拿到數(shù)據(jù)后，建議做一些基礎(chǔ)的數(shù)據(jù)質(zhì)量檢查：
from tickflow import TickFlow
tf = TickFlow.free()
df = tf.klines.get("600000.SH", period="1d", count=5000, adjust="forward", as_dataframe=True)
# 1. 檢查是否有重復(fù)日期
dup = df[df["trade_date"].duplicated()]
print(f"重復(fù)日期: {len(dup)} 個")
# 2. 檢查價格異常（收盤價為 0 或負(fù)數(shù)）
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. 檢查日期連續(xù)性（是否有異常缺失）
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())
九、數(shù)據(jù)緩存與增量更新
每次運(yùn)行策略都重新下載全量數(shù)據(jù)會浪費時間?？梢宰龊唵蔚谋镜鼐彺妫?/p>
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}，增量更新...")
# 獲取緩存之后的新數(shù)據(jù)
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:
# 過濾掉可能重復(fù)的最后一天
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 線")
十、總結(jié)
量化數(shù)據(jù)處理中最容易踩坑的幾個地方：
| 問題 | 正確做法 |
|---|---|
| 不復(fù)權(quán)就算指標(biāo) | 回測必須用前復(fù)權(quán) |
| 多股票日期不對齊 | merge + ffill 或取交集 |
| 缺失值導(dǎo)致計算出錯 | 先檢查再決定填充或刪除 |
| 換手率不知道怎么算 | volume * 100 / float_shares，用 merge_asof |
| 收益率用錯方法 | 策略用簡單收益率，統(tǒng)計分析用對數(shù)收益率 |
| 每次都全量下載 | 做本地緩存 + 增量更新 |
這些技巧看似基礎(chǔ)，但每一個處理不當(dāng)都可能導(dǎo)致回測失真或策略結(jié)論錯誤。
TickFlow 在數(shù)據(jù)層面已經(jīng)做了很多：多種復(fù)權(quán)方式開箱即用、批量接口高效穩(wěn)定、股本數(shù)據(jù)可直接獲取。在此基礎(chǔ)上做好數(shù)據(jù)處理，你的量化研究才有可靠的地基。
相關(guān)鏈接
- 官網(wǎng)：tickflow.org
- 文檔：docs.tickflow.org
- Github：github.com/tickflow-org/tickflow
數(shù)據(jù)處理不性感，但它決定了你的回測結(jié)果是真是假。
以上就是詳解Python中量化數(shù)據(jù)的處理技巧(附實戰(zhàn)代碼)的詳細(xì)內(nèi)容，更多關(guān)于Python量化數(shù)據(jù)處理的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
pycharm執(zhí)行python時,填寫參數(shù)的方法
今天小編就為大家分享一篇pycharm執(zhí)行python時,填寫參數(shù)的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-10-10
python神經(jīng)網(wǎng)絡(luò)MobileNetV3?small模型的復(fù)現(xiàn)詳解
這篇文章主要為大家介紹了python神經(jīng)網(wǎng)絡(luò)MobileNetV3?small模型的復(fù)現(xiàn)詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-05-05
這篇文章主要介紹了Python實現(xiàn)的爬取百度貼吧圖片功能,結(jié)合完整實例形式分析了Python實現(xiàn)的百度貼吧圖片爬蟲相關(guān)操作技巧,需要的朋友可以參考下2019-05-05
在本篇文章里小編給大家整理的是關(guān)于python數(shù)據(jù)爬下來保存的位置，需要的朋友們可以參考下。2020-02-02
python3使用Flask實現(xiàn)api數(shù)據(jù)接口的示例詳解
這篇文章主要為大家詳細(xì)介紹了python3如何使用Flask實現(xiàn)api數(shù)據(jù)接口,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2025-09-09
這篇文章主要介紹了Python的反射機(jī)制，幫助大家更好的理解和使用python，感興趣的朋友可以了解下2020-09-09
Playwright的wait funtion測試的實現(xiàn)
本文主要介紹了Playwright的wait funtion測試的實現(xiàn),文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2026-04-04
Python?GUI利用tkinter皮膚ttkbootstrap實現(xiàn)好看的窗口
這篇文章主要介紹了Python?GUI利用tkinter皮膚ttkbootstrap實現(xiàn)好看的窗口，文章基于python的相關(guān)資料展開詳細(xì)的內(nèi)容介紹，具有一定的參考價值，感興趣的小伙伴可以參考一下2022-06-06
python3實現(xiàn)全角和半角字符轉(zhuǎn)換的方法示例
在自然語言處理過程中，全角、半角的的不一致會導(dǎo)致信息抽取不一致，因此需要統(tǒng)一，下面這篇文章主要給大家介紹了關(guān)于python3中全角和半角字符轉(zhuǎn)換的方法，需要的朋友可以參考借鑒，下面來一起看看吧。2017-09-09
本文介紹了Python實現(xiàn)圖像和辦公文檔處理的方法和技巧，包括使用Pillow庫處理圖像、使用OpenCV庫進(jìn)行圖像識別和處理、使用PyPDF2庫處理PDF文檔、使用docx和xlwt庫處理Word和Excel文檔等，幫助讀者更好地掌握Python在圖像和辦公文檔處理方面的應(yīng)用2023-05-05