：從數(shù)據(jù)獲取到雙均線策略回測的完整閉環(huán))
最近在嘗試將量化交易從理論轉(zhuǎn)向?qū)嵺`時很多開發(fā)者朋友都遇到過類似的困境網(wǎng)上資料要么過于學術(shù)化要么就是零散的代碼片段真正能跑通、能理解、能復現(xiàn)的完整教程少之又少。自己摸索時常常在數(shù)據(jù)獲取、策略回測、實盤對接這些環(huán)節(jié)反復踩坑浪費大量時間。本文旨在整合一套從零到一的 Python 量化交易實戰(zhàn)閉環(huán)。我們將手把手搭建一個完整的量化分析環(huán)境編寫并回測一個簡單的雙均線策略最終探討策略優(yōu)化與風險管理的思路。無論你是對量化感興趣的學生還是希望將策略自動化的開發(fā)者都能從這套保姆級流程中找到可復用的代碼和清晰的路徑避開那些常見的“彎路”。1. 量化交易核心概念與準備工作在開始寫代碼之前我們需要統(tǒng)一對幾個核心概念的理解并準備好相應的“武器庫”。1.1 什么是量化交易量化交易Quantitative Trading的核心是利用數(shù)學模型、統(tǒng)計分析和計算機技術(shù)從海量歷史數(shù)據(jù)中尋找能夠預測未來價格走勢的規(guī)律即“因子”或“Alpha”并據(jù)此構(gòu)建可自動執(zhí)行的交易策略。它與傳統(tǒng)主觀交易的最大區(qū)別在于系統(tǒng)性決策基于明確的規(guī)則和模型排除了情緒干擾?？苫厮莶呗栽跉v史數(shù)據(jù)上的表現(xiàn)回測可以量化評估。自動化策略生成交易信號后可通過程序自動執(zhí)行訂單。一個完整的量化交易流程通常包括數(shù)據(jù)獲取、策略研究、回測驗證、風險控制、實盤交易和績效分析。1.2 環(huán)境準備與工具選型工欲善其事必先利其器。我們選擇 Python 作為實現(xiàn)語言因為它擁有極其豐富的金融數(shù)據(jù)分析庫和社區(qū)生態(tài)。1. 基礎環(huán)境操作系統(tǒng)Windows 10/11, macOS 或 Linux (如 Ubuntu) 均可。本文示例在 Windows 11 下完成。Python 版本推薦使用 Python 3.8 或 3.9這兩個版本與多數(shù)金融庫的兼容性最好。避免使用 Python 3.10 的早期版本可能遇到某些庫的安裝問題。包管理工具使用pip即可。強烈建議使用虛擬環(huán)境如venv或conda來隔離項目依賴。2. 核心工具庫介紹我們將使用以下庫它們構(gòu)成了量化分析的基石pandasnumpy數(shù)據(jù)處理和科學計算的絕對核心用于數(shù)據(jù)清洗、轉(zhuǎn)換和計算。akshare一個免費、開源的 Python 財經(jīng)數(shù)據(jù)接口庫。它提供了國內(nèi)股票、基金、期貨、期權(quán)等大量金融數(shù)據(jù)替代了部分付費數(shù)據(jù)源非常適合學習和個人研究。mplfinance基于 Matplotlib 的金融數(shù)據(jù)可視化庫可以輕松繪制K線圖、均線等。backtrader或zipline專業(yè)的回測框架。backtrader功能強大、靈活社區(qū)活躍zipline是 Quantopian 開源的回測引擎結(jié)構(gòu)嚴謹。本文為求簡單直觀將手動實現(xiàn)回測邏輯以便理解每一步原理。在實際復雜策略中強烈推薦使用這些成熟框架。3. 安裝依賴在你的項目虛擬環(huán)境中執(zhí)行以下命令安裝所需庫pip install pandas numpy akshare mplfinance如果網(wǎng)絡不佳可以使用國內(nèi)鏡像源加速pip install pandas numpy akshare mplfinance -i https://pypi.tuna.tsinghua.edu.cn/simple2. 第一步獲取與處理金融數(shù)據(jù)沒有數(shù)據(jù)量化就是無源之水。可靠、干凈的數(shù)據(jù)是策略成功的首要前提。2.1 使用 AKShare 獲取股票歷史數(shù)據(jù)AKShare 接口非常豐富我們以獲取“貴州茅臺”股票代碼 600519.SH的日線數(shù)據(jù)為例。# 文件data_fetcher.py import akshare as ak import pandas as pd import warnings warnings.filterwarnings(ignore) # 忽略一些不必要的警告 def fetch_stock_data(symbol, start_date, end_date): 獲取指定股票代碼的日線行情數(shù)據(jù) :param symbol: 股票代碼如 600519 (AKShare內(nèi)部會自動補全市場) :param start_date: 開始日期格式 YYYYMMDD :param end_date: 結(jié)束日期格式 YYYYMMDD :return: 包含OHLCV等數(shù)據(jù)的 pandas DataFrame # 使用 ak.stock_zh_a_hist 接口注意參數(shù)名 df ak.stock_zh_a_hist(symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq) # 查看原始數(shù)據(jù)列名 print(原始數(shù)據(jù)列名, df.columns.tolist()) # 重命名列使其更符合通用標準 df.rename(columns{ 日期: date, 開盤: open, 收盤: close, 最高: high, 最低: low, 成交量: volume, 成交額: amount, 振幅: amplitude, 漲跌幅: pct_chg, 漲跌額: change, 換手率: turnover }, inplaceTrue) # 將日期列設置為索引并轉(zhuǎn)換為datetime格式便于后續(xù)時間序列分析 df[date] pd.to_datetime(df[date]) df.set_index(date, inplaceTrue) # 按日期升序排列 df.sort_index(inplaceTrue) return df if __name__ __main__: # 示例獲取貴州茅臺2023年全年的數(shù)據(jù) stock_code 600519 # 貴州茅臺 start 20230101 end 20231231 maotai_data fetch_stock_data(stock_code, start, end) print(f數(shù)據(jù)形狀{maotai_data.shape}) print(maotai_data.head()) # 查看前5行 print(maotai_data.tail()) # 查看后5行運行這段代碼你將得到一個DataFrame索引是日期包含了開盤價、收盤價、最高價、最低價、成交量等關鍵信息。adjustqfq參數(shù)表示獲取前復權(quán)數(shù)據(jù)這能保證歷史價格的可比性是回測的標準做法。2.2 數(shù)據(jù)清洗與特征工程拿到的原始數(shù)據(jù)可能存在缺失值、異常值或需要計算衍生指標如技術(shù)指標。# 文件data_processor.py import pandas as pd import numpy as np def calculate_technical_indicators(df): 計算常用的技術(shù)指標 :param df: 包含OHLCV數(shù)據(jù)的DataFrame :return: 添加了技術(shù)指標列的DataFrame df df.copy() # 1. 簡單移動平均線 (SMA) df[SMA_10] df[close].rolling(window10, min_periods1).mean() df[SMA_30] df[close].rolling(window30, min_periods1).mean() # 2. 指數(shù)移動平均線 (EMA) - 對近期價格賦予更高權(quán)重 df[EMA_12] df[close].ewm(span12, adjustFalse).mean() df[EMA_26] df[close].ewm(span26, adjustFalse).mean() # 3. 計算MACD指標 df[MACD] df[EMA_12] - df[EMA_26] df[MACD_Signal] df[MACD].ewm(span9, adjustFalse).mean() df[MACD_Histogram] df[MACD] - df[MACD_Signal] # 4. 布林帶 (Bollinger Bands) df[BB_Middle] df[close].rolling(window20).mean() bb_std df[close].rolling(window20).std() df[BB_Upper] df[BB_Middle] 2 * bb_std df[BB_Lower] df[BB_Middle] - 2 * bb_std # 5. 相對強弱指數(shù) (RSI) delta df[close].diff() gain (delta.where(delta 0, 0)).rolling(window14).mean() loss (-delta.where(delta 0, 0)).rolling(window14).mean() rs gain / loss df[RSI] 100 - (100 / (1 rs)) # 處理計算產(chǎn)生的初始NaN值例如前29天沒有30日均線 # 這里簡單用前向填充但更嚴謹?shù)淖龇ㄊ翘蕹蚴褂闷渌椒?df.fillna(methodffill, inplaceTrue) df.dropna(inplaceTrue) # 刪除仍為NaN的行 return df if __name__ __main__: # 假設 maotai_data 是上一節(jié)獲取的數(shù)據(jù) from data_fetcher import fetch_stock_data data fetch_stock_data(600519, 20230101, 20231231) processed_data calculate_technical_indicators(data) print(processed_data[[close, SMA_10, SMA_30, RSI]].head(35)) # 查看包含指標的數(shù)據(jù)這個函數(shù)為數(shù)據(jù)添加了多個常用技術(shù)指標。在量化策略中這些指標常被用作產(chǎn)生交易信號的“因子”。3. 第二步策略構(gòu)思與回測實現(xiàn)策略是量化的靈魂。我們從一個最經(jīng)典的趨勢跟蹤策略——雙均線交叉策略開始。3.1 策略邏輯雙均線交叉策略規(guī)則非常簡單買入信號金叉當短期均線如10日SMA從下向上穿越長期均線如30日SMA時買入股票。賣出信號死叉當短期均線從上向下穿越長期均線時賣出持有的股票。持倉其余時間持有股票如果已買入或空倉。這個策略的邏輯是捕捉趨勢的形成金叉意味著上漲趨勢可能開始死叉意味著下跌趨勢可能開始。3.2 手動實現(xiàn)回測引擎為了徹底理解回測的每個細節(jié)我們不依賴框架自己實現(xiàn)一個簡單的回測系統(tǒng)。# 文件backtest_engine.py import pandas as pd import numpy as np class SimpleBacktester: def __init__(self, data, initial_capital100000.0): 初始化回測器 :param data: 包含價格和指標數(shù)據(jù)的DataFrame :param initial_capital: 初始資金 self.data data.copy() self.initial_capital initial_capital self.current_capital initial_capital self.position 0 # 當前持有股票數(shù)量 self.trades [] # 記錄所有交易 self.equity_curve [] # 記錄每日資產(chǎn)總值 def generate_signals(self): 根據(jù)雙均線交叉生成交易信號 # 確保數(shù)據(jù)中有我們需要的均線 if SMA_10 not in self.data.columns or SMA_30 not in self.data.columns: raise ValueError(數(shù)據(jù)中缺少SMA_10或SMA_30列請先計算技術(shù)指標。) # 生成信號1 代表買入 -1 代表賣出 0 代表無操作 self.data[signal] 0 # 金叉SMA_10上穿SMA_30 golden_cross (self.data[SMA_10] self.data[SMA_30]) (self.data[SMA_10].shift(1) self.data[SMA_30].shift(1)) # 死叉SMA_10下穿SMA_30 death_cross (self.data[SMA_10] self.data[SMA_30]) (self.data[SMA_10].shift(1) self.data[SMA_30].shift(1)) self.data.loc[golden_cross, signal] 1 self.data.loc[death_cross, signal] -1 # 為了簡化我們假設信號產(chǎn)生在當天收盤時第二天以開盤價執(zhí)行 self.data[signal] self.data[signal].shift(1) self.data[signal].fillna(0, inplaceTrue) def run_backtest(self): 執(zhí)行回測 print(開始回測...) self.generate_signals() for i, row in self.data.iterrows(): current_price row[close] signal row[signal] # 記錄當前總資產(chǎn)現(xiàn)金 持倉市值 current_equity self.current_capital self.position * current_price self.equity_curve.append({ date: i, equity: current_equity, price: current_price, position: self.position }) # 執(zhí)行交易信號 if signal 1 and self.position 0: # 買入信號且空倉 # 計算可買數(shù)量假設可以買任意股這里取整 # 實際中應考慮手續(xù)費、最小交易單位A股是100股等 shares_to_buy int(self.current_capital // current_price) if shares_to_buy 0: cost shares_to_buy * current_price self.current_capital - cost self.position shares_to_buy self.trades.append({ date: i, type: BUY, price: current_price, shares: shares_to_buy, cost: cost, cash_after: self.current_capital }) print(f{i.date()}: 買入 {shares_to_buy} 股價格 {current_price:.2f}) elif signal -1 and self.position 0: # 賣出信號且持有 proceeds self.position * current_price self.current_capital proceeds self.trades.append({ date: i, type: SELL, price: current_price, shares: self.position, proceeds: proceeds, cash_after: self.current_capital }) print(f{i.date()}: 賣出 {self.position} 股價格 {current_price:.2f}) self.position 0 # 回測結(jié)束平倉如果最后還持有 if self.position 0: last_price self.data[close].iloc[-1] proceeds self.position * last_price self.current_capital proceeds self.trades.append({ date: self.data.index[-1], type: SELL, price: last_price, shares: self.position, proceeds: proceeds, cash_after: self.current_capital }) print(f回測結(jié)束平倉 {self.position} 股價格 {last_price:.2f}) self.position 0 print(回測完成。) def get_results(self): 計算并返回回測績效指標 if not self.equity_curve: raise ValueError(請先運行 run_backtest()) equity_df pd.DataFrame(self.equity_curve) equity_df.set_index(date, inplaceTrue) # 計算累計收益率 initial_equity self.initial_capital final_equity equity_df[equity].iloc[-1] total_return (final_equity - initial_equity) / initial_equity # 計算年化收益率假設一年約242個交易日 days (equity_df.index[-1] - equity_df.index[0]).days years days / 365.0 annualized_return (1 total_return) ** (1 / years) - 1 if years 0 else 0 # 計算最大回撤最大虧損幅度 equity_series equity_df[equity] cumulative_max equity_series.expanding().max() drawdown (equity_series - cumulative_max) / cumulative_max max_drawdown drawdown.min() # 夏普比率簡化版假設無風險利率為0 daily_returns equity_df[equity].pct_change().dropna() if daily_returns.std() ! 0: sharpe_ratio np.sqrt(242) * (daily_returns.mean() / daily_returns.std()) # 年化夏普 else: sharpe_ratio 0 results { 初始資金: initial_equity, 最終資產(chǎn): final_equity, 總收益率: total_return, 年化收益率: annualized_return, 最大回撤: max_drawdown, 夏普比率: sharpe_ratio, 交易次數(shù): len(self.trades) } return results, equity_df, pd.DataFrame(self.trades) if __name__ __main__: # 整合之前的步驟 from data_fetcher import fetch_stock_data from data_processor import calculate_technical_indicators # 1. 獲取并處理數(shù)據(jù) raw_data fetch_stock_data(600519, 20220101, 20231231) # 獲取兩年數(shù)據(jù) processed_data calculate_technical_indicators(raw_data) # 2. 創(chuàng)建回測實例并運行 backtester SimpleBacktester(processed_data, initial_capital100000) backtester.run_backtest() # 3. 獲取結(jié)果 results, equity_curve, trades backtester.get_results() print(\n *50) print(回測績效總結(jié):) print(*50) for key, value in results.items(): if isinstance(value, float): print(f{key}: {value:.4%} if 率 in key or 撤 in key else f{key}: {value:.4f}) else: print(f{key}: {value}) print(f\n交易記錄 (共{len(trades)}筆):) if not trades.empty: print(trades[[date, type, price, shares]].head())4. 第三步結(jié)果可視化與初步分析數(shù)字結(jié)果不夠直觀我們需要圖表來評估策略表現(xiàn)。4.1 繪制資產(chǎn)曲線與買賣點# 文件visualization.py import matplotlib.pyplot as plt import mplfinance as mpf import pandas as pd def plot_equity_curve(equity_df, trades_df, data): 繪制資產(chǎn)曲線和買賣點 :param equity_df: 回測引擎輸出的equity_curve DataFrame :param trades_df: 回測引擎輸出的trades DataFrame :param data: 原始行情數(shù)據(jù)DataFrame fig, axes plt.subplots(3, 1, figsize(14, 10), gridspec_kw{height_ratios: [3, 1, 1]}) # 子圖1價格、均線與買賣點 ax1 axes[0] ax1.plot(data.index, data[close], label收盤價, linewidth1, alpha0.7) ax1.plot(data.index, data[SMA_10], label10日均線, linewidth1, alpha0.9) ax1.plot(data.index, data[SMA_30], label30日均線, linewidth1, alpha0.9) # 標記買賣點 buy_trades trades_df[trades_df[type] BUY] sell_trades trades_df[trades_df[type] SELL] ax1.scatter(buy_trades[date], data.loc[buy_trades[date], close], colorred, marker^, s100, label買入, zorder5) ax1.scatter(sell_trades[date], data.loc[sell_trades[date], close], colorgreen, markerv, s100, label賣出, zorder5) ax1.set_title(雙均線策略 - 價格與交易信號) ax1.set_ylabel(價格 (元)) ax1.legend() ax1.grid(True, alpha0.3) # 子圖2資產(chǎn)曲線 ax2 axes[1] ax2.plot(equity_df.index, equity_df[equity], label總資產(chǎn), colorblue, linewidth2) ax2.fill_between(equity_df.index, equity_df[equity], equity_df[equity].min(), alpha0.3) ax2.set_title(策略資產(chǎn)曲線) ax2.set_ylabel(資產(chǎn) (元)) ax2.legend() ax2.grid(True, alpha0.3) # 子圖3持倉情況 ax3 axes[2] ax3.bar(equity_df.index, equity_df[position], label持倉股數(shù), colororange, alpha0.6) ax3.set_title(持倉變化) ax3.set_xlabel(日期) ax3.set_ylabel(持倉 (股)) ax3.legend() ax3.grid(True, alpha0.3) plt.tight_layout() plt.show() def plot_kline_with_indicators(data): 使用mplfinance繪制K線圖及均線 # 準備mplfinance需要的格式 plot_data data[[open, high, low, close, volume]].copy() plot_data.columns [Open, High, Low, Close, Volume] # 列名需大寫 # 添加均線作為附加繪圖 apds [ mpf.make_addplot(data[SMA_10], colororange, width0.7), mpf.make_addplot(data[SMA_30], colorblue, width0.7), ] mpf.plot(plot_data, typecandle, stylecharles, titleK線圖與雙均線, ylabel價格, volumeTrue, addplotapds, figratio(12, 8), figscale1.2) if __name__ __main__: # 假設已有回測結(jié)果 from backtest_engine import SimpleBacktester from data_fetcher import fetch_stock_data from data_processor import calculate_technical_indicators data calculate_technical_indicators(fetch_stock_data(600519, 20230101, 20231231)) bt SimpleBacktester(data, 100000) bt.run_backtest() results, equity_df, trades_df bt.get_results() # 繪制圖表 plot_equity_curve(equity_df, trades_df, data) plot_kline_with_indicators(data)運行可視化代碼你將看到三張圖第一張是價格走勢、均線以及買賣點標記第二張是你的資產(chǎn)總值變化曲線第三張是持倉數(shù)量變化。這能幫你直觀判斷策略是否跟上了趨勢以及買賣時機是否合理。5. 常見問題與策略缺陷分析第一個能跑起來的策略往往充滿問題。以下是雙均線策略及我們簡易回測中常見的坑點。5.1 回測中的常見陷阱與解決方案問題現(xiàn)象常見原因解決思路與方案回測結(jié)果過于完美年化收益驚人1.未來函數(shù)使用了當時還未產(chǎn)生的數(shù)據(jù)如用當天收盤價計算信號并當天交易。2.忽略交易成本未計算傭金、印花稅、滑點。3.過擬合策略參數(shù)恰好完美匹配了某段歷史數(shù)據(jù)。1.確保信號滯后信號基于t-1日的數(shù)據(jù)交易在t日執(zhí)行如我們代碼中的shift(1)。2.加入成本模型在買賣時扣除固定比例或固定金額的費用。3.樣本外測試用另一段未參與參數(shù)優(yōu)化的歷史數(shù)據(jù)驗證策略。實盤表現(xiàn)與回測差異巨大1.流動性假設回測假設可以立即以收盤價成交實盤可能無法成交或成交價更差。2.數(shù)據(jù)質(zhì)量回測數(shù)據(jù)是清洗過的復權(quán)數(shù)據(jù)實盤數(shù)據(jù)是實時、原始的。3.市場狀態(tài)變化策略有效的市場環(huán)境如強趨勢市可能已改變。1.引入滑點模型假設成交價比信號價差一個固定或隨機比例。2.使用更精細的數(shù)據(jù)回測時使用Tick級或分鐘級數(shù)據(jù)。3.考慮市場機制加入漲跌停板、T1等A股特殊規(guī)則。交易次數(shù)過多或過少1.參數(shù)敏感均線周期設置太短導致頻繁交叉噪音交易太長則信號遲鈍。2.數(shù)據(jù)波動性標的物本身波動性大或小。1.參數(shù)優(yōu)化與穩(wěn)健性檢驗測試不同參數(shù)組合觀察績效是否穩(wěn)定。2.添加過濾器例如要求金叉時價格需同時上穿長期均線一定幅度或結(jié)合波動率指標。最大回撤過大策略在震蕩市中反復“高買低賣”虧損累積。趨勢策略在無趨勢市場中的通病。1.加入止損機制單筆虧損達到一定比例或金額時強制平倉。2.結(jié)合市場擇時增加一個趨勢過濾器例如在大盤指數(shù)處于200日均線之上時才運行該策略。5.2 我們的簡易回測引擎的局限性我們的SimpleBacktester為了教學而極度簡化認識到這些局限是進步的第一步單標的、全倉進出現(xiàn)實中需要管理多只股票和資金分配。無風險控制沒有止損止盈、倉位控制。簡化交易邏輯假設無限流動性、無手續(xù)費、無最小交易單位限制A股需100股整數(shù)倍。未考慮分紅送股雖然用了前復權(quán)數(shù)據(jù)但策略本身未處理除權(quán)除息日的特殊邏輯。6. 策略優(yōu)化與進階方向一個基本的策略跑通后量化工作才真正開始。以下是幾個關鍵的優(yōu)化和進階方向。6.1 參數(shù)優(yōu)化與穩(wěn)健性檢驗不要只尋找歷史數(shù)據(jù)上表現(xiàn)最好的參數(shù)這大概率是過擬合。應該尋找在多個時間段、多只股票上都表現(xiàn)穩(wěn)定的參數(shù)。# 文件parameter_optimization.py import itertools from backtest_engine import SimpleBacktester from data_fetcher import fetch_stock_data from data_processor import calculate_technical_indicators def grid_search_parameters(stock_code, start, end, short_range, long_range): 網(wǎng)格搜索雙均線的最佳參數(shù)組合 :param short_range: 短期均線周期范圍如 range(5, 21, 5) :param long_range: 長期均線周期范圍如 range(20, 61, 10) results_list [] data_raw fetch_stock_data(stock_code, start, end) for short_period, long_period in itertools.product(short_range, long_range): if short_period long_period: continue # 短期必須小于長期 print(f測試參數(shù): SMA({short_period}, {long_period})) # 重新計算指標 data data_raw.copy() data[fSMA_{short_period}] data[close].rolling(windowshort_period).mean() data[fSMA_{long_period}] data[close].rolling(windowlong_period).mean() data.fillna(methodffill, inplaceTrue) data.dropna(inplaceTrue) # 運行回測 bt SimpleBacktester(data, 100000) bt.run_backtest() perf, _, _ bt.get_results() results_list.append({ short_period: short_period, long_period: long_period, total_return: perf[總收益率], max_drawdown: perf[最大回撤], sharpe_ratio: perf[夏普比率], trade_count: perf[交易次數(shù)] }) results_df pd.DataFrame(results_list) # 按夏普比率排序綜合考慮收益與風險 results_df.sort_values(sharpe_ratio, ascendingFalse, inplaceTrue) return results_df if __name__ __main__: # 測試不同參數(shù)組合 best_params_df grid_search_parameters(600519, 20200101, 20231231, range(5, 21, 5), range(30, 101, 15)) print(\n參數(shù)優(yōu)化結(jié)果按夏普比率降序:) print(best_params_df.head(10))6.2 引入風險管理模塊一個沒有風險控制的策略是危險的。至少應加入止損。# 文件risk_management.py class BacktesterWithRisk(SimpleBacktester): 增加了風險管理的回測器 def __init__(self, data, initial_capital100000.0, stop_loss_pct0.05): super().__init__(data, initial_capital) self.stop_loss_pct stop_loss_pct # 止損比例如5% self.entry_price None # 記錄買入成本價 def run_backtest(self): self.generate_signals() for i, row in self.data.iterrows(): current_price row[close] signal row[signal] # --- 止損檢查 --- if self.position 0 and self.entry_price is not None: # 計算從買入后的最大虧損幅度 loss_pct (current_price - self.entry_price) / self.entry_price if loss_pct -self.stop_loss_pct: # 觸發(fā)止損賣出所有持倉 proceeds self.position * current_price self.current_capital proceeds self.trades.append({ date: i, type: SELL_STOPLOSS, price: current_price, shares: self.position, proceeds: proceeds, cash_after: self.current_capital }) print(f{i.date()}: 止損觸發(fā)賣出 {self.position} 股價格 {current_price:.2f}虧損 {loss_pct:.2%}) self.position 0 self.entry_price None # 止損后跳過本日其他信號防止立即反向開倉 continue # 原有交易邏輯... current_equity self.current_capital self.position * current_price self.equity_curve.append({date: i, equity: current_equity, price: current_price, position: self.position}) if signal 1 and self.position 0: shares_to_buy int(self.current_capital // current_price) if shares_to_buy 0: cost shares_to_buy * current_price self.current_capital - cost self.position shares_to_buy self.entry_price current_price # 記錄買入價 self.trades.append({date: i, type: BUY, price: current_price, shares: shares_to_buy, cost: cost}) print(f{i.date()}: 買入 {shares_to_buy} 股價格 {current_price:.2f}) elif signal -1 and self.position 0: proceeds self.position * current_price self.current_capital proceeds self.trades.append({date: i, type: SELL, price: current_price, shares: self.position, proceeds: proceeds}) print(f{i.date()}: 賣出 {self.position} 股價格 {current_price:.2f}) self.position 0 self.entry_price None # 清空買入價 # ... 平倉邏輯同父類6.3 進階學習路線建議完成這個基礎項目后你可以按以下路徑深入學習專業(yè)回測框架掌握backtrader或zipline它們提供了事件驅(qū)動、多資產(chǎn)、復雜訂單類型等專業(yè)功能。探索更多策略類型均值回歸策略如布林帶收縮擴張、RSI超買超賣。套利策略如ETF與一籃子股票之間的價差套利。統(tǒng)計套利尋找一對相關性高的股票做多弱勢股做空強勢股。因子研究與機器學習使用alphalens,zipline進行多因子分析或嘗試用scikit-learn、LightGBM構(gòu)建預測模型。實盤對接了解券商提供的量化交易API如華泰、國信、中泰等但務必先在模擬環(huán)境中充分測試。關注風險與合規(guī)深刻理解市場風險、模型風險、技術(shù)風險并確保所有操作符合相關法律法規(guī)。量化交易是一個將金融理論、統(tǒng)計學、編程技術(shù)和風險管理深度融合的領域。這個教程為你搭建了一個從數(shù)據(jù)到策略再到回測的完整腳手架但請記住在實盤中投入真金白銀之前必須在更長的時間范圍和更多的標的上進行嚴格的驗證并始終對市場保持敬畏。 Python量化交易實戰(zhàn)：從數(shù)據(jù)獲取到雙均線策略回測的完整閉環(huán)
最近在嘗試將量化交易從理論轉(zhuǎn)向?qū)嵺`時很多開發(fā)者朋友都遇到過類似的困境網(wǎng)上資料要么過于學術(shù)化要么就是零散的代碼片段真正能跑通、能理解、能復現(xiàn)的完整教程少之又少。自己摸索時常常在數(shù)據(jù)獲取、策略回測、實盤對接這些環(huán)節(jié)反復踩坑浪費大量時間。本文旨在整合一套從零到一的 Python 量化交易實戰(zhàn)閉環(huán)。我們將手把手搭建一個完整的量化分析環(huán)境編寫并回測一個簡單的雙均線策略最終探討策略優(yōu)化與風險管理的思路。無論你是對量化感興趣的學生還是希望將策略自動化的開發(fā)者都能從這套保姆級流程中找到可復用的代碼和清晰的路徑避開那些常見的“彎路”。1. 量化交易核心概念與準備工作在開始寫代碼之前我們需要統(tǒng)一對幾個核心概念的理解并準備好相應的“武器庫”。1.1 什么是量化交易量化交易Quantitative Trading的核心是利用數(shù)學模型、統(tǒng)計分析和計算機技術(shù)從海量歷史數(shù)據(jù)中尋找能夠預測未來價格走勢的規(guī)律即“因子”或“Alpha”并據(jù)此構(gòu)建可自動執(zhí)行的交易策略。它與傳統(tǒng)主觀交易的最大區(qū)別在于系統(tǒng)性決策基于明確的規(guī)則和模型排除了情緒干擾?？苫厮莶呗栽跉v史數(shù)據(jù)上的表現(xiàn)回測可以量化評估。自動化策略生成交易信號后可通過程序自動執(zhí)行訂單。一個完整的量化交易流程通常包括數(shù)據(jù)獲取、策略研究、回測驗證、風險控制、實盤交易和績效分析。1.2 環(huán)境準備與工具選型工欲善其事必先利其器。我們選擇 Python 作為實現(xiàn)語言因為它擁有極其豐富的金融數(shù)據(jù)分析庫和社區(qū)生態(tài)。1. 基礎環(huán)境操作系統(tǒng)Windows 10/11, macOS 或 Linux (如 Ubuntu) 均可。本文示例在 Windows 11 下完成。Python 版本推薦使用 Python 3.8 或 3.9這兩個版本與多數(shù)金融庫的兼容性最好。避免使用 Python 3.10 的早期版本可能遇到某些庫的安裝問題。包管理工具使用pip即可。強烈建議使用虛擬環(huán)境如venv或conda來隔離項目依賴。2. 核心工具庫介紹我們將使用以下庫它們構(gòu)成了量化分析的基石pandasnumpy數(shù)據(jù)處理和科學計算的絕對核心用于數(shù)據(jù)清洗、轉(zhuǎn)換和計算。akshare一個免費、開源的 Python 財經(jīng)數(shù)據(jù)接口庫。它提供了國內(nèi)股票、基金、期貨、期權(quán)等大量金融數(shù)據(jù)替代了部分付費數(shù)據(jù)源非常適合學習和個人研究。mplfinance基于 Matplotlib 的金融數(shù)據(jù)可視化庫可以輕松繪制K線圖、均線等。backtrader或zipline專業(yè)的回測框架。backtrader功能強大、靈活社區(qū)活躍zipline是 Quantopian 開源的回測引擎結(jié)構(gòu)嚴謹。本文為求簡單直觀將手動實現(xiàn)回測邏輯以便理解每一步原理。在實際復雜策略中強烈推薦使用這些成熟框架。3. 安裝依賴在你的項目虛擬環(huán)境中執(zhí)行以下命令安裝所需庫pip install pandas numpy akshare mplfinance如果網(wǎng)絡不佳可以使用國內(nèi)鏡像源加速pip install pandas numpy akshare mplfinance -i https://pypi.tuna.tsinghua.edu.cn/simple2. 第一步獲取與處理金融數(shù)據(jù)沒有數(shù)據(jù)量化就是無源之水。可靠、干凈的數(shù)據(jù)是策略成功的首要前提。2.1 使用 AKShare 獲取股票歷史數(shù)據(jù)AKShare 接口非常豐富我們以獲取“貴州茅臺”股票代碼 600519.SH的日線數(shù)據(jù)為例。# 文件data_fetcher.py import akshare as ak import pandas as pd import warnings warnings.filterwarnings(ignore) # 忽略一些不必要的警告 def fetch_stock_data(symbol, start_date, end_date): 獲取指定股票代碼的日線行情數(shù)據(jù) :param symbol: 股票代碼如 600519 (AKShare內(nèi)部會自動補全市場) :param start_date: 開始日期格式 YYYYMMDD :param end_date: 結(jié)束日期格式 YYYYMMDD :return: 包含OHLCV等數(shù)據(jù)的 pandas DataFrame # 使用 ak.stock_zh_a_hist 接口注意參數(shù)名 df ak.stock_zh_a_hist(symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq) # 查看原始數(shù)據(jù)列名 print(原始數(shù)據(jù)列名, df.columns.tolist()) # 重命名列使其更符合通用標準 df.rename(columns{ 日期: date, 開盤: open, 收盤: close, 最高: high, 最低: low, 成交量: volume, 成交額: amount, 振幅: amplitude, 漲跌幅: pct_chg, 漲跌額: change, 換手率: turnover }, inplaceTrue) # 將日期列設置為索引并轉(zhuǎn)換為datetime格式便于后續(xù)時間序列分析 df[date] pd.to_datetime(df[date]) df.set_index(date, inplaceTrue) # 按日期升序排列 df.sort_index(inplaceTrue) return df if __name__ __main__: # 示例獲取貴州茅臺2023年全年的數(shù)據(jù) stock_code 600519 # 貴州茅臺 start 20230101 end 20231231 maotai_data fetch_stock_data(stock_code, start, end) print(f數(shù)據(jù)形狀{maotai_data.shape}) print(maotai_data.head()) # 查看前5行 print(maotai_data.tail()) # 查看后5行運行這段代碼你將得到一個DataFrame索引是日期包含了開盤價、收盤價、最高價、最低價、成交量等關鍵信息。adjustqfq參數(shù)表示獲取前復權(quán)數(shù)據(jù)這能保證歷史價格的可比性是回測的標準做法。2.2 數(shù)據(jù)清洗與特征工程拿到的原始數(shù)據(jù)可能存在缺失值、異常值或需要計算衍生指標如技術(shù)指標。# 文件data_processor.py import pandas as pd import numpy as np def calculate_technical_indicators(df): 計算常用的技術(shù)指標 :param df: 包含OHLCV數(shù)據(jù)的DataFrame :return: 添加了技術(shù)指標列的DataFrame df df.copy() # 1. 簡單移動平均線 (SMA) df[SMA_10] df[close].rolling(window10, min_periods1).mean() df[SMA_30] df[close].rolling(window30, min_periods1).mean() # 2. 指數(shù)移動平均線 (EMA) - 對近期價格賦予更高權(quán)重 df[EMA_12] df[close].ewm(span12, adjustFalse).mean() df[EMA_26] df[close].ewm(span26, adjustFalse).mean() # 3. 計算MACD指標 df[MACD] df[EMA_12] - df[EMA_26] df[MACD_Signal] df[MACD].ewm(span9, adjustFalse).mean() df[MACD_Histogram] df[MACD] - df[MACD_Signal] # 4. 布林帶 (Bollinger Bands) df[BB_Middle] df[close].rolling(window20).mean() bb_std df[close].rolling(window20).std() df[BB_Upper] df[BB_Middle] 2 * bb_std df[BB_Lower] df[BB_Middle] - 2 * bb_std # 5. 相對強弱指數(shù) (RSI) delta df[close].diff() gain (delta.where(delta 0, 0)).rolling(window14).mean() loss (-delta.where(delta 0, 0)).rolling(window14).mean() rs gain / loss df[RSI] 100 - (100 / (1 rs)) # 處理計算產(chǎn)生的初始NaN值例如前29天沒有30日均線 # 這里簡單用前向填充但更嚴謹?shù)淖龇ㄊ翘蕹蚴褂闷渌椒?df.fillna(methodffill, inplaceTrue) df.dropna(inplaceTrue) # 刪除仍為NaN的行 return df if __name__ __main__: # 假設 maotai_data 是上一節(jié)獲取的數(shù)據(jù) from data_fetcher import fetch_stock_data data fetch_stock_data(600519, 20230101, 20231231) processed_data calculate_technical_indicators(data) print(processed_data[[close, SMA_10, SMA_30, RSI]].head(35)) # 查看包含指標的數(shù)據(jù)這個函數(shù)為數(shù)據(jù)添加了多個常用技術(shù)指標。在量化策略中這些指標常被用作產(chǎn)生交易信號的“因子”。3. 第二步策略構(gòu)思與回測實現(xiàn)策略是量化的靈魂。我們從一個最經(jīng)典的趨勢跟蹤策略——雙均線交叉策略開始。3.1 策略邏輯雙均線交叉策略規(guī)則非常簡單買入信號金叉當短期均線如10日SMA從下向上穿越長期均線如30日SMA時買入股票。賣出信號死叉當短期均線從上向下穿越長期均線時賣出持有的股票。持倉其余時間持有股票如果已買入或空倉。這個策略的邏輯是捕捉趨勢的形成金叉意味著上漲趨勢可能開始死叉意味著下跌趨勢可能開始。3.2 手動實現(xiàn)回測引擎為了徹底理解回測的每個細節(jié)我們不依賴框架自己實現(xiàn)一個簡單的回測系統(tǒng)。# 文件backtest_engine.py import pandas as pd import numpy as np class SimpleBacktester: def __init__(self, data, initial_capital100000.0): 初始化回測器 :param data: 包含價格和指標數(shù)據(jù)的DataFrame :param initial_capital: 初始資金 self.data data.copy() self.initial_capital initial_capital self.current_capital initial_capital self.position 0 # 當前持有股票數(shù)量 self.trades [] # 記錄所有交易 self.equity_curve [] # 記錄每日資產(chǎn)總值 def generate_signals(self): 根據(jù)雙均線交叉生成交易信號 # 確保數(shù)據(jù)中有我們需要的均線 if SMA_10 not in self.data.columns or SMA_30 not in self.data.columns: raise ValueError(數(shù)據(jù)中缺少SMA_10或SMA_30列請先計算技術(shù)指標。) # 生成信號1 代表買入 -1 代表賣出 0 代表無操作 self.data[signal] 0 # 金叉SMA_10上穿SMA_30 golden_cross (self.data[SMA_10] self.data[SMA_30]) (self.data[SMA_10].shift(1) self.data[SMA_30].shift(1)) # 死叉SMA_10下穿SMA_30 death_cross (self.data[SMA_10] self.data[SMA_30]) (self.data[SMA_10].shift(1) self.data[SMA_30].shift(1)) self.data.loc[golden_cross, signal] 1 self.data.loc[death_cross, signal] -1 # 為了簡化我們假設信號產(chǎn)生在當天收盤時第二天以開盤價執(zhí)行 self.data[signal] self.data[signal].shift(1) self.data[signal].fillna(0, inplaceTrue) def run_backtest(self): 執(zhí)行回測 print(開始回測...) self.generate_signals() for i, row in self.data.iterrows(): current_price row[close] signal row[signal] # 記錄當前總資產(chǎn)現(xiàn)金 持倉市值 current_equity self.current_capital self.position * current_price self.equity_curve.append({ date: i, equity: current_equity, price: current_price, position: self.position }) # 執(zhí)行交易信號 if signal 1 and self.position 0: # 買入信號且空倉 # 計算可買數(shù)量假設可以買任意股這里取整 # 實際中應考慮手續(xù)費、最小交易單位A股是100股等 shares_to_buy int(self.current_capital // current_price) if shares_to_buy 0: cost shares_to_buy * current_price self.current_capital - cost self.position shares_to_buy self.trades.append({ date: i, type: BUY, price: current_price, shares: shares_to_buy, cost: cost, cash_after: self.current_capital }) print(f{i.date()}: 買入 {shares_to_buy} 股價格 {current_price:.2f}) elif signal -1 and self.position 0: # 賣出信號且持有 proceeds self.position * current_price self.current_capital proceeds self.trades.append({ date: i, type: SELL, price: current_price, shares: self.position, proceeds: proceeds, cash_after: self.current_capital }) print(f{i.date()}: 賣出 {self.position} 股價格 {current_price:.2f}) self.position 0 # 回測結(jié)束平倉如果最后還持有 if self.position 0: last_price self.data[close].iloc[-1] proceeds self.position * last_price self.current_capital proceeds self.trades.append({ date: self.data.index[-1], type: SELL, price: last_price, shares: self.position, proceeds: proceeds, cash_after: self.current_capital }) print(f回測結(jié)束平倉 {self.position} 股價格 {last_price:.2f}) self.position 0 print(回測完成。) def get_results(self): 計算并返回回測績效指標 if not self.equity_curve: raise ValueError(請先運行 run_backtest()) equity_df pd.DataFrame(self.equity_curve) equity_df.set_index(date, inplaceTrue) # 計算累計收益率 initial_equity self.initial_capital final_equity equity_df[equity].iloc[-1] total_return (final_equity - initial_equity) / initial_equity # 計算年化收益率假設一年約242個交易日 days (equity_df.index[-1] - equity_df.index[0]).days years days / 365.0 annualized_return (1 total_return) ** (1 / years) - 1 if years 0 else 0 # 計算最大回撤最大虧損幅度 equity_series equity_df[equity] cumulative_max equity_series.expanding().max() drawdown (equity_series - cumulative_max) / cumulative_max max_drawdown drawdown.min() # 夏普比率簡化版假設無風險利率為0 daily_returns equity_df[equity].pct_change().dropna() if daily_returns.std() ! 0: sharpe_ratio np.sqrt(242) * (daily_returns.mean() / daily_returns.std()) # 年化夏普 else: sharpe_ratio 0 results { 初始資金: initial_equity, 最終資產(chǎn): final_equity, 總收益率: total_return, 年化收益率: annualized_return, 最大回撤: max_drawdown, 夏普比率: sharpe_ratio, 交易次數(shù): len(self.trades) } return results, equity_df, pd.DataFrame(self.trades) if __name__ __main__: # 整合之前的步驟 from data_fetcher import fetch_stock_data from data_processor import calculate_technical_indicators # 1. 獲取并處理數(shù)據(jù) raw_data fetch_stock_data(600519, 20220101, 20231231) # 獲取兩年數(shù)據(jù) processed_data calculate_technical_indicators(raw_data) # 2. 創(chuàng)建回測實例并運行 backtester SimpleBacktester(processed_data, initial_capital100000) backtester.run_backtest() # 3. 獲取結(jié)果 results, equity_curve, trades backtester.get_results() print(\n *50) print(回測績效總結(jié):) print(*50) for key, value in results.items(): if isinstance(value, float): print(f{key}: {value:.4%} if 率 in key or 撤 in key else f{key}: {value:.4f}) else: print(f{key}: {value}) print(f\n交易記錄 (共{len(trades)}筆):) if not trades.empty: print(trades[[date, type, price, shares]].head())4. 第三步結(jié)果可視化與初步分析數(shù)字結(jié)果不夠直觀我們需要圖表來評估策略表現(xiàn)。4.1 繪制資產(chǎn)曲線與買賣點# 文件visualization.py import matplotlib.pyplot as plt import mplfinance as mpf import pandas as pd def plot_equity_curve(equity_df, trades_df, data): 繪制資產(chǎn)曲線和買賣點 :param equity_df: 回測引擎輸出的equity_curve DataFrame :param trades_df: 回測引擎輸出的trades DataFrame :param data: 原始行情數(shù)據(jù)DataFrame fig, axes plt.subplots(3, 1, figsize(14, 10), gridspec_kw{height_ratios: [3, 1, 1]}) # 子圖1價格、均線與買賣點 ax1 axes[0] ax1.plot(data.index, data[close], label收盤價, linewidth1, alpha0.7) ax1.plot(data.index, data[SMA_10], label10日均線, linewidth1, alpha0.9) ax1.plot(data.index, data[SMA_30], label30日均線, linewidth1, alpha0.9) # 標記買賣點 buy_trades trades_df[trades_df[type] BUY] sell_trades trades_df[trades_df[type] SELL] ax1.scatter(buy_trades[date], data.loc[buy_trades[date], close], colorred, marker^, s100, label買入, zorder5) ax1.scatter(sell_trades[date], data.loc[sell_trades[date], close], colorgreen, markerv, s100, label賣出, zorder5) ax1.set_title(雙均線策略 - 價格與交易信號) ax1.set_ylabel(價格 (元)) ax1.legend() ax1.grid(True, alpha0.3) # 子圖2資產(chǎn)曲線 ax2 axes[1] ax2.plot(equity_df.index, equity_df[equity], label總資產(chǎn), colorblue, linewidth2) ax2.fill_between(equity_df.index, equity_df[equity], equity_df[equity].min(), alpha0.3) ax2.set_title(策略資產(chǎn)曲線) ax2.set_ylabel(資產(chǎn) (元)) ax2.legend() ax2.grid(True, alpha0.3) # 子圖3持倉情況 ax3 axes[2] ax3.bar(equity_df.index, equity_df[position], label持倉股數(shù), colororange, alpha0.6) ax3.set_title(持倉變化) ax3.set_xlabel(日期) ax3.set_ylabel(持倉 (股)) ax3.legend() ax3.grid(True, alpha0.3) plt.tight_layout() plt.show() def plot_kline_with_indicators(data): 使用mplfinance繪制K線圖及均線 # 準備mplfinance需要的格式 plot_data data[[open, high, low, close, volume]].copy() plot_data.columns [Open, High, Low, Close, Volume] # 列名需大寫 # 添加均線作為附加繪圖 apds [ mpf.make_addplot(data[SMA_10], colororange, width0.7), mpf.make_addplot(data[SMA_30], colorblue, width0.7), ] mpf.plot(plot_data, typecandle, stylecharles, titleK線圖與雙均線, ylabel價格, volumeTrue, addplotapds, figratio(12, 8), figscale1.2) if __name__ __main__: # 假設已有回測結(jié)果 from backtest_engine import SimpleBacktester from data_fetcher import fetch_stock_data from data_processor import calculate_technical_indicators data calculate_technical_indicators(fetch_stock_data(600519, 20230101, 20231231)) bt SimpleBacktester(data, 100000) bt.run_backtest() results, equity_df, trades_df bt.get_results() # 繪制圖表 plot_equity_curve(equity_df, trades_df, data) plot_kline_with_indicators(data)運行可視化代碼你將看到三張圖第一張是價格走勢、均線以及買賣點標記第二張是你的資產(chǎn)總值變化曲線第三張是持倉數(shù)量變化。這能幫你直觀判斷策略是否跟上了趨勢以及買賣時機是否合理。5. 常見問題與策略缺陷分析第一個能跑起來的策略往往充滿問題。以下是雙均線策略及我們簡易回測中常見的坑點。5.1 回測中的常見陷阱與解決方案問題現(xiàn)象常見原因解決思路與方案回測結(jié)果過于完美年化收益驚人1.未來函數(shù)使用了當時還未產(chǎn)生的數(shù)據(jù)如用當天收盤價計算信號并當天交易。2.忽略交易成本未計算傭金、印花稅、滑點。3.過擬合策略參數(shù)恰好完美匹配了某段歷史數(shù)據(jù)。1.確保信號滯后信號基于t-1日的數(shù)據(jù)交易在t日執(zhí)行如我們代碼中的shift(1)。2.加入成本模型在買賣時扣除固定比例或固定金額的費用。3.樣本外測試用另一段未參與參數(shù)優(yōu)化的歷史數(shù)據(jù)驗證策略。實盤表現(xiàn)與回測差異巨大1.流動性假設回測假設可以立即以收盤價成交實盤可能無法成交或成交價更差。2.數(shù)據(jù)質(zhì)量回測數(shù)據(jù)是清洗過的復權(quán)數(shù)據(jù)實盤數(shù)據(jù)是實時、原始的。3.市場狀態(tài)變化策略有效的市場環(huán)境如強趨勢市可能已改變。1.引入滑點模型假設成交價比信號價差一個固定或隨機比例。2.使用更精細的數(shù)據(jù)回測時使用Tick級或分鐘級數(shù)據(jù)。3.考慮市場機制加入漲跌停板、T1等A股特殊規(guī)則。交易次數(shù)過多或過少1.參數(shù)敏感均線周期設置太短導致頻繁交叉噪音交易太長則信號遲鈍。2.數(shù)據(jù)波動性標的物本身波動性大或小。1.參數(shù)優(yōu)化與穩(wěn)健性檢驗測試不同參數(shù)組合觀察績效是否穩(wěn)定。2.添加過濾器例如要求金叉時價格需同時上穿長期均線一定幅度或結(jié)合波動率指標。最大回撤過大策略在震蕩市中反復“高買低賣”虧損累積。趨勢策略在無趨勢市場中的通病。1.加入止損機制單筆虧損達到一定比例或金額時強制平倉。2.結(jié)合市場擇時增加一個趨勢過濾器例如在大盤指數(shù)處于200日均線之上時才運行該策略。5.2 我們的簡易回測引擎的局限性我們的SimpleBacktester為了教學而極度簡化認識到這些局限是進步的第一步單標的、全倉進出現(xiàn)實中需要管理多只股票和資金分配。無風險控制沒有止損止盈、倉位控制。簡化交易邏輯假設無限流動性、無手續(xù)費、無最小交易單位限制A股需100股整數(shù)倍。未考慮分紅送股雖然用了前復權(quán)數(shù)據(jù)但策略本身未處理除權(quán)除息日的特殊邏輯。6. 策略優(yōu)化與進階方向一個基本的策略跑通后量化工作才真正開始。以下是幾個關鍵的優(yōu)化和進階方向。6.1 參數(shù)優(yōu)化與穩(wěn)健性檢驗不要只尋找歷史數(shù)據(jù)上表現(xiàn)最好的參數(shù)這大概率是過擬合。應該尋找在多個時間段、多只股票上都表現(xiàn)穩(wěn)定的參數(shù)。# 文件parameter_optimization.py import itertools from backtest_engine import SimpleBacktester from data_fetcher import fetch_stock_data from data_processor import calculate_technical_indicators def grid_search_parameters(stock_code, start, end, short_range, long_range): 網(wǎng)格搜索雙均線的最佳參數(shù)組合 :param short_range: 短期均線周期范圍如 range(5, 21, 5) :param long_range: 長期均線周期范圍如 range(20, 61, 10) results_list [] data_raw fetch_stock_data(stock_code, start, end) for short_period, long_period in itertools.product(short_range, long_range): if short_period long_period: continue # 短期必須小于長期 print(f測試參數(shù): SMA({short_period}, {long_period})) # 重新計算指標 data data_raw.copy() data[fSMA_{short_period}] data[close].rolling(windowshort_period).mean() data[fSMA_{long_period}] data[close].rolling(windowlong_period).mean() data.fillna(methodffill, inplaceTrue) data.dropna(inplaceTrue) # 運行回測 bt SimpleBacktester(data, 100000) bt.run_backtest() perf, _, _ bt.get_results() results_list.append({ short_period: short_period, long_period: long_period, total_return: perf[總收益率], max_drawdown: perf[最大回撤], sharpe_ratio: perf[夏普比率], trade_count: perf[交易次數(shù)] }) results_df pd.DataFrame(results_list) # 按夏普比率排序綜合考慮收益與風險 results_df.sort_values(sharpe_ratio, ascendingFalse, inplaceTrue) return results_df if __name__ __main__: # 測試不同參數(shù)組合 best_params_df grid_search_parameters(600519, 20200101, 20231231, range(5, 21, 5), range(30, 101, 15)) print(\n參數(shù)優(yōu)化結(jié)果按夏普比率降序:) print(best_params_df.head(10))6.2 引入風險管理模塊一個沒有風險控制的策略是危險的。至少應加入止損。# 文件risk_management.py class BacktesterWithRisk(SimpleBacktester): 增加了風險管理的回測器 def __init__(self, data, initial_capital100000.0, stop_loss_pct0.05): super().__init__(data, initial_capital) self.stop_loss_pct stop_loss_pct # 止損比例如5% self.entry_price None # 記錄買入成本價 def run_backtest(self): self.generate_signals() for i, row in self.data.iterrows(): current_price row[close] signal row[signal] # --- 止損檢查 --- if self.position 0 and self.entry_price is not None: # 計算從買入后的最大虧損幅度 loss_pct (current_price - self.entry_price) / self.entry_price if loss_pct -self.stop_loss_pct: # 觸發(fā)止損賣出所有持倉 proceeds self.position * current_price self.current_capital proceeds self.trades.append({ date: i, type: SELL_STOPLOSS, price: current_price, shares: self.position, proceeds: proceeds, cash_after: self.current_capital }) print(f{i.date()}: 止損觸發(fā)賣出 {self.position} 股價格 {current_price:.2f}虧損 {loss_pct:.2%}) self.position 0 self.entry_price None # 止損后跳過本日其他信號防止立即反向開倉 continue # 原有交易邏輯... current_equity self.current_capital self.position * current_price self.equity_curve.append({date: i, equity: current_equity, price: current_price, position: self.position}) if signal 1 and self.position 0: shares_to_buy int(self.current_capital // current_price) if shares_to_buy 0: cost shares_to_buy * current_price self.current_capital - cost self.position shares_to_buy self.entry_price current_price # 記錄買入價 self.trades.append({date: i, type: BUY, price: current_price, shares: shares_to_buy, cost: cost}) print(f{i.date()}: 買入 {shares_to_buy} 股價格 {current_price:.2f}) elif signal -1 and self.position 0: proceeds self.position * current_price self.current_capital proceeds self.trades.append({date: i, type: SELL, price: current_price, shares: self.position, proceeds: proceeds}) print(f{i.date()}: 賣出 {self.position} 股價格 {current_price:.2f}) self.position 0 self.entry_price None # 清空買入價 # ... 平倉邏輯同父類6.3 進階學習路線建議完成這個基礎項目后你可以按以下路徑深入學習專業(yè)回測框架掌握backtrader或zipline它們提供了事件驅(qū)動、多資產(chǎn)、復雜訂單類型等專業(yè)功能。探索更多策略類型均值回歸策略如布林帶收縮擴張、RSI超買超賣。套利策略如ETF與一籃子股票之間的價差套利。統(tǒng)計套利尋找一對相關性高的股票做多弱勢股做空強勢股。因子研究與機器學習使用alphalens,zipline進行多因子分析或嘗試用scikit-learn、LightGBM構(gòu)建預測模型。實盤對接了解券商提供的量化交易API如華泰、國信、中泰等但務必先在模擬環(huán)境中充分測試。關注風險與合規(guī)深刻理解市場風險、模型風險、技術(shù)風險并確保所有操作符合相關法律法規(guī)。量化交易是一個將金融理論、統(tǒng)計學、編程技術(shù)和風險管理深度融合的領域。這個教程為你搭建了一個從數(shù)據(jù)到策略再到回測的完整腳手架但請記住在實盤中投入真金白銀之前必須在更長的時間范圍和更多的標的上進行嚴格的驗證并始終對市場保持敬畏。