AI Crypto Trading Bots: Complete Guide to Automated Trading 2026

# Example: LSTM for price prediction
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout

def create_lstm_model(sequence_length, features):
    model = Sequential([
        LSTM(128, return_sequences=True, input_shape=(sequence_length, features)),
        Dropout(0.2),
        LSTM(64, return_sequences=False),
        Dropout(0.2),
        Dense(32, activation='relu'),
        Dense(1, activation='linear')  # Price prediction
    ])
    
    model.compile(optimizer='adam', loss='mse', metrics=['mae'])
    return model

# Binary classification for price movement direction
from sklearn.ensemble import RandomForestClassifier

def create_direction_model():
    model = RandomForestClassifier(
        n_estimators=100,
        max_depth=10,
        random_state=42,
        n_jobs=-1
    )
    return model

# Q-Learning trading agent
class TradingAgent:
    def __init__(self, state_size, action_size):
        self.state_size = state_size
        self.action_size = action_size
        self.gamma = 0.95  # Discount factor
        self.epsilon = 1.0  # Exploration rate
        self.learning_rate = 0.001
        self.q_table = {}
    
    def get_action(self, state):
        if random.random() < self.epsilon:
            return random.randrange(self.action_size)
        
        if state not in self.q_table:
            self.q_table[state] = np.zeros(self.action_size)
        
        return np.argmax(self.q_table[state])
    
    def learn(self, state, action, reward, next_state):
        if state not in self.q_table:
            self.q_table[state] = np.zeros(self.action_size)
        if next_state not in self.q_table:
            self.q_table[next_state] = np.zeros(self.action_size)
        
        current_q = self.q_table[state][action]
        max_next_q = np.max(self.q_table[next_state])
        new_q = current_q + self.learning_rate * (reward + self.gamma * max_next_q - current_q)
        
        self.q_table[state][action] = new_q

# Sentiment analysis for market news
from transformers import pipeline

sentiment_analyzer = pipeline("sentiment-analysis")

def analyze_market_sentiment(news_items):
    sentiments = []
    scores = []
    
    for news in news_items:
        result = sentiment_analyzer(news['text'])[0]
        sentiments.append(result['label'])
        scores.append(result['score'])
    
    # Aggregate sentiment
    bullish_count = sum(1 for s in sentiments if s == 'POSITIVE')
    bearish_count = len(sentiments) - bullish_count
    
    sentiment_score = (bullish_count - bearish_count) / len(sentiments)
    
    return {
        'sentiment': 'bullish' if sentiment_score > 0.1 else 'bearish' if sentiment_score < -0.1 else 'neutral',
        'score': sentiment_score,
        'confidence': np.mean(scores)
    }

def engineer_features(df):
    """Create trading features from raw data"""
    
    # Technical indicators
    df['rsi'] = calculate_rsi(df['close'], period=14)
    df['macd'] = calculate_macd(df['close'])
    df['bb_upper'], df['bb_middle'], df['bb_lower'] = calculate_bollinger_bands(df['close'])
    df['atr'] = calculate_atr(df['high'], df['low'], df['close'])
    
    # Price-based features
    df['returns'] = df['close'].pct_change()
    df['volatility'] = df['returns'].rolling(window=20).std()
    df['momentum'] = df['close'] / df['close'].shift(10) - 1
    
    # On-chain features
    df['whale_ratio'] = df['large_transactions'] / df['total_transactions']
    df['exchange_flow'] = df['inflow'] - df['outflow']
    
    # Sentiment features
    df['sentiment_score'] = calculate_sentiment_score(df['news'])
    
    return df

def trend_following_strategy(df, lookback=20):
    """
    Follow trends using moving average crossovers
    """
    # Calculate moving averages
    df['sma_fast'] = df['close'].rolling(window=10).mean()
    df['sma_slow'] = df['close'].rolling(window=lookback).mean()
    
    # Generate signals
    df['signal'] = 0
    df.loc[df['sma_fast'] > df['sma_slow'], 'signal'] = 1  # Long
    df.loc[df['sma_fast'] < df['sma_slow'], 'signal'] = -1  # Short
    
    # Filter with trend strength
    df['trend_strength'] = abs(df['sma_fast'] - df['sma_slow']) / df['sma_slow']
    df.loc[df['trend_strength'] < 0.02, 'signal'] = 0  # No trade in weak trends
    
    return df

