app.py

# Install TA-Lib (see instructions above) then: pip install TA-Lib
import ccxt
import numpy as np
import pandas as pd
import time
from sklearn.neighbors import KNeighborsClassifier
from scipy.linalg import svd
import gradio as gr
import concurrent.futures
import traceback
from datetime import datetime, timezone, timedelta
import logging
import sys
import talib # Import TA-Lib
import threading

# --- Setup Logging ---
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - [%(threadName)s:%(funcName)s] - %(message)s',
    stream=sys.stdout
)
logging.getLogger().handlers[0].flush = sys.stdout.flush

# --- Parameters ---
L = 10
LAG = 11
MINUTES_PER_HOUR = 60
PREDICTION_WINDOW_HOURS = 2
TRAINING_WINDOW_HOURS = 12
TOTAL_WINDOW_HOURS = TRAINING_WINDOW_HOURS + PREDICTION_WINDOW_HOURS
K = TRAINING_WINDOW_HOURS * MINUTES_PER_HOUR # 720
WINDOW = TOTAL_WINDOW_HOURS * MINUTES_PER_HOUR # 840
FEATURES = ['open', 'high', 'low', 'close', 'volume']
D = 5
OVERLAP_STEP = 60
MIN_TRAINING_EXAMPLES = 20
MAX_COINS_TO_DISPLAY = 10
USE_SYNTHETIC_DATA_FOR_LOW_VOLUME = False
NUM_WORKERS_TRAINING = 4
NUM_WORKERS_PREDICTION = 10

# --- TA & Risk Parameters ---
TA_DATA_POINTS = 200 # Candles needed for TA calculation
RSI_PERIOD = 14
MACD_FAST = 12
MACD_SLOW = 26
MACD_SIGNAL = 9
ATR_PERIOD = 14
CONFIDENCE_THRESHOLD = 0.65 # Min confidence for Rise signal
TP1_ATR_MULTIPLIER = 1.5
TP2_ATR_MULTIPLIER = 3.0
SL_ATR_MULTIPLIER = 1.0

# --- CCXT Initialization ---
try:
    exchange = ccxt.bitget({
        'enableRateLimit': True,
        'rateLimit': 1100,
        'timeout': 45000,
        'options': {'adjustForTimeDifference': True}
    })
    logging.info(f"Initialized {exchange.id} exchange.")
except Exception as e:
    logging.exception("FATAL: Could not initialize CCXT exchange.")
    sys.exit()

# --- Global Caches and Variables ---
markets_cache = None
last_markets_update = None
data_cache = {}
trained_models = {}
last_update_time = datetime.now(timezone.utc)

# --- Functions ---

def format_datetime(dt, default="N/A"):
    # (Keep this function as is)
    if pd.isna(dt) or dt is None:
        return default
    try:
        if isinstance(dt, (datetime, pd.Timestamp)):
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=timezone.utc)
            return dt.strftime('%Y-%m-%d %H:%M:%S %Z')
        else:
            return str(dt)
    except Exception:
        return default

def get_all_usdt_pairs():
    # (Keep this function as is - no changes needed)
    global markets_cache, last_markets_update
    current_time = time.time()
    cache_duration = 3600 # 1 hour

    if markets_cache is not None and last_markets_update is not None:
        if current_time - last_markets_update < cache_duration:
            logging.info("Using cached markets list.")
            if isinstance(markets_cache, list) and markets_cache:
                 return markets_cache
            else:
                 logging.warning("Cached market list was invalid, fetching fresh.")


    logging.info("Fetching markets from Bitget...")
    try:
        exchange.load_markets(reload=True)
        all_symbols = list(exchange.markets.keys())
        usdt_pairs = [
            symbol for symbol in all_symbols
            if isinstance(symbol, str)
            and symbol.endswith('/USDT')
            and exchange.markets.get(symbol, {}).get('active', False)
            and exchange.markets.get(symbol, {}).get('spot', False)
            and 'LEVERAGED' not in exchange.markets.get(symbol, {}).get('type', 'spot').upper()
            and not exchange.markets.get(symbol, {}).get('inverse', False)
        ]
        logging.info(f"Found {len(usdt_pairs)} active USDT spot pairs initially.")
        if not usdt_pairs:
             logging.warning("No active USDT spot pairs found.")
             return ['BTC/USDT', 'ETH/USDT', 'SOL/USDT']

        logging.info(f"Fetching tickers for {len(usdt_pairs)} pairs for volume sorting...")
        volumes = {}
        symbols_to_fetch = usdt_pairs[:]
        fetched_tickers = {}
        try:
            if exchange.has['fetchTickers']:
                 batch_size_tickers = 100
                 for i in range(0, len(symbols_to_fetch), batch_size_tickers):
                      batch_symbols = symbols_to_fetch[i:i+batch_size_tickers]
                      logging.info(f"Fetching ticker batch {i//batch_size_tickers + 1}/{ (len(symbols_to_fetch) + batch_size_tickers -1)//batch_size_tickers }...")
                      retries = 2
                      for attempt in range(retries):
                           try:
                                batch_tickers = exchange.fetch_tickers(symbols=batch_symbols)
                                fetched_tickers.update(batch_tickers)
                                time.sleep(exchange.rateLimit / 1000 * 1.5) # Add delay
                                break
                           except (ccxt.RequestTimeout, ccxt.NetworkError) as e_timeout:
                                logging.warning(f"Ticker fetch timeout/network error on attempt {attempt+1}/{retries}: {e_timeout}, retrying after delay...")
                                time.sleep(3 * (attempt + 1))
                           except ccxt.RateLimitExceeded:
                                logging.warning(f"Rate limit exceeded fetching tickers, sleeping...")
                                time.sleep(10 * (attempt+1)) # Longer sleep for rate limit
                           except Exception as e_ticker:
                                logging.error(f"Error fetching ticker batch (attempt {attempt+1}): {e_ticker}")
                                if attempt == retries - 1: raise # Rethrow last error
                                time.sleep(2 * (attempt + 1))

                 logging.info(f"Fetched {len(fetched_tickers)} tickers using fetchTickers.")
            else:
                 raise ccxt.NotSupported("fetchTickers not supported/enabled. Volume sorting requires it.")

        except Exception as e:
             logging.exception(f"Could not fetch tickers for volume sorting: {e}. Volume sorting unavailable.")
             markets_cache = usdt_pairs[:MAX_COINS_TO_DISPLAY]
             last_markets_update = current_time
             logging.warning(f"Returning top {len(markets_cache)} unsorted pairs due to ticker error.")
             return markets_cache

        for symbol, ticker in fetched_tickers.items():
            try:
                quote_volume = ticker.get('info', {}).get('quoteVolume') # Prefer quoteVolume if available
                last_price = ticker.get('last')
                base_volume = ticker.get('baseVolume')

                # Ensure values are convertible to float before calculation
                valid_last = last_price is not None
                valid_base = base_volume is not None
                valid_quote = quote_volume is not None

                if valid_quote:
                    volumes[symbol] = float(quote_volume)
                elif valid_base and valid_last:
                    volumes[symbol] = float(base_volume) * float(last_price)
                else:
                     volumes[symbol] = 0
            except (TypeError, ValueError, KeyError, AttributeError) as e:
                 logging.warning(f"Could not parse volume/price for {symbol} from ticker: {ticker}. Error: {e}")
                 volumes[symbol] = 0

        valid_volume_pairs = {k: v for k, v in volumes.items() if v > 0}
        logging.info(f"Found {len(valid_volume_pairs)} pairs with non-zero volume.")

        if not valid_volume_pairs:
             logging.warning("No pairs with valid volume found. Returning default list.")
             return ['BTC/USDT', 'ETH/USDT', 'SOL/USDT']

        sorted_pairs = sorted(valid_volume_pairs.items(), key=lambda item: item[1], reverse=True)
        num_pairs_to_take = min(MAX_COINS_TO_DISPLAY, len(sorted_pairs))
        top_pairs = [pair[0] for pair in sorted_pairs[:num_pairs_to_take]]
        logging.info(f"Selected Top {len(top_pairs)} pairs by volume. Top 5: {[p[0] for p in sorted_pairs[:5]]}")

        markets_cache = top_pairs
        last_markets_update = current_time
        return top_pairs

    except ccxt.NetworkError as e:
        logging.error(f"Network error getting USDT pairs: {e}")
    except ccxt.ExchangeError as e:
         logging.error(f"Exchange error getting USDT pairs: {e}")
    except Exception as e:
        logging.exception("General error getting USDT pairs.")

    logging.warning("Error fetching markets, returning default fallback list.")
    return ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'BNB/USDT', 'XRP/USDT']

