src/presentation/cli.py - test - akrilly/Markowitz-portfolio-model

src/presentation/cli.py
185 строк9 KB
akrilly
fix bugs
04 окт 2025, 06:50
04 окт 2025, 06:50214f17f
# src/presentation/cli.py
import argparse
import asyncio
import logging
from datetime import datetime, timedelta


from src.domain.export.export_i import ExportI
from src.domain.export.export_to_excel import ExportToExcel
from src.domain.model.profit_report_portfolio_preferences import build_profit_report_portfolio_preferences
from src.infrastructure.database.stock_database_async_impl import StockDatabaseAsyncImpl
from src.infrastructure.datasource.moex_datasource_async_impl import MoexDataSourceAsyncImpl
from src.infrastructure.repository.index_asset_map_provider_impl import IndexAssetMapProviderImpl
from src.infrastructure.repository.stock_repository_async_impl import StockRepositoryAsyncImpl
from src.presentation.report_controller import ReportController

# --------------------------------------------------------------- #
# Настраиваем базовое логирование (можно переопределить в приложении)
# --------------------------------------------------------------- #
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s – %(message)s",
    # filename=f"log/cli_{datetime.now().strftime('%Y-%m-%d')}.log",
)
_DATE_CONST = "%Y-%m-%d"


def parse_date(date_str: str) -> datetime:
    """Парсит дату из строки в формате YYYY-MM-DD."""
    try:
        return datetime.strptime(date_str, _DATE_CONST)
    except ValueError:
        raise argparse.ArgumentTypeError(f"Неправильный формат даты: {date_str}")


