genichin
6e02d92282
- Add acceptance criteria documentation (AC-0013 through AC-0023) - Add Line-0003 discovery documentation - Implement Strategy Catalog v1 with new database migration - Update portfolio API endpoints and schemas - Refactor strategy and portfolio services - Add strategy management CLI commands - Update domain models with strategy catalog support - Add comprehensive tests for v1 strategies - Update design documentation (data model, interfaces, runtime flows)
354 lines
13 KiB
Python
354 lines
13 KiB
Python
"""Portfolio service orchestration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
|
|
from fastapi import Depends
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.models import LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet
|
|
from app.domain.services import (
|
|
apply_buy,
|
|
apply_delete,
|
|
apply_invest,
|
|
apply_reset,
|
|
apply_sell,
|
|
apply_transfer,
|
|
get_portfolio_status,
|
|
)
|
|
from app.repositories.factory import get_portfolio_repository_from_settings
|
|
from app.repositories.portfolio_repository import PortfolioRepository
|
|
from app.schemas.portfolio import (
|
|
BuyRequest,
|
|
CommentCreateRequest,
|
|
CommentResponse,
|
|
CommentUpdateRequest,
|
|
CreatePortfolioRequest,
|
|
InvestRequest,
|
|
PortfolioDetailResponse,
|
|
PortfolioListItem,
|
|
PortfolioTagMutationRequest,
|
|
PortfolioTagResponse,
|
|
ResetPortfolioRequest,
|
|
SellRequest,
|
|
TransferRequest,
|
|
UpdatePortfolioRequest,
|
|
)
|
|
from app.services.strategy_set_service import StrategyInactiveError, StrategyNotFoundError
|
|
|
|
|
|
def _infer_symbol(portfolio_id: str) -> str:
|
|
parts = [part.strip() for part in portfolio_id.split(":") if part.strip()]
|
|
return parts[0] if parts else ""
|
|
|
|
|
|
def _to_portfolio_summary(portfolio: Portfolio):
|
|
from app.schemas.portfolio import PortfolioSummary
|
|
|
|
return PortfolioSummary(
|
|
portfolio_id=portfolio.portfolio_id,
|
|
broker=portfolio.broker,
|
|
status=portfolio.status,
|
|
name=portfolio.name,
|
|
symbol=portfolio.symbol,
|
|
tags=list(portfolio.tags or []),
|
|
owner_user_id=portfolio.owner_user_id,
|
|
cash_balance=portfolio.cash_balance,
|
|
asset_quantity=portfolio.asset_quantity,
|
|
price=portfolio.price,
|
|
profit_buffer=portfolio.profit_buffer,
|
|
active_strategy_id=portfolio.active_strategy_id,
|
|
strategy_name=portfolio.strategy_name,
|
|
strategy_params=portfolio.strategy_params,
|
|
strategy_revision=portfolio.strategy_revision,
|
|
avabile_strategies=portfolio.avabile_strategies,
|
|
managed_by_daemon_id=portfolio.managed_by_daemon_id,
|
|
created_at=portfolio.created_at,
|
|
updated_at=portfolio.updated_at,
|
|
last_trade_at=portfolio.last_trade_at,
|
|
)
|
|
|
|
|
|
def _parse_strategy_config(config: str) -> dict[str, object]:
|
|
try:
|
|
parsed = json.loads(config) if config else {}
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
|
|
|
|
def _merge_strategy_ids(active_strategy_id: str | None, strategy_ids: list[str] | None) -> list[str] | None:
|
|
normalized = list(strategy_ids or [])
|
|
if active_strategy_id and active_strategy_id not in normalized:
|
|
normalized.append(active_strategy_id)
|
|
return normalized or None
|
|
|
|
|
|
def _to_strategy_set_response(strategy_set: StrategySet | None):
|
|
from app.schemas.portfolio import StrategySetResponse
|
|
|
|
if strategy_set is None:
|
|
return None
|
|
return StrategySetResponse(
|
|
id=strategy_set.id,
|
|
name=strategy_set.name,
|
|
config=strategy_set.config,
|
|
nick=strategy_set.nick,
|
|
status=strategy_set.status,
|
|
revision=strategy_set.revision,
|
|
created_at=strategy_set.created_at,
|
|
updated_at=strategy_set.updated_at,
|
|
)
|
|
|
|
|
|
def _to_ledger_entry_response(entry: LedgerEntry):
|
|
from app.schemas.portfolio import LedgerEntryResponse
|
|
|
|
return LedgerEntryResponse(
|
|
portfolio_id=entry.portfolio_id,
|
|
recorded_at=entry.recorded_at,
|
|
action=entry.action,
|
|
price=entry.price,
|
|
quantity=entry.quantity,
|
|
fee_amount=entry.fee_amount,
|
|
tax_amount=entry.tax_amount,
|
|
cash_balance_after=entry.cash_balance_after,
|
|
asset_quantity_after=entry.asset_quantity_after,
|
|
total_value_after=entry.total_value_after,
|
|
profit_buffer_after=entry.profit_buffer_after,
|
|
strategy_set_id=entry.strategy_set_id,
|
|
note=entry.note,
|
|
)
|
|
|
|
|
|
def _to_comment_response(comment: PortfolioComment) -> CommentResponse:
|
|
return CommentResponse(
|
|
id=comment.id,
|
|
portfolio_id=comment.portfolio_id,
|
|
content=comment.content,
|
|
created_at=comment.created_at,
|
|
updated_at=comment.updated_at,
|
|
)
|
|
|
|
|
|
def _to_tag_response(tag: PortfolioTag) -> PortfolioTagResponse:
|
|
return PortfolioTagResponse(portfolio_id=tag.portfolio_id, tag=tag.tag, created_at=tag.created_at)
|
|
|
|
|
|
def _normalize_tag(value: str) -> str:
|
|
normalized = value.strip()
|
|
if not normalized:
|
|
raise ValueError("tag는 비어 있을 수 없습니다")
|
|
return normalized
|
|
|
|
|
|
def _normalize_tag_list(values: list[str] | None) -> list[str] | None:
|
|
if not values:
|
|
return None
|
|
normalized: list[str] = []
|
|
seen: set[str] = set()
|
|
for value in values:
|
|
item = _normalize_tag(value)
|
|
if item in seen:
|
|
continue
|
|
seen.add(item)
|
|
normalized.append(item)
|
|
return normalized or None
|
|
|
|
|
|
def _to_detail_response(status_result) -> PortfolioDetailResponse:
|
|
return PortfolioDetailResponse(
|
|
portfolio=_to_portfolio_summary(status_result.portfolio),
|
|
recent_entries=[_to_ledger_entry_response(entry) for entry in status_result.recent_entries],
|
|
strategy_set=_to_strategy_set_response(status_result.strategy_set),
|
|
live_quantity=status_result.live_quantity,
|
|
)
|
|
|
|
|
|
def get_portfolio_repository(settings: Settings = Depends(get_settings)) -> PortfolioRepository:
|
|
return get_portfolio_repository_from_settings(settings)
|
|
|
|
|
|
class PortfolioService:
|
|
def __init__(self, repository: PortfolioRepository) -> None:
|
|
self._repository = repository
|
|
|
|
def list_portfolios(
|
|
self,
|
|
*,
|
|
broker: str | None = None,
|
|
status: str | None = None,
|
|
owner_user_id: str | None = None,
|
|
include_tags: list[str] | None = None,
|
|
exclude_tags: list[str] | None = None,
|
|
match_mode: str | None = None,
|
|
) -> list[PortfolioListItem]:
|
|
normalized_mode = None if match_mode is None else match_mode.upper()
|
|
if normalized_mode is not None and normalized_mode not in {"AND", "OR"}:
|
|
raise ValueError("matchMode는 AND 또는 OR 여야 합니다")
|
|
portfolios = self._repository.list_portfolios(
|
|
broker=broker,
|
|
status=status,
|
|
owner_user_id=owner_user_id,
|
|
include_tags=_normalize_tag_list(include_tags),
|
|
exclude_tags=_normalize_tag_list(exclude_tags),
|
|
match_mode=normalized_mode,
|
|
)
|
|
return [
|
|
PortfolioListItem(
|
|
**_to_portfolio_summary(portfolio).model_dump(),
|
|
total_value=float(portfolio.cash_balance) + float(portfolio.asset_quantity) * float(portfolio.price),
|
|
)
|
|
for portfolio in portfolios
|
|
]
|
|
|
|
def get_portfolio_status(self, *, portfolio_id: str, recent_trades: int = 5) -> PortfolioDetailResponse:
|
|
status_result = get_portfolio_status(self._repository, portfolio_id, recent_trades=recent_trades)
|
|
return _to_detail_response(status_result)
|
|
|
|
def create_portfolio(
|
|
self,
|
|
payload: CreatePortfolioRequest,
|
|
*,
|
|
owner_user_id: str = "system",
|
|
) -> PortfolioDetailResponse:
|
|
asset = self._repository.get_asset_by_id(payload.asset_id)
|
|
if asset is None:
|
|
raise ValueError(f"asset not found: {payload.asset_id}")
|
|
if asset.status != "active":
|
|
raise ValueError(f"asset is inactive: {payload.asset_id}")
|
|
|
|
strategy_set = self._repository.get_strategy_set(payload.strategy_id)
|
|
if strategy_set is None:
|
|
raise StrategyNotFoundError(f"strategy not found: {payload.strategy_id}")
|
|
if strategy_set.status != "active":
|
|
raise StrategyInactiveError(f"strategy is inactive: {payload.strategy_id}")
|
|
active_strategy_id = strategy_set.id
|
|
portfolio = self._repository.create_portfolio(
|
|
str(uuid.uuid4()),
|
|
name=payload.name,
|
|
owner_user_id=owner_user_id,
|
|
broker=payload.broker,
|
|
status=payload.status,
|
|
asset_id=payload.asset_id,
|
|
cash_balance=payload.cash_balance,
|
|
asset_quantity=payload.asset_quantity,
|
|
asset_price=payload.asset_price,
|
|
profit_buffer=payload.profit_buffer,
|
|
active_strategy_id=active_strategy_id,
|
|
avabile_strategies=[active_strategy_id],
|
|
)
|
|
return self.get_portfolio_status(portfolio_id=portfolio.portfolio_id, recent_trades=5)
|
|
|
|
def update_portfolio(self, portfolio_id: str, payload: UpdatePortfolioRequest) -> PortfolioDetailResponse:
|
|
fields = payload.model_dump(exclude_none=True)
|
|
current = self._repository.get(portfolio_id)
|
|
|
|
if "strategy_id" in fields:
|
|
strategy_id = str(fields.pop("strategy_id"))
|
|
strategy_set = self._repository.get_strategy_set(strategy_id)
|
|
if strategy_set is None:
|
|
raise StrategyNotFoundError(f"strategy not found: {strategy_id}")
|
|
if strategy_set.status != "active":
|
|
raise StrategyInactiveError(f"strategy is inactive: {strategy_id}")
|
|
fields["active_strategy_id"] = strategy_id
|
|
fields["avabile_strategies"] = _merge_strategy_ids(
|
|
strategy_id,
|
|
current.avabile_strategies,
|
|
)
|
|
|
|
self._repository.update_portfolio(portfolio_id, **fields)
|
|
return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5)
|
|
|
|
def buy(self, portfolio_id: str, payload: BuyRequest) -> PortfolioDetailResponse:
|
|
apply_buy(
|
|
self._repository,
|
|
portfolio_id,
|
|
quantity=payload.quantity,
|
|
price=payload.price,
|
|
fee=payload.fee,
|
|
tax=payload.tax,
|
|
note=payload.note,
|
|
)
|
|
return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5)
|
|
|
|
def sell(self, portfolio_id: str, payload: SellRequest) -> PortfolioDetailResponse:
|
|
apply_sell(
|
|
self._repository,
|
|
portfolio_id,
|
|
quantity=payload.quantity,
|
|
price=payload.price,
|
|
fee=payload.fee,
|
|
tax=payload.tax,
|
|
note=payload.note,
|
|
)
|
|
return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5)
|
|
|
|
def invest(self, portfolio_id: str, payload: InvestRequest) -> PortfolioDetailResponse:
|
|
apply_invest(
|
|
self._repository,
|
|
portfolio_id,
|
|
movement_type=payload.movement_type,
|
|
amount=payload.amount,
|
|
note=payload.note,
|
|
)
|
|
return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5)
|
|
|
|
def transfer(self, portfolio_id: str, payload: TransferRequest) -> PortfolioDetailResponse:
|
|
apply_transfer(
|
|
self._repository,
|
|
portfolio_id,
|
|
direction=payload.direction,
|
|
amount=payload.amount,
|
|
note=payload.note,
|
|
)
|
|
return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5)
|
|
|
|
def reset(self, portfolio_id: str, payload: ResetPortfolioRequest) -> PortfolioDetailResponse:
|
|
infer_symbol = payload.infer_symbol or _infer_symbol(portfolio_id)
|
|
apply_reset(self._repository, portfolio_id, infer_symbol=lambda _: infer_symbol)
|
|
return self.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=5)
|
|
|
|
def delete(self, portfolio_id: str, *, keep_ledger: bool = False) -> None:
|
|
apply_delete(self._repository, portfolio_id, clear_ledger=not keep_ledger)
|
|
|
|
def create_comment(self, portfolio_id: str, payload: CommentCreateRequest) -> CommentResponse:
|
|
self._repository.get(portfolio_id)
|
|
comment = self._repository.create_comment(portfolio_id, content=payload.content)
|
|
return _to_comment_response(comment)
|
|
|
|
def list_comments(self, portfolio_id: str) -> list[CommentResponse]:
|
|
self._repository.get(portfolio_id)
|
|
comments = self._repository.list_comments(portfolio_id)
|
|
return [_to_comment_response(c) for c in comments]
|
|
|
|
def update_comment(self, portfolio_id: str, comment_id: str, payload: CommentUpdateRequest) -> CommentResponse:
|
|
self._repository.get(portfolio_id)
|
|
comment = self._repository.update_comment(portfolio_id, comment_id, content=payload.content)
|
|
return _to_comment_response(comment)
|
|
|
|
def delete_comment(self, portfolio_id: str, comment_id: str) -> None:
|
|
self._repository.get(portfolio_id)
|
|
self._repository.delete_comment(portfolio_id, comment_id)
|
|
|
|
def add_tag(self, portfolio_id: str, payload: PortfolioTagMutationRequest) -> PortfolioTagResponse:
|
|
self._repository.get(portfolio_id)
|
|
tag = self._repository.add_tag(portfolio_id, tag=_normalize_tag(payload.tag))
|
|
return _to_tag_response(tag)
|
|
|
|
def list_tags(self, portfolio_id: str) -> list[PortfolioTagResponse]:
|
|
self._repository.get(portfolio_id)
|
|
return [_to_tag_response(tag) for tag in self._repository.list_tags(portfolio_id)]
|
|
|
|
def remove_tag(self, portfolio_id: str, payload: PortfolioTagMutationRequest) -> None:
|
|
self._repository.get(portfolio_id)
|
|
self._repository.remove_tag(portfolio_id, tag=_normalize_tag(payload.tag))
|
|
|
|
|
|
def get_portfolio_service(
|
|
repository: PortfolioRepository = Depends(get_portfolio_repository),
|
|
) -> PortfolioService:
|
|
return PortfolioService(repository)
|