modules/functions.py - master - malkinte/TA_moex

modules/functions.py
175 строк6 KB
Teresa Malkin
first commit
30 июн 2025, 20:46
30 июн 2025, 20:46e22bcdc
import sys
import time
import sqlite3
import requests
import datetime
from typing import List, Tuple, Union
from dateutil.relativedelta import relativedelta


def get_tickers_data(
        tickers_database_path: str,
        without_complex_product: bool=True
        ) -> List[Tuple]:
    """
    Получает данные о тикерах из базы данных.
    
    Args:
        tickers_database_path (str): Путь к файлу БД SQLite
        
    Returns:
        List[Tuple]: Список кортежей с данными тикеров
    """
    # Подготавливаем запрос
    query = 'SELECT * FROM tickers_preferences'
    conditions = []
    params: List[str] = []

    # Если есть аргументы командной строки
    if without_complex_product:
        conditions.append('"complex product" = -100')
    
    if sys.argv[1:]:
        params = [ticker.upper() for ticker in sys.argv[1:]]
        conditions.append(f'ticker IN ({", ".join(["?"] * len(params))})')

    # Выводим предупреждение если нет аргументов
    else:
        sys.stderr.write('\033[31mВнимание: \033[0m' + 
        'Тикер(-ы) не указан(-ы). Выгрузка всего списка тикеров...\n')
    
    if conditions:
        query += ' WHERE ' + ' AND '.join(conditions)

    # Выполняем запрос
    with sqlite3.connect(tickers_database_path) as conn:
        cursor = conn.cursor()
        cursor.execute(query, params)
        return cursor.fetchall()


def get_request_data(
    ticker: Tuple[str, ...],
    interval: str = '24',
    start_index: int = 0
) -> Tuple[List[List[Union[str, float]]], List[str]]:
    """Получает данные свечей с MOEX API для заданного тикера.

    Args:
        ticker: Кортеж с данными тикера:
            [0] - символ, [6] - дата начала, [7] - дата конца
        interval: Интервал свечей ('24' — дневные)
        start_index: Индекс начала пагинации

    Returns:
        Кортеж (данные_свечей, названия_колонок)

    Raises:
        requests.RequestException: При ошибке сетевого запроса
        ValueError: При неверных входных данных
        KeyError: При неожиданной структуре ответа API
    """
    base_url = (
        "https://iss.moex.com/iss/engines/stock/markets/shares/"
        "securities/{ticker}/candles.json?"
        "from={from_date}&till={to_date}"
        "&interval={interval}&start={start_index}"
    )
    
    ticker_data: List[List[Union[str, float]]] = []
    ticker_columns: List[str] = []
    max_retries = 3
    retry_delay = 20

    try:
        while True:
            try:
                response = requests.get(
                    url=base_url.format(
                        ticker=ticker[1],
                        from_date=ticker[7],
                        to_date=ticker[8],
                        interval=interval,
                        start_index=start_index
                    ),
                    headers={'Authorization': 'Token '},
                    timeout=40
                )
                response.raise_for_status()

                data = response.json()['candles']
                if not data.get('data'):
                    break

                if not ticker_columns:
                    ticker_columns = data['columns']

                ticker_data.extend(data['data'])
                start_index += len(data['data'])

                # Задержка для определенных тикеров
                if ticker[0] in {'63', '126', '189'}:
                    time.sleep(retry_delay)

            except requests.exceptions.RequestException as req_err:
                if max_retries > 0:
                    max_retries -= 1
                    time.sleep(retry_delay)
                    continue
                raise requests.exceptions.RequestException(
                    f"API request failed after retries: {req_err}"
                ) from req_err

    except Exception as e:
        sys.stderr.write(
            f"[{datetime.datetime.now()}] Ошибка: {type(e).__name__}: {e}\n"
        )
        raise

    return ticker_data, ticker_columns


def get_range_of_request(month_shift: int = 6, day_shift: int = 0) -> tuple[str, str]:
    """Возвращает диапазон дат в формате YYYY-MM-DD от начала недели до текущей даты.
    
    Args:
        month_shift: Количество месяцев для смещения начальной даты назад (по умолчанию 6)
        day_shift: Количество дней для смещения конечной даты назад (по умолчанию 0)
        
    Returns:
        Кортеж строк с датами в формате (начальная_дата, конечная_дата)
        
    Examples:
        >>> get_range_of_request(1, 0)
        ('2023-01-30', '2023-02-27')  # Примерный результат
    """
    to_date = datetime.datetime.now() - datetime.timedelta(days=day_shift)
    from_date = to_date - relativedelta(months=month_shift)
    
    return (
        move_to_beginning_of_week(from_date).strftime("%Y-%m-%d"),
        to_date.strftime("%Y-%m-%d")
    )


def move_to_beginning_of_week(date: datetime.datetime) -> datetime.datetime:
    """Перемещает указанную дату на начало недели (понедельник).
    
    Args:
        date (datetime.datetime): Исходная дата/время
        
    Returns:
        datetime.datetime: Дата, соответствующая началу недели (понедельнику)
        
    Examples:
        >>> move_to_beginning_of_week(datetime.datetime(2023, 2, 15))
        datetime.datetime(2023, 2, 13, 0, 0)
    """
    days_to_subtract = date.weekday()  # 0 - понедельник, 1 - вторник и т.д.
    # days_to_subtract = (date.weekday() + 1) % 7  # Изменяет начало на воскресенье (day=6)
    return date - relativedelta(days=days_to_subtract)


def date_exists(date, dates) -> bool:
        '''Проверяет, существует ли дата в тикере в таблице ticks_history'''
        return date in dates