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

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)

# 导入yahoofeed模块
from pyalgotrade.barfeed import yahoofeed


# 创建yahoofeed.Feed类创建其实例
feed = yahoofeed.Feed()

# 通过addBarsFromCSV加载本地csv文件
# 传入股票代码名, 文件路径
feed.addBarsFromCSV("orcl", "orcl-2000.csv")

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

pyalgotrade/pyalgotrade/dataseries/bards.py:87

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

# coding: utf8
import abc


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

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

pyalgotrade/pyalgotrade/strategy/__init__.py

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)

        BaseStrategy.__init__(self, barFeed, broker)
        # 默认self.__useAdjustedValue=False
        self.__useAdjustedValues = False
        # 配置日志参数
        self.setUseEventDateTimeInLogs(True)
        self.setDebugMode(True)

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
        # 当交易订单的成交量大于当前bar的成交量的25%则不能成交
        # 没有滑点
        # 没有手续费
        self.__fillStrategy = fillstrategy.DefaultStrategy()
        self.__logger = logger.getLogger(Broker.LOGGER_NAME)

        # 让barfeed对象订阅self.onBars方法
        barFeed.getNewValuesEvent().subscribe(self.onBars)
        self.__barFeed = barFeed
        self.__allowNegativeCash = False
        self.__nextOrderId = 1

pyalgotrade/pyalgotrade/technical/__init__.py

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

    def getEventWindow(self):
        return self.__eventWindow

pyalgotrade/pyalgotrade/dataseries/bards.py

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

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

# 集成strategy.BacktestingStrategy类
class MyStrategy(strategy.BacktestingStrategy):
    def __init__(self, feed, instrument, smaPeriod):
        # 调用父类__init__方法
        super(MyStrategy, self).__init__(feed, 1000)
        # 初始情况下，postion设置为零, postion一般只持仓比例
        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)

    # 省略其他钩子函数

    # 必须实现的onBars函数,用于买卖的主要逻辑
    def onBars(self, bars):
        # 如果没有简单移动平均值则什么都不做
        if self.__sma[-1] is None:
            return

        # 取出指定股票代码的bar对象
        bar = bars[self.__instrument]
        
        # 如果postion is None，即持仓为0
        if self.__position is None:
            # 如果收盘价大于简单移动平均值则买入
            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()