262 lines
10 KiB
Python
262 lines
10 KiB
Python
"""Portfolio CLI commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import typer
|
|
|
|
from app.cli.core.utils import execute_command, render_output
|
|
from app.schemas.portfolio import (
|
|
BuyRequest,
|
|
CreatePortfolioRequest,
|
|
InvestRequest,
|
|
ResetPortfolioRequest,
|
|
SellRequest,
|
|
TransferRequest,
|
|
UpdatePortfolioRequest,
|
|
)
|
|
from app.services.portfolio_service import PortfolioService
|
|
|
|
|
|
portfolio_app = typer.Typer(help="Portfolio lifecycle and trade commands.")
|
|
|
|
|
|
def _parse_strategy_params(value: str) -> dict[str, Any]:
|
|
if not value.strip():
|
|
return {}
|
|
parsed = json.loads(value)
|
|
if not isinstance(parsed, dict):
|
|
raise ValueError("strategy params must be a JSON object")
|
|
return parsed
|
|
|
|
|
|
def _with_service(ctx: typer.Context, callback):
|
|
app_context = ctx.obj
|
|
repository = app_context.repository_factory(app_context.settings)
|
|
service = PortfolioService(repository)
|
|
try:
|
|
result = execute_command(app_context.console, callback=lambda: callback(service, app_context))
|
|
return result
|
|
finally:
|
|
if hasattr(repository, "close"):
|
|
repository.close()
|
|
|
|
|
|
def _confirm(force: bool, message: str) -> None:
|
|
if force:
|
|
return
|
|
confirmed = typer.confirm(message)
|
|
if not confirmed:
|
|
raise typer.Exit(code=0)
|
|
|
|
|
|
@portfolio_app.command("list")
|
|
def list_portfolios(
|
|
ctx: typer.Context,
|
|
broker: str | None = typer.Option(None, help="Filter by broker."),
|
|
status: str | None = typer.Option(None, help="Filter by status."),
|
|
) -> None:
|
|
detail = _with_service(ctx, lambda service, app_context: service.list_portfolios(broker=broker, status=status))
|
|
render_output(ctx.obj.console, ctx.obj.output_format, detail, empty_message="No portfolios found.")
|
|
|
|
|
|
@portfolio_app.command("status")
|
|
def portfolio_status(
|
|
ctx: typer.Context,
|
|
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
|
recent: int = typer.Option(5, "--recent", min=1, help="Number of recent ledger entries."),
|
|
) -> None:
|
|
detail = _with_service(
|
|
ctx,
|
|
lambda service, app_context: service.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=recent),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
|
|
|
|
|
@portfolio_app.command("create")
|
|
def create_portfolio(
|
|
ctx: typer.Context,
|
|
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
|
symbol: str = typer.Option(..., help="Trading symbol."),
|
|
broker: str = typer.Option("kis", help="Broker code."),
|
|
status: str = typer.Option("enabled", help="Portfolio status."),
|
|
cash_balance: float = typer.Option(0.0, min=0.0, help="Starting cash balance."),
|
|
asset_quantity: float = typer.Option(0.0, min=0.0, help="Starting asset quantity."),
|
|
asset_price: float = typer.Option(0.0, min=0.0, help="Starting asset price."),
|
|
profit_buffer: float = typer.Option(0.0, min=0.0, help="Initial profit buffer."),
|
|
strategy_name: str = typer.Option("target_value", help="Strategy name."),
|
|
strategy_params: str = typer.Option("{}", help="Strategy params as JSON object."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
_confirm(force, f"Create portfolio {portfolio_id} for symbol {symbol}?")
|
|
detail = _with_service(
|
|
ctx,
|
|
lambda service, app_context: service.create_portfolio(
|
|
CreatePortfolioRequest(
|
|
portfolio_id=portfolio_id,
|
|
symbol=symbol,
|
|
broker=broker,
|
|
status=status,
|
|
cash_balance=cash_balance,
|
|
asset_quantity=asset_quantity,
|
|
asset_price=asset_price,
|
|
profit_buffer=profit_buffer,
|
|
strategy_name=strategy_name,
|
|
strategy_params=_parse_strategy_params(strategy_params),
|
|
)
|
|
),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
|
|
|
|
|
@portfolio_app.command("update")
|
|
def update_portfolio(
|
|
ctx: typer.Context,
|
|
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
|
broker: str | None = typer.Option(None, help="Broker code."),
|
|
status: str | None = typer.Option(None, help="Portfolio status."),
|
|
symbol: str | None = typer.Option(None, help="Trading symbol."),
|
|
strategy_name: str | None = typer.Option(None, help="Strategy name."),
|
|
strategy_params: str | None = typer.Option(None, help="Strategy params as JSON object."),
|
|
asset_price: float | None = typer.Option(None, min=0.0, help="Current asset price."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
_confirm(force, f"Update portfolio {portfolio_id}?")
|
|
parsed_strategy_params = None if strategy_params is None else _parse_strategy_params(strategy_params)
|
|
detail = _with_service(
|
|
ctx,
|
|
lambda service, app_context: service.update_portfolio(
|
|
portfolio_id,
|
|
UpdatePortfolioRequest(
|
|
broker=broker,
|
|
status=status,
|
|
symbol=symbol,
|
|
strategy_name=strategy_name,
|
|
strategy_params=parsed_strategy_params,
|
|
asset_price=asset_price,
|
|
),
|
|
),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
|
|
|
|
|
@portfolio_app.command("buy")
|
|
def buy_portfolio(
|
|
ctx: typer.Context,
|
|
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
|
quantity: float = typer.Option(..., min=0.0000001, help="Quantity to buy."),
|
|
price: float = typer.Option(..., min=0.0000001, help="Execution price."),
|
|
fee: float = typer.Option(0.0, min=0.0, help="Fee amount."),
|
|
tax: float = typer.Option(0.0, min=0.0, help="Tax amount."),
|
|
note: str = typer.Option("", help="Ledger note."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
total_cost = quantity * price + fee + tax
|
|
_confirm(force, f"Buy {quantity:g} units of {portfolio_id} at {price:,.0f} (total {total_cost:,.0f})?")
|
|
detail = _with_service(
|
|
ctx,
|
|
lambda service, app_context: service.buy(
|
|
portfolio_id,
|
|
BuyRequest(quantity=quantity, price=price, fee=fee, tax=tax, note=note),
|
|
),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
|
|
|
|
|
@portfolio_app.command("sell")
|
|
def sell_portfolio(
|
|
ctx: typer.Context,
|
|
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
|
quantity: float = typer.Option(..., min=0.0000001, help="Quantity to sell."),
|
|
price: float = typer.Option(..., min=0.0000001, help="Execution price."),
|
|
fee: float = typer.Option(0.0, min=0.0, help="Fee amount."),
|
|
tax: float = typer.Option(0.0, min=0.0, help="Tax amount."),
|
|
note: str = typer.Option("", help="Ledger note."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
net_amount = quantity * price - fee - tax
|
|
_confirm(force, f"Sell {quantity:g} units of {portfolio_id} at {price:,.0f} (net {net_amount:,.0f})?")
|
|
detail = _with_service(
|
|
ctx,
|
|
lambda service, app_context: service.sell(
|
|
portfolio_id,
|
|
SellRequest(quantity=quantity, price=price, fee=fee, tax=tax, note=note),
|
|
),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
|
|
|
|
|
@portfolio_app.command("invest")
|
|
def invest_portfolio(
|
|
ctx: typer.Context,
|
|
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
|
movement_type: str = typer.Option(..., "--type", help="deposit, profit, or withdraw."),
|
|
amount: float = typer.Option(..., min=0.0000001, help="Buffer adjustment amount."),
|
|
note: str = typer.Option("", help="Ledger note."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
_confirm(force, f"Apply {movement_type} of {amount:,.0f} to portfolio {portfolio_id}?")
|
|
detail = _with_service(
|
|
ctx,
|
|
lambda service, app_context: service.invest(
|
|
portfolio_id,
|
|
InvestRequest(movement_type=movement_type, amount=amount, note=note),
|
|
),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
|
|
|
|
|
@portfolio_app.command("transfer")
|
|
def transfer_portfolio(
|
|
ctx: typer.Context,
|
|
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
|
direction: str = typer.Option(..., help="to-buffer or from-buffer."),
|
|
amount: float = typer.Option(..., min=0.0000001, help="Transfer amount."),
|
|
note: str = typer.Option("", help="Ledger note."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
_confirm(force, f"Transfer {amount:,.0f} {direction} for portfolio {portfolio_id}?")
|
|
detail = _with_service(
|
|
ctx,
|
|
lambda service, app_context: service.transfer(
|
|
portfolio_id,
|
|
TransferRequest(direction=direction, amount=amount, note=note),
|
|
),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
|
|
|
|
|
@portfolio_app.command("reset")
|
|
def reset_portfolio(
|
|
ctx: typer.Context,
|
|
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
|
infer_symbol: str | None = typer.Option(None, help="Fallback symbol when portfolio does not exist."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
_confirm(force, f"Reset portfolio {portfolio_id}?")
|
|
detail = _with_service(
|
|
ctx,
|
|
lambda service, app_context: service.reset(portfolio_id, ResetPortfolioRequest(infer_symbol=infer_symbol)),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
|
|
|
|
|
@portfolio_app.command("delete")
|
|
def delete_portfolio(
|
|
ctx: typer.Context,
|
|
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
|
keep_ledger: bool = typer.Option(False, help="Preserve ledger history."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
suffix = " and keep ledger" if keep_ledger else ""
|
|
_confirm(force, f"Delete portfolio {portfolio_id}{suffix}?")
|
|
_with_service(
|
|
ctx,
|
|
lambda service, app_context: service.delete(portfolio_id, keep_ledger=keep_ledger),
|
|
)
|
|
if ctx.obj.output_format == "json":
|
|
render_output(ctx.obj.console, ctx.obj.output_format, {"deleted": portfolio_id, "keep_ledger": keep_ledger})
|
|
else:
|
|
ctx.obj.console.print(f"Deleted portfolio: {portfolio_id}") |