iT邦幫忙

第 11 屆 iThome 鐵人賽

DAY 28
2

基礎需求

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

模組實作

策略運行模組發出的訊號,若依基礎需求所述,當第一個 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()

執行結果

行為 時間 價格 獲益 總獲益
買進 2019-01-02 09:08:00 9702
賣出 2019-01-02 09:20:00 9693 -9 -9
買進 2019-01-02 09:33:00 9690
賣出 2019-01-02 10:30:00 9576 -114 -123
買進 2019-01-02 10:37:00 9561
賣出 2019-01-02 10:55:00 9576 +15 -108
買進 2019-01-02 11:05:00 9558
賣出 2019-01-02 11:47:00 9542 -16 -124
買進 2019-01-02 12:27:00 9529
賣出 2019-01-02 12:34:00 9521 -8 -132
買進 2019-01-02 12:40:01 9517
賣出 2019-01-02 12:52:00 9523 +6 -126
買進 2019-01-02 13:05:00 9520
賣出 2019-01-02 13:23:00 9529 +9 -117
買進 2019-01-02 13:34:00 9519
賣出 2019-01-02 13:41:00 9529 +10 -107
買進 2019-01-03 09:03:00 9457
賣出 2019-01-03 09:10:00 9474 +17 -90
買進 2019-01-03 09:25:00 9445
賣出 2019-01-03 09:29:00 9461 +16 -74
買進 2019-01-03 09:39:00 9484
賣出 2019-01-03 09:59:00 9524 +40 -34
買進 2019-01-03 10:13:00 9535
賣出 2019-01-03 10:39:00 9524 -11 -45
...

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


團隊系列文:

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


上一篇
Day-27 回測系統:策略運行模組
下一篇
Day-29 回測系統:交易統計模組
系列文
Python 程式交易 30 天新手入門30
圖片
  直播研討會
圖片
{{ item.channelVendor }} {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言