Add ProofLine criteria documentation and Strategy Catalog v1 implementation

- 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)
This commit is contained in:
2026-08-15 11:18:31 +09:00
parent 5fc389dea8
commit 6e02d92282
48 changed files with 2114 additions and 393 deletions
+61 -47
View File
@@ -1,16 +1,28 @@
"""Strategy set service orchestration."""
"""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)
@@ -19,56 +31,58 @@ class StrategySetService:
def __init__(self, repository: PortfolioRepository) -> None:
self._repository = repository
def list(self) -> list[StrategySet]:
"""List all strategy sets."""
return self._repository.list_strategy_sets()
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) -> StrategySet | None:
"""Get a single strategy set by ID."""
return self._repository.get_strategy_set(strategy_set_id)
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:
"""Create or get existing strategy set.
Uses get_or_create_strategy_set for idempotence.
"""
return self._repository.get_or_create_strategy_set(payload.name, payload.config)
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, status=payload.status)
else:
strategy = self._repository.get_or_create_strategy_set(payload.name, payload.config)
if payload.status != strategy.status:
strategy = self.update(
strategy.id,
UpdateStrategySetRequest(status=payload.status),
)
return strategy
def update(self, strategy_set_id: str, payload: UpdateStrategySetRequest) -> StrategySet:
"""Update a strategy set.
Since strategy sets are identified by (name, config) uniqueness,
updating requires creating a new one if name/config changes.
"""
current = self.get(strategy_set_id)
if current is None:
raise ValueError(f"Strategy set {strategy_set_id} not found")
# Use provided values or fall back to current ones
new_name = payload.name if payload.name is not None else current.name
new_config = payload.config if payload.config is not None else current.config
# If nothing changed, return current
if new_name == current.name and new_config == current.config:
return current
# Create or get the new strategy set
new_strategy_set = self._repository.get_or_create_strategy_set(new_name, new_config)
# If the new strategy set is different from current, delete the old one
# (only if it's not in use - delete_strategy_set will raise if in use)
if new_strategy_set.id != strategy_set_id:
try:
self._repository.delete_strategy_set(strategy_set_id)
except ValueError:
# If old strategy set is in use, keep it and return new one
pass
return new_strategy_set
try:
return self._repository.update_strategy_set(
strategy_set_id,
name=payload.name,
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:
"""Delete a strategy set.
Raises ValueError if the strategy set is in use by any portfolio.
"""
self._repository.delete_strategy_set(strategy_set_id)
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