PyalgoTrade源碼閱讀完結(jié)篇

前言

本文著重于回測(cè)相關(guān)得模塊。

由于上一篇文章實(shí)在是寫得太爛了, 這一篇文章重新開始寫。

Pyalgotrade業(yè)務(wù)邏輯及實(shí)現(xiàn)原理

以官方教程示例為例

下載數(shù)據(jù)

python -c "from pyalgotrade.tools import yahoofinance; yahoofinance.download_daily_bars('orcl', 2000, 'orcl-2000.csv')"

構(gòu)建策略并運(yùn)行

from pyalgotrade import strategy
from pyalgotrade.barfeed import yahoofeed
from pyalgotrade.technical import ma

class MyStrategy(strategy.BacktestingStrategy):
    def __init__(self, feed, instrument, smaPeriod):
        super(MyStrategy, self).__init__(feed, 1000)
        self.__position = None
        self.__instrument = instrument
        # We'll use adjusted close values instead of regular close values.
        self.setUseAdjustedValues(True)
        self.__sma = ma.SMA(feed[instrument].getPriceDataSeries(), smaPeriod)

    def onEnterOk(self, position):
        execInfo = position.getEntryOrder().getExecutionInfo()
        self.info("BUY at $%.2f" % (execInfo.getPrice()))

    def onEnterCanceled(self, position):
        self.__position = None

    def onExitOk(self, position):
        execInfo = position.getExitOrder().getExecutionInfo()
        self.info("SELL at $%.2f" % (execInfo.getPrice()))
        self.__position = None

    def onExitCanceled(self, position):
        # If the exit was canceled, re-submit it.
        self.__position.exitMarket()

    def onBars(self, bars):
        # Wait for enough bars to be available to calculate a SMA.
        if self.__sma[-1] is None:
            return

        bar = bars[self.__instrument]
        # If a position was not opened, check if we should enter a long position.
        if self.__position is None:
            if bar.getPrice() > self.__sma[-1]:
                # Enter a buy market order for 10 shares. The order is good till canceled.
                self.__position = self.enterLong(self.__instrument, 10, True)
        # Check if we have to exit the position.
        elif bar.getPrice() < self.__sma[-1] and not self.__position.exitActive():
            self.__position.exitMarket()

def run_strategy(smaPeriod):
    # Load the yahoo feed from the CSV file
    feed = yahoofeed.Feed()
    feed.addBarsFromCSV("orcl", "orcl-2000.csv")

    # Evaluate the strategy with the feed.
    myStrategy = MyStrategy(feed, "orcl", smaPeriod)
    myStrategy.run()
    print "Final portfolio value: $%.2f" % myStrategy.getBroker().getEquity()

run_strategy(15)

業(yè)務(wù)邏輯概括

1. 創(chuàng)建Feed對(duì)象加載回測(cè)歷史數(shù)據(jù)
2. 創(chuàng)建策略
3. 將Feed對(duì)象傳入策略
4. 內(nèi)部創(chuàng)建Broker對(duì)象
5. 在策略中初始化技術(shù)指標(biāo)
6. 運(yùn)行策略(內(nèi)部會(huì)創(chuàng)建事件循環(huán),依次讀取每一個(gè)bars數(shù)據(jù)調(diào)用策略邏輯,即onBars函)

回測(cè)數(shù)據(jù) Feed對(duì)象

用于承載回測(cè)的數(shù)據(jù)，提供接口訪問，驅(qū)動(dòng)整個(gè)事件循環(huán)。

創(chuàng)建Feed對(duì)象

# 導(dǎo)入yahoofeed模塊
from pyalgotrade.barfeed import yahoofeed

# 創(chuàng)建yahoofeed.Feed類創(chuàng)建其實(shí)例
feed = yahoofeed.Feed()

# 通過addBarsFromCSV加載本地csv文件
# 傳入股票代碼名, 文件路徑
feed.addBarsFromCSV("orcl", "orcl-2000.csv")

Feed對(duì)象繼承鏈

注: 由IntelliJ Idea生成

由上圖可知, 分別繼承不同的BarFeed,最終業(yè)務(wù)邏輯基類pyalgotrade.observer.subject.

Feed數(shù)據(jù)結(jié)構(gòu)構(gòu)建過程

