zwitschi
1feae7ff85
- Introduced OpexValidationError for handling validation errors in processing opex calculations. - Implemented ProjectProcessingOpexRepository and ScenarioProcessingOpexRepository for managing project and scenario-level processing opex snapshots. - Enhanced UnitOfWork to include repositories for processing opex. - Updated sidebar navigation and scenario detail templates to include links to the new Processing Opex Planner. - Created a new template for the Processing Opex Planner with form handling for input components and parameters. - Developed integration tests for processing opex calculations, covering HTML and JSON flows, including validation for currency mismatches and unsupported frequencies. - Added unit tests for the calculation logic, ensuring correct handling of various scenarios and edge cases.
537 lines
17 KiB
Python
537 lines
17 KiB
Python
"""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,
|
|
ProcessingOpexCalculationRequest,
|
|
ProcessingOpexCalculationResult,
|
|
ProcessingOpexCategoryBreakdown,
|
|
ProcessingOpexComponentInput,
|
|
ProcessingOpexMetrics,
|
|
ProcessingOpexParameters,
|
|
ProcessingOpexTotals,
|
|
ProcessingOpexTimelineEntry,
|
|
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,
|
|
initial_capex: float,
|
|
) -> tuple[list[CashFlow], list[CashFlowEntry]]:
|
|
"""Create cash flow structures for financial metric calculations."""
|
|
|
|
cash_flow_models: list[CashFlow] = [
|
|
CashFlow(amount=-initial_capex, period_index=0)
|
|
]
|
|
cash_flow_entries: list[CashFlowEntry] = [
|
|
CashFlowEntry(
|
|
period=0,
|
|
revenue=0.0,
|
|
processing_opex=0.0,
|
|
sustaining_capex=0.0,
|
|
net=-initial_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,
|
|
processing_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.processing_opex) * periods
|
|
sustaining_total = float(request.sustaining_capex) * periods
|
|
initial_capex = float(request.initial_capex)
|
|
|
|
net_per_period = (
|
|
revenue_per_period
|
|
- float(request.processing_opex)
|
|
- float(request.sustaining_capex)
|
|
)
|
|
|
|
cash_flow_models, cash_flow_entries = _generate_cash_flows(
|
|
periods=periods,
|
|
net_per_period=net_per_period,
|
|
initial_capex=initial_capex,
|
|
)
|
|
|
|
# Update per-period entries to include explicit costs for presentation
|
|
for entry in cash_flow_entries[1:]:
|
|
entry.revenue = revenue_per_period
|
|
entry.processing_opex = float(request.processing_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 + initial_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(
|
|
processing_opex_total=processing_total,
|
|
sustaining_capex_total=sustaining_total,
|
|
initial_capex=initial_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_processing_opex(
|
|
request: ProcessingOpexCalculationRequest,
|
|
) -> ProcessingOpexCalculationResult:
|
|
"""Aggregate processing opex components into annual totals and timeline."""
|
|
|
|
if not request.components:
|
|
raise OpexValidationError(
|
|
"At least one processing opex component is required for calculation.",
|
|
["components"],
|
|
)
|
|
|
|
parameters: ProcessingOpexParameters = 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[ProcessingOpexComponentInput] = []
|
|
|
|
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(
|
|
ProcessingOpexComponentInput(
|
|
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[ProcessingOpexTimelineEntry] = []
|
|
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(
|
|
ProcessingOpexTimelineEntry(
|
|
period=period,
|
|
base_cost=base_cost,
|
|
escalated_cost=escalated_cost if apply_escalation else None,
|
|
)
|
|
)
|
|
escalated_values.append(escalated_cost)
|
|
|
|
category_breakdowns: list[ProcessingOpexCategoryBreakdown] = []
|
|
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(
|
|
ProcessingOpexCategoryBreakdown(
|
|
category=category,
|
|
annual_cost=total,
|
|
share=share,
|
|
)
|
|
)
|
|
|
|
metrics = ProcessingOpexMetrics(
|
|
annual_average=fmean(escalated_values) if escalated_values else None,
|
|
cost_per_ton=None,
|
|
)
|
|
|
|
totals = ProcessingOpexTotals(
|
|
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 ProcessingOpexCalculationResult(
|
|
totals=totals,
|
|
timeline=timeline_entries,
|
|
metrics=metrics,
|
|
components=normalised_components,
|
|
parameters=parameters,
|
|
options=request.options,
|
|
currency=currency,
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"calculate_profitability",
|
|
"calculate_initial_capex",
|
|
"calculate_processing_opex",
|
|
]
|