57 lines
2.5 KiB
Python
57 lines
2.5 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
|
|
|
|
STORAGE_QUANTUM = Decimal("0.00000001")
|
|
|
|
|
|
def quantize_stored_decimal(value: Decimal) -> Decimal:
|
|
"""Round a monetary value to the NUMERIC(*, 8) persistence contract."""
|
|
return value.quantize(STORAGE_QUANTUM, rounding=ROUND_HALF_UP)
|
|
|
|
|
|
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")
|
|
# Preserve the authoritative pre-action total independently. A per-unit WAC
|
|
# rounded to the storage scale cannot always multiply back to that total
|
|
# exactly, and that harmless decimal residue is not a fractional share.
|
|
total = quantize_stored_decimal(q0 * wac)
|
|
before = quantize_stored_decimal(wac)
|
|
after = quantize_stored_decimal(total / q1)
|
|
return StockConsolidation(q0, q1, before, 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
|