def mean_reversion_strategy(df, window=20, std_dev=2):
    """
    Trade based on deviation from mean
    """
    df['rolling_mean'] = df['close'].rolling(window=window).mean()
    df['rolling_std'] = df['close'].rolling(window=window).std()
    
    df['z_score'] = (df['close'] - df['rolling_mean']) / df['rolling_std']
    
    # Buy when price is significantly below mean
    df['signal'] = 0
    df.loc[df['z_score'] < -std_dev, 'signal'] = 1
    
    # Sell when price is significantly above mean
    df.loc[df['z_score'] > std_dev, 'signal'] = -1
    
    # Exit when price returns to mean
    df.loc[abs(df['z_score']) < 0.5, 'signal'] = 0
    
    return df

def arbitrage_strategy(exchanges_data):
    """
    Find and execute arbitrage opportunities
    """
    opportunities = []
    
    for i, exchange1 in enumerate(exchanges_data):
        for exchange2 in exchanges_data[i+1:]:
            price_diff = exchange1['price'] - exchange2['price']
            price_diff_pct = price_diff / exchange1['price']
            
            # Account for fees and slippage
            net_profit = price_diff_pct - (exchange1['fee'] + exchange2['fee'])
            
            if net_profit > 0.001:  # 0.1% minimum profit
                opportunities.append({
                    'buy_exchange': exchange1['name'] if exchange1['price'] < exchange2['price'] else exchange2['name'],
                    'sell_exchange': exchange2['name'] if exchange1['price'] < exchange2['price'] else exchange1['name'],
                    'profit_pct': net_profit,
                    'size': calculate_position_size(net_profit)
                })
    
    return opportunities

def market_making_strategy(best_bid, best_ask, spread_pct=0.001):
    """
    Place orders around mid-price to capture spread
    """
    mid_price = (best_bid + best_ask) / 2
    
    # Place orders
    bid_price = mid_price * (1 - spread_pct)
    ask_price = mid_price * (1 + spread_pct)
    
    # Adjust based on inventory
    if inventory > target_inventory:
        # Reduce inventory by lowering ask
        bid_price *= 0.99
    elif inventory < target_inventory:
        # Increase inventory by lowering bid
        ask_price *= 1.01
    
    return {
        'bid': bid_price,
        'ask': ask_price,
        'spread': ask_price - bid_price
    }

def sentiment_strategy(df):
    """
    Trade based on sentiment analysis
    """
    # Get aggregated sentiment
    sentiment = calculate_market_sentiment()
    
    # Generate signals
    df['signal'] = 0
    
    # Strong bullish sentiment + positive price action
    df.loc[(sentiment['score'] > 0.3) & (df['returns'] > 0), 'signal'] = 1
    
    # Strong bearish sentiment + negative price action
    df.loc[(sentiment['score'] < -0.3) & (df['returns'] < 0), 'signal'] = -1
    
    return df

3commas_features = {
    "bots": ["DCA", "Grid", "TradingView"],
    "exchanges": ["Binance", "Coinbase", "Kraken"],
    "ai_features": ["Smart Trade", "AI Strategy Builder"],
    "pricing": "$15-100/month"
}

cryptohopper_features = {
    "bots": ["Market maker", "Arbitrage", "AI"],
    "exchanges": "100+ integrations",
    "ai_features": ["Pattern recognition", "Signal marketplace"],
    "pricing": "$0-100/month"
}

pionex_features = {
    "bots": ["Grid", "DCA", "Martingale"],
    "exchanges": "Native exchange",
    "ai_features": ["AI Portfolio Rebalancer"],
    "pricing": "0.05% trading fee only"
}

bitsgap_features = {
    "bots": ["Arbitrage", "Grid", "DCA"],
    "exchanges": "75+ exchanges",
    "ai_features": ["AI Optimizer"],
    "pricing": "$19-119/month"
}

stoic_features = {
    "approach": "AI quant strategies",
    "portfolio": "$230M+ AUM",
    "ai_features": ["Adaptive rebalancing", "Strategy selection"],
    "pricing": "Performance fee"
}

# Collect market data
import ccxt
import pandas as pd

