import pandas as pd import numpy as np from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, DateTime, TIMESTAMP, text, inspect from sqlalchemy.dialects.mysql import TINYINT, VARCHAR, DATETIME, TIMESTAMP as MySQL_TIMESTAMP from sqlalchemy.exc import SQLAlchemyError import warnings import argparse from datetime import datetime import re import json import sys warnings.filterwarnings('ignore') # Функция для вывода диагностики в stderr (чтобы не мешать JSON) def eprint(*args, **kwargs): print(*args, file=sys.stderr, **kwargs) # Конфигурация подключения DB_CONFIG = { 'host': 'nlbotinterface.ru', 'port': 3306, 'database': 'bitcoin_tickers', 'user': 'bitcoin', 'password': 'g49020007', } # ИЗМЕНЕНО: вместо M5 теперь W1 TIMEFRAME_SUFFIXES = ['W1', 'H1', 'D1'] LIMIT = 100 # количество последних записей для загрузки EMA_FAST = 12 EMA_SLOW = 26 TREND_THRESHOLD = 0.001 # порог для флэта # Создаем engine SQLAlchemy DATABASE_URL = f"mysql+mysqlconnector://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}" engine = create_engine(DATABASE_URL, echo=False) # ------------------------------------------------------------ # Вспомогательная функция очистки имени инструмента # ------------------------------------------------------------ def sanitize_table_name(name): """Очищает имя таблицы от недопустимых символов.""" return re.sub(r'[^a-zA-Z0-9_]', '', name) # ------------------------------------------------------------ # Создание/проверка структуры таблицы trend_analysis # ------------------------------------------------------------ def ensure_trend_table_exists(): """ Создаёт таблицу trend_analysis, если её нет, или приводит существующую к нужной структуре. Добавлена колонка instrument. """ metadata = MetaData() inspector = inspect(engine) # Определяем желаемую структуру таблицы (с колонкой instrument) target_columns = [ Column('id', Integer, primary_key=True, autoincrement=True), Column('table_name', VARCHAR(50), nullable=False), Column('instrument', VARCHAR(50), nullable=False), # новая колонка Column('datetime', DateTime, nullable=False), Column('trend', TINYINT, nullable=False, comment='0 - флэт, 1 - бычий, 2 - медвежий'), Column('created_at', TIMESTAMP, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP')) ] target_table = Table('trend_analysis', metadata, *target_columns, mysql_engine='InnoDB', mysql_default_charset='utf8mb4') # Проверяем существование таблицы if not inspector.has_table('trend_analysis'): # Создаём с нуля metadata.create_all(engine) eprint("Таблица trend_analysis создана (с колонкой instrument).") return # Таблица существует — проверяем её структуру existing_columns = {col['name']: col for col in inspector.get_columns('trend_analysis')} expected_columns = {col.name: col for col in target_columns} # Проверяем наличие всех ожидаемых колонок missing_columns = set(expected_columns.keys()) - set(existing_columns.keys()) extra_columns = set(existing_columns.keys()) - set(expected_columns.keys()) # Если есть лишние колонки или не хватает обязательных, пересоздаём таблицу if missing_columns or extra_columns: eprint("Обнаружено несоответствие структуры таблицы. Пересоздаём trend_analysis...") with engine.connect() as conn: conn.execute(text("DROP TABLE IF EXISTS trend_analysis")) conn.commit() metadata.create_all(engine) eprint("Таблица trend_analysis пересоздана (с колонкой instrument).") return # Проверяем наличие уникального ключа на table_name (оставляем для совместимости) indexes = inspector.get_indexes('trend_analysis') has_unique_table_name = any( idx['column_names'] == ['table_name'] and idx['unique'] for idx in indexes ) if not has_unique_table_name: with engine.connect() as conn: conn.execute(text(""" DELETE t1 FROM trend_analysis t1 INNER JOIN trend_analysis t2 WHERE t1.id > t2.id AND t1.table_name = t2.table_name """)) conn.execute(text("DELETE FROM trend_analysis WHERE table_name IS NULL")) conn.execute(text("ALTER TABLE trend_analysis ADD UNIQUE INDEX (table_name)")) conn.commit() eprint("Добавлен уникальный ключ на table_name.") # ------------------------------------------------------------ # Загрузка последних N записей из указанной таблицы # ------------------------------------------------------------ def load_last_rows(table_name, limit=LIMIT): """ Загружает последние limit записей из таблицы, сортирует по возрастанию времени. Использует колонку timestamp (BIGINT Unix time) для сортировки. Возвращает DataFrame с колонками: datetime, Open, High, Low, Close, Volume. """ query = f""" SELECT timestamp, Open, High, Low, Close, Volume FROM {table_name} ORDER BY timestamp DESC LIMIT {limit} """ df = pd.read_sql(query, engine) if df.empty: return df # Преобразование timestamp в datetime df['datetime'] = pd.to_datetime(df['timestamp'], unit='s') df.drop('timestamp', axis=1, inplace=True) df.sort_values('datetime', inplace=True) df.reset_index(drop=True, inplace=True) # Приведение к числовым типам for col in ['Open', 'High', 'Low', 'Close', 'Volume']: df[col] = pd.to_numeric(df[col], errors='coerce') df.dropna(inplace=True) return df # ------------------------------------------------------------ # Определение тренда на последней свече с помощью EMA # ------------------------------------------------------------ def detect_trend_ema(df, pairs=None, atr_period=14, atr_threshold=0.3, smooth_period=5, threshold_fallback=0.001): """ Адаптивная версия: использует только те пары EMA, для которых достаточно данных. """ if pairs is None: pairs = [(5, 13, 1), (12, 26, 1), (50, 200, 1)] n = len(df) if n < 2: return None, None # Фильтруем пары: нужна длина >= slow valid_pairs = [(f, s, w) for f, s, w in pairs if n >= s] if not valid_pairs: # Нет ни одной подходящей пары return None, None # Проверяем возможность расчёта ATR use_atr = n >= atr_period + 1 if not use_atr: eprint(f" Предупреждение: данных для ATR недостаточно (нужно {atr_period+1}, имеется {n}), используется запасной порог.") df = df.copy() closes = df['Close'].values highs = df['High'].values lows = df['Low'].values # Расчёт ATR, если возможно if use_atr: tr = np.zeros(n) for i in range(n): if i == 0: tr[i] = highs[i] - lows[i] else: tr[i] = max(highs[i] - lows[i], abs(highs[i] - closes[i-1]), abs(lows[i] - closes[i-1])) atr = pd.Series(tr).ewm(span=atr_period, adjust=False).mean().values else: atr = None # Предварительный расчёт всех EMA для отфильтрованных пар ema_dict = {} for fast, slow, w in valid_pairs: ema_dict[(fast, slow)] = { 'fast': pd.Series(closes).ewm(span=fast, adjust=False).mean().values, 'slow': pd.Series(closes).ewm(span=slow, adjust=False).mean().values } sum_votes_history = np.zeros(n) total_weight = sum(w for _, _, w in valid_pairs) for k in range(n): sum_votes = 0.0 for fast, slow, weight in valid_pairs: ema_fast = ema_dict[(fast, slow)]['fast'][k] ema_slow = ema_dict[(fast, slow)]['slow'][k] diff = ema_fast - ema_slow if use_atr and atr[k] > 1e-10: norm_diff = diff / atr[k] if norm_diff > atr_threshold: vote = 1 elif norm_diff < -atr_threshold: vote = -1 else: vote = 0 else: if closes[k] != 0: norm_diff = diff / closes[k] else: norm_diff = 0.0 if norm_diff > threshold_fallback: vote = 1 elif norm_diff < -threshold_fallback: vote = -1 else: vote = 0 sum_votes += vote * weight sum_votes_history[k] = sum_votes # Сглаживание голосов if smooth_period and smooth_period > 1 and n >= smooth_period: smoothed = pd.Series(sum_votes_history).ewm(span=smooth_period, adjust=False).mean().values else: smoothed = sum_votes_history last_smoothed = smoothed[-1] decision_threshold = 0.3 * total_weight if last_smoothed > decision_threshold: trend = 1 elif last_smoothed < -decision_threshold: trend = 2 else: trend = 0 last_dt = df.iloc[-1]['datetime'] return trend, last_dt # ------------------------------------------------------------ # Получение времени последнего анализа для таблицы # ------------------------------------------------------------ def get_last_analysis_datetime(table_name): """ Возвращает datetime последней обработанной свечи из таблицы trend_analysis для указанной таблицы, либо None, если записи нет. """ query = text("SELECT datetime FROM trend_analysis WHERE table_name = :table_name") with engine.connect() as conn: result = conn.execute(query, {"table_name": table_name}).fetchone() return result[0] if result else None # ------------------------------------------------------------ # Обновление записи в таблице trend_analysis # ------------------------------------------------------------ def update_trend_analysis(table_name, instrument, dt, trend): """ Вставляет или обновляет запись для заданной таблицы. Использует INSERT ... ON DUPLICATE KEY UPDATE. """ insert_stmt = text(""" INSERT INTO trend_analysis (table_name, instrument, datetime, trend) VALUES (:table_name, :instrument, :datetime, :trend) ON DUPLICATE KEY UPDATE instrument = VALUES(instrument), datetime = VALUES(datetime), trend = VALUES(trend) """) delete_stmt = text("DELETE FROM trend_analysis WHERE table_name = :table_name") with engine.connect() as conn: if trend is None: conn.execute(delete_stmt, {"table_name": table_name}) else: conn.execute(insert_stmt, { "table_name": table_name, "instrument": instrument, "datetime": dt, "trend": trend }) conn.commit() # ------------------------------------------------------------ # Проверка, нужно ли обрабатывать таблицу в обычном режиме # ------------------------------------------------------------ def should_process_table(table_name, current_time): """ Определяет, нужно ли обрабатывать таблицу в зависимости от системного времени. Для W1: только в понедельник с 00:00 до 00:05. Для H1: первые 5 минут часа. Для D1: первые 5 минут дня. """ # ИЗМЕНЕНО: добавлено условие для W1 if table_name.endswith('_W1'): # Анализ недельных данных выполняется только в понедельник с 00:00 до 00:05 return current_time.weekday() == 0 and current_time.hour == 0 and current_time.minute < 5 elif table_name.endswith('_H1'): return current_time.minute < 5 elif table_name.endswith('_D1'): return current_time.hour == 0 and current_time.minute < 5 else: # Неизвестный суффикс — по умолчанию не обрабатываем return False # ------------------------------------------------------------ # Основная функция # ------------------------------------------------------------ def main(): parser = argparse.ArgumentParser(description='Анализ трендов по свечным таблицам') parser.add_argument('--inst', required=True, help='Название инструмента (например, BTC, ETH)') parser.add_argument('--force', action='store_true', help='Принудительно обработать все таблицы независимо от времени и наличия новых данных') parser.add_argument('--json', action='store_true', help='Вывод результатов в формате JSON (в stdout)') args = parser.parse_args() # Очищаем имя инструмента для использования в именах таблиц instrument_raw = args.inst instrument_clean = sanitize_table_name(instrument_raw) if not instrument_clean: msg = "ОШИБКА: имя инструмента после очистки пустое. Используйте буквы, цифры и подчёркивание." eprint(msg) if args.json: result = { "success": False, "message": msg, "results": [] } print(json.dumps(result, ensure_ascii=False, indent=2)) return # Формируем список таблиц для данного инструмента tables_to_process = [f"{instrument_clean}_{suffix}" for suffix in TIMEFRAME_SUFFIXES] # Проверяем/создаём таблицу с правильной структурой ensure_trend_table_exists() # Получаем inspector для проверки существования таблиц inspector = inspect(engine) current_time = datetime.now() # Список для сбора результатов (если нужен JSON) results = [] for table in tables_to_process: eprint(f"\n--- Обработка таблицы {table} ---") # Структура результата для этой таблицы res = { "instrument": instrument_clean, "table_name": table, "datetime": None, "trend": None, "trend_name": None, "success": False, "message": "" } # Проверка существования таблицы в базе данных if not inspector.has_table(table): msg = f"Таблица {table} не существует в базе данных. Пропускаем." eprint(f" {msg}") res["message"] = msg # Удаляем запись из trend_analysis, если она есть (опционально) last_analyzed_dt = get_last_analysis_datetime(table) if last_analyzed_dt is not None: update_trend_analysis(table, instrument_clean, None, None) results.append(res) continue # Проверяем, нужно ли обрабатывать таблицу в обычном режиме if not args.force and not should_process_table(table, current_time): msg = f"Пропускаем (текущее время {current_time.strftime('%H:%M')} не подходит для этой таблицы)." eprint(f" {msg}") res["message"] = msg results.append(res) continue # Получаем время последнего анализа из trend_analysis last_analyzed_dt = get_last_analysis_datetime(table) # Загружаем данные df = load_last_rows(table, LIMIT) if df.empty: msg = f"Нет данных в таблице {table}. Пропускаем." eprint(f" {msg}") res["message"] = msg # Если данных нет, возможно, нужно удалить запись if last_analyzed_dt is not None: update_trend_analysis(table, instrument_clean, None, None) results.append(res) continue # Определяем время последней свечи в загруженных данных last_candle_dt = df.iloc[-1]['datetime'] # Если не force и время последней свечи совпадает с последним анализом, пропускаем обновление if not args.force and last_analyzed_dt is not None and last_candle_dt == last_analyzed_dt: msg = f"Новых данных нет (последняя свеча {last_candle_dt} уже обработана). Пропускаем." eprint(f" {msg}") res["message"] = msg # Но мы всё равно можем вернуть последний известный тренд? Пока оставим как пропуск. results.append(res) continue # Проверяем, достаточно ли данных для расчёта EMA if len(df) < EMA_SLOW: msg = f"Недостаточно данных в {table} (меньше {EMA_SLOW} записей). Пропускаем." eprint(f" {msg}") res["message"] = msg update_trend_analysis(table, instrument_clean, None, None) results.append(res) continue # Рассчитываем тренд trend, last_dt = detect_trend_ema( df, pairs=[(5,13,1), (12,26,1), (50,200,1)], atr_period=14, atr_threshold=0.3, smooth_period=5 ) trend_names = {0: 'флэт', 1: 'бычий', 2: 'медвежий'} if trend is not None: eprint(f" Тренд на {last_dt}: {trend_names[trend]}") res["datetime"] = last_dt.isoformat() if last_dt else None res["trend"] = trend res["trend_name"] = trend_names[trend] res["success"] = True res["message"] = "OK" else: msg = f"Не удалось определить тренд (ошибка расчёта)" eprint(f" {msg}") res["message"] = msg update_trend_analysis(table, instrument_clean, last_dt, trend) results.append(res) eprint("\nАнализ трендов завершён.") if args.json: # Выводим JSON-массив результатов output = { "success": True, "message": "OK", "results": results } print(json.dumps(output, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()