Building an Automated Trading Bot with Python and Market Data API

Algorithmic trading has become increasingly accessible thanks to powerful APIs and programming libraries. In this tutorial, we'll build a complete automated trading bot using Python that fetches market data, generates signals, and can execute trades automatically.

What We'll Build

By the end of this tutorial, you'll have a trading bot that:

• Connects to the PulseMarkets API for real-time data
• Implements a simple moving average crossover strategy
• Backtests the strategy on historical data
• Generates and logs trading signals
• Has a framework for live execution (paper trading)

Prerequisites

# Install required packages
pip install requests pandas numpy python-dotenv schedule

Project Structure

trading_bot/
├── config.py
├── data_fetcher.py
├── strategy.py
├── backtest.py
├── bot.py
└── .env

Step 1: Configuration Setup

First, let's create the configuration file:

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    API_KEY = os.getenv('PULSE_API_KEY')
    API_BASE_URL = 'https://api.pulse-markets.com'

    # Trading parameters
    SYMBOL = 'EURUSD'
    TIMEFRAME = 'H1'
    FAST_MA = 10
    SLOW_MA = 20

    # Risk management
    RISK_PER_TRADE = 0.02  # 2% of account
    STOP_LOSS_PIPS = 30
    TAKE_PROFIT_PIPS = 60

Step 2: Data Fetcher Module

Create a module to fetch data from the API:

# data_fetcher.py
import requests
import pandas as pd
from config import Config

class DataFetcher:
    def __init__(self):
        self.headers = {'X-API-Key': Config.API_KEY}
        self.base_url = Config.API_BASE_URL

    def get_candles(self, symbol, timeframe, limit=500):
        """Fetch historical candlestick data."""
        url = f"{self.base_url}/api/candles/{symbol}"
        params = {
            'timeframe': timeframe,
            'limit': limit
        }

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

        if response.status_code != 200:
            raise Exception(f"API Error: {response.json()}")

        data = response.json()['data']['candles']

        df = pd.DataFrame(data)
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        df.set_index('timestamp', inplace=True)
        df = df.astype(float)

        return df

    def get_current_price(self, symbol):
        """Fetch current price quote."""
        url = f"{self.base_url}/api/quote/{symbol}"
        response = requests.get(url, headers=self.headers)

        if response.status_code != 200:
            raise Exception(f"API Error: {response.json()}")

        data = response.json()['data']
        return {
            'bid': float(data['bid']),
            'ask': float(data['ask']),
            'price': float(data['price'])
        }

Step 3: Trading Strategy

Implement the moving average crossover strategy:

# strategy.py
import pandas as pd
import numpy as np
from config import Config

class MovingAverageCrossover:
    def __init__(self, fast_period=None, slow_period=None):
        self.fast_period = fast_period or Config.FAST_MA
        self.slow_period = slow_period or Config.SLOW_MA

    def calculate_indicators(self, df):
        """Add technical indicators to dataframe."""
        df = df.copy()
        df['fast_ma'] = df['close'].rolling(window=self.fast_period).mean()
        df['slow_ma'] = df['close'].rolling(window=self.slow_period).mean()
        return df

    def generate_signals(self, df):
        """Generate buy/sell signals based on MA crossover."""
        df = self.calculate_indicators(df)

        # Create signal column
        df['signal'] = 0

        # Buy signal: Fast MA crosses above Slow MA
        df.loc[(df['fast_ma'] > df['slow_ma']) &
               (df['fast_ma'].shift(1) <= df['slow_ma'].shift(1)), 'signal'] = 1

        # Sell signal: Fast MA crosses below Slow MA
        df.loc[(df['fast_ma'] < df['slow_ma']) &
               (df['fast_ma'].shift(1) >= df['slow_ma'].shift(1)), 'signal'] = -1

        return df

    def get_current_signal(self, df):
        """Get the current trading signal."""
        df = self.generate_signals(df)
        last_signal = df['signal'].iloc[-1]

        if last_signal == 1:
            return 'BUY'
        elif last_signal == -1:
            return 'SELL'
        else:
            return 'HOLD'

Step 4: Backtesting Module

Create a backtester to validate the strategy:

# backtest.py
import pandas as pd
import numpy as np
from data_fetcher import DataFetcher
from strategy import MovingAverageCrossover
from config import Config

