Python 建構事件驅動回測框架與交易策略

– 10 分鐘閱讀

事件驅動回測框架是金融領域中常用的工具，用於評估交易策略的有效性。Python 提供了豐富的函式庫和工具，方便我們建構和執行這類框架。本文將逐步介紹如何使用 Python 建構一個事件驅動的回測框架，並示範如何實作交易策略、計算績效指標以及進行策略評估。首先，我們需要定義一個回測基礎類別，包含初始化回測、取得和預處理資料、執行交易策略、評估交易績效以及輸出結果等基本功能。接著，我們可以根據不同的交易策略，例如均值迴歸策略、技術指標策略等，繼承這個基礎類別，並實作具體的交易邏輯。在交易模擬器中，核心方法包括取得日期和價格、列印餘額和淨資產、以及下單買賣等。這些方法用於管理交易過程和計算相關資料，例如模擬買賣過程、檢視餘額和淨資產等，以更好地模擬真實的交易場景。

圖表翻譯：

  flowchart TD
    A[初始化回測基礎類別] --> B[取得和預處理資料]
    B --> C[執行交易策略]
    C --> D[評估交易績效]
    D --> E[輸出結果]

內容解密：

以上程式碼實作了回測基礎類別的初始化、資料取得和預處理、資料視覺化和回測流程。這些功能是回測的基礎，為後續的交易策略執行和績效評估提供了資料支援。

交易模擬器核心方法

在交易模擬器中，幾個核心方法是用於管理交易過程和計算相關資料。這些方法包括取得日期和價格、列印餘額和淨資產、以及下單買賣等。

取得日期和價格

get_date_price 方法用於傳回給定條件下的日期和價格資訊。它接收一個條件 bar，並傳回對應的日期和價格。

def get_date_price(self, bar):
    ''' Return date and price for bar. '''
    date = str(self.data.index[bar])[:10]
    price = self.data.price.iloc[bar]
    return date, price

列印餘額和淨資產

print_balance 和 print_net_wealth 方法分別用於列印當前餘額和淨資產。這兩個方法都會先呼叫 get_date_price 方法來取得日期和價格資訊。

def print_balance(self, bar):
    ''' Print out current cash balance info. '''
    date, price = self.get_date_price(bar)
    print(f'{date} | current balance {self.amount:.2f}')

def print_net_wealth(self, bar):
    ''' Print out current cash balance info. '''
    date, price = self.get_date_price(bar)
    net_wealth = self.units * price + self.amount
    print(f'{date} | current net wealth {net_wealth:.2f}')

下單買賣

place_buy_order 和 place_sell_order 方法用於模擬買賣過程。這些方法會根據給定的條件下單，並計算相關的交易資料。

def place_buy_order(self, bar, units=None, amount=None):
    ''' Place a buy order. '''
    date, price = self.get_date_price(bar)
    # ... 其他買入邏輯

這些核心方法為交易模擬器提供了基本的功能，讓使用者可以模擬買賣過程、檢視餘額和淨資產等。透過這些方法，交易模擬器可以更好地模擬真實的交易場景，幫助使用者進行策略測試和最佳化。

交易策略基礎類別

在進行交易策略的回測時，需要有一個基礎類別來處理基本的交易邏輯。這個基礎類別需要包含買入和賣出訂單的處理、交易記錄的更新，以及交易後的資產狀態的計算。

買入訂單處理

買入訂單的處理需要計算買入的單位數量、更新交易記錄、以及計算交易後的資產狀態。以下是買入訂單處理的程式碼：

def place_buy_order(self, bar, units=None, amount=None):
    ''' 處理買入訂單 '''
    
    # 取得日期和價格
    date, price = self.get_date_price(bar)
    
    # 如果沒有指定單位數量，則根據金額計算單位數量
    if units is None:
        units = int(amount / price)
        
    # 更新交易記錄
    self.amount -= (units * price) * (1 + self.ptc) + self.ftc
    self.units += units
    self.trades += 1
    
    # 列印交易資訊
    if self.verbose:
        print(f'{date} | 買入 {units} 單位 at {price:.2f}')
        
    # 列印當前資產狀態
    self.print_balance(bar)
    self.print_net_wealth(bar)

賣出訂單處理

賣出訂單的處理需要計算賣出的單位數量、更新交易記錄、以及計算交易後的資產狀態。以下是賣出訂單處理的程式碼：

