Code Writer Automated Stock Trading Bot

Here's a Python-based implementation of an automated stock trading bot using Pandas and the Alpaca API, following the moving average crossover strategy. This example assumes you've already registered for Alpaca's API, have API keys, and have set up a funded account. ### Requirements You'll need the following libraries: ```bash pip install alpaca-trade-api pandas numpy ``` ### Source Code The code below connects to Alpaca’s API, retrieves real-time stock data, executes trades based on a moving average crossover strategy, logs activities, and includes risk management with stop-loss and position sizing. ```python import alpaca_trade_api as tradeapi import pandas as pd import numpy as np import logging from datetime import datetime, timedelta import time # Alpaca API credentials API_KEY = 'your_api_key' SECRET_KEY = 'your_secret_key' BASE_URL = 'https://paper-api.alpaca.markets' # Use the paper trading URL # Initialize Alpaca API api = tradeapi.REST(API_KEY, SECRET_KEY, BASE_URL, api_version='v2') # Logging configuration logging.basicConfig(filename='trading_bot.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Bot configuration SYMBOL = 'AAPL' # The stock to trade SHORT_WINDOW = 20 # Short moving average window LONG_WINDOW = 50 # Long moving average window STOP_LOSS_PERCENTAGE = 0.02 # 2% stop loss RISK_PER_TRADE = 0.01 # 1% of account balance per trade TRADE_QUANTITY = 1 # Default number of shares (can be adjusted) def get_historical_data(symbol, start_date, end_date): """Fetch historical bar data from Alpaca.""" logging.info(f"Fetching historical data for {symbol}") return api.get_barset(symbol, 'minute', start=start_date, end=end_date).df[symbol] def calculate_moving_averages(df): """Calculate short and long moving averages.""" df['SMA_short'] = df['close'].rolling(window=SHORT_WINDOW).mean() df['SMA_long'] = df['close'].rolling(window=LONG_WINDOW).mean() return df def should_buy(df): """Return True if a buy signal is present.""" return df['SMA_short'].iloc[-1] > df['SMA_long'].iloc[-1] and df['SMA_short'].iloc[-2] <= df['SMA_long'].iloc[-2] def should_sell(df): """Return True if a sell signal is present.""" return df['SMA_short'].iloc[-1] < df['SMA_long'].iloc[-1] and df['SMA_short'].iloc[-2] >= df['SMA_long'].iloc[-2] def get_account_balance(): """Retrieve account cash balance.""" account = api.get_account() return float(account.cash) def calculate_position_size(price): """Calculate position size based on risk per trade.""" balance = get_account_balance() max_risk = balance * RISK_PER_TRADE return int(max_risk / price) def place_order(symbol, qty, side, price=None, stop_price=None): """Place a trade order.""" try: api.submit_order( symbol=symbol, qty=qty, side=side, type='market' if not price else 'limit', time_in_force='gtc', stop_price=stop_price ) logging.info(f"{side.capitalize()} order for {qty} shares of {symbol} placed at {price or 'market price'}.") except Exception as e: logging.error(f"Error placing {side} order for {symbol}: {e}") def execute_trading_strategy(): """Execute trading strategy with moving average crossover and risk management.""" end_date = datetime.now() start_date = end_date - timedelta(days=10) # Retrieve the last 10 days of data for calculations # Fetch historical data and calculate moving averages df = get_historical_data(SYMBOL, start_date.isoformat(), end_date.isoformat()) df = calculate_moving_averages(df) # Retrieve the latest close price current_price = df['close'].iloc[-1] # Check for buy signal if should_buy(df): # Calculate position size and stop loss qty = calculate_position_size(current_price) stop_loss_price = current_price * (1 - STOP_LOSS_PERCENTAGE) logging.info(f"Buy signal detected. Placing buy order for {qty} shares of {SYMBOL} at ${current_price:.2f} with stop loss at ${stop_loss_price:.2f}.") place_order(SYMBOL, qty, 'buy', stop_price=stop_loss_price) # Check for sell signal elif should_sell(df): # Sell the stock if a sell signal is detected positions = api.list_positions() for position in positions: if position.symbol == SYMBOL: qty = int(position.qty) logging.info(f"Sell signal detected. Placing sell order for {qty} shares of {SYMBOL} at ${current_price:.2f}.") place_order(SYMBOL, qty, 'sell') break else: logging.info("No trade signal. Waiting for the next opportunity.") def main(): """Main function to run the trading bot periodically.""" logging.info("Starting trading bot.") while True: try: # Execute trading strategy execute_trading_strategy() # Wait for the next interval (e.g., 15 minutes) time.sleep(900) except Exception as e: logging.error(f"Error in main loop: {e}") time.sleep(60) # Wait 1 minute before retrying in case of error if __name__ == "__main__": main() ``` ### Explanation 1. **API Connection**: - Uses Alpaca's REST API for trading, retrieving historical and real-time data, and executing orders. 2. **Moving Average Crossover Strategy**: - The `calculate_moving_averages` function computes short-term and long-term moving averages. - `should_buy` triggers a buy signal when the short MA crosses above the long MA, indicating an upward trend. - `should_sell` triggers a sell signal when the short MA crosses below the long MA. 3. **Risk Management**: - A stop-loss is applied based on `STOP_LOSS_PERCENTAGE` below the purchase price. - Position sizing is calculated as a portion of the account’s cash balance (1% per trade). 4. **Logging**: - Records important events and errors, allowing you to review the bot’s decisions and outcomes. 5. **Execution Loop**: - Runs every 15 minutes to check for trade signals, making it suitable for intraday trading. ### Important Notes - This bot trades one symbol (`AAPL` in this example) but can be modified to handle multiple. - Paper trading credentials are used here, so real funds aren’t at risk; remember to switch to live trading credentials for real trading. - Market conditions can vary, so thoroughly test this bot and consider refining parameters based on historical testing and performance.