def clean_and_process_ohlcv(ohlcv_list, symbol, expected_candles):
    # (Keep this function as is - no changes needed)
    if not ohlcv_list:
        return None
    try:
        df = pd.DataFrame(ohlcv_list, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        initial_len = len(df)
        if initial_len == 0: return None

        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
        df = df.drop_duplicates(subset=['timestamp'])
        df = df.sort_values('timestamp')
        len_after_dupes = len(df)

        numeric_cols = ['open', 'high', 'low', 'close', 'volume']
        for col in numeric_cols:
            df[col] = pd.to_numeric(df[col], errors='coerce')

        # Drop rows with NaN in essential price/volume features needed for TA-Lib
        df = df.dropna(subset=numeric_cols)
        len_after_na = len(df)

        df.reset_index(drop=True, inplace=True)

        logging.debug(f"Data cleaning for {symbol}: Initial Fetched={initial_len}, AfterDupes={len_after_dupes}, AfterNA={len_after_na}")

        if len(df) >= expected_candles:
            final_df = df.iloc[-expected_candles:].copy() # Take the most recent ones
            return final_df
        else:
            return None

    except Exception as e:
        logging.exception(f"Error processing DataFrame for {symbol}")
        return None

def fetch_historical_data(symbol, timeframe='1m', total_candles=WINDOW):
    # (Keep this function as is - no changes needed)
    cache_key = f"{symbol}_{timeframe}_{total_candles}"
    current_time = time.time()
    cache_validity_seconds = 300 # 5 minutes

    if cache_key in data_cache:
        cache_time, cached_data = data_cache[cache_key]
        if current_time - cache_time < cache_validity_seconds:
            if isinstance(cached_data, pd.DataFrame) and len(cached_data) == total_candles:
                 logging.debug(f"Using valid cached data for {symbol} ({len(cached_data)} candles)")
                 return cached_data.copy()
            else:
                 logging.warning(f"Cache for {symbol} invalid or wrong size ({len(cached_data) if isinstance(cached_data, pd.DataFrame) else 'N/A'} vs {total_candles}), fetching fresh.")
                 if cache_key in data_cache: del data_cache[cache_key]

    if not exchange.has['fetchOHLCV']:
        logging.error(f"Exchange {exchange.id} does not support fetchOHLCV.")
        return None

    logging.debug(f"Fetching {total_candles} candles for {symbol} (timeframe: {timeframe})")
    final_df = None
    fetch_start_time = time.time()
    duration_ms = exchange.parse_timeframe(timeframe) * 1000
    now_ms = exchange.milliseconds()

    # --- Strategy 1: Try Single Large Fetch ---
    single_fetch_limit = total_candles + 200 # Buffer
    single_fetch_since = now_ms - single_fetch_limit * duration_ms
    try:
        ohlcv_list = exchange.fetch_ohlcv(symbol, timeframe, limit=single_fetch_limit, since=single_fetch_since)
        if ohlcv_list:
            processed_df = clean_and_process_ohlcv(ohlcv_list, symbol, total_candles)
            if processed_df is not None and len(processed_df) == total_candles:
                 final_df = processed_df
    except ccxt.RateLimitExceeded as e:
        logging.warning(f"Rate limit hit during single fetch for {symbol}, falling back: {e}")
        time.sleep(5)
    except (ccxt.RequestTimeout, ccxt.NetworkError) as e:
         logging.warning(f"Timeout/Network error during single fetch for {symbol}, falling back: {e}")
         time.sleep(2)
    except ccxt.ExchangeNotAvailable as e:
         logging.error(f"Exchange not available during fetch for {symbol}: {e}")
         return None
    except ccxt.AuthenticationError as e:
         logging.error(f"Authentication error fetching {symbol}: {e}")
         return None
    except ccxt.ExchangeError as e:
         logging.warning(f"Exchange error during single fetch for {symbol}, falling back: {e}")
    except Exception as e:
        logging.exception(f"Unexpected error during single fetch for {symbol}, falling back.")

    # --- Strategy 2: Fallback to Iterative Chunking ---
    if final_df is None:
        logging.debug(f"Falling back to iterative chunk fetching for {symbol}.")
        limit_per_call = exchange.safe_integer(exchange.limits.get('fetchOHLCV', {}), 'max', 1000)
        limit_per_call = min(limit_per_call, 1000)
        all_ohlcv_chunks = []
        required_start_time_ms = now_ms - (total_candles + 5) * duration_ms
        current_chunk_end_time_ms = now_ms
        max_chunk_attempts = 15
        attempts = 0

        while attempts < max_chunk_attempts:
            attempts += 1
            oldest_ts_in_hand = all_ohlcv_chunks[0][0] if all_ohlcv_chunks else current_chunk_end_time_ms
            if oldest_ts_in_hand <= required_start_time_ms:
                logging.debug(f"Chunking: Collected enough historical range for {symbol}.")
                break

            fetch_limit = limit_per_call
            chunk_fetch_since = oldest_ts_in_hand - fetch_limit * duration_ms
            params = {}
            try:
                ohlcv_chunk = exchange.fetch_ohlcv(symbol, timeframe, since=chunk_fetch_since, limit=fetch_limit, params=params)
                if not ohlcv_chunk:
                    logging.debug(f"Chunking: No more data received for {symbol} from API.")
                    break

                new_chunk = [c for c in ohlcv_chunk if c[0] < oldest_ts_in_hand]
                if not new_chunk:
                    break

                new_chunk.sort(key=lambda x: x[0])
                all_ohlcv_chunks = new_chunk + all_ohlcv_chunks

                if len(new_chunk) < limit_per_call // 20 and attempts > 5:
                    logging.warning(f"Chunking: Received very few new candles ({len(new_chunk)}) repeatedly for {symbol}.")
                    break
                time.sleep(exchange.rateLimit / 1000 * 1.1)

            except ccxt.RateLimitExceeded as e:
                logging.warning(f"Rate limit hit during chunking for {symbol}, sleeping 10s: {e}")
                time.sleep(10 * (attempts/3 + 1))
            except (ccxt.NetworkError, ccxt.RequestTimeout) as e:
                 logging.error(f"Network/Timeout error during chunking for {symbol}: {e}. Stopping.")
                 break
            except ccxt.ExchangeError as e:
                 logging.error(f"Exchange error during chunking for {symbol}: {e}. Stopping.")
                 break
            except Exception as e:
                logging.exception(f"Generic error during chunking fetch for {symbol}")
                break

        if attempts >= max_chunk_attempts:
             logging.warning(f"Max chunk fetch attempts reached for {symbol}.")

        if all_ohlcv_chunks:
             processed_df = clean_and_process_ohlcv(all_ohlcv_chunks, symbol, total_candles)
             if processed_df is not None and len(processed_df) == total_candles:
                final_df = processed_df
        else:
             logging.error(f"No data obtained from chunk fetching for {symbol}.")

    # --- Final Check and Caching ---
    if final_df is not None and len(final_df) == total_candles:
        expected_cols = ['timestamp', 'open', 'high', 'low', 'close', 'volume']
        if all(col in final_df.columns for col in expected_cols):
            data_cache[cache_key] = (current_time, final_df.copy())
            return final_df
        else:
            logging.error(f"Final DataFrame for {symbol} missing expected columns. Won't cache.")
            return None
    else:
        logging.error(f"Failed to fetch exactly {total_candles} candles for {symbol}. Found: {len(final_df) if final_df is not None else 0}")
        return None

# --- Embedding, LLT, Normalize, Training Prep (Largely unchanged) ---
# Keep create_embedding, llt_transform, normalize_data, prepare_training_data, train_model
# as they don't depend on the TA library choice.

def create_embedding(data, l=L, lag=LAG):
    # (Keep this function as is)
    n = len(data)
    rows = n - (l - 1) * lag
    if rows <= 0:
        logging.debug(f"Cannot create embedding: data length {n} too short for L={l}, Lag={lag}")
        return np.array([])
    A = np.zeros((rows, l))
    try:
        for t in range(rows):
            indices = t + np.arange(l) * lag
            A[t] = data[indices]
        return A
    except IndexError as e:
        logging.error(f"IndexError during embedding: n={n}, l={l}, lag={lag}. Error: {e}")
        return np.array([])
    except Exception as e:
        logging.exception("Error in create_embedding")
        return np.array([])

def llt_transform(X_train, y_train, X_test):
    # (Keep this function as is)
    if not isinstance(X_train, np.ndarray) or X_train.ndim != 3 or \
       not isinstance(y_train, np.ndarray) or y_train.ndim != 1 or \
       not isinstance(X_test, np.ndarray) or (X_test.size > 0 and X_test.ndim != 3):
        logging.error(f"LLT input type/shape error.")
        return np.array([]), np.array([])
    if X_train.shape[0] != y_train.shape[0]:
        logging.error(f"LLT input mismatch: len(X_train) != len(y_train)")
        return np.array([]), np.array([])
    if X_train.size == 0 or y_train.size == 0:
        logging.error("LLT requires non-empty training data.")
        return np.array([]), np.array([])
    if X_test.size > 0 and X_test.shape[1:] != X_train.shape[1:]:
         logging.error(f"LLT train/test shape mismatch")
         return np.array([]), np.array([])

    try:
        num_features = X_train.shape[2]
        if num_features != len(FEATURES):
             logging.error(f"LLT: Feature count mismatch.")
             return np.array([]), np.array([])

        V = {j: {'0': [], '1': []} for j in range(num_features)}
        laws_computed_count = {j: {'0': 0, '1': 0} for j in range(num_features)}

        for i in range(len(X_train)):
            label = str(int(y_train[i]))
            if label not in ['0', '1']: continue
            for j in range(num_features):
                feature_data = X_train[i, :, j]
                A = create_embedding(feature_data, l=L, lag=LAG)
                if A.shape[0] < L: continue
                if np.isnan(A).any() or np.isinf(A).any(): continue
                try:
                    S = A.T @ A
                    if np.isnan(S).any() or np.isinf(S).any(): continue
                    U, s, Vt = svd(S, full_matrices=False)
                    if Vt.shape[0] < L or Vt.shape[1] != L: continue
                    if s[-1] < 1e-9: continue
                    v = Vt[-1]
                    norm = np.linalg.norm(v)
                    if norm < 1e-9: continue
                    V[j][label].append(v / norm)
                    laws_computed_count[j][label] += 1
                except np.linalg.LinAlgError: pass
                except Exception: pass

        valid_laws_exist = False
        for j in V:
            for c in ['0', '1']:
                if laws_computed_count[j][c] > 0:
                    valid_vecs = [vec for vec in V[j][c] if isinstance(vec, np.ndarray) and vec.shape == (L,)]
                    if not valid_vecs:
                         V[j][c] = np.zeros((L, 0))
                         continue
                    try:
                        V[j][c] = np.array(valid_vecs).T
                        if V[j][c].shape[0] != L:
                             V[j][c] = np.zeros((L, 0))
                        else:
                             valid_laws_exist = True
                    except Exception: V[j][c] = np.zeros((L, 0))
                else: V[j][c] = np.zeros((L, 0))

        if not valid_laws_exist:
             logging.error("LLT ERROR: No valid laws computed.")
             return np.array([]), np.array([])

        def transform_instance(X_instance):
            transformed_features = []
            if X_instance.ndim != 2 or X_instance.shape[0] != K or X_instance.shape[1] != num_features:
                return np.zeros(num_features * 2 * D)
            for j in range(num_features):
                feature_data = X_instance[:, j]
                A = create_embedding(feature_data, l=L, lag=LAG)
                if A.shape[0] < L:
                    transformed_features.extend([0.0] * (2 * D))
                    continue
                if np.isnan(A).any() or np.isinf(A).any():
                    transformed_features.extend([0.0] * (2 * D))
                    continue
                try:
                    S = A.T @ A
                    if np.isnan(S).any() or np.isinf(S).any():
                        transformed_features.extend([0.0] * (2 * D))
                        continue
                    for c in ['0', '1']:
                        if V[j][c].shape[1] == 0:
                            transformed_features.extend([0.0] * D)
                            continue
                        S_V = S @ V[j][c]
                        if S_V.size == 0 or np.isnan(S_V).any() or np.isinf(S_V).any():
                            transformed_features.extend([0.0] * D)
                            continue
                        variances = np.var(S_V, axis=0)
                        if variances.size == 0:
                            transformed_features.extend([0.0] * D)
                            continue
                        variances = np.nan_to_num(variances, nan=np.finfo(variances.dtype).max, posinf=np.finfo(variances.dtype).max, neginf=np.finfo(variances.dtype).max)
                        num_vars_available = variances.size
                        num_vars_to_select = min(D, num_vars_available)
                        smallest_indices = np.argpartition(variances, num_vars_to_select -1)[:num_vars_to_select]
                        smallest_vars = np.sort(variances[smallest_indices])
                        padded_vars = np.pad(smallest_vars, (0, D - num_vars_to_select), 'constant', constant_values=0.0)
                        if np.isnan(padded_vars).any() or np.isinf(padded_vars).any():
                            padded_vars = np.nan_to_num(padded_vars, nan=0.0, posinf=0.0, neginf=0.0)
                        transformed_features.extend(padded_vars)
                except Exception:
                     current_len = len(transformed_features)
                     expected_len_after_feature = (j + 1) * 2 * D
                     num_missing = expected_len_after_feature - current_len
                     if num_missing > 0: transformed_features.extend([0.0] * num_missing)
                     transformed_features = transformed_features[:expected_len_after_feature]

            correct_len = num_features * 2 * D
            if len(transformed_features) != correct_len:
                if len(transformed_features) < correct_len: transformed_features.extend([0.0] * (correct_len - len(transformed_features)))
                else: transformed_features = transformed_features[:correct_len]
            return np.array(transformed_features)

        X_train_t = np.array([transform_instance(X) for X in X_train])
        X_test_t = np.array([])
        if X_test.size > 0: X_test_t = np.array([transform_instance(X) for X in X_test])

        expected_dim = num_features * 2 * D
        if X_train_t.shape[0] != len(X_train) or (X_train_t.size > 0 and X_train_t.shape[1] != expected_dim):
             logging.error(f"LLT Train transform resulted in unexpected shape.")
             return np.array([]), np.array([])
        if X_test.size > 0 and (X_test_t.shape[0] != len(X_test) or (X_test_t.size > 0 and X_test_t.shape[1] != expected_dim)):
             logging.error(f"LLT Test transform resulted in unexpected shape.")
             return X_train_t, np.array([])

        return X_train_t, X_test_t
    except Exception as e:
        logging.exception("Error in llt_transform function")
        return np.array([]), np.array([])

def normalize_data(df):
    # (Keep this function as is)
    normalized_df = df.copy()
    if not isinstance(df, pd.DataFrame):
        logging.error("Normalize_data received non-DataFrame input.")
        return None
    for feature in FEATURES:
        if feature == 'timestamp': continue
        if feature not in df.columns:
             normalized_df[feature] = 0.0
             continue
        if pd.api.types.is_numeric_dtype(df[feature]):
            mean = df[feature].mean()
            std = df[feature].std()
            if std is not None and not pd.isna(std) and std > 1e-9:
                normalized_df[feature] = (df[feature] - mean) / std
            else:
                normalized_df[feature] = 0.0
            if normalized_df[feature].isnull().any():
                 normalized_df[feature] = normalized_df[feature].fillna(0.0)
        else:
             normalized_df[feature] = 0.0
    return normalized_df

def generate_synthetic_data(symbol, total_candles=WINDOW):
    # (Keep this function as is)
    logging.info(f"Generating synthetic data for {symbol} ({total_candles} candles)")
    np.random.seed(int(time.time() * 1000) % (2**32 - 1))
    end_time = pd.Timestamp.now(tz='UTC')
    timestamps = pd.date_range(end=end_time, periods=total_candles, freq='T')
    volatility = np.random.uniform(0.005, 0.03)
    base_price = np.random.uniform(1, 5000)
    prices = [base_price]
    for _ in range(1, total_candles):
        change = np.random.normal(0, volatility / np.sqrt(1440))
        prices.append(prices[-1] * (1 + change))
    prices = np.maximum(0.01, prices)
    close_prices = np.array(prices)
    open_prices = close_prices * (1 + np.random.normal(0, volatility / np.sqrt(1440) / 2, total_candles))
    high_prices = np.maximum(close_prices, open_prices) * (1 + np.random.uniform(0, volatility / np.sqrt(1440), total_candles))
    low_prices = np.minimum(close_prices, open_prices) * (1 - np.random.uniform(0, volatility / np.sqrt(1440), total_candles))
    high_prices = np.maximum.reduce([high_prices, open_prices, close_prices])
    low_prices = np.minimum.reduce([low_prices, open_prices, close_prices])
    volumes = np.random.poisson(base_price * np.random.uniform(1, 10)) * (1 + np.abs(np.diff(close_prices, prepend=close_prices[0])) / close_prices * 5)
    volumes = np.maximum(1, volumes)
    df = pd.DataFrame({
        'timestamp': timestamps, 'open': open_prices, 'high': high_prices,
        'low': low_prices, 'close': close_prices, 'volume': volumes
    })
    for col in FEATURES: df[col] = pd.to_numeric(df[col])
    df.reset_index(drop=True, inplace=True)
    return df

def prepare_training_data(symbol, total_candles_to_fetch=WINDOW + OVERLAP_STEP * 20):
    # (Keep this function as is)
    logging.info(f"Preparing training data for {symbol}...")
    try:
        required_base_candles = WINDOW
        estimated_candles_needed = required_base_candles + (MIN_TRAINING_EXAMPLES * 2) * OVERLAP_STEP + 500
        fetch_candle_count = max(WINDOW + 500, estimated_candles_needed)

        logging.info(f"Fetching {fetch_candle_count} candles for {symbol} training prep...")
        df = fetch_historical_data(symbol, timeframe='1m', total_candles=fetch_candle_count)

        if df is None or len(df) < WINDOW:
            logging.error(f"Insufficient data fetched for {symbol} ({len(df) if df is not None else 0} < {WINDOW}).")
            if USE_SYNTHETIC_DATA_FOR_LOW_VOLUME:
                 logging.warning(f"Attempting synthetic data generation for {symbol}.")
                 df = generate_synthetic_data(symbol, total_candles=WINDOW + OVERLAP_STEP * 10)
                 if df is None or len(df) < WINDOW:
                      logging.error(f"Synthetic data generation failed or insufficient for {symbol}.")
                      return None, None
                 else: logging.info(f"Using synthetic data ({len(df)} points) for {symbol}.")
            else: return None, None

        df_normalized = normalize_data(df)
        if df_normalized is None:
             logging.error(f"Normalization failed for {symbol}.")
             return None, None
        if df_normalized[FEATURES].isnull().any().any():
             logging.warning(f"NaN values found after normalization for {symbol}. Filling with 0.")
             df_normalized = df_normalized.fillna(0.0)

        X, y = [], []
        end_index = len(df)
        start_index = WINDOW
        num_windows_created = 0

        for i in range(end_index, start_index - 1, -OVERLAP_STEP):
            window_end_idx = i
            window_start_idx = i - WINDOW
            if window_start_idx < 0: continue

            window_orig = df.iloc[window_start_idx:window_end_idx]
            window_norm = df_normalized.iloc[window_start_idx:window_end_idx]

            if len(window_orig) != WINDOW or len(window_norm) != WINDOW: continue

            input_data_norm = window_norm.iloc[:K][FEATURES].values
            if input_data_norm.shape[0] != K or input_data_norm.shape[1] != len(FEATURES): continue
            if np.isnan(input_data_norm).any(): continue

            start_price_iloc_idx = K - 1
            end_price_iloc_idx = WINDOW - 1
            start_price = window_orig['close'].iloc[start_price_iloc_idx]
            end_price = window_orig['close'].iloc[end_price_iloc_idx]

            if pd.isna(start_price) or pd.isna(end_price) or start_price <= 0: continue

            X.append(input_data_norm)
            y.append(1 if end_price > start_price else 0)
            num_windows_created += 1

        if not X:
            logging.error(f"No valid windows created for {symbol}.")
            return None, None

        X = np.array(X)
        y = np.array(y)
        unique_classes, class_counts = np.unique(y, return_counts=True)
        class_dist_str = ", ".join([f"Class {cls}: {count}" for cls, count in zip(unique_classes, class_counts)])
        logging.info(f"Class distribution BEFORE balancing for {symbol}: {class_dist_str}")

        if len(unique_classes) < 2:
            logging.error(f"ONLY ONE CLASS ({unique_classes[0]}) present for {symbol}.")
            return None, None

        min_class_count = min(class_counts)
        if min_class_count * 2 < MIN_TRAINING_EXAMPLES:
             logging.error(f"Minority class count ({min_class_count}) too low for {symbol}.")
             return None, None

        samples_per_class = min_class_count
        balanced_indices = []
        for class_val in unique_classes:
            class_indices = np.where(y == class_val)[0]
            num_to_choose = min(samples_per_class, len(class_indices))
            chosen_indices = np.random.choice(class_indices, size=num_to_choose, replace=False)
            balanced_indices.extend(chosen_indices)

        np.random.shuffle(balanced_indices)
        X_balanced = X[balanced_indices]
        y_balanced = y[balanced_indices]
        final_unique, final_counts = np.unique(y_balanced, return_counts=True)
        logging.info(f"Balanced dataset for {symbol}: {len(X_balanced)} instances. Final counts: {dict(zip(final_unique, final_counts))}")

        if len(X_balanced) < MIN_TRAINING_EXAMPLES:
             logging.error(f"Insufficient data ({len(X_balanced)}) for {symbol} AFTER balancing.")
             return None, None
        if X_balanced.ndim != 3 or X_balanced.shape[0] == 0 or X_balanced.shape[1] != K or X_balanced.shape[2] != len(FEATURES):
             logging.error(f"Final balanced data has unexpected shape {X_balanced.shape} for {symbol}.")
             return None, None

        return X_balanced, y_balanced
    except Exception as e:
        logging.exception(f"Error preparing training data for {symbol}")
        return None, None

def train_model(symbol):
    # (Keep this function as is)
    logging.info(f"--- Attempting to train model for {symbol} ---")
    np.random.seed(int(time.time()) % (2**32 - 1))

    X, y = prepare_training_data(symbol)
    if X is None or y is None:
        logging.error(f"Failed to prepare training data for {symbol}. Training aborted.")
        return None, None, None

    try:
        accuracy = -1.0
        if len(X) < MIN_TRAINING_EXAMPLES + 2:
             logging.warning(f"Dataset for {symbol} too small ({len(X)}). Training on all data.")
             X_train, y_train = X, y
             X_val, y_val = np.array([]), np.array([])
        else:
             indices = np.random.permutation(len(X))
             val_size = max(1, int(len(X) * 0.2))
             split_idx = len(X) - val_size
             train_indices, val_indices = indices[:split_idx], indices[split_idx:]
             if len(train_indices) == 0 or len(val_indices) == 0:
                  logging.error(f"Train/Val split resulted in zero samples. Training on all data.")
                  X_train, y_train = X, y
                  X_val, y_val = np.array([]), np.array([])
             else:
                  X_train, X_val = X[train_indices], X[val_indices]
                  y_train, y_val = y[train_indices], y[val_indices]
                  if len(np.unique(y_train)) < 2:
                       logging.error(f"Only one class in TRAINING set after split for {symbol}. Aborting.")
                       return None, None, None
                  if len(np.unique(y_val)) < 2:
                       logging.warning(f"Only one class in VALIDATION set after split for {symbol}.")

        if X_val.size == 0: X_val_shaped = np.empty((0, K, len(FEATURES)))
        else: X_val_shaped = X_val

        X_train_t, X_val_t = llt_transform(X_train, y_train, X_val_shaped)

        if X_train_t.size == 0:
            logging.error(f"LLT training transformation failed for {symbol}. Training aborted.")
            return None, None, None
        if X_val.size > 0 and X_val_t.size == 0:
             logging.warning(f"LLT validation transformation failed for {symbol}.")
             accuracy = -1.0
        if np.isnan(X_train_t).any() or np.isinf(X_train_t).any():
             logging.error(f"NaN/Inf in LLT transformed TRAINING data for {symbol}. Training aborted.")
             return None, None, None
        if X_val_t.size > 0 and (np.isnan(X_val_t).any() or np.isinf(X_val_t).any()):
             logging.warning(f"NaN/Inf in LLT transformed VALIDATION data for {symbol}.")
             accuracy = -1.0

        n_neighbors = min(5, len(y_train) - 1) if len(y_train) > 1 else 1
        n_neighbors = max(1, n_neighbors)
        if n_neighbors > 1 and n_neighbors % 2 == 0: n_neighbors -= 1

        model = KNeighborsClassifier(n_neighbors=n_neighbors, weights='distance')
        model.fit(X_train_t, y_train)

        if accuracy != -1.0 and X_val_t.size > 0:
            try:
                accuracy = model.score(X_val_t, y_val)
                logging.info(f"Model for {symbol} trained. Validation Accuracy: {accuracy:.3f}")
            except Exception as eval_e:
                 logging.exception(f"Error during KNN validation scoring for {symbol}: {eval_e}")
                 accuracy = -1.0
        elif accuracy == -1.0:
             logging.info(f"Model for {symbol} trained. Validation skipped or failed.")
        else:
            logging.info(f"Model for {symbol} trained. No validation data.")
            accuracy = -1.0

        return model, X_train, y_train
    except Exception as e:
        logging.exception(f"Error during model training pipeline for {symbol}")
        return None, None, None

def predict_real_time(symbol, model_data):
    # (Keep this function as is)
    if model_data is None: return "Model N/A", 0.0
    model, X_train_orig_for_llt, y_train_orig_for_llt = model_data

    if model is None or X_train_orig_for_llt is None or y_train_orig_for_llt is None:
         logging.error(f"Invalid model data tuple for prediction on {symbol}")
         return "Model Error", 0.0
    if X_train_orig_for_llt.size == 0 or y_train_orig_for_llt.size == 0:
         logging.error(f"Training data for LLT laws is empty for {symbol}")
         return "LLT Data Error", 0.0

    try:
        df = fetch_historical_data(symbol, timeframe='1m', total_candles=K + 60)
        if df is None or len(df) < K:
            return "Data Error", 0.0

        df_recent = df.iloc[-K:]
        if len(df_recent) != K:
            return "Data Error", 0.0

        df_recent_normalized = normalize_data(df_recent)
        if df_recent_normalized is None: return "Norm Error", 0.0
        if df_recent_normalized[FEATURES].isnull().any().any():
             df_recent_normalized = df_recent_normalized.fillna(0.0)

        X_predict_input = np.array([df_recent_normalized[FEATURES].values])
        _, X_predict_transformed = llt_transform(X_train_orig_for_llt, y_train_orig_for_llt, X_predict_input)

        if X_predict_transformed.size == 0 or X_predict_transformed.shape[0] != 1:
            return "Transform Error", 0.0
        if np.isnan(X_predict_transformed).any() or np.isinf(X_predict_transformed).any():
             X_predict_transformed = np.nan_to_num(X_predict_transformed, nan=0.0, posinf=0.0, neginf=0.0)

        try:
            probabilities = model.predict_proba(X_predict_transformed)
            if probabilities.shape[0] != 1 or probabilities.shape[1] != 2:
                 return "Predict Error", 0.0

            prob_class_1 = probabilities[0, 1]
            prediction_label = "Rise" if prob_class_1 >= 0.5 else "Fall"
            confidence = prob_class_1 if prediction_label == "Rise" else probabilities[0, 0]
            return prediction_label, confidence

        except Exception as knn_e:
            logging.exception(f"Error during KNN prediction probability for {symbol}")
            return "Predict Error", 0.0

    except Exception as e:
        logging.exception(f"Error in predict_real_time for {symbol}")
        return "Error", 0.0

# --- TA Calculation Function (Using TA-Lib) ---
def calculate_ta_indicators(df_ta):
    """
    Calculates TA indicators (RSI, MACD, VWAP, ATR) using TA-Lib.
    Requires df_ta to have 'open', 'high', 'low', 'close', 'volume' columns.
    """
    indicators = {'RSI': np.nan, 'MACD': np.nan, 'MACD_Signal': np.nan, 'MACD_Hist': np.nan, 'VWAP': np.nan, 'ATR': np.nan}
    required_cols = ['open', 'high', 'low', 'close', 'volume']
    min_len_needed = max(RSI_PERIOD, MACD_SLOW, ATR_PERIOD) + 1 # TA-Lib often needs P+1

    if df_ta is None or len(df_ta) < min_len_needed:
        logging.warning(f"Insufficient data ({len(df_ta) if df_ta is not None else 0} < {min_len_needed}) for TA-Lib calculations.")
        return indicators

    # Ensure columns exist
    if not all(col in df_ta.columns for col in required_cols):
         logging.error(f"Missing required columns for TA-Lib: Have {df_ta.columns}, Need {required_cols}")
         return indicators

    # --- Prepare data for TA-Lib (NumPy arrays, handle NaNs) ---
    df_ta = df_ta[required_cols].copy() # Work on a copy with only needed columns

    # Check for NaNs BEFORE converting to numpy, TA-Lib generally dislikes them
    if df_ta.isnull().values.any():
        nan_count = df_ta.isnull().sum().sum()
        logging.warning(f"Found {nan_count} NaN(s) in TA input data. Applying ffill()...")
        df_ta.ffill(inplace=True) # Forward fill NaNs
        # Check again after ffill - if NaNs remain (e.g., at the start), need more robust handling
        if df_ta.isnull().values.any():
            logging.error(f"NaNs still present after ffill. Cannot proceed with TA-Lib.")
            return indicators # Return NaNs

    try:
        # Convert to NumPy arrays of type float
        open_p = df_ta['open'].values.astype(float)
        high_p = df_ta['high'].values.astype(float)
        low_p = df_ta['low'].values.astype(float)
        close_p = df_ta['close'].values.astype(float)
        volume_p = df_ta['volume'].values.astype(float)

        # --- Calculate Indicators using TA-Lib ---
        # RSI
        rsi_values = talib.RSI(close_p, timeperiod=RSI_PERIOD)
        indicators['RSI'] = rsi_values[-1] if len(rsi_values) > 0 else np.nan

        # MACD
        macd_line, signal_line, hist = talib.MACD(close_p, fastperiod=MACD_FAST, slowperiod=MACD_SLOW, signalperiod=MACD_SIGNAL)
        indicators['MACD'] = macd_line[-1] if len(macd_line) > 0 else np.nan
        indicators['MACD_Signal'] = signal_line[-1] if len(signal_line) > 0 else np.nan
        indicators['MACD_Hist'] = hist[-1] if len(hist) > 0 else np.nan

        # ATR
        atr_values = talib.ATR(high_p, low_p, close_p, timeperiod=ATR_PERIOD)
        indicators['ATR'] = atr_values[-1] if len(atr_values) > 0 else np.nan

        # VWAP (Manual Calculation - TA-Lib doesn't have it built-in)
        typical_price = (high_p + low_p + close_p) / 3.0
        tp_vol = typical_price * volume_p
        cumulative_volume = np.cumsum(volume_p)
        # Avoid division by zero if volume is zero for initial periods
        if cumulative_volume[-1] > 1e-12: # Check if there's significant volume
            vwap_values = np.cumsum(tp_vol) / np.maximum(cumulative_volume, 1e-12) # Avoid div by zero strictly
            indicators['VWAP'] = vwap_values[-1]
        else:
            indicators['VWAP'] = np.nan # VWAP undefined if no volume

        # Final check for NaNs in results (TA-Lib might return NaN for initial periods)
        for key, value in indicators.items():
             if pd.isna(value):
                 indicators[key] = np.nan # Ensure consistent NaN representation

        # logging.debug(f"TA-Lib Indicators calculated: {indicators}")
        return indicators

    except Exception as ta_e:
        logging.exception(f"Error calculating TA indicators using TA-Lib: {ta_e}")
        return {k: np.nan for k in indicators} # Return NaNs on error

# --- Trade Level Calculation (Unchanged) ---
def calculate_trade_levels(prediction, confidence, current_price, atr):
    # (Keep this function as is - no changes needed)
    levels = {'Entry': np.nan, 'TP1': np.nan, 'TP2': np.nan, 'SL': np.nan}
    if pd.isna(current_price) or current_price <= 0 or pd.isna(atr) or atr <= 0:
        return levels
    if prediction == "Rise" and confidence >= CONFIDENCE_THRESHOLD:
        entry_price = current_price
        levels['Entry'] = entry_price
        levels['TP1'] = entry_price + TP1_ATR_MULTIPLIER * atr
        levels['TP2'] = entry_price + TP2_ATR_MULTIPLIER * atr
        levels['SL'] = entry_price - SL_ATR_MULTIPLIER * atr
        levels['SL'] = max(0.01, levels['SL'])
    # Add Fall logic here if needed
    return levels

# --- Concurrency Wrappers (Unchanged) ---
def train_model_task(coin):
    # (Keep this function as is)
    try:
        result = train_model(coin)
        if result != (None, None, None):
            model, X_train_orig, y_train_orig = result
            return coin, (model, X_train_orig, y_train_orig)
        else:
            return coin, None
    except Exception as e:
        logging.exception(f"Unhandled exception in train_model_task for {coin}")
        return coin, None

def train_all_models(coin_list=None, num_workers=NUM_WORKERS_TRAINING):
    # (Keep this function as is)
    global trained_models
    start_time = time.time()
    if coin_list is None or not coin_list:
        logging.info("No coin list provided, fetching top coins by volume...")
        try:
            coin_list = get_all_usdt_pairs()
            if not coin_list:
                 msg = "Failed to fetch coin list even with fallback. Training aborted."
                 logging.error(msg)
                 return msg
        except Exception as e:
             msg = f"Error fetching coin list: {e}. Training aborted."
             logging.exception(msg)
             return msg

    logging.info(f"Starting training for {len(coin_list)} coins using {num_workers} workers...")
    results_log = []
    successful_trains = 0
    failed_trains = 0
    new_models = {}

    with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers, thread_name_prefix='TrainWorker') as executor:
        future_to_coin = {executor.submit(train_model_task, coin): coin for coin in coin_list}
        processed_count = 0
        total_coins = len(coin_list)
        for future in concurrent.futures.as_completed(future_to_coin):
            processed_count += 1
            coin = future_to_coin[future]
            try:
                returned_coin, model_data = future.result()
                if returned_coin == coin and model_data is not None:
                    new_models[returned_coin] = model_data
                    results_log.append(f"✅ {returned_coin}: Model trained successfully.")
                    successful_trains += 1
                else:
                    results_log.append(f"❌ {coin}: Model training failed (check logs).")
                    failed_trains += 1
            except Exception as e:
                results_log.append(f"❌ {coin}: Training task generated exception: {e}")
                failed_trains += 1
                logging.exception(f"Exception from training future for {coin}")
            if processed_count % 10 == 0 or processed_count == total_coins:
                logging.info(f"Training progress: {processed_count}/{total_coins} coins processed.")
            logging.getLogger().handlers[0].flush()

    trained_models.update(new_models)
    logging.info(f"Updated global models dictionary. Total models now: {len(trained_models)}")

    end_time = time.time()
    duration = end_time - start_time
    completion_message = (
        f"Training run completed in {duration:.2f} seconds.\n"
        f"Successfully trained: {successful_trains}\n"
        f"Failed to train: {failed_trains}\n"
        f"Total models available now: {len(trained_models)}"
    )
    logging.info(completion_message)
    return completion_message + "\n\n" + "\n".join(results_log[-20:])

