Crypto Trading Automation: Building Your First Trading Bot

pip install ccxt pandas numpy python-dotenv

# .env file example
BINANCE_API_KEY=your_api_key_here
BINANCE_API_SECRET=your_api_secret_here

# Load API keys safely
from dotenv import load_dotenv
import os

load_dotenv()  # Load environment variables from .env file
API_KEY = os.getenv("BINANCE_API_KEY")
API_SECRET = os.getenv("BINANCE_API_SECRET")

# Connect to the exchange
import ccxt

exchange = ccxt.binance({
    'apiKey': API_KEY,
    'secret': API_SECRET,
    'enableRateLimit': True,  # Prevents API rate limit errors
})

# Check your balance
balance = exchange.fetch_balance()
print(f"BTC balance: {balance['BTC']['free']}")

# Fetch and format OHLCV data
import pandas as pd

def fetch_data(exchange, symbol='BTC/USD', timeframe='1h', limit=100):
    """
    Fetch OHLCV data from the exchange and convert to pandas DataFrame.
    
    Args:
        exchange: CCXT exchange instance
        symbol: Trading pair symbol
        timeframe: Candlestick timeframe
        limit: Number of candles to fetch
        
    Returns:
        DataFrame with OHLCV data
    """
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    return df

# Example usage
btc_data = fetch_data(exchange)
print(btc_data.tail())

# Implement SMA crossover strategy
def calculate_sma(df, period):
    """Calculate Simple Moving Average for a given period."""
    return df['close'].rolling(window=period).mean()

def trading_logic(df):
    """
    Implement SMA crossover trading strategy.
    Buy when short SMA crosses above long SMA.
    Sell when short SMA crosses below long SMA.
    """
    df['short_sma'] = calculate_sma(df, 10)  # 10-period SMA
    df['long_sma'] = calculate_sma(df, 50)   # 50-period SMA
    
    # Check for crossover
    latest = df.iloc[-1]
    previous = df.iloc[-2]
    
    if previous['short_sma'] < previous['long_sma'] and latest['short_sma'] > latest['long_sma']:
        return 'buy'
    elif previous['short_sma'] > previous['long_sma'] and latest['short_sma'] < latest['long_sma']:
        return 'sell'
    return 'hold'

# Main trading loop
import time

def trading_loop(exchange, symbol='BTC/USD', amount=0.001):
    """
    Main trading loop that continuously checks for trading signals
    and executes orders.
    
    Args:
        exchange: CCXT exchange instance
        symbol: Trading pair symbol
        amount: Amount of BTC to buy/sell per trade
    """
    while True:
        try:
            # Fetch latest data
            df = fetch_data(exchange, symbol)
            
            # Determine action
            action = trading_logic(df)
            
            # Execute trades based on the action
            if action == 'buy':
                print(f"Buying {amount} BTC")
                # Uncomment the line below to execute real trades
                # exchange.create_market_buy_order(symbol, amount)
            elif action == 'sell':
                print(f"Selling {amount} BTC")
                # Uncomment the line below to execute real trades
                # exchange.create_market_sell_order(symbol, amount)
            else:
                print("Holding position")
                
            # Wait for the next iteration (1 hour in this example)
            print(f"Waiting for next candlestick... ({time.strftime('%H:%M:%S')})")
            time.sleep(3600)  # 1 hour in seconds
            
        except Exception as e:
            print(f"Error in trading loop: {e}")
            time.sleep(60)  # Wait a minute before retrying

# Start the trading loop
# trading_loop(exchange)

# Momentum trading strategy using Rate of Change (ROC)
def calculate_roc(df, period=10):
    """
    Calculate Rate of Change for a given period.
    ROC = [(Current Price / Price n periods ago) - 1] * 100
    """
    return df['close'].pct_change(periods=period) * 100

def momentum_logic(df):
    """
    Implement momentum trading strategy.
    Buy when ROC is strongly positive.
    Sell when ROC is strongly negative.
    """
    df['roc'] = calculate_roc(df)
    
    # Check momentum
    if df['roc'].iloc[-1] > 5:  # Strong positive momentum
        return 'buy'
    elif df['roc'].iloc[-1] < -5:  # Strong negative momentum
        return 'sell'
    return 'hold'

