pyalgotrade.example.py

from __future__ import print_function
import pandas as pd
import ccxt
import datetime
from pyalgotrade import strategy
from pyalgotrade.barfeed import quandlfeed, csvfeed
from pyalgotrade.technical import ma
from pyalgotrade.bar import Frequency
from pyalgotrade.stratanalyzer import returns, trades
from pyalgotrade import plotter

exchange = ccxt.ftx()

def gather_data(pair):
    data = exchange.fetch_ohlcv(pair)
    df = pd.DataFrame(data)
    df.columns = (["Date Time", "Open", "High", "Low", "Close", "Volume"])
    
    def parse_dates(ts):
        return datetime.datetime.fromtimestamp(ts/1000.0)
    
    df["Date Time"] = df["Date Time"].apply(parse_dates)
    df.to_csv("sampledata.csv")
    

class Accumulator(strategy.BacktestingStrategy):
    def __init__(self, feed, instrument, buy_offset, buy_percent):
        super(Accumulator, self).__init__(feed, 10000) # initial investment
        self.__position = None
        self.__instrument = instrument
        self.__sma = ma.SMA(feed[instrument].getPriceDataSeries(), 60)
        self.offset = buy_offset
        self.buy_percent = buy_percent # not used
    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]
        # self.info(bar.getClose())
        # self.info(self.__sma[-1])
        
        # If a position was not opened, check if we should enter a long position.
        shares = (self.getBroker().getCash() / bars[self.__instrument].getPrice())
        if self.__position is None:
            if (bar.getPrice() * (1 + self.offset) > self.__sma[-1]):
                # Enter a buy market order. The order is good till canceled.
                self.__position = self.enterLong(self.__instrument, shares, True)
        # Check if we have to exit the position.
        elif (bar.getPrice() * (1 - self.offset) < self.__sma[-1]) and not self.__position.exitActive():
            self.__position.exitMarket()
            
    def getSMA(self):
        return self.__sma    
    
def backtest():
    feed = csvfeed.GenericBarFeed(frequency=Frequency.MINUTE)
    feed.addBarsFromCSV("ETH", "sampledata.csv")
    # Evaluate the strategy with the feed's bars.
    myStrategy = Accumulator(feed, "ETH", buy_offset=0.0024, buy_percent=0.1)
    # myStrategy.run()
    returnsAnalyzer = returns.Returns()
    myStrategy.attachAnalyzer(returnsAnalyzer)
    tradesAnalyzer = trades.Trades()
    myStrategy.attachAnalyzer(tradesAnalyzer)
    # Run the strategy.
    plt = plotter.StrategyPlotter(myStrategy)
    # Include the SMA in the instrument's subplot to get it displayed along with the closing prices.
    plt.getInstrumentSubplot("ETH").addDataSeries("SMA", myStrategy.getSMA())
    # Plot the simple returns on each bar.
    plt.getOrCreateSubplot("returns").addDataSeries("Simple returns", returnsAnalyzer.getReturns())
    # Run the strategy.
    myStrategy.run()
    myStrategy.info("Final portfolio value: $%.2f" % myStrategy.getResult())
    # Plot the strategy.
    plt.plot()
    print("Final portfolio value: $%.2f" % myStrategy.getResult())

def main():
    gather_data("ETH/USD") # Get market CSV 
    backtest()
    
if __name__ == "__main__":
    main()