# --- Update Predictions Table (Mostly Unchanged, uses new TA function) ---
def update_predictions_table():
    # (This function structure remains the same, it just calls the new calculate_ta_indicators)
    global last_update_time
    logging.info("--- Updating Predictions Table ---")
    start_time = time.time()
    predictions_data = {}
    current_models = trained_models.copy()

    if not current_models:
        msg = "No models available. Please train first."
        logging.warning(msg)
        cols = ['Rank', 'Coin', 'Prediction', 'Confidence', 'Price', 'Volume (Quote)', 'Entry', 'Entry Time', 'Exit Time', 'TP1', 'TP2', 'SL', 'RSI', 'MACD Hist', 'VWAP', 'ATR']
        return pd.DataFrame([], columns=cols), msg

    symbols_with_models = list(current_models.keys())
    logging.info(f"Step 1: Generating predictions for {len(symbols_with_models)} models...")
    # --- Stage 1: Get Predictions Concurrently ---
    with concurrent.futures.ThreadPoolExecutor(max_workers=NUM_WORKERS_PREDICTION, thread_name_prefix='PredictWorker') as executor:
        future_to_coin_pred = {executor.submit(predict_real_time, coin, model_data): coin for coin, model_data in current_models.items()}
        pred_success = 0
        pred_fail = 0
        for future in concurrent.futures.as_completed(future_to_coin_pred):
            coin = future_to_coin_pred[future]
            try:
                pred, conf = future.result()
                if pred not in ["Model N/A", "Model Error", "Data Error", "Norm Error", "LLT Data Error", "Transform Error", "Predict Error", "Error"]:
                    predictions_data[coin] = {'prediction': pred, 'confidence': float(conf)}
                    pred_success += 1
                else:
                     predictions_data[coin] = {'prediction': pred, 'confidence': 0.0}
                     pred_fail += 1
            except Exception as e:
                 logging.exception(f"Error getting prediction result for {coin}")
                 predictions_data[coin] = {'prediction': "Future Error", 'confidence': 0.0}
                 pred_fail +=1
    logging.info(f"Step 1 Complete: Predictions generated ({pred_success} success, {pred_fail} fail).")

    # --- Stage 2: Fetch Current Tickers & TA Data Concurrently ---
    symbols_to_fetch_data = list(predictions_data.keys())
    if not symbols_to_fetch_data:
        logging.warning("No symbols with predictions to fetch data for.")
        cols = ['Rank', 'Coin', 'Prediction', 'Confidence', 'Price', 'Volume (Quote)', 'Entry', 'Entry Time', 'Exit Time', 'TP1', 'TP2', 'SL', 'RSI', 'MACD Hist', 'VWAP', 'ATR']
        return pd.DataFrame([], columns=cols), "No symbols processed."

    logging.info(f"Step 2: Fetching Tickers and {TA_DATA_POINTS} OHLCV candles for {len(symbols_to_fetch_data)} symbols...")
    tickers_data = {}
    ohlcv_data = {}
    try: # Fetch Tickers
        batch_size_tickers = 100
        fetched_tickers_batch = {}
        for i in range(0, len(symbols_to_fetch_data), batch_size_tickers):
            batch_symbols = symbols_to_fetch_data[i:i+batch_size_tickers]
            try:
                batch_tickers = exchange.fetch_tickers(symbols=batch_symbols)
                fetched_tickers_batch.update(batch_tickers)
                time.sleep(exchange.rateLimit / 1000 * 0.5)
            except Exception as e:
                 logging.error(f"Failed to fetch ticker batch starting with {batch_symbols[0]}: {e}")
        tickers_data = fetched_tickers_batch
        logging.info(f"Fetched {len(tickers_data)} tickers.")
    except Exception as e:
        logging.exception(f"Error fetching tickers in prediction update: {e}")

    # Fetch OHLCV for TA
    with concurrent.futures.ThreadPoolExecutor(max_workers=NUM_WORKERS_PREDICTION, thread_name_prefix='TADataWorker') as executor:
        future_to_coin_ohlcv = {executor.submit(fetch_historical_data, coin, '1m', TA_DATA_POINTS): coin for coin in symbols_to_fetch_data}
        for future in concurrent.futures.as_completed(future_to_coin_ohlcv):
            coin = future_to_coin_ohlcv[future]
            try:
                df_ta = future.result()
                if df_ta is not None and len(df_ta) == TA_DATA_POINTS:
                    # Ensure standard column names expected by calculate_ta_indicators
                    df_ta.columns = ['timestamp', 'open', 'high', 'low', 'close', 'volume']
                    ohlcv_data[coin] = df_ta
            except Exception as e:
                logging.exception(f"Error fetching TA OHLCV data for {coin}")
    logging.info(f"Step 2 Complete: Fetched TA data for {len(ohlcv_data)} symbols.")

    # --- Stage 3: Calculate TA & Trade Levels ---
    logging.info(f"Step 3: Calculating TA (using TA-Lib) and Trade Levels...")
    final_results = []
    processing_time = datetime.now(timezone.utc)

    for symbol in symbols_to_fetch_data:
        pred_info = predictions_data.get(symbol, {'prediction': 'Missing Pred', 'confidence': 0.0})
        ticker = tickers_data.get(symbol)
        df_ta = ohlcv_data.get(symbol) # This df should have standard columns now

        current_price, quote_volume = np.nan, np.nan
        ta_indicators = {k: np.nan for k in ['RSI', 'MACD', 'MACD_Signal', 'MACD_Hist', 'VWAP', 'ATR']}
        trade_levels = {k: np.nan for k in ['Entry', 'TP1', 'TP2', 'SL']}
        entry_time, exit_time = pd.NaT, pd.NaT

        if ticker and isinstance(ticker, dict):
            current_price = ticker.get('last', np.nan)
            quote_volume = ticker.get('info', {}).get('quoteVolume')
            if quote_volume is None:
                 base_volume = ticker.get('baseVolume')
                 if base_volume is not None and current_price is not None:
                      try: quote_volume = float(base_volume) * float(current_price)
                      except (ValueError, TypeError): quote_volume = np.nan
            try: current_price = float(current_price) if current_price is not None else np.nan
            except (ValueError, TypeError): current_price = np.nan
            try: quote_volume = float(quote_volume) if quote_volume is not None else np.nan
            except (ValueError, TypeError): quote_volume = np.nan

        # Calculate TA using the new function
        if df_ta is not None:
            ta_indicators = calculate_ta_indicators(df_ta) # Calls the TA-Lib version

        if pred_info['prediction'] in ["Rise", "Fall"] and not pd.isna(current_price) and not pd.isna(ta_indicators['ATR']):
            trade_levels = calculate_trade_levels(pred_info['prediction'], pred_info['confidence'], current_price, ta_indicators['ATR'])
            if not pd.isna(trade_levels['Entry']):
                entry_time = processing_time
                exit_time = processing_time + timedelta(hours=PREDICTION_WINDOW_HOURS)

        final_results.append({
            'coin': symbol.split('/')[0], 'full_symbol': symbol,
            'prediction': pred_info['prediction'], 'confidence': pred_info['confidence'],
            'price': current_price, 'volume': quote_volume,
            'entry': trade_levels['Entry'], 'entry_time': entry_time, 'exit_time': exit_time,
            'tp1': trade_levels['TP1'], 'tp2': trade_levels['TP2'], 'sl': trade_levels['SL'],
            'rsi': ta_indicators['RSI'], 'macd_hist': ta_indicators['MACD_Hist'],
            'vwap': ta_indicators['VWAP'], 'atr': ta_indicators['ATR']
        })
    logging.info("Step 3 Complete: TA and Trade Levels calculated.")

    # --- Stage 4: Sort and Format (Unchanged) ---
    def sort_key(item):
        pred, conf = item['prediction'], item['confidence']
        if pred == "Rise" and conf >= CONFIDENCE_THRESHOLD and not pd.isna(item['entry']): return (0, -conf)
        elif pred == "Rise": return (1, -conf)
        elif pred == "Fall": return (2, -conf)
        else: return (3, 0)
    final_results.sort(key=sort_key)

    formatted_output = []
    for i, p in enumerate(final_results[:MAX_COINS_TO_DISPLAY]):
        formatted_output.append([
            i + 1, p['coin'], p['prediction'], f"{p['confidence']:.3f}",
            f"{p['price']:.4f}" if not pd.isna(p['price']) else "N/A",
            f"{p['volume']:,.0f}" if not pd.isna(p['volume']) else "N/A",
            f"{p['entry']:.4f}" if not pd.isna(p['entry']) else "N/A",
            format_datetime(p['entry_time'], "N/A"), format_datetime(p['exit_time'], "N/A"),
            f"{p['tp1']:.4f}" if not pd.isna(p['tp1']) else "N/A",
            f"{p['tp2']:.4f}" if not pd.isna(p['tp2']) else "N/A",
            f"{p['sl']:.4f}" if not pd.isna(p['sl']) else "N/A",
            f"{p['rsi']:.2f}" if not pd.isna(p['rsi']) else "N/A",
            f"{p['macd_hist']:.4f}" if not pd.isna(p['macd_hist']) else "N/A",
            f"{p['vwap']:.4f}" if not pd.isna(p['vwap']) else "N/A",
            f"{p['atr']:.4f}" if not pd.isna(p['atr']) else "N/A",
        ])

    output_columns = [
        'Rank', 'Coin', 'Prediction', 'Confidence', 'Price', 'Volume (Quote)',
        'Entry', 'Entry Time', 'Exit Time', 'TP1', 'TP2', 'SL',
        'RSI', 'MACD Hist', 'VWAP', 'ATR'
    ]
    output_df = pd.DataFrame(formatted_output, columns=output_columns)

    end_time = time.time()
    duration = end_time - start_time
    last_update_time = processing_time
    status_message = f"Predictions updated ({len(final_results)} symbols processed) in {duration:.2f}s. Last update: {format_datetime(last_update_time)}"
    logging.info(status_message)

    return output_df, status_message


