""" Transfer of Earnings calculation model. Encapsulates the logic for computing how much money moves between two contracts in a transfer-of-earnings workflow. Each Transfer holds a calculation type (percentage of revenue, percentage of balance, fixed amount, or conditional recouped earnings) and the parameters needed to evaluate it. Used by: - process_all_inputs.py — batch pipeline that generates debit/credit rows - run_transfer.py — interactive CLI for single transfer calculations - example_transfers.py — worked demonstrations of each calculation type Usage: from contract import Contract from transfer import Transfer source = Contract(contract_id="123", name="Source", projected_balance=1000.0) t = Transfer( transfer_id="T001", from_contract_id="123", to_contract_id="456", calculation_type="percentage_balance", calculation_params={"percentage": 85}, ) amount = t.calculate_amount(source) # Decimal('850.00') """ from __future__ import annotations from decimal import Decimal, InvalidOperation from typing import Any, Optional from contract import Contract # Constants ZERO = Decimal("0") PERCENTAGE_DIVISOR = Decimal("100") # Recognized calculation-type identifiers (kept as plain strings so callers # don't need to import an enum for this POC). CALC_PERCENTAGE_REVENUE = "percentage_revenue" CALC_PERCENTAGE_BALANCE = "percentage_balance" CALC_FIXED_AMOUNT = "fixed_amount" CALC_ALL_RECOUPED = "all_recouped" VALID_CALCULATION_TYPES = frozenset({ CALC_PERCENTAGE_REVENUE, CALC_PERCENTAGE_BALANCE, CALC_FIXED_AMOUNT, CALC_ALL_RECOUPED, }) DEFAULT_REVENUE_BASE = "net_revenue" # Transfer model class Transfer: """A single transfer-of-earnings rule between two contracts. A Transfer knows *how* to compute the amount (via ``calculation_type`` and ``calculation_params``) but not *when* to run — scheduling is handled by the calling pipeline. """ def __init__( self, transfer_id: str, from_contract_id: str, to_contract_id: str, calculation_type: str, calculation_params: Optional[dict[str, Any]] = None, ) -> None: if not transfer_id: raise ValueError("transfer_id must be a non-empty string") if not from_contract_id or not to_contract_id: raise ValueError("Both from_contract_id and to_contract_id are required") normalized_type = calculation_type.lower() if normalized_type not in VALID_CALCULATION_TYPES: raise ValueError( f"Unknown calculation_type {calculation_type!r}. " f"Must be one of: {', '.join(sorted(VALID_CALCULATION_TYPES))}" ) self.transfer_id = transfer_id self.from_contract_id = from_contract_id self.to_contract_id = to_contract_id self.calculation_type = normalized_type self.calculation_params: dict[str, Any] = calculation_params or {} # Amount calculation def calculate_amount(self, from_contract: Contract) -> Decimal: """Compute the transfer amount from the source contract's financials. The calculation strategy is determined by ``self.calculation_type``: * **percentage_revenue** — a percentage of a revenue attribute (``net_revenue`` or ``gross_revenue``) on the source contract. * **percentage_balance** — a percentage of the source contract's projected closing balance. * **fixed_amount** — a static dollar amount, independent of the contract's financials. * **all_recouped** — the full value of a revenue attribute, but *only* if the source contract has recouped; otherwise zero. This supports "pay-through" clauses that activate post-recoupment. Returns: The calculated transfer amount as a ``Decimal``. Raises: ValueError: If a required parameter is missing or cannot be converted to ``Decimal``. """ if self.calculation_type == CALC_PERCENTAGE_REVENUE: return self._calc_percentage_revenue(from_contract) if self.calculation_type == CALC_PERCENTAGE_BALANCE: return self._calc_percentage_balance(from_contract) if self.calculation_type == CALC_FIXED_AMOUNT: return self._calc_fixed_amount() if self.calculation_type == CALC_ALL_RECOUPED: return self._calc_all_recouped(from_contract) # Unreachable — __init__ validates the type — but guard defensively. raise ValueError(f"Unhandled calculation_type: {self.calculation_type!r}") # ------------------------------------------------------------------ # Private calculation helpers # ------------------------------------------------------------------ def _calc_percentage_revenue(self, contract: Contract) -> Decimal: """Percentage of a revenue attribute (net or gross).""" percentage = self._require_decimal_param("percentage") base_attr = self.calculation_params.get("base", DEFAULT_REVENUE_BASE) base_value = self._get_contract_attr(contract, base_attr) return base_value * percentage / PERCENTAGE_DIVISOR def _calc_percentage_balance(self, contract: Contract) -> Decimal: """Percentage of the contract's projected closing balance.""" percentage = self._require_decimal_param("percentage") return Decimal(str(contract.projected_balance)) * percentage / PERCENTAGE_DIVISOR def _calc_fixed_amount(self) -> Decimal: """A static transfer amount, independent of contract financials.""" return self._require_decimal_param("amount") def _calc_all_recouped(self, contract: Contract) -> Decimal: """Full revenue value if the contract has recouped, else zero. This implements "pay-through" transfer clauses: once the source contract recoups its advance, all subsequent earnings flow through to the destination contract. """ if not getattr(contract, "is_recouped", False): return ZERO base_attr = self.calculation_params.get("base", DEFAULT_REVENUE_BASE) return self._get_contract_attr(contract, base_attr) # Parameter helpers def _require_decimal_param(self, key: str) -> Decimal: """Retrieve a required param from ``calculation_params`` as Decimal. Raises: ValueError: If the key is missing or the value cannot be converted to ``Decimal``. """ raw = self.calculation_params.get(key) if raw is None: raise ValueError( f"calculation_params[{key!r}] is required for " f"calculation_type={self.calculation_type!r}" ) try: return Decimal(str(raw)) except InvalidOperation as exc: raise ValueError( f"calculation_params[{key!r}]={raw!r} is not a valid number" ) from exc @staticmethod def _get_contract_attr(contract: Contract, attr_name: str) -> Decimal: """Read a numeric attribute from a Contract, converting to Decimal. Raises: ValueError: If the attribute does not exist on the contract. """ if not hasattr(contract, attr_name): raise ValueError( f"Contract {contract.contract_id!r} has no attribute {attr_name!r}" ) return Decimal(str(getattr(contract, attr_name))) # Dunder methods def __repr__(self) -> str: return ( f"Transfer(id={self.transfer_id!r}, " f"{self.from_contract_id}\u2192{self.to_contract_id}, " f"type={self.calculation_type!r})" )