Skip to content

Best Practices for Traderion Bot Development

This guide outlines best practices for developing robust, maintainable, and efficient trading bots with the Traderion framework.

Architecture and Design

graph TD
    A[Bot Architecture] --> B[Modular Design]
    A --> C[Separation of Concerns]
    A --> D[Event-Driven Approach]

    B --> B1[Strategy Modules]
    B --> B2[Risk Management Modules]
    B --> B3[Monitoring Modules]

    C --> C1[Data Processing]
    C --> C2[Decision Logic]
    C --> C3[Execution Logic]

    D --> D1[Event Handlers]
    D --> D2[Callbacks]
    D --> D3[State Management]

    style A fill:#f5f5f5,stroke:#333,stroke-width:2px
    style B fill:#d5e8d4,stroke:#333
    style C fill:#d4f1f9,stroke:#333
    style D fill:#ffe6cc,stroke:#333

Modular Design Principles

  1. Single Responsibility Principle
  2. Each class or module should have one specific responsibility

  3. Strategies should focus on trading logic, not infrastructure concerns

  4. Dependency Injection

  5. Pass dependencies (like API clients) to classes rather than creating them internally

  6. Makes testing and configuration easier

  7. Composition Over Inheritance

  8. Prefer composing functionality from smaller components
  9. Use strategies as modular components rather than deep inheritance hierarchies

Example: Well-Structured Bot

class WellStructuredBot(TraderionBot):
    def __init__(self, username, password, room_id):
        super().__init__(username, password, room_id)

        # Strategy components
        self.signal_generator = SignalGenerator()
        self.risk_manager = RiskManager(max_position=50000)
        self.position_manager = PositionManager(self.api)
        self.monitor = PerformanceMonitor()

        # State
        self.price_history = []
        self.current_position = None

    def on_market_price_change(self, old_prices, new_prices):
        # Update state
        self.price_history.append(new_prices)
        self.current_position = self.api.get_position()

        # Log event
        self.monitor.log_price_change(old_prices, new_prices)

    def main_loop(self):
        # Generate signal
        signal = self.signal_generator.generate_signal(self.price_history)

        # Apply risk management
        adjusted_signal = self.risk_manager.adjust_signal(
            signal, 
            self.current_position
        )

        # Execute trade if needed
        if adjusted_signal != 0:
            self.position_manager.execute_signal(adjusted_signal)
            self.monitor.log_trade(adjusted_signal)

Code Organization

Project Structure

traderion_bot/
├── main.py                  # Entry point
├── config.py                # Configuration
├── bots/
│   ├── __init__.py
│   ├── base_bot.py          # Base bot implementation
│   ├── ema_bot.py           # EMA strategy bot
│   └── composite_bot.py     # Multi-strategy bot
├── strategies/
│   ├── __init__.py
│   ├── base.py              # Base strategy class
│   ├── technical/           # Technical analysis strategies
│   │   ├── __init__.py
│   │   ├── ema.py
│   │   ├── macd.py
│   │   └── rsi.py
│   └── client/              # Client interaction strategies
│       ├── __init__.py
│       └── market_maker.py
├── utils/
│   ├── __init__.py
│   ├── logging.py           # Logging utilities
│   └── indicators.py        # Technical indicators
└── tests/                   # Unit tests
    ├── __init__.py
    ├── test_strategies.py
    └── test_indicators.py

Coding Standards

  1. Follow PEP 8 Style Guide
  2. Consistent indentation (4 spaces)
  3. Maximum line length of 79 characters
  4. Descriptive variable names

  5. Proper docstrings

  6. Use Type Hints 

    def calculate_position_size(price: float, volatility: float) -> int:
    """
    Calculate position size based on price and volatility.

    Args:
        price: Current market price
        volatility: Market volatility (standard deviation)

    Returns:
        Position size in number of shares/contracts
    """
    return int(10000 / (price * volatility))

  7. Separate Configuration from Code 

    # config.py
STRATEGY_PARAMS = {
    'ema': {
        'small_period': 5,
        'large_period': 20,
        'target_amount': 10000
    },
    'rsi': {
        'period': 14,
        'overbought': 70,
        'oversold': 30
    }
}

# bot.py
from config import STRATEGY_PARAMS

class MyBot(TraderionBot):
    def __init__(self, username, password, room_id):
        super().__init__(username, password, room_id)
        self.strategy = EMA(self.api, STRATEGY_PARAMS['ema'])

