基礎需求

通常在簡單交易的情況下，會在安全範圍內買進可交易的最大部位，也會在出場時一次性賣出持有的全部部位，接下來將在該前提下設計交易執行模組。

模組實作

策略運行模組發出的訊號，若依基礎需求所述，當第一個 BUY 訊號被執行後，後續的 BUY 訊號可不理會或加碼，直到 SELL 訊號出現後，賣出持有的全部部位，而賣出後，在下一個 BUY 訊號出現前，所有出現的 SELL 訊號均忽略。

交易執行管理器類別

TradeManager.py

class TradeManager:

    # 將策略與歷史 Tick 資料源傳入
    def __init__(self, strategy, ticks):
        self.Tick = None
        # 使用策略
        self.Strategy = strategy
        # 接管策略的進出場事件
        self.Strategy.OnBuy = self.OnBuy
        self.Strategy.OnSell = self.OnSell
        # 歷史 Tick 資料源
        self.Ticks = ticks

    def OnBuy(self, ds, data):
        # 若未持有部位才買進
        if self.Tick is None:
            tick = self.Ticks.Sequence.pop(0)
            print((
                f'[BUY]\n'
                f'ON PRICE {tick.price} AT {tick.time:%Y-%m-%d %H:%M:%S}\n'
            ))
            self.Tick = tick
    
    def OnSell(self, ds, data):
        # 若已持有部位賣出
        if self.Tick is not None:
            tick = self.Ticks.Sequence.pop(0)
            print((
                f'[SELL] PROFIT {tick.price - self.Tick.price}\n'
                f'ON PRICE {tick.price} AT {tick.time:%Y-%m-%d %H:%M:%S}\n'
            ))
            self.Tick = None

到此已完成按照買賣訊號執行進出場交易的部分，但問題來了，我們並不知道買賣的價格為何？

因此，要透過導入 Tick 歷史資料，在 1 分 K 的訊號發生時，找尋 1 分 K 指標產生時的下 1 個 Tick 訊號 做為假定的成交價，而 1 分 K 指標產生的時間，實際上是在 1 分 K 指標中時間欄位的下一分鐘開始時才產生。

Strategy.py

import abc

import events

class Strategy(abc.ABC, events.Events):
    __events__ = ('OnBuy', 'OnSell')

    @abc.abstractmethod
    def Feed(self, data):
        return NotImplemented

Buy3RK1BKAndSell3BK1RK.py

class Buy3RK1BKAndSell3BK1RK(Strategy):

    def __init__(self):
        self.Bars = []
    
    def Feed(self, data):
        status = 0
        if data.open > data.close:
            status = +1
        if data.open < data.close:
            status = -1
        if sum(self.Bars) == +3 and status == -1:
            if self.OnBuy is not None:
                self.OnBuy(self, data)
        if sum(self.Bars) == -3 and status == +1:
            if self.OnSell is not None:
                self.OnSell(self, data)
        self.Bars.append(status)
        if len(self.Bars) > 3:
            self.Bars.pop(0)

DataSource.py

import abc

import events
import munch

class DataSource(abc.ABC, events.Events):
    __events__ = ('OnData')

    def __init__(self, name, option):
        self.Name = name
        self.Option = munch.munchify(option)
        self.Sequence = []

    @abc.abstractmethod
    def Setup(self):
        return NotImplemented

    @abc.abstractmethod
    def Start(self):
        return NotImplemented

MongoDataSource.py

import munch
import pymongo

class MongoDataSource(DataSource):

    def Setup(self):
        self.Client = pymongo.MongoClient(
            f'mongodb://{self.Option.Username}:{self.Option.Password}@{self.Option.Host}:{self.Option.Port}/'
        )
        self.Database = self.Client[self.Option.Database]

    def Start(self):
        cursor = self.Database[self.Option.Collection].find().sort([('created', 1)])
        for doc in cursor:
            doc = munch.munchify(doc)
            self.Sequence.append(doc)
            if self.OnData is not None:
                self.OnData(self, doc)

main.py

from MongoDataSource import *
from TimeSeriesManager import *
from Buy3RK1BKAndSell3BK1RK import *
from TradeManager import *

def main():

    def k1min_OnData(ds, data):
        # 把 1 分 K 訊號中的時間欄位都往後延遲 1 分鐘，使訊號與 Tick 混和時保持正確時序
        data.time = data.time + datetime.timedelta(minutes=1)
    k1min = MongoDataSource(
        name='TXF-1MINK',
        option={
            'Username': 'root',
            'Password': 'root',
            'Host': 'localhost',
            'Port': 27017,
            'Database': 'backtest',
            'Collection': 'k1min'
        }
    )
    k1min.OnData += k1min_OnData

    ticks = MongoDataSource(
        name='TXF-TICKS',
        option={
            'Username': 'root',
            'Password': 'root',
            'Host': 'localhost',
            'Port': 27017,
            'Database': 'backtest',
            'Collection': 'ticks'
        }
    )

    strategy = Buy3RK1BKAndSell3BK1RK()

    # 將策略與歷史 Tick 資料源傳入交易執行管理器
    trade = TradeManager(strategy, ticks)

    def tsm_OnData(ds, data):
        if ds.Name == 'TXF-1MINK':
            strategy.Feed(data)
    tsm = TimeSeriesManager()
    tsm.DataSources.append(k1min)
    tsm.DataSources.append(ticks)
    tsm.OnData = tsm_OnData
    tsm.Setup()
    tsm.Start()

if __name__ == '__main__':
    main()

目前已完成回測交易如上，接下來就是透過交易統計模組計算報表，以取得回測結果，做為策略驗證使用。

團隊系列文：

CSScoke - 金魚都能懂的這個網頁畫面怎麼切 - 金魚都能懂了你還怕學不會嗎
Clarence - LINE bot 好好玩 30 天玩轉 LINE API
Hina Hina - 陣列大亂鬥
King Tzeng - IoT沒那麼難！新手用JavaScript入門做自己的玩具
Vita Ora - 好 Js 不學嗎 !? JavaScript 入門中的入門。
TaTaMo - 用Python開發的網頁不能放到Github上？Lektor說可以！！