#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Модуль волнового анализа ценовых графиков Интеграция с существующей системой волатильности """ import pandas as pd import numpy as np from sqlalchemy import create_engine, MetaData, Table, Column, Integer, Float, String, DateTime, text, inspect from sqlalchemy.dialects.mysql import VARCHAR from sqlalchemy.types import TIMESTAMP import warnings from datetime import datetime import json import sys import re from contextlib import contextmanager warnings.filterwarnings('ignore') # ------------------------------------------------------------ # Контекстный менеджер для подавления вывода в stdout # ------------------------------------------------------------ @contextmanager def suppress_stdout(): """Временно перенаправляет stdout в stderr, чтобы подавить вывод библиотек.""" old_stdout = sys.stdout sys.stdout = sys.stderr try: yield finally: sys.stdout = old_stdout # ------------------------------------------------------------ # Попытка импорта дополнительных библиотек с подавлением вывода # ------------------------------------------------------------ try: with suppress_stdout(): import taew TAEW_AVAILABLE = True except ImportError: TAEW_AVAILABLE = False print("Предупреждение: taew не установлен. Волны Эллиотта будут недоступны.", file=sys.stderr) try: with suppress_stdout(): import pytrendseries PYTREENDSERIES_AVAILABLE = True except ImportError: PYTREENDSERIES_AVAILABLE = False print("Предупреждение: pytrendseries не установлен. Анализ трендов будет базовым.", file=sys.stderr) try: with suppress_stdout(): from fidat import WaveTrendOscillator FIDAT_AVAILABLE = True except ImportError: FIDAT_AVAILABLE = False print("Предупреждение: fidat не установлен. WTO будет рассчитан вручную.", file=sys.stderr) # ===================== КОНФИГУРАЦИЯ ===================== DB_CONFIG = { 'host': 'nlbotinterface.ru', 'port': 3306, 'database': 'bitcoin_tickers', 'user': 'bitcoin', 'password': 'g49020007', } # Параметры волнового анализа WTO_ESA_PERIOD = 10 # Период EMA для ESA WTO_D_PERIOD = 21 # Период для отклонения WTO_AVG_PERIOD = 4 # Период сглаживания WTO WTO_OVERSOLD = -60 # Уровень перепроданности WTO_OVERBOUGHT = 60 # Уровень перекупленности TREND_WINDOW = 30 # Окно для обнаружения трендов (дней) MIN_TREND_LENGTH = 5 # Минимальная длина тренда 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_wave_analysis_table_exists(): """Создаёт таблицу для хранения результатов волнового анализа.""" metadata = MetaData() inspector = inspect(engine) 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_type', VARCHAR(20), nullable=True, comment='uptrend/downtrend/sideways'), Column('trend_strength', Float, nullable=True, comment='Сила тренда (0-100)'), Column('wave_count', Integer, nullable=True, comment='Количество обнаруженных волн'), Column('elliott_waves', String(200), nullable=True, comment='JSON с точками волн Эллиотта'), Column('wto_value', Float, nullable=True, comment='Wave Trend Oscillator значение'), Column('wto_signal', VARCHAR(50), nullable=True, comment='Сигнал WTO (oversold/overbought/cross)'), Column('fib_levels', String(500), nullable=True, comment='Уровни Фибоначчи'), Column('next_target', Float, nullable=True, comment='Следующая целевая цена'), Column('signals', String(500), nullable=True, comment='Объединённые сигналы'), Column('created_at', TIMESTAMP, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP')) ] target_table = Table('wave_analysis', metadata, *target_columns, mysql_engine='InnoDB', mysql_default_charset='utf8mb4') if not inspector.has_table('wave_analysis'): metadata.create_all(engine) print("Таблица wave_analysis создана.", file=sys.stderr) return # Проверка структуры и обновление при необходимости existing_columns = {col['name'] for col in inspector.get_columns('wave_analysis')} expected_columns = {col.name for col in target_columns if col.name != 'id'} expected_columns.add('id') missing = expected_columns - existing_columns if missing: print(f"Обновление структуры таблицы. Добавление колонок: {missing}", file=sys.stderr) with engine.connect() as conn: conn.execute(text("DROP TABLE IF EXISTS wave_analysis")) conn.commit() metadata.create_all(engine) # ===================== ЗАГРУЗКА ДАННЫХ ===================== def load_data_for_analysis(table_name, limit=200): # <-- ИЗМЕНЕНО: лимит увеличен до 200 """ Загружает данные для волнового анализа. Возвращает 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 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 analyze_trends(df, window=TREND_WINDOW, min_length=MIN_TREND_LENGTH): """ Анализирует тренды в данных. Использует pytrendseries если доступно, иначе простой алгоритм. """ if len(df) < window: return "sideways", 0.0, None close_prices = df['Close'].values dates = df['datetime'].values if PYTREENDSERIES_AVAILABLE: try: # Подготовка данных для pytrendseries trend_data = df[['datetime', 'Close']].copy() trend_data.set_index('datetime', inplace=True) trend_data.columns = ['close_price'] # Поиск нисходящих трендов (подавляем вывод) with suppress_stdout(): downtrends = pytrendseries.detecttrend( trend_data, trend="downtrend", window=window ) uptrends = pytrendseries.detecttrend( trend_data, trend="uptrend", window=window ) # Определяем текущий тренд last_date = dates[-1] current_trend = "sideways" trend_strength = 0.0 fib_levels = None # Проверяем последний нисходящий тренд if not downtrends.empty: last_downtrend = downtrends.iloc[-1] if last_downtrend['to'] >= last_date: current_trend = "downtrend" trend_strength = last_downtrend['drawdown'] * 100 if 'price0' in last_downtrend and 'price1' in last_downtrend: high = last_downtrend['price0'] low = last_downtrend['price1'] fib_levels = calculate_fibonacci_levels(high, low, 'down') # Проверяем последний восходящий тренд if current_trend == "sideways" and not uptrends.empty: last_uptrend = uptrends.iloc[-1] if last_uptrend['to'] >= last_date: current_trend = "uptrend" trend_strength = last_uptrend['drawdown'] * 100 if 'price0' in last_uptrend and 'price1' in last_uptrend: low = last_uptrend['price0'] high = last_uptrend['price1'] fib_levels = calculate_fibonacci_levels(low, high, 'up') return current_trend, trend_strength, fib_levels except Exception as e: print(f"Ошибка в pytrendseries: {e}", file=sys.stderr) # Fallback: простой анализ на основе скользящих средних sma_short = df['Close'].rolling(window=10).mean() sma_long = df['Close'].rolling(window=30).mean() if len(sma_short) < 30 or len(sma_long) < 30: return "sideways", 0.0, None last_short = sma_short.iloc[-1] last_long = sma_long.iloc[-1] prev_short = sma_short.iloc[-2] prev_long = sma_long.iloc[-2] if last_short > last_long and prev_short <= prev_long: current_trend = "uptrend" trend_strength = ((last_short / last_long) - 1) * 100 elif last_short < last_long and prev_short >= prev_long: current_trend = "downtrend" trend_strength = ((last_long / last_short) - 1) * 100 else: current_trend = "sideways" trend_strength = 0.0 return current_trend, trend_strength, None # ===================== ВОЛНЫ ЭЛЛИОТТА ===================== def analyze_elliott_waves(df): """ Анализирует волны Эллиотта в данных. Использует библиотеку taew если доступна. """ if not TAEW_AVAILABLE: return 0, None, None if len(df) < 50: # Минимум данных для волн Эллиотта return 0, None, None close_prices = df['Close'].values try: # Подавляем вывод taew with suppress_stdout(): upward_waves = taew.Alternative_ElliottWave_label_upward(close_prices) downward_waves = taew.Alternative_ElliottWave_label_downward(close_prices) all_waves = [] # Обработка восходящих волн if upward_waves and len(upward_waves) > 0: for wave in upward_waves[:3]: if 'x' in wave and len(wave['x']) >= 5: wave_data = { 'type': 'upward', 'points': wave['x'], 'indices': wave.get('z', []), 'count': len(wave['x']) } all_waves.append(wave_data) # Обработка нисходящих волн if downward_waves and len(downward_waves) > 0: for wave in downward_waves[:3]: if 'x' in wave and len(wave['x']) >= 5: wave_data = { 'type': 'downward', 'points': wave['x'], 'indices': wave.get('z', []), 'count': len(wave['x']) } all_waves.append(wave_data) if all_waves: # Берём самую свежую волну latest_wave = max(all_waves, key=lambda w: max(w['indices']) if w['indices'] else 0) # Прогнозируем следующую цель next_target = predict_next_wave_target(df, latest_wave) return len(all_waves), json.dumps(all_waves, default=str), next_target except Exception as e: print(f"Ошибка в анализе волн Эллиотта: {e}", file=sys.stderr) return 0, None, None def predict_next_wave_target(df, wave_data): """ Прогнозирует следующую цель на основе волновой структуры. """ if wave_data['type'] == 'upward' and len(wave_data['points']) >= 5: if len(wave_data['points']) >= 3: wave3_high = wave_data['points'][3] wave2_low = wave_data['points'][1] wave1_len = wave_data['points'][1] - wave_data['points'][0] target = wave3_high + (wave1_len * 1.618) return float(target) elif wave_data['type'] == 'downward' and len(wave_data['points']) >= 5: if len(wave_data['points']) >= 3: wave3_low = wave_data['points'][3] wave2_high = wave_data['points'][1] wave1_len = wave_data['points'][0] - wave_data['points'][1] target = wave3_low - (wave1_len * 1.618) return float(target) return None # ===================== WAVE TREND OSCILLATOR ===================== def calculate_wto(df, esa_period=WTO_ESA_PERIOD, d_period=WTO_D_PERIOD, avg_period=WTO_AVG_PERIOD): """ Рассчитывает Wave Trend Oscillator. """ if len(df) < max(esa_period, d_period, avg_period) + 10: return None, None df = df.copy() df['ap'] = (df['High'] + df['Low'] + df['Close']) / 3 df['esa'] = df['ap'].ewm(span=esa_period, adjust=False).mean() df['diff'] = (df['ap'] - df['esa']).abs() df['d'] = df['diff'].ewm(span=d_period, adjust=False).mean() df['ci'] = 0.0 mask = df['d'] > 0 df.loc[mask, 'ci'] = (df.loc[mask, 'ap'] - df.loc[mask, 'esa']) / (0.015 * df.loc[mask, 'd']) df['wto'] = df['ci'].ewm(span=avg_period, adjust=False).mean() if pd.isna(df['wto'].iloc[-1]): return None, None current_wto = float(df['wto'].iloc[-1]) prev_wto = float(df['wto'].iloc[-2]) if len(df) > 1 else current_wto signal = None if current_wto < WTO_OVERSOLD: signal = "oversold" elif current_wto > WTO_OVERBOUGHT: signal = "overbought" elif prev_wto < 0 and current_wto >= 0: signal = "bullish_cross" elif prev_wto > 0 and current_wto <= 0: signal = "bearish_cross" return current_wto, signal # ===================== УРОВНИ ФИБОНАЧЧИ ===================== def calculate_fibonacci_levels(start_price, end_price, trend_type): """ Рассчитывает ключевые уровни Фибоначчи. """ fib_levels = { '0.0': start_price, '0.236': start_price + (end_price - start_price) * 0.236, '0.382': start_price + (end_price - start_price) * 0.382, '0.5': start_price + (end_price - start_price) * 0.5, '0.618': start_price + (end_price - start_price) * 0.618, '0.786': start_price + (end_price - start_price) * 0.786, '1.0': end_price, } if trend_type == 'up': fib_levels['1.272'] = end_price + (end_price - start_price) * 0.272 fib_levels['1.618'] = end_price + (end_price - start_price) * 0.618 elif trend_type == 'down': fib_levels['1.272'] = end_price - (start_price - end_price) * 0.272 fib_levels['1.618'] = end_price - (start_price - end_price) * 0.618 return json.dumps(fib_levels, default=float) # ===================== ГЕНЕРАЦИЯ СИГНАЛОВ ===================== def generate_wave_signals(trend_type, trend_strength, wave_count, wto_value, wto_signal, next_target): """ Генерирует текстовые сигналы на основе волнового анализа. """ signals = [] if trend_type == "uptrend": if trend_strength > 5: signals.append(f"📈 СИЛЬНЫЙ ВОСХОДЯЩИЙ ТРЕНД (сила {trend_strength:.1f}%)") elif trend_strength > 2: signals.append(f"📈 Восходящий тренд (сила {trend_strength:.1f}%)") elif trend_type == "downtrend": if trend_strength > 5: signals.append(f"📉 СИЛЬНЫЙ НИСХОДЯЩИЙ ТРЕНД (сила {trend_strength:.1f}%)") elif trend_strength > 2: signals.append(f"📉 Нисходящий тренд (сила {trend_strength:.1f}%)") if wave_count > 0: signals.append(f"🌊 Обнаружено {wave_count} волновых структур") if wto_signal == "oversold": signals.append("💰 WTO: Перепроданность - возможен отскок вверх") elif wto_signal == "overbought": signals.append("💰 WTO: Перекупленность - возможна коррекция вниз") elif wto_signal == "bullish_cross": signals.append("🟢 WTO: Бычий crossover - смена тренда вверх") elif wto_signal == "bearish_cross": signals.append("🔴 WTO: Медвежий crossover - смена тренда вниз") return ' | '.join(signals) if signals else None def clear_old_wave_analysis(table_name, instrument): """Удаляет старые записи для данного инструмента и таблицы.""" delete_stmt = text(""" DELETE FROM wave_analysis WHERE table_name = :table_name AND instrument = :instrument """) with engine.connect() as conn: conn.execute(delete_stmt, {"table_name": table_name, "instrument": instrument}) conn.commit() # ===================== СОХРАНЕНИЕ РЕЗУЛЬТАТОВ ===================== def save_wave_analysis(table_name, instrument, dt, trend_type, trend_strength, wave_count, elliott_waves, wto_value, wto_signal, fib_levels, next_target, signals): """ Сохраняет результаты волнового анализа в БД, предварительно очистив старые записи. """ clear_old_wave_analysis(table_name, instrument) insert_stmt = text(""" INSERT INTO wave_analysis ( table_name, instrument, datetime, trend_type, trend_strength, wave_count, elliott_waves, wto_value, wto_signal, fib_levels, next_target, signals ) VALUES ( :table_name, :instrument, :datetime, :trend_type, :trend_strength, :wave_count, :elliott_waves, :wto_value, :wto_signal, :fib_levels, :next_target, :signals ) """) with engine.connect() as conn: conn.execute(insert_stmt, { "table_name": table_name, "instrument": instrument, "datetime": dt, "trend_type": trend_type, "trend_strength": trend_strength, "wave_count": wave_count, "elliott_waves": elliott_waves, "wto_value": wto_value, "wto_signal": wto_signal, "fib_levels": fib_levels, "next_target": next_target, "signals": signals }) conn.commit() # ===================== ОСНОВНАЯ ФУНКЦИЯ АНАЛИЗА ===================== def analyze_waves(table_name, instrument, limit=200): # <-- ИЗМЕНЕНО: лимит по умолчанию 200 """ Выполняет полный волновой анализ для указанной таблицы. """ df = load_data_for_analysis(table_name, limit) if df.empty: return { "success": False, "message": "Нет данных", "table_name": table_name, "instrument": instrument } # Трендовый анализ trend_type, trend_strength, fib_levels = analyze_trends(df) # Волны Эллиотта wave_count, elliott_waves, next_target = analyze_elliott_waves(df) # Wave Trend Oscillator wto_value, wto_signal = calculate_wto(df) # Если не получили уровни Фибоначчи из трендов, но есть волны if fib_levels is None and wave_count > 0 and elliott_waves: try: waves = json.loads(elliott_waves) if waves and len(waves) > 0: points = waves[0].get('points', []) if len(points) >= 2: if waves[0]['type'] == 'upward': fib_levels = calculate_fibonacci_levels(points[0], points[-1], 'up') else: fib_levels = calculate_fibonacci_levels(points[0], points[-1], 'down') except: pass # Генерация сигналов signals = generate_wave_signals( trend_type, trend_strength, wave_count, wto_value, wto_signal, next_target ) last_dt = df.iloc[-1]['datetime'] # Сохранение результатов save_wave_analysis( table_name, instrument, last_dt, trend_type, trend_strength, wave_count, elliott_waves, wto_value, wto_signal, fib_levels, next_target, signals ) # Формирование результата result = { "success": True, "message": "OK", "table_name": table_name, "instrument": instrument, "datetime": last_dt.isoformat(), "trend_type": trend_type, "trend_strength": trend_strength, "wave_count": wave_count, "wto_value": wto_value, "wto_signal": wto_signal, "next_target": next_target, "signals": signals } return result # ===================== ТОЧКА ВХОДА ===================== if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description='Волновой анализ ценовых графиков') parser.add_argument('--inst', required=True, help='Название инструмента (например, BTC, ETH)') parser.add_argument('--table', help='Конкретная таблица для анализа (опционально)') parser.add_argument('--limit', type=int, default=200, help='Количество свечей для анализа (по умолчанию 200)') # <-- ИЗМЕНЕНО parser.add_argument('--json', action='store_true', help='Вывод результатов в JSON') args = parser.parse_args() instrument_clean = sanitize_table_name(args.inst) ensure_wave_analysis_table_exists() results = [] if args.table: result = analyze_waves(args.table, instrument_clean, args.limit) results.append(result) print(f"Анализ таблицы {args.table} завершён.", file=sys.stderr) else: # ИЗМЕНЕНО: вместо M5 теперь W1 suffixes = ['W1', 'H1', 'D1'] inspector = inspect(engine) for suffix in suffixes: table_name = f"{instrument_clean}_{suffix}" if inspector.has_table(table_name): print(f"\nАнализ {table_name}...", file=sys.stderr) result = analyze_waves(table_name, instrument_clean, args.limit) results.append(result) else: print(f"Таблица {table_name} не найдена", file=sys.stderr) if args.json: output = { "success": True, "message": "Волновой анализ завершён", "results": results } # Печатаем только JSON в stdout print(json.dumps(output, ensure_ascii=False, indent=2))