PyAlgoTrade：アルゴリズミックトレーディングのためのPythonライブラリ

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

class MovingAverageCrossStrategy(strategy.BacktestingStrategy):
    def __init__(self, feed, instrument, smaPeriod):
        super(MovingAverageCrossStrategy, self).__init__(feed)
        self.__instrument = instrument
        self.__position = None
        self.__sma = ma.SMA(feed[instrument].getCloseDataSeries(), smaPeriod)

    def onEnterOk(self, position):
        execInfo = position.getEntryOrder().getExecutionInfo()
        self.info(f"BUY at {execInfo.getPrice()}")

    def onExitOk(self, position):
        execInfo = position.getExitOrder().getExecutionInfo()
        self.info(f"SELL at {execInfo.getPrice()}")

    def onBars(self, bars):
        if self.__sma[-1] is None:
            return

        bar = bars[self.__instrument]
        if self.__position is None:
            if cross.cross_above(bar.getClose(), self.__sma):
                self.__position = self.enterLong(self.__instrument, 10, True)
        elif cross.cross_below(bar.getClose(), self.__sma):
            self.__position.exitMarket()

# Yahoo Financeからデータをロード
feed = yahoofeed.Feed()
feed.addBarsFromCSV("orcl", "orcl-2000.csv")

# ストラテジーの実行
myStrategy = MovingAverageCrossStrategy(feed, "orcl", 20)
myStrategy.run()

from pyalgotrade.technical import indicator

class CustomIndicator(indicator.Indicator):
    def __init__(self, dataSeries, period):
        super(CustomIndicator, self).__init__(dataSeries)
        self.__period = period
        self.__values = []

    def calculate(self):
        if len(self.__values) >= self.__period:
            return sum(self.__values[-self.__period:]) / self.__period
        return None

    def onNewValue(self, dateTime, value):
        self.__values.append(value)
        self.appendWithDateTime(dateTime, self.calculate())

class CustomStrategy(strategy.BacktestingStrategy):
    def __init__(self, feed, instrument, period):
        super(CustomStrategy, self).__init__(feed)
        self.__instrument = instrument
        self.__position = None
        self.__custom_indicator = CustomIndicator(feed[instrument].getCloseDataSeries(), period)

    def onBars(self, bars):
        if self.__custom_indicator[-1] is None:
            return

        bar = bars[self.__instrument]
        if self.__position is None:
            if bar.getClose() > self.__custom_indicator[-1]:
                self.__position = self.enterLong(self.__instrument, 10, True)
        elif bar.getClose() < self.__custom_indicator[-1]:
            self.__position.exitMarket()

# Yahoo Financeからデータをロード
feed = yahoofeed.Feed()
feed.addBarsFromCSV("orcl", "orcl-2000.csv")

# ストラテジーの実行
myStrategy = CustomStrategy(feed, "orcl", 20)
myStrategy.run()