基礎需求

至少要知道交易次數和總獲益點數。

模組實作

交易統計管理器類別

ReportManager.py

class ReportManager:

    def __init__(self):
        # 交易次數
        self.Times = 0
        # 總獲益
        self.Profit = 0

TradeManager.py

class TradeManager:

    def __init__(self, report, strategy, ticks):
        self.Tick = None
        self.Report = report
        self.Strategy = strategy
        self.Strategy.OnBuy = self.OnBuy
        self.Strategy.OnSell = self.OnSell
        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)
            # 遞增交易次數並加入本次獲益至總獲益
            profit = tick.price - self.Tick.price
            self.Report.Times = self.Report.Times + 1
            self.Report.Profit = self.Report.Profit + profit
            print((
                f'[SELL] PROFIT {profit}\n'
                f'ON PRICE {tick.price} AT {tick.time:%Y-%m-%d %H:%M:%S}\n'
            ))
            self.Tick = None

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):
        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(report, 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()

    # 輸出交易統計結果
    print((
        f'[REPORT]\n'
        f'TIMES {report.Times}\n'
        f'PROFIT {report.Profit}\n'
    ))

if __name__ == '__main__':
    main()

今年一月統計

交易：執行 106 次
　　　相當於每個交易日執行 4-5 次交易
　　　若這交易頻率不符心中預想規劃，那就要繼續修正策略
收益：虧損 116 點

恭喜！各位可以開始設計自己的策略，並使用這個簡單的回測系統進行初步的驗證囉！

投資有賺有賠，擁有工具的各位，還請慎重使用做好風控。

團隊系列文：

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