class Backtester:
    def __init__(self, initial_capital=10000):
        self.initial_capital = initial_capital
        self.fetcher = DataFetcher()
        self.strategy = MovingAverageCrossover()

    def run(self, symbol=None, timeframe=None, limit=500):
        """Run backtest and return results."""
        symbol = symbol or Config.SYMBOL
        timeframe = timeframe or Config.TIMEFRAME

        # Fetch data
        df = self.fetcher.get_candles(symbol, timeframe, limit)

        # Generate signals
        df = self.strategy.generate_signals(df)

        # Calculate returns
        df['returns'] = df['close'].pct_change()
        df['strategy_returns'] = df['signal'].shift(1) * df['returns']

        # Calculate cumulative returns
        df['cumulative_market'] = (1 + df['returns']).cumprod()
        df['cumulative_strategy'] = (1 + df['strategy_returns']).cumprod()

        # Calculate metrics
        total_return = df['cumulative_strategy'].iloc[-1] - 1
        market_return = df['cumulative_market'].iloc[-1] - 1

        # Calculate win rate
        winning_trades = (df[df['signal'] != 0]['strategy_returns'] > 0).sum()
        total_trades = (df['signal'] != 0).sum()
        win_rate = winning_trades / total_trades if total_trades > 0 else 0

        # Calculate Sharpe Ratio (annualized)
        risk_free_rate = 0.02
        excess_returns = df['strategy_returns'] - risk_free_rate/252
        sharpe_ratio = np.sqrt(252) * excess_returns.mean() / excess_returns.std()

        # Calculate max drawdown
        cumulative = df['cumulative_strategy']
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max
        max_drawdown = drawdown.min()

        results = {
            'total_return': f"{total_return:.2%}",
            'market_return': f"{market_return:.2%}",
            'total_trades': total_trades,
            'win_rate': f"{win_rate:.2%}",
            'sharpe_ratio': f"{sharpe_ratio:.2f}",
            'max_drawdown': f"{max_drawdown:.2%}",
            'final_capital': f"${self.initial_capital * (1 + total_return):,.2f}"
        }

        return results, df

if __name__ == '__main__':
    backtester = Backtester()
    results, df = backtester.run()

    print("\n=== Backtest Results ===")
    for key, value in results.items():
        print(f"{key}: {value}")

Step 5: Main Bot

Put it all together in the main bot file:

# bot.py
import time
import logging
from datetime import datetime
from data_fetcher import DataFetcher
from strategy import MovingAverageCrossover
from config import Config

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('trading_bot.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class TradingBot:
    def __init__(self):
        self.fetcher = DataFetcher()
        self.strategy = MovingAverageCrossover()
        self.position = None  # None, 'LONG', or 'SHORT'
        self.entry_price = None

    def analyze(self):
        """Analyze market and return signal."""
        try:
            # Get historical data
            df = self.fetcher.get_candles(
                Config.SYMBOL,
                Config.TIMEFRAME,
                limit=100
            )

            # Get current signal
            signal = self.strategy.get_current_signal(df)

            # Get current price
            price = self.fetcher.get_current_price(Config.SYMBOL)

            logger.info(f"Symbol: {Config.SYMBOL}")
            logger.info(f"Current Price: {price['price']}")
            logger.info(f"Signal: {signal}")

            return signal, price

        except Exception as e:
            logger.error(f"Analysis error: {e}")
            return 'HOLD', None

    def execute_signal(self, signal, price):
        """Execute trading signal (paper trading)."""
        if signal == 'BUY' and self.position != 'LONG':
            if self.position == 'SHORT':
                logger.info(f"Closing SHORT position at {price['ask']}")

            logger.info(f"Opening LONG position at {price['ask']}")
            self.position = 'LONG'
            self.entry_price = price['ask']

        elif signal == 'SELL' and self.position != 'SHORT':
            if self.position == 'LONG':
                logger.info(f"Closing LONG position at {price['bid']}")

            logger.info(f"Opening SHORT position at {price['bid']}")
            self.position = 'SHORT'
            self.entry_price = price['bid']

    def run_once(self):
        """Run single analysis cycle."""
        logger.info("=" * 50)
        logger.info(f"Running analysis at {datetime.now()}")

        signal, price = self.analyze()

        if price and signal != 'HOLD':
            self.execute_signal(signal, price)

        logger.info(f"Current position: {self.position or 'FLAT'}")

    def run(self, interval_minutes=60):
        """Run bot continuously."""
        logger.info("Starting Trading Bot...")
        logger.info(f"Symbol: {Config.SYMBOL}")
        logger.info(f"Timeframe: {Config.TIMEFRAME}")
        logger.info(f"Check interval: {interval_minutes} minutes")

        while True:
            try:
                self.run_once()
                time.sleep(interval_minutes * 60)
            except KeyboardInterrupt:
                logger.info("Bot stopped by user")
                break
            except Exception as e:
                logger.error(f"Bot error: {e}")
                time.sleep(60)  # Wait before retry

if __name__ == '__main__':
    bot = TradingBot()

    # Run single analysis
    bot.run_once()

    # Or run continuously (uncomment below)
    # bot.run(interval_minutes=60)

Running the Bot

# Create .env file
echo "PULSE_API_KEY=fx_your_key_here" > .env

# Run backtest first
python backtest.py

# Run bot in analysis mode
python bot.py

Important Considerations

Before Going Live

• Paper trade first: Run for at least a month in paper mode
• Validate backtest: Be wary of overfitting
• Start small: Use minimum position sizes initially
• Monitor closely: Don't leave bots unattended

Risk Warning

Automated trading carries significant risks. This tutorial is for educational purposes only. Never trade with money you can't afford to lose, and always test thoroughly before live trading.

Next Steps

This basic framework can be extended with:

• More sophisticated strategies (RSI, MACD, multiple timeframes)
• Risk management with dynamic position sizing
• WebSocket connection for faster signal detection
• Database storage for trade history
• Telegram or email notifications
• Integration with a broker API for live execution

Happy coding, and trade responsibly!