112 lines
4.4 KiB
Python
112 lines
4.4 KiB
Python
"""Strategy catalog service orchestration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from typing import cast
|
|
|
|
from fastapi import Depends
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.domain.errors import StrategyInactiveError, StrategyNotFoundError
|
|
from app.domain.models import StrategySet
|
|
from app.repositories.factory import get_portfolio_repository_from_settings
|
|
from app.repositories.portfolio_repository import PortfolioRepository
|
|
from app.schemas.portfolio import CreateStrategySetRequest, UpdateStrategySetRequest
|
|
|
|
|
|
class StrategyInUseError(ValueError):
|
|
"""The strategy is referenced by at least one portfolio."""
|
|
|
|
|
|
class StrategyConflictError(ValueError):
|
|
"""The requested catalog mutation conflicts with another strategy."""
|
|
|
|
|
|
def get_strategy_set_repository(settings: Settings = Depends(get_settings)) -> PortfolioRepository:
|
|
return get_portfolio_repository_from_settings(settings)
|
|
|
|
|
|
class StrategySetService:
|
|
def __init__(self, repository: PortfolioRepository) -> None:
|
|
self._repository = repository
|
|
|
|
def list(self, *, include_inactive: bool = True) -> list[StrategySet]:
|
|
return self._repository.list_strategy_sets(include_inactive=include_inactive)
|
|
|
|
def get(self, strategy_set_id: str, *, active_only: bool = False) -> StrategySet:
|
|
strategy = self._repository.get_strategy_set(strategy_set_id)
|
|
if strategy is None:
|
|
raise StrategyNotFoundError(f"strategy not found: {strategy_set_id}")
|
|
if active_only and strategy.status != "active":
|
|
raise StrategyInactiveError(f"strategy is inactive: {strategy_set_id}")
|
|
return strategy
|
|
|
|
def require_active(self, strategy_set_id: str) -> StrategySet:
|
|
return self.get(strategy_set_id, active_only=True)
|
|
|
|
def create(self, payload: CreateStrategySetRequest) -> StrategySet:
|
|
try:
|
|
create_strategy = cast(
|
|
Callable[..., StrategySet] | None,
|
|
getattr(self._repository, "create_strategy_set", None),
|
|
)
|
|
if create_strategy is not None:
|
|
strategy = create_strategy(
|
|
payload.name,
|
|
payload.config,
|
|
nick=payload.nick,
|
|
status=payload.status,
|
|
)
|
|
else:
|
|
strategy = self._repository.get_or_create_strategy_set(payload.name, payload.config)
|
|
if payload.nick is not None and payload.nick != strategy.nick:
|
|
strategy = self.update(
|
|
strategy.id,
|
|
UpdateStrategySetRequest(nick=payload.nick),
|
|
)
|
|
if payload.status != strategy.status:
|
|
strategy = self.update(
|
|
strategy.id,
|
|
UpdateStrategySetRequest(status=payload.status),
|
|
)
|
|
return strategy
|
|
except StrategyConflictError:
|
|
raise
|
|
except ValueError as exc:
|
|
raise StrategyConflictError(str(exc)) from exc
|
|
|
|
def update(self, strategy_set_id: str, payload: UpdateStrategySetRequest) -> StrategySet:
|
|
try:
|
|
if payload.nick is None:
|
|
return self._repository.update_strategy_set(
|
|
strategy_set_id,
|
|
name=payload.name,
|
|
config=payload.config,
|
|
status=payload.status,
|
|
)
|
|
return self._repository.update_strategy_set(
|
|
strategy_set_id,
|
|
name=payload.name,
|
|
nick=payload.nick,
|
|
config=payload.config,
|
|
status=payload.status,
|
|
)
|
|
except ValueError as exc:
|
|
detail = str(exc)
|
|
if "찾을 수 없습니다" in detail:
|
|
raise StrategyNotFoundError(f"strategy not found: {strategy_set_id}") from exc
|
|
raise StrategyConflictError(detail) from exc
|
|
|
|
def delete(self, strategy_set_id: str) -> None:
|
|
self.get(strategy_set_id)
|
|
try:
|
|
self._repository.delete_strategy_set(strategy_set_id)
|
|
except ValueError as exc:
|
|
detail = str(exc)
|
|
if "in use" in detail:
|
|
raise StrategyInUseError(detail) from exc
|
|
if "찾을 수 없습니다" in detail:
|
|
raise StrategyNotFoundError(f"strategy not found: {strategy_set_id}") from exc
|
|
raise
|