主要方法調(diào)用順序如下:

yahooFeed.addBarsFromCSV

-> csvFeed.BarFeed.addBarsFromCSV

-> membf.BarFeed.addBarsFromSequence

-> barfeed.registerInstrument

-> feed.registerDataSeries

-> barfeed.createDataSeries

Feed數(shù)據(jù)結(jié)構(gòu)

在Feed中有兩個(gè)比較重要的數(shù)據(jù)對(duì)象

1. self.__bars = {}
2. self.__ds = BarDataSeries()
   其中BarDataSeries對(duì)象有以下定義

pyalgotrade/pyalgotrade/dataseries/bards.py

class BarDataSeries(dataseries.SequenceDataSeries):
    def __init__(self, maxLen=None):
        super(BarDataSeries, self).__init__(maxLen)
        self.__openDS = dataseries.SequenceDataSeries(maxLen)
        self.__closeDS = dataseries.SequenceDataSeries(maxLen)
        self.__highDS = dataseries.SequenceDataSeries(maxLen)
        self.__lowDS = dataseries.SequenceDataSeries(maxLen)
        self.__volumeDS = dataseries.SequenceDataSeries(maxLen)
        self.__adjCloseDS = dataseries.SequenceDataSeries(maxLen)
        self.__extraDS = {}
        self.__useAdjustedValues = False

BarDataSeries提供一系列方法返回相應(yīng)的數(shù)據(jù)序列，以getOpenDataSeries為例

pyalgotrade/pyalgotrade/dataseries/bards.py:87

    def getOpenDataSeries(self):
        """Returns a :class:`pyalgotrade.dataseries.DataSeries` with the open prices."""
        return self.__openDS

而dataseries.SequenceDataSeries對(duì)象是一個(gè)數(shù)據(jù)存儲(chǔ)在collections.ListDeque對(duì)象上，并集成事件監(jiān)聽的類對(duì)象.

self._bars在membf.BarFeed.addBarsFromSequence方法中讀取csv文件生成.
self._ds在barfeed.createDataSeries方法中創(chuàng)建一個(gè)默認(rèn)長(zhǎng)度為1024的BarDataSeries空數(shù)據(jù)對(duì)象.

小結(jié)

bar是含有時(shí)間, 開盤價(jià), 收盤價(jià), 當(dāng)日最高價(jià), 當(dāng)日最低價(jià), 成交量，復(fù)權(quán)收盤價(jià)的數(shù)據(jù)對(duì)象.

self.__bars是key為股票代碼, value是元素為bars數(shù)據(jù)對(duì)象的列表的字典.

self.__ds是BarDataSeries對(duì)象

事件循環(huán)

事件循環(huán)是PyalgoTrade的數(shù)據(jù)引擎，驅(qū)動(dòng)著整個(gè)策略運(yùn)轉(zhuǎn).

下面是Pyalgotrade內(nèi)部事件循環(huán)的一個(gè)簡(jiǎn)單的實(shí)現(xiàn)。

# coding: utf8
import abc

class Event(object):
    """事件類.
    用于訂閱指定的操作,如函數(shù)
    當(dāng)事件執(zhí)行emit方法的時(shí)候,遍歷訂閱了的操作,并執(zhí)行該操作"""
    def __init__(self):
        # 內(nèi)部handlers列表
        self.__handlers = []

    def subscribe(self, handler):
        if handler not in self.__handlers:
            self.__handlers.append(handler)

    def emit(self, *args, **kwargs):
        """執(zhí)行所有訂閱了的操作"""
        for handler in self.__handlers:
            handler(*args, **kwargs)

class Subject(object):
    """將元類指向abc.ABCMeta元類
    1. 當(dāng)抽象方法未被實(shí)現(xiàn)的時(shí)候,不能新建該類的實(shí)例
    2. abstractmethod相當(dāng)于子類要實(shí)現(xiàn)的接口,如果不實(shí)現(xiàn),則不能新建該類的實(shí)例"""
    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod
    def start(self):
        pass

    @abc.abstractmethod
    def stop(self):
        pass

    @abc.abstractmethod
    def dispatch(self):
        raise NotImplementedError()

    @abc.abstractmethod
    def eof(self):
        raise NotImplementedError()

