API Reference

metricengine public API.

This package provides a pluggable metric engine with optional integrations via entry points. It includes a calculation registry, utility formatters with optional Babel support, and a comprehensive unit system for type-safe financial calculations with explicit unit conversions.

Key Features: - Type-safe FinancialValue objects with unit awareness - Comprehensive unit system with Money, Quantity, and Percent units - Conversion registry with multi-hop routing capabilities - Policy-driven conversion behavior (strict vs permissive) - Provenance tracking for unit conversions and calculations - Unit-aware rendering and formatting

class metricengine.FinancialValue(_value, policy=None, unit=<class 'metricengine.units.Dimensionless'>, _is_percentage=False, _prov=None)[source]

Bases: Generic[U]

as_decimal()[source]

Return type:: Decimal | None

as_float()[source]

Return type:: float | None

as_int()[source]

Return type:: int | None

as_percentage()[source]

Convert this FinancialValue to percentage representation with provenance tracking.

Return type:: FinancialValue[Percent]

as_str()[source]

Return type:: str

classmethod constant(value, policy=None, unit=<class 'metricengine.units.Dimensionless'>)[source]: Create a constant FinancialValue with appropriate provenance.

export_provenance_graph()[source]

Export the complete provenance graph for this FinancialValue.

Return type:: dict[str, Any]
Returns:: Dictionary containing the provenance graph in JSON-serializable format

Example

>>> result = FinancialValue(10) + FinancialValue(5)
>>> graph = result.export_provenance_graph()
>>> print(graph['root'])  # Root provenance ID

get_calculation_summary()[source]

Get a brief summary of how this value was calculated.

Return type:: str
Returns:: Brief string summary of the calculation

Example

>>> result = FinancialValue(10) + FinancialValue(5)
>>> print(result.get_calculation_summary())  # "Op: + | Inputs: 2"

get_input_count()[source]

Get the number of inputs that contributed to this FinancialValue.

Return type:: int
Returns:: Number of input values, 0 for literals or values without provenance

Example

>>> result = FinancialValue(10) + FinancialValue(5)
>>> print(result.get_input_count())  # 2

get_inputs()[source]

Get the input provenance IDs that contributed to this FinancialValue.

Return type:: tuple[str, ...]
Returns:: Tuple of provenance IDs for inputs, empty tuple if no provenance

Example

>>> a = FinancialValue(10)
>>> b = FinancialValue(5)
>>> result = a + b
>>> inputs = result.get_inputs()
>>> print(len(inputs))  # 2

get_operation()[source]

Get the operation that created this FinancialValue.

Return type:: str | None
Returns:: Operation string if provenance is available, None otherwise

Example

>>> a = FinancialValue(10)
>>> b = FinancialValue(5)
>>> result = a + b
>>> print(result.get_operation())  # "+"

get_provenance()[source]

Get the provenance record for this FinancialValue.

Return type:: Provenance | None
Returns:: Provenance record if available, None otherwise

get_provenance_id()[source]

Get the unique provenance ID for this FinancialValue.

Return type:: str | None
Returns:: Provenance ID string if available, None otherwise

Example

>>> value = FinancialValue(100)
>>> prov_id = value.get_provenance_id()
>>> print(prov_id[:8])  # First 8 chars of hash

get_provenance_metadata()[source]

Get the metadata associated with this FinancialValue’s provenance.

Return type:: dict[str, Any]
Returns:: Dictionary of metadata, empty dict if no provenance

Example

>>> with calc_span("analysis"):
...     result = FinancialValue(10) + FinancialValue(5)
>>> meta = result.get_provenance_metadata()
>>> print(meta.get("span"))  # "analysis"

has_operation(operation)[source]

Check if this FinancialValue was created by a specific operation.

Parameters:: operation (str) – Operation string to check for (e.g., “+”, “literal”, “calc:margin”)
Return type:: bool
Returns:: True if the operation matches, False otherwise

Example

