"""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. Note: WAC formula includes buy fee in cost basis (매수 수수료 편입, 2026-06-26). buy: running_wac = (running_wac*qty + price*qty + buy_fee) / new_total_qty """ from __future__ import annotations from datetime import datetime, timezone import pytest from app.domain.pnl import WacPnlResult, calculate_current_wac, 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 # --------------------------------------------------------------------------- # 매수 수수료 WAC 편입 — 단위 테스트 (2026-06-26 변경) # --------------------------------------------------------------------------- class TestBuyFeeInCostBasis: """buy fee를 WAC cost basis에 편입하는 공식 검증. 공식: running_wac = (running_wac*running_qty + price*qty + buy_fee) / new_total_qty """ def test_single_buy_with_fee_realized_pnl_957(self) -> None: """기준 케이스: buy 10@1000 fee=5, sell 10@1100 fee=5 tax=33. 편입 WAC = (1000*10 + 5)/10 = 1000.5 realized = (1100 - 1000.5)*10 - 5 - 33 = 995 - 38 = 957 """ rows = [ _row("buy", 10, 1000, fee=5.0, executed_at=_T0), _row("sell", 10, 1100, fee=5.0, tax=33.0, executed_at=_T1), ] result = calculate_realized_pnl(rows) assert len(result.items) == 1 assert result.items[0].cost_basis_per_unit == pytest.approx(1000.5) assert result.items[0].realized_pnl == pytest.approx(957.0) assert result.total_realized_pnl == pytest.approx(957.0) def test_single_buy_without_fee_is_unchanged(self) -> None: """buy fee=0이면 기존 공식과 동일 결과.""" rows = [ _row("buy", 100, 10_000, fee=0.0, executed_at=_T0), _row("buy", 100, 12_000, fee=0.0, executed_at=_T1), _row("sell", 50, 15_000, fee=500, tax=200, executed_at=_T2), ] result = calculate_realized_pnl(rows) assert result.items[0].cost_basis_per_unit == pytest.approx(11_000.0) assert result.total_realized_pnl == pytest.approx(199_300.0) def test_multi_buy_fee_weighted_average(self) -> None: """다회 매수 시 fee가 WAC에 가중평균으로 반영된다. buy 100@10,000 fee=1,000 → WAC_1 = (0 + 10000*100 + 1000) / 100 = 10010.0 buy 100@12,000 fee=1,200 → WAC_2 = (10010*100 + 12000*100 + 1200) / 200 = (1001000 + 1200000 + 1200) / 200 = 2202200 / 200 = 11011.0 sell 50@15,000 fee=500 tax=200 realized = (15000 - 11011)*50 - 500 - 200 = 199450 - 700 = 198750 """ rows = [ _row("buy", 100, 10_000, fee=1_000.0, executed_at=_T0), _row("buy", 100, 12_000, fee=1_200.0, executed_at=_T1), _row("sell", 50, 15_000, fee=500.0, tax=200.0, executed_at=_T2), ] result = calculate_realized_pnl(rows) assert len(result.items) == 1 assert result.items[0].cost_basis_per_unit == pytest.approx(11_011.0) assert result.total_realized_pnl == pytest.approx(198_750.0) def test_buy_fee_in_cost_basis_no_warning(self) -> None: """매수 fee 편입 시에도 이력 완전한 경우 warning 없음.""" rows = [ _row("buy", 10, 1000, fee=5.0, executed_at=_T0), _row("sell", 10, 1100, fee=5.0, tax=33.0, executed_at=_T1), ] result = calculate_realized_pnl(rows) assert result.data_completeness_warning is None def test_partial_sell_cost_basis_per_unit_reflects_buy_fee(self) -> None: """부분 매도 후 cost_basis_per_unit이 매수 fee 포함 WAC를 반영한다. buy 100@10,000 fee=500 → WAC = (10000*100 + 500)/100 = 10005.0 sell 30@12,000 fee=100 tax=50 realized = (12000 - 10005)*30 - 100 - 50 = 59850 - 150 = 59700 """ rows = [ _row("buy", 100, 10_000, fee=500.0, executed_at=_T0), _row("sell", 30, 12_000, fee=100.0, tax=50.0, executed_at=_T1), ] result = calculate_realized_pnl(rows) assert result.items[0].cost_basis_per_unit == pytest.approx(10_005.0) assert result.items[0].realized_pnl == pytest.approx(59_700.0) class TestCurrentWacBuyFeeInCostBasis: """calculate_current_wac도 매수 fee를 WAC에 편입한다.""" def test_buy_only_with_fee(self) -> None: """buy 10@1000 fee=5 → current_wac = 1000.5.""" from datetime import datetime, timezone rows = [ {"side": "buy", "quantity": 10, "price": 1000, "fee_amount": 5.0, "executed_at": datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)}, ] result = calculate_current_wac(rows) assert result.current_quantity == pytest.approx(10.0) assert result.current_wac == pytest.approx(1000.5) def test_buy_only_no_fee_unchanged(self) -> None: """buy fee=0이면 calculate_current_wac 결과 변화 없음.""" from datetime import datetime, timezone rows = [ {"side": "buy", "quantity": 100, "price": 10_000, "fee_amount": 0.0, "executed_at": datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)}, {"side": "buy", "quantity": 100, "price": 12_000, "fee_amount": 0.0, "executed_at": datetime(2026, 6, 5, 9, 0, tzinfo=timezone.utc)}, ] result = calculate_current_wac(rows) assert result.current_wac == pytest.approx(11_000.0) def test_multi_buy_with_fee_current_wac(self) -> None: """다회 매수 fee가 current_wac에 가중평균 반영된다. buy 100@10,000 fee=1,000 → WAC = 10010.0 buy 100@12,000 fee=1,200 → WAC = 11011.0 sell 150주 → remaining=50, WAC 변화 없음 = 11011.0 """ from datetime import datetime, timezone rows = [ {"side": "buy", "quantity": 100, "price": 10_000, "fee_amount": 1_000.0, "executed_at": datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)}, {"side": "buy", "quantity": 100, "price": 12_000, "fee_amount": 1_200.0, "executed_at": datetime(2026, 6, 5, 9, 0, tzinfo=timezone.utc)}, {"side": "sell", "quantity": 150, "price": 13_000, "fee_amount": 0.0, "executed_at": datetime(2026, 6, 10, 9, 0, tzinfo=timezone.utc)}, ] result = calculate_current_wac(rows) assert result.current_quantity == pytest.approx(50.0) assert result.current_wac == pytest.approx(11_011.0) def test_current_wac_missing_fee_amount_key_defaults_to_zero(self) -> None: """fee_amount 키가 없으면 0으로 폴백 — 기존 동작 호환.""" from datetime import datetime, timezone rows = [ {"side": "buy", "quantity": 100, "price": 10_000, "executed_at": datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc)}, ] result = calculate_current_wac(rows) assert result.current_wac == pytest.approx(10_000.0) # --------------------------------------------------------------------------- # 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}" )