Skip to content

Tutorial

Tick

A Tick represent the value of an Epic at a given point in time.

import arrow

from estrade import Tick


def test_tick():
    now_tokyo = arrow.now("Asia/Tokyo")
    tick = Tick(datetime=now_tokyo, bid=99, ask=101, meta={"my_data": "data"})

    assert tick.bid == 99
    assert tick.ask == 101
    assert tick.datetime == now_tokyo
    assert tick.meta == {"my_data": "data"}

    assert tick.value == 100.0  # value represents the mean between bid and ask
    assert tick.spread == 2.0  # spread is the difference between bid and ask
    assert tick.datetime_utc == now_tokyo.to("UTC")

Epic

An Epic is a financial instrument.

import arrow
from dateutil import tz

from estrade import Epic, Tick


def test_create_epic():
    # GIVEN an epic timezoned on Paris(France) holding a max of 3 ticks in memory
    epic = Epic(timezone="Europe/Paris")

    # WHEN I add a tick to this epic in UTC
    tick = Tick(
        datetime=arrow.utcnow(),
        bid=100,
        ask=101,
    )
    epic.on_new_tick(tick)

    # THEN the tick is set as the Epic last tick
    assert epic.last_tick == tick

    # THEN the tick date was converted from UTC to the Epic timezone (Europe/Paris)
    assert tick.datetime.tzinfo == tz.gettz(epic.timezone)

Tick Provider

A BaseTickProvider is used to dispatch Tick to Epic instances.

import arrow

from estrade import BaseTickProvider, Epic, Tick


class MyTickProvider(BaseTickProvider):
    def run(self):
        # Generates 9 ticks
        for i in range(10):
            # create a new tick
            new_tick = Tick(
                datetime=arrow.utcnow(),
                bid=(i - 0.5),
                ask=(i + 0.5),
            )
            # find epic to attach the tick to
            tick_epic = self.get_epic_by_ref("MY_EPIC_CODE")
            # dispatch tick to epic
            tick_epic.on_new_tick(new_tick)


def test_tick_provider():
    # GIVEN an Epic
    epic = Epic(ref="MY_EPIC_CODE")

    # GIVEN an instance of the tick provider
    provider = MyTickProvider(epics=[epic])

    # WHEN I run the tick provider
    provider.run()

    # THEN 9 ticks are registered on epic
    assert epic.last_tick.value == 9

Trade Provider

A BaseTradeProvider create, update and close Trades.

Backtests

For backtests you do not have to worry about the Trade Provider. A TradeProviderBacktests is automatically added on Epic upon creation when no trade_provider param is provided.

The TradeProviderBacktests automatically set each Trade and TradeClose status to CONFIRMED.

import arrow

from estrade import Epic, Tick
from estrade.enums import TradeDirection, TransactionStatus


def test_trade_provider_backtest():
    # GIVEN an instance of our custom trade provider attached to an Epic
    epic = Epic(ref="MY_EPIC")

    # Add a tick to the epic
    tick = Tick(datetime=arrow.utcnow(), bid=99, ask=101)
    epic.on_new_tick(tick)

    # WHEN I create a new trade add open it with the Trade Provider
    trade = epic.open_trade(direction=TradeDirection.SELL, quantity=2)

    # THEN a new trade is created
    assert len(epic.trade_provider.trades) == 1

    # THEN the trade status is automatically set to CONFIRMED
    trade_in_provider = epic.trade_provider.trades[0]
    assert trade_in_provider.status == TransactionStatus.CONFIRMED

    # WHEN I close the trade
    epic.close_trade(trade=trade, quantity=1)

    # THEN a close of one quantity was created on the opened trade
    assert len(trade_in_provider.closes) == 1

    # THEN the trade close attributes where set
    trade_close_in_provider = trade_in_provider.closes[0]
    assert trade_close_in_provider.status == TransactionStatus.CONFIRMED

Live

When going live you want to specify how to open/update and close trades in a custom BaseTradeProvider.

import arrow

