Building Automated Trading Systems with Python
Complete guide to building automated trading systems with Python. Architecture design, backtesting, risk management, and deployment strategies for algorithmic trading.
Gönderim Güncelleme 
Python Automated Trading System Architecture
Yazan Furkan Köykıran
18 dakikada okunabilir
Building Automated Trading Systems with Python
Automated trading systems have revolutionized the financial markets, enabling traders to execute strategies with precision and speed that humans simply cannot match. In this comprehensive guide, we’ll explore how to build a robust automated trading system using Python, covering everything from architecture design to implementation and deployment.
What is Automated Trading?
Automated trading, also known as algorithmic trading or algo-trading, is the use of computer programs to execute trading strategies automatically based on predefined rules and conditions. These systems can monitor markets, analyze data, and execute trades 24/7 without human intervention.
Key Benefits
- Speed: Execute trades in milliseconds
- Consistency: Eliminate emotional decision-making
- Backtesting: Test strategies on historical data
- Scalability: Monitor multiple markets simultaneously
- Precision: Execute complex strategies with accuracy
System Architecture
A robust automated trading system consists of several key components:
1. Data Layer
The foundation of any trading system is reliable, real-time market data.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import ccxt
import pandas as pd
from datetime import datetime
class DataFeed:
def __init__(self, exchange_name='binance'):
"""Initialize data feed connection"""
self.exchange = getattr(ccxt, exchange_name)({
'enableRateLimit': True,
'options': {'defaultType': 'future'}
})
def fetch_ohlcv(self, symbol, timeframe='1h', limit=100):
"""Fetch OHLCV data from exchange"""
try:
ohlcv = self.exchange.fetch_ohlcv(
symbol,
timeframe,
limit=limit
)
df = pd.DataFrame(
ohlcv,
columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df
except Exception as e:
print(f"Error fetching data: {e}")
return None
def subscribe_ticker(self, symbol, callback):
"""Subscribe to real-time ticker updates"""
while True:
try:
ticker = self.exchange.fetch_ticker(symbol)
callback(ticker)
except Exception as e:
print(f"Ticker error: {e}")
time.sleep(1)
2. Strategy Engine
The strategy engine contains your trading logic and decision-making algorithms.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
import numpy as np
from abc import ABC, abstractmethod
class TradingStrategy(ABC):
"""Base class for trading strategies"""
def __init__(self, symbol, timeframe='1h'):
self.symbol = symbol
self.timeframe = timeframe
self.positions = {}
@abstractmethod
def calculate_signals(self, data):
"""Calculate trading signals from data"""
pass
@abstractmethod
def should_enter(self, data):
"""Determine if should enter position"""
pass
@abstractmethod
def should_exit(self, data):
"""Determine if should exit position"""
pass
class MACrossoverStrategy(TradingStrategy):
"""Moving Average Crossover Strategy"""
def __init__(self, symbol, short_window=20, long_window=50):
super().__init__(symbol)
self.short_window = short_window
self.long_window = long_window
def calculate_signals(self, data):
"""Calculate MA crossover signals"""
data['ma_short'] = data['close'].rolling(
window=self.short_window
).mean()
data['ma_long'] = data['close'].rolling(
window=self.long_window
).mean()
# Generate signals
data['signal'] = 0
data.loc[data['ma_short'] > data['ma_long'], 'signal'] = 1
data.loc[data['ma_short'] < data['ma_long'], 'signal'] = -1
return data
def should_enter(self, data):
"""Check for entry signals"""
if len(data) < 2:
return None
current_signal = data['signal'].iloc[-1]
previous_signal = data['signal'].iloc[-2]
# Long entry
if previous_signal <= 0 and current_signal == 1:
return 'long'
# Short entry
elif previous_signal >= 0 and current_signal == -1:
return 'short'
return None
def should_exit(self, data):
"""Check for exit signals"""
if not self.positions:
return False
current_signal = data['signal'].iloc[-1]
position_type = self.positions.get(self.symbol, {}).get('side')
# Exit long on short signal
if position_type == 'long' and current_signal == -1:
return True
# Exit short on long signal
elif position_type == 'short' and current_signal == 1:
return True
return False
3. Risk Management
Risk management is crucial for long-term trading success.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class RiskManager:
"""Manage risk and position sizing"""
def __init__(self, max_position_size=0.1, max_risk_per_trade=0.02):
"""
Initialize risk manager
Args:
max_position_size: Maximum position size as fraction of capital
max_risk_per_trade: Maximum risk per trade as fraction of capital
"""
self.max_position_size = max_position_size
self.max_risk_per_trade = max_risk_per_trade
def calculate_position_size(self, capital, entry_price, stop_loss):
"""Calculate position size based on risk"""
risk_amount = capital * self.max_risk_per_trade
price_risk = abs(entry_price - stop_loss)
if price_risk == 0:
return 0
position_size = risk_amount / price_risk
max_size = capital * self.max_position_size / entry_price
return min(position_size, max_size)
def calculate_stop_loss(self, entry_price, side, atr, multiplier=2):
"""Calculate stop loss based on ATR"""
if side == 'long':
return entry_price - (atr * multiplier)
else:
return entry_price + (atr * multiplier)
def calculate_take_profit(self, entry_price, stop_loss, risk_reward=2):
"""Calculate take profit based on risk/reward ratio"""
risk = abs(entry_price - stop_loss)
reward = risk * risk_reward
if entry_price > stop_loss: # Long position
return entry_price + reward
else: # Short position
return entry_price - reward
def validate_trade(self, trade, current_positions, capital):
"""Validate if trade meets risk criteria"""
# Check if already in position
if trade['symbol'] in current_positions:
return False, "Already in position"
# Check position size
position_value = trade['quantity'] * trade['price']
if position_value > capital * self.max_position_size:
return False, "Position size too large"
# Check stop loss is set
if 'stop_loss' not in trade:
return False, "No stop loss defined"
return True, "Trade validated"
4. Order Execution
The execution module handles order placement and management.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
from enum import Enum
import time
class OrderType(Enum):
MARKET = 'market'
LIMIT = 'limit'
STOP_LOSS = 'stop_loss'
TAKE_PROFIT = 'take_profit'
class OrderExecutor:
"""Handle order execution and management"""
def __init__(self, exchange):
self.exchange = exchange
self.active_orders = {}
def create_order(self, symbol, side, order_type, quantity, price=None, params={}):
"""Create and execute order"""
try:
if order_type == OrderType.MARKET:
order = self.exchange.create_market_order(
symbol, side, quantity, params
)
elif order_type == OrderType.LIMIT:
if price is None:
raise ValueError("Price required for limit order")
order = self.exchange.create_limit_order(
symbol, side, quantity, price, params
)
elif order_type == OrderType.STOP_LOSS:
order = self.exchange.create_stop_loss_order(
symbol, side, quantity, price, params
)
self.active_orders[order['id']] = order
return order
except Exception as e:
print(f"Order creation failed: {e}")
return None
def cancel_order(self, order_id, symbol):
"""Cancel an existing order"""
try:
result = self.exchange.cancel_order(order_id, symbol)
if order_id in self.active_orders:
del self.active_orders[order_id]
return result
except Exception as e:
print(f"Order cancellation failed: {e}")
return None
def get_order_status(self, order_id, symbol):
"""Get current order status"""
try:
order = self.exchange.fetch_order(order_id, symbol)
return order['status']
except Exception as e:
print(f"Error fetching order status: {e}")
return None
def modify_order(self, order_id, symbol, quantity=None, price=None):
"""Modify an existing order"""
try:
# Cancel old order
self.cancel_order(order_id, symbol)
# Create new order
order = self.active_orders.get(order_id)
if order:
new_order = self.create_order(
symbol,
order['side'],
OrderType(order['type']),
quantity or order['quantity'],
price or order.get('price')
)
return new_order
except Exception as e:
print(f"Order modification failed: {e}")
return None
5. Portfolio Management
Track and manage your overall portfolio performance.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
class Portfolio:
"""Manage trading portfolio"""
def __init__(self, initial_capital):
self.initial_capital = initial_capital
self.current_capital = initial_capital
self.positions = {}
self.trade_history = []
self.equity_curve = []
def open_position(self, symbol, side, quantity, entry_price, stop_loss, take_profit):
"""Open a new position"""
position = {
'symbol': symbol,
'side': side,
'quantity': quantity,
'entry_price': entry_price,
'entry_time': datetime.now(),
'stop_loss': stop_loss,
'take_profit': take_profit,
'unrealized_pnl': 0
}
self.positions[symbol] = position
return position
def close_position(self, symbol, exit_price):
"""Close an existing position"""
if symbol not in self.positions:
return None
position = self.positions[symbol]
# Calculate P&L
if position['side'] == 'long':
pnl = (exit_price - position['entry_price']) * position['quantity']
else:
pnl = (position['entry_price'] - exit_price) * position['quantity']
# Update capital
self.current_capital += pnl
# Record trade
trade = {
**position,
'exit_price': exit_price,
'exit_time': datetime.now(),
'pnl': pnl,
'return_pct': (pnl / (position['entry_price'] * position['quantity'])) * 100
}
self.trade_history.append(trade)
del self.positions[symbol]
# Update equity curve
self.equity_curve.append({
'timestamp': datetime.now(),
'equity': self.current_capital
})
return trade
def update_positions(self, current_prices):
"""Update unrealized P&L for open positions"""
for symbol, position in self.positions.items():
if symbol in current_prices:
current_price = current_prices[symbol]
if position['side'] == 'long':
unrealized_pnl = (current_price - position['entry_price']) * position['quantity']
else:
unrealized_pnl = (position['entry_price'] - current_price) * position['quantity']
position['unrealized_pnl'] = unrealized_pnl
def get_statistics(self):
"""Calculate portfolio statistics"""
if not self.trade_history:
return {}
trades = pd.DataFrame(self.trade_history)
total_trades = len(trades)
winning_trades = len(trades[trades['pnl'] > 0])
losing_trades = len(trades[trades['pnl'] < 0])
win_rate = (winning_trades / total_trades) * 100 if total_trades > 0 else 0
total_pnl = trades['pnl'].sum()
avg_win = trades[trades['pnl'] > 0]['pnl'].mean() if winning_trades > 0 else 0
avg_loss = trades[trades['pnl'] < 0]['pnl'].mean() if losing_trades > 0 else 0
profit_factor = abs(avg_win / avg_loss) if avg_loss != 0 else 0
return {
'total_trades': total_trades,
'winning_trades': winning_trades,
'losing_trades': losing_trades,
'win_rate': win_rate,
'total_pnl': total_pnl,
'avg_win': avg_win,
'avg_loss': avg_loss,
'profit_factor': profit_factor,
'return_pct': ((self.current_capital - self.initial_capital) / self.initial_capital) * 100
}
Building the Trading Bot
Now let’s integrate all components into a complete trading bot.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
import time
from threading import Thread
class TradingBot:
"""Complete automated trading bot"""
def __init__(self, exchange_name, symbol, strategy, initial_capital=10000):
"""Initialize trading bot"""
self.data_feed = DataFeed(exchange_name)
self.symbol = symbol
self.strategy = strategy
self.risk_manager = RiskManager()
self.executor = OrderExecutor(self.data_feed.exchange)
self.portfolio = Portfolio(initial_capital)
self.is_running = False
def start(self):
"""Start the trading bot"""
self.is_running = True
print(f"Starting trading bot for {self.symbol}")
# Start main trading loop
Thread(target=self._trading_loop, daemon=True).start()
# Start position monitoring
Thread(target=self._monitor_positions, daemon=True).start()
def stop(self):
"""Stop the trading bot"""
self.is_running = False
print("Stopping trading bot")
def _trading_loop(self):
"""Main trading loop"""
while self.is_running:
try:
# Fetch latest data
data = self.data_feed.fetch_ohlcv(
self.symbol,
self.strategy.timeframe,
limit=100
)
if data is None:
time.sleep(60)
continue
# Calculate strategy signals
data = self.strategy.calculate_signals(data)
# Check for entry signals
entry_signal = self.strategy.should_enter(data)
if entry_signal and not self.portfolio.positions.get(self.symbol):
self._execute_entry(entry_signal, data)
# Check for exit signals
if self.strategy.should_exit(data):
self._execute_exit(data)
# Sleep until next candle
time.sleep(60) # Check every minute
except Exception as e:
print(f"Error in trading loop: {e}")
time.sleep(60)
def _execute_entry(self, signal, data):
"""Execute entry trade"""
try:
current_price = data['close'].iloc[-1]
atr = self._calculate_atr(data)
# Calculate stop loss and take profit
stop_loss = self.risk_manager.calculate_stop_loss(
current_price, signal, atr
)
take_profit = self.risk_manager.calculate_take_profit(
current_price, stop_loss
)
# Calculate position size
quantity = self.risk_manager.calculate_position_size(
self.portfolio.current_capital,
current_price,
stop_loss
)
if quantity == 0:
print("Position size too small, skipping trade")
return
# Validate trade
trade = {
'symbol': self.symbol,
'side': signal,
'quantity': quantity,
'price': current_price,
'stop_loss': stop_loss,
'take_profit': take_profit
}
is_valid, message = self.risk_manager.validate_trade(
trade, self.portfolio.positions, self.portfolio.current_capital
)
if not is_valid:
print(f"Trade validation failed: {message}")
return
# Execute market order
order = self.executor.create_order(
self.symbol,
signal,
OrderType.MARKET,
quantity
)
if order:
# Open position in portfolio
self.portfolio.open_position(
self.symbol, signal, quantity,
current_price, stop_loss, take_profit
)
# Set stop loss and take profit orders
self.executor.create_order(
self.symbol,
'sell' if signal == 'long' else 'buy',
OrderType.STOP_LOSS,
quantity,
stop_loss
)
self.executor.create_order(
self.symbol,
'sell' if signal == 'long' else 'buy',
OrderType.LIMIT,
quantity,
take_profit
)
print(f"Opened {signal} position: {quantity} @ {current_price}")
except Exception as e:
print(f"Error executing entry: {e}")
def _execute_exit(self, data):
"""Execute exit trade"""
try:
if self.symbol not in self.portfolio.positions:
return
position = self.portfolio.positions[self.symbol]
current_price = data['close'].iloc[-1]
# Execute market order to close
order = self.executor.create_order(
self.symbol,
'sell' if position['side'] == 'long' else 'buy',
OrderType.MARKET,
position['quantity']
)
if order:
# Close position in portfolio
trade = self.portfolio.close_position(self.symbol, current_price)
if trade:
print(f"Closed {position['side']} position: P&L = {trade['pnl']:.2f}")
except Exception as e:
print(f"Error executing exit: {e}")
def _monitor_positions(self):
"""Monitor and update open positions"""
while self.is_running:
try:
if self.portfolio.positions:
# Fetch current prices
current_prices = {}
for symbol in self.portfolio.positions:
ticker = self.data_feed.exchange.fetch_ticker(symbol)
current_prices[symbol] = ticker['last']
# Update positions
self.portfolio.update_positions(current_prices)
time.sleep(10) # Update every 10 seconds
except Exception as e:
print(f"Error monitoring positions: {e}")
time.sleep(10)
def _calculate_atr(self, data, period=14):
"""Calculate Average True Range"""
high = data['high']
low = data['low']
close = data['close']
tr1 = high - low
tr2 = abs(high - close.shift())
tr3 = abs(low - close.shift())
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
atr = tr.rolling(period).mean()
return atr.iloc[-1]
def get_status(self):
"""Get current bot status"""
stats = self.portfolio.get_statistics()
return {
'is_running': self.is_running,
'current_capital': self.portfolio.current_capital,
'open_positions': len(self.portfolio.positions),
'statistics': stats
}
Backtesting Your Strategy
Before deploying with real money, always backtest your strategy.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class Backtester:
"""Backtest trading strategies on historical data"""
def __init__(self, strategy, initial_capital=10000):
self.strategy = strategy
self.initial_capital = initial_capital
self.trades = []
def run(self, data):
"""Run backtest on historical data"""
# Calculate signals
data = self.strategy.calculate_signals(data)
capital = self.initial_capital
position = None
for i in range(len(data)):
current_data = data.iloc[:i+1]
if len(current_data) < 2:
continue
# Check for entry
if position is None:
signal = self.strategy.should_enter(current_data)
if signal:
position = {
'side': signal,
'entry_price': current_data['close'].iloc[-1],
'entry_index': i
}
# Check for exit
elif self.strategy.should_exit(current_data):
exit_price = current_data['close'].iloc[-1]
# Calculate P&L
if position['side'] == 'long':
pnl_pct = ((exit_price - position['entry_price']) / position['entry_price']) * 100
else:
pnl_pct = ((position['entry_price'] - exit_price) / position['entry_price']) * 100
pnl = capital * (pnl_pct / 100)
capital += pnl
# Record trade
self.trades.append({
'entry_time': current_data.index[position['entry_index']],
'exit_time': current_data.index[i],
'side': position['side'],
'entry_price': position['entry_price'],
'exit_price': exit_price,
'pnl': pnl,
'pnl_pct': pnl_pct,
'capital': capital
})
position = None
return self._calculate_metrics(capital)
def _calculate_metrics(self, final_capital):
"""Calculate backtest performance metrics"""
if not self.trades:
return {}
trades_df = pd.DataFrame(self.trades)
# Basic metrics
total_return = ((final_capital - self.initial_capital) / self.initial_capital) * 100
total_trades = len(trades_df)
winning_trades = len(trades_df[trades_df['pnl'] > 0])
losing_trades = len(trades_df[trades_df['pnl'] < 0])
win_rate = (winning_trades / total_trades) * 100
# Risk metrics
returns = trades_df['pnl_pct'].values
sharpe_ratio = (np.mean(returns) / np.std(returns)) * np.sqrt(252) if np.std(returns) != 0 else 0
# Drawdown
equity_curve = trades_df['capital'].values
running_max = np.maximum.accumulate(equity_curve)
drawdown = (equity_curve - running_max) / running_max * 100
max_drawdown = np.min(drawdown)
return {
'total_return': total_return,
'total_trades': total_trades,
'winning_trades': winning_trades,
'losing_trades': losing_trades,
'win_rate': win_rate,
'sharpe_ratio': sharpe_ratio,
'max_drawdown': max_drawdown,
'final_capital': final_capital
}
Deployment and Monitoring
Running the Bot
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# Initialize and run bot
def main():
# Configuration
EXCHANGE = 'binance'
SYMBOL = 'BTC/USDT'
INITIAL_CAPITAL = 10000
# Create strategy
strategy = MACrossoverStrategy(SYMBOL, short_window=20, long_window=50)
# Create and start bot
bot = TradingBot(EXCHANGE, SYMBOL, strategy, INITIAL_CAPITAL)
bot.start()
# Monitor bot
try:
while True:
status = bot.get_status()
print(f"\nBot Status:")
print(f"Running: {status['is_running']}")
print(f"Capital: ${status['current_capital']:.2f}")
print(f"Open Positions: {status['open_positions']}")
if status['statistics']:
stats = status['statistics']
print(f"Total Trades: {stats['total_trades']}")
print(f"Win Rate: {stats['win_rate']:.2f}%")
print(f"Total P&L: ${stats['total_pnl']:.2f}")
time.sleep(300) # Update every 5 minutes
except KeyboardInterrupt:
print("\nShutting down bot...")
bot.stop()
if __name__ == "__main__":
main()
Logging and Monitoring
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import logging
from logging.handlers import RotatingFileHandler
def setup_logging():
"""Configure logging for the trading bot"""
logger = logging.getLogger('TradingBot')
logger.setLevel(logging.INFO)
# File handler
file_handler = RotatingFileHandler(
'trading_bot.log',
maxBytes=10*1024*1024, # 10MB
backupCount=5
)
file_handler.setLevel(logging.INFO)
# Console handler
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# Formatter
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
Best Practices
1. Start Small
Begin with a small amount of capital and test your system thoroughly in live conditions before scaling up.
2. Monitor Continuously
Set up alerts and monitoring to catch issues quickly:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def setup_alerts(bot):
"""Setup trading alerts"""
def check_drawdown():
stats = bot.portfolio.get_statistics()
if stats.get('return_pct', 0) < -10:
send_alert(f"WARNING: Drawdown exceeded 10%: {stats['return_pct']:.2f}%")
def check_win_rate():
stats = bot.portfolio.get_statistics()
if stats.get('total_trades', 0) > 20 and stats.get('win_rate', 100) < 40:
send_alert(f"WARNING: Low win rate: {stats['win_rate']:.2f}%")
# Run checks periodically
schedule.every(1).hours.do(check_drawdown)
schedule.every(1).hours.do(check_win_rate)
3. Implement Circuit Breakers
Automatically stop trading if losses exceed a threshold:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class CircuitBreaker:
"""Automatic trading halt on excessive losses"""
def __init__(self, max_daily_loss_pct=5, max_drawdown_pct=15):
self.max_daily_loss_pct = max_daily_loss_pct
self.max_drawdown_pct = max_drawdown_pct
self.daily_start_capital = None
def check(self, portfolio):
"""Check if circuit breaker should trigger"""
if self.daily_start_capital is None:
self.daily_start_capital = portfolio.current_capital
# Check daily loss
daily_loss_pct = ((portfolio.current_capital - self.daily_start_capital) /
self.daily_start_capital) * 100
if daily_loss_pct < -self.max_daily_loss_pct:
return True, f"Daily loss limit exceeded: {daily_loss_pct:.2f}%"
# Check max drawdown
stats = portfolio.get_statistics()
if abs(stats.get('return_pct', 0)) > self.max_drawdown_pct:
return True, f"Max drawdown exceeded: {stats['return_pct']:.2f}%"
return False, "OK"
4. Version Control Your Strategies
Keep track of strategy versions and their performance:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class StrategyVersion:
"""Track strategy versions and performance"""
def __init__(self, name, version, parameters):
self.name = name
self.version = version
self.parameters = parameters
self.deployed_at = datetime.now()
self.performance_log = []
def log_performance(self, metrics):
"""Log strategy performance"""
self.performance_log.append({
'timestamp': datetime.now(),
**metrics
})
def compare_with(self, other_version):
"""Compare performance with another version"""
if not self.performance_log or not other_version.performance_log:
return None
self_metrics = pd.DataFrame(self.performance_log).mean()
other_metrics = pd.DataFrame(other_version.performance_log).mean()
return {
'version_a': self.version,
'version_b': other_version.version,
'metrics_comparison': self_metrics - other_metrics
}
Common Pitfalls to Avoid
- Over-optimization: Don’t curve-fit your strategy to historical data
- Ignoring transaction costs: Always account for fees and slippage
- No risk management: Always use stop losses and position sizing
- Emotional intervention: Let the system run as designed
- Poor error handling: Expect and handle API failures gracefully
Conclusion
Building an automated trading system requires careful planning, robust architecture, and thorough testing. Start with simple strategies, implement proper risk management, and continuously monitor and improve your system.
Remember:
- Backtest extensively before going live
- Start with small capital
- Monitor continuously
- Implement proper risk management
- Keep learning and adapting
The code examples provided give you a solid foundation to build upon. Customize them for your specific needs and always test thoroughly before deploying with real money.
Resources
- CCXT Documentation
- Algorithmic Trading with Python
- Quantitative Finance with Python
- Trading Strategy Development
Happy trading! 🚀
Bu gönderi CC BY 4.0 lisansı altındadır.