# Combined strategy example
def combined_strategy(df):
    """
    Combine SMA crossover and momentum for a more robust strategy.
    Only buy when both strategies agree.
    """
    # Calculate indicators
    df['short_sma'] = calculate_sma(df, 10)
    df['long_sma'] = calculate_sma(df, 50)
    df['roc'] = calculate_roc(df)
    
    # SMA crossover signals
    latest = df.iloc[-1]
    previous = df.iloc[-2]
    sma_signal = 'hold'
    
    if previous['short_sma'] < previous['long_sma'] and latest['short_sma'] > latest['long_sma']:
        sma_signal = 'buy'
    elif previous['short_sma'] > previous['long_sma'] and latest['short_sma'] < latest['long_sma']:
        sma_signal = 'sell'
    
    # Momentum signals
    momentum_signal = 'hold'
    if latest['roc'] > 5:
        momentum_signal = 'buy'
    elif latest['roc'] < -5:
        momentum_signal = 'sell'
    
    # Combined decision
    if sma_signal == 'buy' and momentum_signal == 'buy':
        return 'buy'
    elif sma_signal == 'sell' or momentum_signal == 'sell':
        return 'sell'
    return 'hold'

# Position sizing and risk management
def calculate_position_size(exchange, symbol, risk_percentage=1.0, stop_loss_percentage=5.0):
    """
    Calculate position size based on account balance and risk tolerance.
    
    Args:
        exchange: CCXT exchange instance
        symbol: Trading pair symbol
        risk_percentage: Percentage of account to risk per trade (1-2% recommended)
        stop_loss_percentage: Percentage below entry price to set stop-loss
        
    Returns:
        Amount of base currency to buy/sell
    """
    # Get account balance
    balance = exchange.fetch_balance()
    quote_currency = symbol.split('/')[1]  # USD in BTC/USD
    available_balance = balance[quote_currency]['free']
    
    # Calculate risk amount
    risk_amount = available_balance * (risk_percentage / 100)
    
    # Get current market price
    ticker = exchange.fetch_ticker(symbol)
    current_price = ticker['last']
    
    # Calculate max loss per unit
    max_loss_per_unit = current_price * (stop_loss_percentage / 100)
    
    # Calculate position size
    position_size = risk_amount / max_loss_per_unit
    
    return position_size

# Implementing stop-loss in the trading loop
def trading_loop_with_risk_management(exchange, symbol='BTC/USD', risk_percentage=1.0, stop_loss_percentage=5.0):
    """
    Enhanced trading loop with proper risk management.
    """
    buy_price = None
    position_size = 0
    in_position = False
    
    while True:
        try:
            # Fetch latest data
            df = fetch_data(exchange, symbol)
            current_price = df['close'].iloc[-1]
            
            # Check for stop-loss if in position
            if in_position and buy_price is not None:
                stop_loss_price = buy_price * (1 - stop_loss_percentage/100)
                if current_price < stop_loss_price:
                    print(f"Stop-loss triggered! Selling at {current_price}")
                    # exchange.create_market_sell_order(symbol, position_size)
                    in_position = False
                    buy_price = None
                    position_size = 0
                    continue
            
            # Determine action based on strategy
            action = trading_logic(df)
            
            # Execute trades based on the action
            if action == 'buy' and not in_position:
                position_size = calculate_position_size(exchange, symbol, risk_percentage, stop_loss_percentage)
                print(f"Buying {position_size} {symbol.split('/')[0]} at {current_price}")
                # order = exchange.create_market_buy_order(symbol, position_size)
                buy_price = current_price
                in_position = True
                
            elif action == 'sell' and in_position:
                print(f"Selling {position_size} {symbol.split('/')[0]} at {current_price}")
                # exchange.create_market_sell_order(symbol, position_size)
                in_position = False
                buy_price = None
                position_size = 0
            
            # Wait for the next iteration
            print(f"Waiting for next update... Current price: {current_price}")
            time.sleep(3600)  # 1 hour
            
        except Exception as e:
            print(f"Error in trading loop: {e}")
            time.sleep(60)  # Wait a minute before retrying

