用 Python 實(shí)現(xiàn)量化交易策略回測(cè)

作者：玩轉(zhuǎn)量化 2024-12-23 07:20:00

今天要給大家分享的例子，就展示了如何基于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)庫的導(dǎo)入

分析過程需要用到的庫如下，其中numpy、pandas等庫用于實(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 MaxNLocator

2. 獲取目標(biāo)個(gè)股歷史行情數(shù)據(jù)

在導(dǎo)入相關(guān)庫后，我們首先需要獲取目標(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 # 長期均線窗口大小

5. 交易信號(hào)計(jì)算

按照上一步給定的參數(shù)，首先計(jì)算出短期、長期均線值：

# 計(jì)算短期均線
history_df['short_mavg'] = history_df['close'].rolling(window=short_window, min_periods=1).mean()
# 計(jì)算長期均線
history_df['long_mavg'] = history_df['close'].rolling(window=long_window, min_periods=1).mean()

接著按照短期均線超過長期均線買入，反之賣出的簡(jiǎn)單均線策略，計(jì)算出不同的「交易信號(hào)」點(diǎn)：

# 初始化交易信號(hào)字段
history_df['signal'] = 0
# 將所有短期均線突破長期均線的交易日，交易信號(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  # 持倉股數(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)，且存在有效持倉時(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_value

7. 回測(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='長期均線', 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ù)分析