def place_sell_order(self, bar, units=None, amount=None):
    ''' 處理賣出訂單 '''
    
    # 取得日期和價格
    date, price = self.get_date_price(bar)
    
    # 如果沒有指定單位數量，則根據金額計算單位數量
    if units is None:
        units = int(amount / price)
        
    # 更新交易記錄
    self.amount += (units * price) * (1 - self.ptc) - self.ftc
    self.units -= units
    self.trades += 1
    
    # 列印交易資訊
    if self.verbose:
        print(f'{date} | 賣出 {units} 單位 at {price:.2f}')
        
    # 列印當前資產狀態
    self.print_balance(bar)
    self.print_net_wealth(bar)

交易結束處理

交易結束處理需要關閉交易位置、更新交易記錄、以及計算交易後的資產狀態。以下是交易結束處理的程式碼：

def close_out(self, bar):
    ''' 關閉交易位置 '''
    
    # 取得日期和價格
    date, price = self.get_date_price(bar)
    
    # 關閉交易位置
    # ... (省略實際關閉交易位置的程式碼)
    
    # 列印交易資訊
    if self.verbose:
        print(f'{date} | 關閉交易位置')
        
    # 列印當前資產狀態
    self.print_balance(bar)
    self.print_net_wealth(bar)

圖表翻譯：

圖表1：交易流程

  flowchart TD
    A[交易開始] --> B[買入訂單]
    B --> C[賣出訂單]
    C --> D[交易結束]
    D --> E[關閉交易位置]

圖表2：交易記錄更新

  flowchart TD
    A[交易記錄] --> B[買入訂單]
    B --> C[賣出訂單]
    C --> D[交易結束]
    D --> E[關閉交易位置]

圖表3：資產狀態更新

  flowchart TD
    A[資產狀態] --> B[買入訂單]
    B --> C[賣出訂單]
    C --> D[交易結束]
    D --> E[關閉交易位置]

建立根據事件的回測框架

在建立根據事件的回測框架時，需要考慮多個因素，包括交易成本、倉位管理、風險控制等。以下是一個簡單的例子，展示如何建立一個根據事件的回測框架。

交易記錄

交易記錄是回測框架的核心，負責記錄每筆交易的詳細資訊，包括交易日期、交易價格、交易量等。

class TradeRecord:
    def __init__(self, date, price, units):
        self.date = date
        self.price = price
        self.units = units

    def __str__(self):
        return f'{self.date} | {self.price:.2f} | {self.units} units'

回測基礎類別

回測基礎類別負責提供基本的回測功能，包括資料匯入、交易記錄、倉位管理等。

class BacktestBase:
    def __init__(self, symbol, start_date, end_date, initial_amount):
        self.symbol = symbol
        self.start_date = start_date
        self.end_date = end_date
        self.initial_amount = initial_amount
        self.amount = initial_amount
        self.units = 0
        self.trades = 0
        self.verbose = True

    def buy(self, date, price):
        self.units += 1
        self.amount -= price
        self.trades += 1
        if self.verbose:
            print(f'{date} | Buy {self.units} units at {price:.2f}')

    def sell(self, date, price):
        self.units -= 1
        self.amount += price
        self.trades += 1
        if self.verbose:
            print(f'{date} | Sell {self.units} units at {price:.2f}')

    def plot_data(self):
        # 繪製資料圖表
        pass

    def __str__(self):
        return f'BacktestBase({self.symbol}, {self.start_date}, {self.end_date}, {self.initial_amount})'

回測執行

回測執行是指根據特定的策略執行回測，以下是簡單的例子。

if __name__ == '__main__':
    bb = BacktestBase('AAPL.O', '2010-1-1', '2019-12-31', 10000)
    bb.plot_data()
    print(bb.data.info())
    print(bb.data.tail())

效果評估

效果評估是指評估回測的結果，包括計算淨績效、交易次數等。

perf = ((bb.amount - bb.initial_amount) / bb.initial_amount * 100)
print(f'Net Performance [%] {perf:.2f}')
print(f'Trades Executed [#] {bb.trades}')

圖表翻譯

以下是圖表的視覺化表示，使用Mermaid語法。

  flowchart TD
    A[資料匯入] --> B[交易記錄]
    B --> C[倉位管理]
    C --> D[風險控制]
    D --> E[效果評估]
    E --> F[圖表視覺化]

內容解密

以上的程式碼展示瞭如何建立根據事件的回測框架，包括交易記錄、回測基礎類別、回測執行、效果評估等。這些內容可以用於建立一個完整的回測系統，幫助使用者評估不同的交易策略。

事件驅動回測的類別構建

在事件驅動回測中，使用類別來組織和實作回測邏輯是一種有效的方法。這種方法允許開發人員建立一個基礎類別，包含回測過程中需要的標準功能，然後根據具體的交易策略建立子類別。

基礎回測類別

