import sys from pathlib import Path import typer from rich.console import Console from rich.table import Table from infrastructure.ai.assistant import AssistantService from domain import CurrencyConverter, FinancialModel from infrastructure.cli.config import app as config_app from infrastructure.cli.i18n import setup_i18n, t from application.forecast import ForecastService from application.scenarios import DEFAULT_SCENARIOS, ScenarioService from infrastructure.repositories import ExcelRepository, JsonFileRepository try: sys.stdout.reconfigure(encoding="utf-8") except (AttributeError, OSError): pass setup_i18n() app = typer.Typer(name="cf", help=t("app.help")) console = Console() DATA_DIR = Path("data") MODEL_PATH = DATA_DIR / "model.json" CURRENCY_OPTION = typer.Option(None, "--currency", "-c", help=t("cmd.forecast.opt.currency")) _json_repo = JsonFileRepository() _excel_repo = ExcelRepository() def _load_model() -> FinancialModel: if MODEL_PATH.exists(): return _json_repo.load(MODEL_PATH) return FinancialModel() def _save_model(model: FinancialModel) -> None: _json_repo.save(model, MODEL_PATH) def _build_services(model: FinancialModel): """Composition root: собрать граф сервисов с зависимостями.""" from application.forecast import ForecastService from application.scenarios import ScenarioService from infrastructure.ai.assistant import AssistantService converter = CurrencyConverter(model.exchange_rates) forecast_service = ForecastService(model) scenario_service = ScenarioService(model) assistant_service = AssistantService( model, forecast_service=forecast_service, converter=converter ) return forecast_service, scenario_service, assistant_service, converter def _get_converter(model: FinancialModel) -> CurrencyConverter: return CurrencyConverter(model.exchange_rates) def _fmt(amount: float, currency: str, symbol: str) -> str: return f"{symbol}{amount:,.2f}" def _resolve_currency(model: FinancialModel, currency: str | None) -> tuple[str, str]: target = currency or model.base_currency converter = _get_converter(model) symbol = converter.get_symbol(target) return target, symbol def _convert_value( converter: CurrencyConverter, amount: float, from_curr: str, to_curr: str ) -> float: if to_curr == from_curr: return amount return converter.convert(amount, from_curr, to_curr) @app.command() def init() -> None: """Создать пустую финансовую модель""" model = FinancialModel() _save_model(model) console.print(t("cmd.init.ok", path=str(MODEL_PATH))) @app.command() def forecast( months: int = typer.Option(12, "--months", "-m", help=t("cmd.forecast.opt.months")), currency: str | None = CURRENCY_OPTION, ) -> None: """Запустить прогноз денежных потоков""" model = _load_model() target_curr, symbol = _resolve_currency(model, currency) forecast_service, _, _, converter = _build_services(model) base_curr = model.base_currency results = forecast_service.forecast_cashflow(months) summary = forecast_service.summary(months) def _cv(val): return _convert_value(converter, val, base_curr, target_curr) def _f(val): return _fmt(val, target_curr, symbol) if results: table = Table(title=t("cmd.forecast.title", months=months, currency=target_curr)) table.add_column(t("cmd.forecast.col.account"), style="cyan") table.add_column(t("cmd.forecast.col.month"), style="white") table.add_column(t("cmd.forecast.col.balance"), justify="right", style="green") table.add_column(t("cmd.forecast.col.income"), justify="right") table.add_column(t("cmd.forecast.col.expenses"), justify="right") for r in results: table.add_row( r["account"], str(r["month"]), _f(_cv(r["balance"])), _f(_cv(r["income"])), _f(_cv(r["expenses"])), ) console.print(table) cb = _cv(summary["total_balance"]) ci = _cv(summary["total_income"]) ce = _cv(summary["total_expenses"]) console.print(t("cmd.forecast.total", balance=_f(cb), income=_f(ci), expenses=_f(ce))) @app.command() def scenario( name: str = typer.Argument("baseline", help=t("cmd.scenario.arg.name")), months: int = typer.Option(12, "--months", "-m", help=t("cmd.scenario.opt.months")), currency: str | None = CURRENCY_OPTION, ) -> None: """Применить сценарий и показать прогноз""" model = _load_model() target_curr, symbol = _resolve_currency(model, currency) _, scenario_service, _, converter = _build_services(model) base_curr = model.base_currency if name in DEFAULT_SCENARIOS: scenario_obj = DEFAULT_SCENARIOS[name] else: console.print(f"[red]{t('cmd.scenario.err.unknown', name=name)}[/red]") console.print(t("cmd.scenario.err.available", scenarios=", ".join(DEFAULT_SCENARIOS.keys()))) raise typer.Exit(1) result = scenario_service.apply(scenario_obj, months) balance = _convert_value(converter, result["total_balance"], base_curr, target_curr) income = _convert_value(converter, result["total_income"], base_curr, target_curr) expenses = _convert_value(converter, result["total_expenses"], base_curr, target_curr) console.print(f"[bold]{t('cmd.scenario.label')}:[/bold] {result['scenario']}") console.print(f"[dim]{result['scenario_description']}[/dim]") console.print(f"{t('cmd.scenario.label.balance')}: {_fmt(balance, target_curr, symbol)}") console.print(f"{t('cmd.scenario.label.income')}: {_fmt(income, target_curr, symbol)}") console.print(f"{t('cmd.scenario.label.expenses')}: {_fmt(expenses, target_curr, symbol)}") @app.command() def whatif( income_mult: float = typer.Option(1.0, "--income", "-i", help=t("cmd.whatif.opt.income")), expense_mult: float = typer.Option(1.0, "--expense", "-e", help=t("cmd.whatif.opt.expense")), growth_mult: float = typer.Option(1.0, "--growth", "-g", help=t("cmd.whatif.opt.growth")), months: int = typer.Option(12, "--months", "-m", help=t("cmd.whatif.opt.months")), currency: str | None = CURRENCY_OPTION, ) -> None: """What-if анализ с произвольными множителями""" model = _load_model() target_curr, symbol = _resolve_currency(model, currency) _, scenario_service, _, converter = _build_services(model) base_curr = model.base_currency result = scenario_service.what_if(income_mult, expense_mult, growth_mult, months) balance = _convert_value(converter, result["total_balance"], base_curr, target_curr) income = _convert_value(converter, result["total_income"], base_curr, target_curr) expenses = _convert_value(converter, result["total_expenses"], base_curr, target_curr) console.print(f"[bold]{t('cmd.whatif.title')}[/bold]") console.print(t("cmd.whatif.params", income=income_mult, expense=expense_mult, growth=growth_mult)) console.print(f"{t('cmd.scenario.label.balance')}: {_fmt(balance, target_curr, symbol)}") console.print(f"{t('cmd.scenario.label.income')}: {_fmt(income, target_curr, symbol)}") console.print(f"{t('cmd.scenario.label.expenses')}: {_fmt(expenses, target_curr, symbol)}") @app.command() def compare( months: int = typer.Option(12, "--months", "-m", help=t("cmd.compare.opt.months")), currency: str | None = CURRENCY_OPTION, ) -> None: """Сравнить все сценарии""" model = _load_model() target_curr, symbol = _resolve_currency(model, currency) _, scenario_service, _, converter = _build_services(model) base_curr = model.base_currency results = scenario_service.compare(months) table = Table(title=t("cmd.compare.title", currency=target_curr)) table.add_column(t("cmd.compare.col.scenario"), style="cyan") table.add_column(t("cmd.forecast.col.balance"), justify="right") table.add_column(t("cmd.forecast.col.income"), justify="right") table.add_column(t("cmd.forecast.col.expenses"), justify="right") for name, r in results.items(): b = _convert_value(converter, r["total_balance"], base_curr, target_curr) i = _convert_value(converter, r["total_income"], base_curr, target_curr) e = _convert_value(converter, r["total_expenses"], base_curr, target_curr) fb = _fmt(b, target_curr, symbol) fi = _fmt(i, target_curr, symbol) fe = _fmt(e, target_curr, symbol) table.add_row(name, fb, fi, fe) console.print(table) @app.command() def import_xlsx( path: str = typer.Argument(..., help=t("cmd.import.arg.path")), ) -> None: """Импорт данных из Excel""" try: model = _excel_repo.load(path) _save_model(model) console.print(t("cmd.import.ok", accounts=len(model.accounts), transactions=len(model.transactions), recurring=len(model.recurring), assets=len(model.assets), liabilities=len(model.liabilities))) console.print(t("cmd.import.base_currency", currency=model.base_currency)) except Exception as e: console.print(f"[red]{t('cmd.import.err', error=str(e))}[/red]") raise typer.Exit(1) @app.command() def export_xlsx( path: str = typer.Argument("exports/forecast.xlsx", help=t("cmd.export.arg.path")), ) -> None: """Экспорт модели в Excel""" model = _load_model() try: _excel_repo.save(model, path) console.print(t("cmd.export.ok", path=path)) except Exception as e: console.print(f"[red]{t('cmd.export.err', error=str(e))}[/red]") raise typer.Exit(1) @app.command() def analyze( months: int = typer.Option(12, "--months", "-m", help=t("cmd.analyze.opt.months")), currency: str | None = CURRENCY_OPTION, ) -> None: """AI-анализ финансовой модели""" model = _load_model() target_curr, _ = _resolve_currency(model, currency) _, _, assistant, _ = _build_services(model) assistant.display_currency = target_curr result = assistant.analyze(months) console.print(f"[bold]{t('cmd.analyze.title.prompt')}:[/bold]") console.print(result["prompt"][:500] + "...\n") console.print(f"[bold]{t('cmd.analyze.title.summary')}:[/bold]") console.print(t("cmd.analyze.base_currency", currency=model.base_currency)) console.print(f"\n[yellow]{t('cmd.analyze.stub')}[/yellow]") @app.command() def info() -> None: """Сводка всей финансовой модели""" model = _load_model() converter = _get_converter(model) symbol = converter.get_symbol(model.base_currency) console.print(f"[bold]{t('cmd.info.metadata')}[/bold]") console.print(f" {t('cmd.info.base_currency', currency=model.base_currency, symbol=symbol)}") console.print(f" {t('cmd.info.accounts', count=len(model.accounts))}") console.print(f" {t('cmd.info.transactions', count=len(model.transactions))}") console.print(f" {t('cmd.info.recurring', count=len(model.recurring))}") console.print(f" {t('cmd.info.assets', count=len(model.assets))}") console.print(f" {t('cmd.info.liabilities', count=len(model.liabilities))}") console.print(f" {t('cmd.info.rates', count=len(model.exchange_rates))}") if model.accounts: tbl = Table(title=t("table.accounts.title")) tbl.add_column(t("table.accounts.col.name"), style="cyan") tbl.add_column(t("table.accounts.col.currency")) tbl.add_column(t("table.accounts.col.balance"), justify="right") for a in model.accounts: tbl.add_row(a.name, a.currency, f"{a.balance:,.2f}") console.print(tbl) if model.transactions: tbl = Table(title=t("table.transactions.title")) tbl.add_column(t("table.transactions.col.date")) tbl.add_column(t("table.transactions.col.category")) tbl.add_column(t("table.transactions.col.amount"), justify="right") tbl.add_column(t("table.transactions.col.description")) for tx in model.transactions[:10]: tbl.add_row(tx.date, tx.category, f"{tx.amount:,.2f}", tx.description) if len(model.transactions) > 10: console.print(tbl) console.print(t("cmd.info.more", count=len(model.transactions) - 10)) else: console.print(tbl) if model.recurring: tbl = Table(title=t("table.recurring.title")) tbl.add_column(t("table.recurring.col.period")) tbl.add_column(t("table.recurring.col.amount"), justify="right") tbl.add_column(t("table.recurring.col.category")) for r in model.recurring: tbl.add_row(r.frequency, f"{r.amount:,.2f}", r.category) console.print(tbl) if model.assets: tbl = Table(title=t("table.assets.title")) tbl.add_column(t("table.assets.col.name"), style="cyan") tbl.add_column(t("table.assets.col.value"), justify="right") tbl.add_column(t("table.assets.col.growth"), justify="right") for a in model.assets: tbl.add_row(a.name, f"{a.value:,.2f}", f"{a.growth_rate:.1f}%") console.print(tbl) if model.liabilities: tbl = Table(title=t("table.liabilities.title")) tbl.add_column(t("table.liabilities.col.name"), style="cyan") tbl.add_column(t("table.liabilities.col.balance"), justify="right") tbl.add_column(t("table.liabilities.col.rate"), justify="right") tbl.add_column(t("table.liabilities.col.payment"), justify="right") for li in model.liabilities: tbl.add_row(li.name, f"{li.balance:,.2f}", f"{li.interest:.1f}%", f"{li.payment:,.2f}") console.print(tbl) app.add_typer(config_app) if __name__ == "__main__": app()