Error Handling and Resilience

Robust Error Handling

from traderion.exceptions import LoginError, RoomNotFound, ServerError, ValidationError, TradingError

def execute_trade(self, direction, amount, price):
    try:
        # Attempt to execute trade
        result = self.api.hit_price(direction, amount, price)
        logger.info(f"Trade executed: {direction}, {amount}, {price}")
        return result
    except TradingError as e:
        # Handle specific trading errors
        logger.warning(f"Trading error: {e}")
        self.handle_trading_error(e)
    except ServerError as e:
        # Handle server errors
        logger.error(f"Server error: {e}")
        self.retry_queue.append((direction, amount, price))
    except Exception as e:
        # Catch-all for unexpected errors
        logger.critical(f"Unexpected error: {e}", exc_info=True)
        # Implement fallback behavior
    return None

Graceful Degradation

class ResilientBot(TraderionBot):
    def __init__(self, username, password, room_id):
        super().__init__(username, password, room_id)
        self.fallback_mode = False
        self.retry_queue = []

    def main_loop(self):
        # Check if we should be in fallback mode
        if self.fallback_mode:
            self.execute_fallback_strategy()
            return

        try:
            # Normal operation
            self.execute_primary_strategy()
        except Exception as e:
            logger.error(f"Switching to fallback mode due to: {e}")
            self.fallback_mode = True
            self.execute_fallback_strategy()

    def execute_fallback_strategy(self):
        """Simple fallback strategy that just maintains current position."""
        # Don't take new positions, just manage existing ones
        position = self.api.get_position()
        if abs(position['amount']) > 0:
            # Check if we need to reduce risk
            if position['pnl_percentage'] < -0.05:  # 5% loss
                self.reduce_position(position)

Comprehensive Logging

Structured Logging

import logging
import json
from datetime import datetime

class StructuredLogger:
    def __init__(self, bot_name, log_file=None):
        self.bot_name = bot_name

        # Configure logger
        self.logger = logging.getLogger(bot_name)
        self.logger.setLevel(logging.INFO)

        # Console handler
        console = logging.StreamHandler()
        console.setLevel(logging.INFO)

        # File handler (optional)
        if log_file:
            file_handler = logging.FileHandler(log_file)
            file_handler.setLevel(logging.DEBUG)
            self.logger.addHandler(file_handler)

        self.logger.addHandler(console)

    def log_event(self, event_type, data):
        """Log a structured event."""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "bot": self.bot_name,
            "event_type": event_type,
            "data": data
        }
        self.logger.info(json.dumps(log_entry))

    def log_trade(self, direction, amount, price, reason=None):
        """Log a trade event."""
        self.log_event("trade", {
            "direction": "buy" if direction == 1 else "sell",
            "amount": amount,
            "price": price,
            "reason": reason
        })

    def log_position(self, position):
        """Log current position."""
        self.log_event("position", {
            "amount": position["amount"],
            "rate": position["rate"],
            "pnl": position["pnl"],
            "pnl_percentage": position["pnl_percentage"]
        })

    def log_error(self, error_type, message, exception=None):
        """Log an error."""
        error_data = {
            "type": error_type,
            "message": message
        }
        if exception:
            error_data["exception"] = str(exception)
        self.log_event("error", error_data)

Usage Example

class LoggingBot(TraderionBot):
    def __init__(self, username, password, room_id):
        super().__init__(username, password, room_id)
        self.logger = StructuredLogger("EMABot", "ema_bot.log")

    def on_market_price_change(self, old_prices, new_prices):
        self.logger.log_event("price_change", {
            "old": old_prices,
            "new": new_prices,
            "delta": {
                "bid": new_prices["bid"] - old_prices["bid"],
                "ask": new_prices["ask"] - old_prices["ask"]
            }
        })

    def on_position_change(self, old_position, new_position):
        self.logger.log_position(new_position)

    def log_info(self):
        while True:
            position = self.api.get_position()
            self.logger.log_position(position)
            time.sleep(60)  # Log position every minute

Testing Strategies