class Dispatcher(object):
    """調(diào)度類
    1. 維護(hù)事件循環(huán)
    2. 不斷的調(diào)度subject的disptch操作并判斷是否結(jié)束"""
    def __init__(self):
        self.__subjects = []
        self.__stop = False

    def run(self):
        """運(yùn)行整個(gè)事件循環(huán)并在調(diào)度之前,之后分別調(diào)用subject的start, stop方法"""
        try:
            for subject in self.__subjects:
                subject.start()

            while not self.__stop:
                eof, dispatched = self.dispatch()
                if eof:
                    self.__stop = True
        finally:
            for subject in self.__subjects:
                subject.stop()

    def dispatch(self):
        ret = False
        eof = False
        for subject in self.__subjects:
            ret = subject.dispatch() is True
            eof = subject.eof()

        return eof, ret

    def addSubject(self, subject):
        self.__subjects.append(subject)

class Broker(Subject):
    """Broker 類"""
    def dispatch(self):
        return None

    def eof(self):
        return None

    def start(self):
        pass

    def stop(self):
        pass

class Feed(Subject):
    """Feed類
    1. 承載數(shù)據(jù)源
    2. 通過數(shù)據(jù)驅(qū)動(dòng)事件循環(huán)"""
    def __init__(self, size):
        self.__data = range(size)
        self.__nextPos = 0
        self.__event = Event()

    def start(self):
        pass

    def stop(self):
        pass

    def dispatch(self):
        value = self.__data[self.__nextPos]
        self.__event.emit(value)
        self.__nextPos += 1
        return True

    def getNewValueEvent(self):
        return self.__event

    def eof(self):
        return self.__nextPos >= len(self.__data)

class Strategy(object):
    def __init__(self, broker, feed):
        self.__dispatcher = Dispatcher()
        self.__feed = feed
        self.__broker = broker
        # 將策略的self.__onBars方法傳入Feed的self.__event里面
        # 當(dāng)Feed調(diào)用dispatch方法的時(shí)候, 會(huì)指定self.__onBars函數(shù)
        self.__feed.getNewValueEvent().subscribe(self.__onBars)
        # 注意順序,Feed對(duì)象必須在最后
        self.__dispatcher.addSubject(self.__broker)
        self.__dispatcher.addSubject(self.__feed)

    def __onBars(self, value):
        print("dispatch before.")
        self.onBars(value)
        print("dispatch after")

    def onBars(self, value):
        print("on Bar: {}".format(value))

    def run(self):
        self.__dispatcher.run()

if __name__ == '__main__':
    feed = Feed(3)
    broker = Broker()
    myStrategy = Strategy(broker, feed)
    myStrategy.run()

output: 
dispatch before.
on Bar: 0
dispatch after
dispatch before.
on Bar: 1
dispatch after
dispatch before.
on Bar: 2
dispatch after

上面的代碼主要說明策略的onBars方法是怎么被調(diào)用的。

關(guān)于Broker怎么被驅(qū)動(dòng)，在后面講解

1. 策略中維護(hù)一個(gè)調(diào)度器dispatcher,當(dāng)策略啟動(dòng)的時(shí)候, 調(diào)度器dipatcher啟動(dòng), 并嘗試調(diào)用feed,broker start方法.
2. 不斷調(diào)用feed, broker的dispatch方法, 判斷是否結(jié)束, 如果結(jié)束, 則做結(jié)束動(dòng)作, 調(diào)用feed, broker的stop方法
3. feed對(duì)象在調(diào)用dispatch方法的時(shí)候, feed對(duì)象會(huì)觸發(fā)自身維護(hù)的self._event. 而self._event在MyStrategy._init_方法中，通過self._feed.getNewValueEvent().subscribe(self._onBars)訂閱了MyStrategy._onBars方法, 所以Feed對(duì)象每次dispatch的時(shí)候，MyStrategy._onBars都會(huì)被調(diào)用.

至此, Feed對(duì)象怎么驅(qū)動(dòng)策略的邏輯已經(jīng)清晰。
接下來，講解BaseStrategy, BacktestingStrategy初始化過程

策略初始化

