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