Building a Complete Automated Trading System

Project Structure

Configuration Layer

Settings Management

load_dotenv()

@dataclass class BrokerConfig: account: int = int(os.getenv('BROKER_ACCOUNT', 0)) password: str = os.getenv('BROKER_PASSWORD', '') server: str = os.getenv('BROKER_SERVER', '')

@dataclass class TradingConfig: symbols: list = None timeframe: str = 'H1' risk_per_trade: float = 1.0 max_daily_loss: float = 3.0 max_drawdown: float = 10.0 max_positions: int = 3

@dataclass class AlertConfig: telegram_token: str = os.getenv('TELEGRAM_TOKEN', '') telegram_chat_id: str = os.getenv('TELEGRAM_CHAT_ID', '') email_enabled: bool = False

class Config: broker = BrokerConfig() trading = TradingConfig(symbols=['EURUSD', 'GBPUSD', 'USDJPY']) alerts = AlertConfig()

@classmethod def validate(cls): """Validate configuration""" errors = []

if not cls.broker.account: errors.append("Broker account not configured")

if not cls.broker.password: errors.append("Broker password not configured")

if not cls.trading.symbols: errors.append("No trading symbols configured")

Data Layer

Price Feed

class DataFeed: def __init__(self, config): self.config = config self.connected = False self.cache = {}

def connect(self) -> bool: """Connect to MT5""" if not mt5.initialize(): return False

if not mt5.login( self.config.broker.account, password=self.config.broker.password, server=self.config.broker.server ): mt5.shutdown() return False

self.connected = True return True

def disconnect(self): """Disconnect from MT5""" mt5.shutdown() self.connected = False

def get_bars(self, symbol: str, timeframe: str, count: int) -> pd.DataFrame: """Get OHLCV bars""" tf_map = { 'M1': mt5.TIMEFRAME_M1, 'M5': mt5.TIMEFRAME_M5, 'M15': mt5.TIMEFRAME_M15, 'H1': mt5.TIMEFRAME_H1, 'H4': mt5.TIMEFRAME_H4, 'D1': mt5.TIMEFRAME_D1, }

rates = mt5.copy_rates_from_pos(symbol, tf_map[timeframe], 0, count)

if rates is None: return pd.DataFrame()

df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') df.set_index('time', inplace=True)

return df

def get_tick(self, symbol: str) -> Optional[Dict]: """Get current tick""" tick = mt5.symbol_info_tick(symbol)

if tick is None: return None

return { 'bid': tick.bid, 'ask': tick.ask, 'time': datetime.fromtimestamp(tick.time) }

def get_symbol_info(self, symbol: str) -> Optional[Dict]: """Get symbol specifications""" info = mt5.symbol_info(symbol)

if info is None: return None

Technical Indicators

class Indicators: @staticmethod def sma(data: pd.Series, period: int) -> pd.Series: return data.rolling(period).mean()

@staticmethod def ema(data: pd.Series, period: int) -> pd.Series: return data.ewm(span=period, adjust=False).mean()

@staticmethod def rsi(data: pd.Series, period: int = 14) -> pd.Series: delta = data.diff() gain = (delta.where(delta > 0, 0)).rolling(period).mean() loss = (-delta.where(delta < 0, 0)).rolling(period).mean() rs = gain / loss return 100 - (100 / (1 + rs))

@staticmethod def atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series: tr1 = high - low tr2 = abs(high - close.shift()) tr3 = abs(low - close.shift()) tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) return tr.rolling(period).mean()

