zwitschi
795a9f99f4
- Updated form template to prefill currency input with default value and added help text for clarity. - Modified integration tests to assert more descriptive error messages for invalid currency codes. - Introduced new tests for currency normalization and validation in various scenarios, including imports and exports. - Added comprehensive tests for pricing calculations, ensuring defaults are respected and overrides function correctly. - Implemented unit tests for pricing settings repository, ensuring CRUD operations and default settings are handled properly. - Enhanced scenario pricing evaluation tests to validate currency handling and metadata defaults. - Added simulation tests to ensure Monte Carlo runs are accurate and handle various distribution scenarios.
177 lines
6.3 KiB
Python
177 lines
6.3 KiB
Python
from __future__ import annotations
|
|
|
|
"""Pricing service implementing commodity revenue calculations.
|
|
|
|
This module exposes data models and helpers for computing product pricing
|
|
according to the formulas outlined in
|
|
``calminer-docs/specifications/price_calculation.md``. It focuses on the core
|
|
calculation steps (payable metal, penalties, net revenue) and is intended to be
|
|
composed within broader scenario evaluation workflows.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Mapping
|
|
|
|
from pydantic import BaseModel, Field, PositiveFloat, field_validator
|
|
from services.currency import require_currency
|
|
|
|
|
|
class PricingInput(BaseModel):
|
|
"""Normalized inputs for pricing calculations."""
|
|
|
|
metal: str = Field(..., min_length=1)
|
|
ore_tonnage: PositiveFloat = Field(
|
|
..., description="Total ore mass processed (metric tonnes)")
|
|
head_grade_pct: PositiveFloat = Field(..., gt=0,
|
|
le=100, description="Head grade as percent")
|
|
recovery_pct: PositiveFloat = Field(..., gt=0,
|
|
le=100, description="Recovery rate percent")
|
|
payable_pct: float | None = Field(
|
|
None, gt=0, le=100, description="Contractual payable percentage")
|
|
reference_price: PositiveFloat = Field(
|
|
..., description="Reference price in base currency per unit")
|
|
treatment_charge: float = Field(0, ge=0)
|
|
smelting_charge: float = Field(0, ge=0)
|
|
moisture_pct: float = Field(0, ge=0, le=100)
|
|
moisture_threshold_pct: float | None = Field(None, ge=0, le=100)
|
|
moisture_penalty_per_pct: float | None = Field(None)
|
|
impurity_ppm: Mapping[str, float] = Field(default_factory=dict)
|
|
impurity_thresholds: Mapping[str, float] = Field(default_factory=dict)
|
|
impurity_penalty_per_ppm: Mapping[str, float] = Field(default_factory=dict)
|
|
premiums: float = Field(0)
|
|
fx_rate: PositiveFloat = Field(
|
|
1, description="Multiplier to convert to scenario currency")
|
|
currency_code: str | None = Field(
|
|
None, description="Optional explicit currency override")
|
|
|
|
@field_validator("impurity_ppm", mode="before")
|
|
@classmethod
|
|
def _validate_impurity_mapping(cls, value):
|
|
if isinstance(value, Mapping):
|
|
return {k: float(v) for k, v in value.items()}
|
|
return value
|
|
|
|
|
|
class PricingResult(BaseModel):
|
|
"""Structured output summarising pricing computation results."""
|
|
|
|
metal: str
|
|
ore_tonnage: float
|
|
head_grade_pct: float
|
|
recovery_pct: float
|
|
payable_metal_tonnes: float
|
|
reference_price: float
|
|
gross_revenue: float
|
|
moisture_penalty: float
|
|
impurity_penalty: float
|
|
treatment_smelt_charges: float
|
|
premiums: float
|
|
net_revenue: float
|
|
currency: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PricingMetadata:
|
|
"""Metadata defaults applied when explicit inputs are omitted."""
|
|
|
|
default_payable_pct: float = 100.0
|
|
default_currency: str | None = "USD"
|
|
moisture_threshold_pct: float = 8.0
|
|
moisture_penalty_per_pct: float = 0.0
|
|
impurity_thresholds: Mapping[str, float] = field(default_factory=dict)
|
|
impurity_penalty_per_ppm: Mapping[str, float] = field(default_factory=dict)
|
|
|
|
|
|
def calculate_pricing(
|
|
pricing_input: PricingInput,
|
|
*,
|
|
metadata: PricingMetadata | None = None,
|
|
currency: str | None = None,
|
|
) -> PricingResult:
|
|
"""Calculate pricing metrics for the provided commodity input.
|
|
|
|
Parameters
|
|
----------
|
|
pricing_input:
|
|
Normalised input data including ore tonnage, grades, charges, and
|
|
optional penalties.
|
|
metadata:
|
|
Optional default metadata applied when specific values are omitted from
|
|
``pricing_input``.
|
|
currency:
|
|
Optional override for the output currency label. Falls back to
|
|
``metadata.default_currency`` when not provided.
|
|
"""
|
|
|
|
applied_metadata = metadata or PricingMetadata()
|
|
|
|
payable_pct = (
|
|
pricing_input.payable_pct
|
|
if pricing_input.payable_pct is not None
|
|
else applied_metadata.default_payable_pct
|
|
)
|
|
moisture_threshold = (
|
|
pricing_input.moisture_threshold_pct
|
|
if pricing_input.moisture_threshold_pct is not None
|
|
else applied_metadata.moisture_threshold_pct
|
|
)
|
|
moisture_penalty_factor = (
|
|
pricing_input.moisture_penalty_per_pct
|
|
if pricing_input.moisture_penalty_per_pct is not None
|
|
else applied_metadata.moisture_penalty_per_pct
|
|
)
|
|
|
|
impurity_thresholds = {
|
|
**applied_metadata.impurity_thresholds,
|
|
**pricing_input.impurity_thresholds,
|
|
}
|
|
impurity_penalty_factors = {
|
|
**applied_metadata.impurity_penalty_per_ppm,
|
|
**pricing_input.impurity_penalty_per_ppm,
|
|
}
|
|
|
|
q_metal = pricing_input.ore_tonnage * (pricing_input.head_grade_pct / 100.0) * (
|
|
pricing_input.recovery_pct / 100.0
|
|
)
|
|
payable_metal = q_metal * (payable_pct / 100.0)
|
|
|
|
gross_revenue_ref = payable_metal * pricing_input.reference_price
|
|
charges = pricing_input.treatment_charge + pricing_input.smelting_charge
|
|
|
|
moisture_excess = max(0.0, pricing_input.moisture_pct - moisture_threshold)
|
|
moisture_penalty = moisture_excess * moisture_penalty_factor
|
|
|
|
impurity_penalty_total = 0.0
|
|
for impurity, value in pricing_input.impurity_ppm.items():
|
|
threshold = impurity_thresholds.get(impurity, 0.0)
|
|
penalty_factor = impurity_penalty_factors.get(impurity, 0.0)
|
|
impurity_penalty_total += max(0.0, value - threshold) * penalty_factor
|
|
|
|
net_revenue_ref = (
|
|
gross_revenue_ref - charges - moisture_penalty - impurity_penalty_total
|
|
)
|
|
net_revenue_ref += pricing_input.premiums
|
|
|
|
net_revenue = net_revenue_ref * pricing_input.fx_rate
|
|
|
|
currency_code = require_currency(
|
|
currency or pricing_input.currency_code,
|
|
default=applied_metadata.default_currency,
|
|
)
|
|
|
|
return PricingResult(
|
|
metal=pricing_input.metal,
|
|
ore_tonnage=pricing_input.ore_tonnage,
|
|
head_grade_pct=pricing_input.head_grade_pct,
|
|
recovery_pct=pricing_input.recovery_pct,
|
|
payable_metal_tonnes=payable_metal,
|
|
reference_price=pricing_input.reference_price,
|
|
gross_revenue=gross_revenue_ref,
|
|
moisture_penalty=moisture_penalty,
|
|
impurity_penalty=impurity_penalty_total,
|
|
treatment_smelt_charges=charges,
|
|
premiums=pricing_input.premiums,
|
|
net_revenue=net_revenue,
|
|
currency=currency_code,
|
|
)
|