""" Модуль работы с базой данных MySQL и комбинированной загрузки данных. Функции: fetch_ohlcv — загрузка OHLCV-данных из MySQL fetch_ohlcv_combined — комбинированная загрузка: MySQL (история) + T-Bank API (текущие свечи) get_available_tickers — получение списка тикеров """ from typing import Any, Optional import mysql.connector import pandas as pd DB_CONFIG: dict[str, Any] = { "host": "nlbotinterface.ru", "port": 3306, "database": "bitcoin_tickers", "user": "bitcoin", "password": "g49020007", "autocommit": True, "charset": "utf8mb4", } # Импорт T-Bank API для свежих данных (lazy — только если нужен) _tbank_available = True try: from src.api.tbank import fetch_candles except ImportError: _tbank_available = False def get_connection() -> Any: """Создание подключения к БД.""" return mysql.connector.connect(**DB_CONFIG) def fetch_ohlcv( ticker: str, tf: str, limit: int = 500, date_from: Optional[str] = None, ) -> pd.DataFrame: """ Загрузка OHLCV-данных из БД MySQL. Args: ticker: тикер инструмента (напр. 'X5'). tf: таймфрейм ('H1', 'D1', 'W1'). limit: максимальное количество свечей. date_from: фильтр по дате в формате 'YYYY.MM.DD' (включая). Returns: DataFrame с колонками [timestamp, Date, Time, Open, High, Low, Close, Volume], отсортированный по возрастанию timestamp. """ table = f"{ticker}_{tf}" conn = get_connection() try: if date_from: query = ( f"SELECT timestamp, Date, Time, Open, High, Low, Close, Volume " f"FROM `{table}` " f"WHERE Date >= %s " f"ORDER BY timestamp DESC " f"LIMIT %s" ) cursor = conn.cursor() cursor.execute(query, (date_from, limit)) rows = cursor.fetchall() columns = ['timestamp', 'Date', 'Time', 'Open', 'High', 'Low', 'Close', 'Volume'] df = pd.DataFrame(rows, columns=columns) else: query = ( f"SELECT timestamp, Date, Time, Open, High, Low, Close, Volume " f"FROM `{table}` " f"ORDER BY timestamp DESC " f"LIMIT %s" ) cursor = conn.cursor() cursor.execute(query, (limit,)) rows = cursor.fetchall() columns = ['timestamp', 'Date', 'Time', 'Open', 'High', 'Low', 'Close', 'Volume'] df = pd.DataFrame(rows, columns=columns) # Сортировка по возрастанию timestamp df = df.sort_values('timestamp').reset_index(drop=True) # Преобразование типов for col in ['Open', 'High', 'Low', 'Close']: df[col] = pd.to_numeric(df[col], errors='coerce') df['Volume'] = pd.to_numeric(df['Volume'], errors='coerce') df['timestamp'] = pd.to_numeric(df['timestamp'], errors='coerce') return df finally: conn.close() def fetch_ohlcv_combined( ticker: str, tf: str, limit: int = 200, date_from: Optional[str] = None, use_tbank: bool = True, ) -> pd.DataFrame: """ Комбинированная загрузка OHLCV: история из MySQL + текущая свеча из T-Bank API. Особенности: - Загружает исторические данные из MySQL (как fetch_ohlcv). - Если use_tbank=True, догружает последние свечи из T-Bank API через GetCandles и заменяет последнюю свечу из БД на свежую (или добавляет, если в БД нет данных за текущий период). - Удаляет дубликаты по timestamp, оставляя приоритет за T-Bank. Args: ticker: тикер инструмента (напр. 'X5'). tf: таймфрейм ('H1', 'D1', 'W1'). limit: количество свечей из БД. date_from: фильтр по дате в формате 'YYYY.MM.DD' (включая). use_tbank: если True, догружает текущую свечу через T-Bank API. Returns: DataFrame с колонками [timestamp, Date, Time, Open, High, Low, Close, Volume], отсортированный по возрастанию timestamp. Если T-Bank данные недоступны — возвращает чистые данные из БД. """ # 1. Загружаем историю из MySQL df = fetch_ohlcv(ticker, tf, limit=limit, date_from=date_from) if not use_tbank or not _tbank_available: return df # 2. Проверяем, является ли инструмент MOEX (ищем FIGI) try: from src.api.tbank import _find_figi _find_figi(ticker) # пробрасывает ValueError, если нет FIGI has_figi = True except (ValueError, ImportError): has_figi = False if not has_figi: return df # 3. Получаем последние свечи из T-Bank API # Запрашиваем 5 свечей, чтобы гарантированно получить текущую tbank_candles_raw = fetch_candles(ticker, tf, count=5) if not tbank_candles_raw: return df # 4. Объединяем: удаляем дубликаты по timestamp из БД, # заменяем их на T-Bank свечи df_tbank = pd.DataFrame(tbank_candles_raw) if df_tbank.empty: return df # Приводим типы как в fetch_ohlcv for col in ['Open', 'High', 'Low', 'Close']: df_tbank[col] = pd.to_numeric(df_tbank[col], errors='coerce') df_tbank['Volume'] = pd.to_numeric(df_tbank['Volume'], errors='coerce') df_tbank['timestamp'] = pd.to_numeric(df_tbank['timestamp'], errors='coerce') # Есть ли уже данные из MySQL? if df.empty: return df_tbank # 5. Определяем минимальный timestamp среди T-Bank свечей, # чтобы удалить пересекающиеся свечи из БД min_tbank_ts = df_tbank['timestamp'].min() # Оставляем в БД только те свечи, что СТАРШЕ (timestamp <) самой старой T-Bank свечи # Это гарантирует, что все пересечения будут разрешены в пользу T-Bank df_old = df[df['timestamp'] < min_tbank_ts].copy() # Объединяем df_combined = pd.concat([df_old, df_tbank], ignore_index=True) # Сортировка по возрастанию timestamp df_combined = df_combined.sort_values('timestamp').reset_index(drop=True) # Удаляем дубликаты timestamp (на случай, если граница неточная) df_combined = df_combined.drop_duplicates(subset='timestamp', keep='last') # 6. Ограничиваем до запрошенного количества if len(df_combined) > limit: df_combined = df_combined.tail(limit).reset_index(drop=True) return df_combined def get_available_tickers() -> list[str]: """Получение списка уникальных тикеров в БД.""" conn = get_connection() try: cursor = conn.cursor() cursor.execute("SHOW TABLES") tables = cursor.fetchall() tickers: set[str] = set() for (table_name,) in tables: table_str: str = str(table_name) if '_' in table_str: parts = table_str.rsplit('_', 1) if len(parts) == 2 and parts[1] in ('H1', 'D1', 'W1'): tickers.add(parts[0]) return sorted(tickers) finally: conn.close()