import pandas as pd import numpy as np from sqlalchemy import create_engine, MetaData, Table, Column, Integer, Float, String, DateTime, TIMESTAMP, text, inspect from sqlalchemy.dialects.mysql import VARCHAR import warnings import argparse from datetime import datetime import re import json import sys warnings.filterwarnings('ignore') # ===================== ВЫВОД В STDERR ===================== 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 = 500 ATR_PERIOD = 14 BB_PERIOD = 20 BB_STD_DEV = 2 # Пороги (оставлены для классических интерпретаций, но динамические будут использоваться в приоритете) BB_WIDTH_SQUEEZE_STRONG = 0.05 BB_WIDTH_SQUEEZE_MODERATE = 0.1 BB_WIDTH_WIDE = 0.3 BB_PERCENT_OVERSOLD = 0.1 BB_PERCENT_OVERBOUGHT = 0.9 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) # ===================== РАБОТА С ТАБЛИЦЕЙ ===================== def ensure_volatility_table_exists(): 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('atr', Float, nullable=True, comment='Average True Range'), Column('bb_width', Float, nullable=True, comment='(Upper BB - Lower BB) / Middle BB'), Column('bb_percent_b', Float, nullable=True, comment='(Close - Lower BB) / (Upper BB - Lower BB)'), Column('keltner_width', Float, nullable=True, comment='Keltner Channel width (2*ATR/EMA)'), Column('signals', String(500), nullable=True, comment='Текстовые сигналы через разделитель'), Column('created_at', TIMESTAMP, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP')) ] target_table = Table('volatility_analysis', metadata, *target_columns, mysql_engine='InnoDB', mysql_default_charset='utf8mb4') if not inspector.has_table('volatility_analysis'): metadata.create_all(engine) eprint("Таблица volatility_analysis создана (с колонкой instrument).") return # Таблица существует — проверяем структуру existing_columns = {col['name'] for col in inspector.get_columns('volatility_analysis')} expected_columns = {col.name for col in target_columns if col.name != 'id'} expected_columns.add('id') missing = expected_columns - existing_columns extra = existing_columns - expected_columns if missing or extra: eprint(f"Несоответствие структуры. Пропущенные: {missing}, лишние: {extra}. Пересоздаём таблицу.") with engine.connect() as conn: conn.execute(text("DROP TABLE IF EXISTS volatility_analysis")) conn.commit() metadata.create_all(engine) eprint("Таблица volatility_analysis пересоздана (с колонкой instrument).") return # Проверяем уникальный ключ (оставляем по table_name, так как имя таблицы уникально) indexes = inspector.get_indexes('volatility_analysis') has_unique = any(idx['column_names'] == ['table_name'] and idx.get('unique', False) for idx in indexes) if not has_unique: with engine.connect() as conn: conn.execute(text(""" DELETE t1 FROM volatility_analysis t1 INNER JOIN volatility_analysis t2 WHERE t1.id > t2.id AND t1.table_name = t2.table_name """)) conn.execute(text("DELETE FROM volatility_analysis WHERE table_name IS NULL")) conn.execute(text("ALTER TABLE volatility_analysis ADD UNIQUE INDEX (table_name)")) conn.commit() eprint("Добавлен уникальный ключ на table_name.") # ===================== ЗАГРУЗКА ДАННЫХ ===================== def load_last_rows(table_name, limit=LIMIT): """ Загружает последние limit записей из таблицы, используя колонку timestamp (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 # ===================== РАСЧЁТ ИНДИКАТОРОВ ===================== def calculate_atr(df, period=ATR_PERIOD): if len(df) < period: return None, None high = df['High'] low = df['Low'] close_prev = df['Close'].shift(1) tr1 = high - low tr2 = (high - close_prev).abs() tr3 = (low - close_prev).abs() tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) atr_series = tr.rolling(window=period).mean() return atr_series, float(atr_series.iloc[-1]) if not atr_series.empty else None def calculate_bb(df, period=BB_PERIOD, std_dev=BB_STD_DEV): if len(df) < period: return None, None, None, None close = df['Close'] sma = close.rolling(window=period).mean() std = close.rolling(window=period).std() upper_band = sma + (std * std_dev) lower_band = sma - (std * std_dev) bb_width_series = (upper_band - lower_band) / sma bb_percent_b_series = (close - lower_band) / (upper_band - lower_band) return (float(bb_width_series.iloc[-1]) if not bb_width_series.empty else None, float(bb_percent_b_series.iloc[-1]) if not bb_percent_b_series.empty else None, bb_width_series, bb_percent_b_series) def calculate_keltner(df, atr_period=ATR_PERIOD, ema_period=BB_PERIOD, multiplier=BB_STD_DEV): """ Рассчитывает ширину канала Кельтнера (верхняя - нижняя) / EMA. Возвращает текущую ширину в процентах. """ if len(df) < max(atr_period, ema_period): return None ema = df['Close'].ewm(span=ema_period, adjust=False).mean() atr_series, current_atr = calculate_atr(df, atr_period) if atr_series is None or len(atr_series) < 1: return None last_ema = ema.iloc[-1] last_atr = atr_series.iloc[-1] if pd.isna(last_ema) or pd.isna(last_atr) or last_ema == 0: return None width = (2 * multiplier * last_atr) / last_ema return float(width) # ===================== ГЕНЕРАЦИЯ СООБЩЕНИЙ ===================== def generate_volatility_signals(df, atr_series, current_atr, bb_width, bb_percent_b, bb_width_series, keltner_width=None): """ Анализирует показатели и возвращает список строк с интерпретацией и торговыми сигналами. """ messages = [] # ---- Динамика ATR ---- if atr_series is not None and len(atr_series) > ATR_PERIOD * 2: atr_valid = atr_series.dropna() if len(atr_valid) > 1: current = atr_valid.iloc[-1] historical = atr_valid.iloc[:-1].mean() if historical > 0: ratio = current / historical if ratio > 1.5: messages.append(f"⚠️ ATR резко вырос: в {ratio:.1f} раз выше среднего. Высокая волатильность, возможны резкие движения.") elif ratio < 0.5: messages.append(f"✅ ATR сильно упал: в {1/ratio:.1f} раз ниже среднего. Рынок затихает, возможна консолидация.") elif ratio > 1.2: messages.append(f"📈 ATR выше среднего (в {ratio:.1f} раз). Волатильность растёт, тренд может ускориться.") elif ratio < 0.8: messages.append(f"📉 ATR ниже среднего (в {1/ratio:.1f} раз). Волатильность падает, рынок входит в фазу накопления.") # ---- Динамическое определение СЖАТИЯ (Squeeze) на основе скользящей средней BB Width ---- if bb_width is not None and bb_width_series is not None and len(bb_width_series.dropna()) > 10: bb_valid = bb_width_series.dropna() lookback = 10 if len(bb_valid) > lookback: avg_bb_width = bb_valid.iloc[-lookback-1:-1].mean() if pd.notna(avg_bb_width): if bb_width < avg_bb_width * 0.8: messages.append("💥 АДАПТИВНОЕ СИЛЬНОЕ СЖАТИЕ полос Боллинджера! Высокая вероятность взрыва волатильности.") elif bb_width < avg_bb_width: messages.append("🔍 АДАПТИВНОЕ УМЕРЕННОЕ СЖАТИЕ полос Боллинджера. Рынок готовится к движению.") elif bb_width > avg_bb_width * 1.5: messages.append("🌊 АДАПТИВНОЕ РАСШИРЕНИЕ полос Боллинджера. Тренд может быть сильным.") else: messages.append(f"📊 Адаптивная ширина полос Боллинджера в норме (текущая {bb_width:.2%}, средняя {avg_bb_width:.2%}).") else: if bb_width is not None: if bb_width < BB_WIDTH_SQUEEZE_STRONG: messages.append("💥 СИЛЬНОЕ СЖАТИЕ полос Боллинджера! (статический порог)") elif bb_width < BB_WIDTH_SQUEEZE_MODERATE: messages.append("🔍 Умеренное сжатие полос Боллинджера. (статический порог)") elif bb_width > BB_WIDTH_WIDE: messages.append("🌊 Широкие полосы Боллинджера. (статический порог)") else: messages.append(f"📊 Ширина полос Боллинджера в норме ({bb_width:.2%}).") # ---- Положение цены (%B) ---- if bb_percent_b is not None: if bb_percent_b < BB_PERCENT_OVERSOLD: messages.append("📉 Цена у НИЖНЕЙ границы Боллинджера. Возможен отскок вверх.") elif bb_percent_b > BB_PERCENT_OVERBOUGHT: messages.append("📈 Цена у ВЕРХНЕЙ границы Боллинджера. Возможна коррекция вниз.") elif 0.4 < bb_percent_b < 0.6: messages.append("⚖️ Цена в нейтральной зоне Боллинджера. Определённого сигнала нет.") else: messages.append(f"📐 %B = {bb_percent_b:.2f} — цена смещена, но не экстремально.") # ---- ТОРГОВЫЕ СИГНАЛЫ с подтверждением от ATR ---- if bb_width is not None and bb_percent_b is not None: atr_confirmed = False if atr_series is not None and len(atr_series) > ATR_PERIOD: atr_valid = atr_series.dropna() if len(atr_valid) > ATR_PERIOD: current_atr_val = atr_valid.iloc[-1] historical_atr = atr_valid.iloc[-ATR_PERIOD-1:-1].mean() if historical_atr > 0 and (current_atr_val / historical_atr) > 1.3: atr_confirmed = True if bb_width < BB_WIDTH_SQUEEZE_STRONG: if bb_percent_b < BB_PERCENT_OVERSOLD: msg = "💰 СИГНАЛ К ПОКУПКЕ: Сильное сжатие + перепроданность." if atr_confirmed: msg += " ПОДТВЕРЖДЕНО ростом ATR." else: msg += " (подтверждение ATR отсутствует)." messages.append(msg) elif bb_percent_b > BB_PERCENT_OVERBOUGHT: msg = "💰 СИГНАЛ К ПРОДАЖЕ: Сильное сжатие + перекупленность." if atr_confirmed: msg += " ПОДТВЕРЖДЕНО ростом ATR." else: msg += " (подтверждение ATR отсутствует)." messages.append(msg) if bb_percent_b < BB_PERCENT_OVERSOLD and atr_confirmed: messages.append("💰 СИГНАЛ К ПОКУПКЕ: Перепроданность + подтверждённый рост волатильности (ATR).") elif bb_percent_b > BB_PERCENT_OVERBOUGHT and atr_confirmed: messages.append("💰 СИГНАЛ К ПРОДАЖЕ: Перекупленность + подтверждённый рост волатильности (ATR).") # ---- Сравнение с Каналом Кельтнера ---- if keltner_width is not None and bb_width is not None: if bb_width > keltner_width * 1.2: messages.append(f"📊 Полосы Боллинджера ({bb_width:.2%}) шире канала Кельтнера ({keltner_width:.2%}). Возможны резкие движения, но фильтруйте ложные пробои.") elif keltner_width > bb_width * 1.2: messages.append(f"📊 Канал Кельтнера ({keltner_width:.2%}) шире полос Боллинджера. Движение более плавное, в соответствии с ATR.") return messages # ===================== РАБОТА С ЗАПИСЯМИ ===================== def get_last_volatility_datetime(table_name): query = text("SELECT datetime FROM volatility_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 def update_volatility_analysis(table_name, instrument, dt, atr, bb_width, bb_percent_b, keltner_width, signals_str): delete_stmt = text("DELETE FROM volatility_analysis WHERE table_name = :table_name") insert_stmt = text(""" INSERT INTO volatility_analysis (table_name, instrument, datetime, atr, bb_width, bb_percent_b, keltner_width, signals) VALUES (:table_name, :instrument, :datetime, :atr, :bb_width, :bb_percent_b, :keltner_width, :signals) ON DUPLICATE KEY UPDATE instrument = VALUES(instrument), datetime = VALUES(datetime), atr = VALUES(atr), bb_width = VALUES(bb_width), bb_percent_b = VALUES(bb_percent_b), keltner_width = VALUES(keltner_width), signals = VALUES(signals) """) with engine.connect() as conn: if atr is None and bb_width is None and bb_percent_b is None and keltner_width is None and not signals_str: conn.execute(delete_stmt, {"table_name": table_name}) else: conn.execute(insert_stmt, { "table_name": table_name, "instrument": instrument, "datetime": dt, "atr": atr, "bb_width": bb_width, "bb_percent_b": bb_percent_b, "keltner_width": keltner_width, "signals": signals_str }) 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 # Формируем список таблиц для данного инструмента all_possible_tables = [f"{instrument_clean}_{suffix}" for suffix in TIMEFRAME_SUFFIXES] # Проверяем существование таблиц в базе данных inspector = inspect(engine) existing_tables = [] for table in all_possible_tables: if inspector.has_table(table): existing_tables.append(table) else: eprint(f"Предупреждение: таблица {table} не найдена в базе данных. Пропускаем.") if not existing_tables: msg = f"ОШИБКА: для инструмента {instrument_clean} не найдено ни одной таблицы (суффиксы {TIMEFRAME_SUFFIXES})." eprint(msg) if args.json: result = { "success": False, "message": msg, "results": [] } print(json.dumps(result, ensure_ascii=False, indent=2)) return ensure_volatility_table_exists() current_time = datetime.now() results = [] for table in existing_tables: eprint(f"\n{'='*60}") eprint(f"Таблица: {table} (инструмент: {instrument_clean})") eprint('='*60) res = { "instrument": instrument_clean, "table_name": table, "datetime": None, "atr": None, "bb_width": None, "bb_percent_b": None, "keltner_width": None, "signals": None, "success": False, "message": "" } 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 last_analyzed_dt = get_last_volatility_datetime(table) df = load_last_rows(table, LIMIT) if df.empty: msg = f"Нет данных." eprint(f" {msg}") res["message"] = msg if last_analyzed_dt is not None: update_volatility_analysis(table, instrument_clean, None, None, None, None, None, None) results.append(res) continue last_candle_dt = df.iloc[-1]['datetime'] 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 min_required = max(ATR_PERIOD, BB_PERIOD) if len(df) < min_required: msg = f"Недостаточно данных (нужно {min_required}, имеется {len(df)})." eprint(f" {msg}") update_volatility_analysis(table, instrument_clean, None, None, None, None, None, None) res["message"] = msg results.append(res) continue atr_series, current_atr = calculate_atr(df, ATR_PERIOD) bb_width, bb_percent_b, bb_width_series, bb_percent_b_series = calculate_bb(df, BB_PERIOD, BB_STD_DEV) keltner_width = calculate_keltner(df, ATR_PERIOD, BB_PERIOD, BB_STD_DEV) eprint(f" Последняя свеча: {last_candle_dt}") if current_atr is not None: eprint(f" ATR({ATR_PERIOD}): {current_atr:.2f}") if bb_width is not None: eprint(f" BB Width({BB_PERIOD}): {bb_width:.4f} ({bb_width:.2%})") if bb_percent_b is not None: eprint(f" %B: {bb_percent_b:.4f}") if keltner_width is not None: eprint(f" Keltner Width: {keltner_width:.4f} ({keltner_width:.2%})") signals = generate_volatility_signals(df, atr_series, current_atr, bb_width, bb_percent_b, bb_width_series, keltner_width) signals_str = ' | '.join(signals) if signals else None if signals: eprint("\n 📋 ИНТЕРПРЕТАЦИЯ И СИГНАЛЫ:") for msg in signals: eprint(f" {msg}") else: eprint("\n ➖ Нет выраженных сигналов.") update_volatility_analysis(table, instrument_clean, last_candle_dt, current_atr, bb_width, bb_percent_b, keltner_width, signals_str) # Заполняем результат res["datetime"] = last_candle_dt.isoformat() if last_candle_dt else None res["atr"] = current_atr res["bb_width"] = bb_width res["bb_percent_b"] = bb_percent_b res["keltner_width"] = keltner_width res["signals"] = signals_str res["success"] = True res["message"] = "OK" results.append(res) eprint("\n✅ Анализ волатильности завершён.") if args.json: output = { "success": True, "message": "OK", "results": results } print(json.dumps(output, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()