>>> result = FinancialValue(10) + FinancialValue(5)
>>> print(result.has_operation("+"))  # True
>>> print(result.has_operation("*"))  # False

has_provenance()[source]

Check if this FinancialValue has provenance information.

Return type:: bool
Returns:: True if provenance is available, False otherwise

is_computed()[source]

Check if this FinancialValue was computed from other values.

Return type:: bool
Returns:: True if this value was computed, False if it’s a literal

Example

>>> literal = FinancialValue(100)
>>> computed = FinancialValue(50) + FinancialValue(50)
>>> print(literal.is_computed())   # False
>>> print(computed.is_computed())  # True

is_literal()[source]

Check if this FinancialValue is a literal (not computed from other values).

Return type:: bool
Returns:: True if this is a literal value, False if computed

Example

>>> literal = FinancialValue(100)
>>> computed = FinancialValue(50) + FinancialValue(50)
>>> print(literal.is_literal())   # True
>>> print(computed.is_literal())  # False

is_none()[source]

Return type:: bool

is_percentage()[source]

Return type:: bool

classmethod none(policy=None)[source]

Create a None FinancialValue with appropriate provenance.

Return type:: FinancialValue

classmethod none_with_unit(unit, policy=None)[source]

Create a None FinancialValue with specific unit and appropriate provenance.

Return type:: FinancialValue

policy: Policy | None = None

ratio()[source]

Convert this FinancialValue to ratio representation with provenance tracking.

Return type:: FinancialValue[Ratio]

render(fmt='text', **context)[source]

Render this FinancialValue using a registered renderer.

Parameters:

fmt (str) – Name of the renderer to use (default: “text”)
**context – Additional context passed to the renderer

Return type:

str

Returns:

Rendered string representation

Raises:

KeyError – If the specified renderer is not registered

Example

>>> amount = money(1234.56)
>>> amount.render("html")  # '<span class="fv positive">$1,234.56</span>'
>>> amount.render("html", css_classes="highlight")

to(unit, *, at=None, meta=None)[source]

Convert this FinancialValue to a different unit.

This method performs explicit unit conversion using the registered conversion functions. It creates a new FinancialValue with the target unit and converted value, preserving the original policy and percentage flag.

Parameters:

unit (NewUnit) – Target unit to convert to
at (str | None) – Optional timestamp for rate lookups (e.g., “2025-09-06T10:30:00Z”)
meta (dict[str, str] | None) – Optional metadata dictionary for conversion context (e.g., {“rate”: “0.79”, “source”: “ECB”})

Return type:

FinancialValue

Returns:

New FinancialValue with the target unit and converted value

Raises:

KeyError – If no conversion is registered between the units
ValueError – If this FinancialValue has a None value
TypeError – If this FinancialValue doesn’t use the new unit system

Example

>>> from metricengine import FinancialValue as FV
>>> from metricengine.units import MoneyUnit
>>>
>>> usd = MoneyUnit("USD")
>>> gbp = MoneyUnit("GBP")
>>> amount = FV(100, unit=usd)
>>>
>>> # Convert with default context
>>> converted = amount.to(gbp)
>>>
>>> # Convert with specific rate and timestamp
>>> converted = amount.to(gbp, at="2025-09-06T10:30:00Z",
...                      meta={"rate": "0.79", "source": "ECB"})

trace_calculation(max_depth=10)[source]

Generate a human-readable trace of how this value was calculated.

This method provides a detailed explanation of the calculation chain that led to this FinancialValue, useful for debugging and auditing.

Parameters:: max_depth (int) – Maximum depth to traverse in the calculation tree
Return type:: str
Returns:: Formatted string showing the calculation trace

Example

>>> revenue = FinancialValue(1000)
>>> cost = FinancialValue(600)
>>> profit = revenue - cost
>>> print(profit.trace_calculation())

unit: alias of Dimensionless

with_policy(policy)[source]

Create a new FinancialValue with a different policy, maintaining provenance.

Return type:: FinancialValue

classmethod zero(policy=None, unit=<class 'metricengine.units.Dimensionless'>)[source]: Create a zero FinancialValue with appropriate provenance.

metricengine.FV: alias of FinancialValue

class metricengine.Unit[source]

Bases: object

Base unit class.

class metricengine.Dimensionless[source]

Bases: Unit

Unit for dimensionless values.

class metricengine.Ratio[source]

Bases: Unit

Unit for ratio values.

class metricengine.Percent[source]

Bases: Ratio

Unit for percentage values (inherits from Ratio).

class metricengine.Money[source]

Bases: Unit

Unit for monetary values.

code: str = 'USD'

class metricengine.NewUnit(category, code)[source]

Bases: object

Generic unit with category and code dimensions.

A unit represents a measurement dimension with both a category (the type of measurement) and a specific code (the particular unit within that category). Units are immutable and hashable, making them suitable for use as dictionary keys in conversion registries.

category: The category of measurement (e.g., “Money”, “Quantity”, “Percent”)

code: The specific unit code within the category (e.g., “USD”, “kg”, “ratio”)

Examples

>>> usd = NewUnit("Money", "USD")
>>> str(usd)
'Money[USD]'
>>> kg = NewUnit("Quantity", "kg")
>>> str(kg)
'Quantity[kg]'
>>> ratio = NewUnit("Percent", "ratio")
>>> str(ratio)
'Percent[ratio]'

category: str

code: str

metricengine.MoneyUnit(code)[source]

Create a Money unit with the specified currency code.

Parameters:: code (str) – Currency code (e.g., “USD”, “GBP”, “EUR”)
Return type:: NewUnit
Returns:: NewUnit with “Money” category and the specified code

Example

>>> usd = MoneyUnit("USD")
>>> str(usd)
'Money[USD]'

metricengine.Qty(code)[source]

Create a Quantity unit with the specified unit code.

Parameters:: code (str) – Quantity unit code (e.g., “kg”, “L”, “m”)
Return type:: NewUnit
Returns:: NewUnit with “Quantity” category and the specified code

Example

>>> kg = Qty("kg")
>>> str(kg)
'Quantity[kg]'

metricengine.Pct(code='ratio')[source]

Create a Percent unit with the specified code.

Parameters:: code (str) – Percent unit code, defaults to “ratio”
Return type:: NewUnit
Returns:: NewUnit with “Percent” category and the specified code

Example

>>> ratio = Pct()
>>> str(ratio)
'Percent[ratio]'
>>> bp = Pct("bp")
>>> str(bp)
'Percent[bp]'

class metricengine.ConversionContext(at=None, meta=<factory>)[source]

Bases: object

Context information for unit conversions.

Provides metadata and timing information that conversion functions can use to perform dynamic rate lookups or apply business rules. This allows conversion functions to access external data sources like exchange rate APIs or historical rate databases.

at: Optional timestamp or date string for rate lookups

meta: Dictionary of additional metadata (rates, tenant info, etc.)

Examples

>>> # Simple context with timestamp
>>> ctx = ConversionContext(at="2025-09-06T10:30:00Z")
>>>
>>> # Context with metadata
>>> ctx = ConversionContext(
...     at="2025-09-06",
...     meta={"rate": "0.79", "source": "ECB"}
... )

at: Optional[str] = None

meta: dict[str, str]

class metricengine.ConversionPolicy(strict=True, allow_paths=True)[source]

Bases: object

Policy configuration for unit conversions.

Controls the behavior of the conversion system, including whether to raise errors on missing conversions and whether to allow multi-hop conversion paths through intermediate units.

strict: If True, raise KeyError on missing conversions; if False, return original value

allow_paths: If True, enable multi-hop conversions; if False, only direct conversions

Examples

>>> # Strict policy (default) - raises on missing conversions
>>> strict_policy = ConversionPolicy(strict=True, allow_paths=True)
>>>
>>> # Permissive policy - returns original value on missing conversions
>>> permissive_policy = ConversionPolicy(strict=False, allow_paths=True)
>>>
>>> # Direct-only policy - no multi-hop conversions
>>> direct_only = ConversionPolicy(strict=True, allow_paths=False)