flowchart TD
    A[Testing Strategy] --> B[Unit Tests]
    A --> C[Integration Tests]
    A --> D[Simulation Tests]

    B --> B1[Test Indicators]
    B --> B2[Test Signal Generation]
    B --> B3[Test Risk Management]

    C --> C1[Test Bot with Mock API]
    C --> C2[Test Strategy Combinations]

    D --> D1[Test with Historical Data]
    D --> D2[Test in Simulator]

    style A fill:#f5f5f5,stroke:#333,stroke-width:2px
    style B fill:#d5e8d4,stroke:#333
    style C fill:#d4f1f9,stroke:#333
    style D fill:#ffe6cc,stroke:#333

Unit Testing

import unittest
from unittest.mock import Mock, patch
from traderion.strategies.ema import CrossingEMAs

class TestEMAStrategy(unittest.TestCase):
    def setUp(self):
        # Create mock API
        self.mock_api = Mock()

        # Configure mock to return test data
        self.mock_api.get_eb_depth.return_value = {
            0: [{'price': 100.0, 'amount': 10}],
            1: [{'price': 101.0, 'amount': 10}]
        }

        # Create strategy with mock API
        self.strategy = CrossingEMAs(self.mock_api, {
            'target_amount': 1000,
            'small_ema_period': 5,
            'big_ema_period': 20
        })

    def test_compute_emas(self):
        # Test EMA calculation
        self.strategy.price_curve = [100.0, 101.0, 102.0, 101.5, 101.8]
        self.strategy.compute_emas()

        # Check results (approximate values)
        self.assertAlmostEqual(self.strategy.ema_5, 101.26, places=2)

    def test_get_cross_direction_bullish(self):
        # Set up a bullish cross
        self.strategy.ema_5 = 101.0
        self.strategy.ema_20 = 100.0
        self.strategy.last_cross = 0  # BEARISH_CROSS

        # Test detection
        direction = self.strategy.get_new_cross_direction()
        self.assertEqual(direction, 1)  # BULLISH_CROSS

    def test_load_executes_trades(self):
        # Set up strategy state
        self.strategy.last_cross = 1  # BULLISH_CROSS
        self.mock_api.ticket_unit = 1
        self.mock_api.max_ticket = 10

        # Mock position
        self.mock_api.get_position.return_value = {
            'amount': 0,
            'limit': 10000
        }

        # Mock hit_price to return successful trade
        self.mock_api.hit_price.return_value = {'amount': 10}

        # Call load method
        self.strategy.load(1)  # BULLISH_CROSS

        # Verify hit_price was called with correct parameters
        self.mock_api.hit_price.assert_called_with(
            1,  # direction (BULLISH_CROSS)
            10,  # amount (max_ticket)
            100.0  # price from mock depth
        )
<!--

Integration Testing

class TestBotIntegration(unittest.TestCase):
    def setUp(self):
        # Create a test bot with mock API
        self.api_mock = Mock()
        self.bot = TestBot('test', 'test', 123)
        self.bot.api = self.api_mock

        # Set up mock API responses
        self.api_mock.get_market_prices.return_value = {
            'bid': 100.0,
            'ask': 101.0,
            'open': 99.0
        }
        self.api_mock.get_price_curve.return_value = [
            (datetime.now(), 99.0),
            (datetime.now(), 100.0),
            (datetime.now(), 101.0)
        ]
        self.api_mock.get_position.return_value = {
            'amount': 0,
            'rate': 0,
            'pnl': 0,
            'limit': 10000
        }

    def test_market_price_change_triggers_strategy(self):
        # Create a spy on the strategy method
        strategy_spy = Mock()
        self.bot.strategy.on_price_curve_change = strategy_spy

        # Trigger the event
        self.bot.on_market_price_change(
            {'bid': 99.0, 'ask': 100.0, 'open': 99.0},
            {'bid': 100.0, 'ask': 101.0, 'open': 99.0}
        )

        # Verify strategy method was called
        strategy_spy.assert_called_once()
``` -->

## Configuration Management

```python
import os
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Configuration settings
CONFIG = {
    "credentials": {
        "username": os.getenv("TRADERION_USERNAME"),
        "password": os.getenv("TRADERION_PASSWORD"),
    },
    "room_id": int(os.getenv("ROOM_ID", 0)),
    "risk_parameters": {
        "max_position": int(os.getenv("MAX_POSITION", 50000)),
        "max_drawdown": float(os.getenv("MAX_DRAWDOWN", 0.05)),
    },
    "strategy_parameters": {
        "ema_small": int(os.getenv("EMA_SMALL", 5)),
        "ema_big": int(os.getenv("EMA_BIG", 20)),
    }
}

Next Steps