portfolio-hub/app/cli/core/utils.py

"""Formatting and execution helpers for the CLI."""

from __future__ import annotations

import json
from dataclasses import asdict, is_dataclass
from typing import Any, Callable, TypeVar

import typer
from pydantic import BaseModel, ValidationError
from rich.console import Console
from rich.panel import Panel
from rich.table import Table

from app.core.errors import FeatureNotReadyError
from app.schemas.portfolio import PortfolioDetailResponse, PortfolioListItem

T = TypeVar("T")


def format_currency(value: float | None) -> str:
    if value is None:
        return "N/A"
    return f"{value:,.0f}"


def to_plain_data(value: Any) -> Any:
    if isinstance(value, BaseModel):
        return value.model_dump(mode="json")
    if is_dataclass(value):
        return asdict(value)
    if isinstance(value, list):
        return [to_plain_data(item) for item in value]
    if isinstance(value, dict):
        return {key: to_plain_data(item) for key, item in value.items()}
    return value


def print_portfolio_table(console: Console, portfolios: list[PortfolioListItem]) -> None:
    table = Table(title="Portfolios", expand=True)
    table.add_column("ID", style="cyan", no_wrap=True)
    table.add_column("Broker")
    table.add_column("Status", no_wrap=True)
    table.add_column("Symbol", no_wrap=True)
    table.add_column("Cash", justify="right")
    table.add_column("Qty", justify="right")
    table.add_column("Price", justify="right")
    table.add_column("Buffer", justify="right")
    table.add_column("Total", justify="right")
    for item in portfolios:
        table.add_row(
            item.portfolio_id,
            item.broker,
            item.status,
            item.symbol,
            format_currency(item.cash_balance),
            f"{item.asset_quantity:g}",
            format_currency(item.price),
            format_currency(item.profit_buffer),
            format_currency(item.total_value),
        )
    console.print(table)


def print_portfolio_detail(console: Console, detail: PortfolioDetailResponse) -> None:
    portfolio = detail.portfolio
    position_value = float(portfolio.asset_quantity) * float(portfolio.price)
    bot_total = float(portfolio.cash_balance) + position_value
    user_total = bot_total + float(portfolio.profit_buffer)
    panel = Panel.fit(
        "\n".join(
            [
                f"id: {portfolio.portfolio_id}",
                f"broker: {portfolio.broker}",
                f"status: {portfolio.status}",
                f"symbol: {portfolio.symbol}",
                f"active_strategy_id: {portfolio.active_strategy_id or 'N/A'}",
                f"avabile_strategies: {', '.join(portfolio.avabile_strategies) if portfolio.avabile_strategies else 'N/A'}",
                f"cash_balance: {format_currency(portfolio.cash_balance)}",
                f"asset_quantity: {portfolio.asset_quantity:g}",
                f"price: {format_currency(portfolio.price)}",
                f"profit_buffer: {format_currency(portfolio.profit_buffer)}",
                f"position_value: {format_currency(position_value)}",
                f"bot_total: {format_currency(bot_total)}",
                f"user_total: {format_currency(user_total)}",
                f"live_quantity: {detail.live_quantity if detail.live_quantity is not None else 'N/A'}",
            ]
        ),
        title="Portfolio Status",
    )
    console.print(panel)

    if detail.strategy_set is not None:
        console.print(
            Panel.fit(
                "\n".join(
                    [
                        f"id: {detail.strategy_set.id}",
                        f"name: {detail.strategy_set.name}",
                        f"config: {detail.strategy_set.config}",
                        f"nick: {detail.strategy_set.nick}",
                    ]
                ),
                title="Strategy Set",
            )
        )

    ledger_table = Table(title="Recent Ledger Entries", expand=True)
    ledger_table.add_column("Recorded At", style="cyan", no_wrap=True)
    ledger_table.add_column("Action")
    ledger_table.add_column("Qty", justify="right")
    ledger_table.add_column("Price", justify="right")
    ledger_table.add_column("Cash After", justify="right")
    ledger_table.add_column("Qty After", justify="right")
    ledger_table.add_column("Buffer After", justify="right")
    for entry in detail.recent_entries:
        ledger_table.add_row(
            entry.recorded_at.isoformat(timespec="seconds"),
            entry.action,
            f"{entry.quantity:g}",
            format_currency(entry.price),
            format_currency(entry.cash_balance_after),
            f"{entry.asset_quantity_after:g}",
            format_currency(entry.profit_buffer_after),
        )
    console.print(ledger_table)


def print_json(console: Console, payload: Any) -> None:
    console.print_json(data=to_plain_data(payload))


def render_output(console: Console, output_format: str, payload: Any, *, empty_message: str | None = None) -> None:
    if output_format == "json":
        print_json(console, payload)
        return

    if isinstance(payload, list):
        if not payload and empty_message:
            console.print(empty_message)
            return
        print_portfolio_table(console, payload)
        return

    if isinstance(payload, PortfolioDetailResponse):
        print_portfolio_detail(console, payload)
        return

    console.print(payload)


def execute_command(console: Console, callback: Callable[[], T]) -> T:
    try:
        return callback()
    except ValidationError as exc:
        console.print(f"Input validation failed: {exc}")
        raise typer.Exit(code=1) from exc
    except FeatureNotReadyError as exc:
        console.print(str(exc))
        raise typer.Exit(code=1) from exc
    except ValueError as exc:
        console.print(str(exc))
        raise typer.Exit(code=1) from exc