前言在學(xué)習(xí) Python 量化交易的過程中很多人一開始就想著策略回測自動(dòng)交易但實(shí)際上一個(gè)量化系統(tǒng)最基礎(chǔ)的部分往往是數(shù)據(jù)處理。如果行情數(shù)據(jù)本身存在缺失重復(fù)時(shí)間錯(cuò)誤價(jià)格異常字段錯(cuò)誤那么后面的策略回測即使代碼完全正確也可能得到錯(cuò)誤結(jié)果。因此一個(gè)比較完整的量化研究流程應(yīng)該從數(shù)據(jù)開始。本文使用 Python 和 Pandas搭建一個(gè)簡單的歷史行情分析流程。一、準(zhǔn)備一份行情數(shù)據(jù)假設(shè) CSV 文件包含datetimeopenhighlowclosevolume例如2026-08-01,3200,3250,3180,3230,1250002026-08-02,3230,3280,3210,3260,1380002026-08-03,3260,3290,3220,3240,119000二、使用 Pandas 讀取數(shù)據(jù)首先import pandas as pd讀取CSVdf pd.read_csv(“market_data.csv”)查看前幾行print(df.head())查看數(shù)據(jù)結(jié)構(gòu)print(df.info())三、處理時(shí)間字段量化數(shù)據(jù)中時(shí)間字段非常重要。先轉(zhuǎn)換df[“datetime”] pd.to_datetime(df[“datetime”])然后排序df df.sort_values(“datetime”)再重新設(shè)置索引df df.set_index(“datetime”)這樣后續(xù)進(jìn)行時(shí)間序列分析會方便很多。四、檢查重復(fù)數(shù)據(jù)可以使用duplicates df.index.duplicated()print(duplicates.sum())如果結(jié)果大于0說明存在重復(fù)時(shí)間記錄?？梢赃M(jìn)一步df df[~df.index.duplicated(keep“l(fā)ast”)]保留最后一條記錄。五、檢查缺失值使用print(df.isnull().sum())例如open 0high 0low 0close 3volume 0說明close存在3條缺失數(shù)據(jù)。不要直接忽略。需要根據(jù)數(shù)據(jù)頻率和數(shù)據(jù)來源決定如何處理。六、檢查價(jià)格邏輯OHLC數(shù)據(jù)應(yīng)該滿足基本關(guān)系High OpenHigh CloseLow OpenLow Close可以檢查invalid df[(df[“high”] df[“open”]) |(df[“high”] df[“close”]) |(df[“l(fā)ow”] df[“open”]) |(df[“l(fā)ow”] df[“close”])]print(invalid)如果出現(xiàn)異常記錄就應(yīng)該進(jìn)一步檢查原始數(shù)據(jù)。七、計(jì)算收益率最簡單的收益率df[“return”] (df[“close”].pct_change())查看print(df[[“close”, “return”]].tail())這樣就得到了每個(gè)時(shí)間點(diǎn)的價(jià)格變化。八、計(jì)算20日移動(dòng)平均線df[“ma20”] (df[“close”].rolling(20).mean())60日均線df[“ma60”] (df[“close”].rolling(60).mean())現(xiàn)在數(shù)據(jù)中就多出了MA20MA60九、使用均線判斷趨勢可以定義一個(gè)非常簡單的趨勢狀態(tài)df[“trend”] 0df.loc[df[“ma20”] df[“ma60”],“trend”] 1df.loc[df[“ma20”] df[“ma60”],“trend”] -1含義1短期趨勢偏強(qiáng)以及-1短期趨勢偏弱十、計(jì)算波動(dòng)率可以使用收益率標(biāo)準(zhǔn)差df[“volatility”] (df[“return”].rolling(20).std())這樣可以觀察最近20個(gè)周期的價(jià)格波動(dòng)情況。十一、計(jì)算成交量變化例如計(jì)算20周期平均成交量df[“volume_ma20”] (df[“volume”].rolling(20).mean())再計(jì)算成交量比df[“volume_ratio”] (df[“volume”] /df[“volume_ma20”])如果volume_ratio 1說明當(dāng)前成交量高于近期平均水平。十二、建立一個(gè)簡單的研究信號現(xiàn)在假設(shè)我們研究一個(gè)非常簡單的趨勢條件MA20 MA60并且成交量 20周期平均成交量Pythondf[“signal”] ((df[“ma20”] df[“ma60”])(df[“volume”] df[“volume_ma20”]))得到TrueFalse這樣的信號。注意這只是一個(gè)研究示例并不代表一個(gè)完整交易策略。十三、為什么信號和交易執(zhí)行應(yīng)該分開這是量化系統(tǒng)設(shè)計(jì)中非常重要的一點(diǎn)。不要把所有邏輯寫在一個(gè)函數(shù)里。例如數(shù)據(jù)模塊↓指標(biāo)模塊↓信號模塊↓交易模塊↓風(fēng)險(xiǎn)控制↓回測模塊這樣以后修改策略時(shí)不需要把整個(gè)程序推倒重來。十四、把數(shù)據(jù)處理封裝成函數(shù)例如def prepare_data(path):df pd.read_csv(path) df[datetime] pd.to_datetime( df[datetime] ) df df.sort_values( datetime ) df df.set_index( datetime ) df[return] ( df[close] .pct_change() ) df[ma20] ( df[close] .rolling(20) .mean() ) df[ma60] ( df[close] .rolling(60) .mean() ) return df使用df prepare_data(“market_data.csv”)print(df.tail())以后換數(shù)據(jù)文件只需要修改路徑。十五、數(shù)據(jù)處理完成后再進(jìn)入回測比較合理的研究流程原始行情數(shù)據(jù)↓數(shù)據(jù)清洗↓數(shù)據(jù)驗(yàn)證↓指標(biāo)計(jì)算↓信號生成↓回測↓資金曲線↓風(fēng)險(xiǎn)指標(biāo)而不是CSV↓直接買賣十六、為什么數(shù)據(jù)質(zhì)量比策略復(fù)雜度更重要假設(shè)一個(gè)策略只有兩條均線但是數(shù)據(jù)非常準(zhǔn)確。另一個(gè)策略使用20個(gè)指標(biāo)但是行情數(shù)據(jù)存在缺失重復(fù)未來數(shù)據(jù)時(shí)間錯(cuò)位那么后者即使回測收益看起來非常漂亮也沒有太大意義。因此量化研究中經(jīng)常需要遵循先保證數(shù)據(jù)正確再討論策略效果。十七、進(jìn)一步加入數(shù)據(jù)質(zhì)量檢查以后可以建立一個(gè)統(tǒng)一的數(shù)據(jù)檢查函數(shù)def validate_data(df):errors [] if df.empty: errors.append( 數(shù)據(jù)為空 ) if df.index.duplicated().any(): errors.append( 存在重復(fù)時(shí)間 ) if df[close].isnull().any(): errors.append( 存在缺失收盤價(jià) ) if ( df[high] df[low] ).any(): errors.append( 存在High Low ) return errors調(diào)用errors validate_data(df)if errors:print(“數(shù)據(jù)存在問題”)for error in errors: print(error)else:print(“數(shù)據(jù)檢查通過”)這樣以后建立自動(dòng)化數(shù)據(jù)倉庫時(shí)就可以把數(shù)據(jù)檢查作為固定步驟。十八、從簡單腳本逐漸發(fā)展成量化研究系統(tǒng)一個(gè)小型 Python 量化項(xiàng)目最終可以逐漸變成quant_project/│├── data/│ ├── raw/│ └── processed/│├── indicators/│├── strategies/│├── backtest/│├── risk/│├── reports/│└── main.py這樣比把所有代碼全部放到main.py里面更加容易維護(hù)。十九、AI 也可以加入這個(gè)流程如果以后結(jié)合上一篇文章中的 AI Agent可以形成用戶↓AI Agent↓讀取行情數(shù)據(jù)↓Python數(shù)據(jù)清洗↓計(jì)算指標(biāo)↓執(zhí)行策略分析↓生成研究報(bào)告例如“幫我分析RB最近一個(gè)月的趨勢?！盇gent可以自動(dòng)調(diào)用讀取數(shù)據(jù)↓計(jì)算MA↓計(jì)算波動(dòng)率↓分析成交量↓生成報(bào)告這就是AI Python 量化數(shù)據(jù)分析比較適合進(jìn)一步開發(fā)的方向?？偨Y(jié)一個(gè)完整的 Python 量化研究流程并不是從“寫買賣策略”開始。更加合理的順序是獲取數(shù)據(jù)↓清洗數(shù)據(jù)↓驗(yàn)證數(shù)據(jù)↓計(jì)算指標(biāo)↓生成信號↓回測↓分析資金曲線↓評估風(fēng)險(xiǎn)其中最基礎(chǔ)、也最容易被忽略的就是數(shù)據(jù)質(zhì)量。只有確保數(shù)據(jù)可靠后面的策略收益、最大回撤、Sharpe 等指標(biāo)才有研究價(jià)值。如果后面再結(jié)合 AI Agent還可以進(jìn)一步把數(shù)據(jù)獲取指標(biāo)計(jì)算策略分析報(bào)告生成整合成一個(gè)自動(dòng)化的 AI 量化研究助手。