import os
import pandas as pd
import numpy as np
import talib
import ccxt
import logging
import getpass
import dotenv
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
import tweepy
from newsapi import NewsApiClient
import time
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.preprocessing import StandardScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, LSTM
from transformers import pipeline
import matplotlib.pyplot as plt

# Load .env file if it exists
dotenv.load_dotenv()

# Configuration for logging
if not os.path.exists(‘logs’):
os.makedirs(‘logs’)

logging.basicConfig(filename=‘logs/trading_bot.log’, level=logging.INFO,
format=‘%(asctime)s:%(levelname)s:%(message)s’)


class CryptoTradingBot:
def init(self, api_key, api_secret, twitter_api_key, twitter_api_secret_key, twitter_access_token, twitter_access_token_secret, news_api_key, timeframe=‘1h’, balance=10000, max_risk=0.02, paper_trading=True):
self.exchange = ccxt.binance({
‘apiKey’: api_key,
‘secret’: api_secret,
})
self.timeframe = timeframe
self.balance = balance
self.max_risk = max_risk
self.paper_trading = paper_trading
self.sentiment_analyzer = SentimentIntensityAnalyzer()
self.setup_twitter_api(twitter_api_key, twitter_api_secret_key, twitter_access_token, twitter_access_token_secret)
self.news_api = NewsApiClient(api_key=news_api_key)
self.symbols = self.get_usdt_symbols()
self.sentiment_analyzer_advanced = pipeline(“sentiment-analysis”)
self.rf_model, self.scaler_rf = self.train_random_forest_model()
self.lstm_model, self.scaler_lstm = self.train_lstm_model()

def setup_twitter_api(self, twitter_api_key, twitter_api_secret_key, twitter_access_token, twitter_access_token_secret):
auth = tweepy.OAuthHandler(twitter_api_key, twitter_api_secret_key)
auth.set_access_token(twitter_access_token, twitter_access_token_secret)
self.twitter_api = tweepy.API(auth)

def get_usdt_symbols(self):
markets = self.exchange.load_markets()
usdt_symbols = [symbol for symbol in markets if symbol.endswith(‘/USDT’)]
return usdt_symbols

