Free Python

Basana Backtesting library with CryptoDataDownload Data Sources

Jan 10, 2024 · by CryptoDataDownload

PyAlgoTrader vs Basana

Per the Official Github page, PyAlgoTrade is basically deprecated and no longer maintained, and they suggest using Basana. Basana is a great library for backtesting and is often used to evaluate the effectiveness of a trading strategy using historical data before risking real capital. Some features of Basana is that it is asynchronous and event driven (making it faster) and already has some built in features for cryptocurrencies. The purpose of this article is to demonstrate how to modify the Basana backtesting CSV input to accept cryptocurrency historical data files dynamically from CryptoDataDownload (without pre-downloading any files)!

Code

We take the basic Backtesting using SMA example from the Basana github homepage and add a function that allows us to use any file from CryptoDataDownload. The function will download from any exchange and for any cryptocurrency pair, and then format the file to feed the Basana script. Please feel free to modify to fit your purposes!

Live codebase Python

# CryptoDataDownload
# No Express Warranty
# MIT License
from decimal import Decimal
import asyncio
import logging

from talipp.indicators import SMA
from basana.backtesting import charts
from basana.external.bitstamp import csv
import basana as bs
import basana.backtesting.exchange as backtesting_exchange
import pandas as pd
import os

def return_CDD_filepath(exchange='Binance', pair='BTC/USDT'):
    """This function is designed for you to be able to dynamically pull CryptoDataDownload historical data files directly
    into Basana examples for backtesting strategies"""
    sym = pair.split("/")[0]  # get the symbol in front of the slash, ie BTC from BTC/USD
    sym_base = pair.split("/")[1]  # get the base symbol, ie USD in BTC/USD
    new_filename = f'{exchange}_{sym + sym_base}_d.csv'
    URL = f"https://www.cryptodatadownload.com/cdd/{new_filename}"  # path to file
    data = pd.read_csv(URL, skiprows=1)  # use pandas to read file from site
    data.sort_values('Date', ascending=True, inplace=True)  # we need to sort the data to have most recent data last in file
    data['Date'] = pd.to_datetime(data['Unix'], unit='ms')
    data['Date'] = data['Date'].dt.strftime('%Y-%m-%d %H:%M:%S')
    data.drop(columns=[f'Volume {sym_base}', 'tradecount', 'Symbol', 'Unix'], inplace=True)  # drop columns we dont need
    data.rename(columns={'Date': 'datetime', 'Open': 'open', 'High': 'high', 'Low': 'low', 'Close': 'close',
                         f'Volume {sym}': 'volume'}, inplace=True)

data.to_csv(new_filename, index=False)
    filepath_to_csv = f'{os.path.join(os.getcwd(), new_filename)}'
    print(f'CryptoDataDownload file written: {filepath_to_csv}')

return filepath_to_csv  # return our csv filepath for Basana to open

# The strategy implements the set of rules that define when to enter or exit a trade based on market conditions.
class SMA_Strategy(bs.TradingSignalSource):
    def __init__(self, dispatcher: bs.EventDispatcher, period: int):
        super().__init__(dispatcher)
        self.sma = SMA(period)
        self._values = (None, None)

async def on_bar_event(self, bar_event: bs.BarEvent):
        # Feed the technical indicator.
        value = float(bar_event.bar.close)
        self.sma.add_input_value(value)

# Keep a small window of values to check if there is a crossover.
        self._values = (self._values[-1], value)

# Is the indicator ready ?
        if len(self.sma) < 2:
            return

# Price crossed below SMA ?
        if self._values[-2] >= self.sma[-2] and self._values[-1] < self.sma[-1]:
            self.push(bs.TradingSignal(bar_event.when, bs.OrderOperation.SELL, bar_event.bar.pair))
        # Price crossed above SMA ?
        elif self._values[-2] <= self.sma[-2] and self._values[-1] > self.sma[-1]:
            self.push(bs.TradingSignal(bar_event.when, bs.OrderOperation.BUY, bar_event.bar.pair))

# The position manager is responsible for executing trades and managing positions.
class PositionManager:
    def __init__(self, exchange: backtesting_exchange.Exchange, position_amount: Decimal):
        assert position_amount > 0
        self._exchange = exchange
        self._position_amount = position_amount

async def on_trading_signal(self, trading_signal: bs.TradingSignal):
        logging.info("Trading signal: operation=%s pair=%s", trading_signal.operation, trading_signal.pair)
        try:
            # Calculate the order size.
            balances = await self._exchange.get_balances()
            if trading_signal.operation == bs.OrderOperation.BUY:
                _, ask = await self._exchange.get_bid_ask(trading_signal.pair)
                balance = balances[trading_signal.pair.quote_symbol]
                order_size = min(self._position_amount, balance.available) / ask
            else:
                balance = balances[trading_signal.pair.base_symbol]
                order_size = balance.available
            pair_info = await self._exchange.get_pair_info(trading_signal.pair)
            order_size = bs.truncate_decimal(order_size, pair_info.base_precision)
            if not order_size:
                return