from estrade import BaseTradeProvider, Epic, Tick
from estrade.enums import TradeDirection, TransactionStatus


class MyTradeProvider(BaseTradeProvider):
    def open_trade_request(self, trade):
        # call your api to open a new trade
        # eg: response = requests.get("http://my_provider.com/open_trade", params={
        #     "epic": trade.epic.ref,
        #     "direction": trade.direction,
        #     "quantity": trade.quantity,
        #     ...
        # })

        # update trade with your provider response
        trade.meta["provider_id"] = 123
        trade.status = TransactionStatus.PENDING

        return trade

    def close_trade_request(self, trade_close):
        # call your api to close a trade.
        # eg: response = requests.get("http://my_provider.com/close_trade", params={
        #     "epic": trade_close.trade.epic.ref,
        #     "quantity": trade_close.quantity,
        #     ...
        # })

        # update trade with your provider response
        trade_close.meta["provider_close_id"] = 123
        trade_close.status = TransactionStatus.REQUIRED

        return trade_close


def test_trade_provider_custom():
    # GIVEN an instance of our custom trade provider attached to an Epic
    trade_provider = MyTradeProvider()
    epic = Epic(ref="MY_EPIC", trade_provider=trade_provider)

    # Add a tick to the epic
    tick = Tick(datetime=arrow.utcnow(), bid=99, ask=101)
    epic.on_new_tick(tick)

    # WHEN I create a new trade add open it with the Trade Provider
    trade = epic.open_trade(direction=TradeDirection.SELL, quantity=2)

    # THEN a new trade is created
    assert len(epic.trade_provider.trades) == 1

    # THEN the trade attribute were updated by the trade provider
    trade_in_provider = epic.trade_provider.trades[0]
    assert trade_in_provider.status == TransactionStatus.PENDING
    assert trade_in_provider.meta["provider_id"] == 123

    # WHEN I close the trade
    epic.close_trade(trade=trade, quantity=1)

    # THEN a close of one quantity was created on the opened trade
    assert len(trade_in_provider.closes) == 1

    # THEN the trade close attributes where set
    trade_close_in_provider = trade_in_provider.closes[0]
    assert trade_close_in_provider.status == TransactionStatus.REQUIRED
    assert trade_close_in_provider.meta["provider_close_id"] == 123

Strategy

A Strategy is the object where you connect everything to open/close your trades based on the epic, its indicators etc.

On your Strategy define a method that will be triggered on every new Tick received by one of its Epics.

import arrow

from estrade import BaseStrategy, Epic, Tick
from estrade.enums import TradeDirection


class MyStrategy(BaseStrategy):
    """
    Define a basic Strategy.

    Open a trade when the tick value is 100 and close all opened trades otherwise.
    """

    def on_every_tick(self, epic: "Epic") -> None:
        # on every new tick received this method is triggered.
        if epic.last_tick.value == 100:
            # create a new trade
            self.open_trade(epic=epic, quantity=5, direction=TradeDirection.BUY)
        else:
            # close all opened trades
            self.close_opened_trades()


def test_basic_strategy():
    epic = Epic(ref="MY_EPIC_CODE")

    # GIVEN a instance of MyStrategy on an Epic
    strategy = MyStrategy()
    epic.add_strategy(strategy)

    # WHEN a tick of value 100 is received
    tick = Tick(arrow.utcnow(), 99, 101)
    epic.on_new_tick(tick)
    # THEN a trade is created
    assert len(epic.trade_provider.opened_trades) == 1

    # WHEN a tick of value != 100 is received
    tick = Tick(arrow.utcnow(), 101, 103)
    epic.on_new_tick(tick)
    # THEN the previously opened trade is closed
    assert len(epic.trade_provider.opened_trades) == 0

Note

here we used the on_every_tick_method refer to Strategy to view other methods available.

Graphical

Frame Set & Frames

A FrameSet is registered on Epic and generate (or update) Frames on every Tick received.

import arrow

from estrade import BaseTickProvider, Epic, Tick
from estrade.enums import Unit
from estrade.graph import FrameSet


