"""v1 portfolio CRUD, tag, trade history and PnL endpoints. REQ-005, REQ-025~031: Portfolio CRUD, tags, comments. REQ-039: GET /v1/portfolios/{id}/trades — trade history with date filter. REQ-040: GET /v1/portfolios/{id}/pnl/realized — WAC realized PnL. REQ-043: GET /v1/portfolios/{id}/holdings/cost-basis — current holding WAC cost basis. """ from __future__ import annotations import base64 import binascii import hashlib import hmac import json import logging import uuid from datetime import date, datetime, timedelta, timezone from decimal import Decimal from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, status from app.core.auth import AuthContext, Permission, require_read, require_trade_write from app.core.config import get_settings from app.domain.errors import DaemonInactiveError, DaemonNotFoundError from app.domain.models import CorporateAction, Portfolio, PortfolioComment from app.domain.pnl import calculate_current_wac, calculate_realized_pnl from app.domain.services import PORTFOLIO_NOT_FOUND_MESSAGE from app.repositories.factory import get_portfolio_repository_from_settings from app.repositories.portfolio_repository import LedgerTradeConflictError from app.schemas.corporate_actions import ( CorporateActionRequest, CorporateActionResponse, ) from app.schemas.v1 import ( CorporateActionLedgerHistoryItem, DividendLedgerHistoryItem, DividendManualLedgerResponse, HoldingsCostBasisResponse, LedgerHistoryItem, LedgerHistoryResponse, LedgerTradeRequest, LedgerTradeResponse, ManualLedgerHistoryResponse, ManualLedgerRequest, ManualLedgerResponse, PortfolioCommentItem, PortfolioCommentListResponse, PortfolioCommentRequest, PortfolioCommentUpdateRequest, PortfolioCreateRequest, PortfolioItem, PortfolioListResponse, PortfolioReturnSummaryResponse, PortfolioTagListResponse, PortfolioUpdateRequest, RealizedPnlItem, RealizedPnlResponse, TagApplyRequest, TagApplyResponse, TagItem, TradeHistoryItem, TradeHistoryResponse, ) from app.services.strategy_set_service import ( StrategyInactiveError, StrategyNotFoundError, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/v1/portfolios", tags=["v1"]) def _get_repo(settings=None): from app.core.config import get_settings return get_portfolio_repository_from_settings(settings or get_settings()) def _close_repo(repo) -> None: if callable(getattr(type(repo), "close", None)): repo.close() def _require_write_token(auth: AuthContext) -> None: if auth.token_type in ("service", "client"): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"{auth.token_type} token cannot create or modify portfolios", ) 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, *, return_summary: dict[str, float] | None = None, ) -> PortfolioItem: summary = return_summary or {} return PortfolioItem( portfolio_id=portfolio.portfolio_id, asset_id=portfolio.asset_id, symbol=portfolio.symbol, display_name=portfolio.display_name, owner_user_id=portfolio.owner_user_id, broker=portfolio.broker, status=portfolio.status, name=portfolio.name, cash_balance=portfolio.cash_balance, asset_quantity=portfolio.asset_quantity, price=portfolio.price, profit_buffer=portfolio.profit_buffer, net_contributed_capital=summary.get("net_contributed_capital", 0.0), cumulative_profit_withdrawn=summary.get("cumulative_profit_withdrawn", 0.0), tag_ids=[str(t) for t in (portfolio.tag_ids or [])], managed_by_daemon_id=portfolio.managed_by_daemon_id, created_at=portfolio.created_at, updated_at=portfolio.updated_at, # REQ-055: 전략·시장 메타데이터 additive 필드 active_strategy_id=portfolio.active_strategy_id, strategy_name=portfolio.strategy_name, strategy_nick=portfolio.strategy_nick, strategy_params=portfolio.strategy_params, strategy_revision=portfolio.strategy_revision, markets=portfolio.markets, ) def _comment_to_item(comment: PortfolioComment) -> PortfolioCommentItem: return PortfolioCommentItem( comment_id=comment.id, portfolio_id=comment.portfolio_id, content=comment.content, created_at=comment.created_at, updated_at=comment.updated_at, ) def _get_portfolio_or_404(auth: AuthContext, portfolio_id: str, repo) -> Portfolio: try: portfolio = repo.get(portfolio_id) except ValueError: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"portfolio를 찾을 수 없습니다: {portfolio_id}") if auth.permission != Permission.ADMIN and portfolio.owner_user_id != auth.user_id: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Portfolio ownership mismatch") return portfolio def _get_owner_portfolio_or_404(auth: AuthContext, portfolio_id: str, repo) -> Portfolio: """Fetch a portfolio for APIs that are strictly owner-scoped. Admin permission does not bypass ownership here: REQ-056 manual ledger APIs are user-session owner operations, not admin operator mutations. """ try: portfolio = repo.get(portfolio_id) except ValueError: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"portfolio를 찾을 수 없습니다: {portfolio_id}") if portfolio.owner_user_id != auth.user_id: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Portfolio ownership mismatch") return portfolio # --------------------------------------------------------------------------- # POST /v1/portfolios — REQ-025 # --------------------------------------------------------------------------- @router.post("", response_model=PortfolioItem, status_code=status.HTTP_201_CREATED) def create_portfolio_v1( body: PortfolioCreateRequest, auth: AuthContext = Depends(require_read), ) -> PortfolioItem: _require_write_token(auth) owner_user_id: str = ( body.user_id if (auth.permission == Permission.ADMIN and body.user_id) else auth.user_id ) or "system" repo = _get_repo() try: asset = repo.get_asset_by_id(body.asset_id) if asset is None or asset.status != "active": raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"asset not found or inactive: {body.asset_id}", ) name = body.name or f"{body.broker}-{body.asset_id[:8]}" strategy_set = None if body.strategy_id is not None: strategy_set = repo.get_strategy_set(body.strategy_id) if strategy_set is None: raise StrategyNotFoundError(f"strategy not found: {body.strategy_id}") 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, broker=body.broker, status="enabled", name=name, asset_id=body.asset_id, cash_balance=body.cash_balance, asset_quantity=body.asset_quantity, asset_price=body.asset_price, profit_buffer=body.profit_buffer, active_strategy_id=strategy_set.id if strategy_set is not None else None, avabile_strategies=[strategy_set.id] if strategy_set is not None else None, managed_by_daemon_id=body.managed_by_daemon_id, ) except HTTPException: raise except StrategyNotFoundError as exc: 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: _close_repo(repo) return _portfolio_to_item(portfolio) # --------------------------------------------------------------------------- # PATCH /v1/portfolios/{portfolio_id} — REQ-026 # --------------------------------------------------------------------------- @router.patch("/{portfolio_id}", response_model=PortfolioItem) def update_portfolio_v1( portfolio_id: str, body: PortfolioUpdateRequest, auth: AuthContext = Depends(require_read), ) -> PortfolioItem: _require_write_token(auth) repo = _get_repo() try: portfolio = _get_portfolio_or_404(auth, portfolio_id, repo) if portfolio.status == "disabled": non_status_fields = {k for k, v in body.model_dump(exclude_none=True).items() if k != "status"} if non_status_fields: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="disabled portfolio: only status restoration is allowed", ) fields = body.model_dump(exclude_none=True) if fields: if "strategy_id" in fields: strategy_id = str(fields.pop("strategy_id")) strategy_set = repo.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_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) except HTTPException: raise except StrategyNotFoundError as exc: 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: _close_repo(repo) return _portfolio_to_item(portfolio) # --------------------------------------------------------------------------- # DELETE /v1/portfolios/{portfolio_id} — REQ-027 # --------------------------------------------------------------------------- @router.delete("/{portfolio_id}", response_model=PortfolioItem) def delete_portfolio_v1( portfolio_id: str, auth: AuthContext = Depends(require_read), ) -> PortfolioItem: _require_write_token(auth) repo = _get_repo() try: _get_portfolio_or_404(auth, portfolio_id, repo) portfolio = repo.soft_delete_portfolio(portfolio_id) except HTTPException: raise except ValueError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc finally: _close_repo(repo) return _portfolio_to_item(portfolio) # --------------------------------------------------------------------------- # GET /v1/portfolios — REQ-028 # --------------------------------------------------------------------------- @router.get("", response_model=PortfolioListResponse) def list_portfolios_v1( market: str | None = Query(default=None), include_disabled: bool = Query(default=False), user_id: str | None = Query(default=None), auth: AuthContext = Depends(require_read), ) -> PortfolioListResponse: if auth.token_type == "client": raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="client token cannot list portfolios") status_filter: str | None = None if include_disabled else "enabled" repo = _get_repo() try: if auth.token_type == "service": if auth.daemon_id is None: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="service token without daemon_id cannot list portfolios", ) portfolios = repo.list_portfolios( managed_by_daemon_id=auth.daemon_id, market=market, status=status_filter, ) else: if auth.permission == Permission.ADMIN and user_id is not None: owner = user_id else: owner = auth.user_id portfolios = repo.list_portfolios( owner_user_id=owner, market=market, status=status_filter, ) return_summaries = repo.get_return_summaries([portfolio.portfolio_id for portfolio in portfolios]) except HTTPException: raise finally: _close_repo(repo) items = [ _portfolio_to_item(p, return_summary=return_summaries.get(p.portfolio_id)) for p in portfolios ] return PortfolioListResponse(portfolios=items, total=len(items)) @router.post("/{portfolio_id}/tags", response_model=TagApplyResponse) def apply_portfolio_tag( portfolio_id: str, body: TagApplyRequest, auth: AuthContext = Depends(require_trade_write), ) -> TagApplyResponse: if auth.token_source != "db": raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="DB token required") repo = _get_repo() try: try: portfolio = repo.get(portfolio_id) except ValueError: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"portfolio를 찾을 수 없습니다: {portfolio_id}") is_admin = auth.permission == Permission.ADMIN is_owner = portfolio.owner_user_id == auth.user_id if not is_admin and not is_owner: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Portfolio ownership mismatch") _require_write_token(auth) if portfolio.status == "disabled": raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="disabled portfolio") tag = repo.get_tag_by_id(body.tag_id) if tag is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"태그를 찾을 수 없습니다: {body.tag_id}") if not tag.is_active: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="비활성화된 태그는 적용할 수 없습니다") if not is_admin and not tag.is_admin_tag and tag.created_by_user_id != portfolio.owner_user_id: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="태그 적용 정책을 위반했습니다") repo.append_portfolio_tag_id(portfolio_id, body.tag_id) except HTTPException: raise except ValueError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc finally: _close_repo(repo) return TagApplyResponse(tag_id=body.tag_id, status="applied") @router.delete("/{portfolio_id}/tags/{tag_id}", response_model=TagApplyResponse) def remove_portfolio_tag( portfolio_id: str, tag_id: str, auth: AuthContext = Depends(require_trade_write), ) -> TagApplyResponse: if auth.token_source != "db": raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="DB token required") repo = _get_repo() try: try: portfolio = repo.get(portfolio_id) except ValueError: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"portfolio를 찾을 수 없습니다: {portfolio_id}") is_admin = auth.permission == Permission.ADMIN is_owner = portfolio.owner_user_id == auth.user_id if not is_admin and not is_owner: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Portfolio ownership mismatch") _require_write_token(auth) if portfolio.status == "disabled": raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="disabled portfolio") repo.remove_portfolio_tag_id(portfolio_id, tag_id) except HTTPException: raise except ValueError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc finally: _close_repo(repo) return TagApplyResponse(tag_id=tag_id, status="removed") # --------------------------------------------------------------------------- # GET /v1/portfolios/{portfolio_id}/tags — REQ-030 # --------------------------------------------------------------------------- @router.get("/{portfolio_id}/tags", response_model=PortfolioTagListResponse) def get_portfolio_tags( portfolio_id: str, auth: AuthContext = Depends(require_read), ) -> PortfolioTagListResponse: if auth.token_type in ("service", "client"): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"{auth.token_type} token cannot access portfolio tags", ) repo = _get_repo() try: portfolio = _get_portfolio_or_404(auth, portfolio_id, repo) tags_dict = repo.get_tags_by_ids([str(t) for t in (portfolio.tag_ids or [])]) tag_items = [ TagItem( tag_id=t.tag_id, name=t.name, is_admin_tag=t.is_admin_tag, is_active=t.is_active, created_at=t.created_at, ) for t in tags_dict.values() ] except HTTPException: raise finally: _close_repo(repo) return PortfolioTagListResponse(tags=tag_items, total=len(tag_items)) # --------------------------------------------------------------------------- # POST /v1/portfolios/{portfolio_id}/comments — REQ-031 # --------------------------------------------------------------------------- @router.post("/{portfolio_id}/comments", response_model=PortfolioCommentItem, status_code=status.HTTP_201_CREATED) def create_portfolio_comment( portfolio_id: str, body: PortfolioCommentRequest, auth: AuthContext = Depends(require_read), ) -> PortfolioCommentItem: _require_write_token(auth) repo = _get_repo() try: portfolio = _get_portfolio_or_404(auth, portfolio_id, repo) if portfolio.status == "disabled": raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="disabled portfolio") comment = repo.create_comment(portfolio_id, content=body.content) except HTTPException: raise except ValueError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc finally: _close_repo(repo) return _comment_to_item(comment) # --------------------------------------------------------------------------- # GET /v1/portfolios/{portfolio_id}/comments — REQ-031 # --------------------------------------------------------------------------- @router.get("/{portfolio_id}/comments", response_model=PortfolioCommentListResponse) def list_portfolio_comments( portfolio_id: str, auth: AuthContext = Depends(require_read), ) -> PortfolioCommentListResponse: if auth.token_type in ("service", "client"): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"{auth.token_type} token cannot access portfolio comments", ) repo = _get_repo() try: _get_portfolio_or_404(auth, portfolio_id, repo) comments = repo.list_comments(portfolio_id) except HTTPException: raise finally: _close_repo(repo) items = [_comment_to_item(c) for c in comments] return PortfolioCommentListResponse(comments=items, total=len(items)) # --------------------------------------------------------------------------- # PATCH /v1/portfolios/{portfolio_id}/comments/{comment_id} — REQ-031 # --------------------------------------------------------------------------- @router.patch("/{portfolio_id}/comments/{comment_id}", response_model=PortfolioCommentItem) def update_portfolio_comment( portfolio_id: str, comment_id: str, body: PortfolioCommentUpdateRequest, auth: AuthContext = Depends(require_read), ) -> PortfolioCommentItem: _require_write_token(auth) repo = _get_repo() try: portfolio = _get_portfolio_or_404(auth, portfolio_id, repo) if portfolio.status == "disabled": raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="disabled portfolio") comment = repo.update_comment(portfolio_id, comment_id, content=body.content) except HTTPException: raise except ValueError: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"댓글을 찾을 수 없습니다: {comment_id}") finally: _close_repo(repo) return _comment_to_item(comment) # --------------------------------------------------------------------------- # DELETE /v1/portfolios/{portfolio_id}/comments/{comment_id} — REQ-031 # --------------------------------------------------------------------------- @router.delete("/{portfolio_id}/comments/{comment_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_portfolio_comment( portfolio_id: str, comment_id: str, auth: AuthContext = Depends(require_read), ) -> None: _require_write_token(auth) repo = _get_repo() try: portfolio = _get_portfolio_or_404(auth, portfolio_id, repo) if portfolio.status == "disabled": raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="disabled portfolio") repo.delete_comment(portfolio_id, comment_id) except HTTPException: raise except ValueError: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"댓글을 찾을 수 없습니다: {comment_id}") finally: _close_repo(repo) # --------------------------------------------------------------------------- # POST /v1/portfolios/{portfolio_id}/ledger/trades — REQ-029 # --------------------------------------------------------------------------- def _require_service_token(auth: AuthContext = Depends(require_trade_write)) -> AuthContext: if auth.token_source != "db" or auth.token_type != "service": raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Service token required", ) if auth.daemon_id is None: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="daemon_id not assigned to this token", ) return auth def _economic_payloads_match(existing, body: LedgerTradeRequest) -> bool: """Defence in depth for non-Postgres repository implementations. The Postgres repository performs this comparison under the portfolio and ledger row locks; this check preserves the public contract for test or alternate repository implementations. """ return ( existing.side == body.side and abs(existing.quantity - float(body.quantity)) < 1e-9 and abs(existing.price - float(body.price)) < 1e-9 and abs(existing.fee_amount - float(body.fee_amount)) < 1e-9 and abs(existing.tax_amount - float(body.tax_amount)) < 1e-9 ) @router.post( "/{portfolio_id}/ledger/trades", response_model=LedgerTradeResponse, status_code=status.HTTP_201_CREATED, ) def record_ledger_trade_v1( portfolio_id: str, body: LedgerTradeRequest, response: Response, auth: AuthContext = Depends(_require_service_token), ) -> LedgerTradeResponse: repo = _get_repo() try: try: portfolio = repo.get(portfolio_id) except Exception: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"portfolio를 찾을 수 없습니다: {portfolio_id}", ) if portfolio.status == "disabled": raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="disabled portfolio") if portfolio.managed_by_daemon_id is None: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Asset is not assigned to any daemon", ) if portfolio.managed_by_daemon_id != auth.daemon_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Token daemon does not match asset daemon assignment", ) trade, created = repo.insert_ledger_trade( portfolio_id=portfolio_id, order_id=body.order_id, execution_id=body.execution_id, side=body.side, quantity=float(body.quantity), price=float(body.price), executed_at=body.executed_at, fee_amount=float(body.fee_amount), tax_amount=float(body.tax_amount), execution_venue=body.execution_venue, order_venue=body.order_venue, ) except LedgerTradeConflictError: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Duplicate key with different payload", ) from None except HTTPException: raise finally: _close_repo(repo) if not created: if not _economic_payloads_match(trade, body): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Duplicate key with different payload", ) response.status_code = status.HTTP_200_OK trade_status = "duplicate" else: trade_status = "new" return LedgerTradeResponse( ledger_entry_id=trade.ledger_entry_id, portfolio_id=trade.portfolio_id, order_id=trade.order_id, execution_id=trade.execution_id, side=trade.side, quantity=trade.quantity, price=trade.price, fee_amount=trade.fee_amount, tax_amount=trade.tax_amount, executed_at=trade.executed_at, execution_venue=trade.execution_venue, order_venue=trade.order_venue, recorded_at=trade.recorded_at, cash_balance_after=trade.cash_balance_after, asset_quantity_after=trade.asset_quantity_after, status=trade_status, ) def _manual_entry_to_response(entry) -> ManualLedgerResponse | DividendManualLedgerResponse: action = (entry.action or entry.side or "").upper() if action == "DIVIDEND": amount_decimal = entry.price_decimal if entry.price_decimal is not None else Decimal(str(entry.price)) tax_decimal = entry.tax_amount_decimal if entry.tax_amount_decimal is not None else Decimal(str(entry.tax_amount)) return DividendManualLedgerResponse( ledger_entry_id=entry.ledger_entry_id, portfolio_id=entry.portfolio_id, action=action, amount=str(amount_decimal), quantity=entry.quantity, price=entry.price, fee=entry.fee_amount, tax=str(tax_decimal), executed_at=entry.executed_at, note=entry.note, recorded_at=entry.recorded_at, cash_balance_after=entry.cash_balance_after, asset_quantity_after=entry.asset_quantity_after, profit_buffer_after=entry.profit_buffer_after, total_value_after=entry.total_value_after, source=entry.source or "manual", target_ledger_entry_id=entry.target_ledger_entry_id, confirmation_evidence=entry.confirmation_evidence, ) return ManualLedgerResponse( ledger_entry_id=entry.ledger_entry_id, portfolio_id=entry.portfolio_id, action=action, amount=float(entry.price) if action in {"INVEST", "BUFFER_OUT", "BUFFER_IN", "WITHDRAW", "PROFIT"} else None, quantity=entry.quantity, price=entry.price, fee=entry.fee_amount, tax=float(entry.tax_amount), executed_at=entry.executed_at, note=entry.note, recorded_at=entry.recorded_at, cash_balance_after=entry.cash_balance_after, asset_quantity_after=entry.asset_quantity_after, profit_buffer_after=entry.profit_buffer_after, total_value_after=entry.total_value_after, source=entry.source or "manual", target_ledger_entry_id=entry.target_ledger_entry_id, confirmation_evidence=entry.confirmation_evidence, ) def _manual_error_code(message: str) -> str: if "POTENTIAL_DUPLICATE_DIVIDEND" in message: return "POTENTIAL_DUPLICATE_DIVIDEND" if "LEDGER_ENTRY_NOT_FOUND" in message or "원장 항목을 찾을 수 없습니다" in message: return "LEDGER_ENTRY_NOT_FOUND" if "INSUFFICIENT_CASH" in message or "현금" in message: return "INSUFFICIENT_CASH" if "INSUFFICIENT_QUANTITY" in message or "수량" in message: return "INSUFFICIENT_QUANTITY" if "INSUFFICIENT_BUFFER" in message or "버퍼" in message: return "INSUFFICIENT_BUFFER" if "LEDGER_ENTRY_ALREADY_REVERSED" in message: return "LEDGER_ENTRY_ALREADY_REVERSED" return "MANUAL_LEDGER_CONFLICT" def _extract_duplicate_dividend_entry_id(message: str) -> int | None: prefix = "POTENTIAL_DUPLICATE_DIVIDEND:" if not message.startswith(prefix): return None try: return int(message.removeprefix(prefix)) except ValueError: return None def _manual_error_message(code: str) -> str: if code == "LEDGER_ENTRY_NOT_FOUND": return "Target ledger entry not found" if code == "POTENTIAL_DUPLICATE_DIVIDEND": return "Potential duplicate dividend entry detected" if code == "INSUFFICIENT_CASH": return "Insufficient cash balance" if code == "INSUFFICIENT_QUANTITY": return "Insufficient asset quantity" if code == "INSUFFICIENT_BUFFER": return "Insufficient profit buffer" if code == "LEDGER_ENTRY_ALREADY_REVERSED": return "Ledger entry is already reversed" return "Manual ledger request could not be processed" def _encode_ledger_cursor(*, portfolio_id: str, source: str | None, recorded_at: datetime, ledger_entry_id: int) -> str: payload = { "portfolio_id": portfolio_id, "source": source, "recorded_at": recorded_at.isoformat(), "ledger_entry_id": ledger_entry_id, } raw = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() signature = hmac.new(get_settings().ledger_cursor_secret.encode(), raw, hashlib.sha256).digest() return base64.urlsafe_b64encode(raw + signature).decode().rstrip("=") def _decode_ledger_cursor(cursor: str, *, portfolio_id: str, source: str | None) -> tuple[datetime, int]: try: encoded = cursor.encode() raw_and_signature = base64.urlsafe_b64decode(encoded + b"=" * (-len(encoded) % 4)) if base64.urlsafe_b64encode(raw_and_signature).rstrip(b"=") != encoded: raise ValueError raw, signature = raw_and_signature[:-32], raw_and_signature[-32:] expected = hmac.new(get_settings().ledger_cursor_secret.encode(), raw, hashlib.sha256).digest() payload = json.loads(raw) recorded_at = datetime.fromisoformat(payload["recorded_at"]) ledger_entry_id = int(payload["ledger_entry_id"]) if not hmac.compare_digest(signature, expected) or payload["portfolio_id"] != portfolio_id or payload["source"] != source: raise ValueError if recorded_at.tzinfo is None: raise ValueError return recorded_at, ledger_entry_id except (KeyError, TypeError, ValueError, json.JSONDecodeError, UnicodeDecodeError, binascii.Error): raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="invalid ledger cursor") from None def _ledger_entry_to_history_item(entry) -> LedgerHistoryItem | DividendLedgerHistoryItem | CorporateActionLedgerHistoryItem: if getattr(entry, "corporate_action", None): event = entry.corporate_action return CorporateActionLedgerHistoryItem( ledger_entry_id=entry.ledger_entry_id, portfolio_id=entry.portfolio_id, source="corporate_action", event_id=str(event["corporate_action_id"]), event_type=event["action_type"], effective_date=event["effective_date"], quantity_before=event["quantity_before"], quantity_after=event["quantity_after"], average_price_before=event["average_price_before"], average_price_after=event["average_price_after"], total_cost_basis=event["total_cost_basis"], cash_balance=event["cash_balance"], profit_buffer=event["profit_buffer"], evidence_ref=event["evidence_ref"], actor_user_id=event.get("actor_user_id"), recorded_at=entry.recorded_at, execution_venue=entry.execution_venue, order_venue=entry.order_venue, ) action = (entry.action or entry.side or "").upper() if action == "DIVIDEND": amount_decimal = entry.price_decimal if entry.price_decimal is not None else Decimal(str(entry.price)) tax_decimal = entry.tax_amount_decimal if entry.tax_amount_decimal is not None else Decimal(str(entry.tax_amount)) return DividendLedgerHistoryItem( ledger_entry_id=entry.ledger_entry_id, portfolio_id=entry.portfolio_id, source=entry.source or "broker", action=action, recorded_at=entry.recorded_at, executed_at=entry.executed_at, order_id=entry.order_id, execution_id=entry.execution_id, quantity=entry.quantity, amount=str(amount_decimal), price=entry.price, fee_amount=entry.fee_amount, tax_amount=str(tax_decimal), note=entry.note, cash_balance_after=entry.cash_balance_after, asset_quantity_after=entry.asset_quantity_after, profit_buffer_after=entry.profit_buffer_after, total_value_after=entry.total_value_after, target_ledger_entry_id=entry.target_ledger_entry_id, confirmation_evidence=entry.confirmation_evidence, execution_venue=entry.execution_venue, order_venue=entry.order_venue, ) return LedgerHistoryItem( ledger_entry_id=entry.ledger_entry_id, portfolio_id=entry.portfolio_id, source=entry.source or "broker", action=action, recorded_at=entry.recorded_at, executed_at=entry.executed_at, order_id=entry.order_id, execution_id=entry.execution_id, quantity=entry.quantity, amount=float(entry.price) if action in {"INVEST", "BUFFER_OUT", "BUFFER_IN", "WITHDRAW", "PROFIT"} else None, price=entry.price, fee_amount=entry.fee_amount, tax_amount=float(entry.tax_amount), note=entry.note, cash_balance_after=entry.cash_balance_after, asset_quantity_after=entry.asset_quantity_after, profit_buffer_after=entry.profit_buffer_after, total_value_after=entry.total_value_after, target_ledger_entry_id=entry.target_ledger_entry_id, confirmation_evidence=entry.confirmation_evidence, execution_venue=entry.execution_venue, order_venue=entry.order_venue, ) @router.post( "/{portfolio_id}/ledger/manual", response_model=ManualLedgerResponse | DividendManualLedgerResponse, status_code=status.HTTP_201_CREATED, ) def record_manual_ledger_v1( portfolio_id: str, body: ManualLedgerRequest, auth: AuthContext = Depends(require_read), ) -> ManualLedgerResponse | DividendManualLedgerResponse: if auth.token_type in {"service", "client"} or auth.user_id is None: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="manual ledger requires user/session token") repo = _get_repo() try: portfolio = _get_owner_portfolio_or_404(auth, portfolio_id, repo) if portfolio.status == "disabled": raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail={"code": "PORTFOLIO_DISABLED"}) if portfolio.managed_by_daemon_id and not body.confirm_daemon_managed_manual_write: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={ "code": "MANUAL_WRITE_REQUIRES_CONFIRMATION", "managed_by_daemon_id": portfolio.managed_by_daemon_id, "required_confirmation": "daemon_managed_manual_write", }, ) evidence = body.confirmation_evidence if portfolio.managed_by_daemon_id and body.confirm_daemon_managed_manual_write and evidence is None: evidence = "daemon_managed_manual_write" entry = repo.record_manual_ledger( portfolio_id, action=body.action, amount=body.amount, quantity=body.quantity, price=body.price, fee=body.fee, tax=body.tax, executed_at=( datetime( body.executed_at.year, body.executed_at.month, body.executed_at.day, tzinfo=timezone.utc, ) if body.executed_at is not None else None ), target_ledger_entry_id=body.target_ledger_entry_id, note=body.note, actor_user_id=auth.user_id, confirmation_evidence=evidence, confirm_duplicate=body.confirm_duplicate, ) except HTTPException: raise except ValueError as exc: message = str(exc) code = _manual_error_code(message) if code == "LEDGER_ENTRY_NOT_FOUND": status_code = status.HTTP_404_NOT_FOUND detail: str | dict[str, object] = {"code": code, "message": _manual_error_message(code)} elif code == "POTENTIAL_DUPLICATE_DIVIDEND": status_code = status.HTTP_409_CONFLICT detail = { "code": code, "message": _manual_error_message(code), "matched_ledger_entry_id": _extract_duplicate_dividend_entry_id(message), } else: status_code = status.HTTP_409_CONFLICT if code != "MANUAL_LEDGER_CONFLICT" else status.HTTP_422_UNPROCESSABLE_ENTITY detail = {"code": code, "message": _manual_error_message(code)} raise HTTPException(status_code=status_code, detail=detail) from exc finally: _close_repo(repo) return _manual_entry_to_response(entry) @router.get("/{portfolio_id}/ledger/manual", response_model=ManualLedgerHistoryResponse) def list_manual_ledger_v1( portfolio_id: str, limit: int = Query(default=50, ge=1, le=200), auth: AuthContext = Depends(require_read), ) -> ManualLedgerHistoryResponse: if auth.token_type in {"service", "client"} or auth.user_id is None: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="manual ledger requires user/session token") repo = _get_repo() try: _get_owner_portfolio_or_404(auth, portfolio_id, repo) entries = repo.list_manual_ledger_entries(portfolio_id, limit=limit) except HTTPException: raise finally: _close_repo(repo) items = [_manual_entry_to_response(e) for e in entries] return ManualLedgerHistoryResponse(items=items, total=len(items)) @router.get("/{portfolio_id}/ledger", response_model=LedgerHistoryResponse) def list_ledger_v1( portfolio_id: str, source: str | None = Query(default=None, min_length=1), cursor: str | None = Query(default=None), limit: int = Query(default=50, ge=1, le=200), auth: AuthContext = Depends(require_read), ) -> LedgerHistoryResponse: if auth.token_type in {"service", "client"} or auth.user_id is None: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="ledger requires user/session token") before_recorded_at = before_ledger_entry_id = None if cursor is not None: before_recorded_at, before_ledger_entry_id = _decode_ledger_cursor(cursor, portfolio_id=portfolio_id, source=source) repo = _get_repo() try: _get_owner_portfolio_or_404(auth, portfolio_id, repo) entries = repo.list_ledger_entries( portfolio_id, source=source, before_recorded_at=before_recorded_at, before_ledger_entry_id=before_ledger_entry_id, limit=limit + 1, ) finally: _close_repo(repo) page, extra = entries[:limit], entries[limit:] next_cursor = None if extra and page: last = page[-1] if last.recorded_at is None: raise RuntimeError("ledger entry is missing recorded_at") next_cursor = _encode_ledger_cursor( portfolio_id=portfolio_id, source=source, recorded_at=last.recorded_at, ledger_entry_id=last.ledger_entry_id ) return LedgerHistoryResponse(items=[_ledger_entry_to_history_item(entry) for entry in page], next_cursor=next_cursor) @router.get("/{portfolio_id}/returns/summary", response_model=PortfolioReturnSummaryResponse) def get_portfolio_return_summary_v1( portfolio_id: str, auth: AuthContext = Depends(require_read), ) -> PortfolioReturnSummaryResponse: if auth.token_type in {"service", "client"} or auth.user_id is None: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="return summary requires user/session token") repo = _get_repo() try: _get_owner_portfolio_or_404(auth, portfolio_id, repo) summary = repo.get_return_summary(portfolio_id) except HTTPException: raise finally: _close_repo(repo) return PortfolioReturnSummaryResponse(portfolio_id=portfolio_id, **summary) # --------------------------------------------------------------------------- # Helpers for date parameter parsing (REQ-039, REQ-040) # --------------------------------------------------------------------------- def _parse_date_param(value: str, param_name: str) -> datetime: """Parse YYYY-MM-DD string to UTC datetime (start of day). Raises 422 on bad format. Strictly requires YYYY-MM-DD with dashes (e.g. 2026-06-15). Python's date.fromisoformat() also accepts YYYYMMDD in 3.11+, so we explicitly validate the format first. """ import re if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"날짜 형식 오류: {param_name}은 YYYY-MM-DD 형식이어야 합니다. 입력값: {value!r}", ) try: d = date.fromisoformat(value) except ValueError: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"날짜 형식 오류: {param_name}은 YYYY-MM-DD 형식이어야 합니다. 입력값: {value!r}", ) return datetime(d.year, d.month, d.day, 0, 0, 0, tzinfo=timezone.utc) # --------------------------------------------------------------------------- # GET /v1/portfolios/{portfolio_id}/trades — REQ-039 # --------------------------------------------------------------------------- @router.get("/{portfolio_id}/trades", response_model=TradeHistoryResponse) def list_portfolio_trades( portfolio_id: str, date_param: str | None = Query(default=None, alias="date", description="단일 날짜 필터 (YYYY-MM-DD)"), from_param: str | None = Query(default=None, alias="from", description="범위 시작 날짜 (YYYY-MM-DD, 포함)"), to_param: str | None = Query(default=None, alias="to", description="범위 종료 날짜 (YYYY-MM-DD, 포함)"), auth: AuthContext = Depends(require_read), ) -> TradeHistoryResponse: """체결 내역 조회 — 날짜/범위 필터 (REQ-039). - `date`: YYYY-MM-DD 단일 날짜 - `from` + `to`: 날짜 범위 (경계 포함) - 날짜 파라미터 없으면 HTTP 422 """ if auth.token_type in ("service", "client"): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"{auth.token_type} token cannot query trade history", ) # Validate date params — AC-3: at least one of date or from/to required has_date = date_param is not None has_range = from_param is not None or to_param is not None if not has_date and not has_range: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="날짜 파라미터가 필요합니다: date 또는 from/to 중 하나를 지정하세요.", ) # If range is partially specified, require both if has_range and (from_param is None or to_param is None): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="from과 to 파라미터는 둘 다 필요합니다.", ) repo = _get_repo() try: _get_portfolio_or_404(auth, portfolio_id, repo) if has_date: date_dt = _parse_date_param(date_param, "date") # type: ignore[arg-type] trades = repo.list_trades_by_date(portfolio_id, date=date_dt) else: from_dt = _parse_date_param(from_param, "from") # type: ignore[arg-type] to_dt = _parse_date_param(to_param, "to") # type: ignore[arg-type] trades = repo.list_trades_by_date(portfolio_id, from_=from_dt, to_=to_dt) except HTTPException: raise finally: _close_repo(repo) items = [ TradeHistoryItem( ledger_entry_id=t.ledger_entry_id, portfolio_id=t.portfolio_id, order_id=t.order_id, execution_id=t.execution_id, side=t.side or "", quantity=t.quantity, price=t.price, fee_amount=t.fee_amount, tax_amount=t.tax_amount, executed_at=t.executed_at, # type: ignore[arg-type] recorded_at=t.recorded_at, ) for t in trades ] return TradeHistoryResponse(items=items, total=len(items)) # --------------------------------------------------------------------------- # GET /v1/portfolios/{portfolio_id}/pnl/realized — REQ-040 # --------------------------------------------------------------------------- @router.get("/{portfolio_id}/pnl/realized", response_model=RealizedPnlResponse) def get_realized_pnl( portfolio_id: str, from_param: str | None = Query(default=None, alias="from", description="집계 시작 날짜 (YYYY-MM-DD, 필수)"), to_param: str | None = Query(default=None, alias="to", description="집계 종료 날짜 (YYYY-MM-DD, 필수)"), symbol: str | None = Query(default=None, description="종목 필터 (선택)"), auth: AuthContext = Depends(require_read), ) -> RealizedPnlResponse: """실현 PnL 집계 조회 — WAC 방식 (REQ-040). - `from`, `to`: 집계 기간 (필수, 경계 포함) - `symbol`: 선택적 종목 필터 - 응답에 data_completeness_warning (이력 불완전 시), cost_basis_from 포함 """ if auth.token_type in ("service", "client"): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"{auth.token_type} token cannot query realized PnL", ) # AC-3: both from and to are required if from_param is None or to_param is None: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="from과 to 파라미터는 둘 다 필요합니다.", ) from_dt = _parse_date_param(from_param, "from") to_dt = _parse_date_param(to_param, "to") # to_dt is inclusive: extend to end of day (start of next day is used in WAC filter) to_dt_inclusive_end = to_dt + timedelta(days=1) repo = _get_repo() try: portfolio = _get_portfolio_or_404(auth, portfolio_id, repo) # Fetch all buy/sell history for this portfolio (WAC requires full history) ledger_entries = repo.get_all_trades_for_pnl(portfolio_id, symbol=symbol) except HTTPException: raise finally: _close_repo(repo) # Convert LedgerEntry objects to dicts for the domain function # symbol is taken from the portfolio (one portfolio = one symbol) portfolio_symbol = portfolio.symbol rows = [ { "side": e.side, "quantity": e.quantity, "price": e.price, "fee_amount": e.fee_amount, "tax_amount": e.tax_amount, "executed_at": e.executed_at, "order_id": e.order_id, "execution_id": e.execution_id, # symbol is set from portfolio; symbol filter applied in WAC function "symbol": portfolio_symbol if symbol is None else (portfolio_symbol if portfolio_symbol == symbol else None), } for e in ledger_entries ] result = calculate_realized_pnl(rows, from_dt=from_dt, to_dt=to_dt_inclusive_end, symbol=symbol) pnl_items = [ RealizedPnlItem( symbol=item.symbol, executed_at=item.executed_at, quantity=item.quantity, sell_price=item.sell_price, cost_basis_per_unit=item.cost_basis_per_unit, fee_amount=item.fee_amount, tax_amount=item.tax_amount, realized_pnl=item.realized_pnl, order_id=item.order_id, execution_id=item.execution_id, ) for item in result.items ] return RealizedPnlResponse( total_realized_pnl=result.total_realized_pnl, items=pnl_items, cost_basis_from=result.cost_basis_from, data_completeness_warning=result.data_completeness_warning, ) # --------------------------------------------------------------------------- # GET /v1/portfolios/{portfolio_id}/holdings/cost-basis — REQ-043 # --------------------------------------------------------------------------- @router.get("/{portfolio_id}/holdings/cost-basis", response_model=HoldingsCostBasisResponse) def get_holdings_cost_basis( portfolio_id: str, auth: AuthContext = Depends(require_read), ) -> HoldingsCostBasisResponse: """현재 보유 평단(WAC) 조회 — 미매도 보유분의 cost basis (REQ-043). - service token: daemon_id == portfolio.managed_by_daemon_id 일치 시만 허용 (불일치/null → 403). - user token: 소유자 또는 admin만 허용 (비소유자 → 403). - 존재하지 않는 portfolio_id → 404. - 보유수량 0인 경우 → 200 + current_quantity=0, cost_basis_per_unit=null, total_cost_basis=null. """ repo = _get_repo() try: # Fetch portfolio (existence check) try: portfolio = repo.get(portfolio_id) except ValueError: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"portfolio를 찾을 수 없습니다: {portfolio_id}", ) # Ownership / scope check if auth.token_type == "service": # daemon scope: daemon_id must match managed_by_daemon_id if portfolio.managed_by_daemon_id is None: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="daemon 미배정 portfolio에 daemon 접근 불가", ) if auth.daemon_id is None or auth.daemon_id != portfolio.managed_by_daemon_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Token daemon does not match portfolio daemon assignment", ) else: # user/admin token: non-admin must own the portfolio if auth.permission != Permission.ADMIN and portfolio.owner_user_id != auth.user_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Portfolio ownership mismatch", ) # Fetch all ledger history for WAC replay ledger_entries = repo.get_all_trades_for_pnl(portfolio_id, symbol=None) except HTTPException: raise finally: _close_repo(repo) # Build rows for domain function, preserving joined corporate-action metadata. rows = [] corporate_action_fields = ( "corporate_action_id", "action_type", "quantity_before", "quantity_after", "average_price_after", "total_cost_basis", ) for entry in ledger_entries: row = { "side": entry.side, "quantity": entry.quantity, "price": entry.price, "fee_amount": entry.fee_amount, "executed_at": entry.executed_at, "ledger_entry_id": entry.ledger_entry_id, } if entry.corporate_action: row.update( { field: entry.corporate_action.get(field) for field in corporate_action_fields } ) rows.append(row) wac_result = calculate_current_wac(rows) current_quantity = portfolio.asset_quantity if wac_result.total_cost_basis is not None and wac_result.current_quantity > 0: current_quantity = wac_result.current_quantity total_cost_basis = wac_result.total_cost_basis cost_basis_per_unit = total_cost_basis / current_quantity else: cost_basis_per_unit = wac_result.current_wac if current_quantity > 0 else None total_cost_basis = ( cost_basis_per_unit * current_quantity if cost_basis_per_unit is not None and current_quantity > 0 else None ) return HoldingsCostBasisResponse( portfolio_id=portfolio_id, symbol=portfolio.symbol, current_quantity=current_quantity, cost_basis_per_unit=cost_basis_per_unit, total_cost_basis=total_cost_basis, cost_basis_from=wac_result.cost_basis_from, data_completeness_warning=wac_result.data_completeness_warning, ) @router.post("/{portfolio_id}/corporate-actions", response_model=CorporateActionResponse, status_code=status.HTTP_201_CREATED) def create_corporate_action( portfolio_id: str, payload: CorporateActionRequest, response: Response, auth: AuthContext = Depends(require_trade_write), # noqa: B008 - FastAPI dependency declaration idempotency_key: str = Header(..., alias="Idempotency-Key"), ): if not idempotency_key or not idempotency_key.strip(): raise HTTPException(status_code=422, detail="Idempotency-Key must not be empty") repo = _get_repo() try: portfolio = repo.get(portfolio_id) if portfolio is None: raise HTTPException(status_code=404, detail="Portfolio not found") if portfolio.owner_user_id != auth.user_id and auth.permission != Permission.ADMIN: raise HTTPException(status_code=403, detail="Portfolio ownership mismatch") if portfolio.managed_by_daemon_id and not payload.confirm_daemon_managed: raise HTTPException(status_code=409, detail="DAEMON_MANAGED_CONFIRMATION_REQUIRED") from app.domain.corporate_actions import canonical_payload_hash digest = canonical_payload_hash(payload) # The repository derives all WAC and immutable snapshot fields under its row lock. action = CorporateAction(str(uuid.uuid4()), portfolio_id, portfolio.asset_id, payload.action_type, payload.effective_date, payload.quantity_before, payload.quantity_after, Decimal(0), Decimal(0), Decimal(0), Decimal(str(portfolio.cash_balance)), Decimal(str(portfolio.profit_buffer)), payload.evidence_ref, idempotency_key, digest, 0, auth.user_id) saved = repo.record_corporate_action( portfolio_id, action=action, idempotency_key=idempotency_key, payload_hash=digest, resume_partial=payload.resume_partial, ) return CorporateActionResponse(**saved.__dict__) except ValueError as exc: code = str(exc) if code == PORTFOLIO_NOT_FOUND_MESSAGE: raise HTTPException(status_code=404, detail="Portfolio not found") from exc raise HTTPException(status_code=409 if code.startswith(("IDEMPOTENCY", "CORPORATE_ACTION_")) else 400, detail=code) from exc finally: _close_repo(repo)