belikovme
61b6f03e26
- Добавлен yfinance в requirements.txt - Расширен .gitignore для исключения служебных файлов Python - Исправлен импорт в currency.py для корректной работы с модулями
139 lines
5.6 KiB
Python
139 lines
5.6 KiB
Python
import yfinance as yf
|
||
from datetime import datetime, timedelta
|
||
import logging
|
||
|
||
class ForexDataHandler:
|
||
def __init__(self):
|
||
"""
|
||
Инициализация обработчика валютных данных
|
||
"""
|
||
self.logger = logging.getLogger(__name__)
|
||
# logging.basicConfig(level=logging.INFO)
|
||
self.logger = logging.getLogger(__name__)
|
||
# logging.basicConfig(level=logging.INFO)
|
||
|
||
# Словари с тикерами
|
||
self.russian_indices = {
|
||
'IMOEX': 'IMOEX.ME', # Индекс МосБиржи
|
||
'RTSI': 'RTSI.ME', # Индекс РТС
|
||
'MOEXBC': 'MOEXBC.ME' # Индекс голубых фишек
|
||
}
|
||
self.russian_sectors = {
|
||
'oil_gas': 'MOEXOG.ME', # Нефть и газ
|
||
'finance': 'MOEXFN.ME', # Финансы
|
||
'telecom': 'MOEXTL.ME', # Телекоммуникации
|
||
'metals': 'MOEXMM.ME', # Металлы и добыча
|
||
'consumer': 'MOEXCN.ME', # Потребительский сектор
|
||
'transport': 'MOEXTN.ME' # Транспорт
|
||
}
|
||
self.russian_stocks = {
|
||
'Gazprom': 'GAZP.ME',
|
||
'Sberbank': 'SBER.ME',
|
||
'Lukoil': 'LKOH.ME',
|
||
'Rosneft': 'ROSN.ME',
|
||
'VTB': 'VTBR.ME'
|
||
}
|
||
|
||
def get_forex_rate(self, from_currency: str, to_currency: str) -> dict:
|
||
"""
|
||
Получает текущий курс валют с Yahoo Finance
|
||
|
||
Args:
|
||
from_currency: исходная валюта (например, 'USD')
|
||
to_currency: целевая валюта (например, 'RUB')
|
||
|
||
Returns:
|
||
dict: Словарь с данными о курсе валют
|
||
"""
|
||
try:
|
||
# Формируем тикер для валютной пары
|
||
ticker = f"{from_currency}{to_currency}=X"
|
||
self.logger.info(f"Получение данных для пары {ticker}")
|
||
|
||
currency_pair = yf.Ticker(ticker)
|
||
|
||
# Получаем данные за последний день
|
||
end_date = datetime.now()
|
||
start_date = end_date - timedelta(days=1)
|
||
hist = currency_pair.history(start=start_date, end=end_date)
|
||
|
||
if hist.empty:
|
||
raise ValueError(f"Данные не получены для пары {ticker}")
|
||
|
||
latest_data = hist.iloc[-1]
|
||
|
||
return {
|
||
'close': float(latest_data['Close']),
|
||
'timestamp': latest_data.name.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'open': float(latest_data['Open']),
|
||
'high': float(latest_data['High']),
|
||
'low': float(latest_data['Low']),
|
||
'volume': float(latest_data['Volume']) if 'Volume' in latest_data else None
|
||
}
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"Ошибка при получении данных для пары {from_currency}/{to_currency}: {e}")
|
||
raise
|
||
|
||
def get_market_data(self, ticker: str, period: str = "1d") -> dict:
|
||
"""
|
||
Получает данные по указанному тикеру
|
||
|
||
Args:
|
||
ticker: тикер инструмента
|
||
period: период данных ("1d", "5d", "1mo", "3mo", "6mo", "1y", "2y", "5y", "10y", "ytd", "max")
|
||
"""
|
||
try:
|
||
stock = yf.Ticker(ticker)
|
||
info = stock.info
|
||
hist = stock.history(period=period)
|
||
|
||
if hist.empty:
|
||
raise ValueError(f"Данные не получены для {ticker}")
|
||
|
||
latest_data = hist.iloc[-1]
|
||
|
||
return {
|
||
'close': float(latest_data['Close']),
|
||
'open': float(latest_data['Open']),
|
||
'high': float(latest_data['High']),
|
||
'low': float(latest_data['Low']),
|
||
'volume': float(latest_data['Volume']) if 'Volume' in latest_data else None,
|
||
'timestamp': latest_data.name.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'name': info.get('longName', None),
|
||
'sector': info.get('sector', None),
|
||
'market_cap': info.get('marketCap', None),
|
||
'pe_ratio': info.get('trailingPE', None),
|
||
'dividend_yield': info.get('dividendYield', None),
|
||
'currency': info.get('currency', None)
|
||
}
|
||
except Exception as e:
|
||
self.logger.error(f"Ошибка при получении данных для {ticker}: {e}")
|
||
raise
|
||
|
||
def get_sector_performance(self, sector: str) -> dict:
|
||
"""
|
||
Получает данные по определенному сектору
|
||
|
||
Args:
|
||
sector: ключ сектора из словаря russian_sectors
|
||
"""
|
||
if sector not in self.russian_sectors:
|
||
raise ValueError(f"Неизвестный сектор: {sector}")
|
||
|
||
return self.get_market_data(self.russian_sectors[sector])
|
||
|
||
def get_index_data(self, index: str) -> dict:
|
||
"""
|
||
Получает данные по индексу
|
||
|
||
Args:
|
||
index: ключ индекса из словаря russian_indices
|
||
"""
|
||
if index not in self.russian_indices:
|
||
raise ValueError(f"Неизвестный индекс: {index}")
|
||
|
||
return self.get_market_data(self.russian_indices[index])
|
||
|
||
|