"""Service functions for financial calculations.""" from __future__ import annotations from collections import defaultdict from statistics import fmean from services.currency import CurrencyValidationError, normalise_currency from services.exceptions import ( CapexValidationError, OpexValidationError, ProfitabilityValidationError, ) from services.financial import ( CashFlow, ConvergenceError, PaybackNotReachedError, internal_rate_of_return, net_present_value, payback_period, ) from services.pricing import PricingInput, PricingMetadata, PricingResult, calculate_pricing from schemas.calculations import ( CapexCalculationRequest, CapexCalculationResult, CapexCategoryBreakdown, CapexComponentInput, CapexParameters, CapexTotals, CapexTimelineEntry, CashFlowEntry, OpexCalculationRequest, OpexCalculationResult, OpexCategoryBreakdown, OpexComponentInput, OpexMetrics, OpexParameters, OpexTotals, OpexTimelineEntry, ProfitabilityCalculationRequest, ProfitabilityCalculationResult, ProfitabilityCosts, ProfitabilityMetrics, ) _FREQUENCY_MULTIPLIER = { "daily": 365, "weekly": 52, "monthly": 12, "quarterly": 4, "annually": 1, } def _build_pricing_input( request: ProfitabilityCalculationRequest, ) -> PricingInput: """Construct a pricing input instance including impurity overrides.""" impurity_values: dict[str, float] = {} impurity_thresholds: dict[str, float] = {} impurity_penalties: dict[str, float] = {} for impurity in request.impurities: code = impurity.name.strip() if not code: continue code = code.upper() if impurity.value is not None: impurity_values[code] = float(impurity.value) if impurity.threshold is not None: impurity_thresholds[code] = float(impurity.threshold) if impurity.penalty is not None: impurity_penalties[code] = float(impurity.penalty) pricing_input = PricingInput( metal=request.metal, ore_tonnage=request.ore_tonnage, head_grade_pct=request.head_grade_pct, recovery_pct=request.recovery_pct, payable_pct=request.payable_pct, reference_price=request.reference_price, treatment_charge=request.treatment_charge, smelting_charge=request.smelting_charge, moisture_pct=request.moisture_pct, moisture_threshold_pct=request.moisture_threshold_pct, moisture_penalty_per_pct=request.moisture_penalty_per_pct, impurity_ppm=impurity_values, impurity_thresholds=impurity_thresholds, impurity_penalty_per_ppm=impurity_penalties, premiums=request.premiums, fx_rate=request.fx_rate, currency_code=request.currency_code, ) return pricing_input def _generate_cash_flows( *, periods: int, net_per_period: float, capex: float, ) -> tuple[list[CashFlow], list[CashFlowEntry]]: """Create cash flow structures for financial metric calculations.""" cash_flow_models: list[CashFlow] = [ CashFlow(amount=-capex, period_index=0) ] cash_flow_entries: list[CashFlowEntry] = [ CashFlowEntry( period=0, revenue=0.0, opex=0.0, sustaining_capex=0.0, net=-capex, ) ] for period in range(1, periods + 1): cash_flow_models.append( CashFlow(amount=net_per_period, period_index=period)) cash_flow_entries.append( CashFlowEntry( period=period, revenue=0.0, opex=0.0, sustaining_capex=0.0, net=net_per_period, ) ) return cash_flow_models, cash_flow_entries def calculate_profitability( request: ProfitabilityCalculationRequest, *, metadata: PricingMetadata, ) -> ProfitabilityCalculationResult: """Calculate profitability metrics using pricing inputs and cost data.""" if request.periods <= 0: raise ProfitabilityValidationError( "Evaluation periods must be at least 1.", ["periods"] ) pricing_input = _build_pricing_input(request) try: pricing_result: PricingResult = calculate_pricing( pricing_input, metadata=metadata ) except CurrencyValidationError as exc: raise ProfitabilityValidationError( str(exc), ["currency_code"]) from exc periods = request.periods revenue_total = float(pricing_result.net_revenue) revenue_per_period = revenue_total / periods processing_total = float(request.opex) * periods sustaining_total = float(request.sustaining_capex) * periods capex = float(request.capex) net_per_period = ( revenue_per_period - float(request.opex) - float(request.sustaining_capex) ) cash_flow_models, cash_flow_entries = _generate_cash_flows( periods=periods, net_per_period=net_per_period, capex=capex, ) # Update per-period entries to include explicit costs for presentation for entry in cash_flow_entries[1:]: entry.revenue = revenue_per_period entry.opex = float(request.opex) entry.sustaining_capex = float(request.sustaining_capex) entry.net = net_per_period discount_rate = (request.discount_rate or 0.0) / 100.0 npv_value = net_present_value(discount_rate, cash_flow_models) try: irr_value = internal_rate_of_return(cash_flow_models) * 100.0 except (ValueError, ZeroDivisionError, ConvergenceError): irr_value = None try: payback_value = payback_period(cash_flow_models) except (ValueError, PaybackNotReachedError): payback_value = None total_costs = processing_total + sustaining_total + capex total_net = revenue_total - total_costs if revenue_total == 0: margin_value = None else: margin_value = (total_net / revenue_total) * 100.0 currency = request.currency_code or pricing_result.currency try: currency = normalise_currency(currency) except CurrencyValidationError as exc: raise ProfitabilityValidationError( str(exc), ["currency_code"]) from exc costs = ProfitabilityCosts( opex_total=processing_total, sustaining_capex_total=sustaining_total, capex=capex, ) metrics = ProfitabilityMetrics( npv=npv_value, irr=irr_value, payback_period=payback_value, margin=margin_value, ) return ProfitabilityCalculationResult( pricing=pricing_result, costs=costs, metrics=metrics, cash_flows=cash_flow_entries, currency=currency, ) def calculate_initial_capex( request: CapexCalculationRequest, ) -> CapexCalculationResult: """Aggregate capex components into totals and timelines.""" if not request.components: raise CapexValidationError( "At least one capex component is required for calculation.", ["components"], ) parameters = request.parameters base_currency = parameters.currency_code if base_currency: try: base_currency = normalise_currency(base_currency) except CurrencyValidationError as exc: raise CapexValidationError( str(exc), ["parameters.currency_code"] ) from exc overall = 0.0 category_totals: dict[str, float] = defaultdict(float) timeline_totals: dict[int, float] = defaultdict(float) normalised_components: list[CapexComponentInput] = [] for index, component in enumerate(request.components): amount = float(component.amount) overall += amount category_totals[component.category] += amount spend_year = component.spend_year or 0 timeline_totals[spend_year] += amount component_currency = component.currency if component_currency: try: component_currency = normalise_currency(component_currency) except CurrencyValidationError as exc: raise CapexValidationError( str(exc), [f"components[{index}].currency"] ) from exc if base_currency is None and component_currency: base_currency = component_currency elif ( base_currency is not None and component_currency is not None and component_currency != base_currency ): raise CapexValidationError( ( "Component currency does not match the global currency. " f"Expected {base_currency}, got {component_currency}." ), [f"components[{index}].currency"], ) normalised_components.append( CapexComponentInput( id=component.id, name=component.name, category=component.category, amount=amount, currency=component_currency, spend_year=component.spend_year, notes=component.notes, ) ) contingency_pct = float(parameters.contingency_pct or 0.0) contingency_amount = overall * (contingency_pct / 100.0) grand_total = overall + contingency_amount category_breakdowns: list[CapexCategoryBreakdown] = [] if category_totals: for category, total in sorted(category_totals.items()): share = (total / overall * 100.0) if overall else None category_breakdowns.append( CapexCategoryBreakdown( category=category, amount=total, share=share, ) ) cumulative = 0.0 timeline_entries: list[CapexTimelineEntry] = [] for year, spend in sorted(timeline_totals.items()): cumulative += spend timeline_entries.append( CapexTimelineEntry(year=year, spend=spend, cumulative=cumulative) ) try: currency = normalise_currency(base_currency) if base_currency else None except CurrencyValidationError as exc: raise CapexValidationError( str(exc), ["parameters.currency_code"] ) from exc totals = CapexTotals( overall=overall, contingency_pct=contingency_pct, contingency_amount=contingency_amount, with_contingency=grand_total, by_category=category_breakdowns, ) return CapexCalculationResult( totals=totals, timeline=timeline_entries, components=normalised_components, parameters=parameters, options=request.options, currency=currency, ) def calculate_opex( request: OpexCalculationRequest, ) -> OpexCalculationResult: """Aggregate opex components into annual totals and timeline.""" if not request.components: raise OpexValidationError( "At least one opex component is required for calculation.", ["components"], ) parameters: OpexParameters = request.parameters base_currency = parameters.currency_code if base_currency: try: base_currency = normalise_currency(base_currency) except CurrencyValidationError as exc: raise OpexValidationError( str(exc), ["parameters.currency_code"] ) from exc evaluation_horizon = parameters.evaluation_horizon_years or 1 if evaluation_horizon <= 0: raise OpexValidationError( "Evaluation horizon must be at least 1 year.", ["parameters.evaluation_horizon_years"], ) escalation_pct = float(parameters.escalation_pct or 0.0) apply_escalation = bool(parameters.apply_escalation) category_totals: dict[str, float] = defaultdict(float) timeline_totals: dict[int, float] = defaultdict(float) timeline_escalated: dict[int, float] = defaultdict(float) normalised_components: list[OpexComponentInput] = [] max_period_end = evaluation_horizon for index, component in enumerate(request.components): frequency = component.frequency.lower() multiplier = _FREQUENCY_MULTIPLIER.get(frequency) if multiplier is None: raise OpexValidationError( f"Unsupported frequency '{component.frequency}'.", [f"components[{index}].frequency"], ) unit_cost = float(component.unit_cost) quantity = float(component.quantity) annual_cost = unit_cost * quantity * multiplier period_start = component.period_start or 1 period_end = component.period_end or evaluation_horizon if period_end < period_start: raise OpexValidationError( ( "Component period_end must be greater than or equal to " "period_start." ), [f"components[{index}].period_end"], ) max_period_end = max(max_period_end, period_end) component_currency = component.currency if component_currency: try: component_currency = normalise_currency(component_currency) except CurrencyValidationError as exc: raise OpexValidationError( str(exc), [f"components[{index}].currency"] ) from exc if base_currency is None and component_currency: base_currency = component_currency elif ( base_currency is not None and component_currency is not None and component_currency != base_currency ): raise OpexValidationError( ( "Component currency does not match the global currency. " f"Expected {base_currency}, got {component_currency}." ), [f"components[{index}].currency"], ) category_totals[component.category] += annual_cost for period in range(period_start, period_end + 1): timeline_totals[period] += annual_cost normalised_components.append( OpexComponentInput( id=component.id, name=component.name, category=component.category, unit_cost=unit_cost, quantity=quantity, frequency=frequency, currency=component_currency, period_start=period_start, period_end=period_end, notes=component.notes, ) ) evaluation_horizon = max(evaluation_horizon, max_period_end) try: currency = normalise_currency(base_currency) if base_currency else None except CurrencyValidationError as exc: raise OpexValidationError( str(exc), ["parameters.currency_code"] ) from exc timeline_entries: list[OpexTimelineEntry] = [] escalated_values: list[float] = [] overall_annual = timeline_totals.get(1, 0.0) escalated_total = 0.0 for period in range(1, evaluation_horizon + 1): base_cost = timeline_totals.get(period, 0.0) if apply_escalation: factor = (1 + escalation_pct / 100.0) ** (period - 1) else: factor = 1.0 escalated_cost = base_cost * factor timeline_escalated[period] = escalated_cost escalated_total += escalated_cost timeline_entries.append( OpexTimelineEntry( period=period, base_cost=base_cost, escalated_cost=escalated_cost if apply_escalation else None, ) ) escalated_values.append(escalated_cost) category_breakdowns: list[OpexCategoryBreakdown] = [] total_base = sum(category_totals.values()) for category, total in sorted(category_totals.items()): share = (total / total_base * 100.0) if total_base else None category_breakdowns.append( OpexCategoryBreakdown( category=category, annual_cost=total, share=share, ) ) metrics = OpexMetrics( annual_average=fmean(escalated_values) if escalated_values else None, cost_per_ton=None, ) totals = OpexTotals( overall_annual=overall_annual, escalated_total=escalated_total if apply_escalation else None, escalation_pct=escalation_pct if apply_escalation else None, by_category=category_breakdowns, ) return OpexCalculationResult( totals=totals, timeline=timeline_entries, metrics=metrics, components=normalised_components, parameters=parameters, options=request.options, currency=currency, ) __all__ = [ "calculate_profitability", "calculate_initial_capex", "calculate_opex", ]