class MyTickProvider(BaseTickProvider):
    def run(self):
        # Generate 10 ticks (1 tick every minute from 2020-01-01 00:00:12)
        dt = arrow.get("2020-01-01 00:00:12")
        for i in range(10):
            dt = dt.shift(minutes=1)
            # create a new tick
            new_tick = Tick(
                datetime=dt,
                bid=(i - 0.5),
                ask=(i + 0.5),
            )
            # find epic to attach the tick to
            tick_epic = self.get_epic_by_ref("MY_EPIC_CODE")
            # dispatch tick to epic
            tick_epic.on_new_tick(new_tick)


def test_frame_sets():
    # GIVEN A FrameSet grouping ticks by 5
    ut5ticks = FrameSet(ref="UT5TICKS", unit=Unit.TICK, unit_quantity=5)

    # GIVEN a FrameSet grouping ticks by 3 minutes
    ut3mn = FrameSet(ref="UT3MN", unit=Unit.MINUTE, unit_quantity=3)

    # GIVEN an Epic holding both frame sets
    epic = Epic(ref="MY_EPIC_CODE")
    epic.add_frame_set(ut5ticks)
    epic.add_frame_set(ut3mn)

    # GIVEN an instance of my provider
    provider = MyTickProvider(epics=[epic])

    # WHEN I run the tick provider
    provider.run()

    # THEN in UT 5 ticks
    # 2 frames were created
    assert len(epic.frame_sets["UT5TICKS"].frames) == 2

    # THEN in UT 3 minutes
    # 4 frames where created
    assert len(epic.frame_sets["UT3MN"].frames) == 4
    # the period_start of each frame is rounded properly
    assert [f.period_start for f in epic.frame_sets["UT3MN"].frames] == [
        arrow.get("2020-01-01 00:00:00"),
        arrow.get("2020-01-01 00:03:00"),
        arrow.get("2020-01-01 00:06:00"),
        arrow.get("2020-01-01 00:09:00"),
    ]

Indicators

Indicators are generators of Indicator Values.

On every Frame created by a FrameSet indicators values values are created on the new created Frame.

Candle Sets

import arrow

from estrade import BaseTickProvider, Epic, Tick
from estrade.enums import CandleType, Unit
from estrade.graph import CandleSet, FrameSet


class MyTickProvider(BaseTickProvider):
    def run(self):
        # Generate 8 ticks (1 tick every minute from 2020-01-01 00:00:12)
        dt = arrow.get("2020-01-01 00:00:00")
        for i in [12, 13.4, 8.2, 10, 12.6, 9.4, 14.7, 11.8]:
            # create a new tick
            new_tick = Tick(
                datetime=dt,
                bid=(i - 0.5),
                ask=(i + 0.5),
            )
            # find epic to attach the tick to
            tick_epic = self.get_epic_by_ref("MY_EPIC_CODE")
            # dispatch tick to epic
            tick_epic.on_new_tick(new_tick)
            dt = dt.shift(minutes=1)