基礎回測類別（BacktestBase）提供了回測過程中需要的基本功能，例如資料載入、交易記錄和績效評估。這個類別可以被其他特定策略的類別繼承，以實作具體的回測邏輯。

class BacktestBase:
    def __init__(self, symbol, start_date, end_date):
        self.symbol = symbol
        self.start_date = start_date
        self.end_date = end_date
        self.data = self.load_data()

    def load_data(self):
        # 載入資料
        data = pd.read_csv(f'{self.symbol}.csv', index_col='Date', parse_dates=['Date'])
        return data

    def run_strategy(self):
        # 執行策略
        pass

只做多回測類別

某些投資者偏好或法規可能禁止做空，因此交易員或投資組合經理只允許進入做多頭寸或將資金存放在現金或類似的低風險資產中。BacktestLongOnly類別繼承自BacktestBase類別，實作了只做多的回測邏輯。

class BacktestLongOnly(BacktestBase):
    def __init__(self, symbol, start_date, end_date):
        super().__init__(symbol, start_date, end_date)

    def run_mean_reversion_strategy(self, SMA, threshold):
        # 實作均值迴歸策略的回測邏輯
        signals = pd.DataFrame(index=self.data.index)
        signals['signal'] = 0.0

        signals['signal'] = np.where(self.data['price'] < self.data['price'].rolling(window=SMA).mean() - threshold, 1.0, 0.0)
        signals['signal'] = np.where(self.data['price'] > self.data['price'].rolling(window=SMA).mean() + threshold, -1.0, signals['signal'])

        # 計算績效
        self.data['return'] = self.data['price'].pct_change()
        self.data['strategy_return'] = self.data['return'] * signals['signal'].shift(1)

        # 評估績效
        print('策略績效：')
        print(self.data['strategy_return'].cumsum().plot())

策略評估

最後，需要評估策略的績效。這可以透過計算策略的累積回報率、夏普比率等指標來實作。

def evaluate_strategy(self):
    # 計算績效指標
    strategy_return = self.data['strategy_return'].cumsum()
    sharpe_ratio = strategy_return.std() / strategy_return.mean()

    print('策略績效指標：')
    print(f'累積回報率：{strategy_return.iloc[-1]}')
    print(f'夏普比率：{sharpe_ratio}')

內容解密：

以上程式碼實作了事件驅動回測的基礎類別和只做多回測類別。基礎類別提供了資料載入和交易記錄的功能，而只做多回測類別則實作了均值迴歸策略的回測邏輯。最終，需要評估策略的績效，以確定其是否有效。

圖表翻譯：

  flowchart TD
    A[載入資料] --> B[執行策略]
    B --> C[計算績效]
    C --> D[評估績效]
    D --> E[輸出結果]

此圖表描述了回測過程的流程，從載入資料到輸出結果。每一步驟都對應到程式碼中的特定功能，例如載入資料、執行策略、計算績效等。

平均迴歸策略引數設定

在進行平均迴歸策略時，需要設定兩個重要引數：SMA（簡單移動平均）和threshold（門檻值）。SMA是用來計算移動平均的天數，threshold則是用來判斷價格偏離移動平均的絕對值。

引數設定

  • SMA：簡單移動平均的天數，決定了移動平均的計算範圍。
  • threshold：門檻值，決定了價格偏離移動平均的程度，超過此值則觸發買賣訊號。

策略執行

當設定好SMA和threshold後，策略便會開始執行。首先，計算每日的移動平均價格，然後根據價格與移動平均價格的差異，判斷是否需要進行買賣操作。

# 設定引數
SMA = 20  # 移動平均天數
threshold = 0.05  # 閾值

# 初始化變數
position = 0  # 初始持倉狀態
trades = 0  # 初始交易次數
amount = initial_amount  # 初始資金

# 計算移動平均價格
data['SMA'] = data['price'].rolling(SMA).mean()

# 逐日執行策略
for bar in range(SMA, len(data)):
    if position == 0:
        # 判斷是否買入
        if data['price'].iloc[bar] < data['SMA'].iloc[bar] * (1 - threshold):
            # 買入操作
            position = 1
            trades += 1
            print(f"買入於{data.index[bar]}，價格{data['price'].iloc[bar]}")
    elif position == 1:
        # 判斷是否賣出
        if data['price'].iloc[bar] > data['SMA'].iloc[bar] * (1 + threshold):
            # 賣出操作
            position = 0
            trades += 1
            print(f"賣出於{data.index[bar]}，價格{data['price'].iloc[bar]}")

策略評估

平均迴歸策略的績效可以透過多個指標來評估，例如勝率、賺率、最大回撤等。這些指標可以幫助投資者瞭解策略的風險和收益特性。

