Creating an AI Trading Bot using Machine Learning with help of AI¶

KEY_ID = "your API KEY" #replace it with your own KEY_ID from Alpaca: https://alpaca.markets/
SECRET_KEY = "your SECRET KEY" #replace it with your own SECRET_KEY from Alpaca

import warnings
warnings.filterwarnings('ignore')

pip install alpaca-trade-api pandas scikit-learn xgboost numpy pandas-ta

import pandas_ta as ta

def create_features(df):
    if df.empty or 'close' not in df.columns:
        print("DataFrame is empty or 'close' column is missing.")
        return df

    df.ta.sma(length=10, append=True, col_names=('SMA_10'))
    df.ta.sma(length=30, append=True, col_names=('SMA_30'))
    df.ta.ema(length=10, append=True, col_names=('EMA_10'))
    df.ta.rsi(length=14, append=True, col_names=('RSI_14'))
    df.ta.macd(append=True, col_names=('MACD_12_26_9', 'MACDh_12_26_9', 'MACDs_12_26_9'))
    df.ta.bbands(length=20, append=True, col_names=('BBL_20_2.0', 'BBM_20_2.0', 'BBU_20_2.0', 'BBB_20_2.0', 'BBP_20_2.0'))
    df.ta.atr(length=14, append=True, col_names=('ATR_14'))

    # Lagged returns
    for lag in [1, 3, 5, 10]:
        df[f'return_{lag}m'] = df['close'].pct_change(periods=lag)

    # Add more features: volatility, momentum, volume-based if available, etc.
    # e.g., log returns, price relative to moving average, etc.

    df.dropna(inplace=True) # Remove rows with NaNs created by indicators
    return df

if not btc_data.empty:
    featured_data = create_features(btc_data.copy()) # Use .copy() to avoid modifying original
    print("\nData with features:")
    print(featured_data.head())
else:
    print("Cannot create features, btc_data is empty.")
    featured_data = pd.DataFrame() # ensure it's a DataFrame

from sklearn.model_selection import train_test_split, TimeSeriesSplit
from sklearn.ensemble import RandomForestClassifier # Example model
from sklearn.metrics import accuracy_score, classification_report
import xgboost as xgb

def prepare_data_for_model(df, target_horizon=5, test_size=0.2):
    """
    Prepares X (features) and y (target) for the ML model.
    Target: 1 if price increases by more than a small threshold after target_horizon periods, 0 otherwise.
            -1 if price decreases by more than a small threshold. (Optional: make it 3 classes)
    """
    if df.empty or 'close' not in df.columns:
        print("DataFrame is empty or 'close' column is missing.")
        return pd.DataFrame(), pd.Series(dtype='float64'), pd.DataFrame(), pd.Series(dtype='float64')

    # Define target: 1 if price goes up in `target_horizon` minutes, 0 otherwise
    # A small threshold can help avoid noise around 0% change
    # price_threshold = 0.0005 # e.g., 0.05% change
    # df['future_price'] = df['close'].shift(-target_horizon)
    # df['price_change'] = (df['future_price'] - df['close']) / df['close']
    # df['target'] = 0 # Hold
    # df.loc[df['price_change'] > price_threshold, 'target'] = 1 # Buy
    # df.loc[df['price_change'] < -price_threshold, 'target'] = -1 # Sell (for 3-class)
    # For 2-class (Up/Not Up):
    df['target'] = (df['close'].shift(-target_horizon) > df['close']).astype(int)

    df.dropna(inplace=True) # Remove rows with NaN target (due to shift)

    feature_columns = [col for col in df.columns if col not in ['open', 'high', 'low', 'close', 'volume', 'trade_count', 'vwap', 'target', 'future_price', 'price_change']]
    X = df[feature_columns]
    y = df['target']

    # Time series split is crucial: DO NOT shuffle time series data for training
    # For a simple split:
    split_index = int(len(X) * (1 - test_size))
    X_train, X_test = X[:split_index], X[split_index:]
    y_train, y_test = y[:split_index], y[split_index:]

    # For more robust cross-validation, use TimeSeriesSplit
    # tscv = TimeSeriesSplit(n_splits=5)
    # for train_index, test_index in tscv.split(X):
    #     X_train, X_test = X.iloc[train_index], X.iloc[test_index]
    #     y_train, y_test = y.iloc[train_index], y.iloc[test_index]
        # Train and evaluate your model here

    return X_train, X_test, y_train, y_test, feature_columns


