import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler from sklearn.linear_model import LinearRegression from sklearn.metrics import classification_report, confusion_matrix from sklearn.utils.class_weight import compute_class_weight import tensorflow as tf from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout, BatchNormalization from tensorflow.keras.callbacks import EarlyStopping from tensorflow.keras.optimizers import Adam import matplotlib.pyplot as plt import warnings import mysql.connector # добавлен импорт для MySQL import joblib import json import tensorflow as tf warnings.filterwarnings('ignore') # ====================== # 1. Загрузка данных из MySQL # ====================== def load_data_from_mysql(host='nlbotinterface.ru', port=3306, database='bitcoin_tickers', user='bitcoin', password='g49020007', table='tickers', ticker=None): """ Загружает данные из таблицы MySQL. Параметры подключения заданы по умолчанию. Если указан ticker, добавляется условие WHERE. Возвращает DataFrame с колонками open, high, low, close, volume """ conn = mysql.connector.connect( host=host, port=port, database=database, user=user, password=password, ssl_disabled=True ) # Формируем запрос: выбираем нужные поля. # Предполагаем, что Date и Time хранятся в строковом формате '%Y.%m.%d' и '%H:%M'. # Если это не так, можно преобразовать в SQL, например: # DATE_FORMAT(Date, '%Y.%m.%d') as Date, DATE_FORMAT(Time, '%H:%i') as Time query = f"SELECT Date, Time, Open, High, Low, Close, Volume FROM {table}" if ticker: query += f" WHERE ticker = '{ticker}'" query += " ORDER BY Date, Time" df = pd.read_sql_query(query, conn) conn.close() # Переименовываем столбцы в соответствии с ожидаемыми именами df.columns = ['date', 'time', 'open', 'high', 'low', 'close', 'volume'] # Преобразуем дату и время в datetime # Если Date и Time уже строки в нужном формате, объединяем и парсим df['datetime'] = pd.to_datetime(df['date'] + ' ' + df['time'], format='%Y.%m.%d %H:%M', errors='coerce') df.set_index('datetime', inplace=True) df.drop(['date', 'time'], axis=1, inplace=True) # Приводим числовые колонки к float for col in ['open', 'high', 'low', 'close', 'volume']: df[col] = pd.to_numeric(df[col], errors='coerce') df.dropna(inplace=True) df.sort_index(inplace=True) print(f"Загружено записей: {len(df)}") return df # ====================== # 2. Добавление признаков # ====================== def add_features(df): # Логарифмическая доходность df['log_ret'] = np.log(df['close'] / df['close'].shift(1)) # Относительные изменения цены за разные периоды df['ret_5'] = df['close'].pct_change(5) df['ret_10'] = df['close'].pct_change(10) df['ret_20'] = df['close'].pct_change(20) # RSI delta = df['close'].diff() gain = delta.where(delta > 0, 0).rolling(14).mean() loss = (-delta.where(delta < 0, 0)).rolling(14).mean() rs = gain / loss df['rsi'] = 100 - 100 / (1 + rs) # MACD ema12 = df['close'].ewm(span=12).mean() ema26 = df['close'].ewm(span=26).mean() df['macd'] = ema12 - ema26 df['macd_signal'] = df['macd'].ewm(span=9).mean() df['macd_hist'] = df['macd'] - df['macd_signal'] df['macd_hist_diff'] = df['macd_hist'].diff() # ATR, нормированный на цену закрытия high_low = df['high'] - df['low'] high_close = (df['high'] - df['close'].shift()).abs() low_close = (df['low'] - df['close'].shift()).abs() tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1) df['atr'] = tr.rolling(14).mean() df['atr_ratio'] = df['atr'] / df['close'] # Bollinger Bands ma20 = df['close'].rolling(20).mean() std20 = df['close'].rolling(20).std() df['bb_width'] = (2 * std20) / ma20 df['bb_percent'] = (df['close'] - (ma20 - 2*std20)) / (4*std20) # Волатильность df['volatility'] = df['log_ret'].rolling(20).std() # Расстояния до экстремумов df['dist_from_high'] = (df['high'].rolling(20).max() - df['close']) / df['close'] df['dist_from_low'] = (df['close'] - df['low'].rolling(20).min()) / df['close'] # Отношения цены к EMA ema10 = df['close'].ewm(span=10, adjust=False).mean() ema30 = df['close'].ewm(span=30, adjust=False).mean() df['close_ema10'] = df['close'] / ema10 - 1 df['close_ema30'] = df['close'] / ema30 - 1 df['ema10_ema30'] = ema10 / ema30 - 1 df['ema10_slope'] = (ema10 / ema10.shift(1) - 1).fillna(0) df['ema30_slope'] = (ema30 / ema30.shift(1) - 1).fillna(0) # Объёмные индикаторы df['volume_ma20'] = df['volume'].rolling(20).mean() df['volume_ratio'] = df['volume'] / df['volume_ma20'] df['volume_change'] = df['volume'].pct_change() df['volume_vola'] = df['volume_ratio'] / df['volatility'] df['adx'] = compute_adx(df, period=14) df['chop'] = add_choppiness(df) ema10 = df['close'].ewm(span=10).mean() ema30 = df['close'].ewm(span=30).mean() cross = ((ema10 > ema30) & (ema10.shift(1) <= ema30.shift(1))) | ((ema10 < ema30) & (ema10.shift(1) >= ema30.shift(1))) df['ema_cross_count'] = cross.rolling(20).sum() # ADF p-value (требуется statsmodels) from statsmodels.tsa.stattools import adfuller def adf_pvalue(series): try: return adfuller(series, autolag='AIC')[1] except: return np.nan df['adf_pvalue'] = df['close'].rolling(60).apply(adf_pvalue, raw=False) df.dropna(inplace=True) return df # ====================== # Вычисление ADX (упрощённо) # ====================== def compute_adx(df, period=14): high = df['high'] low = df['low'] close = df['close'] plus_dm = high.diff() minus_dm = low.diff() plus_dm[plus_dm < 0] = 0 minus_dm[minus_dm > 0] = 0 minus_dm = abs(minus_dm) tr = pd.concat([high - low, (high - close.shift()).abs(), (low - close.shift()).abs()], axis=1).max(axis=1) atr = tr.rolling(period).mean() plus_di = 100 * (plus_dm.ewm(alpha=1/period).mean() / atr) minus_di = 100 * (minus_dm.ewm(alpha=1/period).mean() / atr) dx = (abs(plus_di - minus_di) / (plus_di + minus_di)) * 100 adx = dx.rolling(period).mean() return adx # ====================== # Вычисление индикатора CI # ====================== def add_choppiness(df, period=14): high = df['high'] low = df['low'] close = df['close'] tr = pd.concat([high - low, (high - close.shift()).abs(), (low - close.shift()).abs()], axis=1).max(axis=1) atr = tr.rolling(period).mean() highest_high = high.rolling(period).max() lowest_low = low.rolling(period).min() ci = 100 * np.log10((atr.rolling(period).sum()) / (highest_high - lowest_low)) / np.log10(period) return ci # ====================== # 3. Разметка тренда (адаптивный порог) # ====================== def label_trend_adaptive(df, window=30, atr_multiplier=0.5, use_atr_percentile=True): """ Определяет классы: 0 - флэт (|slope| < threshold) 1 - бычий (slope > threshold) 2 - медвежий (slope < -threshold) """ prices = df['close'].values slopes = np.full(len(df), np.nan) for i in range(window, len(df)): y = prices[i-window:i] x = np.arange(window).reshape(-1, 1) model = LinearRegression().fit(x, y) slopes[i] = model.coef_[0] df['slope'] = slopes if use_atr_percentile: # Динамический порог: скользящий процентиль от абсолютных наклонов abs_slope = np.abs(slopes) rolling_percentile = pd.Series(abs_slope).rolling(window=window*10, min_periods=window).quantile(0.15) # 20-й процентиль threshold = rolling_percentile.fillna(rolling_percentile.median()).values df['threshold'] = threshold print(f"Порог флэта: скользящий 20-й процентиль от |slope|") else: # Альтернативный вариант: порог как доля от среднего ATR за окно # (требуется колонка atr_ratio) atr_rolling = df['atr_ratio'].rolling(window=window*10, min_periods=window).mean() threshold = atr_multiplier * atr_rolling df['threshold'] = threshold.fillna(threshold.median()) print(f"Порог флэта: {atr_multiplier:.2f} * средний ATR за окно") # Присвоение классов df['trend'] = 0 df.loc[df['slope'] > threshold, 'trend'] = 1 df.loc[df['slope'] < -threshold, 'trend'] = 2 # Удалён блок, который присваивал класс 3 df.dropna(subset=['slope', 'threshold'], inplace=True) return df # ====================== # 4. Создание последовательностей # ====================== def create_sequences(data, targets, lookback): X, y = [], [] for i in range(len(data) - lookback): X.append(data[i:i+lookback]) y.append(targets[i+lookback]) return np.array(X), np.array(y) # ====================== # 5. Аугментация (oversampling) для нескольких классов # ====================== def oversample_classes(X_train, y_train, target_ratios, noise_scale=0.01): X_new, y_new = X_train.copy(), y_train.copy() for class_label, target_ratio in target_ratios.items(): class_idx = np.where(y_new == class_label)[0] current_count = len(class_idx) total_count = len(y_new) current_ratio = current_count / total_count if current_ratio >= target_ratio: print(f"Класс {class_label}: {current_ratio:.2%} >= {target_ratio:.2%}, oversampling не требуется") continue target_count = int(total_count * target_ratio) n_to_add = target_count - current_count print(f"Класс {class_label}: {current_count} -> {target_count} (доля {target_ratio:.2%})") X_add = [] for idx in np.random.choice(class_idx, n_to_add, replace=True): seq = X_new[idx].copy() noise = np.random.normal(0, noise_scale, seq.shape) seq += noise X_add.append(seq) X_new = np.vstack([X_new, np.array(X_add)]) y_new = np.hstack([y_new, [class_label] * n_to_add]) shuffle_idx = np.random.permutation(len(X_new)) X_new = X_new[shuffle_idx] y_new = y_new[shuffle_idx] return X_new, y_new # ====================== # 6. Модель LSTM # ====================== def weighted_sum(x): """Суммирование по временной оси для механизма внимания.""" return tf.reduce_sum(x, axis=1) def build_model(input_shape, num_classes): inputs = tf.keras.Input(shape=input_shape) # Свёртка для локальных признаков x = tf.keras.layers.Conv1D(filters=64, kernel_size=5, padding='same', activation='relu')(inputs) x = tf.keras.layers.BatchNormalization()(x) x = tf.keras.layers.Dropout(0.2)(x) # Двунаправленная LSTM x = tf.keras.layers.Bidirectional(tf.keras.layers.LSTM(128, return_sequences=True))(x) x = tf.keras.layers.BatchNormalization()(x) x = tf.keras.layers.Dropout(0.3)(x) # Механизм внимания attention = tf.keras.layers.Dense(1, activation='tanh')(x) attention = tf.keras.layers.Flatten()(attention) attention = tf.keras.layers.Activation('softmax')(attention) attention = tf.keras.layers.RepeatVector(256)(attention) # 256 = 128*2 attention = tf.keras.layers.Permute((2, 1))(attention) x = tf.keras.layers.Multiply()([x, attention]) x = tf.keras.layers.Lambda(weighted_sum, output_shape=lambda input_shape: (input_shape[0], input_shape[2]))(x) x = tf.keras.layers.Dense(64, activation='relu')(x) x = tf.keras.layers.Dropout(0.3)(x) outputs = tf.keras.layers.Dense(num_classes, activation='softmax')(x) model = tf.keras.Model(inputs, outputs) model.compile(optimizer=Adam(0.00005), loss='sparse_categorical_crossentropy', metrics=['accuracy']) return model #def build_model(input_shape, num_classes): # inputs = tf.keras.Input(shape=input_shape) # x = LSTM(128, return_sequences=True)(inputs) # x = BatchNormalization()(x) # x = Dropout(0.3)(x) # Механизм внимания # attention = Dense(1, activation='tanh')(x) # (batch, time, 1) # attention = tf.keras.layers.Flatten()(attention) # (batch, time) # attention = tf.keras.layers.Activation('softmax')(attention) # attention = tf.keras.layers.RepeatVector(128)(attention) # attention = tf.keras.layers.Permute((2, 1))(attention) # x = tf.keras.layers.Multiply()([x, attention]) # Используем именованную функцию вместо лямбды # x = tf.keras.layers.Lambda(weighted_sum, output_shape=(128,))(x) # x = Dense(64, activation='relu')(x) # x = Dropout(0.3)(x) # outputs = Dense(num_classes, activation='softmax')(x) # model = tf.keras.Model(inputs, outputs) # model.compile(optimizer=Adam(0.0005), # loss='sparse_categorical_crossentropy', # metrics=['accuracy']) # return model # ====================== # 7. Основной блок # ====================== def main(): print("Загрузка данных из MySQL...") # Если нужно выбрать конкретный тикер, передайте ticker='BTCUSD' df = load_data_from_mysql(ticker=None) # или укажите ticker, например 'BTCUSD' print("Добавление признаков...") df = add_features(df) print("Разметка тренда...") df = label_trend_adaptive(df, window=14, atr_multiplier=0.5) print("Распределение классов (исходные метки):") print(df['trend'].value_counts().sort_index()) # Отбор признаков feature_cols = [ 'ema10_slope', 'ema30_slope', 'log_ret', 'rsi', 'macd', 'macd_signal', 'macd_hist', 'atr', 'volatility', 'bb_width', 'bb_percent', 'dist_from_high', 'dist_from_low', 'close_ema10', 'close_ema30', 'ema10_ema30', 'volume_ratio', 'chop', 'ema_cross_count', 'adf_pvalue' ] feature_cols = [c for c in feature_cols if c in df.columns] with open('feature_cols.json', 'w') as f: json.dump(feature_cols, f) print(f"Список признаков сохранён в feature_cols.json: {feature_cols}") data_raw = df[feature_cols].values targets_raw = df['trend'].values.astype(int) # Перенумерация классов в 0..n-1 unique_classes = np.unique(targets_raw) class_mapping = {old: new for new, old in enumerate(unique_classes)} targets = np.array([class_mapping[x] for x in targets_raw]) print("Отображение исходных классов в новые:", class_mapping) print("Новое распределение классов:", pd.Series(targets).value_counts().sort_index().to_dict()) # Нормализация scaler = StandardScaler() data_scaled = scaler.fit_transform(data_raw) # Создание последовательностей lookback = 60 X, y = create_sequences(data_scaled, targets, lookback) print(f"Форма X: {X.shape}, y: {y.shape}") # Гарантируем наличие класса 3 в тестовой выборке split = int(len(X) * 0.8) X_train, X_test = X[:split], X[split:] y_train, y_test = y[:split], y[split:] # Oversampling для редких классов 0 и 3 target_ratios = {} if 0 in class_mapping: target_ratios[class_mapping[0]] = 0.2 # 10% класса 0 if target_ratios: X_train, y_train = oversample_classes( X_train, y_train, target_ratios=target_ratios, noise_scale=0.02 ) print("После oversampling распределение классов в train:") print(pd.Series(y_train).value_counts().sort_index().to_dict()) # Веса классов classes_in_train = np.unique(y_train) class_weights = compute_class_weight('balanced', classes=classes_in_train, y=y_train) class_weight_dict = dict(zip(classes_in_train, class_weights)) if 0 in class_mapping: class_weight_dict[class_mapping[0]] *= 2.0 print("Веса классов (новые метки):", class_weight_dict) # Модель num_classes = len(unique_classes) model = build_model((X.shape[1], X.shape[2]), num_classes) model.summary() early_stop = EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True) history = model.fit(X_train, y_train, validation_data=(X_test, y_test), epochs=300, batch_size=32, class_weight=class_weight_dict, callbacks=[early_stop], verbose=1) # Оценка y_pred_prob = model.predict(X_test) y_pred_new = np.argmax(y_pred_prob, axis=1) inv_mapping = {v: k for k, v in class_mapping.items()} y_test_orig = np.array([inv_mapping[x] for x in y_test]) y_pred_orig = np.array([inv_mapping[x] for x in y_pred_new]) unique_orig_in_test = np.unique(y_test_orig) target_names = [f'Класс {u}' for u in unique_orig_in_test] print(classification_report(y_test_orig, y_pred_orig, labels=unique_orig_in_test, target_names=target_names)) print("Confusion Matrix (исходные метки):") print(confusion_matrix(y_test_orig, y_pred_orig, labels=unique_orig_in_test)) # Графики plt.figure(figsize=(12, 4)) plt.subplot(1, 2, 1) plt.plot(history.history['loss'], label='Train') plt.plot(history.history['val_loss'], label='Val') plt.legend() plt.title('Loss') plt.subplot(1, 2, 2) plt.plot(history.history['accuracy'], label='Train') plt.plot(history.history['val_accuracy'], label='Val') plt.legend() plt.title('Accuracy') plt.show() model.save('bitcoin_trend_improved.keras') print("Модель сохранена как 'bitcoin_trend_improved.keras'") # Сохраняем стандартизатор joblib.dump(scaler, 'scaler.pkl') print("Стандартизатор сохранён как 'scaler.pkl'") # Сохраняем маппинг классов (исходные метки -> индексы модели) joblib.dump(class_mapping, 'class_mapping.pkl') print("Маппинг классов сохранён как 'class_mapping.pkl'") if __name__ == "__main__": main()