EMA Bot Example¶
This example demonstrates how to create a bot using the Exponential Moving Average (EMA) strategy. The bot implements a simple crossing moving average strategy using the 5 period EMA as the signal and the 20 period EMA as the base. When a crossing happens, the bot fills its position to the position limit in the respective direction and waits for a signal in the other direction. When this happens, it unloads its positions and loads in the new direction using the same position limit as the target.
Bot Implementation¶
```python import pandas as pd import logging import traceback import sys import time from traderion.bot import TraderionBot from traderion.dashboard import Dashboard, format_number import curses
Direction constants¶
BEARISH_CROSS, BULLISH_CROSS = (0, 1)
class CrossingEMAsBot(TraderionBot): """ The bot implements a simple crossing moving average strategy using the 5 period EMA as the signal and the 20 period EMA as the base.
When a crossing happens the bot fills its position to the position limit
in the respective direction and waits for a signal in the other direction.
When this happens it unloads its positions and loads in the new direction
using the same position limit as target.
In case we get the same values for the two EMAs from the initial price curve,
the bot will sell/buy when the signal drops below/rises above the base EMA.
"""
def __init__(self, username, password, room_id, loop_sleep=1, dashboard=None):
super().__init__(username, password, room_id, loop_sleep)
self.dashboard = dashboard
# Session attributes
self.position = self.api.get_position()
self.depth = self.api.get_eb_depth()
self.price_curve = [price for (time, price) in self.api.get_price_curve()]
self.room_parameters = self.api.get_room_parameters()
# Strategy attributes
self.ema_5 = None
self.ema_20 = None
self.last_cross = None
# Dashboard integration
self.ema_data = {
'small_ema': self.ema_5 if self.ema_5 is not None else 0.0000,
'big_ema': self.ema_20 if self.ema_20 is not None else 0.0000,
'last_cross': 'None',
'target_amount': self.position.get('limit', 0)
}
self.compute_emas()
# Add EMA panel to dashboard if available
try:
if self.dashboard is not None:
self.dashboard.update_core_data('status', 'Connected')
self.dashboard.add_custom_panel(self.get_dashboard_panel())
# Initialize dashboard with current position
self.dashboard.update_core_data('position', {
'amount': self.position.get('amount', 0),
'pnl': self.position.get('pnl', 0),
'rate': self.position.get('rate', 0)
})
# Initialize dashboard with market prices
market_prices = self.api.get_market_prices()
self.dashboard.update_core_data('market_prices', market_prices)
except Exception as e:
print(f"Error initializing dashboard: {e}")
import traceback
traceback.print_exc()
self.dashboard = None
def get_dashboard_panel(self):
def ema_panel(stdscr, y, x, height, width):
stdscr.addstr(y, x, "EMA Strategy", curses.A_BOLD)
stdscr.addstr(y + 1, x + 2, f"Small EMA: {self.ema_data.get('small_ema', 0.0000):10.4f}")
stdscr.addstr(y + 2, x + 2, f"Big EMA: {self.ema_data.get('big_ema', 0.0000):10.4f}")
stdscr.addstr(y + 3, x + 2, f"Last Cross: {self.ema_data.get('last_cross', 'N/A')}")
stdscr.addstr(y + 4, x + 2, f"Target Amount: {format_number(self.ema_data.get('target_amount', 0))}")
return ema_panel
def on_price_curve_change(self, old_price_curve, new_price_curve):
"""
Callback for price curve changes.
"""
self.price_curve = [price for (time, price) in self.api.get_price_curve()]
self.compute_emas()
if self.dashboard:
self.dashboard.add_market_event('Price Curve Change', "Price curve updated")
self.ema_data['small_ema'] = self.ema_5
self.ema_data['big_ema'] = self.ema_20
def on_position_change(self, old_position, new_position):
"""
Callback for position changes.
"""
self.position = new_position
if self.dashboard:
self.dashboard.update_core_data('position', {
'amount': new_position['amount'],
'pnl': new_position['pnl'],
'rate': new_position.get('rate', 0)
})
self.dashboard.add_performance_metric('Position Change', new_position['pnl'])
def on_eb_depth_change(self, old_depth, new_depth):
"""
Callback for electronic broker depth changes.
"""
self.depth = new_depth
if self.dashboard:
self.dashboard.add_market_event('EB Depth Change', "Depth updated")
# Update market prices in dashboard
if len(new_depth[0]) > 0 and len(new_depth[1]) > 0:
self.dashboard.update_core_data('market_prices', {
'bid': new_depth[0][0]['price'],
'ask': new_depth[1][0]['price'],
'open': self.dashboard.core_data['market_prices'].get('open', 0)
})
def main_loop(self):
"""
Main loop of the bot.
"""
self.run_strategy()
def compute_emas(self):
"""
Compute the EMAs based on the price curve.
"""
self.ema_5 = round(pd.Series(self.price_curve).ewm(span=5).mean().tolist()[-1],
self.room_parameters['price_decimals'])
self.ema_20 = round(pd.Series(self.price_curve).ewm(span=20).mean().tolist()[-1],
self.room_parameters['price_decimals'])
if self.dashboard:
self.ema_data['small_ema'] = self.ema_5
self.ema_data['big_ema'] = self.ema_20
def run_strategy(self):
"""
Run the EMA crossing strategy.
"""
cross_direction = self.get_new_cross_direction()
if cross_direction is not None:
self.last_cross = cross_direction
if self.dashboard:
cross_name = 'Bullish' if cross_direction == BULLISH_CROSS else 'Bearish'
self.ema_data['last_cross'] = cross_name
self.dashboard.add_bot_event('EMA Cross', f"{cross_name} cross detected")
self.dashboard.update_core_data('last_action', f"Loading on {cross_name} cross")
self.load(cross_direction)
def get_new_cross_direction(self):
"""
Determine if there's a new EMA cross.
"""
cross_direction = None
if self.last_cross in [BEARISH_CROSS, None] and self.ema_5 > self.ema_20:
cross_direction = BULLISH_CROSS
elif self.last_cross in [BULLISH_CROSS, None] and self.ema_5 < self.ema_20:
cross_direction = BEARISH_CROSS
return cross_direction
def load(self, direction):
"""
Load a position based on the cross direction.
"""
current_amount = self.position['amount'] / self.api.ticket_unit # short amount
target_amount = self.position['limit'] / self.api.ticket_unit # short amount, absolute
sgn = (-1, 1)[direction == BULLISH_CROSS]
remaining_amount = target_amount - sgn * current_amount
if remaining_amount < 0:
return
else:
while remaining_amount > 0 and direction == self.last_cross:
ticket = self.api.max_ticket \
if remaining_amount >= self.api.max_ticket else remaining_amount
if self.dashboard:
self.dashboard.update_core_data('last_action', f"Hitting {'BID' if direction == BULLISH_CROSS else 'ASK'} for {ticket}")
try:
result = self.api.hit_price(direction, ticket, self.depth[direction][0]['price'])
amount = result['amount']
remaining_amount -= amount
if self.dashboard:
self.dashboard.add_bot_event('Hit Success', f"{'BID' if direction == BULLISH_CROSS else 'ASK'} for {amount}")
except Exception as e:
if self.dashboard:
self.dashboard.add_bot_event('Hit Failure', str(e))
self