策略的繼承鏈并不復(fù)雜, 所有策略的基類是BaseStartegy, BacktestingStrategy是提供給用戶使用的策略，至少實(shí)現(xiàn)onBars函數(shù)則可以回測(cè)。

BaseStrategy, BacktestingStrategy的初始化源代碼如下

pyalgotrade/pyalgotrade/strategy/__init__.py

class BaseStartegy(object):
    def __init__(self, barFeed, broker):
        # 綁定barFeed對(duì)象
        self.__barFeed = barFeed
        # 綁定broker對(duì)象
        self.__broker = broker
        # 交易相關(guān)的倉位
        self.__activePositions = set()
        # 訂單處理順序
        self.__orderToPosition = {}
        # bar被處理后的事件
        self.__barsProcessedEvent = observer.Event()
        # analyzer列表
        self.__analyzers = []
        # 命名的analyzer列表
        self.__namedAnalyzers = {}
        # 重新取樣的feed對(duì)象列表
        self.__resampledBarFeeds = []
        # 調(diào)度器對(duì)象
        self.__dispatcher = dispatcher.Dispatcher()
        # broker的訂單被更新時(shí)的事件, 訂閱self.__onOrderEvent方法
        self.__broker.getOrderUpdatedEvent().subscribe(self.__onOrderEvent)
        # barfeed值被更新的時(shí)候的事件(當(dāng)barfeed被調(diào)度的時(shí)候),訂閱self.__onBars方法
        self.__barFeed.getNewValuesEvent().subscribe(self.__onBars)

        # 調(diào)度器的開始事件，訂閱self.onStart方法
        self.__dispatcher.getStartEvent().subscribe(self.onStart)
        # 調(diào)度器的空閑事件, 訂閱self.__onIdle方法
        self.__dispatcher.getIdleEvent().subscribe(self.__onIdle)

        # 分別將繼承了Subject類的broker,barFeed對(duì)象加入到調(diào)度器的subject列表
        self.__dispatcher.addSubject(self.__broker)
        self.__dispatcher.addSubject(self.__barFeed)

        # 日志級(jí)別的初始化
        self.__logger = logger.getLogger(BaseStrategy.LOGGER_NAME)

class BacktestingStrategy(BaseStrategy):
    # 默認(rèn)初始化一個(gè)持有100w現(xiàn)金的虛擬賬戶
    def __init__(self, barFeed, cash_or_brk=1000000):

        # 如果沒有傳入cash_or_brk參數(shù), 或者傳入數(shù)值類型的值
        # 則傳入cash_or_brk,barFeed對(duì)象新建一個(gè)backtesting.Broker實(shí)例，并調(diào)用父類的__init__方法
        # 如果傳入的cash_or_brk參數(shù)值是backtesting.Broker的實(shí)例, 則直接使用
        if isinstance(cash_or_brk, pyalgotrade.broker.Broker):
            broker = cash_or_brk
        else:
            broker = backtesting.Broker(cash_or_brk, barFeed)

        BaseStrategy.__init__(self, barFeed, broker)
        # 默認(rèn)self.__useAdjustedValue=False
        self.__useAdjustedValues = False
        # 配置日志參數(shù)
        self.setUseEventDateTimeInLogs(True)
        self.setDebugMode(True)

總的來說真正Strategy對(duì)象,barFeed對(duì)象,broker對(duì)象訂閱了更多的事件, 以及更多的判斷。但，內(nèi)核都是調(diào)度器驅(qū)動(dòng)著barFeed, broker對(duì)象不斷的被調(diào)度(調(diào)用dispatch方法), 而barFeed對(duì)象會(huì)不斷的從self._bars中取數(shù)據(jù)追加到self._ds對(duì)象中，并將取出來的數(shù)據(jù)提交的self._event中,而self._event訂閱了Strategy.__onBars方法, 所以不斷的驅(qū)動(dòng)著Strategy的自定義策略(onBars里面定義的交易邏輯).

交易賬戶 Broker對(duì)象

在Strategy對(duì)象初始化時(shí)候, 會(huì)初始化一個(gè)虛擬的回測(cè)賬戶.

回測(cè)賬戶broker需要傳入barfeed對(duì)象, 并在barfeed的event對(duì)象里面訂閱自己的onBars函數(shù)，源碼如下:

