metricengine.engine

Execution engine for financial calculations with DAG resolution.

class metricengine.engine.Engine(default_policy=None)

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)

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

constant(value)

Create a constant FinancialValue.

Parameters:

value (int | float | Decimal | None) – The constant value to wrap

Return type:

FinancialValue

zero()

Create a zero FinancialValue.

Return type:

FinancialValue

none()

Create a None FinancialValue.

Return type:

FinancialValue

calculate_many(targets, ctx=None, *, policy=None, allow_partial=False, **kwargs)

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 … )

set_metric_policy(name, policy)

Return type:

None

clear_metric_policy(name)

Return type:

None

get_dependencies(target)

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

validate_dependencies(target)

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

get_all_calculations()

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

Engine Class

class metricengine.engine.Engine(default_policy=None)

Bases: object

Execution engine for financial calculations.

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

__init__(default_policy=None)

Initialize the engine with an optional default policy.

Parameters:

default_policy (Policy | None) – Default policy for calculations. Uses DEFAULT_POLICY if None.

calculate(name, ctx=None, *, policy=None, allow_partial=False, **kwargs)

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

constant(value)

Create a constant FinancialValue.

Parameters:

value (int | float | Decimal | None) – The constant value to wrap

Return type:

FinancialValue

zero()

Create a zero FinancialValue.

Return type:

FinancialValue

none()

Create a None FinancialValue.

Return type:

FinancialValue

calculate_many(targets, ctx=None, *, policy=None, allow_partial=False, **kwargs)

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 … )

set_metric_policy(name, policy)

Return type:

None

clear_metric_policy(name)

Return type:

None

get_dependencies(target)

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

validate_dependencies(target)

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

get_all_calculations()

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

Engine Functions

Engine Configuration

Global Engine