strategies

Computation and Sampling Strategies

This module defines strategies for computing distribution characteristics and generating random samples.

class pysatl_core.distributions.strategies.ComputationStep(target, sources, edge_kind, method_name, options_descriptor=<factory>)[source]

Bases: object

One step of a strategy’s execution plan for a target characteristic.

Parameters:

  • target (GenericCharacteristicName)

  • sources (tuple[GenericCharacteristicName, ...])

  • edge_kind (str)

  • method_name (str)

  • options_descriptor (EdgeOptionsDescriptor)

target

Characteristic produced by this step.

Type:

GenericCharacteristicName

sources

Characteristics consumed by this step.

Type:

tuple[GenericCharacteristicName, ]

edge_kind

Underlying edge kind: "analytical_loop", "transformation_loop" or "computation".

Type:

str

method_name

Human-readable identifier of the underlying method (descriptor name when available, otherwise target).

Type:

str

options_descriptor

Compact descriptor describing which user-supplied options will be consumed at this step. Empty for self-loop steps.

Type:

EdgeOptionsDescriptor

target: TypeAliasType
sources: tuple[TypeAliasType, ...]
edge_kind: str
method_name: str
options_descriptor: EdgeOptionsDescriptor
__init__(target, sources, edge_kind, method_name, options_descriptor=<factory>)
Parameters:

  • target (GenericCharacteristicName)

  • sources (tuple[GenericCharacteristicName, ...])

  • edge_kind (str)

  • method_name (str)

  • options_descriptor (EdgeOptionsDescriptor)

Return type:

None

class pysatl_core.distributions.strategies.ComputationPlan(target, source, steps)[source]

Bases: object

Plan describing how a strategy will compute target for a distribution.

Parameters:

  • target (GenericCharacteristicName)

  • source (GenericCharacteristicName)

  • steps (tuple[ComputationStep, ...])

target

Characteristic the plan resolves.

Type:

GenericCharacteristicName

source

Starting characteristic of the plan (a self-loop characteristic in Distribution.analytical_computations).

Type:

GenericCharacteristicName

steps

Ordered sequence of steps. For a single-loop plan there is exactly one step source -> source; for a conversion plan the first step starts at source and the last step targets target.

Type:

tuple[ComputationStep, ]

target: TypeAliasType
source: TypeAliasType
steps: tuple[ComputationStep, ...]
required_options()[source]

Return the names of all options that may be consumed by the plan.

Useful for users that want to know which keyword arguments are meaningful for a particular query_method / calculate_characteristic call.

Return type:

tuple[str, ...]

required_characteristic_options()[source]

Return the names of all characteristic options across all steps.

These are the options that are intrinsic to the characteristic and should be broadcast to every step that declares them. They also affect the cache key.

Return type:

tuple[str, ...]

required_computation_options()[source]

Return the names of all computation options across all steps.

These are fitter-specific options that control numerical algorithms. They do not affect the cache key.

Return type:

tuple[str, ...]

with_options(step_index, **kwargs)[source]

Create a StepOptions mapping with validated options for one step.

This is the recommended way to build the options parameter for ComputationStrategy.query_method(). Call it once per step that needs non-default options and merge the results:

plan = distr.explain_computation_path("ppf")
opts = plan.with_options(0, tol=0.1) | plan.with_options(1, eps=1e-3)
ppf = distr.query_method("ppf", options=opts)

Option values are validated eagerly (type-cast + predicate check) so errors surface here rather than deep inside the strategy.

Parameters:

  • step_index (int) – 0-based index into steps.

  • **kwargs (Any) – Option values for that step.

Returns:

A single-entry StepOptions mapping that can be merged with other such mappings via |.

Return type:

dict[int, ResolvedEdgeOptions]

Raises:

  • IndexError – If step_index is out of range.

  • TypeError – If a value cannot be cast to the declared type.

  • ValueError – If a value fails the option’s validation predicate.

__init__(target, source, steps)
Parameters:

  • target (GenericCharacteristicName)

  • source (GenericCharacteristicName)

  • steps (tuple[ComputationStep, ...])

Return type:

None

class pysatl_core.distributions.strategies.ComputationStrategy(*args, **kwargs)[source]

Bases: Protocol

Protocol for strategies that resolve computation methods for characteristics.

enable_caching

Whether to cache fitted computation methods.

Type:

bool

query_method(state, distr, options=None, *, characteristic_options=None, computation_defaults=None)[source]
Return type:

GenericAlias[Any, Any]

Parameters:
explain_computation_path(state, distr)[source]

Describe how this strategy will compute state for distr.

Returns an ComputationPlan that lists every step the strategy will perform along with the option descriptors it will consult at each step. Implementations are expected to fix the returned plan internally so that a subsequent call to query_method() for the same (distr, state) follows the same path even if the strategy makes non-deterministic choices.

Return type:

ComputationPlan

Parameters:
__init__(*args, **kwargs)
class pysatl_core.distributions.strategies.DefaultComputationStrategy(enable_caching=False, computation_defaults=None)[source]

Bases: object

Default strategy for resolving characteristic computation methods.

This strategy first checks for analytical implementations provided by the distribution. If none exists, it walks the characteristic graph to find a conversion path from an analytical characteristic to the target characteristic.

Parameters:

  • enable_caching (bool, default False) – If True, cache fitted conversions to avoid repeated fitting.

  • computation_defaults (Mapping[str, Any] | None, default None) –

    Strategy-level defaults for computation options. These sit between the hardcoded ComputationOption.default and any per-step caller override. Resolution order (highest priority first):

    1. Per-step caller override (options argument to query_method()).

    2. computation_defaults supplied here.

    3. Hardcoded ComputationOption.default on the descriptor.

    Example:

    strategy = DefaultComputationStrategy(
    enable_caching=True,
    computation_defaults={"max_iter": 100, "limit": 50},
)

_enable_caching

Whether caching is enabled.

Type:

bool

_computation_defaults

Strategy-level computation option defaults.

Type:

dict[str, Any]

_cache

Cache of fitted computation methods keyed by (distr_id, edge_id, target, frozen_all_options) so that different option sets (both characteristic and computation) produce independent cache entries. Characteristic options affect the meaning of the result; computation options affect the accuracy of the fitted callable — both must be part of the key.

Type:

dict

_path_cache

Cache of resolved execution plans keyed by (distr_id, target). Lets repeated query_method calls reuse the path produced by a previous explain_computation_path / query_method and keeps both methods in sync for non-deterministic strategies.

Type:

dict

_resolving

Tracking of currently resolving characteristics to detect cycles.

Type:

dict[int, set[str]]

_char_options_stack

Stack of characteristic-options dicts, one entry per active query_method call. When a fitter on step N calls distribution.query_method(intermediate) recursively, the strategy picks up the characteristic options from the top of this stack so they are propagated automatically without the fitter needing to forward them explicitly.

Type:

list[dict[str, Any]]

__init__(enable_caching=False, computation_defaults=None)[source]
Parameters:
Return type:

None

property is_caching_enabled: bool
explain_computation_path(state, distr)[source]

Describe and pin the plan that query_method() will follow.

The returned ComputationPlan lists every step (loop or conversion edge) and the EdgeOptionsDescriptor consulted at that step. The plan is cached per (distr, state) so that a subsequent query_method() call goes through the very same edges – this matters for non-deterministic strategy variants and is also what allows the second call to skip the BFS pass.

Return type:

ComputationPlan

Parameters:
query_method(state, distr, options=None, *, characteristic_options=None, computation_defaults=None)[source]

Resolve a computation method for the target characteristic.

Resolution order: 1. Cached fitted method (if caching enabled) 2. Analytical implementation for non-registry characteristics 3. First self-loop from the registry view 4. Conversion path from loop characteristics via the graph

Parameters:

  • state (str) – Target characteristic name (e.g., “pdf”, “cdf”).

  • distr (Distribution) – Distribution to compute the characteristic for.

  • options (StepOptions | None, default None) – Per-step options built via ComputationPlan.with_options(). Each key is a 0-based step index and each value is a ResolvedEdgeOptions produced by EdgeOptionsDescriptor.with_values(). When None, every edge uses its declared defaults.

  • characteristic_options (Mapping[str, Any] | None, default None) – Shared characteristic options broadcast to every step that declares a matching CharacteristicOption. These are intrinsic to the characteristic (e.g. eps, x0 for PPF) and affect the meaning of the result. Per-step overrides in options take precedence over this dict; the dict takes precedence over the hardcoded CharacteristicOption.default.

  • computation_defaults (Mapping[str, Any] | None, default None) – Per-call computation option defaults. These override the strategy-level computation_defaults set at construction time and the hardcoded ComputationOption.default, but are overridden by per-step values in options.

Returns:

Callable that computes the characteristic.

Return type:

Method

Raises:

RuntimeError – If no analytical base exists, no conversion path is found, or a cycle is detected.

class pysatl_core.distributions.strategies.SamplingStrategy(*args, **kwargs)[source]

Bases: Protocol

Protocol for strategies that generate samples from distributions.

sample(n, distr, **options)[source]
Return type:

TypeAliasType

Parameters:
__init__(*args, **kwargs)