pyalgotrade/pyalgotrade/broker/__init__.py

class Broker(broker.Broker):
    LOGGER_NAME = "broker.backtesting"

    def __init__(self, cash, barFeed, commission=None):
        super(Broker, self).__init__()

        assert(cash >= 0)
        self.__cash = cash
        if commission is None:
            self.__commission = NoCommission()
        else:
            self.__commission = commission
        self.__shares = {}
        self.__activeOrders = {}
        self.__useAdjustedValues = False
        # 持倉策略, 使用DefaultStrategy
        # 使用DefaultStrategy.volumeLimit = 0.25
        # 當(dāng)交易訂單的成交量大于當(dāng)前bar的成交量的25%則不能成交
        # 沒有滑點(diǎn)
        # 沒有手續(xù)費(fèi)
        self.__fillStrategy = fillstrategy.DefaultStrategy()
        self.__logger = logger.getLogger(Broker.LOGGER_NAME)

        # 讓barfeed對(duì)象訂閱self.onBars方法
        barFeed.getNewValuesEvent().subscribe(self.onBars)
        self.__barFeed = barFeed
        self.__allowNegativeCash = False
        self.__nextOrderId = 1

由上可知，當(dāng)barFeed對(duì)象數(shù)據(jù)更新的時(shí)候，還會(huì)調(diào)用BackTestBroker.onBars方法.

交易倉位 Position對(duì)象

當(dāng)使用enterLong之類交易方法，則會(huì)返回一個(gè)Postion的對(duì)象，這個(gè)對(duì)象承載著當(dāng)前各股的持倉比例，以及持有現(xiàn)金.

以enterLong方法說明持倉流程.

1. 實(shí)例化一個(gè)LongPosition對(duì)象
2. 調(diào)用broker的createMarketOrder方法創(chuàng)建一個(gè)MarketOrder.
3. 注冊(cè)order, 以便barFeed對(duì)象數(shù)據(jù)驅(qū)動(dòng)的時(shí)候，使用該order

以exitMarket方法說明平倉流程.

1. 使用Position對(duì)象的exitMarket方法提交平倉訂單.
2. 注冊(cè)order, 以便barFeed對(duì)象數(shù)據(jù)驅(qū)動(dòng)的時(shí)候，使用該order

源代碼調(diào)用鏈太長(zhǎng)....所以文字概括.

交易訂單 Order對(duì)象

當(dāng)我們買入或者賣出的時(shí)候，其實(shí)是提交一個(gè)訂單給交易賬戶(broker), 交易賬戶會(huì)根據(jù)交易訂單的類型,動(dòng)作等相關(guān)信息執(zhí)行相關(guān)的操作.

交易訂單的類型參考: https://www.thebalance.com/understanding-stock-orders-3141318

一般有買入(做多), 賣出(做空)兩種交易類型, 但是這兩種類型成交的方式分別由市價(jià)成交, 限價(jià)成交.

所以一共由以下四種類型，對(duì)應(yīng)Strategy的四個(gè)方法:

1. enterLong 以市價(jià)(下一個(gè)Bar的開盤價(jià))買入
2. enterLongLimit 當(dāng)市價(jià)(下一個(gè)Bar的開盤價(jià))低于或等于指定的價(jià)格時(shí)買入
3. enterShort 與enterLong相反
4. enterShortLimit 與enterLongShort相反.

以enter開頭是更加上層的方法, 建議使用.

goodTillCanceled為了適配實(shí)盤接口, 實(shí)盤接口可能有前一天的訂單不會(huì)再執(zhí)行的限制，所以設(shè)置goodTillCanceled=True保證第二天或者更后的時(shí)間，訂單依然有效，直至手動(dòng)取消.

除了提交交易訂單還可以提交止損訂單, 分別對(duì)應(yīng)Strategy的兩個(gè)方法.

1. StopOrder 提交一個(gè)止損訂單, 傳入止損價(jià)格, 當(dāng)價(jià)格突破止損價(jià)位, 以市價(jià)成交進(jìn)行止損.
2. StopLimitOrder 提交一個(gè)止損訂單, 傳入止損價(jià)格, 當(dāng)價(jià)格突破止損價(jià)位, 并且價(jià)格在限定的價(jià)格區(qū)間才會(huì)止損.