def fetch_ohlcv(exchange, symbol, timeframe='1h', limit=1000):
    ohlcv = 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

# Usage
exchange = ccxt.binance()
btc_data = fetch_ohlcv(exchange, 'BTC/USDT')

def create_features(df):
    # Technical indicators
    df['rsi'] = ta.rsi(df['close'])
    df['macd'] = ta.macd(df['close'])
    df['bbands'] = ta.bbands(df['close'])
    
    # On-chain metrics (from external API)
    df['whale_activity'] = get_whale_activity()
    
    return df

def train_model(df):
    X = df[features].dropna()
    y = df['target'].dropna()
    
    # Align X and y
    X = X.loc[y.index]
    
    # Train-test split
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, shuffle=False
    )
    
    # Train model
    model = create_lstm_model()
    model.fit(X_train, y_train, epochs=100, batch_size=32)
    
    return model

def backtest(model, df, initial_capital=10000):
    df['predictions'] = model.predict(df[features])
    df['signal'] = (df['predictions'] > 0.5).astype(int)
    
    # Simulate trading
    capital = initial_capital
    position = 0
    
    for i in range(len(df)):
        if df['signal'].iloc[i] == 1 and position == 0:
            # Buy
            position = capital / df['close'].iloc[i]
            capital = 0
        elif df['signal'].iloc[i] == 0 and position > 0:
            # Sell
            capital = position * df['close'].iloc[i]
            position = 0
    
    final_value = capital + position * df['close'].iloc[-1]
    return (final_value - initial_capital) / initial_capital

class LiveTradingBot:
    def __init__(self, exchange, model, config):
        self.exchange = exchange
        self.model = model
        self.config = config
        self.position = 0
    
    def run(self):
        while True:
            # Get latest data
            df = fetch_recent_data(self.exchange)
            
            # Generate signal
            signal = self.model.predict(df)
            
            # Execute trade
            if signal == 1 and self.position == 0:
                self.buy()
            elif signal == 0 and self.position > 0:
                self.sell()
            
            # Sleep until next iteration
            time.sleep(self.config['interval'])
    
    def buy(self):
        order = self.exchange.create_market_buy_order(
            self.config['symbol'],
            self.config['size']
        )
        self.position = self.config['size']
    
    def sell(self):
        order = self.exchange.create_market_sell_order(
            self.config['symbol'],
            self.position
        )
        self.position = 0

def calculate_position_size(account_balance, risk_per_trade, stop_loss_pct):
    """
    Calculate position size based on risk management
    """
    risk_amount = account_balance * risk_per_trade
    position_size = risk_amount / stop_loss_pct
    
    return position_size

# Example
position = calculate_position_size(
    account_balance=10000,
    risk_per_trade=0.02,  # 2% risk
    stop_loss_pct=0.05     # 5% stop loss
)

def set_risk_parameters(entry_price, risk_reward_ratio=2):
    """
    Set stop loss and take profit levels
    """
    stop_loss = entry_price * 0.95  # 5% stop loss
    take_profit = entry_price * (1 + 0.05 * risk_reward_ratio)
    
    return {
        'entry': entry_price,
        'stop_loss': stop_loss,
        'take_profit': take_profit,
        'risk': 0.05,
        'reward': 0.05 * risk_reward_ratio
    }

def calculate_portfolio_risk(positions, portfolio_value):
    """
    Calculate overall portfolio risk
    """
    total_exposure = sum([p['value'] for p in positions])
    leverage = total_exposure / portfolio_value
    
    # Maximum recommended leverage
    max_leverage = 3.0
    
    return {
        'total_exposure': total_exposure,
        'leverage': leverage,
        'max_leverage': max_leverage,
        'risk_level': 'high' if leverage > 2 else 'moderate' if leverage > 1 else 'low'
    }

future_predictions = {
    "2026": [
        "Mainstream AI bot adoption",
        "Regulatory frameworks emerging",
        "Decentralized AI trading protocols"
    ],
    "2027": [
        "LLM-native trading strategies",
        "Cross-exchange AI coordination",
        "AI regulation mandates"
    ],
    "2028": [
        "Quantum-enhanced trading",
        "Fully autonomous trading AIs",
        "AI trading as a service"
    ]
}