Crypto Trading Bot in Alpaca. The goal of this story is to create the… | by Trade Mamba | Coinmonks | Jan, 2025

The goal of this story is to create the simplest paper trading bot in Alpaca which will use WebSocket 1-minute bars. The strategy isBuy 0.1 BTC at Market if RSI_14 < 30 and HAVE_NO_POSITIONSell 0.1 BTC at Market if RSI_14 > 70 and HAVE_POSITIONInstall alpaca-py!pip install aplaca-py2. Install pandas_ta!pip install pandas_tafrom google.colab import userdataALPACA_API_KEY=userdata.get(‘ALPACA_API_KEY’)ALPACA_API_SECRET=userdata.get(‘ALPACA_API_SECRET’)Allow nested use of asyncio within Jupyter notebooks, Google Colab, or other environments that already have an event loop running.Written by the legend Ewald De Wit RIPimport nest_asyncio nest_asyncio.apply()from alpaca.trading.client import TradingClienttrading_client = TradingClient(ALPACA_API_KEY, ALPACA_API_SECRET, paper=True)Initialize CryptoDataStreamUsed to manage real-time Crypto Websocket data.from alpaca.data.live import CryptoDataStreamcrypto_data_stream = CryptoDataStream(ALPACA_API_KEY, ALPACA_API_SECRET)Define Event Handler when New Bar ArrivesThis is where we put our trading logicAnalogous to callback in ib_insyncfrom alpaca.trading.requests import MarketOrderRequestfrom alpaca.trading.enums import OrderSide, TimeInForcefrom pandas_ta.momentum import rsibars = []has_position = Falseasync def on_new_bar(bar):global barsglobal has_positionbars.append(bar)if len(bars)<15:print(“Not Enough Data”)else:rsi_value = rsi(util.df(bars[-15:]).close).iloc[-1]print(f”Current RSI is {rsi_value}”)if has_position and rsi_value>=70:print(“RSI is overbought, sell BTC”)trading_client.submit_order(order_data=MarketOrderRequest(symbol=”BTC/USD”,qty=.1,side=OrderSide.SELL,time_in_force=TimeInForce.GTC))has_position = Falseelif not has_position and rsi_value<=30:print(“RSI is oversold, buy BTC”)trading_client.submit_order(order_data=MarketOrderRequest(symbol=”BTC/USD”,qty=.1,side=OrderSide.BUY,time_in_force=TimeInForce.GTC))has_position = TrueNote we used the following helper function util.df to convert a list of bars into a Pandas DataFrameimport pandas as pdclass util:@staticmethoddef df(json_like):if isinstance(json_like, list):json_like = [element.__dict__ for element in json_like]return pd.DataFrame(json_like)else:json_like = json_like.__dict__return pd.DataFrame(json_like.values(), index=json_like.keys())Subscribe to Websocket 1-minute Bars and attach Event Handlercrypto_data_stream.subscribe_bars(on_new_bar, “BTC/USD”)Run the Algo (BLOCKING)crypto_data_stream.run()Run the Algo (NON BLOCKING)import asyncioasyncio.create_task(crypto_data_stream._run_forever())If you are using Jupyter/Colab and still want it to be Interactive this is usefulSample Output: