ml/main.py - llmcode - dimaWebexplorer/investmentAssistant

ml/main.py
596 строк28 KB
Дима
added ml + backend new version
07 ноя 2025, 02:21
07 ноя 2025, 02:219d8b192
import pandas as pd
import requests
from datetime import datetime
import random

try:
    from transformers import AutoTokenizer, AutoModelForSequenceClassification
    import torch
    FINBERT_AVAILABLE = True
except ImportError:
    print("Библиотеки transformers или torch не установлены. Функция анализа новостей будет использовать заглушку.")
    FINBERT_AVAILABLE = False
    # Заглушка для ML модели
    class MockTokenizer:
        def __call__(self, texts, *args, **kwargs):
            # Заглушка для токенизации
            return {'input_ids': [[1]] * len(texts), 'attention_mask': [[1]] * len(texts)}
    class MockModel:
        def __init__(self):
            self.config = type('Config', (), {'num_labels': 3})()
        def __call__(self, **kwargs):
            # Заглушка для модели, возвращающей случайные логиты
            import numpy as np
            logits = torch.tensor([[random.random(), random.random(), random.random()]], dtype=torch.float32)
            return type('Output', (), {'logits': logits})()
    mock_tokenizer = MockTokenizer()
    mock_model = MockModel()

try:
    from tavily import TavilyClient
    TAVILY_AVAILABLE = True
    TAVILY_API_KEY = "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"
    if TAVILY_API_KEY == None:
        print("Предупреждение: API ключ Tavily не установлен. Функция анализа новостей будет использовать заглушку.")
        TAVILY_AVAILABLE = False
    else:
        tavily_client = TavilyClient(api_key=TAVILY_API_KEY)
except ImportError:
    print("Библиотека tavily не установлена. Функция анализа новостей будет использовать заглушку.")
    TAVILY_AVAILABLE = False
except Exception as e:
    print(f"Ошибка инициализации Tavily: {e}. Функция анализа новостей будет использовать заглушку.")
    TAVILY_AVAILABLE = False


def parse_moex_stocks(boards=['TQBR', 'TQTF']):
    """
    Парсит данные об акциях с MOEX ISS API

    Parameters:
    boards: list - список площадок ('TQBR' - акции, 'TQTF' - биржевые фонды)

    Returns:
    pd.DataFrame - объединенный DataFrame с данными
    """

    base_url = "https://iss.moex.com/iss/engines/stock/markets/shares/boards/{board}/securities.json"

    all_data = []

    # Словарь для перевода названий колонок
    column_translation = {
        'SECID': 'Тикер',
        'SHORTNAME': 'Краткое название',
        'SECNAME': 'Полное название',
        'LAST': 'Цена',
        'PREVLEGALCLOSEPRICE': 'Цена закрытия'
    }

    for board in boards:
        try:
            url = base_url.format(board=board)
            print(f"Загружаем данные с площадки: {board}")

            response = requests.get(url)
            response.raise_for_status()
            data = response.json()

            # Извлекаем данные
            securities_data = data['securities']['data']
            securities_columns = data['securities']['columns']
            marketdata_data = data['marketdata']['data']
            marketdata_columns = data['marketdata']['columns']

            # Создаем DataFrame
            df_securities = pd.DataFrame(securities_data, columns=securities_columns)
            df_marketdata = pd.DataFrame(marketdata_data, columns=marketdata_columns)

            # Объединяем данные
            df_combined = pd.merge(df_securities, df_marketdata, on='SECID', suffixes=('', '_market'))

            # Добавляем информацию о площадке
            df_combined['BOARD'] = board

            all_data.append(df_combined)
            print(f"Успешно загружено {len(df_combined)} инструментов с площадки {board}")

        except requests.exceptions.RequestException as e:
            print(f"Ошибка при загрузке данных с площадки {board}: {e}")
            continue

    if not all_data:
        print("Не удалось загрузить данные ни с одной площадки")
        return None

    # Объединяем все данные
    df_final = pd.concat(all_data, ignore_index=True)

    # Переименовываем колонки
    df_final = df_final.rename(columns=column_translation)

    # Выбираем только нужные колонки (те, которые есть в DataFrame)
    available_columns = [col for col in column_translation.values() if col in df_final.columns]

    # Добавляем колонку с площадкой
    available_columns.append('BOARD')

    # Преобразование числовых столбцов
    numeric_cols = ['Цена', 'Цена закрытия']
    for col in numeric_cols:
        if col in df_final.columns:
            df_final[col] = pd.to_numeric(df_final[col], errors='coerce')

    return df_final[available_columns]


def parse_moex_bonds():
    """
    Парсит данные об облигациях с MOEX ISS API.

    Returns:
        pd.DataFrame: DataFrame с данными.
    """
    url = "https://iss.moex.com/iss/engines/stock/markets/bonds/boards/TQOB/securities.json"

    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()

        df_securities = pd.DataFrame(data['securities']['data'], columns=data['securities']['columns'])
        df_marketdata = pd.DataFrame(data['marketdata']['data'], columns=data['marketdata']['columns'])

        df = pd.merge(df_securities, df_marketdata, on='SECID')

        translation_dict = {
            'SECID': 'Тикер',
            'SHORTNAME': 'Краткое название',
            'SECNAME': 'Полное название',
            'LAST': 'Цена',
            'YIELD': 'Доходность к погашению',
            'MODDURATION': 'Модифицированная дюрация',
            'COUPONPERCENT': 'Купон (%)',
            'LOTVALUE': 'Номинал',
            'ACCRUEDINT': 'НКД',
            'COUPONPERIOD': 'Период купона',
            'NEXTCOUPON': 'Следующий купон',
            'MATDATE': 'Дата погашения'
        }

        existing_columns = {}
        for eng, rus in translation_dict.items():
            if eng in df.columns:
                existing_columns[eng] = rus

        df = df.rename(columns=existing_columns)
        result_columns = list(existing_columns.values())

        numeric_cols = ['Цена', 'Доходность к погашению', 'НКД']
        for col in numeric_cols:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors='coerce')

        return df[result_columns]

    except Exception as e:
        print(f"Ошибка при получении данных с MOEX ISS: {e}")
        return None


def get_stocks_above_yield(target_yield, num_stocks=5):
    """
    Находит акции с дивидендной доходностью выше или равной заданной,
    обеспечивая уникальность эмитента (на основе краткого названия).
    В реальности колонка 'Дивидендная доходность' может отсутствовать, поэтому генерируется.

    Args:
        target_yield (float): Минимальная требуемая дивидендная доходность (например, 5.0 для 5.0%).
        num_stocks (int): Количество уникальных акций для поиска (по умолчанию 5).

    Returns:
        pd.DataFrame: DataFrame с найденными акциями.
    """
    df_real = parse_moex_stocks(['TQBR', 'TQTF'])

    if df_real is None or df_real.empty:
        print("Не удалось получить данные об акциях.")
        return pd.DataFrame()

    # Проверяем, есть ли реальная колонка доходности
    if 'Дивидендная доходность' in df_real.columns and not df_real['Дивидендная доходность'].isna().all():
        df = df_real
    else:
        # Генерация доходности, как в вашем примере
        mu = target_yield + 3
        sigma = 2.5
        min_yield = target_yield + 0.01

        generated_yields = []
        for _ in range(len(df_real)):
            val = random.normalvariate(mu, sigma)
            val = max(val, min_yield)
            val = min(val, 30.0)
            generated_yields.append(round(val, 2))

        df_real['Дивидендная доходность'] = generated_yields
        df = df_real

    # Фильтрация по доходности
    df_filtered = df[df['Дивидендная доходность'] >= target_yield].copy()

    if df_filtered.empty:
        print(f"Не найдено акций с дивидендной доходностью >= {target_yield}%.")
        return pd.DataFrame()

    # Сортировка по убыванию доходности
    df_filtered.sort_values(by='Дивидендная доходность', ascending=False, inplace=True)

    # Уникальность эмитента
    def extract_issuer(name):
        if pd.isna(name):
            return "Неизвестный"
        parts = name.split()
        if parts:
            first_part = parts[0]
            issuer = first_part.split(',')[0]
            return issuer
        return "Неизвестный"

    df_filtered['Эмитент'] = df_filtered['Краткое название'].apply(extract_issuer)

    # Убираем дубликаты по эмитенту, оставляя первую (с наивысшей доходностью)
    unique_issuers_df = df_filtered.drop_duplicates(subset=['Эмитент'], keep='first')

    # Выбираем все подходящие (или до num_stocks)
    final_stocks = unique_issuers_df.head(num_stocks)

    # Убираем вспомогательную колонку 'Эмитент' из финального результата
    result_df = final_stocks.drop(columns=['Эмитент'])

    if len(result_df) < num_stocks:
        print(f"Внимание: Найдено только {len(result_df)} уникальных эмитентов с акциями, доходность которых >= {target_yield}%.")

    return result_df


def get_bonds_by_yield(target_yield):
    """
    Находит облигации с доходностью выше или равной заданной.
    В реальности колонка 'Доходность к погашению' может отсутствовать, поэтому генерируется.

    Args:
        target_yield (float): Минимальная требуемая доходность.

    Returns:
        pd.DataFrame: DataFrame с подходящими облигациями.
    """
    df_real = parse_moex_bonds()

    if df_real is None or df_real.empty:
        print("Не удалось получить данные об облигациях.")
        return pd.DataFrame()

    # Проверяем, есть ли реальная колонка доходности
    if 'Доходность к погашению' in df_real.columns and not df_real['Доходность к погашению'].isna().all():
        print("Используются реальные данные о доходности облигаций.")
        df = df_real
    else:
        print("Колонка 'Доходность к погашению' отсутствует или пуста. Генерируем случайные значения.")
        # Генерация доходности для облигаций
        mu = target_yield + 1 # Облигации обычно имеют доходность ближе к целевой
        sigma = 1.5
        min_yield = target_yield

        generated_yields = []
        for _ in range(len(df_real)):
            val = random.normalvariate(mu, sigma)
            val = max(val, min_yield)
            val = min(val, 25.0)
            generated_yields.append(round(val, 2))

        df_real['Доходность к погашения'] = generated_yields
        df = df_real

    # Фильтрация по доходности
    df_filtered = df[df['Доходность к погашению'] >= target_yield].copy()

    if df_filtered.empty:
        print(f"Не найдено облигаций с доходностью >= {target_yield}%.")
        return pd.DataFrame()

    # Сортировка по убыванию доходности
    df_filtered.sort_values(by='Доходность к погашению', ascending=False, inplace=True)

    # Уникальность эмитента (аналогично акциям)
    def extract_issuer(name):
        if pd.isna(name):
            return "Неизвестный"
        parts = name.split()
        if parts:
            first_part = parts[0]
            issuer = first_part.split(',')[0]
            return issuer
        return "Неизвестный"

    df_filtered['Эмитент'] = df_filtered['Краткое название'].apply(extract_issuer)

    # Убираем дубликаты по эмитенту, оставляя первую (с наивысшей доходностью)
    unique_issuers_df = df_filtered.drop_duplicates(subset=['Эмитент'], keep='first')

    # Выбираем все подходящие (не ограничиваем количество, как в предыдущей версии)
    result_df = unique_issuers_df.drop(columns=['Эмитент'])

    return result_df


def calculate_regular_deposit(future_value, annual_rate, years, frequency=12):
    """
    Рассчитывает сумму регулярного внесения (R) по формуле R = FV * i / ((1 + i)^n - 1).

    Args:
        future_value (float): Будущая стоимость (FV).
        annual_rate (float): Годовая ставка доходности (i).
        years (int): Количество лет (n).
        frequency (int): Частота внесений в год.

    Returns:
        float: Сумма регулярного внесения.
    """
    rate_per_period = annual_rate / frequency
    total_periods = years * frequency

    if rate_per_period == 0:
        return future_value / total_periods

    numerator = future_value * rate_per_period
    denominator = (1 + rate_per_period) ** total_periods - 1
    regular_deposit = numerator / denominator

    return regular_deposit


def generate_portfolio_schedule(target_amount, years, annual_rate, initial_deposit=0):
    """
    Генерирует график роста портфеля, используя регулярные внесения.

    Args:
        target_amount (float): Целевая сумма.
        years (int): Срок в годах.
        annual_rate (float): Годовая доходность.
        initial_deposit (float): Начальный депозит (по умолчанию 0).

    Returns:
        pd.DataFrame: Таблица с годами и накопленной суммой.
    """
    schedule_data = []
    current_amount = initial_deposit
    monthly_deposit = calculate_regular_deposit(target_amount, annual_rate, years, 12)

    for year in range(1, years + 1):
        # Применяем сложные проценты за год
        current_amount *= (1 + annual_rate)
        # Добавляем внесения за год
        current_amount += monthly_deposit * 12
        schedule_data.append({'Год': year, 'Сумма накопления': current_amount})

    return pd.DataFrame(schedule_data)


def calculate_positions(portfolio_amount, securities_list, num_issuers=5):
    """
    Рассчитывает количество бумаг для покупки на основе суммы портфеля,
    распределяя средства равномерно между заданным количеством эмитентов.

    Args:
        portfolio_amount (float): Сумма портфеля.
        securities_list (pd.DataFrame): Список бумаг с ценами.
        num_issuers (int): Количество эмитентов для включения в портфель.

    Returns:
        pd.DataFrame: DataFrame с тикерами, ценами и количеством бумаг.
    """
    if securities_list.empty:
        print("Список бумаг пуст.")
        return pd.DataFrame()

    # Выбираем первые num_issuers бумаг из списка (предполагается, что они уже отсортированы и уникальны по эмитенту)
    selected_securities = securities_list.head(num_issuers).copy()

    if selected_securities.empty:
        print("Недостаточно бумаг для формирования портфеля.")
        return pd.DataFrame()

    # Вычисляем долю на каждую бумагу (равномерно распределяем)
    amount_per_security = portfolio_amount / num_issuers

    # Рассчитываем количество бумаг
    selected_securities['Количество бумаг'] = (amount_per_security / selected_securities['Цена']).round().astype(int)

    # Создаем результат
    positions_df = selected_securities[['Тикер', 'Цена', 'Количество бумаг']].copy()
    positions_df.rename(columns={'Тикер': 'Бумаги'}, inplace=True)

    return positions_df

def get_news_text(ticker):
    """
    Получает текст новости для указанной бумаги (тикера).
    Использует Tavily или заглушку.
    """
    if TAVILY_AVAILABLE:
        try:
            query = f"Новости об акции {ticker} за последние 24 часа"
            response = tavily_client.search(query=query, max_results=1)
            if response.get('results') and len(response['results']) > 0:
                summary = response['results'][0].get('content', 'Новости не найдены')
                return summary
            else:
                return "Новости не найдены"
        except Exception as e:
            print(f"Ошибка при запросе к Tavily для {ticker}: {e}")
            return "Новости не найдены"
    else:
        # Заглушка: генерируем случайный текст новости для демонстрации
        # В реальности вы бы использовали реальный текст новости
        news_texts = {
            "Сбер": "Сбербанк объявил о росте чистой прибыли за третий квартал.",
            "Газпром": "Газпром сообщил о снижении объемов добычи газа в связи с техническими проблемами.",
            "РЖД": "РЖД увеличили тарифы на перевозки, что может повлиять на инфляцию.",
            "Альфа-банк, БО-39": "Облигации Альфа-банка показали высокую доходность.",
            "Газпромбанк, 001Р-22": "Газпромбанк успешно разместил новые облигации.",
            "Полюс-золото": "Цены на золото выросли, что положительно влияет на акции Полюса.",
            "default": f"Текст новости о компании {ticker}."
        }
        return news_texts.get(ticker, news_texts["default"])

def analyze_news(positions_df):
    """
    Анализирует новости по каждой бумаге в портфеле с использованием FinBert.

    Args:
        positions_df (pd.DataFrame): DataFrame с позициями портфеля.

    Returns:
        pd.DataFrame: DataFrame с новостным фоном.
    """
    if positions_df.empty:
        print("Нет позиций для анализа новостей.")
        return pd.DataFrame()

    news_results = []
    if FINBERT_AVAILABLE:
        print("Инициализация FinBert для анализа новостей...")
        # Загружаем предварительно обученные токенизатор и модель FinBert
        tokenizer = AutoTokenizer.from_pretrained("ProsusAI/finbert")
        model = AutoModelForSequenceClassification.from_pretrained("ProsusAI/finbert")

        # Определяем метки классов (в FinBert по умолчанию: positive, negative, neutral)
        labels = {0: 'позитив', 1: 'негатив', 2: 'нейтрально'}

        for index, row in positions_df.iterrows():
            ticker = row['Бумаги']
            news_text = get_news_text(ticker)
            print(f"  Обрабатываем новость для {ticker}: {news_text[:50]}...") # Вывод первых 50 символов

            # Токенизируем текст
            inputs = tokenizer(news_text, return_tensors="pt", truncation=True, padding=True, max_length=512)

            # Прогноз с помощью модели
            with torch.no_grad(): # Отключаем вычисление градиентов для ускорения
                outputs = model(**inputs)

            # Получаем логиты и применяем softmax для получения вероятностей
            logits = outputs.logits
            probabilities = torch.nn.functional.softmax(logits, dim=-1)
            predicted_class_id = logits.argmax().item()
            predicted_label = labels.get(predicted_class_id, "неизвестно")
            confidence = probabilities[0][predicted_class_id].item()

            print(f"    -> {predicted_label} (уверенность: {confidence:.2f})")

            news_results.append({
                'Бумаги': ticker,
                'Новостной фон': predicted_label
            })

    else: # Используем заглушку
        print("FinBert недоступен. Используем случайный анализ новостей.")
        for index, row in positions_df.iterrows():
            ticker = row['Бумаги']
            # Получаем текст новости (только для логирования, не используется в заглушке)
            news_text = get_news_text(ticker)
            # Просто случайный выбор
            sentiment = random.choice(["позитив", "негатив", "нейтрально"])
            news_results.append({
                'Бумаги': ticker,
                'Новостной фон': sentiment
            })

    news_df = pd.DataFrame(news_results)
    return news_df


# --- Основные функции для пользователя ---

def main_investment_calculator():
    """Основная функция для расчета инвестиционного портфеля."""
    print("=" * 80)
    print("ИНВЕСТИЦИОННЫЙ КАЛЬКУЛЯТОР")
    print("=" * 80)

    try:
        target_amount = float(input("Введите целевую сумму накопления: "))
        years = int(input("Введите срок накопления (лет): "))
        annual_rate = float(input("Введите требуемую годовую доходность (%): ")) / 100.0
        equity_ratio = float(input("Введите долю акций в портфеле (например, 0.5 для 50%): "))
        num_issuers = int(input("Введите количество эмитентов в портфеле: "))
    except ValueError:
        print("Ошибка ввода. Используем значения по умолчанию.")
        target_amount = 1_000_000
        years = 5
        annual_rate = 0.1609
        equity_ratio = 0.5
        num_issuers = 5

    # Шаг 1: Расчет графика накопления
    print("\n--- Шаг 1: Расчет графика накопления ---")
    schedule_df = generate_portfolio_schedule(target_amount, years, annual_rate)
    print(schedule_df.to_string(index=False))

    # Шаг 2: Получение списка бумаг
    print("\n--- Шаг 2: Получение списка бумаг ---")

    # Получаем акции
    stocks_df = get_stocks_above_yield(annual_rate, num_stocks=100) # Получаем все возможные
    print(f"\nНайдено акций с подходящей доходностью: {len(stocks_df)}")

    # Получаем облигации
    bonds_df = get_bonds_by_yield(annual_rate) # Получаем все возможные
    print(f"\nНайдено облигаций с подходящей доходностью: {len(bonds_df)}")

    # Шаг 3: Расчет позиций
    print("\n--- Шаг 3: Расчет позиций ---")
    # Берем сумму на конец первого года из графика
    first_year_amount = schedule_df.iloc[0]['Сумма накопления'] if not schedule_df.empty else target_amount / years

    # Определяем, сколько денег идет на акции и облигации
    equity_amount = first_year_amount * equity_ratio
    bond_amount = first_year_amount * (1 - equity_ratio)

    print(f"Распределяем {first_year_amount:.2f} руб. (1-й год): {equity_amount:.2f} руб. на акции, {bond_amount:.2f} руб. на облигации.")

    # Рассчитываем позиции для акций
    equity_positions = calculate_positions(equity_amount, stocks_df, num_issuers=max(1, int(num_issuers * equity_ratio)))
    print(f"\nРассчитанные позиции акций (на сумму {equity_amount:.2f}, {len(equity_positions)} эмитентов):")
    if not equity_positions.empty:
        print(equity_positions.to_string(index=False))
    else:
        print("Не удалось сформировать позиции акций.")

    # Рассчитываем позиции для облигаций
    bond_positions = calculate_positions(bond_amount, bonds_df, num_issuers=max(1, num_issuers - len(equity_positions)))
    print(f"\nРассчитанные позиции облигаций (на сумму {bond_amount:.2f}, {len(bond_positions)} эмитентов):")
    if not bond_positions.empty:
        print(bond_positions.to_string(index=False))
    else:
        print("Не удалось сформировать позиции облигаций.")

    # Объединяем позиции
    all_positions = pd.concat([equity_positions, bond_positions], ignore_index=True)

    print(f"\n--- Итоговый портфель (первый год, {len(all_positions)} позиций) ---")
    if not all_positions.empty:
        print(all_positions.to_string(index=False))
    else:
        print("Портфель пуст. Проверьте параметры и доступные бумаги.")

    # Шаг 4: Анализ новостей
    print("\n--- Шаг 4: Анализ новостей ---")
    news_df = analyze_news(all_positions)
    print("\nНовостной фон по бумагам:")
    print(news_df.to_string(index=False))


if __name__ == "__main__":
    main_investment_calculator()