Connecting Python to Broker APIs for Live Trading

Requirements:

• Windows OS (MT5 limitation)
• MT5 terminal installed
• Demo or live account

Basic Connection

def connect_mt5(): """Initialize MT5 connection""" if not mt5.initialize(): print(f"MT5 initialization failed: {mt5.last_error()}") return False

# Login to account account = 12345678 # Your account number password = "your_password" server = "YourBroker-Server"

if not mt5.login(account, password=password, server=server): print(f"Login failed: {mt5.last_error()}") mt5.shutdown() return False

print(f"Connected to account #{account}") return True

def disconnect_mt5(): """Shutdown MT5 connection""" mt5.shutdown() print("Disconnected from MT5")

Getting Market Data

def get_historical_data(symbol, timeframe, bars): """ Get historical OHLCV data """ # Timeframe mapping 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, }

tf = tf_map.get(timeframe, mt5.TIMEFRAME_H1)

# Get rates rates = mt5.copy_rates_from_pos(symbol, tf, 0, bars)

if rates is None: print(f"Failed to get data: {mt5.last_error()}") return None

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

return df

Getting Live Prices

if tick is None: return None, None

return tick.bid, tick.ask

def get_symbol_info(symbol): """Get symbol specifications""" info = mt5.symbol_info(symbol)

if info is None: return None

return { 'name': info.name, 'digits': info.digits, 'point': info.point, 'spread': info.spread, 'min_lot': info.volume_min, 'max_lot': info.volume_max, 'lot_step': info.volume_step }

# Usage bid, ask = get_current_price('EURUSD') print(f"EURUSD: Bid={bid}, Ask={ask}")

Placing Orders

order_type: 'buy' or 'sell' volume: lot size sl_pips: stop loss in pips tp_pips: take profit in pips """ # Get symbol info info = mt5.symbol_info(symbol) if info is None: print(f"Symbol {symbol} not found") return None

point = info.point

# Get current prices tick = mt5.symbol_info_tick(symbol) if tick is None: print("Failed to get price") return None

# Determine price and order type if order_type.lower() == 'buy': price = tick.ask mt5_type = mt5.ORDER_TYPE_BUY sl = price - sl_pips point 10 if sl_pips else 0 tp = price + tp_pips point 10 if tp_pips else 0 else: price = tick.bid mt5_type = mt5.ORDER_TYPE_SELL sl = price + sl_pips point 10 if sl_pips else 0 tp = price - tp_pips point 10 if tp_pips else 0

# Prepare request request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": symbol, "volume": volume, "type": mt5_type, "price": price, "sl": sl, "tp": tp, "deviation": 10, "magic": 234000, "comment": "Python trade", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, }

# Send order result = mt5.order_send(request)

if result.retcode != mt5.TRADE_RETCODE_DONE: print(f"Order failed: {result.retcode}") return None

print(f"Order placed: {result.order}") return result

Managing Positions

if positions is None: return []

return list(positions)

def close_position(ticket): """Close position by ticket""" position = mt5.positions_get(ticket=ticket) if not position: print(f"Position {ticket} not found") return False

pos = position[0]

# Prepare close request request = { "action": mt5.TRADE_ACTION_DEAL, "symbol": pos.symbol, "volume": pos.volume, "type": mt5.ORDER_TYPE_SELL if pos.type == 0 else mt5.ORDER_TYPE_BUY, "position": ticket, "price": mt5.symbol_info_tick(pos.symbol).bid if pos.type == 0 else mt5.symbol_info_tick(pos.symbol).ask, "deviation": 10, "magic": 234000, "comment": "Close by Python", "type_time": mt5.ORDER_TIME_GTC, "type_filling": mt5.ORDER_FILLING_IOC, }

result = mt5.order_send(request)

if result.retcode != mt5.TRADE_RETCODE_DONE: print(f"Close failed: {result.retcode}") return False

print(f"Position {ticket} closed") return True

OANDA API Example

For non-MT5 brokers:

class OandaAPI: def __init__(self, api_key, account_id, practice=True): self.api_key = api_key self.account_id = account_id

if practice: self.base_url = "https://api-fxpractice.oanda.com" else: self.base_url = "https://api-fxtrade.oanda.com"

self.headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }

def get_prices(self, instruments): """Get current prices""" url = f"{self.base_url}/v3/accounts/{self.account_id}/pricing" params = {"instruments": instruments}

response = requests.get(url, headers=self.headers, params=params) return response.json()

def place_order(self, instrument, units, sl=None, tp=None): """Place market order""" url = f"{self.base_url}/v3/accounts/{self.account_id}/orders"

order = { "type": "MARKET", "instrument": instrument, "units": str(units), # Positive for buy, negative for sell "timeInForce": "FOK" }

if sl: order["stopLossOnFill"] = {"price": str(sl)} if tp: order["takeProfitOnFill"] = {"price": str(tp)}

data = {"order": order} response = requests.post(url, headers=self.headers, json=data) return response.json()

Complete Trading Script

class MT5Trader: def __init__(self, account, password, server): self.account = account self.password = password self.server = server self.connected = False

def connect(self): if not mt5.initialize(): return False

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

self.connected = True return True

def disconnect(self): mt5.shutdown() self.connected = False

def get_data(self, symbol, timeframe, bars): tf_map = {'H1': mt5.TIMEFRAME_H1, 'D1': mt5.TIMEFRAME_D1} rates = mt5.copy_rates_from_pos(symbol, tf_map[timeframe], 0, bars) df = pd.DataFrame(rates) df['time'] = pd.to_datetime(df['time'], unit='s') return df

def calculate_signal(self, df, fast=20, slow=50): df['fast_ma'] = df['close'].rolling(fast).mean() df['slow_ma'] = df['close'].rolling(slow).mean()

if df['fast_ma'].iloc[-1] > df['slow_ma'].iloc[-1]: if df['fast_ma'].iloc[-2] <= df['slow_ma'].iloc[-2]: return 'BUY' elif df['fast_ma'].iloc[-1] < df['slow_ma'].iloc[-1]: if df['fast_ma'].iloc[-2] >= df['slow_ma'].iloc[-2]: return 'SELL' return None

def run(self, symbol='EURUSD', timeframe='H1'): if not self.connect(): print("Connection failed") return

print(f"Starting trading on {symbol}")

try: while True: df = self.get_data(symbol, timeframe, 100) signal = self.calculate_signal(df)

if signal: print(f"{datetime.now()}: {signal} signal on {symbol}") # Execute trade logic here

time.sleep(60) # Check every minute

except KeyboardInterrupt: print("Stopping...") finally: self.disconnect()