allow_paths: bool = True

strict: bool = True

class metricengine.Conversion(src, dst, fn)[source]

Bases: object

Represents a registered conversion between two units.

Contains the source unit, destination unit, and the function that performs the actual conversion calculation. Conversion functions receive the value to convert and a context object that can provide additional information like exchange rates.

src: Source unit for the conversion

dst: Destination unit for the conversion

fn: Function that performs the conversion, taking (Decimal, ConversionContext) -> Decimal

Examples

>>> usd = MoneyUnit("USD")
>>> gbp = MoneyUnit("GBP")
>>>
>>> def usd_to_gbp(value: Decimal, ctx: ConversionContext) -> Decimal:
...     return value * Decimal("0.79")
>>>
>>> conversion = Conversion(usd, gbp, usd_to_gbp)

src: NewUnit

dst: NewUnit

fn: Callable[[Decimal, ConversionContext], Decimal]

metricengine.register_conversion(src, dst)[source]

Decorator for registering conversion functions between units.

The decorated function must accept a Decimal value and ConversionContext, and return a Decimal result.

Parameters:

src (NewUnit) – Source unit for the conversion
dst (NewUnit) – Destination unit for the conversion

Returns:

Decorator function that registers the conversion

Raises:

ValueError – If the function signature is invalid

Example

>>> usd = MoneyUnit("USD")
>>> gbp = MoneyUnit("GBP")
>>>
>>> @register_conversion(usd, gbp)
... def usd_to_gbp(value: Decimal, ctx: ConversionContext) -> Decimal:
...     # Simple fixed rate for example
...     return value * Decimal("0.79")

metricengine.get_conversion(src, dst)[source]

Get a registered conversion between two units.

Parameters:

src (NewUnit) – Source unit
dst (NewUnit) – Destination unit

Return type:

Conversion

Returns:

Conversion object containing the conversion function

Raises:

KeyError – If no conversion is registered for the unit pair

metricengine.list_conversions()[source]

Get a copy of all registered conversions.

Return type:: dict[tuple[NewUnit, NewUnit], Conversion]
Returns:: Dictionary mapping unit pairs to their conversions

metricengine.convert_decimal(value, src, dst, *, at=None, meta=None)[source]

Convert a decimal value from one unit to another.

This function performs unit-to-unit conversion using the registered conversion functions. It handles same-unit conversions by returning the original value unchanged, and supports multi-hop conversions through intermediate units when no direct conversion exists.

The behavior is controlled by the current ConversionPolicy: - strict=True: Raises KeyError on missing conversions - strict=False: Returns original value on missing conversions - allow_paths=True: Enables multi-hop conversions - allow_paths=False: Only allows direct conversions

Parameters:

value (Decimal) – The decimal value to convert
src (NewUnit) – Source unit
dst (NewUnit) – Destination unit
at (Optional[str]) – Optional timestamp for rate lookups
meta (Optional[dict[str, str]]) – Optional metadata dictionary for conversion context

Return type:

Decimal

Returns:

Converted decimal value, or original value if conversion fails and strict=False

Raises:

KeyError – If no conversion path exists and strict=True
ValueError – If conversion function raises an exception and strict=True

Example

>>> usd = MoneyUnit("USD")
>>> gbp = MoneyUnit("GBP")
>>>
>>> # Strict mode (default) - raises on missing conversion
>>> result = convert_decimal(Decimal("100"), usd, gbp)
>>>
>>> # Permissive mode - returns original on missing conversion
>>> policy = ConversionPolicy(strict=False)
>>> with use_conversions(policy):
...     result = convert_decimal(Decimal("100"), usd, gbp)
...     # Returns Decimal("100") if no conversion exists

metricengine.use_conversions(policy)[source]

Context manager for scoped conversion policy.

Temporarily sets the conversion policy for the duration of the context. The previous policy is restored when the context exits.

