PyalgoTrade源碼閱讀完結篇

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()

# 創建yahoofeed.Feed類創建其實例
feed = yahoofeed.Feed()

class Event(object):
   """事件類.
   用於訂閱指定的操作,如函數
   當事件執行emit方法的時候,遍歷訂閱了的操作,並執行該操作"""
   def __init__(self):
       # 內部handlers列表
       self.__handlers = []

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

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

class Subject(object):
   """將元類指向abc.ABCMeta元類
   1. 當抽象方法未被實現的時候,不能新建該類的實例
   2. abstractmethod相當於子類要實現的介面,如果不實現,則不能新建該類的實例"""
   __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):
   """調度類
   1. 維護事件循環
   2. 不斷的調度subject的disptch操作並判斷是否結束"""
   def __init__(self):
       self.__subjects = []
       self.__stop = False

   def run(self):
       """運行整個事件循環並在調度之前,之後分別調用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. 承載數據源
   2. 通過數據驅動事件循環"""
   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裡面
       # 當Feed調用dispatch方法的時候, 會指定self.__onBars函數
       self.__feed.getNewValueEvent().subscribe(self.__onBars)
       # 注意順序,Feed對象必須在最後
       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()

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

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

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

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

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

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

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
       # 當交易訂單的成交量大於當前bar的成交量的25%則不能成交
       # 沒有滑點
       # 沒有手續費
       self.__fillStrategy = fillstrategy.DefaultStrategy()
       self.__logger = logger.getLogger(Broker.LOGGER_NAME)

class EventWindow(object):
   """數據實際承載類
   數據保存在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是否已經填滿"""
       return len(self.__values) == self.__windowSize

   def getValue(self):
       """子類須實現的類"""
       raise NotImplementedError()

class EventBasedFilter(dataseries.SequenceDataSeries):
   def __init__(self, dataSeries, eventWindow, maxLen=None):
       super(EventBasedFilter, self).__init__(maxLen)
       self.__dataSeries = dataSeries
       # 當dataseries數據有新值的時候，調用self.__onNewValues方法
       self.__dataSeries.getNewValueEvent().subscribe(self.__onNewValue)
       self.__eventWindow = eventWindow

   def __onNewValue(self, dataSeries, dateTime, value):
       # 讓EventWindow對象計算新值
       self.__eventWindow.onNewValue(dateTime, value)
       # 獲取計算後的結果
       newValue = self.__eventWindow.getValue()
       # 將值保存到自身實例裡面, 即self.__values
       # 因為繼承了dataseries.SequenceDataSeries類
       # 而dataseries.SequenceDataSeries父類實現了__getitem__方法, 所以可以使用索引取值.
       self.appendWithDateTime(dateTime, newValue)

   def getDataSeries(self):
       return self.__dataSeries

# 首先調用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

# 然後調用SequenceDataSeries對象的appendWithDateTime
# 在這個方法中提交數據更新的事件
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)

   # 省略其他鉤子函數

   # 必須實現的onBars函數,用於買賣的主要邏輯
   def onBars(self, bars):
       # 如果沒有簡單移動平均值則什麼都不做
       if self.__sma[-1] is None:
           return

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