Advanced Topics¶
This guide covers advanced concepts and techniques for developing sophisticated trading bots with Traderion.
Current and Upcoming Features¶
The Traderion framework currently provides core functionality for building trading bots with technical analysis strategies and basic position management. This page includes both currently implemented features and templates for future extensions you can implement yourself.
graph TD
A[TraderionBot] --> B[Event Handlers]
A --> C[Strategy Manager]
A -.-> D[Risk Manager]
A -.-> E[Performance Monitor]
C --> F[Technical Strategies]
C --> G[Client Strategies]
C --> H[Custom Strategies]
F --> I[EMA]
F --> J[MACD]
F --> K[RSI]
F -.-> L[Custom Indicators]
D -.-> M[Position Limits]
D -.-> N[Stop Loss]
D -.-> O[Take Profit]
D -.-> P[Dynamic Sizing]
E -.-> Q[Metrics Collection]
E -.-> R[Reporting]
E -.-> S[Alerting]
style A fill:#d4f1f9,stroke:#333,stroke-width:2px
style C fill:#d5e8d4,stroke:#333
style D fill:#ffe6cc,stroke:#333,stroke-dasharray: 5 5
style E fill:#f8cecc,stroke:#333,stroke-dasharray: 5 5
style L fill:#d5e8d4,stroke:#333,stroke-dasharray: 5 5
style M fill:#ffe6cc,stroke:#333,stroke-dasharray: 5 5
style N fill:#ffe6cc,stroke:#333,stroke-dasharray: 5 5
style O fill:#ffe6cc,stroke:#333,stroke-dasharray: 5 5
style P fill:#ffe6cc,stroke:#333,stroke-dasharray: 5 5
style Q fill:#f8cecc,stroke:#333,stroke-dasharray: 5 5
style R fill:#f8cecc,stroke:#333,stroke-dasharray: 5 5
style S fill:#f8cecc,stroke:#333,stroke-dasharray: 5 5
classDef upcoming stroke-dasharray: 5 5;
linkStyle 2,3,7,8,9,10,11,12,13,14,15 stroke-dasharray: 5 5;Note: Dashed lines indicate upcoming/template features that are not yet fully implemented in the core framework.
Custom Technical Indicators (Template)¶
The following is a template for creating your own technical indicators to implement unique trading strategies:
class CustomIndicator:
"""
A template for creating custom technical indicators.
This example implements a weighted moving average with more
weight given to recent prices.
"""
def __init__(self, period, weight_factor=0.1):
self.period = period
self.weight_factor = weight_factor
self.values = []
self.current_value = None
def update(self, price):
# Add new price to values
self.values.append(price)
# Keep only the most recent 'period' values
if len(self.values) > self.period:
self.values.pop(0)
# Calculate weighted average
weights = [(1 + self.weight_factor * i) for i in range(len(self.values))]
total_weight = sum(weights)
self.current_value = sum(p * w for p, w in zip(self.values, weights)) / total_weight
return self.current_value
def get_value(self):
return self.current_value
Implementing in a Bot¶
class AdvancedBot(TraderionBot):
def __init__(self, username, password, room_id):
super().__init__(username, password, room_id)
self.wma = CustomIndicator(period=14, weight_factor=0.2)
self.price_history = []
def on_price_curve_change(self, old_price_curve, new_price_curve):
# Extract prices from price curve
prices = [price for (_, price) in new_price_curve]
self.price_history = prices
# Update indicator with latest price
latest_price = prices[-1] if prices else None
if latest_price:
indicator_value = self.wma.update(latest_price)
print(f"Latest price: {latest_price}, WMA: {indicator_value}")
def main_loop(self):
if not self.price_history:
return
current_price = self.price_history[-1]
indicator_value = self.wma.get_value()
if current_price > indicator_value * 1.02: # 2% above WMA
# Consider selling
print("Sell signal generated")
elif current_price < indicator_value * 0.98: # 2% below WMA
# Consider buying
print("Buy signal generated")
Advanced Strategy Composition (Template)¶
The following is a template for combining multiple strategies with weighted decision-making:
class StrategyManager:
def __init__(self, api):
self.api = api
self.strategies = {}
self.weights = {}
def add_strategy(self, name, strategy, weight=1.0):
"""Add a strategy with an optional weight."""
self.strategies[name] = strategy
self.weights[name] = weight
def get_decision(self):
"""
Get a weighted decision from all strategies.
Returns a value between -1 (strong sell) and 1 (strong buy).
"""
if not self.strategies:
return 0
total_weight = sum(self.weights.values())
weighted_decision = 0
for name, strategy in self.strategies.items():
# Get strategy signal (-1 to 1)
signal = strategy.get_signal()
# Apply weight
weight = self.weights[name] / total_weight
weighted_decision += signal * weight
return weighted_decision
def execute(self):
"""Execute trades based on the weighted decision."""
decision = self.get_decision()
# Strong signals (above 0.7 or below -0.7)
if decision > 0.7:
# Strong buy signal
self.execute_buy(size="large")
elif decision > 0.3:
# Moderate buy signal
self.execute_buy(size="medium")
elif decision > 0.1:
# Weak buy signal
self.execute_buy(size="small")
elif decision < -0.7:
# Strong sell signal
self.execute_sell(size="large")
elif decision < -0.3:
# Moderate sell signal
self.execute_sell(size="medium")
elif decision < -0.1:
# Weak sell signal
self.execute_sell(size="small")
# Otherwise hold
Implementation Example¶
class CompositeBot(TraderionBot):
def __init__(self, username, password, room_id):
super().__init__(username, password, room_id)
# Create strategy manager
self.manager = StrategyManager(self.api)
# Add strategies with weights
self.manager.add_strategy("ema", CrossingEMAs(self.api, {
'target_amount': 10000,
'small_ema_period': 5,
'big_ema_period': 20
}), weight=1.5) # Higher weight for EMA
self.manager.add_strategy("rsi", Rsi(self.api, {
'target_amount': 10000,
'period': 14,
'overbought_threshold': 70,
'oversold_threshold': 30
}), weight=1.0)
self.manager.add_strategy("macd", Macd(self.api, {
'target_amount': 10000,
'small_ema_period': 12,
'big_ema_period': 26,
'signal_ema_period': 9
}), weight=1.2)
def main_loop(self):
# Let the strategy manager make decisions
self.manager.execute()
Advanced Risk Management (Template)¶
The following is a template for implementing sophisticated risk management techniques:
class RiskManager:
def __init__(self, api, config):
self.api = api
# Configuration
self.max_position = config.get('max_position', 50000)
self.max_drawdown = config.get('max_drawdown', 0.05) # 5%
self.stop_loss_pct = config.get('stop_loss', 0.02) # 2%
self.take_profit_pct = config.get('take_profit', 0.03) # 3%
# State
self.entry_prices = {} # Track entry prices for positions
self.position_sizes = {} # Track position sizes
def check_stop_loss(self, position):
"""Check if stop loss has been triggered."""
if position['amount'] == 0:
return False
direction = 1 if position['amount'] > 0 else -1
entry_price = self.entry_prices.get(direction, position['rate'])
current_price = self.api.get_market_prices()
if direction > 0: # Long position
stop_price = entry_price * (1 - self.stop_loss_pct)
return current_price['bid'] <= stop_price
else: # Short position
stop_price = entry_price * (1 + self.stop_loss_pct)
return current_price['ask'] >= stop_price
def check_take_profit(self, position):
"""Check if take profit has been triggered."""
if position['amount'] == 0:
return False
direction = 1 if position['amount'] > 0 else -1
entry_price = self.entry_prices.get(direction, position['rate'])
current_price = self.api.get_market_prices()
if direction > 0: # Long position
take_profit_price = entry_price * (1 + self.take_profit_pct)
return current_price['bid'] >= take_profit_price
else: # Short position
take_profit_price = entry_price * (1 - self.take_profit_pct)
return current_price['ask'] <= take_profit_price
def check_drawdown(self, position):
"""Check if maximum drawdown has been exceeded."""
return position['pnl_percentage'] < -self.max_drawdown
def manage_risk(self):
"""Main risk management function."""
position = self.api.get_position()
# Check stop loss
if self.check_stop_loss(position):
print("Stop loss triggered")
self.close_position(position)
return True
# Check take profit
if self.check_take_profit(position):
print("Take profit triggered")
self.close_position(position)
return True
# Check drawdown
if self.check_drawdown(position):
print("Maximum drawdown exceeded")
self.reduce_position(position, reduction_pct=0.5)
return True
return False
def close_position(self, position):
"""Close the current position."""
if position['amount'] == 0:
return
direction = 0 if position['amount'] > 0 else 1 # Opposite direction to close
amount = abs(position['amount']) / self.api.ticket_unit # Convert to short amount
# Execute the trade to close position
depth = self.api.get_eb_depth()
self.api.hit_price(direction, amount, depth[direction][0]['price'])
def reduce_position(self, position, reduction_pct=0.5):
"""Reduce the current position by a percentage."""
if position['amount'] == 0:
return
direction = 0 if position['amount'] > 0 else 1 # Opposite direction to reduce
amount = abs(position['amount']) * reduction_pct / self.api.ticket_unit # Convert to short amount
# Execute the trade to reduce position
depth = self.api.get_eb_depth()
self.api.hit_price(direction, amount, depth[direction][0]['price'])
def on_new_position(self, direction, amount, price):
"""Track new positions."""
self.entry_prices[direction] = price
self.position_sizes[direction] = amount
Implementation in a Bot¶
class RiskManagedBot(TraderionBot):
def __init__(self, username, password, room_id):
super().__init__(username, password, room_id)
# Initialize risk manager
self.risk_manager = RiskManager(self.api, {
'max_position': 50000,
'max_drawdown': 0.05,
'stop_loss': 0.02,
'take_profit': 0.03
})
# Initialize strategy
self.strategy = CrossingEMAs(self.api, {
'target_amount': 10000,
'small_ema_period': 5,
'big_ema_period': 20
})
def on_position_change(self, old_position, new_position):
# Update risk manager with position changes
if old_position['amount'] == 0 and new_position['amount'] != 0:
# New position opened
direction = 1 if new_position['amount'] > 0 else 0
amount = abs(new_position['amount'])
price = new_position['rate']
self.risk_manager.on_new_position(direction, amount, price)
def main_loop(self):
# First check risk management rules
if self.risk_manager.manage_risk():
# If risk management took action, skip strategy execution
return
# Execute strategy if risk management didn't intervene
self.strategy.run()