#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Скрипт-агрегатор и распределитель данных анализов. Запускается с параметром --server для запуска WebSocket-сервера и TCP-приёмника. При обычном запуске читает JSON из stdin и отправляет его на TCP-порт сервера для обновления хранилища и рассылки клиентам. """ import asyncio import websockets import socket import threading import json import sys import argparse import logging from datetime import datetime logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) WEBSOCKET_PORT = 8765 TCP_PORT = 9050 TCP_HOST = 'localhost' HISTORY_LIMIT = 3 connected_clients = set() # Структура хранилища: # { # "BTC": { # "BTC_H1": { # "trend": {...}, # "volatility": {...}, # "candle": {...}, # "levels": [...], # "wave": {...}, # "pa_volume": {...}, # "pullback": {...}, # "trading_signal": {...}, # "next_extreme": {...}, # "candle_direction": {...}, # "candle_change": {...} # }, # ... # }, # ... # } data_store = {} store_lock = asyncio.Lock() # ------------------------------------------------------------ # WebSocket сервер # ------------------------------------------------------------ async def websocket_handler(websocket): connected_clients.add(websocket) client_id = id(websocket) logger.info(f"WebSocket клиент {client_id} подключился. Всего клиентов: {len(connected_clients)}") try: async with store_lock: if data_store: for instrument, inst_data in data_store.items(): payload = { 'instrument': instrument, 'data': inst_data, 'timestamp': datetime.now().isoformat() } await websocket.send(json.dumps(payload, default=str)) await websocket.wait_closed() finally: connected_clients.remove(websocket) logger.info(f"WebSocket клиент {client_id} отключился. Осталось: {len(connected_clients)}") async def broadcast_data(instrument_data): if not connected_clients: return message = json.dumps(instrument_data, ensure_ascii=False, default=str) tasks = [asyncio.create_task(client.send(message)) for client in connected_clients] if tasks: await asyncio.gather(*tasks, return_exceptions=True) logger.debug(f"Отправлено сообщение {len(tasks)} клиентам") async def run_websocket_server(): async with websockets.serve(websocket_handler, "0.0.0.0", WEBSOCKET_PORT): logger.info(f"WebSocket сервер запущен на порту {WEBSOCKET_PORT}") await asyncio.Future() # ------------------------------------------------------------ # Обновление хранилища из JSON (новая версия) # ------------------------------------------------------------ async def update_store_from_json(json_data): global data_store if not isinstance(json_data, dict): logger.error("Получен некорректный JSON (не словарь)") return success = json_data.get('success', False) if not success: logger.warning("JSON помечен как неуспешный, пропускаем") return results = json_data.get('results', []) if not results: logger.warning("JSON не содержит результатов") return logger.info(f"Получен JSON: {json_data.get('message', '')}, results: {len(results)}") # Определяем инструмент (предполагаем, что все результаты для одного инструмента) instrument = None for res in results: if res.get('instrument'): instrument = res.get('instrument') break if not instrument: logger.error("Не удалось определить инструмент из результатов") return logger.info(f"Инструмент: {instrument}") async with store_lock: if instrument not in data_store: data_store[instrument] = {} for res in results: table_name = res.get('table_name') if not table_name: continue if table_name not in data_store[instrument]: data_store[instrument][table_name] = {} # Сохраняем время последней свечи if 'last_candle_time' in res: data_store[instrument][table_name]['last_candle_time'] = res['last_candle_time'] analysis_results = res.get('results', {}) if not isinstance(analysis_results, dict): continue # Обновление истории для каждого типа прогноза for pred_key in ['next_extreme', 'candle_direction', 'candle_change']: if pred_key in analysis_results and analysis_results[pred_key] is not None: new_entry = analysis_results[pred_key].copy() # Добавляем timestamp, если его нет if 'timestamp' not in new_entry: new_entry['timestamp'] = datetime.now().isoformat() update_history(data_store[instrument][table_name], pred_key, new_entry) # Остальные данные (trend, levels, candle и т.д.) обновляются как обычно # candle_patterns → candle if 'candle_patterns' in analysis_results and analysis_results['candle_patterns'] is not None: candle_data = analysis_results['candle_patterns'] if isinstance(candle_data, dict) and 'pattern_history' in candle_data and isinstance(candle_data['pattern_history'], list): sorted_history = sorted( candle_data['pattern_history'], key=lambda x: x.get('end_time', ''), reverse=True ) candle_data['pattern_history'] = sorted_history[:7] data_store[instrument][table_name]['candle'] = candle_data # stats_summary if 'stats_summary' in analysis_results and analysis_results['stats_summary'] is not None: data_store[instrument][table_name]['stats_summary'] = analysis_results['stats_summary'] # levels if 'levels' in analysis_results and analysis_results['levels'] is not None: data_store[instrument][table_name]['levels'] = analysis_results['levels'] # pa_volume if 'pa_volume' in analysis_results and analysis_results['pa_volume'] is not None: data_store[instrument][table_name]['pa_volume'] = analysis_results['pa_volume'] # trend if 'trend' in analysis_results and analysis_results['trend'] is not None: data_store[instrument][table_name]['trend'] = analysis_results['trend'] # volatility if 'volatility' in analysis_results and analysis_results['volatility'] is not None: data_store[instrument][table_name]['volatility'] = analysis_results['volatility'] # wave if 'wave' in analysis_results and analysis_results['wave'] is not None: data_store[instrument][table_name]['wave'] = analysis_results['wave'] # pullback if 'pullback_analysis' in analysis_results and analysis_results['pullback_analysis'] is not None: data_store[instrument][table_name]['pullback'] = analysis_results['pullback_analysis'] # trading_signal if 'trading_signal' in analysis_results and analysis_results['trading_signal'] is not None: data_store[instrument][table_name]['trading_signal'] = analysis_results['trading_signal'] # atr_analysis if 'atr_analysis' in analysis_results and analysis_results['atr_analysis'] is not None: data_store[instrument][table_name]['atr_analysis'] = analysis_results['atr_analysis'] # wave_range_analysis if 'wave_range_analysis' in analysis_results and analysis_results['wave_range_analysis'] is not None: data_store[instrument][table_name]['wave_range_analysis'] = analysis_results['wave_range_analysis'] instrument_payload = { 'instrument': instrument, 'data': data_store[instrument], 'timestamp': datetime.now().isoformat() } return instrument_payload # ------------------------------------------------------------ # TCP сервер для приёма данных от локальных отправителей # ------------------------------------------------------------ def tcp_server(loop): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind((TCP_HOST, TCP_PORT)) s.listen() logger.info(f"TCP сервер запущен на {TCP_HOST}:{TCP_PORT}") while True: conn, addr = s.accept() thread = threading.Thread(target=handle_tcp_connection, args=(conn, addr, loop)) thread.daemon = True thread.start() def handle_tcp_connection(conn, addr, loop): logger.info(f"TCP соединение от {addr}") try: data = b"" while True: chunk = conn.recv(4096) if not chunk: break data += chunk if data: try: json_data = json.loads(data.decode('utf-8')) logger.info(f"Получены данные от {addr}, размер: {len(data)} байт") # Добавим лог содержимого (первые 500 символов) logger.debug(f"Содержимое: {data[:500]}") future = asyncio.run_coroutine_threadsafe(update_and_broadcast(json_data), loop) future.add_done_callback(lambda f: logger.debug("Обновление завершено")) except json.JSONDecodeError: logger.error(f"Получены некорректные JSON от {addr}: {data[:200]}") except Exception as e: logger.error(f"Ошибка обработки TCP соединения: {e}") finally: conn.close() async def update_and_broadcast(json_data): payload = await update_store_from_json(json_data) if payload: await broadcast_data(payload) # ------------------------------------------------------------ # Функция отправки данных на TCP-порт сервера # ------------------------------------------------------------ def send_data_via_tcp(data_json): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((TCP_HOST, TCP_PORT)) s.sendall(data_json.encode('utf-8')) logger.info("Данные отправлены на TCP-сервер") except ConnectionRefusedError: logger.error("Не удалось подключиться к TCP-серверу. Возможно, сервер не запущен.") sys.exit(1) # ------------------------------------------------------------ # Вспомогательная функция обновления истории # ------------------------------------------------------------ def update_history(store_dict, key, new_entry, limit=HISTORY_LIMIT): """Обновляет историю для заданного ключа, вставляя новую запись в начало.""" history_key = f"{key}_history" history = store_dict.get(history_key, []) history.insert(0, new_entry) store_dict[history_key] = history[:limit] # Сохраняем также последний прогноз в поле без истории для обратной совместимости store_dict[key] = new_entry # ------------------------------------------------------------ # Главная функция # ------------------------------------------------------------ def main(): parser = argparse.ArgumentParser(description='Агрегатор и распределитель данных анализов') parser.add_argument('--server', action='store_true', help='Запустить сервер (WebSocket + TCP)') args = parser.parse_args() if args.server: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) tcp_thread = threading.Thread(target=tcp_server, args=(loop,), daemon=True) tcp_thread.start() try: loop.run_until_complete(run_websocket_server()) except KeyboardInterrupt: logger.info("Сервер остановлен пользователем") finally: loop.close() else: data = sys.stdin.read() if not data: logger.error("Нет данных в stdin") sys.exit(1) send_data_via_tcp(data) if __name__ == "__main__": main()