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