portfolio-hub/app/domain/pnl.py

"""WAC (weighted average cost) based realized PnL calculation (REQ-041).

This module contains pure Python domain logic with no repository or DB dependency.
All inputs are plain dicts or dataclasses; no database driver imports are used here.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any


@dataclass
class PnlItem:
    """Realized PnL for a single sell trade."""

    symbol: str | None
    executed_at: datetime
    quantity: float
    sell_price: float
    cost_basis_per_unit: float
    fee_amount: float
    tax_amount: float
    realized_pnl: float
    order_id: str | None = None
    execution_id: str | None = None


@dataclass
class WacPnlResult:
    """Result of WAC realized PnL calculation over a set of ledger rows."""

    items: list[PnlItem] = field(default_factory=list)
    total_realized_pnl: float = 0.0
    cost_basis_from: datetime | None = None
    data_completeness_warning: str | None = None


def calculate_realized_pnl(
    rows: list[dict[str, Any]],
    *,
    from_dt: datetime | None = None,
    to_dt: datetime | None = None,
    symbol: str | None = None,
) -> WacPnlResult:
    """Calculate WAC-based realized PnL from a time-ordered list of ledger rows.

    Args:
        rows: List of portfolio_ledger row dicts, each containing:
              side (str), quantity (float), price (float),
              fee_amount (float), tax_amount (float),
              executed_at (datetime), order_id (str|None),
              execution_id (str|None), symbol (str|None, optional).
              Must be sorted by executed_at ASC.
        from_dt: Inclusive lower bound for sell trade date filter (UTC datetime).
                 If None, no lower bound is applied.
        to_dt:   Inclusive upper bound for sell trade date filter (UTC datetime).
                 If None, no upper bound is applied.
        symbol:  Optional symbol filter for sell trades. Ignored if None.

    Returns:
        WacPnlResult with per-sell PnL items, total, cost_basis_from,
        and data_completeness_warning when history is incomplete.

    WAC Rules (REQ-041):
        buy:  running_wac = (running_wac * running_qty + price * qty) / (running_qty + qty)
              running_qty += qty
        sell: realized_pnl = (sell_price - running_wac) * qty - fee - tax
              running_qty -= qty
        Incomplete history (running_qty < sell_qty or sell with running_qty=0):
              data_completeness_warning is set; calculation continues with best estimate.
    """
    if not rows:
        return WacPnlResult()

    running_wac: float = 0.0
    running_qty: float = 0.0
    cost_basis_from: datetime | None = None
    warning_triggered: bool = False
    items: list[PnlItem] = []
    total_pnl: float = 0.0

    for row in rows:
        side = str(row.get("side", "")).lower()
        qty = float(row.get("quantity", 0.0))
        price = float(row.get("price", 0.0))
        fee = float(row.get("fee_amount", 0.0))
        tax = float(row.get("tax_amount", 0.0))
        executed_at: datetime | None = row.get("executed_at")
        row_symbol: str | None = row.get("symbol")
        order_id: str | None = row.get("order_id")
        execution_id: str | None = row.get("execution_id")

        if side == "buy":
            # Update running WAC
            new_total_qty = running_qty + qty
            if new_total_qty > 0:
                running_wac = (running_wac * running_qty + price * qty) / new_total_qty
            running_qty = new_total_qty
            # Track earliest buy for cost_basis_from
            if cost_basis_from is None and executed_at is not None:
                cost_basis_from = executed_at

        elif side == "sell":
            # Check for incomplete history
            if running_qty <= 0:
                warning_triggered = True
            elif qty > running_qty + 1e-9:
                warning_triggered = True

            # Apply date filter for sell trades (all buy rows participate in WAC regardless)
            in_range = True
            if from_dt is not None and executed_at is not None:
                # Make both timezone-aware or both naive for comparison
                exec_dt = _normalize_tz(executed_at, from_dt)
                if exec_dt < from_dt:
                    in_range = False
            if to_dt is not None and executed_at is not None and in_range:
                exec_dt_to = _normalize_tz(executed_at, to_dt)
                if exec_dt_to > to_dt:
                    in_range = False

            # Apply symbol filter
            if symbol is not None and row_symbol is not None and row_symbol != symbol:
                in_range = False

            if in_range:
                realized = (price - running_wac) * qty - fee - tax
                total_pnl += realized
                items.append(
                    PnlItem(
                        symbol=row_symbol,
                        executed_at=executed_at,  # type: ignore[arg-type]
                        quantity=qty,
                        sell_price=price,
                        cost_basis_per_unit=running_wac,
                        fee_amount=fee,
                        tax_amount=tax,
                        realized_pnl=realized,
                        order_id=order_id,
                        execution_id=execution_id,
                    )
                )

            # Update running qty regardless of date filter (WAC state must be correct)
            running_qty -= qty
            if running_qty < 0:
                running_qty = 0.0

    warning: str | None = None
    if warning_triggered:
        warning = (
            "이력 불완전: 보유 수량보다 많은 매도 또는 buy 이력 없는 sell이 감지되어 "
            "cost_basis 산출이 부정확할 수 있습니다. "
            "portfolio_ledger 이력이 포트폴리오 생성 초기부터 완전한지 확인하세요."
        )

    return WacPnlResult(
        items=items,
        total_realized_pnl=total_pnl,
        cost_basis_from=cost_basis_from,
        data_completeness_warning=warning,
    )


def _normalize_tz(dt: datetime, reference: datetime) -> datetime:
    """Align timezone-awareness of dt to match reference.

    If reference is timezone-aware and dt is naive, treat dt as UTC.
    If reference is naive and dt is aware, strip tzinfo.
    """
    import datetime as _dt_module

    if reference.tzinfo is not None and dt.tzinfo is None:
        return dt.replace(tzinfo=_dt_module.timezone.utc)
    if reference.tzinfo is None and dt.tzinfo is not None:
        return dt.replace(tzinfo=None)
    return dt