386 lines
11 KiB
Python
386 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
|
|
import app.cli.commands.health as health_commands
|
|
from typer.testing import CliRunner
|
|
|
|
from app.cli.main import create_app
|
|
from app.core.config import Settings
|
|
from app.domain.models import LedgerEntry, Portfolio, StrategySet
|
|
|
|
|
|
class FakeRepo:
|
|
def __init__(self, portfolio: Portfolio | None = None) -> None:
|
|
self._portfolio = portfolio
|
|
self.entries: list[LedgerEntry] = []
|
|
self.strategy_set = StrategySet(id="set-1", name="target_value", config="{}", nick="target_value:99914b932bd3")
|
|
self.closed = False
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
def ensure_schema(self) -> None:
|
|
return None
|
|
|
|
def ping(self) -> None:
|
|
return None
|
|
|
|
def get(self, portfolio_id: str) -> Portfolio:
|
|
if self._portfolio is None or self._portfolio.portfolio_id != portfolio_id:
|
|
raise ValueError("포트폴리오를 찾을 수 없습니다")
|
|
return self._portfolio
|
|
|
|
def get_or_create(
|
|
self,
|
|
portfolio_id: str,
|
|
symbol: str,
|
|
*,
|
|
broker: str = "kis",
|
|
status: str = "enabled",
|
|
strategy_name: str = "",
|
|
strategy_params: dict | None = None,
|
|
initial_cash: float = 0.0,
|
|
) -> Portfolio:
|
|
if self._portfolio is None:
|
|
self._portfolio = Portfolio(
|
|
portfolio_id=portfolio_id,
|
|
broker=broker,
|
|
status=status,
|
|
symbol=symbol,
|
|
strategy_name=strategy_name,
|
|
strategy_params=strategy_params or {},
|
|
cash_balance=initial_cash,
|
|
)
|
|
return self._portfolio
|
|
|
|
def create_portfolio(
|
|
self,
|
|
portfolio_id: str,
|
|
*,
|
|
broker: str,
|
|
status: str,
|
|
symbol: str,
|
|
strategy_name: str,
|
|
strategy_params: dict,
|
|
cash_balance: float,
|
|
asset_quantity: float,
|
|
asset_price: float,
|
|
profit_buffer: float,
|
|
) -> Portfolio:
|
|
self._portfolio = Portfolio(
|
|
portfolio_id=portfolio_id,
|
|
broker=broker,
|
|
status=status,
|
|
symbol=symbol,
|
|
strategy_name=strategy_name,
|
|
strategy_params=strategy_params,
|
|
cash_balance=cash_balance,
|
|
asset_quantity=asset_quantity,
|
|
price=asset_price,
|
|
profit_buffer=profit_buffer,
|
|
)
|
|
return self._portfolio
|
|
|
|
def update_portfolio(self, portfolio_id: str, **fields: object) -> Portfolio:
|
|
portfolio = self.get(portfolio_id)
|
|
for key, value in fields.items():
|
|
if value is not None:
|
|
setattr(portfolio, key, value)
|
|
return portfolio
|
|
|
|
def list_portfolios(self, *, broker: str | None = None, status: str | None = None) -> list[Portfolio]:
|
|
if self._portfolio is None:
|
|
return []
|
|
if broker is not None and self._portfolio.broker != broker:
|
|
return []
|
|
if status is not None and self._portfolio.status != status:
|
|
return []
|
|
return [self._portfolio]
|
|
|
|
def get_or_create_strategy_set(self, name: str, config: str) -> StrategySet:
|
|
self.strategy_set = StrategySet(id="set-1", name=name, config=config, nick=f"{name}:99914b932bd3")
|
|
return self.strategy_set
|
|
|
|
def get_strategy_set(self, id: str) -> StrategySet | None:
|
|
return self.strategy_set if self.strategy_set.id == id else None
|
|
|
|
def save(
|
|
self,
|
|
portfolio: Portfolio,
|
|
*,
|
|
current_price: float,
|
|
action: str,
|
|
trade_price: float,
|
|
quantity: float,
|
|
fee_amount: float,
|
|
tax_amount: float,
|
|
total_value_after: float,
|
|
strategy_set_id: str = "manual",
|
|
note: str = "",
|
|
) -> None:
|
|
self._portfolio = portfolio
|
|
self.entries.insert(
|
|
0,
|
|
LedgerEntry(
|
|
portfolio_id=portfolio.portfolio_id,
|
|
recorded_at=datetime.now(timezone.utc),
|
|
action=action,
|
|
price=trade_price,
|
|
quantity=quantity,
|
|
fee_amount=fee_amount,
|
|
tax_amount=tax_amount,
|
|
cash_balance_after=portfolio.cash_balance,
|
|
asset_quantity_after=portfolio.asset_quantity,
|
|
total_value_after=total_value_after,
|
|
profit_buffer_after=portfolio.profit_buffer,
|
|
strategy_set_id=strategy_set_id,
|
|
note=note,
|
|
),
|
|
)
|
|
|
|
def list_recent_trades(self, portfolio_id: str, limit: int = 5) -> list[LedgerEntry]:
|
|
return self.entries[:limit]
|
|
|
|
def reset(self, portfolio_id: str) -> Portfolio:
|
|
portfolio = self.get(portfolio_id)
|
|
portfolio.cash_balance = 0.0
|
|
portfolio.asset_quantity = 0.0
|
|
portfolio.price = 0.0
|
|
portfolio.profit_buffer = 0.0
|
|
portfolio.current_strategy_set_id = None
|
|
self.entries = []
|
|
return portfolio
|
|
|
|
def adjust_buffer(self, portfolio_id: str, amount: float, action: str, note: str = "") -> Portfolio:
|
|
portfolio = self.get(portfolio_id)
|
|
if action == "INVEST":
|
|
portfolio.profit_buffer += amount
|
|
else:
|
|
portfolio.profit_buffer -= amount
|
|
return portfolio
|
|
|
|
def transfer_buffer(self, portfolio_id: str, amount: float, direction: str, note: str = "") -> Portfolio:
|
|
portfolio = self.get(portfolio_id)
|
|
if direction == "to-buffer":
|
|
portfolio.cash_balance -= amount
|
|
portfolio.profit_buffer += amount
|
|
else:
|
|
portfolio.cash_balance += amount
|
|
portfolio.profit_buffer -= amount
|
|
return portfolio
|
|
|
|
def delete(self, portfolio_id: str, *, clear_ledger: bool = True) -> None:
|
|
self._portfolio = None
|
|
if clear_ledger:
|
|
self.entries = []
|
|
|
|
|
|
def _settings() -> Settings:
|
|
return Settings(db_enabled=True, allowed_origins_raw="http://localhost:3000")
|
|
|
|
|
|
def build_runner(repo: FakeRepo) -> CliRunner:
|
|
app = create_app(settings_factory=_settings, repository_factory=lambda settings: repo)
|
|
runner = CliRunner()
|
|
runner.app = app
|
|
return runner
|
|
|
|
|
|
def _portfolio() -> Portfolio:
|
|
return Portfolio(
|
|
portfolio_id="005930:tv",
|
|
broker="kis",
|
|
status="enabled",
|
|
symbol="005930",
|
|
strategy_name="target_value",
|
|
strategy_params={"target_asset_ratio": 0.7},
|
|
cash_balance=1_000_000.0,
|
|
asset_quantity=1.0,
|
|
price=70_000.0,
|
|
profit_buffer=50_000.0,
|
|
current_strategy_set_id="set-1",
|
|
)
|
|
|
|
|
|
def test_cli_list_renders_table() -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
runner = build_runner(repo)
|
|
|
|
result = runner.invoke(runner.app, ["portfolio", "list"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "005930:tv" in result.stdout
|
|
assert "Portfolios" in result.stdout
|
|
|
|
|
|
def test_cli_status_supports_json_output() -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
runner = build_runner(repo)
|
|
|
|
result = runner.invoke(runner.app, ["--format", "json", "portfolio", "status", "005930:tv"])
|
|
|
|
assert result.exit_code == 0
|
|
payload = json.loads(result.stdout)
|
|
assert payload["portfolio"]["portfolio_id"] == "005930:tv"
|
|
|
|
|
|
def test_cli_create_buy_and_delete_flow() -> None:
|
|
repo = FakeRepo()
|
|
runner = build_runner(repo)
|
|
|
|
create_result = runner.invoke(
|
|
runner.app,
|
|
[
|
|
"portfolio",
|
|
"create",
|
|
"005930:tv",
|
|
"--symbol",
|
|
"005930",
|
|
"--cash-balance",
|
|
"1000000",
|
|
"--strategy-params",
|
|
'{"target_asset_ratio": 0.5}',
|
|
"--force",
|
|
],
|
|
)
|
|
buy_result = runner.invoke(
|
|
runner.app,
|
|
[
|
|
"portfolio",
|
|
"buy",
|
|
"005930:tv",
|
|
"--quantity",
|
|
"2",
|
|
"--price",
|
|
"50000",
|
|
"--fee",
|
|
"100",
|
|
"--force",
|
|
],
|
|
)
|
|
assert repo.entries[0].action == "BUY"
|
|
delete_result = runner.invoke(runner.app, ["portfolio", "delete", "005930:tv", "--keep-ledger", "--force"])
|
|
|
|
assert create_result.exit_code == 0
|
|
assert buy_result.exit_code == 0
|
|
assert delete_result.exit_code == 0
|
|
assert "Deleted portfolio: 005930:tv" in delete_result.stdout
|
|
|
|
|
|
def test_cli_write_command_prompts_before_create() -> None:
|
|
repo = FakeRepo()
|
|
runner = build_runner(repo)
|
|
|
|
result = runner.invoke(
|
|
runner.app,
|
|
[
|
|
"portfolio",
|
|
"create",
|
|
"005930:tv",
|
|
"--symbol",
|
|
"005930",
|
|
"--cash-balance",
|
|
"1000000",
|
|
],
|
|
input="y\n",
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert "Create portfolio 005930:tv for symbol 005930?" in result.stdout
|
|
|
|
|
|
def test_cli_write_command_can_abort() -> None:
|
|
repo = FakeRepo()
|
|
runner = build_runner(repo)
|
|
|
|
result = runner.invoke(
|
|
runner.app,
|
|
[
|
|
"portfolio",
|
|
"create",
|
|
"005930:tv",
|
|
"--symbol",
|
|
"005930",
|
|
"--cash-balance",
|
|
"1000000",
|
|
],
|
|
input="n\n",
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert repo._portfolio is None
|
|
|
|
|
|
def test_cli_health_check_reports_repository() -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
runner = build_runner(repo)
|
|
|
|
result = runner.invoke(runner.app, ["health-check"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "repository: FakeRepo" in result.stdout
|
|
|
|
|
|
def test_cli_init_db_reports_created(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
runner = build_runner(repo)
|
|
|
|
def fake_initialize_database(*, settings, maintenance_db: str = "postgres") -> dict[str, object]:
|
|
return {
|
|
"database": settings.active_db_name,
|
|
"host": settings.active_db_host,
|
|
"port": settings.active_db_port,
|
|
"created": True,
|
|
"schema_initialized": True,
|
|
"migration_applied": "head",
|
|
}
|
|
|
|
monkeypatch.setattr(health_commands, "initialize_database", fake_initialize_database)
|
|
|
|
result = runner.invoke(runner.app, ["init-db"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "created: True" in result.stdout
|
|
assert "schema_initialized: True" in result.stdout
|
|
assert "migration_applied: head" in result.stdout
|
|
|
|
|
|
def test_cli_init_db_supports_json(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
runner = build_runner(repo)
|
|
|
|
def fake_initialize_database(*, settings, maintenance_db: str = "postgres") -> dict[str, object]:
|
|
return {
|
|
"database": "portfoliohub",
|
|
"host": "localhost",
|
|
"port": 5432,
|
|
"created": False,
|
|
"schema_initialized": True,
|
|
"migration_applied": "head",
|
|
}
|
|
|
|
monkeypatch.setattr(health_commands, "initialize_database", fake_initialize_database)
|
|
|
|
result = runner.invoke(runner.app, ["--format", "json", "init-db"])
|
|
|
|
assert result.exit_code == 0
|
|
payload = json.loads(result.stdout)
|
|
assert payload["database"] == "portfoliohub"
|
|
assert payload["schema_initialized"] is True
|
|
|
|
|
|
def test_cli_migrate_db_reports_head(monkeypatch) -> None:
|
|
repo = FakeRepo(_portfolio())
|
|
runner = build_runner(repo)
|
|
|
|
def fake_bootstrap_or_upgrade_to_head(settings) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(health_commands, "bootstrap_or_upgrade_to_head", fake_bootstrap_or_upgrade_to_head)
|
|
|
|
result = runner.invoke(runner.app, ["migrate-db"])
|
|
|
|
assert result.exit_code == 0
|
|
assert "migration_applied: head" in result.stdout |