41 lines
2.4 KiB
Python
41 lines
2.4 KiB
Python
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
from app.domain.corporate_actions import canonical_payload_hash
|
|
from app.domain.pnl import calculate_current_wac, calculate_realized_pnl
|
|
from app.schemas.corporate_actions import CorporateActionRequest
|
|
|
|
|
|
def test_buy_consolidation_sell_replays_wac_and_realized_pnl():
|
|
rows = [
|
|
{"side": "buy", "quantity": 10, "price": 100, "fee_amount": 0, "tax_amount": 0, "executed_at": datetime(2026, 1, 1, tzinfo=timezone.utc)},
|
|
{"side": "stock_consolidation", "quantity": 0, "price": 0, "quantity_before": 10, "quantity_after": 5, "average_price_after": 200, "executed_at": datetime(2026, 1, 2, tzinfo=timezone.utc)},
|
|
{"side": "sell", "quantity": 2, "price": 250, "fee_amount": 0, "tax_amount": 0, "executed_at": datetime(2026, 1, 3, tzinfo=timezone.utc)},
|
|
]
|
|
assert calculate_current_wac(rows).current_wac == 200
|
|
result = calculate_realized_pnl(rows)
|
|
assert result.items[0].cost_basis_per_unit == 200
|
|
assert result.total_realized_pnl == 100
|
|
|
|
|
|
def test_canonical_hash_same_payload_and_changed_payload_differ():
|
|
a = CorporateActionRequest(effective_date=date(2026, 1, 2), quantity_before=Decimal("10"), quantity_after=Decimal("5"), evidence_ref="ev")
|
|
b = CorporateActionRequest(effective_date=date(2026, 1, 2), quantity_before=Decimal("10"), quantity_after=Decimal("4"), evidence_ref="ev")
|
|
assert canonical_payload_hash(a) == canonical_payload_hash(a.model_copy())
|
|
assert canonical_payload_hash(a) != canonical_payload_hash(b)
|
|
|
|
|
|
def test_runtime_schema_has_one_corporate_action_ddl_and_replay_mapping_starts_at_18():
|
|
source = Path("app/repositories/postgres.py").read_text()
|
|
assert source.count("CREATE TABLE IF NOT EXISTS portfolio_corporate_actions") == 1
|
|
assert "strategy_set_id, note, source," in source
|
|
assert "ca.corporate_action_id, ca.action_type" in source
|
|
assert 'row[18] is not None' in source
|
|
|
|
|
|
def test_joined_corporate_action_metadata_is_complete_in_mapping_contract():
|
|
source = Path("app/repositories/postgres.py").read_text()
|
|
for field in ("action_type", "quantity_before", "quantity_after", "average_price_before", "average_price_after", "total_cost_basis", "cash_balance", "profit_buffer", "evidence_ref", "actor_user_id", "idempotency_key", "payload_hash", "recorded_at"):
|
|
assert f'"{field}"' in source
|