def fetch_data(self, symbol, limit=500):
try:
bars = self.exchange.fetch_ohlcv(symbol, timeframe=self.timeframe, limit=limit)
df = pd.DataFrame(bars, columns=[‘timestamp’, ‘open’, ‘high’, ‘low’, ‘close’, ‘volume’])
df[‘timestamp’] = pd.to_datetime(df[‘timestamp’], unit=‘ms’)
return df
except Exception as e:
logging.error(f"Failed to fetch data for {symbol}: {e}“)
return pd.DataFrame()

def compute_indicators(self, df):
try:
df[‘macd’], df[‘macd_signal’], _ = talib.MACD(df[‘close’], fastperiod=12, slowperiod=26, signalperiod=9)
df[‘rsi’] = talib.RSI(df[‘close’], timeperiod=14)
df[‘ema’] = talib.EMA(df[‘close’], timeperiod=20)
df[‘atr’] = talib.ATR(df[‘high’], df[‘low’], df[‘close’], timeperiod=14)
df[‘sma50’] = talib.SMA(df[‘close’], timeperiod=50)
df[‘sma200’] = talib.SMA(df[‘close’], timeperiod=200)

# Additional indicators
df[‘bollinger_upper’], df[‘bollinger_middle’], df[‘bollinger_lower’] = talib.BBANDS(df[‘close’], timeperiod=20)
df[‘stoch_k’], df[‘stoch_d’] = talib.STOCH(df[‘high’], df[‘low’], df[‘close’])
df[‘ichimoku_conversion’], df[‘ichimoku_base’], df[‘ichimoku_span_a’], df[‘ichimoku_span_b’] = self.ichimoku_cloud(df)
df.dropna(inplace=True)
return df
except Exception as e:
logging.error(f"Failed to compute indicators: {e}”)
return df

def ichimoku_cloud(self, df):
nine_period_high = df[‘high’].rolling(window=9).max()
nine_period_low = df[‘low’].rolling(window=9).min()
conversion_line = (nine_period_high + nine_period_low) / 2

twenty_six_period_high = df[‘high’].rolling(window=26).max()
twenty_six_period_low = df[‘low’].rolling(window=26).min()
base_line = (twenty_six_period_high + twenty_six_period_low) / 2

leading_span_a = ((conversion_line + base_line) / 2).shift(26)

fifty_two_period_high = df[‘high’].rolling(window=52).max()
fifty_two_period_low = df[‘low’].rolling(window=52).min()
leading_span_b = ((fifty_two_period_high + fifty_two_period_low) / 2).shift(26)

return conversion_line, base_line, leading_span_a, leading_span_b

def fetch_twitter_sentiment(self, keyword=‘Bitcoin’):
try:
tweets = tweepy.Cursor(self.twitter_api.search, q=keyword, lang=“en”).items(100)
tweet_texts = [tweet.text for tweet in tweets]
sentiment_scores = [self.sentiment_analyzer_advanced(text)[0][‘score’] for text in tweet_texts]
avg_sentiment = np.mean(sentiment_scores)
return avg_sentiment
except Exception as e:
logging.error(f"Failed to fetch Twitter sentiment: {e}“)
return 0

def fetch_news_sentiment(self, keyword=‘Bitcoin’):
try:
articles = self.news_api.get_everything(q=keyword, language=‘en’, page_size=10)
article_texts = [article[‘title’] + " " + article[‘description’] for article in articles[‘articles’]]
sentiment_scores = [self.sentiment_analyzer_advanced(text)[0][‘score’] for text in article_texts]
avg_sentiment = np.mean(sentiment_scores)
return avg_sentiment
except Exception as e:
logging.error(f"Failed to fetch news sentiment: {e}”)
return 0

def price_action_signals(self, df):
df[‘higher_highs’] = (df[‘high’] > df[‘high’].shift(1)) & (df[‘high’] > df[‘high’].shift(2))
df[‘higher_lows’] = (df[‘low’] > df[‘low’].shift(1)) & (df[‘low’] > df[‘low’].shift(2))
df[‘lower_highs’] = (df[‘high’] < df[‘high’].shift(1)) & (df[‘high’] < df[‘high’].shift(2))
df[‘lower_lows’] = (df[‘low’] < df[‘low’].shift(1)) & (df[‘low’] < df[‘low’].shift(2))

df[‘uptrend’] = df[‘higher_highs’] & df[‘higher_lows’]
df[‘downtrend’] = df[‘lower_highs’] & df[‘lower_lows’]
df[‘sideways’] = ~(df[‘uptrend’] | df[‘downtrend’])

return df

def generate_signals(self, df):
try:
df = self.price_action_signals(df)
df[‘signal’] = 0

# Technical indicator signals
df[‘tech_signal’] = 0
tech_buy_conditions = [
(df[‘macd’] > df[‘macd_signal’]),
(df[‘rsi’] < 30),
(df[‘close’] > df[‘ema’]),
(df[‘close’] > df[‘sma50’]),
(df[‘close’] > df[‘sma200’]),
(df[‘close’] > df[‘bollinger_lower’]),
(df[‘stoch_k’] < 20)
]
tech_sell_conditions = [
(df[‘macd’] < df[‘macd_signal’]),
(df[‘rsi’] > 70),
(df[‘close’] < df[‘ema’]),
(df[‘close’] < df[‘sma50’]),
(df[‘close’] < df[‘sma200’]),
(df[‘close’] < df[‘bollinger_upper’]),
(df[‘stoch_k’] > 80)
]
df.loc[np.sum(tech_buy_conditions, axis=0) >= 3, ‘tech_signal’] = 1
df.loc[np.sum(tech_sell_conditions, axis=0) >= 3, ‘tech_signal’] = -1

return df
except Exception as e:
logging.error(f"Failed to generate signals: {e}“)
return df

def sentiment_analysis(self, df):
twitter_sentiment = self.fetch_twitter_sentiment()
news_sentiment = self.fetch_news_sentiment()
avg_sentiment = (twitter_sentiment + news_sentiment) / 2
sentiment_signal = 1 if avg_sentiment > 0 else -1 if avg_sentiment < 0 else 0

return sentiment_signal

def train_random_forest_model(self):
df = pd.read_csv(‘historical_data.csv’)
features = [‘macd’, ‘macd_signal’, ‘rsi’, ‘ema’, ‘sma50’, ‘sma200’, ‘bollinger_upper’, ‘bollinger_lower’, ‘stoch_k’, ‘stoch_d’]
X = df[features]
y = df[‘target’]

scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

X_train, X_test, y_train, y_test = train_test_split(X_scaled, y, test_size=0.2, random_state=42)

rf_model = RandomForestClassifier(n_estimators=100, random_state=42)
rf_model.fit(X_train, y_train)

logging.info(f"Random Forest Model Accuracy: {rf_model.score(X_test, y_test)}”)
return rf_model, scaler

def train_lstm_model(self):
df = pd.read_csv(‘historical_data.csv’)
features = [‘macd’, ‘macd_signal’, ‘rsi’, ‘ema’, ‘sma50’, ‘sma200’, ‘bollinger_upper’, ‘bollinger_lower’, ‘stoch_k’, ‘stoch_d’]
X = df[features]
y = df[‘target’]

scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

X_train, X_test, y_train, y_test = train_test_split(X_scaled, y, test_size=0.2, random_state=42)

X_train = np.reshape(X_train, (X_train.shape[0], 1, X_train.shape[1]))
X_test = np.reshape(X_test, (X_test.shape[0], 1, X_test.shape[1]))

lstm_model = Sequential()
lstm_model.add(LSTM(50, return_sequences=True, input_shape=(1, X_train.shape[2])))
lstm_model.add(LSTM(50))
lstm_model.add(Dense(1, activation=‘sigmoid’))

lstm_model.compile(optimizer=‘adam’, loss=‘binary_crossentropy’, metrics=[‘accuracy’])
lstm_model.fit(X_train, y_train, epochs=10, batch_size=32, validation_data=(X_test, y_test))

loss, accuracy = lstm_model.evaluate(X_test, y_test)
logging.info(f"LSTM Model Accuracy: {accuracy}“)

return lstm_model, scaler

def predict_signal(self, df):
features = [‘macd’, ‘macd_signal’, ‘rsi’, ‘ema’, ‘sma50’, ‘sma200’, ‘bollinger_upper’, ‘bollinger_lower’, ‘stoch_k’, ‘stoch_d’]
X = df[features].tail(1)

X_scaled_rf = self.scaler_rf.transform(X)
rf_prediction = self.rf_model.predict(X_scaled_rf)

X_scaled_lstm = self.scaler_lstm.transform(X)
lstm_input = np.reshape(X_scaled_lstm, (X_scaled_lstm.shape[0], 1, X_scaled_lstm.shape[1]))
lstm_prediction = self.lstm_model.predict(lstm_input)
ensemble_prediction = (rf_prediction + lstm_prediction) / 2
signal = 1 if ensemble_prediction > 0.5 else 0 if ensemble_prediction == 0.5 else -1

return signal

def place_order(self, symbol, order_type, side, amount, price=None):
try:
if self.paper_trading:
logging.info(f"Paper trade order: {order_type} {side} {amount} {symbol} at {price}”)
return {
‘symbol’: symbol,
‘order_type’: order_type,
‘side’: side,
‘amount’: amount,
‘price’: price,
‘status’: ‘filled’
}

order = None
if order_type == ‘market’:
order = self.exchange.create_market_order(symbol, side, amount)
elif order_type == ‘limit’:
if price:
order = self.exchange.create_limit_order(symbol, side, amount, price)
else:
logging.error(“Price must be specified for a limit order.”)
return None
elif order_type == ‘stop_limit’:
if price:
order = self.exchange.create_order(symbol, order_type, side, amount, price, params={‘stopPrice’: price})
else:
logging.error(“Price must be specified for a stop-limit order.”)
return None
logging.info(f"Order placed: {order}“)
return order
except (ccxt.NetworkError, ccxt.ExchangeError) as e:
logging.error(f"Failed to place order for {symbol}: {e}”)
return None

def advanced_risk_management(self, df):
atr = df[‘atr’].iloc[-1]
recent_close = df[‘close’].iloc[-1]
position_size = (self.max_risk * self.balance) / atr

stop_loss = recent_close * (1 - self.max_risk)
take_profit = recent_close * (1 + 2 * self.max_risk) # Risk/Reward ratio of 1:2

return position_size, stop_loss, take_profit

def trailing_stop_loss(self, df, entry_price, position_size, trail_offset=0.02):
df[‘highest_close’] = df[‘close’].expanding().max()
df[‘trailing_stop’] = df[‘highest_close’] * (1 - trail_offset)
return df[‘trailing_stop’].iloc[-1]

def exit_strategy(self, df):
current_rsi = df[‘rsi’].iloc[-1]
current_macd = df[‘macd’].iloc[-1]
current_macd_signal = df[‘macd_signal’].iloc[-1]

exit_signal = 0
if current_rsi > 70 or current_macd < current_macd_signal:
exit_signal = -1 # Signal to exit long positions
elif current_rsi < 30 or current_macd > current_macd_signal:
exit_signal = 1 # Signal to exit short positions

return exit_signal

def manage_positions(self, df, symbol, last_signal, position_size):
entry_price = df[‘close’].iloc[-1]
stop_loss, take_profit = self.advanced_risk_management(df)
trailing_stop_price = self.trailing_stop_loss(df, entry_price, position_size)

self.place_order(symbol, ‘market’, ‘buy’ if last_signal == 1 else ‘sell’, position_size)

if last_signal == 1:
while df[‘close’].iloc[-1] > trailing_stop_price:
time.sleep(60) # Check every minute
df = self.fetch_data(symbol)
trailing_stop_price = self.trailing_stop_loss(df, entry_price, position_size)
self.place_order(symbol, ‘market’, ‘sell’, position_size)
elif last_signal == -1:
while df[‘close’].iloc[-1] < trailing_stop_price:
time.sleep(60) # Check every minute
df = self.fetch_data(symbol)
trailing_stop_price = self.trailing_stop_loss(df, entry_price, position_size)
self.place_order(symbol, ‘market’, ‘buy’, position_size)

def backtest(self, symbol, start_date, end_date):
df = self.fetch_data(symbol)
df = df[(df[‘timestamp’] >= start_date) & (df[‘timestamp’] <= end_date)]
df = self.compute_indicators(df)
backtest_results = []
position = None # None: no position, “long”: long position, “short”: short position
entry_price = 0
for index, row in df.iterrows():
df_slice = df.iloc[:index+1]
signal = self.predict_signal(df_slice)
sentiment_signal = self.sentiment_analysis(df_slice)
combined_signal = signal * sentiment_signal
if combined_signal == 1 and position is None: # Enter long position
entry_price = row[‘close’]
position_size, stop_loss, take_profit = self.advanced_risk_management(df_slice)
trailing_stop = self.trailing_stop_loss(df_slice, entry_price, position_size)
position = “long”
backtest_results.append({
‘timestamp’: row[‘timestamp’], ‘action’: ‘buy’, ‘price’: entry_price,
‘stop_loss’: stop_loss, ‘take_profit’: take_profit, ‘trailing_stop’: trailing_stop
})
elif combined_signal == -1 and position is None: # Enter short position
entry_price = row[‘close’]
position_size, stop_loss, take_profit = self.advanced_risk_management(df_slice)
trailing_stop = self.trailing_stop_loss(df_slice, entry_price, position_size)
position = “short”
backtest_results.append({
‘timestamp’: row[‘timestamp’], ‘action’: ‘sell’, ‘price’: entry_price,
‘stop_loss’: stop_loss, ‘take_profit’: take_profit, ‘trailing_stop’: trailing_stop
})
elif position == “long”:
if row[‘close’] >= take_profit or row[‘close’] <= stop_loss or row[‘close’] <= trailing_stop:
# Exit long position
profit = (row[‘close’] - entry_price) * position_size
backtest_results.append({
‘timestamp’: row[‘timestamp’], ‘action’: ‘sell’, ‘price’: row[‘close’],
‘profit’: profit
})
position = None
elif position == “short”:
if row[‘close’] <= take_profit or row[‘close’] >= stop_loss or row[‘close’] >= trailing_stop:
# Exit short position
profit = (entry_price - row[‘close’]) * position_size
backtest_results.append({
‘timestamp’: row[‘timestamp’], ‘action’: ‘buy’, ‘price’: row[‘close’],
‘profit’: profit
})
position = None
# Calculate total profit
summary = pd.DataFrame(backtest_results)
summary[‘profit’].fillna(0, inplace=True)
total_profit = summary[‘profit’].sum()
logging.info(f"Total Profit: {total_profit}")
return summary

def run(self):
results = {‘long’: [], ‘short’: [], ‘sideways’: []}
for symbol in self.symbols:
df = self.fetch_data(symbol)
if df.empty:
continue
df = self.compute_indicators(df)
df = self.generate_signals(df)
sentiment_signal = self.sentiment_analysis(df)
df[‘signal_with_sentiment’] = df[‘tech_signal’] * sentiment_signal
last_signal = df[‘signal_with_sentiment’].iloc[-1]
position_size, stop_loss, take_profit = self.advanced_risk_management(df)
trailing_stop_price = self.trailing_stop_loss(df, entry_price=df[‘close’].iloc[-1], position_size=position_size)
exit_signal = self.exit_strategy(df)
if last_signal == 1:
results[‘long’].append((symbol, df[‘close’].iloc[-1], last_signal, stop_loss, take_profit, trailing_stop_price, exit_signal))
elif last_signal == -1:
results[‘short’].append((symbol, df[‘close’].iloc[-1], last_signal, stop_loss, take_profit, trailing_stop_price, exit_signal))
else:
results[‘sideways’].append((symbol, df[‘close’].iloc[-1], last_signal, stop_loss, take_profit, trailing_stop_price, exit_signal))
for key in results:
results[key] = sorted(results[key], key=lambda x: x[1], reverse=True)[:5]
print(“Top 5 Long Opportunities:”, results[‘long’])
print(“Top 5 Short Opportunities:”, results[‘short’])
print(“Top 5 Sideways Opportunities:”, results[‘sideways’])
for trade_type, opportunities in results.items():
for opportunity in opportunities:
symbol, current_price, signal, stop_loss, take_profit, trailing_stop_price, exit_signal = opportunity
df = self.fetch_data(symbol)
self.manage_positions(df, symbol, signal, position_size)

if name == “main”:
print(“Please enter your API keys and other information.”)
binance_api_key = getpass.getpass("Binance API Key: ")
binance_api_secret = getpass.getpass("Binance API Secret: ")
twitter_api_key = getpass.getpass("Twitter API Key: ")
twitter_api_secret_key = getpass.getpass("Twitter API Secret Key: ")
twitter_access_token = getpass.getpass("Twitter Access Token: ")
twitter_access_token_secret = getpass.getpass("Twitter Access Token Secret: ")
news_api_key = getpass.getpass("News API Key: ")

bot = CryptoTradingBot(
binance_api_key, binance_api_secret,
twitter_api_key, twitter_api_secret_key,
twitter_access_token, twitter_access_token_secret,
news_api_key
)
bot.run()
We broke the code into different parts based on functionalities i.e 1. Environment and Logging Setup
2. Exchange Connectivity and Data Fetching
3. Technical Indicators Calculation
4. Sentiment Analysis
5. Machine Learning Models
6. Risk Management and Order Placement
7. Signal Generation and Strategy Execution
8. Backtesting and Running the Bot,
and work on those parts one by one starting with part 1. We finished part 1 to 3 and were to deal with part 4. What can we include to enhance the performance. Remember when working on each part individually you don’t have to write the whole code, just write the part that we are working but maintain references if any from other parts of code