feat(portfolios): persist and validate daemon assignment on create/update

POST /v1/portfolios now forwards a non-null managed_by_daemon_id into
create persistence, and both create and PATCH apply the same daemon
validation contract (UUID format -> 422, missing/foreign -> 404,
inactive -> 409, admin target-owner basis). Daemon validation locks the
daemon row FOR SHARE inside the portfolio write transaction so
validation and write form one atomic decision boundary.

Adds domain errors, repository protocol/signature updates, API/PostgreSQL
tests (POST->GET persistence, no partial state on failure, assignment vs
deactivation race), and contract/data-model documentation.
This commit is contained in:
2026-08-15 12:56:53 +09:00
parent bb3bb5d3f0
commit e2102b6f2b
10 changed files with 861 additions and 7 deletions
+25
View File
@@ -15,6 +15,7 @@ from datetime import date, datetime, timedelta, timezone
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from app.core.auth import AuthContext, Permission, require_read, require_trade_write
from app.domain.errors import DaemonInactiveError, DaemonNotFoundError
from app.domain.models import Portfolio, PortfolioComment
from app.domain.pnl import calculate_current_wac, calculate_realized_pnl
from app.repositories.factory import get_portfolio_repository_from_settings
@@ -67,6 +68,18 @@ def _require_write_token(auth: AuthContext) -> None:
)
def _require_valid_daemon_uuid(daemon_id: str | None) -> None:
if daemon_id is None:
return
try:
uuid.UUID(daemon_id)
except (ValueError, AttributeError, TypeError):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"managed_by_daemon_id는 UUID 형식이어야 합니다: {daemon_id}",
)
def _portfolio_to_item(
portfolio: Portfolio,
*,
@@ -167,6 +180,8 @@ def create_portfolio_v1(
if strategy_set.status != "active":
raise StrategyInactiveError(f"strategy is inactive: {body.strategy_id}")
_require_valid_daemon_uuid(body.managed_by_daemon_id)
portfolio = repo.create_portfolio(
str(uuid.uuid4()),
owner_user_id=owner_user_id,
@@ -180,6 +195,7 @@ def create_portfolio_v1(
profit_buffer=body.profit_buffer,
active_strategy_id=strategy_set.id,
avabile_strategies=[strategy_set.id],
managed_by_daemon_id=body.managed_by_daemon_id,
)
except HTTPException:
raise
@@ -187,6 +203,10 @@ def create_portfolio_v1(
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
except StrategyInactiveError as exc:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
except DaemonNotFoundError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
except DaemonInactiveError as exc:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
finally:
@@ -230,6 +250,7 @@ def update_portfolio_v1(
raise StrategyInactiveError(f"strategy is inactive: {strategy_id}")
fields["active_strategy_id"] = strategy_set.id
fields["avabile_strategies"] = [strategy_set.id]
_require_valid_daemon_uuid(fields.get("managed_by_daemon_id"))
repo.update_portfolio(portfolio_id, **fields)
portfolio = repo.get(portfolio_id)
@@ -239,6 +260,10 @@ def update_portfolio_v1(
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
except StrategyInactiveError as exc:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
except DaemonNotFoundError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
except DaemonInactiveError as exc:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
finally:
+9 -1
View File
@@ -1,4 +1,4 @@
"""Shared strategy-catalog domain errors used across API, service, and persistence layers."""
"""Shared domain errors used across API, service, and persistence layers."""
class StrategyNotFoundError(ValueError):
@@ -7,3 +7,11 @@ class StrategyNotFoundError(ValueError):
class StrategyInactiveError(ValueError):
"""Raised when a new portfolio selection targets an inactive strategy."""
class DaemonNotFoundError(ValueError):
"""Raised when a referenced daemon ID does not exist or is not owned by the portfolio owner."""
class DaemonInactiveError(ValueError):
"""Raised when a new portfolio assignment targets an inactive daemon."""
+1
View File
@@ -162,6 +162,7 @@ class PortfolioRepository(Protocol):
profit_buffer: float,
active_strategy_id: str | None,
avabile_strategies: list[str] | None,
managed_by_daemon_id: str | None = None,
) -> Portfolio: ...
def get_asset_by_id(self, asset_id: str) -> Asset | None: ...
+37 -3
View File
@@ -13,7 +13,7 @@ from typing import Any
import psycopg
from psycopg import sql
from app.domain.errors import StrategyInactiveError, StrategyNotFoundError
from app.domain.errors import DaemonInactiveError, DaemonNotFoundError, StrategyInactiveError, StrategyNotFoundError
from app.domain.models import Asset, AuthToken, BrokerCredential, Daemon, LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, Tag, TokenPrincipal, User
logger = logging.getLogger(__name__)
@@ -159,6 +159,26 @@ class PostgresPortfolioRepository:
if row[0] != "active":
raise StrategyInactiveError(f"strategy is inactive: {strategy_id}")
@staticmethod
def _lock_active_daemon(cur: psycopg.Cursor, daemon_id: str, owner_user_id: str) -> None:
"""Serialize new portfolio assignments with daemon deactivation/owner transfer.
Locks the daemon row FOR SHARE inside the portfolio write transaction and
enforces existence, same-owner, and active status so validation and write
form a single atomic decision boundary.
"""
cur.execute(
"SELECT owner_user_id, is_active FROM daemons WHERE id = %s FOR SHARE",
(daemon_id,),
)
row = cur.fetchone()
if row is None:
raise DaemonNotFoundError(f"daemon not found: {daemon_id}")
if row[0] != owner_user_id:
raise DaemonNotFoundError(f"daemon not found: {daemon_id}")
if not row[1]:
raise DaemonInactiveError(f"daemon is inactive: {daemon_id}")
def ensure_schema(self) -> None:
conn = self._connect()
with conn.cursor() as cur:
@@ -1128,12 +1148,15 @@ class PostgresPortfolioRepository:
profit_buffer: float,
active_strategy_id: str | None,
avabile_strategies: list[str] | None,
managed_by_daemon_id: str | None = None,
) -> Portfolio:
self.ensure_schema()
conn = self._connect()
with conn.transaction(), conn.cursor() as cur:
if active_strategy_id is not None:
self._lock_active_strategy(cur, active_strategy_id)
if managed_by_daemon_id is not None:
self._lock_active_daemon(cur, managed_by_daemon_id, owner_user_id)
cur.execute("SELECT 1 FROM portfolios WHERE portfolio_id = %s", (portfolio_id,))
if cur.fetchone() is not None:
raise ValueError(f"이미 존재하는 포트폴리오입니다: {portfolio_id}")
@@ -1142,8 +1165,8 @@ class PostgresPortfolioRepository:
INSERT INTO portfolios (
portfolio_id, broker, status, name, asset_id,
owner_user_id, cash_balance, asset_quantity, asset_price, profit_buffer,
active_strategy_id, avabile_strategies
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
active_strategy_id, avabile_strategies, managed_by_daemon_id
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
(
portfolio_id,
@@ -1158,6 +1181,7 @@ class PostgresPortfolioRepository:
float(profit_buffer),
active_strategy_id,
avabile_strategies,
managed_by_daemon_id,
),
)
portfolio = self.get(portfolio_id)
@@ -1195,6 +1219,16 @@ class PostgresPortfolioRepository:
strategy_id = fields.get("active_strategy_id")
if isinstance(strategy_id, str):
self._lock_active_strategy(cur, strategy_id)
daemon_id = fields.get("managed_by_daemon_id")
if isinstance(daemon_id, str):
cur.execute(
"SELECT owner_user_id FROM portfolios WHERE portfolio_id = %s",
(portfolio_id,),
)
owner_row = cur.fetchone()
if owner_row is None:
raise ValueError(f"portfolio를 찾을 수 없습니다: {portfolio_id}")
self._lock_active_daemon(cur, daemon_id, owner_row[0])
cur.execute(
f"""
UPDATE portfolios