🤖
Trading Bots Development
Build professional automated trading systems on Solana
Advanced45 minutes
⚠️
Trading Risk Warning
Automated trading involves substantial risk. Never invest more than you can afford to lose. Always test strategies thoroughly on devnet/testnet before deploying with real funds. This guide is for educational purposes only.
🏗️ Bot Architecture Overview
A well-designed trading bot follows a modular architecture with clear separation of concerns:
Core Components
📊
Data Layer
WebSocket streams, REST API queries, data normalization
🧠
Strategy Engine
Trading logic, signal generation, decision making
💰
Execution Layer
Order placement, position management, trade execution
🛡️
Risk Management
Position sizing, stop-loss, drawdown limits
📝
Logging & Monitoring
Trade history, performance metrics, error tracking
🔧 Modular Bot Framework
base_bot.py - Reusable Bot Framework
import asyncio
import logging
from abc import ABC, abstractmethod
from datetime import datetime
from axiomtradeapi import AxiomTradeClient
logger = logging.getLogger(__name__)
class BaseTradingBot(ABC):
"""Base class for all trading bots"""
def __init__(self, config):
self.config = config
self.client = AxiomTradeClient(
auth_token=config.get('auth_token'),
refresh_token=config.get('refresh_token')
)
# Risk management
self.max_position_size = config.get('max_position_size', 1.0)
self.max_daily_trades = config.get('max_daily_trades', 50)
self.stop_loss_pct = config.get('stop_loss_pct', 0.05)
# Statistics
self.trades_today = 0
self.total_profit = 0
self.winning_trades = 0
self.losing_trades = 0
self.start_time = datetime.now()
logger.info(f"✅ {self.__class__.__name__} initialized")
@abstractmethod
async def analyze_opportunity(self, data):
"""
Analyze trading opportunity and generate signal
Returns: dict with 'action' ('buy'/'sell'/'hold') and 'confidence' (0-1)
"""
pass
@abstractmethod
async def execute_trade(self, signal, data):
"""Execute the trade based on signal"""
pass
async def check_risk_limits(self, proposed_trade):
"""Verify trade doesn't violate risk limits"""
# Check daily trade limit
if self.trades_today >= self.max_daily_trades:
logger.warning("⚠️ Daily trade limit reached")
return False
# Check position size
if proposed_trade['size'] > self.max_position_size:
logger.warning(f"⚠️ Position size {proposed_trade['size']} exceeds limit")
return False
# Check wallet balance
balance = self.client.GetBalance(proposed_trade['wallet'])
if balance['sol'] < proposed_trade['size']:
logger.warning("⚠️ Insufficient balance")
return False
return True
async def log_trade(self, trade_result):
"""Log trade execution and update statistics"""
self.trades_today += 1
if trade_result['profit'] > 0:
self.winning_trades += 1
else:
self.losing_trades += 1
self.total_profit += trade_result['profit']
logger.info(f"📝 Trade logged: {trade_result}")
def get_statistics(self):
"""Get bot performance statistics"""
total_trades = self.winning_trades + self.losing_trades
win_rate = (self.winning_trades / total_trades * 100) if total_trades > 0 else 0
return {
'total_trades': total_trades,
'winning_trades': self.winning_trades,
'losing_trades': self.losing_trades,
'win_rate': win_rate,
'total_profit': self.total_profit,
'runtime': (datetime.now() - self.start_time).total_seconds()
}
def print_statistics(self):
"""Print bot statistics"""
stats = self.get_statistics()
print(f"\n📊 BOT STATISTICS")
print(f" Total Trades: {stats['total_trades']}")
print(f" Win Rate: {stats['win_rate']:.1f}%")
print(f" Total Profit: {stats['total_profit']:.6f} SOL")
print(f" Runtime: {stats['runtime']:.0f}s")
@abstractmethod
async def run(self):
"""Main bot loop"""
pass🎯 Example: Momentum Trading Bot
A practical implementation using the base framework:
momentum_bot.py
from base_bot import BaseTradingBot
import asyncio
import logging
logger = logging.getLogger(__name__)
class MomentumTradingBot(BaseTradingBot):
"""Trades based on token momentum indicators"""
def __init__(self, config):
super().__init__(config)
# Strategy parameters
self.min_liquidity = 20.0 # Minimum 20 SOL liquidity
self.momentum_threshold = 0.15 # 15% price increase threshold
self.volume_multiplier = 2.0 # 2x average volume
# Tracking
self.token_history = {}
async def analyze_opportunity(self, token):
"""Analyze token for momentum signals"""
token_address = token['tokenAddress']
current_price = token.get('priceUsd', 0)
volume = token.get('volume24h', 0)
liquidity = token.get('liquiditySol', 0)
# Check basic criteria
if liquidity < self.min_liquidity:
return {'action': 'hold', 'confidence': 0, 'reason': 'Low liquidity'}
# Calculate momentum if we have history
if token_address in self.token_history:
prev_data = self.token_history[token_address]
prev_price = prev_data.get('price', current_price)
# Calculate price change
if prev_price > 0:
price_change = (current_price - prev_price) / prev_price
# Strong upward momentum
if price_change >= self.momentum_threshold:
confidence = min(price_change / self.momentum_threshold, 1.0)
return {
'action': 'buy',
'confidence': confidence,
'reason': f'Strong momentum: {price_change*100:.1f}%'
}
# Strong downward momentum (sell signal)
if price_change <= -self.momentum_threshold:
return {
'action': 'sell',
'confidence': 0.8,
'reason': f'Negative momentum: {price_change*100:.1f}%'
}
# Update history
self.token_history[token_address] = {
'price': current_price,
'volume': volume,
'timestamp': token.get('timestamp')
}
return {'action': 'hold', 'confidence': 0}
async def execute_trade(self, signal, token):
"""Execute the trade"""
if signal['action'] == 'hold':
return
# Prepare trade
trade = {
'wallet': self.config['wallet_address'],
'token': token['tokenAddress'],
'size': self.max_position_size * signal['confidence'],
'action': signal['action']
}
# Check risk limits
if not await self.check_risk_limits(trade):
return
logger.info(f"🎯 Executing {signal['action']} for {token['tokenName']}")
logger.info(f" Confidence: {signal['confidence']:.2f}")
logger.info(f" Reason: {signal['reason']}")
# PLACEHOLDER: Add your actual trade execution here
# For example, using Solana SDK to swap tokens
# Log the trade
await self.log_trade({
'token': token['tokenAddress'],
'action': signal['action'],
'size': trade['size'],
'profit': 0 # Would be calculated after trade completes
})
async def handle_new_tokens(self, tokens):
"""Process incoming token stream"""
for token in tokens:
try:
signal = await self.analyze_opportunity(token)
if signal['action'] != 'hold':
await self.execute_trade(signal, token)
except Exception as e:
logger.error(f"Error processing token: {e}")
async def run(self):
"""Start the momentum trading bot"""
print("🚀 Momentum Trading Bot Started")
print(f" Min Liquidity: {self.min_liquidity} SOL")
print(f" Momentum Threshold: {self.momentum_threshold*100}%")
print("\nPress Ctrl+C to stop\n")
try:
await self.client.subscribe_new_tokens(
callback=self.handle_new_tokens
)
except KeyboardInterrupt:
print("\n⏹️ Stopping bot...")
self.print_statistics()
# Run the bot
async def main():
config = {
'auth_token': 'your-auth-token',
'refresh_token': 'your-refresh-token',
'wallet_address': 'your-wallet-address',
'max_position_size': 0.5, # 0.5 SOL max per trade
'max_daily_trades': 20, # Max 20 trades per day
'stop_loss_pct': 0.10 # 10% stop-loss
}
bot = MomentumTradingBot(config)
await bot.run()
if __name__ == "__main__":
asyncio.run(main())🛡️ Essential Risk Management
1. Position Sizing
Never risk more than 1-2% of your capital on a single trade:
# Calculate position size based on risk
def calculate_position_size(account_balance, risk_pct, stop_loss_pct):
risk_amount = account_balance * risk_pct
position_size = risk_amount / stop_loss_pct
return min(position_size, account_balance * 0.1) # Max 10% per trade2. Stop-Loss Orders
Always implement automated stop-losses:
async def monitor_position(self, position):
entry_price = position['entry_price']
stop_loss = entry_price * (1 - self.stop_loss_pct)
while position['active']:
current_price = await self.get_current_price(position['token'])
if current_price <= stop_loss:
await self.close_position(position, reason='stop_loss')
break
await asyncio.sleep(1)3. Daily Limits
Set maximum daily trades and loss limits to prevent emotional trading and system errors.
4. Diversification
Don't put all capital in one token. Spread risk across multiple positions.
🎨 Common Bot Strategies
🎯
Token Sniping
Buy tokens immediately at launch
- • Ultra-low latency required
- • High risk, high reward
- • Requires quick exits
📈
Trend Following
Ride momentum in established trends
- • Analyze price patterns
- • Medium-term holds
- • Lower risk profile
🔄
Arbitrage
Exploit price differences across DEXs
- • Low risk when done right
- • Requires fast execution
- • Small but consistent profits
📊
Market Making
Provide liquidity and earn spreads
- • Continuous buy/sell orders
- • Requires significant capital
- • Steady passive income
🧪 Testing & Backtesting
💡
Always Test First
Never deploy a trading bot with real funds without extensive testing. Start on Solana devnet, then paper trade on mainnet, before risking real capital.
Backtesting Framework
Test your strategy against historical data:
class Backtester:
def __init__(self, strategy, historical_data):
self.strategy = strategy
self.data = historical_data
self.results = []
def run(self):
for token_data in self.data:
signal = self.strategy.analyze(token_data)
if signal['action'] != 'hold':
result = self.simulate_trade(signal, token_data)
self.results.append(result)
return self.calculate_metrics()Paper Trading
Run your bot in simulation mode with live data but without executing real trades. Track performance for at least a week before going live.