Parameters:: policy (ConversionPolicy) – ConversionPolicy to use within the context

Example

>>> permissive_policy = ConversionPolicy(strict=False, allow_paths=True)
>>> with use_conversions(permissive_policy):
...     # Conversions within this block use permissive policy
...     result = convert_decimal(value, usd, gbp)

metricengine.get_current_conversion_policy()[source]

Get the current conversion policy from context.

Return type:: ConversionPolicy
Returns:: Current ConversionPolicy in effect

class metricengine.Policy(decimal_places=2, rounding='ROUND_HALF_UP', none_text='—', percent_display='percent', cap_percentage_at=<factory>, percent_style='percent', quantizer_factory=<function default_quantizer_factory>, negative_sales_is_none=True, compare_none_as_minus_infinity=False, arithmetic_strict=False, thousands_sep=True, currency_symbol=None, currency_position='prefix', negative_parentheses=False, locale=None, display=None)[source]

Bases: object

Immutable configuration for financial calculations and formatting.

arithmetic_strict: bool = False

compare_none_as_minus_infinity: bool = False

currency_position: Literal['prefix', 'suffix'] = 'prefix'

currency_symbol: Optional[str] = None

decimal_places: int = 2

display: Optional[DisplayPolicy] = None

format_decimal(d, unit)[source]

Format a decimal with thousands separators, currency, and negative style.

This method is deprecated and will delegate to the built-in formatter for backward compatibility.

Return type:: str

format_percent(ratio_value)[source]

Render ratio (0..1) as percent text. Always clamp to cap_percentage_at if provided.

Return type:: str

locale: Optional[str] = None

negative_parentheses: bool = False

negative_sales_is_none: bool = True

none_text: str = '—'

percent_display: Literal['ratio', 'percent'] = 'percent'

percent_style: str = 'percent'

quantize(d)[source]

Quantize according to policy. Supports arbitrary step sizes (e.g., 0.5) by rounding to nearest step.

Return type:: Decimal

quantizer_factory()

Exact, fast quantizer for given dp: e.g., dp=2 -> Decimal(‘0.01’).

Return type:: Decimal

rounding: str = 'ROUND_HALF_UP'

thousands_sep: bool = True

cap_percentage_at: Optional[Decimal]

metricengine.get_policy()[source]

May return None if no ambient policy has been set.

Return type:: Optional[Policy]

class metricengine.use_policy(policy)[source]: Bases: object

class metricengine.PolicyResolution(value)[source]

Bases: Enum

CONTEXT = 1

LEFT_OPERAND = 2

STRICT_MATCH = 3

exception metricengine.MetricEngineError[source]

Bases: Exception

Base exception for all Metric Engine errors.

exception metricengine.MissingInputError(message, missing_inputs=None)[source]

Bases: MetricEngineError

Raised when required inputs are missing for a calculation.

exception metricengine.CircularDependencyError(cycle)[source]

Bases: MetricEngineError

Raised when a circular dependency is detected in calculations.

exception metricengine.CalculationError(message, calculation_name=None)[source]

Bases: MetricEngineError

Generic calculation error.

metricengine.calc(name, *, depends_on=())[source]

Decorator to register a calculation function with its dependencies.

Parameters:

name (str) – Unique name for the calculation
depends_on (tuple[str, ...]) – Tuple of calculation names this function depends on

Raises:

CalculationError – If the name is invalid, already registered, or self-dependent.

Return type:

Callable[[Callable[..., Any]], Callable[..., Any]]

metricengine.get(name)[source]

Get a registered calculation function by name.

Return type:: Callable[..., Any]

metricengine.list_calculations()[source]

List all registered calculations and their dependencies (copies).

Return type:: dict[str, set[str]]

metricengine.deps(name)[source]

Get dependencies for a calculation (copy).

Return type:: set[str]

metricengine.is_registered(name)[source]

Check if a calculation is registered.

Return type:: bool

class metricengine.Engine(default_policy=None)[source]

