48 lines
2.2 KiB
Python
48 lines
2.2 KiB
Python
"""Pure domain rules for portfolio stock consolidations."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from decimal import ROUND_HALF_UP, Decimal
|
|
|
|
|
|
def canonical_payload_hash(payload: object) -> str:
|
|
if hasattr(payload, "model_dump"):
|
|
payload = payload.model_dump(mode="json", exclude_none=True)
|
|
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
class CorporateActionError(ValueError):
|
|
code = "INVALID_CORPORATE_ACTION_RATIO"
|
|
def __init__(self, message: str, code: str | None = None):
|
|
super().__init__(message); self.code = code or self.code
|
|
|
|
@dataclass(frozen=True)
|
|
class StockConsolidation:
|
|
quantity_before: Decimal
|
|
quantity_after: Decimal
|
|
average_price_before: Decimal
|
|
average_price_after: Decimal
|
|
total_cost_basis: Decimal
|
|
ratio: Decimal
|
|
|
|
def calculate_stock_consolidation(quantity_before: Decimal, quantity_after: Decimal, average_price_before: Decimal) -> StockConsolidation:
|
|
q0, q1, wac = map(Decimal, (quantity_before, quantity_after, average_price_before))
|
|
if q0 <= 0 or q1 <= 0 or wac < 0:
|
|
raise CorporateActionError("quantities must be positive and price non-negative")
|
|
ratio = q1 / q0
|
|
if ratio >= 1 or ratio <= 0:
|
|
raise CorporateActionError("stock consolidation must reduce quantity")
|
|
if q1 != q1.to_integral_value() or q0 != q0.to_integral_value():
|
|
raise CorporateActionError("fractional result unsupported", "FRACTIONAL_RESULT_UNSUPPORTED")
|
|
total = (q0 * wac).quantize(Decimal("0.00000001"), rounding=ROUND_HALF_UP)
|
|
after = (total / q1).quantize(Decimal("0.00000001"), rounding=ROUND_HALF_UP)
|
|
if after * q1 != total:
|
|
raise CorporateActionError("fractional result unsupported", "FRACTIONAL_RESULT_UNSUPPORTED")
|
|
return StockConsolidation(q0, q1, wac, after, total, ratio)
|
|
|
|
def apply_corporate_action_to_wac(quantity: Decimal, wac: Decimal, quantity_after: Decimal) -> tuple[Decimal, Decimal]:
|
|
result = calculate_stock_consolidation(quantity, quantity_after, wac)
|
|
return result.quantity_after, result.average_price_after
|