@staticmethod def macd(data: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> Dict: fast_ema = data.ewm(span=fast, adjust=False).mean() slow_ema = data.ewm(span=slow, adjust=False).mean() macd_line = fast_ema - slow_ema signal_line = macd_line.ewm(span=signal, adjust=False).mean() histogram = macd_line - signal_line

Strategy Layer

Base Strategy

class BaseStrategy(ABC): def __init__(self, config: Dict): self.config = config self.name = "BaseStrategy"

@abstractmethod def calculate_indicators(self, data: pd.DataFrame) -> pd.DataFrame: """Calculate strategy indicators""" pass

@abstractmethod def generate_signal(self, data: pd.DataFrame) -> Optional[Dict]: """ Generate trading signal

Returns: Dict with keys: direction ('buy'/'sell'), reason, strength None if no signal """ pass

MA Crossover Strategy

class MACrossoverStrategy(BaseStrategy): def __init__(self, fast_period: int = 20, slow_period: int = 50): self.fast_period = fast_period self.slow_period = slow_period self.name = f"MA_Cross_{fast_period}_{slow_period}"

def calculate_indicators(self, data: pd.DataFrame) -> pd.DataFrame: df = data.copy() df['fast_ma'] = Indicators.ema(df['close'], self.fast_period) df['slow_ma'] = Indicators.ema(df['close'], self.slow_period) df['rsi'] = Indicators.rsi(df['close']) return df

def generate_signal(self, data: pd.DataFrame) -> Optional[Dict]: df = self.calculate_indicators(data)

if len(df) < self.slow_period + 2: return None

# Current and previous values fast_curr = df['fast_ma'].iloc[-1] fast_prev = df['fast_ma'].iloc[-2] slow_curr = df['slow_ma'].iloc[-1] slow_prev = df['slow_ma'].iloc[-2] rsi = df['rsi'].iloc[-1]

# Buy signal: Fast crosses above slow if fast_prev <= slow_prev and fast_curr > slow_curr: if rsi < 70: # Not overbought return { 'direction': 'buy', 'reason': f'Fast MA crossed above Slow MA, RSI={rsi:.1f}', 'strength': min(abs(fast_curr - slow_curr) / slow_curr * 100, 1) }

# Sell signal: Fast crosses below slow if fast_prev >= slow_prev and fast_curr < slow_curr: if rsi > 30: # Not oversold return { 'direction': 'sell', 'reason': f'Fast MA crossed below Slow MA, RSI={rsi:.1f}', 'strength': min(abs(fast_curr - slow_curr) / slow_curr * 100, 1) }

Risk Management Layer

Position Sizing

class PositionSizer: def __init__(self, data_feed: DataFeed, config): self.feed = data_feed self.config = config

def calculate_lots(self, symbol: str, stop_loss_pips: float, risk_percent: float = None) -> float: """Calculate position size based on risk""" if risk_percent is None: risk_percent = self.config.trading.risk_per_trade

# Get account info import MetaTrader5 as mt5 account = mt5.account_info() balance = account.balance

# Risk amount risk_amount = balance * risk_percent / 100

# Get symbol info info = self.feed.get_symbol_info(symbol) if info is None: return info['min_lot']

# Calculate pip value pip_value = (info['tick_value'] / info['tick_size']) info['point'] 10

# Calculate lots lots = risk_amount / (stop_loss_pips * pip_value)

# Normalize to lot step lots = int(lots / info['lot_step']) * info['lot_step']

# Apply limits lots = max(lots, info['min_lot']) lots = min(lots, info['max_lot'])

Risk Limits

class RiskLimits: def __init__(self, config): self.config = config self.daily_start_balance = None self.peak_balance = None self.last_reset = None

def initialize(self): """Initialize risk tracking""" balance = mt5.account_info().balance self.daily_start_balance = balance self.peak_balance = balance self.last_reset = datetime.now().date()

def reset_daily(self): """Reset daily tracking""" if datetime.now().date() != self.last_reset: balance = mt5.account_info().balance self.daily_start_balance = balance self.last_reset = datetime.now().date()

def check_daily_loss(self) -> bool: """Check if daily loss limit reached""" self.reset_daily()

current = mt5.account_info().balance loss_pct = (self.daily_start_balance - current) / self.daily_start_balance * 100

return loss_pct < self.config.trading.max_daily_loss

def check_drawdown(self) -> bool: """Check if max drawdown reached""" current = mt5.account_info().balance

if current > self.peak_balance: self.peak_balance = current

dd_pct = (self.peak_balance - current) / self.peak_balance * 100

return dd_pct < self.config.trading.max_drawdown

def check_position_count(self, magic: int) -> bool: """Check if max positions reached""" count = 0 for i in range(mt5.positions_total()): pos = mt5.positions_get_by_ticket(mt5.PositionGetTicket(i)) if pos and pos[0].magic == magic: count += 1

return count < self.config.trading.max_positions

def can_trade(self, magic: int) -> tuple: """Check all risk limits""" if not self.check_daily_loss(): return False, "Daily loss limit reached"

if not self.check_drawdown(): return False, "Max drawdown reached"

if not self.check_position_count(magic): return False, "Max positions reached"

Execution Layer

Order Manager

logger = logging.getLogger(__name__)

class OrderManager: def __init__(self, magic_number: int = 123456): self.magic = magic_number

def place_order(self, symbol: str, direction: str, lots: float, sl_price: float = 0, tp_price: float = 0) -> Optional[int]: """Place market order""" tick = mt5.symbol_info_tick(symbol) info = mt5.symbol_info(symbol)

if direction == 'buy': price = tick.ask order_type = mt5.ORDER_TYPE_BUY else: price = tick.bid order_type = mt5.ORDER_TYPE_SELL

request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": lots, "type": order_type, "price": price, "sl": sl_price, "tp": tp_price, "deviation": 10, "magic": self.magic, "comment": "TradingSystem", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, }

result = mt5.order_send(request)

if result.retcode != mt5.TRADE_RETCODE_DONE: logger.error(f"Order failed: {result.retcode} - {result.comment}") return None

logger.info(f"Order placed: {direction} {lots} {symbol} @ {price}") return result.order

def close_position(self, ticket: int) -> bool: """Close position by ticket""" position = mt5.positions_get(ticket=ticket)

if not position: return False

pos = position[0] tick = mt5.symbol_info_tick(pos.symbol)

if pos.type == mt5.POSITION_TYPE_BUY: price = tick.bid order_type = mt5.ORDER_TYPE_SELL else: price = tick.ask order_type = mt5.ORDER_TYPE_BUY

request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": pos.symbol, "volume": pos.volume, "type": order_type, "position": ticket, "price": price, "deviation": 10, "magic": self.magic, "comment": "Close", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, }

result = mt5.order_send(request) return result.retcode == mt5.TRADE_RETCODE_DONE

def get_open_positions(self) -> list: """Get all open positions for this EA""" positions = mt5.positions_get()

if positions is None: return []

Main Trading Loop

from config.settings import Config from data.feed import DataFeed from strategy.ma_crossover import MACrossoverStrategy from risk.position_sizing import PositionSizer from risk.limits import RiskLimits from execution.orders import OrderManager from monitoring.alerts import TelegramNotifier

# Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s | %(levelname)s | %(message)s' ) logger = logging.getLogger(__name__)

class TradingSystem: def __init__(self): Config.validate()

self.feed = DataFeed(Config) self.strategy = MACrossoverStrategy(fast_period=20, slow_period=50) self.sizer = PositionSizer(self.feed, Config) self.risk = RiskLimits(Config) self.orders = OrderManager() self.notifier = TelegramNotifier( Config.alerts.telegram_token, Config.alerts.telegram_chat_id )

self.running = False

def start(self): """Start trading system""" logger.info("Starting Trading System")

if not self.feed.connect(): logger.error("Failed to connect to broker") return

self.risk.initialize() self.running = True self.notifier.send("Trading System Started")

try: self.run() except KeyboardInterrupt: logger.info("Shutdown requested") except Exception as e: logger.exception(f"Critical error: {e}") self.notifier.send_error(str(e)) finally: self.stop()

def run(self): """Main trading loop""" last_bar_time = {}

while self.running: for symbol in Config.trading.symbols: try: # Check for new bar data = self.feed.get_bars(symbol, Config.trading.timeframe, 100)

if data.empty: continue

current_bar = data.index[-1]

if symbol in last_bar_time and last_bar_time[symbol] == current_bar: continue # No new bar

last_bar_time[symbol] = current_bar logger.debug(f"New bar on {symbol}")

# Process symbol self.process_symbol(symbol, data)

except Exception as e: logger.error(f"Error processing {symbol}: {e}")

# Manage existing positions self.manage_positions()

# Sleep before next iteration time.sleep(1)

def process_symbol(self, symbol: str, data): """Process trading logic for symbol""" # Check risk limits can_trade, reason = self.risk.can_trade(self.orders.magic) if not can_trade: logger.warning(f"Cannot trade: {reason}") return

# Generate signal signal = self.strategy.generate_signal(data)

if signal is None: return

logger.info(f"Signal on {symbol}: {signal['direction']} - {signal['reason']}")

# Calculate stop loss atr = data['high'].iloc[-14:].max() - data['low'].iloc[-14:].min() sl_distance = atr * 2

tick = self.feed.get_tick(symbol) info = self.feed.get_symbol_info(symbol)

if signal['direction'] == 'buy': sl_price = tick['bid'] - sl_distance tp_price = tick['ask'] + sl_distance * 2 else: sl_price = tick['ask'] + sl_distance tp_price = tick['bid'] - sl_distance * 2

# Calculate position size sl_pips = sl_distance / (info['point'] * 10) lots = self.sizer.calculate_lots(symbol, sl_pips)

# Execute trade order_id = self.orders.place_order( symbol=symbol, direction=signal['direction'], lots=lots, sl_price=sl_price, tp_price=tp_price )

if order_id: self.notifier.send_trade( signal['direction'].upper(), symbol, tick['ask'] if signal['direction'] == 'buy' else tick['bid'], lots, sl_price, tp_price )

def manage_positions(self): """Manage open positions""" positions = self.orders.get_open_positions()

for pos in positions: # Add trailing stop, break-even, etc. pass

def stop(self): """Stop trading system""" self.running = False self.feed.disconnect() self.notifier.send("Trading System Stopped") logger.info("Trading System stopped")