if not featured_data.empty:
    X_train, X_test, y_train, y_test, feature_cols = prepare_data_for_model(featured_data.copy(), target_horizon=5)

    if not X_train.empty:
        print(f"\nTraining data shape: X_train: {X_train.shape}, y_train: {y_train.shape}")
        print(f"Test data shape: X_test: {X_test.shape}, y_test: {y_test.shape}")
        print(f"Features used: {feature_cols}")

        # Example Model: Random Forest
        # model = RandomForestClassifier(n_estimators=100, random_state=42, class_weight='balanced')

        # Example Model: XGBoost (often performs well)
        model = xgb.XGBClassifier(
            objective='binary:logistic', # or 'multi:softprob' for multi-class
            n_estimators=100,
            learning_rate=0.1,
            max_depth=3,
            use_label_encoder=False, # Suppress a warning
            eval_metric='logloss' # or 'mlogloss' for multi-class
        )

        model.fit(X_train, y_train)

        # Evaluate on test set
        y_pred = model.predict(X_test)
        print("\nModel Evaluation on Test Set:")
        print(f"Accuracy: {accuracy_score(y_test, y_pred):.4f}")
        print(classification_report(y_test, y_pred, zero_division=0))

        # Feature importance (for tree-based models)
        if hasattr(model, 'feature_importances_'):
            importances = pd.Series(model.feature_importances_, index=X_train.columns).sort_values(ascending=False)
            print("\nFeature Importances:")
            print(importances.head(10))
    else:
        print("Not enough data to create training/test sets after feature engineering and target creation.")
else:
    print("Cannot prepare data for model, featured_data is empty.")
    model = None # Ensure model is defined even if training fails

def run_simple_backtest(df_with_predictions, initial_capital=10000, trade_size_usd=1000, transaction_cost_pct=0.003): # Alpaca crypto fee
    """
    A very simplified vectorized backtest.
    Assumes df_with_predictions has a 'signal' column (1 for buy, 0 for hold/nothing, -1 for sell if implementing).
    For this example, we'll assume our model's prediction (0 or 1) is the signal.
    1 = Go Long, 0 = Exit Long (or do nothing if not in position)
    """
    if df_with_predictions.empty or 'predicted_signal' not in df_with_predictions.columns:
        print("DataFrame for backtest is empty or 'predicted_signal' column missing.")
        return

    capital = initial_capital
    position_btc = 0  # Amount of BTC held
    portfolio_value = []

    # Assume 'predicted_signal' comes from your model (1 for predicted up, 0 for predicted down/neutral)
    # Let's assume a simple strategy: if signal is 1, buy. If signal is 0 and we have a position, sell.

    for i in range(len(df_with_predictions)):
        current_price = df_with_predictions['close'].iloc[i]
        signal = df_with_predictions['predicted_signal'].iloc[i]

        # Decision logic
        if signal == 1 and position_btc == 0: # Buy signal and no current position
            # Buy
            amount_to_buy_btc = trade_size_usd / current_price
            cost = amount_to_buy_btc * current_price * (1 + transaction_cost_pct)
            if capital >= cost:
                capital -= cost
                position_btc += amount_to_buy_btc
                # print(f"{df_with_predictions.index[i]}: BUY {amount_to_buy_btc:.6f} BTC @ {current_price:.2f}")

        elif signal == 0 and position_btc > 0: # Sell signal (or neutral) and have a position
            # Sell
            proceeds = position_btc * current_price * (1 - transaction_cost_pct)
            capital += proceeds
            # print(f"{df_with_predictions.index[i]}: SELL {position_btc:.6f} BTC @ {current_price:.2f}")
            position_btc = 0

        current_portfolio_value = capital + (position_btc * current_price)
        portfolio_value.append(current_portfolio_value)

    df_with_predictions['portfolio_value'] = portfolio_value
    print("\nBacktest Results:")
    print(f"Initial Capital: ${initial_capital:.2f}")
    print(f"Final Portfolio Value: ${df_with_predictions['portfolio_value'].iloc[-1]:.2f}")
    returns = (df_with_predictions['portfolio_value'].iloc[-1] / initial_capital - 1) * 100
    print(f"Total Return: {returns:.2f}%")

    # Plotting (optional)
    # import matplotlib.pyplot as plt
    # plt.figure(figsize=(12,6))
    # plt.plot(df_with_predictions.index, df_with_predictions['portfolio_value'])
    # plt.title('Portfolio Value Over Time')
    # plt.xlabel('Date')
    # plt.ylabel('Portfolio Value ($)')
    # plt.show()

if model and not X_test.empty:
    # Use the model to predict on the entire test set for backtesting
    # For a more realistic backtest, you'd re-train periodically or use a walk-forward approach.
    # Here, we're just using the single model trained on X_train.
    all_featured_data_for_backtest = featured_data.loc[X_test.index].copy() # Get original data rows for X_test
    all_featured_data_for_backtest['predicted_signal'] = model.predict(X_test) # Use the trained model

    run_simple_backtest(all_featured_data_for_backtest)
else:
    print("Skipping backtest as model or test data is not available.")

SYMBOL = "BTC/USD"
TRADE_QTY_USD = 100 # Amount in USD to trade per signal. Adjust based on risk tolerance.
TARGET_HORIZON_MINUTES = 5 # Same as used in training

# Global model and feature_cols (assuming they are trained and available)
# model = ... (your trained model)
# feature_cols = ... (list of feature column names used for training)