# 策略績效評估
def evaluate_strategy(data, SMA, threshold):
    # 初始化變數
    position = 0
    trades = 0
    amount = initial_amount
    
    # 計算移動平均價格
    data['SMA'] = data['price'].rolling(SMA).mean()
    
    # 逐日執行策略
    for bar in range(SMA, len(data)):
        if position == 0:
            # 判斷是否買入
            if data['price'].iloc[bar] < data['SMA'].iloc[bar] * (1 - threshold):
                # 買入操作
                position = 1
                trades += 1
        elif position == 1:
            # 判斷是否賣出
            if data['price'].iloc[bar] > data['SMA'].iloc[bar] * (1 + threshold):
                # 賣出操作
                position = 0
                trades += 1
                
    # 計算績效指標
    win_rate = trades / (trades + 1)
    profit_rate = (amount - initial_amount) / initial_amount
    max_drawdown = (amount - initial_amount) / initial_amount
    
    return win_rate, profit_rate, max_drawdown

# 執行評估
win_rate, profit_rate, max_drawdown = evaluate_strategy(data, SMA, threshold)
print(f"勝率：{win_rate:.2f}, 賺率：{profit_rate:.2f}, 最大回撤：{max_drawdown:.2f}")

技術指標策略的實作

在進行技術指標策略的實作時，需要考慮多個因素，包括策略的邏輯、交易成本、市場狀況等。在這個例子中，我們將實作一個簡單的移動平均線（SMA）策略。

策略邏輯

策略的邏輯如下：

  1. 當市場價格低於SMA時，買入。
  2. 當市場價格高於SMA時，賣出。

程式碼實作

import pandas as pd

class SMA_strategy:
    def __init__(self, data, sma_window):
        self.data = data
        self.sma_window = sma_window
        self.position = 0

    def calculate_sma(self):
        self.data['SMA'] = self.data['price'].rolling(window=self.sma_window).mean()

    def execute_strategy(self):
        for bar in range(len(self.data)):
            if self.position == 0:
                if self.data['price'].iloc[bar] < self.data['SMA'].iloc[bar]:
                    self.place_buy_order(bar, amount=1000)
                    self.position = 1
            elif self.position == 1:
                if self.data['price'].iloc[bar] >= self.data['SMA'].iloc[bar]:
                    self.place_sell_order(bar, units=1000)
                    self.position = 0

    def place_buy_order(self, bar, amount):
        # 執行買入交易
        print(f"買入 {amount} 單位於 {self.data['price'].iloc[bar]}")

    def place_sell_order(self, bar, units):
        # 執行賣出交易
        print(f"賣出 {units} 單位於 {self.data['price'].iloc[bar]}")

# 範例使用
data = pd.DataFrame({
    'price': [10, 12, 15, 18, 20, 22, 25, 28, 30, 32]
})

sma_window = 3
strategy = SMA_strategy(data, sma_window)
strategy.calculate_sma()
strategy.execute_strategy()

結果分析

在這個例子中，SMA策略的結果取決於SMA的視窗大小。不同的視窗大小會導致不同的交易結果。

圖表翻譯：

  flowchart TD
    A[開始] --> B[計算SMA]
    B --> C[執行策略]
    C --> D[買入]
    C --> E[賣出]
    D --> F[更新位置]
    E --> F
    F --> G[結束]

內容解密：

  1. calculate_sma方法計算SMA值。
  2. execute_strategy方法執行策略邏輯。
  3. place_buy_order和place_sell_order方法執行買入和賣出交易。

從使用者經驗視角來看，建構一個兼具彈性與易用性的回測框架是提升交易策略開發效率的關鍵。本文深入探討了回測框架的核心方法、事件驅動機制、交易策略基礎類別的建構，以及技術指標策略的實作，並佐以程式碼範例和流程圖表，完整呈現了回測系統的開發流程。分析段落中，我們剖析了買賣訂單處理、交易記錄更新和資產狀態計算等核心模組，並闡述瞭如何透過繼承基礎類別來實作客製化交易策略，例如只做多策略。同時，也強調了引數設定的重要性，例如SMA和threshold在均值迴歸策略中的作用。然而，目前提供的程式碼範例仍相對簡化，缺乏對真實市場環境的模擬，例如滑價、交易費用等，這也是未來需要強化的方向。展望未來，整合更多市場資料、最佳化績效評估指標，並加入風險管理模組，將是回測框架演進的重點。玄貓認為，一個完善的回測框架不僅應具備精確的模擬能力，更需提供使用者友善的操作介面和視覺化工具，才能真正賦能交易策略的開發與最佳化，進而提升投資績效。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。