logging.info(
                "Creating %s market order for %s: amount=%s", trading_signal.operation, trading_signal.pair, order_size
            )
            await self._exchange.create_market_order(trading_signal.operation, trading_signal.pair, order_size)
        except Exception as e:
            logging.error(e)

async def main():
    logging.basicConfig(level=logging.INFO, format="[%(asctime)s %(levelname)s] %(message)s")

event_dispatcher = bs.backtesting_dispatcher()
    pair = bs.Pair("BTC", "USD")
    exchange = backtesting_exchange.Exchange(
        event_dispatcher,
        initial_balances={"BTC": Decimal(0), "USD": Decimal(10000)}
    )
    exchange.set_pair_info(pair, bs.PairInfo(8, 2))

# Connect the strategy to the bar events from the exchange.
    strategy = SMA_Strategy(event_dispatcher, 15)
    exchange.subscribe_to_bar_events(pair, strategy.on_bar_event)

# Connect the position manager to the strategy signals.
    position_mgr = PositionManager(exchange, Decimal(1000))
    strategy.subscribe_to_trading_signals(position_mgr.on_trading_signal)

# Load bars from CSV files.
    exchange.add_bar_source(csv.BarSource(pair, f"{return_CDD_filepath(exchange='Binance', pair='BTC/USDT')}", "1d"))

# Setup the chart.
    chart = charts.LineCharts(exchange)
    chart.add_pair(pair)
    chart.add_pair_indicator("SMA", pair, charts.DataPointFromSequence(strategy.sma))
    chart.add_portfolio_value("USD")

# Run the backtest.
    await event_dispatcher.run()

# Log balances.
    balances = await exchange.get_balances()
    for currency, balance in balances.items():
        logging.info("%s balance: %s", currency, balance.available)

chart.show()

if __name__ == "__main__":
    asyncio.run(main())

Notice: Information contained herein is not and should not be construed as an offer, solicitation, or recommendation to buy or sell securities. The information has been obtained from sources we believe to be reliable; however no guarantee is made or implied with respect to its accuracy, timeliness, or completeness. Author does not own the any crypto currency discussed. The information and content are subject to change without notice. CryptoDataDownload and its affiliates do not provide investment, tax, legal or accounting advice.

This material has been prepared for informational purposes only and is the opinion of the author, and is not intended to provide, and should not be relied on for, investment, tax, legal, accounting advice. You should consult your own investment, tax, legal and accounting advisors before engaging in any transaction. All content published by CryptoDataDownload is not an endorsement whatsoever. CryptoDataDownload was not compensated to submit this article. Please also visit our Privacy policy; disclaimer; and terms and conditions page for further information.

THE PERFORMANCE OF TRADING SYSTEMS IS BASED ON THE USE OF COMPUTERIZED SYSTEM LOGIC. IT IS HYPOTHETICAL. PLEASE NOTE THE FOLLOWING DISCLAIMER. CFTC RULE 4.41: HYPOTHETICAL OR SIMULATED PERFORMANCE RESULTS HAVE CERTAIN LIMITATIONS. UNLIKE AN ACTUAL PERFORMANCE RECORD, SIMULATED RESULTS DO NOT REPRESENT ACTUAL TRADING. ALSO, SINCE THE TRADES HAVE NOT BEEN EXECUTED, THE RESULTS MAY HAVE UNDER-OR-OVER COMPENSATED FOR THE IMPACT, IF ANY, OF CERTAIN MARKET FACTORS, SUCH AS LACK OF LIQUIDITY. SIMULATED TRADING PROGRAMS IN GENERAL ARE ALSO SUBJECT TO THE FACT THAT THEY ARE DESIGNED WITH THE BENEFIT OF HINDSIGHT. NO REPRESENTATION IS BEING MADE THAT ANY ACCOUNT WILL OR IS LIKELY TO ACHIEVE PROFIT OR LOSSES SIMILAR TO THOSE SHOWN. U.S. GOVERNMENT REQUIRED DISCLAIMER: COMMODITY FUTURES TRADING COMMISSION. FUTURES AND OPTIONS TRADING HAS LARGE POTENTIAL REWARDS, BUT ALSO LARGE POTENTIAL RISK. YOU MUST BE AWARE OF THE RISKS AND BE WILLING TO ACCEPT THEM IN ORDER TO INVEST IN THE FUTURES AND OPTIONS MARKETS. DON’T TRADE WITH MONEY YOU CAN’T AFFORD TO LOSE. THIS IS NEITHER A SOLICITATION NOR AN OFFER TO BUY/SELL FUTURES OR OPTIONS. NO REPRESENTATION IS BEING MADE THAT ANY ACCOUNT WILL OR IS LIKELY TO ACHIEVE PROFITS OR LOSSES SIMILAR TO THOSE DISCUSSED ON THIS WEBSITE. THE PAST PERFORMANCE OF ANY TRADING SYSTEM OR METHODOLOGY IS NOT NECESSARILY INDICATIVE OF FUTURE RESULTS.