# --- Gradio UI Handlers (Unchanged) ---
def handle_train_click(coin_input, num_workers):
    # (Keep this function as is)
    logging.info(f"Train button clicked. Workers: {num_workers}")
    coins = None
    num_workers = int(num_workers)
    if coin_input and coin_input.strip():
        raw_coins = coin_input.replace(',', ' ').split()
        coins = []
        valid = True
        for c in raw_coins:
            coin_upper = c.strip().upper()
            if '/' not in coin_upper: coin_upper += '/USDT'
            if coin_upper.endswith('/USDT'): coins.append(coin_upper)
            else:
                 valid = False
                 logging.error(f"Invalid coin format: {c}. Must be SYMBOL or SYMBOL/USDT.")
                 break
        if not valid: return "Error: Custom coins must be valid SYMBOL or SYMBOL/USDT pairs."
        logging.info(f"Training requested for custom coin list: {coins}")
    else:
        logging.info("Training requested for top coins by volume.")
    train_status = train_all_models(coin_list=coins, num_workers=num_workers)
    return f"--- Training Run ---:\n{train_status}\n\n---> Press 'Refresh Predictions' <---"

def handle_refresh_click():
    # (Keep this function as is)
    logging.info("Refresh button clicked.")
    try:
        df, status = update_predictions_table()
        return df, status
    except Exception as e:
        logging.exception("Error during handle_refresh_click")
        cols = ['Rank', 'Coin', 'Prediction', 'Confidence', 'Price', 'Volume (Quote)', 'Entry', 'Entry Time', 'Exit Time', 'TP1', 'TP2', 'SL', 'RSI', 'MACD Hist', 'VWAP', 'ATR']
        return pd.DataFrame([], columns=cols), f"Error updating predictions: {e}"