Bases: object

Execution engine for financial calculations.

Builds dependency graphs, detects circular dependencies, caches results, and executes calculations in the correct order.

calculate(name, ctx=None, *, policy=None, allow_partial=False, **kwargs)[source]

Calculate a target metric given a context of input values.

The engine follows a “let calculations validate” philosophy: - None values propagate naturally through calculations - No need for defensive checks before calling calculate - Each calculation determines what inputs are valid - FinancialValue results can be passed directly to other calculations

Parameters:

name (str) – Name of the calculation to compute
ctx (dict[str, Union[int, float, str, Decimal, SupportsFloat, None, FinancialValue]] | None) – Dictionary of input values (optional if using kwargs)
policy (Policy | None) – Optional policy to override default
allow_partial (bool) – If True, return None on failure instead of raising
**kwargs (Union[int, float, str, Decimal, SupportsFloat, None, FinancialValue]) – Input values as keyword arguments (can include None)

Return type:

FinancialValue

Returns:

FinancialValue containing the result (may wrap None)

Raises:

MissingInputError – If required non-None inputs are missing
CircularDependencyError – If circular dependencies are detected
CalculationError – If calculation fails

calculate_many(targets, ctx=None, *, policy=None, allow_partial=False, **kwargs)[source]

Resolve all targets in one pass with shared dependency resolution.

Parameters:

targets (Set of metric names you want)
ctx (Inputs you already have (optional if using kwargs))
policy (Optional Policy override)
allow_partial (If True, return what can be computed and) – leave missing ones out instead of raising.
**kwargs (Input values as keyword arguments)

Return type:

Dictionary mapping metric name to FinancialValue

Raises:

MissingInputError – If any targets cannot be computed (unless allow_partial=True):
CircularDependencyError – If circular dependencies are detected:
CalculationError – If any calculation fails:
Examples: – # Using context dict >>> results = engine.calculate_many( … {“gross_profit”, “gross_margin_percentage”}, … {“sales”: 1000, “cost”: 650} … ) # Using keyword arguments >>> results = engine.calculate_many( … {“gross_profit”, “gross_margin_percentage”}, … sales=1000, cost=650 … )

clear_metric_policy(name)[source]

Return type:: None

constant(value)[source]

Create a constant FinancialValue.

Parameters:: value (int | float | Decimal | None) – The constant value to wrap
Return type:: FinancialValue

get_all_calculations()[source]

Get information about all registered calculations.

Returns:

function: The calculation function
depends_on: Set of dependencies
docstring: The function’s docstring

Return type:

Dict mapping calculation names to their metadata including

get_dependencies(target)[source]

Get all dependencies (direct and transitive) for a calculation.

Parameters:: target (str) – Name of the calculation
Return type:: set[str]
Returns:: Set of all dependency names
Raises:: CircularDependencyError – If circular dependencies detected

none()[source]

Create a None FinancialValue.

Return type:: FinancialValue

set_metric_policy(name, policy)[source]

Return type:: None

validate_dependencies(target)[source]

Validate dependencies for a calculation.

Parameters:: target (str) – Name of the calculation to validate
Return type:: tuple[set[str], set[str]]
Returns:: Tuple of (registered_deps, unregistered_deps)
Raises:: CircularDependencyError – If circular dependencies detected

zero()[source]

Create a zero FinancialValue.

Return type:: FinancialValue

metricengine.format_currency(amount, currency='USD', *, locale=None)[source]

Format currency, using Babel if available and a locale is provided.

Falls back to a simple “1,234.56 USD” style when Babel or locale is not available.

Return type:: str

metricengine.format_percent(value, *, locale=None, precision=2)[source]

Format a percentage, using Babel if available and a locale is provided.

Falls back to a simple “12.34%” style when Babel or locale is not available.

Return type:: str

class metricengine.Renderer(*args, **kwargs)[source]

Bases: Protocol

Protocol for custom FinancialValue renderers.

Renderers must implement a render method that takes a FinancialValue and optional context, returning a string representation.

