zwitschi
d9fd82b2e3
- Added CapexComponentInput, CapexParameters, CapexCalculationRequest, CapexCalculationResult, and related schemas for capex calculations. - Introduced calculate_initial_capex function to aggregate capex components and compute totals and timelines. - Created ProjectCapexRepository and ScenarioCapexRepository for managing capex snapshots in the database. - Developed capex.html template for capturing and displaying initial capex data. - Registered common Jinja2 filters for formatting currency and percentages. - Implemented unit and integration tests for capex calculation functionality. - Updated unit of work to include new repositories for capex management.
336 lines
10 KiB
Python
336 lines
10 KiB
Python
"""Service functions for financial calculations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
|
|
from services.currency import CurrencyValidationError, normalise_currency
|
|
from services.exceptions import CapexValidationError, 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,
|
|
ProfitabilityCalculationRequest,
|
|
ProfitabilityCalculationResult,
|
|
ProfitabilityCosts,
|
|
ProfitabilityMetrics,
|
|
)
|
|
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
__all__ = ["calculate_profitability", "calculate_initial_capex"]
|