Advanced Features for Your Trading Bot with Python - Part 2
1. Advanced Strategies
Enhance your bot with more sophisticated trading algorithms, such as:
a. RSI (Relative Strength Index) Strategy
RSI identifies overbought or oversold conditions.
# Calculate RSI
def calculate_rsi(data, window=14):
delta = data['close'].diff(1)
gain = delta.where(delta > 0, 0.0)
loss = -delta.where(delta < 0, 0.0)
avg_gain = gain.rolling(window=window).mean()
avg_loss = loss.rolling(window=window).mean()
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
return rsi
data['RSI'] = calculate_rsi(data)
# Generate RSI signals
data['RSI_Signal'] = 0
data.loc[data['RSI'] < 30, 'RSI_Signal'] = 1 # Buy
data.loc[data['RSI'] > 70, 'RSI_Signal'] = -1 # Sell
Integrate this with your trading logic:
if data['RSI_Signal'][i] == 1:
trade(symbol, 'buy')
elif data['RSI_Signal'][i] == -1:
trade(symbol, 'sell')
b. Mean Reversion Strategy
This assumes prices revert to their mean over time.
# Calculate mean and standard deviation
data['Mean'] = data['close'].rolling(window=20).mean()
data['Std'] = data['close'].rolling(window=20).std()
# Define upper and lower bands
data['Upper_Band'] = data['Mean'] + (2 * data['Std'])
data['Lower_Band'] = data['Mean'] - (2 * data['Std'])
# Plot the bands
plt.plot(data['close'], label='Close Price')
plt.plot(data['Upper_Band'], label='Upper Band', linestyle='--')
plt.plot(data['Lower_Band'], label='Lower Band', linestyle='--')
plt.legend()
plt.show()
# Generate signals
data['MR_Signal'] = 0
data.loc[data['close'] > data['Upper_Band'], 'MR_Signal'] = -1 # Sell
data.loc[data['close'] < data['Lower_Band'], 'MR_Signal'] = 1 # Buy
2. Integrate Other APIs
You can expand beyond Alpaca by integrating APIs like Binance, Interactive Brokers, or Coinbase.
a. Binance API Integration
Install the library: pip install python-binance.
Fetch real-time cryptocurrency data:
from binance.client import Client
# Binance API keys
BINANCE_API_KEY = 'your_binance_api_key'
BINANCE_API_SECRET = 'your_binance_api_secret'
# Initialize Binance client
client = Client(BINANCE_API_KEY, BINANCE_API_SECRET)
# Fetch historical data
symbol = 'BTCUSDT'
data = client.get_historical_klines(symbol, Client.KLINE_INTERVAL_1HOUR, "1 week ago UTC")
# Convert to a DataFrame
import pandas as pd
df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume', ..., 'ignore'])
df['close'] = pd.to_numeric(df['close'])
b. Interactive Brokers API (IBKR)
Install the TWS API: IBKR API Installation Guide.
Automate trading in stocks, options, and futures.
3. Risk Management
Implement safeguards to manage risk:
Position Sizing: Never invest more than a small percentage of your account in one trade.
Stop Loss: Automatically exit losing trades to limit losses.
Take Profit: Lock in profits once a target is reached.
Example:
def place_order(symbol, qty, side, stop_loss=None, take_profit=None):
order = {
'symbol': symbol,
'qty': qty,
'side': side,
'type': 'market',
'time_in_force': 'gtc'
}
if stop_loss or take_profit:
order['stop_loss'] = stop_loss
order['take_profit'] = take_profit
api.submit_order(**order)
4. Backtesting
Use libraries like Backtrader to evaluate strategy performance on historical data.
import backtrader as bt
class MyStrategy(bt.Strategy):
def __init__(self):
self.sma10 = bt.indicators.SimpleMovingAverage(self.data.close, period=10)
self.sma50 = bt.indicators.SimpleMovingAverage(self.data.close, period=50)
def next(self):
if self.sma10 > self.sma50:
self.buy()
elif self.sma10 < self.sma50:
self.sell()
# Load data into Backtrader
data = bt.feeds.PandasData(dataname=data)
# Run backtest
cerebro = bt.Cerebro()
cerebro.addstrategy(MyStrategy)
cerebro.adddata(data)
cerebro.run()
cerebro.plot()
5. Deployment
Cloud Deployment: Host your bot on cloud platforms like AWS, Google Cloud, or Heroku.
Cron Jobs: Schedule your bot to run at specific intervals using cron or task schedulers like APScheduler.
6. Dashboard for Monitoring
Create a dashboard for real-time monitoring using Flask or Dash.
from flask import Flask, render_template
app = Flask(__name__)
@app.route('/')
def home():
account = api.get_account()
return f"Account Balance: ${account.equity}"
if __name__ == '__main__':
app.run(debug=True)
Comments
Post a Comment