每個(gè)提交的訂單會(huì)到下一個(gè)事件循環(huán)才會(huì)判斷條件是否符合，才會(huì)執(zhí)行.

技術(shù)指標(biāo) EventBasedFilter對(duì)象

通過借助自定義指標(biāo)或者自帶的指標(biāo)，如SMA,EMA,MACD等可以更全面的看待股票的走勢(shì)以及信號(hào).

下面是技術(shù)指標(biāo)基類的初始化過程.

pyalgotrade/pyalgotrade/technical/__init__.py

class EventWindow(object):
    """數(shù)據(jù)實(shí)際承載類
    數(shù)據(jù)保存在self__values里面
    """
    def __init__(self, windowSize, dtype=float, skipNone=True):
        assert(windowSize > 0)
        assert(isinstance(windowSize, int))
        self.__values = collections.NumPyDeque(windowSize, dtype)
        self.__windowSize = windowSize
        self.__skipNone = skipNone

    def onNewValue(self, dateTime, value):
        """提供onNewValue方法將新的值傳入"""
        if value is not None or not self.__skipNone:
            self.__values.append(value)

    def getValues(self):
        """獲取EventWindows的所有值"""
        return self.__values.data()

    def getWindowSize(self):
        """獲取EventWindow Size"""
        return self.__windowSize

    def windowFull(self):
        """eventWindow是否已經(jīng)填滿"""
        return len(self.__values) == self.__windowSize

    def getValue(self):
        """子類須實(shí)現(xiàn)的類"""
        raise NotImplementedError()

class EventBasedFilter(dataseries.SequenceDataSeries):
    def __init__(self, dataSeries, eventWindow, maxLen=None):
        super(EventBasedFilter, self).__init__(maxLen)
        self.__dataSeries = dataSeries
        # 當(dāng)dataseries數(shù)據(jù)有新值的時(shí)候，調(diào)用self.__onNewValues方法
        self.__dataSeries.getNewValueEvent().subscribe(self.__onNewValue)
        self.__eventWindow = eventWindow

    def __onNewValue(self, dataSeries, dateTime, value):
        # 讓EventWindow對(duì)象計(jì)算新值
        self.__eventWindow.onNewValue(dateTime, value)
        # 獲取計(jì)算后的結(jié)果
        newValue = self.__eventWindow.getValue()
        # 將值保存到自身實(shí)例里面, 即self.__values
        # 因?yàn)槔^承了dataseries.SequenceDataSeries類
        # 而dataseries.SequenceDataSeries父類實(shí)現(xiàn)了__getitem__方法, 所以可以使用索引取值.
        self.appendWithDateTime(dateTime, newValue)

    def getDataSeries(self):
        return self.__dataSeries

    def getEventWindow(self):
        return self.__eventWindow

在Feed對(duì)象初始過程中，會(huì)初始化兩個(gè)比較重要的數(shù)據(jù)結(jié)構(gòu), 一個(gè)是self._bars, 一個(gè)是self._ds,在整個(gè)事件驅(qū)動(dòng)中, 策略不停的從self_bars中取數(shù)據(jù)，然后使用appendWithDateTime方法將數(shù)據(jù)追加的self._ds里面。
源碼如下:

pyalgotrade/pyalgotrade/dataseries/bards.py

# 首先調(diào)用BarDataSeries的appendWithDateTime方法
class BarDataSeries(dataseries.SequenceDataSeries):
    def appendWithDateTime(self, dateTime, bar):
        assert(dateTime is not None)
        assert(bar is not None)
        bar.setUseAdjustedValue(self.__useAdjustedValues)

        super(BarDataSeries, self).appendWithDateTime(dateTime, bar)

        self.__openDS.appendWithDateTime(dateTime, bar.getOpen())
        self.__closeDS.appendWithDateTime(dateTime, bar.getClose())
        self.__highDS.appendWithDateTime(dateTime, bar.getHigh())
        self.__lowDS.appendWithDateTime(dateTime, bar.getLow())
        self.__volumeDS.appendWithDateTime(dateTime, bar.getVolume())
        self.__adjCloseDS.appendWithDateTime(dateTime, bar.getAdjClose())

        # Process extra columns.
        for name, value in bar.getExtraColumns().iteritems():
            extraDS = self.__getOrCreateExtraDS(name)
            extraDS.appendWithDateTime(dateTime, value)

pyalgotrade/dataseries/__init__.py

# 然后調(diào)用SequenceDataSeries對(duì)象的appendWithDateTime
# 在這個(gè)方法中提交數(shù)據(jù)更新的事件
class SequenceDataSeries(DataSeries):
    def appendWithDateTime(self, dateTime, value):
        """
        Appends a value with an associated datetime.

        .. note::
            If dateTime is not None, it must be greater than the last one.
        """

        if dateTime is not None and len(self.__dateTimes) != 0 and self.__dateTimes[-1] >= dateTime:
            raise Exception("Invalid datetime. It must be bigger than that last one")

        assert(len(self.__values) == len(self.__dateTimes))
        self.__dateTimes.append(dateTime)
        self.__values.append(value)

        self.getNewValueEvent().emit(self, dateTime, value)            

小結(jié)

使用技術(shù)指標(biāo)需要傳入dataSeries對(duì)象, 可以通過getPriceDataSeries, getOpenDataSeries等獲得.

創(chuàng)建策略

由于上面已經(jīng)有完整版本的代碼，這里做一定的刪減, 并做注解.

# 集成strategy.BacktestingStrategy類
class MyStrategy(strategy.BacktestingStrategy):
    def __init__(self, feed, instrument, smaPeriod):
        # 調(diào)用父類__init__方法
        super(MyStrategy, self).__init__(feed, 1000)
        # 初始情況下，postion設(shè)置為零, postion一般只持倉比例
        self.__position = None
        # 股票代碼
        self.__instrument = instrument
        # We'll use adjusted close values instead of regular close values.
        # 是否使用復(fù)權(quán)收盤價(jià)
        self.setUseAdjustedValues(True)
        # 初始化策略指標(biāo)
        self.__sma = ma.SMA(feed[instrument].getPriceDataSeries(), smaPeriod)

    # 省略其他鉤子函數(shù)

    # 必須實(shí)現(xiàn)的onBars函數(shù),用于買賣的主要邏輯
    def onBars(self, bars):
        # 如果沒有簡(jiǎn)單移動(dòng)平均值則什么都不做
        if self.__sma[-1] is None:
            return

        # 取出指定股票代碼的bar對(duì)象
        bar = bars[self.__instrument]

        # 如果postion is None，即持倉為0
        if self.__position is None:
            # 如果收盤價(jià)大于簡(jiǎn)單移動(dòng)平均值則買入
            if bar.getPrice() > self.__sma[-1]:
                # 買入,enterLong=做多
                self.__position = self.enterLong(self.__instrument, 10, True)
        # 反之賣出
        elif bar.getPrice() < self.__sma[-1] and not self.__position.exitActive():
            self.__position.exitMarket()

總結(jié)

BarFeed像是PyalgoTrade中的燃料，不斷的供給給策略的Dispatcher調(diào)度器, 使整個(gè)策略不斷運(yùn)行，直至沒有燃料(沒有新的數(shù)據(jù).)

BarFeed使數(shù)據(jù)源的一個(gè)抽象，里面保存著兩個(gè)重要的數(shù)據(jù)結(jié)構(gòu), self._bars, self._ds.

self.__bars是key為股票代碼, value是元素為bar數(shù)據(jù)對(duì)象的列表的字典.

self.__ds為BarDataSeries對(duì)象.

Broker維護(hù)著虛擬賬戶里面的現(xiàn)金以及相關(guān)股票的倉位.接收訂單并實(shí)時(shí)的處理訂單, 計(jì)算收益等.

Position為股票倉位持有情況的對(duì)象, 提供交易的相關(guān)接口.

EventBasedFilter為技術(shù)指標(biāo), 可以計(jì)算相關(guān)指標(biāo)如MACD, SMA等, 也可以自定義自己的技術(shù)指標(biāo).

Strategy為自定義策略,只需實(shí)現(xiàn)onBars函數(shù)即可完成買賣邏輯, 將Broker,Position相關(guān)接口放在Strategy實(shí)例方法里面, 同一調(diào)用接口.