globalsy_mvp/classes/moex_history.py

import pandas as pd
import aiohttp
import asyncio
from datetime import datetime, timedelta
import sys
import os

# Добавляем родительскую директорию в путь для импорта
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from config import sector_indices


class MOEXHistoricalData:
    """Класс для получения исторических данных с Московской биржи"""
    
    def __init__(self):
        """Инициализация класса для работы с историческими данными MOEX"""
        self.base_url = "https://iss.moex.com/iss"
    
    async def _make_request(self, url: str) -> dict:
        """
        Выполняет асинхронный запрос к API MOEX
        
        Args:
            url: URL для запроса
            
        Returns:
            dict: Ответ от API в формате JSON
        """
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                if response.status == 200:
                    return await response.json()
                else:
                    raise Exception(f"Ошибка при запросе к API MOEX: {response.status}")
    
    async def get_official_sector_index(self, sector: str, start_date: str, end_date: str) -> pd.DataFrame:
        """
        Получает исторические данные по официальному индексу сектора
        
        Args:
            sector: Название сектора
            start_date: Начальная дата в формате 'YYYY-MM-DD'
            end_date: Конечная дата в формате 'YYYY-MM-DD'
            
        Returns:
            pd.DataFrame: Исторические данные по индексу
        """
        # Получаем код индекса для сектора
        index_code = sector_indices.get(sector)
        
        if not index_code:
            raise ValueError(f"Неизвестный сектор: {sector}")
        
        # Формируем URL для запроса
        url = f"{self.base_url}/history/engines/stock/markets/index/securities/{index_code}/candles.json"
        url += f"?from={start_date}&till={end_date}&interval=24"
        
        # Выполняем запрос
        response = await self._make_request(url)
        
        # Преобразуем ответ в DataFrame
        if 'candles' in response and 'data' in response['candles']:
            df = pd.DataFrame(response['candles']['data'], columns=response['candles']['columns'])
            
            # Переименовываем колонки для соответствия формату
            df = df.rename(columns={
                'begin': 'TRADEDATE',
                'open': 'OPEN',
                'high': 'HIGH',
                'low': 'LOW',
                'close': 'CLOSE',
                'volume': 'VOLUME'
            })
            
            # Преобразуем дату в формат datetime
            df['TRADEDATE'] = pd.to_datetime(df['TRADEDATE']).dt.date
            
            return df
        else:
            return pd.DataFrame()
    
    async def get_security_history(self, ticker: str, start_date: str, end_date: str) -> pd.DataFrame:
        """
        Получает исторические данные по ценной бумаге
        
        Args:
            ticker: Тикер ценной бумаги
            start_date: Начальная дата в формате 'YYYY-MM-DD'
            end_date: Конечная дата в формате 'YYYY-MM-DD'
            
        Returns:
            pd.DataFrame: Исторические данные по ценной бумаге
        """
        # Формируем URL для запроса
        url = f"{self.base_url}/history/engines/stock/markets/shares/securities/{ticker}/candles.json"
        url += f"?from={start_date}&till={end_date}&interval=24"
        
        # Выполняем запрос
        response = await self._make_request(url)
        
        # Преобразуем ответ в DataFrame
        if 'candles' in response and 'data' in response['candles']:
            df = pd.DataFrame(response['candles']['data'], columns=response['candles']['columns'])
            
            # Переименовываем колонки для соответствия формату
            df = df.rename(columns={
                'begin': 'TRADEDATE',
                'open': 'OPEN',
                'high': 'HIGH',
                'low': 'LOW',
                'close': 'CLOSE',
                'volume': 'VOLUME'
            })
            
            # Преобразуем дату в формат datetime
            df['TRADEDATE'] = pd.to_datetime(df['TRADEDATE']).dt.date
            
            return df
        else:
            return pd.DataFrame()


# Пример использования
if __name__ == "__main__":
    async def test():
        moex = MOEXHistoricalData()
        
        # Получаем данные по индексу нефти и газа за последний месяц
        end_date = datetime.now()
        start_date = end_date - timedelta(days=30)
        
        df = await moex.get_official_sector_index(
            'oil_gas',
            start_date.strftime('%Y-%m-%d'),
            end_date.strftime('%Y-%m-%d')
        )
        
        print(df.head())
    
    asyncio.run(test())