# Simple backtesting function
def backtest_strategy(df, initial_capital=1000.0, position_size_pct=10.0):
    """
    Backtest a trading strategy on historical data.
    
    Args:
        df: DataFrame with OHLCV data
        initial_capital: Starting capital in USD
        position_size_pct: Percentage of capital to use per trade
        
    Returns:
        Dictionary with backtest results
    """
    # Add strategy signals
    df['short_sma'] = calculate_sma(df, 10)
    df['long_sma'] = calculate_sma(df, 50)
    
    # Initialize backtest variables
    capital = initial_capital
    position = 0
    trades = []
    
    # Simulate trading
    for i in range(1, len(df)):
        previous = df.iloc[i-1]
        current = df.iloc[i]
        
        # Check for buy signal
        if previous['short_sma'] < previous['long_sma'] and current['short_sma'] > current['long_sma']:
            # Calculate position size
            position_size = (capital * position_size_pct / 100) / current['close']
            # Record trade
            trades.append({
                'type': 'buy',
                'date': current['timestamp'],
                'price': current['close'],
                'position_size': position_size,
                'capital': capital
            })
            # Update position and capital
            position = position_size
            capital -= position_size * current['close']
            
        # Check for sell signal
        elif previous['short_sma'] > previous['long_sma'] and current['short_sma'] < current['long_sma'] and position > 0:
            # Calculate sell value
            sell_value = position * current['close']
            # Record trade
            trades.append({
                'type': 'sell',
                'date': current['timestamp'],
                'price': current['close'],
                'position_size': position,
                'capital': capital + sell_value
            })
            # Update position and capital
            capital += sell_value
            position = 0
    
    # Calculate final portfolio value
    final_value = capital
    if position > 0:
        final_value += position * df['close'].iloc[-1]
    
    # Calculate performance metrics
    total_return = (final_value / initial_capital - 1) * 100
    num_trades = len(trades)
    winning_trades = sum(1 for i in range(1, len(trades), 2) if i < len(trades) and trades[i]['capital'] > trades[i-1]['capital'])
    win_rate = winning_trades / (num_trades // 2) * 100 if num_trades > 0 else 0
    
    return {
        'initial_capital': initial_capital,
        'final_value': final_value,
        'total_return': total_return,
        'num_trades': num_trades,
        'win_rate': win_rate,
        'trades': trades
    }

# Example usage
# historical_data = pd.read_csv('btc_usd_history.csv')
# backtest_results = backtest_strategy(historical_data)
# print(f"Total Return: {backtest_results['total_return']:.2f}%")
# print(f"Number of Trades: {backtest_results['num_trades']}")
# print(f"Win Rate: {backtest_results['win_rate']:.2f}%")

# Run the bot in the background and keep it running after logout
nohup python trading_bot.py > trading_bot.log 2>&1 &

# Check if it's running
ps aux | grep trading_bot.py

# Create a systemd service file
sudo nano /etc/systemd/system/trading-bot.service

# Content of trading-bot.service
[Unit]
Description=Crypto Trading Bot
After=network.target

[Service]
User=your_username
WorkingDirectory=/path/to/your/bot
ExecStart=/usr/bin/python3 /path/to/your/bot/trading_bot.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

# Enable and start the service
sudo systemctl enable trading-bot.service
sudo systemctl start trading-bot.service

# Check status
sudo systemctl status trading-bot.service

# Edit crontab
crontab -e

# Add a daily restart at 00:00
0 0 * * * pkill -f trading_bot.py && cd /path/to/your/bot && python3 trading_bot.py > trading_bot.log 2>&1 &

# Email notifications
import smtplib
from email.mime.text import MIMEText

def send_email_alert(subject, message, to_email, from_email, password):
    """Send email notification."""
    msg = MIMEText(message)
    msg['Subject'] = subject
    msg['From'] = from_email
    msg['To'] = to_email
    
    try:
        server = smtplib.SMTP('smtp.gmail.com', 587)
        server.starttls()
        server.login(from_email, password)
        server.send_message(msg)
        server.quit()
        print("Email notification sent")
    except Exception as e:
        print(f"Failed to send email: {e}")

# Example usage in trading loop
if action == 'buy':
    # Execute buy order
    send_email_alert(
        subject="Trading Bot: BUY Signal Executed",
        message=f"Bought {position_size} BTC at {current_price} USD",
        to_email="your-email@example.com",
        from_email="your-bot-email@gmail.com",
        password="your-app-password"  # Use app-specific password for security
    )