data/streamer.py - main - eugenesic/MetaStream-Flask

data/streamer.py
137 строк6 KB
Evgenii Zagorodskikh
refresh
04 дек 2025, 12:34
04 дек 2025, 12:34e3f1346
# data/streamer.py
import os
import time
import logging
import threading
from datetime import datetime
import pandas as pd
import MetaTrader5 as mt5
import pytz

# Автоопределение твоей локальной зоны
try:
    import tzlocal
    LOCAL_TZ = tzlocal.get_localzone()
    logging.getLogger(__name__).info(f"[STREAM] Твоя зона времени: {LOCAL_TZ}")
except ImportError:
    LOCAL_TZ = pytz.timezone("Europe/Moscow")

from .mt5_data import TIMEFRAMES

logger = logging.getLogger(__name__)


def _ensure_file_exists(filepath: str):
    if os.path.exists(filepath):
        return
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    header = ["time", "open", "high", "low", "close", "tick_volume", "spread", "real_volume"]
    pd.DataFrame(columns=header).to_csv(filepath, index=False)
    logger.info(f"[STREAM] Создан файл: {os.path.basename(filepath)}")


def _append_bar(filepath: str, bar: pd.Series):
    bar_str = bar.copy()
    bar_str["time"] = bar["time"].strftime("%Y-%m-%d %H:%M:%S%z")
    bar_str.to_frame().T.to_csv(filepath, mode="a", header=False, index=False)


def _replace_last_bar(filepath: str, new_bar: pd.Series):
    if not os.path.exists(filepath) or os.path.getsize(filepath) == 0:
        return
    lines = open(filepath, "r", encoding="utf-8").readlines()
    if len(lines) <= 1:  # только заголовок
        return
    # Удаляем последнюю строку
    with open(filepath, "w", encoding="utf-8", newline='') as f:
        f.writelines(lines[:-1])
    _append_bar(filepath, new_bar)


def stream_symbol_all_timeframes(
    symbol: str = "EURUSD",
    base_folder: str = "data/csv",
    check_interval: int = 8
):
    folder = os.path.join(base_folder, symbol.lower())
    last_closed = {}  # последняя сохранённая завершённая свеча по таймфрейму

    logger.info(f"[STREAM] Запуск умного стриминга: {symbol}")

    while True:
        if not mt5.initialize():
            logger.error(f"[MT5] Не удалось подключиться: {mt5.last_error()}")
            time.sleep(check_interval)
            continue

        try:
            for tf_name, tf_value in TIMEFRAMES.items():
                filepath = os.path.join(folder, f"{symbol}_{tf_name}.csv")
                _ensure_file_exists(filepath)

                # КЛЮЧЕВОЕ ИСПРАВЛЕНИЕ: правильно проверяем, что данные пришли
                rates = mt5.copy_rates_from_pos(symbol, tf_value, 0, 2)
                if rates is None or (hasattr(rates, "__len__") and len(rates) == 0):
                    continue

                df = pd.DataFrame(rates)

                # Исправляем имя колонки времени, если нужно
                if df.columns[0] != "time":
                    df.rename(columns={df.columns[0]: "time"}, inplace=True)

                # Конвертируем время в твою локальную зону
                df["time"] = pd.to_datetime(df["time"], unit="s", utc=True).dt.tz_convert(LOCAL_TZ)

                current_bar = df.iloc[-1]
                prev_bar = df.iloc[-2] if len(df) >= 2 else None

                current_time_key = current_bar["time"]

                # 1. Если появилась новая завершённая свеча — сохраняем её навсегда
                if prev_bar is not None:
                    prev_time_key = prev_bar["time"]
                    if tf_name not in last_closed or prev_time_key > last_closed[tf_name]:
                        _append_bar(filepath, prev_bar)
                        last_closed[tf_name] = prev_time_key
                        logger.debug(f"[STREAM] Новая свеча {symbol} {tf_name}: {prev_time_key}")

                # 2. Обновляем текущую (живую) свечу — только если она уже есть в файле
                try:
                    last_line = pd.read_csv(filepath, nrows=1, skiprows=max(0, os.path.getsize(filepath)))
                    if os.path.getsize(filepath) > 100:  # грубо: файл не пустой
                        last_line_time = pd.to_datetime(last_line["time"].iloc[-1], utc=True).tz_convert(LOCAL_TZ)
                        if last_line_time.floor("min") == current_time_key.floor("min"):
                            # Это текущая свеча — обновляем
                            if (abs(last_line["close"].iloc[-1] - current_bar["close"]) > 0.00001 or
                                last_line["high"].iloc[-1] != current_bar["high"] or
                                last_line["low"].iloc[-1] != current_bar["low"]):
                                _replace_last_bar(filepath, current_bar)
                except:
                    pass  # если ошибка чтения — просто пропустим

        except Exception as e:
            logger.error(f"[STREAM] Ошибка в стриминге {symbol}: {e}", exc_info=True)
        finally:
            mt5.shutdown()

        time.sleep(check_interval)


def start_streaming(symbols, base_folder="data/csv", check_interval=8):
    for symbol in symbols:
        t = threading.Thread(
            target=stream_symbol_all_timeframes,
            args=(symbol, base_folder, check_interval),
            daemon=True
        )
        t.start()
        logger.info(f"[STREAM] Поток запущен: {symbol}")

    # Держим основной поток живым
    while True:
        time.sleep(60)


if __name__ == "__main__":
    start_streaming(["EURUSD", "GBPUSD"])