Quickstart¶

Ready to get started ? This page gives a good introduction on how to get started with Basana.

Installation¶

Basana requires Python 3.8.1 or above and to install the package you can use the following command:

(.venv) $ pip install basana[charts]

As mentioned before, technical indicators are not included and the examples that follow take advantage of TALIpp that you can install using the following command:

(.venv) $ pip install talipp

Backtesting¶

The examples that follow are structured like this:

A trading strategy that implements the set of rules that define when to enter or exit a trade based on market conditions.
A position manager that is responsible for executing trades and managing positions. It receives trading signals from the strategy and submits orders to the exchange or broker.

Note

The way the examples are structured is just one way to do it. You’re free to structure the code in any other way.

The strategy that we’re going to use for backtesting is based on Bollinger Bands and the purpose of this example is just to give you an overview on how to connect the different pieces together.

At a high level this is how this strategy will work:

There are two types of events taking place in this example. Bars (OHLC) coming from the exchange and trading signals.
When a new bar is received by the strategy, a technical indicator will be fed using the bar’s closing price. If the technical indicator is ready we’ll check if the price moved below the lower band or if the price moved above upper band, and we’ll generate a buy or sell trading signal.
When a trading signal is received by the position manager, a buy or sell market order will be submitted to the exchange in order to open or close a position.

We’ll be using market orders to keep this example short, but you’ll probably want to use limit orders when writing your own position managers.

The first thing we’ll need in order to execute the backtest is historical data. Use the following command to download daily bars from Binance for year 2021:

(.venv) $ python -m basana.external.binance.tools.download_bars -c BTC/USDT -p 1d -s 2021-01-01 -e 2021-12-31 -o binance_btcusdt_day.csv

Next, save the following strategy code as bbands.py:

from talipp.indicators import BB

import basana as bs


# Strategy based on Bollinger Bands: https://www.investopedia.com/articles/trading/07/bollinger.asp
class Strategy(bs.TradingSignalSource):
    def __init__(self, dispatcher: bs.EventDispatcher, period: int, std_dev_multiplier: float):
        super().__init__(dispatcher)
        self.bb = BB(period, std_dev_multiplier)
        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.bb.add(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.bb) < 2 or self.bb[-2] is None:
            return

        # Price moved below lower band ?
        if self._values[-2] >= self.bb[-2].lb and self._values[-1] < self.bb[-1].lb:
            self.push(bs.TradingSignal(bar_event.when, bs.OrderOperation.BUY, bar_event.bar.pair))
        # Price moved above upper band ?
        elif self._values[-2] <= self.bb[-2].ub and self._values[-1] > self.bb[-1].ub:
            self.push(bs.TradingSignal(bar_event.when, bs.OrderOperation.SELL, bar_event.bar.pair))

and the following code as backtesting_bbands.py:

from decimal import Decimal
import asyncio
import logging

from basana.backtesting import charts
from basana.external.binance import csv
import basana as bs
import basana.backtesting.exchange as backtesting_exchange
import bbands


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", "USDT")
    exchange = backtesting_exchange.Exchange(
        event_dispatcher,
        initial_balances={"BTC": Decimal(0), "USDT": Decimal(10000)}
    )
    exchange.set_pair_info(pair, bs.PairInfo(8, 2))

    # Connect the strategy to the bar events from the exchange.
    strategy = bbands.Strategy(event_dispatcher, 10, 1.5)
    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, "binance_btcusdt_day.csv", "1d"))

    # Setup chart.
    chart = charts.LineCharts(exchange)
    chart.add_pair(pair)
    chart.add_pair_indicator(
        "Upper", pair, lambda _: strategy.bb[-1].ub if len(strategy.bb) and strategy.bb[-1] else None
    )
    chart.add_pair_indicator(
        "Central", pair, lambda _: strategy.bb[-1].cb if len(strategy.bb) and strategy.bb[-1] else None
    )
    chart.add_pair_indicator(
        "Lower", pair, lambda _: strategy.bb[-1].lb if len(strategy.bb) and strategy.bb[-1] else None
    )
    chart.add_portfolio_value("USDT")

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

