277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""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.auth import AuthTokenResponse
|
|
from app.schemas.portfolio import CommentResponse, PortfolioDetailResponse, PortfolioListItem
|
|
from app.schemas.user import UserResponse
|
|
|
|
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_user_table(console: Console, users: list[UserResponse]) -> None:
|
|
table = Table(title="Users", expand=True)
|
|
table.add_column("ID", style="cyan", no_wrap=True)
|
|
table.add_column("Username", no_wrap=True)
|
|
table.add_column("Email")
|
|
table.add_column("Created At", no_wrap=True)
|
|
table.add_column("Updated At", no_wrap=True)
|
|
for user in users:
|
|
table.add_row(
|
|
user.id,
|
|
user.username,
|
|
user.email or "N/A",
|
|
user.created_at.isoformat(timespec="seconds") if user.created_at else "N/A",
|
|
user.updated_at.isoformat(timespec="seconds") if user.updated_at else "N/A",
|
|
)
|
|
console.print(table)
|
|
|
|
|
|
def print_user_detail(console: Console, user: UserResponse) -> None:
|
|
console.print(
|
|
Panel.fit(
|
|
"\n".join(
|
|
[
|
|
f"id: {user.id}",
|
|
f"username: {user.username}",
|
|
f"email: {user.email or 'N/A'}",
|
|
f"created_at: {user.created_at.isoformat(timespec='seconds') if user.created_at else 'N/A'}",
|
|
f"updated_at: {user.updated_at.isoformat(timespec='seconds') if user.updated_at else 'N/A'}",
|
|
]
|
|
),
|
|
title="User",
|
|
)
|
|
)
|
|
|
|
|
|
def print_service_token_table(console: Console, tokens: list[AuthTokenResponse]) -> None:
|
|
table = Table(title="Service Tokens", expand=True)
|
|
table.add_column("Name", style="cyan", no_wrap=True)
|
|
table.add_column("Permission", no_wrap=True)
|
|
table.add_column("User ID", no_wrap=True)
|
|
table.add_column("Prefix", no_wrap=True)
|
|
table.add_column("Expires At", no_wrap=True)
|
|
table.add_column("Revoked At", no_wrap=True)
|
|
table.add_column("Created At", no_wrap=True)
|
|
for token in tokens:
|
|
table.add_row(
|
|
token.name,
|
|
token.permission,
|
|
token.user_id,
|
|
token.token_prefix,
|
|
token.expires_at.isoformat(timespec="seconds") if token.expires_at else "never",
|
|
token.revoked_at.isoformat(timespec="seconds") if token.revoked_at else "-",
|
|
token.created_at.isoformat(timespec="seconds") if token.created_at else "N/A",
|
|
)
|
|
console.print(table)
|
|
|
|
|
|
def print_comment_table(console: Console, comments: list[CommentResponse]) -> None:
|
|
table = Table(title="Portfolio Comments", expand=True)
|
|
table.add_column("ID", style="cyan", no_wrap=True)
|
|
table.add_column("Portfolio ID", no_wrap=True)
|
|
table.add_column("Content")
|
|
table.add_column("Created At", no_wrap=True)
|
|
table.add_column("Updated At", no_wrap=True)
|
|
for comment in comments:
|
|
table.add_row(
|
|
comment.id,
|
|
comment.portfolio_id,
|
|
comment.content,
|
|
comment.created_at.isoformat(timespec="seconds") if comment.created_at else "N/A",
|
|
comment.updated_at.isoformat(timespec="seconds") if comment.updated_at else "N/A",
|
|
)
|
|
console.print(table)
|
|
|
|
|
|
def print_comment_detail(console: Console, comment: CommentResponse) -> None:
|
|
console.print(
|
|
Panel.fit(
|
|
"\n".join(
|
|
[
|
|
f"id: {comment.id}",
|
|
f"portfolio_id: {comment.portfolio_id}",
|
|
f"content: {comment.content}",
|
|
f"created_at: {comment.created_at.isoformat(timespec='seconds') if comment.created_at else 'N/A'}",
|
|
f"updated_at: {comment.updated_at.isoformat(timespec='seconds') if comment.updated_at else 'N/A'}",
|
|
]
|
|
),
|
|
title="Comment",
|
|
)
|
|
)
|
|
|
|
|
|
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
|
|
if all(isinstance(item, PortfolioListItem) for item in payload):
|
|
print_portfolio_table(console, payload)
|
|
return
|
|
if all(isinstance(item, UserResponse) for item in payload):
|
|
print_user_table(console, payload)
|
|
return
|
|
if all(isinstance(item, CommentResponse) for item in payload):
|
|
print_comment_table(console, payload)
|
|
return
|
|
if all(isinstance(item, AuthTokenResponse) for item in payload):
|
|
print_service_token_table(console, payload)
|
|
return
|
|
return
|
|
|
|
if isinstance(payload, PortfolioDetailResponse):
|
|
print_portfolio_detail(console, payload)
|
|
return
|
|
|
|
if isinstance(payload, UserResponse):
|
|
print_user_detail(console, payload)
|
|
return
|
|
|
|
if isinstance(payload, CommentResponse):
|
|
print_comment_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
|