genichinandClaude Opus 4.8
bd34450ee9
- GET /v1/portfolios/{id}/trades and GET /v1/portfolios/{id}/pnl/realized endpoints
- Weighted Average Cost (WAC) realized PnL calculation in app/domain/pnl.py
- cost_basis and data_completeness_warning returned in pnl API response
- Test coverage: 50 passed (WAC 20 + trades 12 + pnl 18), full 427 passed regression ✓
- WAC validated against 199,300 unit e2e test cases
- DB migration: none required
- QC: qc-reviewer PASS-WITH-NOTES; team lead confirm-batch-verification → release-candidate
- Remaining non-blocking: symbol filter edge case testing, HTTP_422 deprecation
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5
278 lines
12 KiB
Python
278 lines
12 KiB
Python
"""Pure unit tests for WAC cost_basis PnL calculation (REQ-041, REQ-042).
|
|
|
|
These tests have NO DB dependency and must pass without any database container.
|
|
All inputs are plain Python dicts; no psycopg/DB imports are used.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from app.domain.pnl import WacPnlResult, calculate_realized_pnl
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_T0 = datetime(2026, 6, 1, 9, 0, 0, tzinfo=timezone.utc)
|
|
_T1 = datetime(2026, 6, 1, 9, 30, 0, tzinfo=timezone.utc)
|
|
_T2 = datetime(2026, 6, 1, 10, 0, 0, tzinfo=timezone.utc)
|
|
_T3 = datetime(2026, 6, 15, 10, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _row(side: str, qty: float, price: float, *, fee: float = 0.0, tax: float = 0.0,
|
|
executed_at: datetime = _T0, symbol: str | None = "005930") -> dict:
|
|
return {
|
|
"side": side,
|
|
"quantity": qty,
|
|
"price": price,
|
|
"fee_amount": fee,
|
|
"tax_amount": tax,
|
|
"executed_at": executed_at,
|
|
"order_id": None,
|
|
"execution_id": None,
|
|
"symbol": symbol,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-042 AC-1 / REQ-041 AC-1: WAC 199,300 scenario
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestWac199300:
|
|
"""buy 100주@10,000 + buy 100주@12,000 → sell 50주@15,000 fee=500 tax=200.
|
|
|
|
WAC = (100*10,000 + 100*12,000) / 200 = 11,000
|
|
realized_pnl = (15,000 - 11,000) * 50 - 500 - 200 = 200,000 - 700 = 199,300
|
|
"""
|
|
|
|
def _make_rows(self) -> list[dict]:
|
|
return [
|
|
_row("buy", 100, 10_000, executed_at=_T0),
|
|
_row("buy", 100, 12_000, executed_at=_T1),
|
|
_row("sell", 50, 15_000, fee=500, tax=200, executed_at=_T2),
|
|
]
|
|
|
|
def test_total_realized_pnl(self) -> None:
|
|
result = calculate_realized_pnl(self._make_rows())
|
|
assert result.total_realized_pnl == pytest.approx(199_300.0)
|
|
|
|
def test_cost_basis_per_unit(self) -> None:
|
|
result = calculate_realized_pnl(self._make_rows())
|
|
assert len(result.items) == 1
|
|
assert result.items[0].cost_basis_per_unit == pytest.approx(11_000.0)
|
|
|
|
def test_no_warning_when_history_complete(self) -> None:
|
|
result = calculate_realized_pnl(self._make_rows())
|
|
assert result.data_completeness_warning is None
|
|
|
|
def test_cost_basis_from_is_first_buy(self) -> None:
|
|
result = calculate_realized_pnl(self._make_rows())
|
|
assert result.cost_basis_from == _T0
|
|
|
|
def test_sell_item_fields(self) -> None:
|
|
result = calculate_realized_pnl(self._make_rows())
|
|
item = result.items[0]
|
|
assert item.sell_price == pytest.approx(15_000.0)
|
|
assert item.quantity == pytest.approx(50.0)
|
|
assert item.fee_amount == pytest.approx(500.0)
|
|
assert item.tax_amount == pytest.approx(200.0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-041 AC-2: buy only, no sell → items=[], total=0, cost_basis_from set
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestBuyOnlyNoSell:
|
|
def test_empty_items(self) -> None:
|
|
rows = [_row("buy", 100, 10_000, executed_at=_T0)]
|
|
result = calculate_realized_pnl(rows)
|
|
assert result.items == []
|
|
assert result.total_realized_pnl == pytest.approx(0.0)
|
|
|
|
def test_cost_basis_from_set(self) -> None:
|
|
rows = [_row("buy", 100, 10_000, executed_at=_T0)]
|
|
result = calculate_realized_pnl(rows)
|
|
assert result.cost_basis_from == _T0
|
|
|
|
def test_no_warning(self) -> None:
|
|
rows = [_row("buy", 100, 10_000, executed_at=_T0)]
|
|
result = calculate_realized_pnl(rows)
|
|
assert result.data_completeness_warning is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-042 AC-1 / REQ-041 AC-6: empty input → total=0, cost_basis_from=None
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestEmptyInput:
|
|
def test_empty_rows(self) -> None:
|
|
result = calculate_realized_pnl([])
|
|
assert result.items == []
|
|
assert result.total_realized_pnl == pytest.approx(0.0)
|
|
assert result.cost_basis_from is None
|
|
assert result.data_completeness_warning is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-042 AC-1 / REQ-041 AC-3: sell with no prior buy → warning
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSellWithNoBuyHistory:
|
|
def test_warning_is_set(self) -> None:
|
|
rows = [_row("sell", 10, 15_000, fee=100, tax=50, executed_at=_T0)]
|
|
result = calculate_realized_pnl(rows)
|
|
assert result.data_completeness_warning is not None
|
|
assert len(result.data_completeness_warning) > 0
|
|
|
|
def test_calculation_continues(self) -> None:
|
|
"""REQ-041: 중단 없이 최선 추정 계속."""
|
|
rows = [_row("sell", 10, 15_000, fee=100, tax=50, executed_at=_T0)]
|
|
result = calculate_realized_pnl(rows)
|
|
# WAC=0 when no buy history; realized = (15000-0)*10 - 100 - 50
|
|
assert len(result.items) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-042 AC-1 / REQ-041 AC-4: sell quantity exceeds running_qty → warning
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSellExceedsRunningQty:
|
|
def test_warning_when_oversell(self) -> None:
|
|
rows = [
|
|
_row("buy", 10, 10_000, executed_at=_T0),
|
|
_row("sell", 20, 15_000, executed_at=_T1), # oversell: 20 > 10
|
|
]
|
|
result = calculate_realized_pnl(rows)
|
|
assert result.data_completeness_warning is not None
|
|
assert len(result.data_completeness_warning) > 0
|
|
|
|
def test_calculation_not_aborted(self) -> None:
|
|
rows = [
|
|
_row("buy", 10, 10_000, executed_at=_T0),
|
|
_row("sell", 20, 15_000, executed_at=_T1),
|
|
]
|
|
result = calculate_realized_pnl(rows)
|
|
assert len(result.items) == 1 # sell item still produced
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-041 AC-5: multiple buy/sell cycles — WAC accumulates correctly
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMultipleBuySellCycles:
|
|
def test_wac_is_cumulative(self) -> None:
|
|
"""After buy100@10k then buy100@12k then sell50@15k, remaining WAC stays 11k."""
|
|
rows = [
|
|
_row("buy", 100, 10_000, executed_at=datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)),
|
|
_row("buy", 100, 12_000, executed_at=datetime(2026, 6, 1, 9, 30, tzinfo=timezone.utc)),
|
|
_row("sell", 50, 15_000, fee=500, tax=200, executed_at=datetime(2026, 6, 1, 10, 0, tzinfo=timezone.utc)),
|
|
# After sell: running_qty = 150, WAC = 11,000
|
|
# Buy more at 13,000: new WAC = (11000*150 + 13000*50) / 200 = (1,650,000 + 650,000) / 200 = 11,500
|
|
_row("buy", 50, 13_000, executed_at=datetime(2026, 6, 10, 9, 0, tzinfo=timezone.utc)),
|
|
_row("sell", 100, 16_000, fee=300, tax=100, executed_at=datetime(2026, 6, 15, 10, 0, tzinfo=timezone.utc)),
|
|
]
|
|
result = calculate_realized_pnl(rows)
|
|
assert len(result.items) == 2
|
|
|
|
# First sell: WAC=11,000, qty=50, fee=500, tax=200
|
|
item0 = result.items[0]
|
|
assert item0.cost_basis_per_unit == pytest.approx(11_000.0)
|
|
assert item0.realized_pnl == pytest.approx((15_000 - 11_000) * 50 - 500 - 200)
|
|
|
|
# Second sell: WAC=11,500, qty=100, fee=300, tax=100
|
|
item1 = result.items[1]
|
|
assert item1.cost_basis_per_unit == pytest.approx(11_500.0)
|
|
assert item1.realized_pnl == pytest.approx((16_000 - 11_500) * 100 - 300 - 100)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Date filter: from_dt / to_dt
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDateFilter:
|
|
def _make_rows(self) -> list[dict]:
|
|
"""buy once, then two sells on different days."""
|
|
return [
|
|
_row("buy", 200, 10_000, executed_at=datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)),
|
|
_row("sell", 50, 15_000, fee=100, tax=50,
|
|
executed_at=datetime(2026, 6, 10, 10, 0, tzinfo=timezone.utc)),
|
|
_row("sell", 50, 16_000, fee=100, tax=50,
|
|
executed_at=datetime(2026, 6, 20, 10, 0, tzinfo=timezone.utc)),
|
|
]
|
|
|
|
def test_filter_includes_only_sells_in_range(self) -> None:
|
|
from_dt = datetime(2026, 6, 10, 0, 0, 0, tzinfo=timezone.utc)
|
|
to_dt = datetime(2026, 6, 11, 0, 0, 0, tzinfo=timezone.utc) # exclusive upper
|
|
result = calculate_realized_pnl(self._make_rows(), from_dt=from_dt, to_dt=to_dt)
|
|
assert len(result.items) == 1
|
|
assert result.items[0].sell_price == pytest.approx(15_000.0)
|
|
|
|
def test_filter_both_sells_in_range(self) -> None:
|
|
from_dt = datetime(2026, 6, 1, 0, 0, 0, tzinfo=timezone.utc)
|
|
to_dt = datetime(2026, 6, 30, 0, 0, 0, tzinfo=timezone.utc)
|
|
result = calculate_realized_pnl(self._make_rows(), from_dt=from_dt, to_dt=to_dt)
|
|
assert len(result.items) == 2
|
|
|
|
def test_wac_uses_all_buys_regardless_of_filter(self) -> None:
|
|
"""WAC must be computed from ALL buy history, even if buy is before from_dt."""
|
|
rows = [
|
|
_row("buy", 100, 10_000, executed_at=datetime(2026, 5, 1, 9, 0, tzinfo=timezone.utc)),
|
|
_row("sell", 50, 15_000, executed_at=datetime(2026, 6, 15, 10, 0, tzinfo=timezone.utc)),
|
|
]
|
|
from_dt = datetime(2026, 6, 1, 0, 0, 0, tzinfo=timezone.utc)
|
|
to_dt = datetime(2026, 6, 30, 0, 0, 0, tzinfo=timezone.utc)
|
|
result = calculate_realized_pnl(rows, from_dt=from_dt, to_dt=to_dt)
|
|
assert len(result.items) == 1
|
|
assert result.items[0].cost_basis_per_unit == pytest.approx(10_000.0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Symbol filter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSymbolFilter:
|
|
def test_symbol_filter_excludes_other_symbols(self) -> None:
|
|
rows = [
|
|
_row("buy", 100, 10_000, symbol="005930"),
|
|
_row("sell", 50, 15_000, symbol="005930"),
|
|
_row("sell", 30, 20_000, symbol="OTHER"),
|
|
]
|
|
result = calculate_realized_pnl(rows, symbol="005930")
|
|
# Only the 005930 sell should appear
|
|
assert len(result.items) == 1
|
|
assert result.items[0].sell_price == pytest.approx(15_000.0)
|
|
|
|
def test_no_symbol_filter_includes_all(self) -> None:
|
|
rows = [
|
|
_row("buy", 200, 10_000),
|
|
_row("sell", 50, 15_000),
|
|
_row("sell", 30, 16_000),
|
|
]
|
|
result = calculate_realized_pnl(rows)
|
|
assert len(result.items) == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-041 AC-7: no psycopg import in domain module (import-time check)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestNoPsycopgInDomainModule:
|
|
def test_pnl_module_has_no_db_import(self) -> None:
|
|
import importlib
|
|
import sys
|
|
module_name = "app.domain.pnl"
|
|
# Force reimport to check
|
|
mod = sys.modules.get(module_name)
|
|
if mod is None:
|
|
mod = importlib.import_module(module_name)
|
|
source_file = getattr(mod, "__file__", "")
|
|
with open(source_file) as f:
|
|
source = f.read()
|
|
for forbidden in ("psycopg", "asyncpg", "psycopg2"):
|
|
assert forbidden not in source, (
|
|
f"app/domain/pnl.py contains forbidden DB import: {forbidden!r}"
|
|
)
|