def test_indicator_candle_set():
    # GIVEN A FrameSet in UT4 minutes
    ut4mn = FrameSet(ref="UT4MN", unit=Unit.MINUTE, unit_quantity=4)

    # GIVEN a Classic candle set indicator set on the UT4MN FrameSet
    candle_set_classic = CandleSet(ref="UT4CS", candle_type=CandleType.CLASSIC)
    ut4mn.add_indicator(candle_set_classic)

    # GIVEN a HEIKIN ASHI candle set indicator set on the UT4MN FrameSet
    candle_set_ha = CandleSet(ref="UT4HA", candle_type=CandleType.HEIKIN_ASHI)
    ut4mn.add_indicator(candle_set_ha)

    # GIVEN an Epic holding the UT4MN FrameSet
    epic = Epic(ref="MY_EPIC_CODE")
    epic.add_frame_set(ut4mn)

    # GIVEN an instance of my provider
    provider = MyTickProvider(epics=[epic])

    # WHEN I run the tick provider
    provider.run()

    # THEN the last classic candle values match the input ticks
    first_frame_candle_classic = epic.get_indicator_value(
        frame_set_ref="UT4MN", indicator_ref="UT4CS", offset=1
    )
    assert first_frame_candle_classic.open == 12
    assert first_frame_candle_classic.low == 8.2
    assert first_frame_candle_classic.high == 13.4
    assert first_frame_candle_classic.last == 10

    last_frame_candle_classic = epic.get_indicator_value(
        frame_set_ref="UT4MN", indicator_ref="UT4CS"
    )
    assert last_frame_candle_classic.open == 12.6
    assert last_frame_candle_classic.low == 9.4
    assert last_frame_candle_classic.high == 14.7
    assert last_frame_candle_classic.last == 11.8

    # THEN the HEIKIN ASHI candle values match the input ticks
    first_frame_candle_ha = epic.get_indicator_value(
        frame_set_ref="UT4MN", indicator_ref="UT4HA", offset=1
    )
    assert first_frame_candle_ha.open == 12
    assert first_frame_candle_ha.low == 8.2
    assert first_frame_candle_ha.high == 13.4
    assert first_frame_candle_ha.last == 10.9

    last_frame_candle_ha = epic.get_indicator_value(
        frame_set_ref="UT4MN", indicator_ref="UT4HA"
    )
    assert last_frame_candle_ha.open == 11.45
    assert last_frame_candle_ha.low == 9.4
    assert last_frame_candle_ha.high == 14.7
    assert last_frame_candle_ha.last == 12.12

Simple Moving Average

import arrow

from estrade import BaseTickProvider, Epic, Tick
from estrade.enums import Unit
from estrade.graph import FrameSet, SimpleMovingAverage


class MyTickProvider(BaseTickProvider):
    def run(self):
        # Generate 8 ticks (1 tick every minute from 2020-01-01 00:00:12)
        dt = arrow.get("2020-01-01 00:00:00")
        for i in [12, 13.4, 8.2, 10, 12.6, 9.4, 14.7, 11.8]:
            # create a new tick
            new_tick = Tick(
                datetime=dt,
                bid=(i - 0.5),
                ask=(i + 0.5),
            )
            # find epic to attach the tick to
            tick_epic = self.get_epic_by_ref("MY_EPIC_CODE")
            # dispatch tick to epic
            tick_epic.on_new_tick(new_tick)
            dt = dt.shift(minutes=1)


def test_simple_moving_average():
    # GIVEN A FrameSet in UT1 minutes
    ut1mn = FrameSet(ref="UT1MN", unit=Unit.MINUTE, unit_quantity=1)

    # GIVEN a SMA with a max number of periods of 10
    sma = SimpleMovingAverage(ref="SMA", max_periods=10)
    ut1mn.add_indicator(sma)

    # GIVEN an Epic holding the UT4MN FrameSet
    epic = Epic(ref="MY_EPIC_CODE")
    epic.add_frame_set(ut1mn)

    # GIVEN an instance of my provider
    provider = MyTickProvider(epics=[epic])

    # WHEN I run the tick provider
    provider.run()

    # THEN the SMA on the last 3 periods is properly calculated
    sma3 = epic.get_indicator_value(
        frame_set_ref="UT1MN",
        indicator_ref="SMA",
    ).get_value(periods=3)
    assert sma3 == round((9.4 + 14.7 + 11.8) / 3, 2)

    # WHEN I update the epic with a new tick, the SMA is updated
    new_tick = Tick(datetime=epic.last_tick.datetime, bid=15.6, ask=17.6)
    epic.on_new_tick(new_tick)

    # THEN the SMA on the last 3 periods is properly updated
    sma3 = epic.get_indicator_value(
        frame_set_ref="UT1MN",
        indicator_ref="SMA",
    ).get_value(periods=3)
    assert sma3 == round((9.4 + 14.7 + 16.6) / 3, 2)

Other Indicators

Reporting

Csv reporting

Estrade includes a very basic CSV reporting functionality that build a CSV file from an Epic instance.

see: ReportingCSV