Composite Strategy Bot Example¶
This example demonstrates how to create a bot that combines multiple trading strategies. The bot uses the CrossingEMAs, Macd, Rsi, Clients, and Cover strategies to make trading decisions. Each strategy is run in sequence, and the bot updates the dashboard with the status and performance of each strategy.
Bot Implementation¶
import random
import time
import threading
import logging
import traceback
import sys
from traderion.bot import TraderionBot
from traderion.strategies.clients import Clients
from traderion.strategies.ema import CrossingEMAs
from traderion.strategies.macd import Macd
from traderion.strategies.rsi import Rsi
from traderion.strategies.cover import Cover
from traderion.dashboard import Dashboard
import curses
# Direction constants
BEARISH_CROSS, BULLISH_CROSS = (0, 1)
# Strategy constants
CROSSING_EMA, MACD, RSI, CLIENTS, COVER = range(5)
STRATEGY_MAP = {
CROSSING_EMA: CrossingEMAs,
MACD: Macd,
RSI: Rsi,
CLIENTS: Clients,
COVER: Cover,
}
class CompositeBot(TraderionBot):
"""
A bot that combines multiple trading strategies.
"""
def __init__(self, username, password, room_id, strategy_map=None, loop_sleep=1, dashboard=None):
super().__init__(username, password, room_id, loop_sleep)
self.strategies = []
self.dashboard = dashboard
self.strategies_data = {}
if strategy_map is None or not isinstance(strategy_map, dict):
raise Exception("Provide a valid strategy configuration.")
for strategy_type, options in strategy_map.items():
strategy_ctor = STRATEGY_MAP[strategy_type]
strategy = strategy_ctor(self.api, options)
self.strategies.append(strategy)
strategy_name = strategy_ctor.__name__
self.strategies_data[strategy_name] = "Waiting for signals"
if self.dashboard:
self.dashboard.update_core_data('status', 'Connecting')
self.dashboard.add_custom_panel(self.get_dashboard_panel())
self.dashboard.update_core_data('position', {'amount': 0, 'pnl': 0, 'rate': 0})
self.dashboard.update_core_data('market_prices', {'bid': 0.0000, 'ask': 0.0000, 'open': 0.0000})
def get_dashboard_panel(self):
def composite_panel(stdscr, y, x, height, width):
stdscr.addstr(y, x, "Composite Strategies", curses.A_BOLD)
line = 1
for strategy_name, strategy_info in self.strategies_data.items():
if line < height - 1:
stdscr.addstr(y + line, x + 2, f"{strategy_name}: {strategy_info}")
line += 1
return composite_panel
def update_dashboard_data(self):
""" Updates position and market prices on the dashboard """
if not self.dashboard:
return
# Update market prices
market_prices = self.api.get_market_prices()
self.dashboard.update_core_data('market_prices', market_prices)
# Update position
position = self.api.get_position()
self.dashboard.update_core_data('position', {
'amount': position['amount'],
'pnl': position['pnl'],
'rate': position.get('rate', 0)
})
def on_position_change(self, old_position, new_position):
""" Callback for position changes """
for strategy in self.strategies:
strategy.on_position_change(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 """
for strategy in self.strategies:
strategy.on_eb_depth_change(new_depth)
if self.dashboard and 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 on_room_status_change(self, old_status, new_status):
""" Callback for room status updates """
if self.dashboard:
status_text = new_status
self.dashboard.update_core_data('status', status_text)
if old_status != status_text:
self.dashboard.add_bot_event('Room Status Change', f"From {old_status} to {status_text}")
def main_loop(self):
""" Main bot loop """
for strategy in self.strategies:
strategy.run()
if self.dashboard:
strategy_name = strategy.__class__.__name__
strategy_info = strategy.log_info()
self.strategies_data[strategy_name] = strategy_info
self.dashboard.update_core_data('last_action', f"Running {strategy_name}")
self.dashboard.add_bot_event('Strategy Executed', f"Ran {strategy_name}")
def update_dashboard(self):
""" Periodically updates strategy data on the dashboard """
if not self.dashboard:
return
while True:
for strategy in self.strategies:
strategy_name = strategy.__class__.__name__
self.strategies_data[strategy_name] = strategy.log_info()
time.sleep(1)
def run_bot(dashboard):
""" Run the bot with the dashboard """
logger = logging.getLogger(__name__)
try:
dashboard.update_core_data('status', 'Starting')
dashboard.update_core_data('last_action', 'Initializing bot')
print("Creating bot...")
str_map = {
CROSSING_EMA: {'target_amount': 10000, 'small_ema_period': 5, 'big_ema_period': 20},
RSI: {'target_amount': 10000, 'period': 14, 'overbought_threshold': 70, 'oversold_threshold': 30},
CLIENTS: {'client_quote_spread': lambda: random.randint(-5, 4), 'own_quote_spread': lambda: -22},
COVER: {'position_limit': 40000},
}
bot = CompositeBot('vc_trainee1', 'traderion', 19311, str_map, dashboard=dashboard)
print("Getting room status...")
room_status = bot.api.room_status
dashboard.update_core_data('status', room_status)
dashboard.add_bot_event('Room Status', f"Current status: {room_status}")
if not bot.api.is_playing:
dashboard.add_bot_event('Warning', f"Room not playing. Status: {room_status}")
print("Getting market prices...")
try:
bot.update_dashboard_data()
except Exception as e:
logger.error(f"Error fetching market data: {e}")
dashboard.add_bot_event('Error', f"Failed to fetch market data: {e}")
# Start background dashboard updates
update_thread = threading.Thread(target=bot.update_dashboard)
update_thread.daemon = True
update_thread.start()
print("Starting bot...")
dashboard.add_bot_event('Bot Starting', 'Running main loop')
bot.run()
except Exception as e:
print(f"CRITICAL ERROR: {e}")
print(traceback.format_exc())
logger.critical(f"Bot initialization failed: {e}")
dashboard.update_core_data('status', f"ERROR: {e}")
time.sleep(5)
sys.exit(1)
if __name__ == '__main__':
"""
Entry point for the bot.
"""
dashboard = Dashboard()
dashboard.run_dashboard(run_bot, (dashboard,))
Running the Bot¶
if __name__ == "__main__":
dashboard = Dashboard()
dashboard.run_dashboard(run_bot, (dashboard,))
Detailed Explanation¶
Initialization¶
__init__Method:- The bot initializes with the provided
username,password, androom_id. - It sets up multiple strategies using the
strategy_mapparameter, which is a dictionary mapping strategy types to their configuration options. - The bot initializes the dashboard and adds a custom panel to display the status of each strategy.
- It also initializes the position and market prices on the dashboard.
Handling Position Changes¶
on_position_changeMethod:- This method is a callback that handles position changes.
- It updates the position for each strategy and updates the dashboard with the new position data.
Handling Electronic Broker Depth Changes¶
on_eb_depth_changeMethod:- This method is a callback that handles electronic broker depth changes.
- It updates the depth for each strategy and updates the