# --- Gradio Interface Definition (Unchanged) ---
with gr.Blocks(theme=gr.themes.Soft()) as demo:
    gr.Markdown("# Cryptocurrency Prediction & TA Signal Explorer (LLT-KNN + TA-Lib)") # Updated title slightly
    gr.Markdown(f"""
        Predicts **{PREDICTION_WINDOW_HOURS}-hour** price direction (Rise/Fall) using LLT-KNN.
        Displays current price, volume, TA indicators (RSI, MACD, VWAP, ATR calculated using **TA-Lib**), and potential trade levels for **Rise** signals meeting confidence >= **{CONFIDENCE_THRESHOLD}**.
        TP/SL levels based on **{TP1_ATR_MULTIPLIER}x / {TP2_ATR_MULTIPLIER}x / {SL_ATR_MULTIPLIER}x ATR({ATR_PERIOD})**.
        **Warning:** Educational. High risk. Not financial advice. Ensure TA-Lib is correctly installed.
        """)

    with gr.Row():
        with gr.Column(scale=4):
            prediction_df = gr.Dataframe(
                headers=[
                    'Rank', 'Coin', 'Prediction', 'Confidence', 'Price', 'Volume (Quote)',
                    'Entry', 'Entry Time', 'Exit Time', 'TP1', 'TP2', 'SL',
                    'RSI', 'MACD Hist', 'VWAP', 'ATR'
                ],
                datatype=[
                    'number', 'str', 'str', 'str', 'str', 'str',
                    'str', 'str', 'str', 'str', 'str', 'str',
                    'str', 'str', 'str', 'str'
                    ],
                row_count=15, col_count=(16, "fixed"), label="Predictions & TA Signals", wrap=True,
            )
        with gr.Column(scale=1):
            with gr.Accordion("Train Models", open=True):
                 coin_input = gr.Textbox(label="Train Specific Coins (e.g., BTC, ETH/USDT)", placeholder="Leave empty for top coins by volume")
                 max_workers_slider = gr.Slider(minimum=1, maximum=10, value=NUM_WORKERS_TRAINING, step=1, label="Parallel Training Workers")
                 train_button = gr.Button("Start Training", variant="primary")
            refresh_button = gr.Button("Refresh Predictions", variant="secondary")
            status_text = gr.Textbox(label="Status Log", lines=15, interactive=False, max_lines=30)

    gr.Markdown(
        """
        ## Notes
        - **TA-Lib**: This version uses the TA-Lib library for indicators. Ensure it's installed correctly (can be tricky).
        - **Data**: Fetches OHLCV data (Bitget, 1-min). Uses cache. Handles rate limits.
        - **Training**: Uses past ~14h data (12h train, 2h predict). Normalizes, balances classes, applies LLT, trains KNN.
        - **Prediction**: Uses latest 12h data for KNN input.
        - **Trade Levels**: Only shown for 'Rise' predictions above confidence threshold. Based on current price and ATR volatility. **Highly speculative.**
        - **Sorting**: Table sorted by (Potential Rise Signals > Other Rise > Fall > Errors), then by confidence descending.
        - **Refresh**: Fetches latest prices/TA and re-evaluates signals.
        """
    )
    train_button.click(fn=handle_train_click, inputs=[coin_input, max_workers_slider], outputs=status_text)
    refresh_button.click(fn=handle_refresh_click, inputs=None, outputs=[prediction_df, status_text])

