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

data/mt5_data.py
63 строки2 KB
Evgenii Zagorodskikh
Initial commit
02 дек 2025, 17:48
02 дек 2025, 17:483eb2d35
# data/mt5_data.py
import MetaTrader5 as mt5
import pandas as pd
import os
import logging
from datetime import datetime, timedelta

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

TIMEFRAMES = {
    'M1': mt5.TIMEFRAME_M1, 'M5': mt5.TIMEFRAME_M5, 'M15': mt5.TIMEFRAME_M15,
    'M30': mt5.TIMEFRAME_M30, 'H1': mt5.TIMEFRAME_H1, 'H4': mt5.TIMEFRAME_H4,
    'D1': mt5.TIMEFRAME_D1, 'W1': mt5.TIMEFRAME_W1, 'MN1': mt5.TIMEFRAME_MN1,
    # Добавляй любые другие, если нужно
}

def _get_rates(symbol: str, timeframe: int) -> pd.DataFrame | None:
    # Вместо жёсткого 2000 года просим только последние 5 лет
    # Это гарантированно есть почти у всех брокеров даже без прокрутки графика
    rates = mt5.copy_rates_range(
        symbol, timeframe,
        datetime.now() - timedelta(days=365*5),   # ← вот эта строчка спасает всё
        datetime.now()
    )
    if rates is None or len(rates) == 0:
        logger.warning(f"Нет данных: {symbol} {timeframe}")
        return None
    df = pd.DataFrame(rates)
    df['time'] = pd.to_datetime(df['time'], unit='s')
    return df[['time', 'open', 'high', 'low', 'close', 'tick_volume', 'spread', 'real_volume']]

def export_symbol_all_timeframes(
    symbol: str = "EURUSD",
    base_folder: str = "data/csv"
):
    if not mt5.initialize():
        logger.error(f"Не удалось подключиться к MT5: {mt5.last_error()}")
        return

    # Проверка символа
    if mt5.symbol_info(symbol) is None or not mt5.symbol_select(symbol, True):
        logger.error(f"Символ {symbol} недоступен")
        mt5.shutdown()
        return

    # Создаём папку вида data/csv/eurusd
    folder = os.path.join(base_folder, symbol.lower())
    os.makedirs(folder, exist_ok=True)
    logger.info(f"Экспорт {symbol} → {folder}")

    for name, tf in TIMEFRAMES.items():
        logger.info(f"  → {name}")
        df = _get_rates(symbol, tf)
        if df is not None and len(df) > 0:
            path = os.path.join(folder, f"{symbol}_{name}.csv")
            df.to_csv(path, index=False)
            logger.info(f"     сохранено {len(df):,} свечей")
        else:
            logger.warning(f"     пусто")

    mt5.shutdown()
    logger.info(f"Экспорт {symbol} завершён\n")