import sys from pathlib import Path import typer from rich.console import Console from rich.table import Table from ai.assistant import AssistantService from cashflow_model import CurrencyConverter, FinancialModel from cli.config import app as config_app from cli.i18n import setup_i18n, t from engine.forecast import ForecastService from engine.scenarios import DEFAULT_SCENARIOS, ScenarioService from sync.excel_sync import ExcelSync 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")) def _load_model() -> FinancialModel: if MODEL_PATH.exists(): return FinancialModel.load(MODEL_PATH) return FinancialModel() def _save_model(model: FinancialModel) -> None: model.save(MODEL_PATH) 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) converter = _get_converter(model) base_curr = model.base_currency service = ForecastService(model) results = service.forecast_cashflow(months) summary = 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) converter = _get_converter(model) base_curr = model.base_currency service = ScenarioService(model) 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 = 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) converter = _get_converter(model) base_curr = model.base_currency service = ScenarioService(model) result = 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) converter = _get_converter(model) base_curr = model.base_currency service = ScenarioService(model) results = 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""" sync = ExcelSync() try: model = sync.import_model(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() sync = ExcelSync() try: sync.export_model(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) converter = _get_converter(model) assistant = AssistantService(model, converter=converter, 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()