"""Deterministic reference implementation of the payroll calculator.""" from __future__ import annotations import hashlib import json import re from datetime import date, datetime from decimal import Context, Decimal, DecimalException, localcontext from typing import Protocol from .balances import carryover_contribution from .domain import ( CalculationWarning, ExpectedTotals, InputLine, LineKind, PayrollLine, PayrollPeriod, PayrollRequest, PayrollResult, PayrollRules, PayrollTotals, PointPolicy, RoundingPolicy, decimal_text, ) from .exceptions import PayrollIssue, PayrollValidationError _CURRENCY_RE = re.compile(r"^[A-Z]{3}$") _CODE_RE = re.compile(r"^[A-Z][A-Z0-9_.-]{0,63}$") _SYSTEM_LINE_PREFIX = "system:" class PayrollCalculator(Protocol): """Port implemented by any compatible calculation provider.""" def calculate( self, request: PayrollRequest, rules: PayrollRules, ) -> PayrollResult: ... def _canonical_hash(payload: object) -> str: encoded = json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _is_finite_decimal(value: object) -> bool: return isinstance(value, Decimal) and value.is_finite() class DeterministicPayrollCalculator: """Pure calculator with explicit Decimal and rounding semantics.""" CALCULATOR_VERSION = "1.4.0" DECIMAL_PRECISION = 80 MAX_DECIMAL_DIGITS = 30 MAX_FRACTIONAL_PLACES = 12 MAX_INPUT_LINES = 10_000 def calculate( self, request: PayrollRequest, rules: PayrollRules, ) -> PayrollResult: issues = self._validate(request, rules) if issues: raise PayrollValidationError(issues) decimal_context = Context( prec=self.DECIMAL_PRECISION, rounding=rules.rounding.value, ) with localcontext(decimal_context): quantize = self._quantizer(rules) lines: list[PayrollLine] = [ PayrollLine( line_id="system:base", code="BASE", label="Base accrual", kind=LineKind.EARNING, amount=quantize(request.base_accrual), source_ref=request.base_source_ref, calculated=True, ) ] opening_balance = quantize(request.opening_balance) warnings: list[CalculationWarning] = [] point_delta: Decimal | None = None if rules.point_policy is PointPolicy.PROPORTIONAL_WITH_EXCESS: point_delta = request.actual_points - request.target_points point_base_accrual = ( request.point_base_accrual if request.point_base_accrual is not None else request.base_accrual ) raw_point_amount = ( point_delta * request.point_rate if point_delta > 0 and request.point_rate_overridden else point_base_accrual * (request.actual_points / request.target_points) - point_base_accrual ) point_kind = ( LineKind.ADJUSTMENT_DEBIT if raw_point_amount < 0 else LineKind.ADJUSTMENT_CREDIT ) lines.append( PayrollLine( line_id="system:point-adjustment", code="POINT_ADJUSTMENT", label="Point-based recalculation", kind=point_kind, amount=quantize(abs(raw_point_amount)), source_ref="calculation:point-policy", calculated=True, ) ) elif rules.point_policy is PointPolicy.EXCESS_ONLY: point_delta = request.actual_points - request.target_points eligible_points = max(point_delta, Decimal("0")) point_amount = quantize(eligible_points * request.point_rate) lines.append( PayrollLine( line_id="system:point-excess", code="POINT_EXCESS", label="Points above target", kind=LineKind.EARNING, amount=point_amount, source_ref="calculation:point-policy", calculated=True, ) ) if point_delta < 0: warnings.append( CalculationWarning( code="POINTS_BELOW_TARGET_NO_DEDUCTION", field="actual_points", message=( "Actual points are below target; the active " "policy does not reduce pay." ), context=(("point_delta", decimal_text(point_delta)),), ) ) for input_line in sorted( request.lines, key=lambda item: item.line_id, ): lines.append(self._to_result_line(input_line, quantize)) totals = self._calculate_totals( lines, opening_balance=opening_balance, balance_rules_version=request.balance_rules_version, ) if totals.payable < 0 and not rules.allow_negative_payable: raise PayrollValidationError( [ PayrollIssue( code="NEGATIVE_PAYABLE_NOT_ALLOWED", field="lines", message=( "Deductions and recorded payments exceed available pay." ), context={"payable": decimal_text(totals.payable)}, ) ] ) warnings.extend( self._reconcile_expected( request.expected_totals, lines, totals, quantize, ) ) input_hash = _canonical_hash(request.calculation_dict()) rules_payload = rules.to_dict() ruleset_hash = _canonical_hash(rules_payload) calculator_version = ( "1.3.0" if request.balance_rules_version == 1 else self.CALCULATOR_VERSION ) result_payload = { "employee_ref": request.employee_ref, "period": request.period.to_dict(), "currency": request.currency, "lines": [line.to_dict() for line in lines], "point_delta": ( decimal_text(point_delta) if point_delta is not None else None ), "totals": totals.to_dict(), "ruleset_id": rules.ruleset_id, "ruleset_version": rules.version, "ruleset_hash": ruleset_hash, "calculator_version": calculator_version, "rounding_policy": rules.rounding.value, "input_hash": input_hash, } result_hash = _canonical_hash(result_payload) reconciliation_hash = _canonical_hash( { "result_hash": result_hash, "warnings": [warning.to_dict() for warning in warnings], } ) return PayrollResult( employee_ref=request.employee_ref, period=request.period, currency=request.currency, lines=tuple(lines), point_delta=point_delta, totals=totals, ruleset_id=rules.ruleset_id, ruleset_version=rules.version, ruleset_hash=ruleset_hash, calculator_version=calculator_version, rounding_policy=rules.rounding.value, input_hash=input_hash, result_hash=result_hash, reconciliation_hash=reconciliation_hash, warnings=tuple(warnings), ) @staticmethod def _quantizer(rules: PayrollRules): def quantize(value: Decimal) -> Decimal: try: units = (value / rules.money_quantum).quantize( Decimal("1"), rounding=rules.rounding.value, ) return units * rules.money_quantum except DecimalException as exc: raise PayrollValidationError( [ PayrollIssue( code="ARITHMETIC_ERROR", message="A monetary value cannot be rounded safely.", ) ] ) from exc return quantize @staticmethod def _to_result_line(input_line: InputLine, quantize) -> PayrollLine: return PayrollLine( line_id=input_line.line_id, code=input_line.code, label=input_line.label, kind=input_line.kind, amount=quantize(input_line.amount), source_ref=input_line.source_ref, reason=input_line.reason, source_period=input_line.source_period, is_retro=input_line.is_retro, calculated=False, include_in_carryover=input_line.include_in_carryover, include_in_salary=input_line.include_in_salary, ) @staticmethod def _calculate_totals( lines: list[PayrollLine], *, opening_balance: Decimal = Decimal("0"), balance_rules_version: int = 2, ) -> PayrollTotals: earnings = sum( ( line.amount for line in lines if line.kind is LineKind.EARNING and line.include_in_salary ), Decimal("0"), ) adjustment_credits = sum( ( line.amount for line in lines if line.kind is LineKind.ADJUSTMENT_CREDIT and line.include_in_salary ), Decimal("0"), ) adjustment_debits = sum( ( line.amount for line in lines if line.kind is LineKind.ADJUSTMENT_DEBIT and line.include_in_salary ), Decimal("0"), ) deductions = sum( ( line.amount for line in lines if line.kind is LineKind.DEDUCTION and line.include_in_salary ), Decimal("0"), ) payments = sum( (line.amount for line in lines if line.kind is LineKind.PAYMENT), Decimal("0"), ) non_salary_payments = sum( ( line.amount for line in lines if line.kind is LineKind.PAYMENT and not line.include_in_salary ), Decimal("0"), ) carryover_change = sum( ( carryover_contribution( line.kind.value, line.amount, version=balance_rules_version ) for line in lines if not line.calculated and line.include_in_carryover ), Decimal("0"), ) adjustment_total = adjustment_credits - adjustment_debits gross = earnings + adjustment_total net_pay = gross - deductions return PayrollTotals( gross_before_adjustments=earnings, adjustment_total=adjustment_total, gross_after_adjustments=gross, deduction_total=deductions, net_pay=net_pay, payment_total=payments, payable=net_pay - payments + non_salary_payments, non_salary_payment_total=non_salary_payments, closing_balance=opening_balance + carryover_change, ) @staticmethod def _reconcile_expected( expected: ExpectedTotals | None, lines: list[PayrollLine], totals: PayrollTotals, quantize, ) -> list[CalculationWarning]: if expected is None: return [] point_line = next( ( line for line in lines if line.line_id in {"system:point-excess", "system:point-adjustment"} ), None, ) point_amount = ( -point_line.amount if point_line is not None and point_line.kind is LineKind.ADJUSTMENT_DEBIT else point_line.amount if point_line is not None else Decimal("0") ) gross_before_point_recalculation = totals.gross_after_adjustments - point_amount checks = ( ("point_amount", expected.point_amount, point_amount), ("gross", expected.gross, gross_before_point_recalculation), ( "recalculated_gross", expected.recalculated_gross, totals.recalculated_gross, ), ("payable", expected.payable, totals.payable), ) warnings = [] for field_name, expected_value, actual_value in checks: if expected_value is None: continue normalized_expected = quantize(expected_value) if normalized_expected != actual_value: warnings.append( CalculationWarning( code="LEGACY_EXPECTED_TOTAL_MISMATCH", field=f"expected_totals.{field_name}", message=( "The calculated amount differs from the legacy " "control total." ), context=( ("expected", decimal_text(normalized_expected)), ("actual", decimal_text(actual_value)), ), ) ) return warnings def _validate( self, request: PayrollRequest, rules: PayrollRules, ) -> list[PayrollIssue]: if not isinstance(request, PayrollRequest): return [ PayrollIssue( code="INVALID_REQUEST", field="request", message="Request must be a PayrollRequest instance.", ) ] if not isinstance(rules, PayrollRules): return [ PayrollIssue( code="INVALID_RULESET", field="rules", message="Rules must be a PayrollRules instance.", ) ] issues: list[PayrollIssue] = [] period_is_valid = self._validate_request_fields(request, issues) self._validate_rules(rules, request, period_is_valid, issues) self._validate_point_metrics(request, rules, issues) self._validate_input_lines(request, period_is_valid, issues) self._validate_expected_totals(request, rules, issues) return issues def _validate_request_fields( self, request: PayrollRequest, issues: list[PayrollIssue], ) -> bool: if not self._is_nonempty_string(request.employee_ref): issues.append( PayrollIssue( code="MISSING_EMPLOYEE_REF", field="employee_ref", message="Employee reference is required.", ) ) if type( request.balance_rules_version ) is not int or request.balance_rules_version not in (1, 2): issues.append( PayrollIssue( code="INVALID_BALANCE_RULES_VERSION", field="balance_rules_version", message="Unsupported carryover rules version.", ) ) period_is_valid = self._validate_period(request.period, "period", issues) if not isinstance(request.currency, str) or not _CURRENCY_RE.fullmatch( request.currency ): issues.append( PayrollIssue( code="INVALID_CURRENCY", field="currency", message="Currency must be a three-letter uppercase code.", ) ) self._validate_decimal( request.base_accrual, "base_accrual", issues, minimum=Decimal("0") ) self._validate_decimal( request.opening_balance, "opening_balance", issues, minimum=None ) self._validate_decimal( request.point_base_accrual, "point_base_accrual", issues, minimum=Decimal("0"), allow_none=True, ) if not self._is_nonempty_string(request.base_source_ref): issues.append( PayrollIssue( code="MISSING_SOURCE_REF", field="base_source_ref", message="Base accrual must identify its source.", ) ) self._validate_decimal( request.point_rate, "point_rate", issues, minimum=Decimal("0") ) if not isinstance(request.point_rate_overridden, bool): issues.append( PayrollIssue( code="INVALID_POINT_RATE_MODE", field="point_rate_overridden", message="Point rate override marker must be boolean.", ) ) return period_is_valid def _validate_rules( self, rules: PayrollRules, request: PayrollRequest, period_is_valid: bool, issues: list[PayrollIssue], ) -> None: if not self._is_nonempty_string(rules.ruleset_id): issues.append( PayrollIssue( code="INVALID_RULESET", field="rules.ruleset_id", message="Ruleset identifier is required.", ) ) if not self._is_nonempty_string(rules.version): issues.append( PayrollIssue( code="INVALID_RULESET", field="rules.version", message="Ruleset version is required.", ) ) rules_from_valid = isinstance(rules.effective_from, date) and not isinstance( rules.effective_from, datetime ) rules_to_valid = rules.effective_to is None or ( isinstance(rules.effective_to, date) and not isinstance(rules.effective_to, datetime) ) if not rules_from_valid: issues.append( PayrollIssue( code="INVALID_RULESET_PERIOD", field="rules.effective_from", message="Ruleset effective_from must be a date.", ) ) if not rules_to_valid: issues.append( PayrollIssue( code="INVALID_RULESET_PERIOD", field="rules.effective_to", message="Ruleset effective_to must be a date or None.", ) ) if ( rules_from_valid and rules_to_valid and rules.effective_to is not None and rules.effective_from > rules.effective_to ): issues.append( PayrollIssue( code="INVALID_RULESET_PERIOD", field="rules.effective_to", message="Ruleset effective dates are invalid.", ) ) if ( period_is_valid and rules_from_valid and rules_to_valid and ( request.period.start < rules.effective_from or rules.effective_to is not None and request.period.end > rules.effective_to ) ): issues.append( PayrollIssue( code="RULESET_NOT_EFFECTIVE", field="period", message="Ruleset is not effective for the full period.", ) ) if not isinstance(rules.point_policy, PointPolicy): issues.append( PayrollIssue( code="UNSUPPORTED_POINT_POLICY", field="rules.point_policy", message="Point policy is not supported.", ) ) if not isinstance(rules.rounding, RoundingPolicy): issues.append( PayrollIssue( code="UNSUPPORTED_ROUNDING_POLICY", field="rules.rounding", message="Rounding policy is not supported.", ) ) if not isinstance(rules.allow_negative_payable, bool): issues.append( PayrollIssue( code="INVALID_RULE_OPTION", field="rules.allow_negative_payable", message="allow_negative_payable must be a boolean.", ) ) self._validate_decimal( rules.money_quantum, "rules.money_quantum", issues, minimum=Decimal("0"), exclusive_minimum=True, ) def _validate_point_metrics( self, request: PayrollRequest, rules: PayrollRules, issues: list[PayrollIssue], ) -> None: if rules.point_policy in { PointPolicy.EXCESS_ONLY, PointPolicy.PROPORTIONAL_WITH_EXCESS, }: self._validate_decimal( request.target_points, "target_points", issues, minimum=Decimal("0"), exclusive_minimum=True, ) self._validate_decimal( request.actual_points, "actual_points", issues, minimum=Decimal("0"), ) return if rules.point_policy is not PointPolicy.DISABLED: return has_target = request.target_points is not None has_actual = request.actual_points is not None if has_target != has_actual: issues.append( PayrollIssue( code="INCOMPLETE_POINT_METRICS", field="target_points", message="Target and actual points must be supplied together.", ) ) if has_target: self._validate_decimal( request.target_points, "target_points", issues, minimum=Decimal("0"), exclusive_minimum=True, ) if has_actual: self._validate_decimal( request.actual_points, "actual_points", issues, minimum=Decimal("0"), ) def _validate_input_lines( self, request: PayrollRequest, period_is_valid: bool, issues: list[PayrollIssue], ) -> None: if not isinstance(request.lines, tuple): issues.append( PayrollIssue( code="INVALID_LINES_CONTAINER", field="lines", message="Input lines must be an immutable tuple.", ) ) lines = () else: lines = request.lines if len(lines) > self.MAX_INPUT_LINES: issues.append( PayrollIssue( code="TOO_MANY_INPUT_LINES", field="lines", message=f"At most {self.MAX_INPUT_LINES} input lines are allowed.", ) ) seen_line_ids: set[str] = set() for index, line in enumerate(lines): self._validate_input_line( line, index=index, request=request, period_is_valid=period_is_valid, seen_line_ids=seen_line_ids, issues=issues, ) def _validate_input_line( self, line: object, *, index: int, request: PayrollRequest, period_is_valid: bool, seen_line_ids: set[str], issues: list[PayrollIssue], ) -> None: prefix = f"lines[{index}]" if not isinstance(line, InputLine): issues.append( PayrollIssue( code="INVALID_COMPONENT", field=prefix, message="Every input line must be an InputLine instance.", ) ) return for flag in ("include_in_salary", "include_in_carryover"): if not isinstance(getattr(line, flag), bool): issues.append( PayrollIssue( code="INVALID_COMPONENT_FLAG", field=f"{prefix}.{flag}", message="Component participation flags must be boolean.", ) ) if not self._is_nonempty_string(line.line_id): issues.append( PayrollIssue( code="MISSING_COMPONENT_ID", field=f"{prefix}.line_id", message="Input line identifier is required.", ) ) else: if line.line_id.startswith(_SYSTEM_LINE_PREFIX): issues.append( PayrollIssue( code="RESERVED_COMPONENT_ID", field=f"{prefix}.line_id", message="Input line uses a reserved system prefix.", ) ) if line.line_id in seen_line_ids: issues.append( PayrollIssue( code="DUPLICATE_COMPONENT_ID", field=f"{prefix}.line_id", message="Input line identifiers must be unique.", ) ) seen_line_ids.add(line.line_id) if not isinstance(line.code, str) or not _CODE_RE.fullmatch(line.code): issues.append( PayrollIssue( code="INVALID_COMPONENT_CODE", field=f"{prefix}.code", message="Component code must use canonical uppercase form.", ) ) if not self._is_nonempty_string(line.label): issues.append( PayrollIssue( code="MISSING_COMPONENT_LABEL", field=f"{prefix}.label", message="Input line label is required.", ) ) kind_is_valid = isinstance(line.kind, LineKind) if not kind_is_valid: issues.append( PayrollIssue( code="INVALID_COMPONENT_KIND", field=f"{prefix}.kind", message="Input line kind is not supported.", ) ) self._validate_decimal( line.amount, f"{prefix}.amount", issues, minimum=Decimal("0") ) if not self._is_nonempty_string(line.source_ref): issues.append( PayrollIssue( code="MISSING_SOURCE_REF", field=f"{prefix}.source_ref", message="Every input line must identify its source.", ) ) self._validate_retro_line( line, prefix=prefix, request=request, period_is_valid=period_is_valid, issues=issues, ) def _validate_retro_line( self, line: InputLine, *, prefix: str, request: PayrollRequest, period_is_valid: bool, issues: list[PayrollIssue], ) -> None: is_retro_valid = isinstance(line.is_retro, bool) if not is_retro_valid: issues.append( PayrollIssue( code="INVALID_RETRO_FLAG", field=f"{prefix}.is_retro", message="is_retro must be a boolean.", ) ) if line.is_retro is True and line.source_period is None: issues.append( PayrollIssue( code="MISSING_RETRO_SOURCE_PERIOD", field=f"{prefix}.source_period", message="Retro input must reference the affected period.", ) ) if line.source_period is None: return source_period_valid = self._validate_period( line.source_period, f"{prefix}.source_period", issues ) if is_retro_valid and line.is_retro is not True: issues.append( PayrollIssue( code="UNEXPECTED_SOURCE_PERIOD", field=f"{prefix}.source_period", message="A source period requires is_retro=True.", ) ) elif ( is_retro_valid and line.is_retro is True and source_period_valid and period_is_valid and line.source_period.end >= request.period.start ): issues.append( PayrollIssue( code="INVALID_RETRO_SOURCE_PERIOD", field=f"{prefix}.source_period", message="Retro source period must end before this period.", ) ) def _validate_expected_totals( self, request: PayrollRequest, rules: PayrollRules, issues: list[PayrollIssue], ) -> None: if request.expected_totals is None: return if not isinstance(request.expected_totals, ExpectedTotals): issues.append( PayrollIssue( code="INVALID_EXPECTED_TOTALS", field="expected_totals", message="Expected totals must be an ExpectedTotals instance.", ) ) return for field_name in ("point_amount", "gross", "recalculated_gross", "payable"): allow_negative = field_name == "point_amount" or ( field_name == "payable" and rules.allow_negative_payable ) self._validate_decimal( getattr(request.expected_totals, field_name), f"expected_totals.{field_name}", issues, minimum=None if allow_negative else Decimal("0"), allow_none=True, ) @staticmethod def _is_nonempty_string(value: object) -> bool: return isinstance(value, str) and bool(value.strip()) @staticmethod def _validate_period( value: object, field_name: str, issues: list[PayrollIssue], ) -> bool: if not isinstance(value, PayrollPeriod): issues.append( PayrollIssue( code="INVALID_PERIOD", field=field_name, message="Period must be a PayrollPeriod instance.", ) ) return False if ( not isinstance(value.start, date) or isinstance(value.start, datetime) or not isinstance(value.end, date) or isinstance(value.end, datetime) ): issues.append( PayrollIssue( code="INVALID_PERIOD", field=field_name, message="Period boundaries must be dates.", ) ) return False if value.start > value.end: issues.append( PayrollIssue( code="INVALID_PERIOD", field=field_name, message="Period start must not be after period end.", ) ) return False return True @staticmethod def _validate_decimal( value: object, field_name: str, issues: list[PayrollIssue], *, minimum: Decimal | None, exclusive_minimum: bool = False, allow_none: bool = False, ) -> None: if value is None and allow_none: return if not isinstance(value, Decimal): issues.append( PayrollIssue( code="INVALID_DECIMAL_TYPE", field=field_name, message="Monetary and quantity values must use Decimal.", ) ) return if not _is_finite_decimal(value): issues.append( PayrollIssue( code="NON_FINITE_DECIMAL", field=field_name, message="NaN and infinite values are not supported.", ) ) return decimal_tuple = value.as_tuple() fractional_places = max(-decimal_tuple.exponent, 0) integer_digits = max(value.adjusted() + 1, 0) if value else 0 if ( len(decimal_tuple.digits) > DeterministicPayrollCalculator.MAX_DECIMAL_DIGITS or integer_digits > DeterministicPayrollCalculator.MAX_DECIMAL_DIGITS or fractional_places > DeterministicPayrollCalculator.MAX_FRACTIONAL_PLACES ): issues.append( PayrollIssue( code="DECIMAL_OUT_OF_RANGE", field=field_name, message="Decimal precision or scale exceeds safe limits.", ) ) return below_minimum = minimum is not None and value < minimum equals_exclusive_minimum = ( minimum is not None and exclusive_minimum and value == minimum ) if below_minimum or equals_exclusive_minimum: comparator = "greater than" if exclusive_minimum else "at least" issues.append( PayrollIssue( code="NEGATIVE_OR_INVALID_AMOUNT", field=field_name, message=(f"Value must be {comparator} {decimal_text(minimum)}."), ) )