class CLI:
    """Командная строка для запуска расчёта Risk-Return отчёта."""

    def __init__(self, export: ExportI):
        self._export = export

    async def main(self) -> None:
        """Основной метод CLI - теперь полностью асинхронный."""
        parser = argparse.ArgumentParser(
            description="Запуск расчёта Risk‑Return отчёта и экспорт в Excel."
        )
        parser.add_argument(
            "--start",
            type=parse_date,
            default=parse_date("2022-10-03"),
            help="Дата начала периода (YYYY‑MM‑DD)",
        )
        parser.add_argument(
            "--end",
            type=parse_date,
            default=parse_date("2025-10-03"),
            help="Дата окончания периода (YYYY‑MM‑DD)",
        )
        parser.add_argument(
            "--rf",
            type=float,
            default=0.14,
            help="Безрисковая ставка (по умолчанию 3%).",
        )
        # todo if check_profit is True build ProfitReportPortfolioPreferences() and add to controller
        # parser.add_argument(
        #     "--check_profit",
        #     dest="legacy_export",
        #     action="store_false",  # Если флаг указан → export == False
        #     default=True,
        #     help="Не сохранять результат в Excel (полученный список будет выведен в stdout).",
        # )

        # NOTE: Флаг отключает экспорт в Excel.
        parser.add_argument(
            "--no-legacy.export",
            dest="legacy_export",
            action="store_false",  # Если флаг указан → export == False
            default=True,
            help="Не сохранять результат в Excel (полученный список будет выведен в stdout).",
        )

        args = parser.parse_args()

        # Валидация аргументов
        if args.start > args.end:
            parser.error("Дата начала должна быть раньше даты окончания")

        logging.info("Аргументы обработаны")
        logging.info(f"Поиск данных с: {args.start} по: {args.end}")
        # ----------------------------------------------------------- #
        #   Создаём провайдера индексов и активов
        # ----------------------------------------------------------- #
        map_provider = IndexAssetMapProviderImpl(
            risk_free_rate=args.rf,  # Используем переданную ставку
            index_to_assets = {
                'MOEXOG': ['ROSN', 'GAZP', 'NVTK', 'SNGS', 'SIBN', 'TATN', 'TRNFP'],
                'MOEXFN': ['RENI', 'MOEX', 'VTBR', 'SBER', 'AFKS'],
                'MOEXCN': ['MDMG', 'OZPH', 'PRMD', 'APTK', 'EUTR', 'AQUA', 'LENT', 'MVID'],
                'MOEXTN': ['AFLT', 'FESH', 'NKHP', 'NMTP', 'FLOT'],
                "MOEXTL": ["RTKM", "MGTS", "TTLK", "MTSS"],
                'MOEXMM': ['CHMF', 'ALRS', 'PLZL', 'GMKN', 'NLMK', 'ENPG', 'MAGN', 'SELG', 'UGLD', 'VSMO'],
                'MOEXEU': ["FEES", "HYDR", "UPRO", "MSRS", "LSNG"],
                "MOEXIT": ["YDEX", "VKCO", "OZON", "CNRU"],  # IT
                "MOEXRE": ["PIKK", "LSRG", "SMLT"],
            }
        )

        # ----------------------------------------------------------- #
        #   ИСПРАВЛЕНО: Используем контекстные менеджеры правильно
        # ----------------------------------------------------------- #
        try:
            async with StockDatabaseAsyncImpl() as db:
                async with MoexDataSourceAsyncImpl() as source:
                    stock_repo = StockRepositoryAsyncImpl(
                        database=db,
                        datasource=source,
                    )

                    # ИСПРАВЛЕНО: Создаем асинхронный контроллер
                    preferences = build_profit_report_portfolio_preferences(from_date=args.end,
                                                                            till_date=datetime.now(),
                                                                            delta=timedelta(days=int(365 / 12)))
                    preferences = None
                    controller = ReportController(
                        map_provider=map_provider,
                        stock_repo=stock_repo,
                        risk_free_rate=args.rf,
                        is_portfolio_need_profit_report=preferences
                    )

                    # ----------------------------------------------------------- #
                    #   ИСПРАВЛЕНО: Используем await для асинхронного запуска
                    # ----------------------------------------------------------- #
                    logging.info("Запуск расчёта отчёта...")
                    metrics = await controller.run_async(
                        start=args.start,
                        end=args.end,
                        export=args.legacy_export
                    )

                    # Если экспорт отключён – просто выводим таблицу в консоль
                    if not args.legacy_export:
                        try:
                            import pandas as pd
                            df = pd.DataFrame([
                                {
                                    'Актив': m.name,
                                    'Доходность': f"{m.revenue:.4f}",
                                    'Риск': f"{m.risk:.4f}",
                                    'Альфа': f"{m.alpha:.4f}",
                                    'Бета': f"{m.beta:.4f}",
                                    'Дисперсия': f"{m.variance:.6f}",
                                    'Ковариация': f"{m.covariance:.6f}"
                                }
                                for m in metrics
                            ])
                            print("\nРезультаты расчёта:")
                            print("=" * 80)
                            print(df.to_string(index=False))
                        except ImportError:
                            # Если pandas недоступен, выводим простую таблицу
                            print("\nРезультаты расчёта:")
                            print("=" * 80)
                            print(f"{'Актив':<8} {'Доходность':<12} {'Риск':<10} {'Альфа':<10} {'Бета':<8}")
                            print("-" * 80)
                            for m in metrics:
                                print(f"{m.name:<8} {m.revenue:<12.4f} {m.risk:<10.4f} {m.alpha:<10.4f} {m.beta:<8.4f}")

                    logging.info(f"Обработано активов: {len(metrics)}")

        except Exception as exc:
            logging.error("Фатальная ошибка: %s", exc, exc_info=True)
            raise  # Прокидываем исключение дальше


# ИСПРАВЛЕНО: Используем async main функцию
async def main():
    """Точка входа в приложение."""
    cli = CLI(export=ExportToExcel())
    await cli.main()

if __name__ == "__main__":
    # Пример запуска: python -m src.presentation.cli --start 2023-01-01 --end 2023-12-31
    asyncio.run(main())