render(fv, *, context=None)[source]

Render a FinancialValue to a string.

Parameters:

fv (FinancialValue) – The FinancialValue to render
context (dict[str, Any] | None) – Optional context dictionary for rendering customization

Return type:

str

Returns:

String representation of the FinancialValue

metricengine.register_renderer(name, renderer)[source]

Register a renderer with the given name.

Parameters:

name (str) – Unique name for the renderer (e.g., “html”, “markdown”)
renderer (Renderer) – Renderer instance implementing the Renderer protocol

Raises:

TypeError – If renderer doesn’t implement the Renderer protocol

Return type:

None

Example

>>> class CustomRenderer:
...     def render(self, fv, *, context=None):
...         return f"Custom: {fv.as_str()}"
>>> register_renderer("custom", CustomRenderer())

metricengine.get_renderer(name)[source]

Get a registered renderer by name.

Parameters:: name (str) – Name of the renderer to retrieve
Return type:: Renderer
Returns:: The registered renderer instance
Raises:: KeyError – If no renderer is registered with the given name

Example

>>> renderer = get_renderer("html")
>>> output = renderer.render(my_value)

metricengine.list_renderers()[source]

List all registered renderer names.

Return type:: list[str]
Returns:: List of registered renderer names

Example

>>> list_renderers()
['text', 'html', 'markdown']

metricengine.inputs_needed_for(targets)[source]

Determine base inputs needed for the given targets by walking dependencies.

A “base input” is any dependency name that is not a registered calculation. Registered calculations that have no dependencies are not counted as inputs.

Return type:: set[str]

class metricengine.CalculationService(policy=None)[source]

Bases: object

Utility class for calculation services.

metricengine.get_nulls()[source]

Return the currently active null behavior.

Return type:: NullBehavior

class metricengine.NullBinaryMode(value)[source]

Bases: Enum

How None is handled in binary ops (a ⊕ b).

PROPAGATE = 1

RAISE = 2

class metricengine.EqualityMode(value)[source]

Bases: Enum

VALUE_ONLY = 1

VALUE_AND_UNIT = 2

VALUE_UNIT_AND_POLICY = 3

metricengine.load_plugins(context=None)[source]

Load all available plugins with optional context.

Return type:: int

metricengine.get_calc(name)[source]

Get a registered calculation function by name.

Parameters:: name (str) – The name of the calculation to retrieve
Return type:: Callable[..., Any]
Returns:: The calculation function
Raises:: CalculationError – If the calculation is not found or name is invalid

Example

>>> cogs = get_calc("cogs")
>>> result = cogs(opening_inventory=FV(100), purchases=FV(150), closing_inventory=FV(100))

metricengine.calc_span(name, **attrs)[source]

Context manager for grouping calculations under a named span.

This context manager allows grouping related financial calculations under a named span, which will be included in the provenance metadata of all operations performed within the span context.

Parameters:

name (str) – Name of the calculation span
**attrs – Additional attributes to associate with the span

Yields:

None

Return type:

Generator[None, None, None]

Example

>>> with calc_span("quarterly_analysis", quarter="Q1", year=2024):
...     revenue = FinancialValue(1000)
...     cost = FinancialValue(600)
...     profit = revenue - cost  # Will include span info in provenance

>>> prov = profit.get_provenance()
>>> print(prov.meta.get("span"))  # "quarterly_analysis"
>>> print(prov.meta.get("span_attrs"))  # {"quarter": "Q1", "year": 2024}

metricengine.explain(fv, max_depth=10)[source]

Generate human-readable explanation of calculation.

This function creates a formatted text representation of how a FinancialValue was calculated, showing the operation tree in a readable format. This is useful for debugging and understanding complex calculations.

Parameters:

fv (FinancialValue) – FinancialValue to explain
max_depth (int) – Maximum depth to traverse (prevents infinite recursion)

Return type:

str

Returns:

Human-readable string explaining the calculation

Example