and execute the backtest like this:

(.venv) $ python backtesting_bbands.py

A chart similar to this one should open in a browser:

Live trading¶

The strategy that we’re going to use for live trading is the exact same one that we used for backtesting, but instead of using a backtesting exchange we’ll use Binance crypto currency exchange.

Note

The examples provided are for educational purposes only. If you decide to execute them with real credentials you are doing that at your own risk.

If you decide to move forward, save the following code as binance_bbands.py and update your api_key and api_secret:

from decimal import Decimal
import asyncio
import logging

from basana.external.binance import exchange as binance_exchange
import basana as bs
import bbands


class PositionManager:
    def __init__(self, exchange: binance_exchange.Exchange, position_amount: Decimal):
        assert position_amount > 0
        self._exchange = exchange
        self._position_amount = position_amount

    async def calculate_price(self, trading_signal: bs.TradingSignal):
        bid, ask = await self._exchange.get_bid_ask(trading_signal.pair)
        return {
            bs.OrderOperation.BUY: ask,
            bs.OrderOperation.SELL: bid,
        }[trading_signal.operation]

    async def cancel_open_orders(self, pair: bs.Pair, order_operation: bs.OrderOperation):
        await asyncio.gather(*[
            self._exchange.spot_account.cancel_order(pair, order_id=open_order.id)
            for open_order in await self._exchange.spot_account.get_open_orders(pair)
            if open_order.operation == order_operation
        ])

    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:
            # Cancel any open orders in the opposite direction.
            await self.cancel_open_orders(
                trading_signal.pair,
                bs.OrderOperation.BUY if trading_signal.operation == bs.OrderOperation.SELL
                else bs.OrderOperation.SELL
            )

            # Calculate the order price and size.
            balances, price, pair_info = await asyncio.gather(
                self._exchange.spot_account.get_balances(),
                self.calculate_price(trading_signal),
                self._exchange.get_pair_info(trading_signal.pair)
            )
            if trading_signal.operation == bs.OrderOperation.BUY:
                balance = balances[trading_signal.pair.quote_symbol]
                order_size = min(balance.available, self._position_amount) / price
            else:
                balance = balances[trading_signal.pair.base_symbol]
                order_size = balance.available
            order_size = bs.truncate_decimal(order_size, pair_info.base_precision)
            if not order_size:
                return

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

    async def on_bar_event(self, bar_event: bs.BarEvent):
        logging.info(
            "Bar event: pair=%s open=%s high=%s low=%s close=%s volume=%s",
            bar_event.bar.pair, bar_event.bar.open, bar_event.bar.high, bar_event.bar.low, bar_event.bar.close,
            bar_event.bar.volume
        )


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

    event_dispatcher = bs.realtime_dispatcher()
    api_key = "YOUR_API_KEY"
    api_secret = "YOUR_API_SECRET"
    exchange = binance_exchange.Exchange(event_dispatcher, api_key=api_key, api_secret=api_secret)
    position_mgr = PositionManager(exchange, Decimal(30))

    pairs = [
        bs.Pair("BTC", "USDT"),
        bs.Pair("ETH", "USDT"),
    ]
    for pair in pairs:
        # Connect the strategy to the bar events from the exchange.
        strategy = bbands.Strategy(event_dispatcher, 20, 1.5)
        exchange.subscribe_to_bar_events(pair, "1m", strategy.on_bar_event)

        # Connect the position manager to the strategy signals and to bar events just for logging.
        strategy.subscribe_to_trading_signals(position_mgr.on_trading_signal)
        exchange.subscribe_to_bar_events(pair, "1m", position_mgr.on_bar_event)

    await event_dispatcher.run()


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

Next, start live trading using the following command:

(.venv) $ python binance_bbands.py

Next steps¶

The examples presented here, and many others, can be found at the examples folder at GitHub.

Quickstart¶

Installation¶

Backtesting¶

Live trading¶

Next steps¶

Basana

Navigation

Related Topics