portfolio-hub/app/schemas/portfolio.py

"""Portfolio request and response schemas."""

from __future__ import annotations

from datetime import datetime
from typing import Any, Literal

import re as _re

from pydantic import BaseModel, ConfigDict, Field, field_validator

_PORTFOLIO_NAME_RE = _re.compile(r"^[a-zA-Z0-9._:\-]{1,100}$")


class StrategySetResponse(BaseModel):
    id: str
    name: str
    config: str
    nick: str
    created_at: datetime | None = None


class LedgerEntryResponse(BaseModel):
    portfolio_id: str
    recorded_at: datetime
    action: str
    price: float
    quantity: float
    fee_amount: float = 0.0
    tax_amount: float = 0.0
    cash_balance_after: float
    asset_quantity_after: float
    total_value_after: float
    profit_buffer_after: float = 0.0
    strategy_set_id: str = "manual"
    note: str = ""


class PortfolioBase(BaseModel):
    broker: str = "kis"
    status: str = "enabled"
    name: str | None = None
    symbol: str


class PortfolioSummary(BaseModel):
    portfolio_id: str
    broker: str
    status: str
    name: str
    symbol: str
    tags: list[str] = Field(default_factory=list)
    owner_user_id: str | None = None
    cash_balance: float = 0.0
    asset_quantity: float = 0.0
    price: float = 0.0
    profit_buffer: float = 0.0
    active_strategy_id: str | None = None
    avabile_strategies: list[str] | None = None
    created_at: datetime | None = None
    updated_at: datetime | None = None
    last_trade_at: datetime | None = None


class PortfolioListItem(PortfolioSummary):
    total_value: float = 0.0


class PortfolioDetailResponse(BaseModel):
    portfolio: PortfolioSummary
    recent_entries: list[LedgerEntryResponse] = Field(default_factory=list)
    strategy_set: StrategySetResponse | None = None
    live_quantity: float | None = None


class CreatePortfolioRequest(PortfolioBase):
    name: str  # required, overrides Optional from PortfolioBase; used as (owner_user_id, name) identifier
    strategy_name: str = "target_value"
    strategy_params: dict[str, Any] = Field(default_factory=dict)
    active_strategy_id: str | None = None
    avabile_strategies: list[str] | None = None
    cash_balance: float = Field(default=0.0, ge=0.0)
    asset_quantity: float = Field(default=0.0, ge=0.0)
    asset_price: float = Field(default=0.0, ge=0.0)
    profit_buffer: float = Field(default=0.0, ge=0.0)

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str) -> str:
        if not _PORTFOLIO_NAME_RE.match(v):
            raise ValueError(
                f"이름은 [a-zA-Z0-9._:-] 문자만 허용하며 1~100자여야 합니다: {v!r}"
            )
        return v


class UpdatePortfolioRequest(BaseModel):
    broker: str | None = None
    status: str | None = None
    name: str | None = None
    symbol: str | None = None
    strategy_name: str | None = None
    strategy_params: dict[str, Any] | None = None
    active_strategy_id: str | None = None
    avabile_strategies: list[str] | None = None
    asset_price: float | None = Field(default=None, ge=0.0)


class BuyRequest(BaseModel):
    quantity: float = Field(gt=0.0)
    price: float = Field(gt=0.0)
    fee: float = Field(default=0.0, ge=0.0)
    tax: float = Field(default=0.0, ge=0.0)
    note: str = ""


class SellRequest(BaseModel):
    quantity: float = Field(gt=0.0)
    price: float = Field(gt=0.0)
    fee: float = Field(default=0.0, ge=0.0)
    tax: float = Field(default=0.0, ge=0.0)
    note: str = ""


class InvestRequest(BaseModel):
    movement_type: Literal["deposit", "profit", "withdraw"]
    amount: float = Field(gt=0.0)
    note: str = ""


class TransferRequest(BaseModel):
    direction: Literal["to-buffer", "from-buffer"]
    amount: float = Field(gt=0.0)
    note: str = ""


class ResetPortfolioRequest(BaseModel):
    infer_symbol: str | None = None


class CommentCreateRequest(BaseModel):
    content: str


class CommentUpdateRequest(BaseModel):
    content: str


class CommentResponse(BaseModel):
    id: str
    portfolio_id: str
    content: str
    created_at: datetime | None = None
    updated_at: datetime | None = None


class PortfolioTagMutationRequest(BaseModel):
    tag: str = Field(min_length=1)

    @field_validator("tag")
    @classmethod
    def normalize_tag(cls, value: str) -> str:
        normalized = value.strip()
        if not normalized:
            raise ValueError("tag는 비어 있을 수 없습니다")
        return normalized


class PortfolioTagResponse(BaseModel):
    portfolio_id: str
    tag: str
    created_at: datetime | None = None


class ErrorResponse(BaseModel):
    detail: str
    model_config = ConfigDict(extra="ignore")


class CreateStrategySetRequest(BaseModel):
    name: str
    config: str


class UpdateStrategySetRequest(BaseModel):
    name: str | None = None
    config: str | None = None