用 Python 實現量化交易策略回測
作者：玩轉量化
今天要給大家分享的例子，就展示了如何基于Python中常用的numpy、pandas等常用數據分析處理框架，針對目標個股簡單實現「均線策略回測」。
Python憑借其在數據科學領域積累的豐富生態，已然成為專業「量化分析」中必不可少的技術手段。今天要給大家分享的例子，就展示了如何基于Python中常用的numpy、pandas等常用數據分析處理框架，針對目標個股簡單實現「均線策略回測」。
1. 相關庫的導入
分析過程需要用到的庫如下，其中numpy、pandas等庫用于實現分析過程的「數據處理」及「運算」，xtquant用于快捷「獲取」股票歷史行情數據，matplotlib則用于對策略過程及效果進行「可視化」：
import numpy as np
import pandas as pd
from xtquant import xtdata # qmt行情數據模塊
from functools import reduce
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
from matplotlib.ticker import MaxNLocator2. 獲取目標個股歷史行情數據
在導入相關庫后，我們首先需要獲取目標個股的「歷史行情數據」，以601127.SH賽力斯為例，我們基于xtquant中的行情數據模塊，直接下載并提取目標個股「近5年」的日線行情數據(xtquant調用股票行情數據使用需配合本機QMT程序)：
# 以601127.SH賽力斯為例
target_stock = '601127.SH'
# 獲取近5年歷史日線行情數據
start_time = (datetime.now() - timedelta(days=365 * 5)).strftime('%Y%m%d')
# 下載數據
xtdata.download_history_data(
target_stock,
period='1d',
start_time=start_time
)
# 提取數據
raw_history = xtdata.get_market_data(
stock_list=[target_stock],
period='1d',
start_time=start_time,
field_list=['open', 'high', 'low', 'close', 'volume']
)3. 歷史行情數據清洗轉換
為了進行下一步的策略回測模擬，我們需要對先前下載提取的日線歷史行情數據進行「清洗轉換」，通過下面的代碼，即可將上一步的原始數據轉換為標準的「數據框」格式：
# 轉換為回測所需標準數據框格式
history_df = (
reduce(
lambda left, right: left.join(right),
[
value
.T
.rename(
columns={
target_stock: key
}
)
for key, value in raw_history.items()
]
)
.reset_index(drop=False)
.rename(columns={'index': 'datetime'})
)
history_df.head()4. 回測模擬相關參數設置
接下來我們需要定義策略模擬相關的初始資金、交易傭金率、交易最低傭金等基本參數：
# 回測模擬相關參數設置
initial_cash = 100000 # 初始資金
commission_rate = 0.0001 # 交易傭金率
min_commission = 5 # 交易最低傭金
short_window = 15 # 短期均線窗口大小
long_window = 60 # 長期均線窗口大小5. 交易信號計算
按照上一步給定的參數，首先計算出短期、長期均線值：
# 計算短期均線
history_df['short_mavg'] = history_df['close'].rolling(window=short_window, min_periods=1).mean()
# 計算長期均線
history_df['long_mavg'] = history_df['close'].rolling(window=long_window, min_periods=1).mean()接著按照短期均線超過長期均線買入，反之賣出的簡單均線策略，計算出不同的「交易信號」點：
# 初始化交易信號字段
history_df['signal'] = 0
# 將所有短期均線突破長期均線的交易日，交易信號標記為1
history_df.loc[short_window:, 'signal'] = np.where(
history_df.loc[short_window:, 'short_mavg'] > history_df.loc[short_window:, 'long_mavg'],
1,
0
)
# 通過差分運算，輔助計算交易時機
history_df['positions'] = history_df['signal'].diff()6. 基于交易信號模擬交易過程
接著我們就可以在前面計算結果的基礎上，執行模擬交易過程，并妥善記錄中間過程「賬戶值變化」情況：
cash = initial_cash # 資金
total_shares = 0 # 持倉股數
portfolio_value = [] # 記錄每日賬戶總值
total_commission = 0 # 記錄累計傭金支出
# 執行交易模擬過程
for row in history_df.itertuples():
# 當出現買入信號時
if row.positions == 1:
num_shares = cash // row.close # 計算本次可買入的股數
total_commission += max(min_commission, commission_rate * num_shares * row.close) # 累加本次交易傭金
cash -= num_shares * row.close # 更新最新資金量
total_shares += num_shares # 更新最新持股數量
# 當出現賣出信號，且存在有效持倉時，假設全部賣出
elif row.positions == -1 and total_shares > 0:
total_commission += max(min_commission, commission_rate * total_shares * row.close) # 計算本次交易傭金
cash += total_shares * row.close # 更新最新資金量
total_shares = 0 # 更新最新持股數量
# 計算每日的賬戶總值
portfolio_value.append(cash + total_shares * row.close)
# 添加賬戶總值字段
history_df['portfolio_value'] = portfolio_value7. 回測結果可視化
最后，我們將整個回測過程，以及最終的賬戶結果值、傭金成本等信息整合到一張圖中展示：
# 設置中文字體
plt.rcParams['font.family'] = ['SimHei']
# 設置負號顯示
plt.rcParams['axes.unicode_minus'] = False
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8), sharex=True)
# 繪制上圖 - 交易過程
ax1.plot(history_df['datetime'], history_df['close'], label='收盤價', linewidth=1)
ax1.plot(history_df['datetime'], history_df['short_mavg'], label='短期均線', linewidth=1)
ax1.plot(history_df['datetime'], history_df['long_mavg'], label='長期均線', linewidth=1)
ax1.plot(history_df[history_df['positions'] == 1]['datetime'], history_df['short_mavg'][history_df['positions'] == 1], '^', markersize=6, color='g', label='買入')
ax1.plot(history_df[history_df['positions'] == -1]['datetime'], history_df['short_mavg'][history_df['positions'] == -1], 'v', markersize=6, color='r', label='賣出')
ax1.legend()
ax1.set_title('模擬交易過程')
# 繪制下圖 - 賬戶值波動變化
ax2.plot(history_df['datetime'], history_df['portfolio_value'], label='賬戶總值變化')
ax2.legend()
ax2.set_title(f'賬戶初始資金：{initial_cash} 回測結束賬戶值：{round(portfolio_value[-1])} 傭金成本：{round(total_commission)}')
ax2.set_xlabel('日期')
# 設置共享的x軸刻度和標簽角度
ax2.xaxis.set_major_locator(MaxNLocator(nbins=20)) # 設置刻度數量
plt.setp(ax2.get_xticklabels(), rotation=45, ha='right') # 設置標簽角度和對齊方式
plt.tight_layout()
# 導出圖片
plt.savefig('回測結果可視化.png', dpi=300)
plt.show()
責任編輯：趙寧寧
來源：
Python大數據分析