def get_latest_bar_features():
    """Fetches latest bars, calculates features for the most recent one."""
    # Fetch enough data to calculate all features (e.g., max lookback of your indicators)
    # If SMA_30 is longest, need at least 30 + target_horizon previous bars
    # Let's fetch more to be safe, e.g., 100 bars
    now = datetime.now()
    start_fetch_dt = (now - timedelta(minutes=150)).strftime('%Y-%m-%d %H:%M:%S') # fetch last 150 mins

    latest_bars_df = api.get_crypto_bars(
        SYMBOL,
        tradeapi. টাইমফ্রেম.মিনিট,
        start=start_fetch_dt, # Alpaca needs ISO format with T
        # end defaults to now
        limit=150 # fetch a bit more than needed for features
    ).df
    
    if latest_bars_df.empty or len(latest_bars_df) < 35: # Min needed for SMA_30 + some buffer
        print("Not enough recent bars to calculate features.")
        return None

    featured_bars = create_features(latest_bars_df.copy())
    if featured_bars.empty:
        print("Failed to create features for latest bars.")
        return None
    # Return only the features for the most recent complete bar
    return featured_bars[feature_cols].iloc[-1:] # Return as DataFrame

def check_and_place_trade():
    global model, feature_cols # Ensure these are accessible

    if model is None or feature_cols is None:
        print("Model not trained or feature columns not defined. Skipping trade check.")
        return

    print(f"\n{datetime.now()}: Checking for trading signal...")
    current_features_df = get_latest_bar_features()

    if current_features_df is None or current_features_df.empty:
        print("Could not get features for the latest bar.")
        return

    # Ensure columns are in the same order as during training
    current_features_df = current_features_df[feature_cols]

    prediction = model.predict(current_features_df)
    signal = prediction[0] # 0 for down/neutral, 1 for up

    print(f"Raw features for prediction: {current_features_df.iloc[0].to_dict()}")
    print(f"Model prediction: {signal}")

    try:
        positions = api.list_positions()
        btc_position = next((p for p in positions if p.symbol == SYMBOL), None)
        current_price_info = api.get_latest_crypto_quote(SYMBOL) # Use quote for more current price
        current_price = (current_price_info.ap + current_price_info.bp) / 2 # Mid price

        if not current_price:
            print("Could not get current price for BTC/USD.")
            return


        if signal == 1: # Predicted UP - Potential BUY
            if btc_position is None or float(btc_position.qty) == 0:
                qty_to_buy = TRADE_QTY_USD / current_price
                print(f"BUY signal. Attempting to buy {qty_to_buy:.6f} {SYMBOL} at ~${current_price:.2f}")
                api.submit_order(
                    symbol=SYMBOL,
                    qty=round(qty_to_buy, 6), # Alpaca crypto needs precision
                    side='buy',
                    type='market',
                    time_in_force='gtc' # Good 'til canceled
                )
                print("BUY order submitted.")
            else:
                print(f"BUY signal, but already have a position of {btc_position.qty} {SYMBOL}. Holding.")

        elif signal == 0: # Predicted DOWN/NEUTRAL - Potential SELL
            if btc_position and float(btc_position.qty) > 0:
                qty_to_sell = float(btc_position.qty) # Sell entire position
                print(f"SELL signal. Attempting to sell {qty_to_sell:.6f} {SYMBOL} at ~${current_price:.2f}")
                api.submit_order(
                    symbol=SYMBOL,
                    qty=round(qty_to_sell, 6),
                    side='sell',
                    type='market',
                    time_in_force='gtc'
                )
                print("SELL order submitted.")
            else:
                print("SELL signal, but no open position to sell. Doing nothing.")
        else:
            print("Neutral signal or unrecognized signal. Doing nothing.")

    except Exception as e:
        print(f"Error during trade execution: {e}")

# Main loop (very basic scheduler)
# For a robust bot, use APScheduler or run it in a more managed environment (e.g., cloud server with cron)
if __name__ == "__main__" and model is not None: # Ensure model is trained
    # This is a simplified loop. In a real bot, you'd schedule this.
    # For example, using APScheduler to run exactly at the start of each minute.
    print("Starting dummy trading loop (runs a few times for demo). Press Ctrl+C to stop.")
    print("IMPORTANT: This is for PAPER TRADING ONLY.")
    print(f"Will use model: {type(model).__name__} and features: {feature_cols}")
    try:
        # Initial run
        check_and_place_trade()
        for i in range(5): # Run for a few iterations for demo
            # Wait for the next minute (approximately)
            # A more precise scheduler (like APScheduler) is better for live trading
            time.sleep(60)
            check_and_place_trade()
    except KeyboardInterrupt:
        print("Trading loop stopped by user.")
    except Exception as e:
        print(f"An error occurred in the trading loop: {e}")
else:
    if model is None:
        print("Model is not trained. Cannot start trading loop.")

# Example of a market buy order with a trailing stop-loss
# api.submit_order(
#     symbol=SYMBOL,
#     qty=qty_to_buy,
#     side='buy',
#     type='market',
#     time_in_force='day',
#     trail_percent='1.5' # Trail stop loss 1.5% below high water mark
# )

import warnings
warnings.filterwarnings('ignore')