# --- Startup Initialization (Unchanged) ---
def initialize_models_on_startup():
    # (Keep this function as is)
    logging.info("----- Initializing Models (Startup Thread) -----")
    default_coins = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'XRP/USDT', 'DOGE/USDT']
    try:
        initial_status = train_all_models(default_coins, num_workers=2)
        logging.info("----- Initial Model Training Complete -----")
        logging.info(initial_status)
    except Exception as e:
         logging.exception("Error during startup initialization.")

# --- Main Execution (Unchanged) ---
if __name__ == "__main__":
    logging.info("Starting application...")
    # Check if TA-Lib import worked (basic check)
    try:
        # Try accessing a TA-Lib function
        _ = talib.RSI(np.array([1.0, 2.0, 3.0, 4.0, 5.0, 6.0]))
        logging.info("TA-Lib library seems accessible.")
    except NameError:
        logging.error("FATAL: TA-Lib library not found or import failed. Please install it correctly.")
        sys.exit(1)
    except Exception as ta_init_e:
        logging.error(f"FATAL: Error testing TA-Lib library: {ta_init_e}. Please check installation.")
        sys.exit(1)

    init_thread = threading.Thread(target=initialize_models_on_startup, name="StartupTrainThread", daemon=True)
    init_thread.start()

    logging.info("Launching Gradio Interface...")
    try:
        demo.launch(server_name="0.0.0.0")
    except Exception as e:
        logging.exception("Failed to launch Gradio interface.")
    finally:
        logging.info("Gradio Interface stopped.")