import ccxt
import pandas as pd
import numpy as np
import sqlite3
from apscheduler.schedulers.background import BackgroundScheduler
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
import joblib
from keras.models import Sequential
from keras.layers import LSTM, Dense, Dropout, Bidirectional
import talib

# Initialize exchange
exchange = ccxt.binance()

# Database connection
conn = sqlite3.connect(‘trading_data.db’)

# Load models and scalers
rf_model = joblib.load(‘rf_model.pkl’)
gbm_model = joblib.load(‘gbm_model.pkl’)
rnn_model = Sequential() # Define your model architecture and load weights
rnn_model.load_weights(‘rnn_model.h5’)
scaler = joblib.load(‘scaler.pkl’)
pca = joblib.load(‘pca.pkl’)

def fetch_real_time_data(symbol, timeframe=‘1m’, limit=100):
“”“Fetches real-time OHLCV data”“”
try:
bars = exchange.fetch_ohlcv(symbol, timeframe=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:
print(f"Error fetching real-time data: {e}“)
return None

def preprocess_data(df):
“”“Preprocess incoming data for model prediction””“
df[‘price_change’] = df[‘close’].pct_change()
df[‘volatility’] = df[‘price_change’].rolling(window=20).std()
df[‘rsi’] = talib.RSI(df[‘close’], timeperiod=14)
df[‘sma50’] = talib.SMA(df[‘close’], timeperiod=50)
df[‘sma200’] = talib.SMA(df[‘close’], timeperiod=200)
df[‘bollinger_upper’], df[‘bollinger_middle’], df[‘bollinger_lower’] = talib.BBANDS(df[‘close’], timeperiod=20)
df[‘macd’], df[‘macd_signal’], df[‘macd_hist’] = talib.MACD(df[‘close’])

df.dropna(inplace=True)

features = df[[‘price_change’, ‘volatility’, ‘rsi’, ‘sma50’, ‘sma200’, ‘bollinger_upper’, ‘bollinger_middle’, ‘bollinger_lower’, ‘macd’, ‘macd_signal’, ‘macd_hist’]]
return features


### 2. Model Deployment and Signal Generation

#### Model Prediction and Signal Generation

def generate_realtime_signals(models, feature_df):
rf_model, gbm_model, rnn_model = models

# Preprocessing input
scaled_features = scaler.transform(feature_df)
pca_features = pca.transform(scaled_features)

X_test_array = pca_features.reshape((pca_features.shape[0], 1, pca_features.shape[1]))

rf_preds = rf_model.predict(pca_features)
gbm_preds = gbm_model.predict(pca_features)
rnn_preds = (rnn_model.predict(X_test_array) > 0.5).astype(int).flatten()

# Weighted averaging based on validation set performance
weights = [0.4, 0.4, 0.2] # Assuming these weights are based on validation performance
aggregated_preds = np.array([rf_preds, gbm_preds, rnn_preds]).T
final_preds = [np.round(np.average(preds, weights=weights)).astype(int) for preds in aggregated_preds]

return final_preds[-1] # Return the latest prediction


#### Signal Execution with Risk Management

def execute_trade(signal, symbol, current_price, position, balance):
“””
Execute trades based on signal, with basic risk management
“”“
if signal == 1 and position == 0: # Buy signal
print(f"Buying {symbol}”)
position = 1
entry_price = current_price
balance[‘initial’] = balance[‘current’] # Track initial balance for comparison
elif signal == -1 and position == 0: # Sell signal
print(f"Selling {symbol}“)
position = -1
entry_price = current_price
balance[‘initial’] = balance[‘current’]
elif signal == 0 and position != 0: # Close position
if position == 1: # Closing long position
balance[‘current’] += (current_price - entry_price) * balance[‘initial’] / entry_price
elif position == -1: # Closing short position
balance[‘current’] += (entry_price - current_price) * balance[‘initial’] / entry_price
print(f"Closing position at {current_price}. Updated balance: {balance[‘current’]}”)
position = 0

return position, balance

# Load or initialize balance
balance = {‘initial’: 10000, ‘current’: 10000}


### 3. Real-Time Scheduler and Monitoring

#### Scheduler Setup

def run_trading_strategy(symbol):
df = fetch_real_time_data(symbol)
if df is not None and len(df) > 0:
feature_df = preprocess_data(df)
signal = generate_realtime_signals((rf_model, gbm_model, rnn_model), feature_df)
current_price = df.iloc[-1][‘close’]
global position, balance

position, balance = execute_trade(signal, symbol, current_price, position, balance)
else:
print(“No data available to process.”)

if name == “main”:
symbol = ‘BTC/USDT’
scheduler = BackgroundScheduler()
scheduler.add_job(func=lambda: run_trading_strategy(symbol), trigger=“interval”, minutes=1)
scheduler.start()

print(“Scheduler started. Press Ctrl+C to exit.”)

try:
while True:
pass
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()
print(“Scheduler stopped.”)
import ccxt
import pandas as pd
import logging
import sqlite3
import requests_cache
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from time import sleep
import numpy as np
import traceback
import talib

class CryptoTradingBot:
    def init(self, api_key, api_secret, timeframe=‘1h’, balance=10000, max_risk=0.02, paper_trading=True, config_file=‘config.json’):
        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
        
        # Load configuration
        with open(config_file) as f:
            self.config = json.load(f)
        
        # Setup requests cache to minimize redundant API calls
        self.session = requests_cache.CachedSession(‘ccxt_cache’, backend=‘memory’, expire_after=self.config.get(“cache_expires”, 300))

        # Setup database connection
        self.conn = sqlite3.connect(‘trading_data.db’)
        self.cursor = self.conn.cursor()
        self.setup_database()

        # Fetch symbols
        self.symbols = self.get_usdt_symbols(
            min_volume=self.config.get(“min_volume”),
            min_volatility=self.config.get(“min_volatility”)
        )

    def setup_database(self):
        self.cursor.execute(“”“
        CREATE TABLE IF NOT EXISTS ohlcv (
            symbol TEXT,
            timestamp TEXT,
            open REAL,
            high REAL,
            low REAL,
            close REAL,
            volume REAL,
            PRIMARY KEY (symbol, timestamp)
        )”“”)
        self.cursor.execute(“”“
        CREATE TABLE IF NOT EXISTS indicators (
            symbol TEXT,
            timestamp TEXT,
            indicator_name TEXT,
            indicator_value REAL,
            PRIMARY KEY (symbol, timestamp, indicator_name)
        )”“”)
        self.conn.commit()

    def save_data(self, df, symbol, table_name):
        df[‘symbol’] = symbol
        df.to_sql(table_name, self.conn, if_exists=‘append’, index=False)

    def get_usdt_symbols(self, min_volume=None, min_volatility=None):
        try:
            markets = self.exchange.load_markets()
            usdt_symbols = [symbol for symbol in markets if symbol.endswith(‘/USDT’)]
            filtered_symbols = []

            for symbol in usdt_symbols:
                ticker = self.fetch_real_time_data(symbol)
                if min_volume and ticker[‘quoteVolume’] < min_volume:
                    continue

                if min_volatility:
                    df = self.fetch_data(symbol, limit=100)
                    if df is not None:
                        volatility = df[‘close’].pct_change().std()
                        if volatility < min_volatility:
                            continue
                filtered_symbols.append(symbol)

            logging.info(“Successfully fetched and filtered USDT symbols.”)
            return filtered_symbols
        except (ccxt.ExchangeError, ccxt.NetworkError, Exception) as e:
            logging.error(f"Failed to fetch USDT symbols: {e}“)
            logging.error(traceback.format_exc())
            return []

    def fetch_data(self, symbol, limit=500, retries=3, backoff_factor=1.0):
        for attempt in range(retries):
            try:
                url = self.exchange.urls[‘api’][‘public’] + f”/klines?symbol={symbol.replace(‘/’, ‘’)}&interval={self.timeframe}&limit={limit}“
                response = self.session.get(url)
                response.raise_for_status()
                bars = response.json()
                df = pd.DataFrame(bars, columns=[‘timestamp’, ‘open’, ‘high’, ‘low’, ‘close’, ‘volume’, ‘’, ‘’, ‘’, ‘’, ‘’, ‘’])[[‘timestamp’, ‘open’, ‘high’, ‘low’, ‘close’, ‘volume’]]
                df[‘timestamp’] = pd.to_datetime(df[‘timestamp’], unit=‘ms’)
                if self.validate_data(df):
                    df = self.clean_data(df)
                    self.save_data(df, symbol, ‘ohlcv’)
                    logging.info(f"Successfully fetched data for {symbol} on attempt {attempt + 1}.”)
                    return df
                else:
                    logging.warning(f"Validation failed for data fetched for {symbol}. Attempt {attempt + 1}/{retries}.“)
            except requests.RequestException as e:
                logging.error(f"Network/HTTP error while fetching data for {symbol}: {e}. Retrying in {backoff_factor * (2 ** attempt)} seconds.”)
                logging.error(traceback.format_exc())
            except ccxt.ExchangeError as e:
                logging.error(f"Exchange error while fetching data for {symbol}: {e}. Retrying in {backoff_factor * (2 ** attempt)} seconds.“)
                logging.error(traceback.format_exc())
            except Exception as e:
                logging.error(f"Unexpected error while fetching data for {symbol}: {e}. Retrying in {backoff_factor * (2 ** attempt)} seconds.”)
                logging.error(traceback.format_exc())
            sleep(backoff_factor * (2 ** attempt))  # Exponential backoff
        logging.error(f"Failed to fetch data for {symbol} after {retries} attempts.“)
        return None

    def fetch_real_time_data(self, symbol):
        try:
            ticker = self.exchange.fetch_ticker(symbol)
            logging.info(f"Successfully fetched real-time data for {symbol}.”)
            return ticker
        except (ccxt.NetworkError, ccxt.ExchangeError, Exception) as e:
            logging.error(f"Failed to fetch real-time data for {symbol}: {e}“)
            logging.error(traceback.format_exc())
            return None

    @staticmethod
    def validate_data(df):
        if df.isnull().values.any():
            return False
        if df.empty:
            return False
        return True

    def clean_data(self, df):
        df.replace([np.inf, -np.inf], np.nan, inplace=True)
        df.dropna(inplace=True)
        return df

    def invalidate_cache(self):
        self.session.cache.clear()
        logging.info(“Cache invalidated manually.”)

    def fetch_data_concurrently(self, symbols, limit=500, max_workers=5):
        results = {}
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_symbol = {executor.submit(self.fetch_data, symbol, limit): symbol for symbol in symbols}
            for future in as_completed(future_to_symbol):
                symbol = future_to_symbol[future]
                try:
                    data = future.result()
                    if data is not None:
                        results[symbol] = data
                except Exception as e:
                    logging.error(f"Error fetching data for {symbol}: {e}”)
                    logging.error(traceback.format_exc())
        return results

    def compute_indicators(self, df):
        try:
            indicator_count = 0
            all_indicators = {
                ‘macd’: lambda: self.add_macd(df),
                ‘rsi’: lambda: self.add_rsi(df),
                ‘ema’: lambda: self.add_ema(df),
                ‘atr’: lambda: self.add_atr(df),
                ‘sma50’: lambda: self.add_sma(df, ‘sma50’, 50),
                ‘sma200’: lambda: self.add_sma(df, ‘sma200’, 200),
                ‘bollinger’: lambda: self.add_bollinger_bands(df),
                ‘stochastic’: lambda: self.add_stochastic(df),
                ‘ichimoku’: lambda: self.add_ichimoku_cloud(df),
                ‘aroon’: lambda: self.add_aroon(df),
                ‘cci’: lambda: self.add_cci(df),
                ‘obv’: lambda: self.add_obv(df),
                ‘volatility’: lambda: self.add_volatility(df),
                ‘price_range’: lambda: self.add_price_range(df),
                ‘adx’: lambda: self.add_adx(df),
                ‘williams_r’: lambda: self.add_williams_r(df),
            }
            selected_indicators = self.config.get(“indicators”, [])
            for indicator in selected_indicators[:5]:
                if indicator in all_indicators:
                    all_indicatorsindicator
                    indicator_count += 1
            if self.config.get(“lag_features”, False):
                df = self.add_lag_features(df)

            df.dropna(inplace=True)
            logging.info(f"Successfully computed {indicator_count} indicators.“)
            return df
        except Exception as e:
            logging.error(f"Failed to compute indicators: {e}”)
            logging.error(traceback.format_exc())
            return df

    @staticmethod
    def add_macd(df):
        df[‘macd’], df[‘macd_signal’], _ = talib.MACD(df[‘close’], fastperiod=12, slowperiod=26, signalperiod=9)

    @staticmethod
    def add_rsi(df):
        df[‘rsi’] = talib.RSI(df[‘close’], timeperiod=14)

    @staticmethod
    def add_ema(df):
        df[‘ema’] = talib.EMA(df[‘close’], timeperiod=20)

    @staticmethod
    def add_atr(df):
        df[‘atr’] = talib.ATR(df[‘high’], df[‘low’], df[‘close’], timeperiod=14)

    @staticmethod
    def add_sma(df, column_name, timeperiod):
        df[column_name] = talib.SMA(df[‘close’], timeperiod=timeperiod)

    @staticmethod
    def add_bollinger_bands(df):
        df[‘bollinger_upper’], df[‘bollinger_middle’], df[‘bollinger_lower’] = talib.BBANDS(df[‘close’], timeperiod=20)

    @staticmethod
    def add_stochastic(df):
        df[‘stoch_k’], df[‘stoch_d’] = talib.STOCH(df[‘high’], df[‘low’], df[‘close’])

    def add_ichimoku_cloud(self, df):
        df[‘ichimoku_conversion’], df[‘ichimoku_base’], df[‘ichimoku_span_a’], df[‘ichimoku_span_b’] = self.ichimoku_cloud(df)

    @staticmethod
    def add_aroon(df):
        df[‘aroon_down’], df[‘aroon_up’] = talib.AROON(df[‘high’], df[‘low’], timeperiod=25)
        df[‘aroon_oscillator’] = df[‘aroon_up’] - df[‘aroon_down’]

    @staticmethod
    def add_cci(df):
        df[‘cci’] = talib.CCI(df[‘high’], df[‘low’], df[‘close’], timeperiod=20)

    @staticmethod
    def add_obv(df):
        df[‘obv’] = talib.OBV(df[‘close’], df[‘volume’])

    @staticmethod
    def add_volatility(df):
        df[‘volatility’] = df[‘close’].pct_change().rolling(window=20).std()

    @staticmethod
    def add_price_range(df):
        df[‘price_range’] = df[‘high’] - df[‘low’]

    @staticmethod
    def add_adx(df):
        df[‘adx’] = talib.ADX(df[‘high’], df[‘low’], df[‘close’], timeperiod=14)

    @staticmethod
    def add_williams_r(df):
        df[‘williams_r’] = talib.WILLR(df[‘high’], df[‘low’], df[‘close’], timeperiod=14)

    def add_lag_features(self, df):
        lag_features = [‘close’, ‘volume’, ‘rsi’, ‘ema’]  # Example set of features to create lagged versions for
        for feature in lag_features:
            for lag in range(1, 6):  # Create lagged features from 1 to 5
                df[f’{feature}lag{lag}'] = df[feature].shift(lag)
        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 identify_market_condition(self, df):
        # Market condition criteria
        short_sma = df[‘sma50’].iloc[-1]
        long_sma = df[‘sma200’].iloc[-1]
        atr = df[‘atr’].iloc[-1]
        adx = df[‘adx’].iloc[-1]
        closing_price = df[‘close’].iloc[-1]

        # Volatility threshold for identifying trends
        volatility_threshold = self.config.get(“min_volatility”, 0.02)

        # Bullish markets have short SMA above long SMA, price above long SMA, and low to moderate volatility
        if short_sma > long_sma and closing_price > long_sma and atr < volatility_threshold and adx >= 20:
            return ‘bullish’
        # Bearish markets have short SMA below long SMA, price below long SMA, and low to moderate volatility
        elif short_sma < long_sma and closing_price < long_sma and atr < volatility_threshold and adx >= 20:
            return ‘bearish’
        # Sideways markets have high volatility or low ADX (indicating lack of trend)
        elif atr >= volatility_threshold or adx < 20:
            return ‘sideways’
        # Default to sideways if conditions are unclear
        else:
            return ‘sideways’

    def configure_indicators_for_market_condition(self, df):
        market_condition = self.identify_market_condition(df)
        indicators_by_market_condition = self.config.get(“indicator_selection_criteria”, {}).get(“market_condition”, {})
        selected_indicators = indicators_by_market_condition.get(market_condition, [])

        # Recompute indicators based on the market condition
        for indicator in selected_indicators:
            if indicator == ‘macd’:
                self.add_macd(df)
            elif indicator == ‘rsi’:
                self.add_rsi(df)
            elif indicator == ‘ema’:
                self.add_ema(df)
            elif indicator == ‘atr’:
                self.add_atr(df)
            elif indicator == ‘sma50’:
                self.add_sma(df, ‘sma50’, 50)
            elif indicator == ‘sma200’:
                self.add_sma(df, ‘sma200’, 200)
            elif indicator == ‘bollinger’:
                self.add_bollinger_bands(df)
            elif indicator == ‘stochastic’:
                self.add_stochastic(df)
            elif indicator == ‘ichimoku’:
                self.add_ichimoku_cloud(df)
            elif indicator == ‘aroon’:
                self.add_aroon(df)
            elif indicator == ‘cci’:
                self.add_cci(df)
            elif indicator == ‘obv’:
                self.add_obv(df)
            elif indicator == ‘volatility’:
                self.add_volatility(df)
            elif indicator == ‘price_range’:
                self.add_price_range(df)
            elif indicator == ‘adx’:
                self.add_adx(df)
            elif indicator == ‘williams_r’:
                self.add_williams_r(df)

        df.dropna(inplace=True)
        logging.info(f"Indicators configured for {market_condition} market condition.")
        return df

if name == “main”:
    # Replace with your actual API keys
    binance_api_key = ‘your_binance_api_key’
    binance_api_secret = ‘your_binance_api_secret’

    # Create an instance of CryptoTradingBot
    bot = CryptoTradingBot(
        api_key=binance_api_key,
        api_secret=binance_api_secret,
        timeframe=‘1h’  # User can change this to any supported timeframe (e.g., ‘5m’, ‘1d’)
    )

    # Fetch and display USDT trading symbols with optional volume and volatility filtering
    usdt_symbols = bot.get_usdt_symbols()
    print(“Filtered USDT Trading Symbols:”, usdt_symbols)

    # Fetch and display historical OHLCV data for a sample symbol
    sample_data = bot.fetch_data(‘BTC/USDT’)
    if sample_data is not None:
        sample_data_with_indicators = bot.configure_indicators_for_market_condition(sample_data)
        print(sample_data_with_indicators.head())

    # Manually invalidate the cache (optional)
    bot.invalidate_cache()
Above is code for machine learning models generating signals and code for technical analysis. My goal is to use technical analysis as a secondary filter to validate signals generated by machine learning models.

This is a sound approach to risk management and improving overall strategy performance.

Implementing Signal Confirmation
Here's a basic outline of how to incorporate technical analysis for signal confirmation:

Generate Machine Learning Signals:

Use your existing machine learning models to produce buy, sell, or hold signals.
Fetch Real-Time Data:

Obtain the latest market data for the asset being analyzed.
Calculate Technical Indicators:

Apply relevant technical indicators to the real-time data (e.g., RSI, moving averages, support/resistance levels).
Signal Confirmation:

Compare the machine learning signal with the current technical analysis indicators.
Establish specific criteria for confirming a signal:
Bullish Signal Confirmation: Requires a bullish signal from the machine learning model and a bullish technical setup (e.g., golden cross, RSI oversold).
Bearish Signal Confirmation: Requires a bearish signal from the machine learning model and a bearish technical setup (e.g., death cross, RSI overbought).
Sideways: This is where price will be moving up and down within the support and resistance level and will be buying at support and selling at resistance and selling again at resistance and buy at support. Please note that I will be trading futures mostly.