>>> revenue = FinancialValue(1000)
>>> cost = FinancialValue(600)
>>> profit = revenue - cost
>>> print(explain(profit))
# Result (400.00):
#   Operation: -
#   Left: 1000.00 (literal)
#   Right: 600.00 (literal)

metricengine.get_provenance_graph(fv)[source]

Extract complete provenance graph as dictionary.

This function traverses the complete provenance graph starting from the given FinancialValue and returns a dictionary mapping provenance IDs to their Provenance records. This is useful for analysis and debugging of calculation lineage.

Note: This implementation can only traverse the provenance records that are directly accessible from the root FinancialValue. In the current architecture, we don’t maintain a global provenance store, so we can only include the root provenance record. A full implementation would require either: 1. A global provenance registry, or 2. Maintaining references to parent FinancialValue instances

Parameters:: fv (FinancialValue) – FinancialValue to extract provenance graph from
Return type:: dict[str, Provenance]
Returns:: Dictionary mapping provenance IDs to Provenance records

Example

>>> revenue = FinancialValue(1000)
>>> cost = FinancialValue(600)
>>> profit = revenue - cost
>>> graph = get_provenance_graph(profit)
>>> print(len(graph))  # 1 (only profit, as we can't traverse to inputs)
>>> print(list(graph.keys()))  # ['profit_id']

metricengine.to_trace_json(fv)[source]

Export complete provenance graph as JSON-serializable dictionary.

This function creates a complete JSON representation of the provenance graph that can be serialized, stored, or transmitted. The format includes a root node identifier and a nodes dictionary containing all provenance records.

Parameters:: fv (FinancialValue) – FinancialValue to export provenance graph from
Return type:: dict[str, Any]
Returns:: Dictionary with ‘root’ and ‘nodes’ keys containing the complete graph

Example

>>> revenue = FinancialValue(1000)
>>> cost = FinancialValue(600)
>>> profit = revenue - cost
>>> trace = to_trace_json(profit)
>>> print(trace['root'])  # profit provenance ID
>>> print(len(trace['nodes']))  # 3 nodes

Formatting

metricengine.formatting.format_currency(amount, currency='USD', *, locale=None)[source]

Format currency, using Babel if available and a locale is provided.

Falls back to a simple “1,234.56 USD” style when Babel or locale is not available.

Return type:: str

metricengine.formatting.format_percent(value, *, locale=None, precision=2)[source]

Format a percentage, using Babel if available and a locale is provided.

Falls back to a simple “12.34%” style when Babel or locale is not available.

Return type:: str

Registry

Registry for financial calculations with dependency tracking.

metricengine.registry.calc(name, *, depends_on=())[source]

Decorator to register a calculation function with its dependencies.

Parameters:

name (str) – Unique name for the calculation
depends_on (tuple[str, ...]) – Tuple of calculation names this function depends on

Raises:

CalculationError – If the name is invalid, already registered, or self-dependent.

Return type:

Callable[[Callable[..., Any]], Callable[..., Any]]

metricengine.registry.get(name)[source]

Get a registered calculation function by name.

Return type:: Callable[..., Any]

metricengine.registry.deps(name)[source]

Get dependencies for a calculation (copy).

Return type:: set[str]

metricengine.registry.list_calculations()[source]

List all registered calculations and their dependencies (copies).

Return type:: dict[str, set[str]]

metricengine.registry.clear_registry()[source]

Clear all registered calculations. Primarily for testing.

Return type:: None

metricengine.registry.is_registered(name)[source]

Check if a calculation is registered.

Return type:: bool

metricengine.registry.unregister(name)[source]

Remove a calculation from the registry (and its edges).

Return type:: None

metricengine.registry.dependency_graph()[source]

Get a read-only view of the dependency graph (copies of sets).

Return type:: Mapping[str, set[str]]

metricengine.registry.detect_cycles()[source]

Return a set of cycles detected in the dependency graph (as tuples). Simple DFS; fine for small graphs.

Return type:: set[tuple[str, ...]]