API Reference

class causaloop.ARNoise(alpha: float, sigma: float = 1.0)

Bases: object

Autoregressive noise model for temporal correlations.

Parameters:

• alpha (float) – Autoregressive coefficient (0 < alpha < 1).

• sigma (float, default=1.0) – Innovation standard deviation.

Examples

>>> from causaloop import ARNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = ARNoise(alpha=0.8, sigma=0.1)
>>> # First sample
>>> noise(rng)
array([0.03...])
>>> # Second sample (correlated with first)
>>> noise(rng)
array([-0.07...])

alpha: float

reset() → None

Reset AR process state.

sigma: float = 1.0

class causaloop.BaseVariable(name: str, domain: Domain, value: Any | None = None, units: str | None = None)

Bases: object

Minimal variable abstraction without causal features.

BaseVariable provides the foundation for all variables in the CausaLoop system. It encapsulates a named value within a defined domain, with automatic validation and type inference. This class does not include causal tracking, multi-source resolution, or temporal logic - those features are added in subclasses.

Parameters:

• name (str) – Unique identifier for the variable within its context. Should be descriptive and follow naming conventions of the domain.

• domain (Domain) – The domain defining valid values for this variable. Determines validation rules and variable type.

• value (Any, optional) – Initial value for the variable. Must be valid according to the domain’s validation rules. If None, variable starts uninitialized.

• units (str, optional) – Physical or logical units for the variable’s value. Used for dimensional analysis and unit conversion. Examples: “mg/dL”, “seconds”, “USD”, “count”.

name

Variable identifier.

Type:

str

domain

Domain defining valid values.

Type:

Domain

units

Measurement units, if applicable.

Type:

Optional[str]

_value

Internal storage for the current value.

Type:

Optional[Any]

Raises:

ValueError – If the provided initial value fails domain validation.

Examples

>>> from causaloop import BaseVariable, Interval
>>> # Valid construction
>>> temp = BaseVariable("temperature", Interval(-273.15, 1000), 25.0, "°C")
>>> temp.value
25.0
>>> # Invalid construction - value outside domain
>>> try:
...     var = BaseVariable("var", Interval(0, 1), 1.1, "g/mol")
... except ValueError as e:
...     print(e)
Value 1.1 invalid for domain Interval(0, 1)
>>> # Construction with None value
>>> uninitialized = BaseVariable("pressure", Interval(0, 100))
>>> uninitialized.value is None
True

See also

Variable

Extended variable with multi-source resolution.

CausalVariable

Variable with causal tracking and temporal logic.

Domain

Base class for value domains.

clear() → None

Clear the variable’s value, setting it to None.

Examples

>>> from causaloop import BaseVariable, Interval
>>> var = BaseVariable("counter", Interval(0, 100), 42)
>>> var.value
42
>>> var.clear()
>>> var.value is None
True

Notes

Clearing a variable sets its value to None, representing an uninitialized state. This is different from setting a value that happens to be None within the domain (e.g., for optional categorical variables).

Use this method when you need to reset a variable without destroying the object.

is_valid(value: Any) → bool

Check if a value would be valid for the variable’s domain.

Parameters:

value (Any) – Value to check against the variable’s domain.

Returns:

True if the value is valid for this variable’s domain, False otherwise.

Return type:

bool

Examples

>>> from causaloop import BaseVariable, Interval
>>> var = BaseVariable("age", Interval(0, 120))
>>> var.is_valid(25)
True
>>> var.is_valid(-5)
False
>>> var.is_valid("twenty")  # Non-numeric string
False

Notes

This method performs the same validation as the value setter but without actually setting the value. Useful for pre-checking values before assignment.

See also

value

Property setter that uses the same validation.

to_dict() → dict

Convert variable to dictionary representation.

Returns:

Dictionary containing variable metadata and current state.

Return type:

dict

Examples

>>> from causaloop import BaseVariable, Interval
>>> var = BaseVariable("pressure", Interval(0, 100), 85.5, "kPa")
>>> var.to_dict()
{
    'name': 'pressure',
    'type': 'continuous',
    'value': 85.5,
    'units': 'kPa',
    'domain': 'Interval(0, 100)'
}

Notes

This representation is suitable for serialization (JSON, YAML) and persistence. The domain is represented as its string representation for compactness.

property type: VariableType

Infer the variable’s data type from its domain.

Returns:

The variable type inferred from the domain class.

Return type:

VariableType

Notes

Type inference is based on the domain class hierarchy: - Interval → CONTINUOUS - Categorical → CATEGORICAL - Boolean → BOOLEAN - All others → DISCRETE

This inference is used for serialization, visualization, and automatic solver selection.

Examples

>>> from causaloop import BaseVariable, Interval, Boolean, Categorical
>>> var = BaseVariable("temp", Interval(0, 100), 25)
>>> var.type
<VariableType.CONTINUOUS: 'continuous'>
>>> var = BaseVariable("flag", Boolean(), True)
>>> var.type
<VariableType.BOOLEAN: 'boolean'>
>>> var = BaseVariable("category", Categorical(["A", "B"]), "A")
>>> var.type
<VariableType.CATEGORICAL: 'categorical'>

property value: Any | None

Get the current value of the variable.

Returns:

The current value, or None if the variable is uninitialized.

Return type:

Optional[Any]

Examples

>>> from causaloop import BaseVariable, Interval, Categorical
>>> var = BaseVariable("count", Interval(0, 100), 42.)
>>> var.value
42.0
>>> var = BaseVariable("status", Categorical(["on", "off"]))
>>> var.value is not None
False

class causaloop.Boolean

Bases: Domain

Binary true/false domain.

Represents variables that can take only boolean values (True/False) or their integer equivalents (1/0).

Examples

>>> from causaloop import Boolean
>>> domain = Boolean()
>>> domain.validate(True)
True
>>> domain.validate(1)
True
>>> domain.validate(0)
True
>>> domain.validate(False)
True
>>> domain.validate(2)
False
>>> domain.validate("true")
False

See also

Domain

Base class for all domains.

validate(value: Any) → bool

Check if a value is a valid boolean representation.

Parameters:

value (Any) – The value to validate as boolean.

Returns:

True if value is boolean (True/False) or integer (0/1), False otherwise.

Return type:

bool

Notes

Only accepts Python bool type or integers 0/1. Does not accept string representations like “true” or “false”.

Examples

>>> from causaloop import Boolean
>>> domain = Boolean()
>>> domain.validate(True)
True
>>> domain.validate(1)
True
>>> domain.validate(0.0)
False
>>> domain.validate(2)
False
>>> domain.validate(None)
False

class causaloop.Categorical(categories: List[Any])

Bases: Domain

Domain of unordered categorical values.

Represents variables that can take values from a finite set of discrete categories. Categories can be of any hashable type.

Parameters:

categories (List[Any]) – List of valid categorical values. Values should be hashable.

categories

The list of valid categorical values.

Type:

List[Any]

Examples

>>> from causaloop import Categorical
>>> domain = Categorical(["cat", "dog", "bird"])
>>> domain.validate("cat")
True
>>> domain.validate("fish")
False
>>> domain = Categorical([1, 2, 3])
>>> domain.validate(2)
True

See also

Domain

Base class for all domains.

validate(value: Any) → bool

Check if a value is in the list of valid categories.

Parameters:

value (Any) – The value to check for category membership.

Returns:

True if value is in the categories list, False otherwise.

Return type:

bool

Notes

Uses exact equality comparison (==), not type conversion. Case-sensitive for strings.

Examples

>>> from causaloop import Categorical
>>> domain = Categorical(["red", "green", "blue"])
>>> domain.validate("red")
True
>>> domain.validate("Red")  # Case sensitive
False
>>> domain.validate("yellow")
False

exception causaloop.CausalError

Bases: Exception

Exception raised for causal consistency violations.

This exception indicates that a causal operation would violate temporal consistency, create causal loops, or otherwise break the causal integrity of the system.

Examples

>>> from causaloop import Interval, CausalVariable, CausalError
>>> var = CausalVariable("temp", Interval(0, 100))
>>> var.update("heater", 25.0, time=10.0)
'update_id_1'
>>> try:
...     var.update("heater", 30.0, time=9.0)  # Earlier time!
... except CausalError as e:
...     print(e)
Temporal violation: heater at 9.0 (last update at 10.0)

class causaloop.CausalUpdate(source: str, value: ~typing.Any, timestamp: float, id: str = '', deps: ~typing.List[str] = <factory>, status: ~causaloop.core.variable.types.TemporalStatus = TemporalStatus.CONSEQUENT, confidence: float = 1.0, meta: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Record of a causal update with full temporal and dependency metadata.

A CausalUpdate represents a single causal event where a variable’s value changes due to a specific mechanism. It captures not just the new value, but the complete causal context: when it happened, what caused it, and with what confidence.

Parameters:

• source (str) – Identifier of the mechanism or process that caused this update (e.g., “metabolism”, “insulin_injection”, “sensor_reading”).

• value (Any) – The new value assigned by this update.

• timestamp (float) – Simulation time when this update occurred. Must be monotonically non-decreasing for each source to maintain causal consistency.

• id (str, optional) – Unique identifier for this update. Auto-generated if not provided.

• deps (List[str], optional) – List of causal IDs that this update depends on. Forms the causal dependency graph when combined across updates.

• status (TemporalStatus, default=TemporalStatus.CONSEQUENT) – Temporal status indicating the update’s position in causal ordering.

• confidence (float, default=1.0) – Confidence level (0.0 to 1.0) in this update’s value. Used in weighted resolution strategies.

• meta (Dict[str, Any], optional) – Additional metadata for domain-specific information or analysis.

source

Source mechanism identifier.

Type:

str

value

The update value.

Type:

Any

timestamp

Update timestamp.

Type:

float

id

Unique causal identifier.

Type:

str

deps

Causal dependencies.

Type:

List[str]

status

Temporal status.

Type:

TemporalStatus

confidence

Update confidence.

Type:

float

meta

Additional metadata.

Type:

Dict[str, Any]

Examples

>>> from causaloop import CausalUpdate
>>> update = CausalUpdate(
...     source="metabolism",
...     value=95.0,
...     timestamp=10.5,
...     deps=["prev_update_id"],
...     confidence=0.95
... )
>>> update.id  # Auto-generated
'a1b2c3d4e5f6g7h8'
>>> update.timestamp
10.5

Notes

CausalUpdates form the building blocks of causal reasoning. By tracking dependencies between updates, the system can reconstruct causal pathways and detect inconsistencies.

The auto-generated ID uses SHA-256 hashing for uniqueness while keeping the identifier reasonably short. For extremely high-frequency systems, consider providing custom IDs to avoid collisions.

confidence: float = 1.0

deps: List[str]

id: str = ''

meta: Dict[str, Any]

source: str

status: TemporalStatus = 2

timestamp: float

value: Any

class causaloop.CausalVariable(name: str, domain: Domain, value: Any | None = None, units: str | None = None, noise: Callable[[Generator, int], ndarray] | None = None, rng: Generator | None = None, max_history: int = 1000)

Bases: Variable

Variable with causal-temporal tracking and counterfactual reasoning.

CausalVariable extends Variable with the ability to track the complete causal history of value changes, enforce temporal consistency, and support counterfactual reasoning. It maintains a causal graph of updates and provides methods for analyzing causal pathways.

Key features: 1. Temporal consistency: Prevents retroactive causation and causal loops 2. Causal history: Complete record of all updates with metadata 3. Counterfactual reasoning: Alternative causal scenarios for what-if analysis 4. Causal resolution: Value determination considering temporal context 5. Dependency tracking: Explicit causal dependencies between updates

Parameters:

• name (str) – Variable identifier.

• domain (Domain) – Value domain with validation rules.

• value (Any, optional) – Initial deterministic value.

• units (str, optional) – Measurement units.

• noise (Distribution, optional) – Noise model for stochastic perturbations.

• rng (np.random.Generator, optional) – Random number generator for noise (required if noise specified).

• max_history (int, default=1000) – Maximum number of causal updates to retain in history. Older updates are discarded when this limit is exceeded.

max_history

Maximum causal history length.

Type:

int

_history

Complete causal history in chronological order.

Type:

List[CausalUpdate]

_constraints

Temporal consistency constraints.

Type:

List[Callable]

_pending

Updates pending causal resolution.

Type:

Dict[str, CausalUpdate]

_last_time

Timestamp of most recent update.

Type:

float

_branches

Counterfactual causal branches.

Type:

Dict[str, List[CausalUpdate]]

Examples

>>> import numpy as np
>>> from causaloop import CausalVariable, Interval

>>> # Create causal variable
>>> var = CausalVariable("glucose", Interval(50, 300), 95.0, "mg/dL")

>>> # Record causal updates
>>> update_id = var.update(
...     source="metabolism",
...     value=95.0,
...     time=10.0,
...     deps=["previous_event"],
...     confidence=0.95
... )

>>> # Query causal history
>>> history = var.history(start=0, end=20)
>>> len(history)
1

>>> # Create counterfactual scenario
>>> var.branch(
...     branch_id="insulin_intervention",
...     time=10.0,
...     value=85.0,
...     source="intervention"
... )

>>> # Query counterfactual value
>>> cf_value = var.get_branch_value("insulin_intervention", time=11.0)

See also

Variable

Parent class with deterministic resolution and noise handling.

CausalUpdate

Individual causal event record.

add_constraint(constraint: Callable[[CausalVariable, str, float], bool]) → None

Add a temporal consistency constraint.

Parameters:

constraint (Callable) – Constraint function with signature: constraint(variable: CausalVariable, source: str, time: float) -> bool Should return True if the update is allowed, False otherwise.

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("pressure", Interval(0, 100))

>>> # Constraint: updates from "valve" can only occur at integer times
>>> def integer_time_constraint(var, source, time):
...     if source == "valve":
...         return time.is_integer()
...     return True

>>> var.add_constraint(integer_time_constraint)

>>> # This will work
>>> var.update("valve", 50.0, time=10.0)
'update_id'

>>> # This will fail the constraint
>>> try:
...     var.update("valve", 60.0, time=10.5)
... except CausalError:
...     print("Constraint violation")
Constraint violation

Notes

Constraints allow domain-specific temporal logic to be enforced. Common uses include: - Minimum time between updates from certain sources - Synchronization requirements between variables - Domain-specific timing patterns - Resource availability constraints

Constraints are checked in registration order. The first constraint that returns False prevents the update.

For performance, keep constraint functions simple and fast.

branch(branch_id: str, time: float, value: Any, source: str = 'counterfactual') → None

Create a counterfactual causal branch.

Parameters:

• branch_id (str) – Unique identifier for this counterfactual branch.

• time (float) – Intervention time where the counterfactual diverges from actual history.

• value (Any) – Counterfactual intervention value at the divergence point.

• source (str, default="counterfactual") – Source identifier for the counterfactual intervention.

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("glucose", Interval(50, 300), 95.0, "mg/dL")
>>> var.update("metabolism", 95.0, time=9.0)
'id1'
>>> var.update("meal", 150.0, time=10.0)
'id2'

>>> # Create counterfactual: what if insulin was given instead of meal?
>>> var.branch(
...     branch_id="insulin_intervention",
...     time=10.0,  # Same time as the meal
...     value=85.0,  # Insulin would lower glucose
...     source="insulin_injection"
... )

>>> # Query counterfactual value at later time
>>> cf_value = var.get_branch_value("insulin_intervention", time=11.0)

Notes

Counterfactual branches allow what-if analysis by creating alternative causal histories. Each branch: 1. Copies all actual history up to the intervention time 2. Adds the counterfactual intervention as a new update 3. Can be extended with additional hypothetical updates

Branches are stored separately from actual history and don’t affect the variable’s current value or normal operations.

Multiple branches can be created for different intervention scenarios. Branch IDs must be unique within a variable.

clear_pending() → None

Clear pending updates without affecting history.

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("output", Interval(0, 100))
>>> var.update("process", 50.0, time=10.0)
'id1'
>>> len(var._pending)
1

>>> var.clear_pending()
>>> len(var._pending)
0
>>> len(var._history)  # History preserved
1

Notes

Pending updates are used for current value resolution but can be cleared to reset the variable’s “current state” while preserving historical records. This is useful between simulation runs or analysis phases.

Clearing pending updates does not affect the causal history or counterfactual branches.

get_branch_value(branch_id: str, time: float, strategy: str = 'causal') → Any | None

Get value in a counterfactual branch at specified time.

Parameters:

• branch_id (str) – Counterfactual branch identifier.

• time (float) – Time to query in the counterfactual scenario.

• strategy (str, default="causal") – Resolution strategy for determining the value. See _resolve_causal for available strategies.

Returns:

Value in the counterfactual scenario at the specified time, or None if the branch doesn’t exist or has no relevant updates.

Return type:

Optional[Any]

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100), 20.0)
>>> var.update("heater", 25.0, time=10.0)
'id1'

>>> # Create counterfactual: what if heater was stronger?
>>> var.branch("stronger_heater", time=10.0, value=30.0)

>>> # Query counterfactual value
>>> cf_value = var.get_branch_value("stronger_heater", time=11.0)
>>> cf_value
30.0

Notes

This method reconstructs the counterfactual state by: 1. Finding all updates in the branch up to the query time 2. Applying the specified resolution strategy to those updates

The resolution considers only updates within the branch, ignoring actual history after the intervention point.

For complex counterfactuals with multiple hypothetical updates after the intervention, additional updates would need to be added to the branch manually.

get_causal_path(start_time: float, end_time: float) → List[CausalUpdate]

Get the causal pathway between two times.

Parameters:

• start_time (float) – Start time for causal pathway.

• end_time (float) – End time for causal pathway.

Returns:

Updates in the causal pathway, in chronological order.

Return type:

List[CausalUpdate]

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("signal", Interval(0, 10))
>>> var.update("A", 1.0, time=1.0, deps=[])
'id1'
>>> var.update("B", 2.0, time=2.0, deps=["id1"])
'id2'
>>> var.update("C", 3.0, time=3.0, deps=["id2"])
'id3'

>>> # Get causal pathway
>>> path = var.get_causal_path(start_time=1.0, end_time=3.0)
>>> [u.source for u in path]
['A', 'B', 'C']

Notes

A causal pathway is a sequence of updates where each update (after the first) depends on the previous one. This method reconstructs such pathways by following dependency links.

If multiple dependency paths exist, this method returns one valid path (not necessarily the shortest or only one).

Circular dependencies will cause infinite recursion.

get_most_recent(source: str | None = None) → CausalUpdate | None

Get the most recent causal update.

Parameters:

source (str, optional) – If provided, returns most recent update from this specific source.

Returns:

Most recent update matching criteria, or None if no updates exist.

Return type:

Optional[CausalUpdate]

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100))
>>> var.update("sensor1", 25.0, time=10.0)
'id1'
>>> var.update("sensor2", 26.0, time=11.0)
'id2'

>>> # Most recent overall
>>> recent = var.get_most_recent()
>>> recent.source
'sensor2'
>>> recent.timestamp
11.0

>>> # Most recent from specific source
>>> sensor1_recent = var.get_most_recent(source="sensor1")
>>> sensor1_recent.value
25.0

Notes

This method is optimized for frequent access by tracking the most recent update time internally. For source-specific queries, it performs a reverse linear search through history.

For variables with very long histories, consider maintaining source-specific pointers for O(1) access.

history(start: float = 0.0, end: float | None = None, source: str | None = None) → List[CausalUpdate]

Retrieve causal history with optional filtering.

Parameters:

• start (float, default=0.0) – Start time (inclusive) for history query.

• end (float, optional) – End time (inclusive) for history query. If None, includes all updates from start onward.

• source (str, optional) – If provided, only returns updates from this source.

Returns:

List of causal updates matching the filter criteria, in chronological order.

Return type:

List[CausalUpdate]

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("pressure", Interval(0, 100))
>>> var.update("valve", 50.0, time=10.0)
'id1'
>>> var.update("pump", 60.0, time=12.0)
'id2'
>>> var.update("valve", 55.0, time=15.0)
'id3'

>>> # All history
>>> all_history = var.history()
>>> len(all_history)
3

>>> # Time window
>>> window = var.history(start=11.0, end=14.0)
>>> len(window)
1
>>> window[0].source
'pump'

>>> # Source-specific
>>> valve_history = var.history(source="valve")
>>> len(valve_history)
2

Notes

The history is stored in chronological order, so this method performs linear filtering. For large histories with frequent queries, consider maintaining additional indexing structures.

If max_history is set and old updates have been discarded, queries for times before the oldest retained update will return empty results.

reset() → None

Reset all causal tracking (history, branches, pending updates).

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("signal", Interval(0, 10))
>>> var.update("source", 5.0, time=1.0)
'id1'
>>> var.branch("counterfactual", time=1.0, value=6.0)

>>> len(var._history)
1
>>> len(var._branches)
1

>>> var.reset()
>>> len(var._history)
0
>>> len(var._branches)
0
>>> var._last_time
0.0

Notes

This method completely resets causal tracking while preserving the variable’s basic configuration (domain, units, noise model).

Use this method to start fresh causal tracking, such as when beginning a new simulation or analysis from time zero.

Source registrations and weights from the parent class are preserved. Only causal-specific state is reset.

resolve_causal(strategy: str = 'causal') → Any

Get value using specified causal resolution strategy.

Parameters:

strategy (str, default="causal") – Resolution strategy (see _resolve_causal for options).

Returns:

Resolved value according to specified strategy.

Return type:

Any

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("signal", Interval(0, 10))
>>> var.update("source1", 5.0, time=1.0, confidence=0.8)
'id1'
>>> var.update("source2", 7.0, time=2.0, confidence=0.9)
'id2'

>>> # Different resolution strategies
>>> var.resolve_causal("causal")  # Weighted by time and confidence
np.float64(6.10...)
>>> var.resolve_causal("most_recent")  # Most recent update
7.0
>>> var.resolve_causal("most_confident")  # Highest confidence
7.0

Notes

This method provides explicit control over causal resolution, which can be important for different analysis purposes:

• “causal”: For simulation where recent, confident updates dominate

• “most_recent”: For real-time monitoring or state tracking

• “most_confident”: When reliability is more important than recency

to_dict() → Dict

Convert variable to detailed dictionary representation.

Returns:

Dictionary containing variable metadata, source information, causal history summary, and configuration.

Return type:

Dict

Notes

The full causal history can be large, so this method only includes summary statistics. For complete serialization of causal state, additional methods would be needed to export/import the full history and branches.

Noise models and RNGs require special handling for serialization and are not included in the basic dictionary representation.

update(source: str, value: Any, time: float, deps: List[str] | None = None, confidence: float = 1.0, meta: Dict[str, Any] | None = None) → str

Record a causal update with temporal consistency checking.

Parameters:

• source (str) – Source mechanism causing this update.

• value (Any) – New deterministic value (must satisfy domain constraints).

• time (float) – Simulation time when this update occurs.

• deps (List[str], optional) – List of causal IDs that this update depends on.

• confidence (float, default=1.0) – Confidence in this update (0.0 to 1.0).

• meta (Dict[str, Any], optional) – Additional metadata for analysis or domain-specific information.

Returns:

Unique causal ID of the created update.

Return type:

str

Raises:

• CausalError – If the update violates temporal consistency constraints.

• ValueError – If the value violates domain constraints.

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100))

>>> # Simple update
>>> update_id = var.update("heater", 25.0, time=10.0)

>>> # Update with dependencies
>>> update_id2 = var.update(
...     source="thermostat",
...     value=26.0,
...     time=11.0,
...     deps=[update_id],  # Depends on previous update
...     confidence=0.9,
...     meta={"room": "living_room"}
... )

>>> # Invalid: retroactive update
>>> try:
...     var.update("heater", 30.0, time=9.0)
... except CausalError as e:
...     print(e)
Temporal violation: heater at 9.0 (last update at 10.0)

Notes

This method enforces causal consistency by: 1. Checking that time >= last update time for this source 2. Validating against all registered temporal constraints 3. Recording dependencies to build the causal graph

The update is stored in causal history and also registered as a normal source output for deterministic value resolution.

Dependencies should reference valid causal IDs from previous updates. Circular dependencies are not automatically detected but will cause issues in causal analysis.

property value: Any

Get current value using causal-aware resolution.

Returns:

Current deterministic value resolved using causal strategy.

Return type:

Any

Notes

The causal resolution strategy considers: 1. Temporal recency: More recent updates have higher weight 2. Confidence: Updates with higher confidence have higher weight 3. Source-specific weighting: As configured via add_source()

The resolution uses exponential decay for temporal weighting:

weight = confidence * exp(-decay_rate * time_since_update)

This gives a smooth transition between updates while respecting causal ordering.

For the pure deterministic value without causal weighting, use the parent class’s resolution directly:

>>> det_value = super(CausalVariable, var).value

Or access the most recent update:

>>> recent = var.get_most_recent()
>>> det_value = recent.value if recent else None

class causaloop.CompositeNoise(models: list = <factory>, weights: list | None = None)

Bases: object

Composite noise model combining multiple noise sources.

Parameters:

• models (list) – List of noise models to combine.

• weights (list, optional) – Relative weights for each model. If None, equal weights.

Examples

>>> from causaloop import GaussianNoise, UniformNoise, CompositeNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> models = [
...     GaussianNoise(mean=0, std=0.1),
...     UniformNoise(low=-0.05, high=0.05)
... ]
>>> noise = CompositeNoise(models=models, weights=[0.7, 0.3])
>>> noise(rng, 3)
array([ 0.02..., -0.08...,  0.06...])

models: list

weights: list | None = None

class causaloop.Domain

Bases: ABC

Abstract base class for variable domains.

A Domain defines the set of valid values a variable can take and provides validation logic. Concrete subclasses implement specific types of domains (numeric intervals, categorical sets, etc.).

Subclasses must implement the validate method and may override __repr__ for better debugging output.

See also

Interval

Continuous numeric range domain.

Categorical

Finite set of unordered categories.

Boolean

Binary true/false domain.

abstractmethod validate(value: Any) → bool

Check if a value is valid for this domain.

Parameters:

value (Any) – The value to check for domain membership.

Returns:

True if the value is within the domain’s definition, False otherwise.

Return type:

bool

Notes

Implementations should handle type conversion gracefully where appropriate. For example, numeric domains should attempt to convert string representations to numbers before validation.

Raises:

NotImplementedError – When called directly on the abstract base class.

class causaloop.DomainProtocol(*args, **kwargs)

Bases: Protocol

Protocol defining the interface for variable domain implementations.

A Domain defines the valid value space for a variable and provides validation logic. This protocol ensures all domain implementations provide consistent validation and representation.

validate(value: Any) → bool

Check if a value is within the domain’s definition.

__repr__() → str

Return a string representation of the domain.

See also

Domain

Abstract base class implementing this protocol.

Interval

Continuous numeric domain implementation.

Categorical

Discrete categorical domain implementation.

Boolean

Binary domain implementation.

validate(value: Any) → bool

Validate that a value belongs to this domain.

Parameters:

value (Any) – The value to validate against the domain’s constraints.

Returns:

True if the value is valid for this domain, False otherwise.

Return type:

bool

Notes

Validation should be permissive where possible, attempting type conversions for numeric domains. For example, an Interval domain should accept string representations of numbers.

class causaloop.GaussianNoise(mean: float = 0.0, std: float = 1.0)

Bases: object

Gaussian (normal) noise model.

Parameters:

• mean (float, default=0.0) – Mean of the Gaussian distribution.

• std (float, default=1.0) – Standard deviation of the Gaussian distribution.

Examples

>>> from causaloop import GaussianNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = GaussianNoise(mean=0, std=0.1)
>>> noise(rng)
array([0.030...])

mean: float = 0.0

std: float = 1.0

class causaloop.Interval(low: float = -inf, high: float = inf)

Bases: Domain

Continuous numeric domain defined by lower and upper bounds.

Represents variables that can take any real value within a specified range [low, high]. The bounds can be infinite for unbounded domains.

Parameters:

• low (float, optional) – Lower bound of the interval (inclusive). Default is -inf.

• high (float, optional) – Upper bound of the interval (inclusive). Default is +inf.

low

Lower bound of the interval.

Type:

float

high

Upper bound of the interval.

Type:

float

Examples

>>> from causaloop import Interval
>>> domain = Interval(0, 100)
>>> domain.validate(50)
True
>>> domain.validate(-10)
False
>>> domain.validate("75")  # String representation
True

See also

Domain

Base class for all domains.

validate(value: Any) → bool

Check if a value is within the interval bounds.

Parameters:

value (Any) – The value to validate. Can be numeric or string representation.

Returns:

True if value can be converted to float and lies within [low, high] (inclusive), False otherwise.

Return type:

bool

Notes

Attempts to convert the value to float before comparison. Returns False if conversion fails or value is outside bounds.

Examples

>>> from causaloop import Interval
>>> domain = Interval(0, 10)
>>> domain.validate(5)
True
>>> domain.validate(15)
False
>>> domain.validate("7.5")
True
>>> domain.validate("not a number")
False

class causaloop.LogNormalNoise(mean: float = 0.0, sigma: float = 1.0)

Bases: object

Log-normal noise model for positive-valued variables.

Parameters:

• mean (float, default=0.0) – Mean of the underlying normal distribution.

• sigma (float, default=1.0) – Standard deviation of the underlying normal distribution.

Examples

>>> from causaloop import LogNormalNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = LogNormalNoise(mean=0, sigma=0.1)
>>> noise(rng)
array([1.03...])

mean: float = 0.0

sigma: float = 1.0

class causaloop.PoissonNoise(lam: float = 1.0)

Bases: object

Poisson noise for count variables.

Parameters:

lam (float, default=1.0) – Expected value (lambda) of Poisson distribution.

Examples

>>> from causaloop import PoissonNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = PoissonNoise(lam=5.0)
>>> noise(rng)
array([8])

lam: float = 1.0

class causaloop.TemporalStatus(*values)

Bases: Enum

Enumeration of temporal positions in causal ordering.

These statuses track where a variable update sits in the causal-temporal sequence, crucial for maintaining causal consistency and enabling counterfactual reasoning.

ANTECEDENT

Update occurs before a causal event (in the causal past). Used for establishing baseline conditions.

Type:

auto

CONSEQUENT

Update occurs after a causal event (in the causal future). Represents effects or outcomes of causal mechanisms.

Type:

auto

PENDING

Update is scheduled but not yet resolved in the causal order. Used for asynchronous or concurrent process execution.

Type:

auto

VIOLATED

Update violates causal consistency constraints. Indicates a causal paradox or constraint violation.

Type:

auto

COUNTERFACTUAL

Update exists in a hypothetical counterfactual scenario. Used for what-if analysis and intervention testing.

Type:

auto

ANTECEDENT = 1

CONSEQUENT = 2

COUNTERFACTUAL = 5

PENDING = 3

VIOLATED = 4

class causaloop.UniformNoise(low: float = -1.0, high: float = 1.0)

Bases: object

Uniform noise model.

Parameters:

• low (float, default=-1.0) – Lower bound of uniform distribution.

• high (float, default=1.0) – Upper bound of uniform distribution.

Examples

>>> from causaloop import UniformNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = UniformNoise(low=-0.5, high=0.5)
>>> noise(rng, 4)
array([ 0.27..., -0.06...,  0.35...,  0.19...])

high: float = 1.0

low: float = -1.0

class causaloop.Variable(name: str, domain: Domain, value: Any | None = None, units: str | None = None, noise: Callable[[Generator, int], ndarray] | None = None, rng: Generator | None = None)

Bases: BaseVariable

Multi-source variable with deterministic resolution and optional stochastic noise.

Variable extends BaseVariable with the ability to receive values from multiple sources (e.g., mechanisms, processes, measurements) and provides configurable strategies for resolving conflicts. Following scientific convention, deterministic values and stochastic noise are clearly separated:

1. Deterministic value: Computed from source values using weighted averaging

2. Noise model: Optional stochastic perturbation applied separately

3. RNG: Explicit random number generator for reproducible noise

This separation ensures causal integrity while enabling flexible stochastic modeling. Users can choose when and how to apply noise based on their specific requirements.

Parameters:

• name (str) – Unique identifier for the variable (e.g., “temperature”, “glucose_level”).

• domain (Domain) – Mathematical domain defining valid values (e.g., Interval, Categorical).

• value (Any, optional) – Initial deterministic value. Must satisfy domain constraints if provided.

• units (str, optional) – Physical or logical units (e.g., “°C”, “mg/dL”, “count”).

• noise (Distribution, optional) – Noise model implementing the Distribution protocol. Signature: noise(rng: np.random.Generator, n: int) -> np.ndarray

• rng (np.random.Generator, optional) – Random number generator for reproducible stochasticity. Required if noise is specified.

noise

Noise model for stochastic perturbations.

Type:

Optional[Distribution]

rng

Random number generator for noise sampling.

Type:

Optional[np.random.Generator]

_sources

List of registered source identifiers.

Type:

List[str]

_weights

Source weights for weighted averaging.

Type:

Dict[str, float]

_source_values

Current values provided by each source.

Type:

Dict[str, Any]

Raises:

ValueError – If initial value violates domain constraints, or if noise is specified without providing an RNG.

Examples

>>> from causaloop import Variable, Interval
>>> import numpy as np

>>> # Create deterministic variable
>>> var = Variable("temperature", Interval(0, 100), 25.0, "°C")
>>> var.value  # Deterministic value
25.0

>>> # Create variable with noise model
>>> def gaussian_noise(rng: np.random.Generator, n: int) -> np.ndarray:
...     return rng.normal(loc=0.0, scale=0.1, size=n)
>>>
>>> rng = np.random.default_rng(seed=42)  # For reproducibility
>>> noisy_var = Variable("signal", Interval(-1, 1), 0.0,
...                      noise=gaussian_noise, rng=rng)
>>>
>>> # Deterministic value (always the same)
>>> noisy_var.value
0.0
>>>
>>> # Apply noise (different each call due to RNG state)
>>> noisy_var.apply_noise()
(0.0, 0.03..)
>>>
>>> # Sample noise separately
>>> noise_samples = noisy_var.sample_noise(n=5)
>>> noise_samples.shape
(5,)

See also

BaseVariable

Parent class with domain validation.

CausalVariable

Extension with temporal causal tracking.

property active_sources: List[str]

Get list of sources that have provided current values.

Returns:

List of source identifiers that have non-None values.

Return type:

List[str]

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("score", Interval(0, 100))
>>> var.add_source("judge1")
Variable('score', value=None, sources=1)
>>> var.add_source("judge2")
Variable('score', value=None, sources=2)
>>> var.register_output("judge1", 85)
>>> var.active_sources
['judge1']

Notes

This property helps distinguish between potential influences (all registered sources) and actual current influences (active sources).

Inactive sources (registered but not providing values) don’t affect the variable’s current deterministic value.

add_source(source: str, weight: float = 1.0) → Variable

Register a potential source for this variable.

Parameters:

• source (str) – Unique identifier for the source (e.g., “metabolism”, “sensor_A”, “process_1”). Should be descriptive and consistent.

• weight (non-negative float, default=1.0) – Relative weight of this source in weighted averaging. Higher weights give the source more influence. Weights are relative, not normalized.

Returns:

Self for method chaining.

Return type:

Variable

Raises:

ValueError – If weight is negative.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("process_A", weight=2.0)
Variable('output', value=None, sources=1)
>>> var.add_source("process_B", weight=1.0)
Variable('output', value=None, sources=2)
>>> var.get_source_weight("process_A")
2.0

Notes

Sources can be registered before they provide values, allowing pre-configuration of the variable’s dependency structure. This is useful for declaring expected influences in complex systems.

Weights are used in the default “weighted” resolution strategy:

weighted_sum = Σ(weight_i * value_i) total_weight = Σ(weight_i) result = weighted_sum / total_weight

Weight values are relative: a source with weight 2.0 has twice the influence of a source with weight 1.0.

apply_noise(value: Any | None = None) → Tuple[Any, Any]

Apply noise to a value, returning both deterministic and noisy results.

Parameters:

value (Any, optional) – Value to apply noise to. If None, uses the variable’s current deterministic value.

Returns:

Tuple containing (deterministic_value, noisy_value). If no noise model is configured, returns (value, value).

Return type:

Tuple[Any, Any]

Raises:

ValueError – If the provided value violates domain constraints.

Examples

>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> def uniform_noise(rng, n):
...     return rng.uniform(-0.1, 0.1, n)
>>>
>>> rng = np.random.default_rng(seed=42)
>>> var = Variable("measurement", Interval(0, 10),
...                noise=uniform_noise, rng=rng)
>>> var.register_output("sensor", 5.0)
>>>
>>> # Apply noise to current value
>>> deterministic, noisy = var.apply_noise()
>>> deterministic
5.0
>>> noisy  # 5.0 + noise
5.05...
>>>
>>> # Apply noise to specific value
>>> var.apply_noise(7.5)
(7.5, 7.48...)
>>>
>>> # Deterministic variable
>>> det_var = Variable("constant", Interval(0, 1), 0.5)
>>> det_var.apply_noise()  # No noise model
(0.5, 0.5)

Notes

This method follows the scientific convention of clearly separating deterministic values from stochastic perturbations. It always returns both components, allowing users to:

1. Track the underlying deterministic signal

2. Apply domain-specific handling of noisy values

3. Compute error statistics (noise = noisy - deterministic)

4. Implement custom noise handling (clipping, rejection, etc.)

The method validates that the input value satisfies domain constraints but does NOT validate the noisy result. Users are responsible for handling cases where noise pushes values outside valid domains.

This design supports flexible modeling strategies: - Measurement error: deterministic = true value, noisy = observed - Process noise: deterministic = expected, noisy = realized - Stochastic processes: deterministic = mean, noisy = sample

clear_sources() → None

Clear all source values while preserving source registrations.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("temperature", Interval(0, 100))
>>> var.register_output("sensor1", 25.0)
>>> var.register_output("sensor2", 26.0)
>>> var.value
25.5
>>> var.clear_sources()
>>> var.value is None
True
>>> var._sources  # Sources still registered
['sensor1', 'sensor2']
>>> var._source_values  # Values cleared
{}

Notes

This method clears only the current values from sources, not the source registrations themselves. This is useful for:

• Resetting variable state between simulation runs

• Clearing temporary measurements while keeping configurations

• Reusing variable structures with different input data

Source weights and noise models remain unchanged. To completely reset a variable, create a new instance.

get_deterministic_and_noisy() → Tuple[Any, Any]

Get both deterministic value and noisy version in one call.

Returns:

Tuple containing (deterministic_value, noisy_value). Convenience method equivalent to apply_noise(None).

Return type:

Tuple[Any, Any]

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("signal", Interval(-1, 1), noise=..., rng=...)
>>> var.register_output("source", 0.5)
>>> det, noisy = var.get_deterministic_and_noisy()
>>> det  # Deterministic value
0.5

Notes

This is a convenience method for the common pattern of wanting both the deterministic value and a noisy version. It’s equivalent to:

>>> det, noisy = var.apply_noise(var.value)

The method always returns a tuple, even for deterministic variables (where both elements are identical).

get_source_value(source: str) → Any | None

Get the current value from a specific source.

Parameters:

source (str) – Source identifier.

Returns:

Current deterministic value from the specified source, or None if the source hasn’t provided a value.

Return type:

Optional[Any]

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("pressure", Interval(0, 100))
>>> var.register_output("sensor_A", 85.5)
>>> var.get_source_value("sensor_A")
85.5
>>> var.get_source_value("sensor_B") is None  # Not provided
True

Notes

This method retrieves the raw value provided by a specific source, before any resolution or noise application. It’s useful for:

• Debugging source contributions

• Implementing custom resolution logic

• Monitoring individual component outputs

• Validating source behavior

get_source_weight(source: str) → float

Get the weight of a registered source.

Parameters:

source (str) – Source identifier.

Returns:

Current weight of the source. Returns 0.0 if source is not registered.

Return type:

float

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("process1", weight=2.0)
Variable('output', value=None, sources=1)
>>> var.get_source_weight("process1")
2.0
>>> var.get_source_weight("unknown_process")  # Not registered
0.0

Notes

Source weights determine influence in weighted averaging. A weight of 0.0 effectively disables a source’s contribution (unless using other strategy).

Weights are relative, not normalized. Changing one source’s weight doesn’t automatically adjust others.

property has_noise: bool

Check if the variable has a configured noise model.

Returns:

True if a noise model and RNG are configured, False otherwise.

Return type:

bool

Examples

>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> var = Variable("signal", Interval(-1, 1))
>>> var.has_noise
False

>>> noisy_var = Variable("noisy_signal", Interval(-1, 1),
...                      noise=lambda rng, n: rng.normal(0, 0.1, n),
...                      rng=np.random.default_rng())
>>> noisy_var.has_noise
True

Notes

This property checks both the presence of a noise model and a valid RNG. A variable with a noise model but no RNG is considered invalid and will raise an error during initialization.

register_output(source: str, value: Any) → None

Register a deterministic value from a specific source.

Parameters:

• source (str) – Source identifier that generated this value.

• value (Any) – The deterministic value to register. Must satisfy all domain constraints.

Raises:

ValueError – If the value violates domain constraints.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("glucose", Interval(50, 300))
>>> var.add_source("metabolism", weight=1.0)
Variable('glucose', value=None, sources=1)
>>> var.register_output("metabolism", 95.0)
>>> var.value  # Deterministic value
95.0

>>> # Multiple sources with different weights
>>> var.add_source("diet", weight=0.5)
Variable('glucose', value=95.0, sources=2)
>>> var.register_output("diet", 105.0)
>>> var.value  # Weighted average: (1.0*95 + 0.5*105) / 1.5
98.333...

Notes

This method updates the variable’s deterministic value immediately based on all registered source values. The computation is purely deterministic and does not involve any stochastic components.

The method performs full domain validation to ensure mathematical consistency. If validation fails, the variable’s state remains unchanged.

Sources are automatically registered if not already present (with default weight 1.0). This allows flexible usage patterns.

resolve(strategy: str = 'weighted') → Any

Get deterministic value using specified resolution strategy.

Parameters:

strategy (str, default="weighted") – Resolution strategy for handling multiple source values: - “weighted”: Weighted average (recommended for most applications) - “latest”: Most recently registered value - “priority”: Value from source with highest weight

Returns:

Deterministic value according to the specified strategy.

Return type:

Any

Raises:

ValueError – If an unknown resolution strategy is specified.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("score", Interval(0, 100))
>>> var.register_output("judge1", 85)
>>> var.register_output("judge2", 90)
>>> var.add_source("judge1", weight=2.0)
Variable('score', value=86.66..., sources=2)

>>> # Different resolution strategies
>>> var.resolve("weighted")  # (2*85 + 1*90) / 3
86.666...
>>> var.resolve("latest")    # Most recent value
90
>>> var.resolve("priority")  # Judge1 has higher weight
85

Notes

This method provides explicit control over the resolution strategy, which can be important in specific modeling contexts:

• “weighted”: Appropriate when sources have different reliabilities

• “latest”: Useful for temporal sequences or state updates

• “priority”: When certain sources should dominate others

All strategies produce deterministic results. For stochastic perturbations, use apply_noise() on the resolved value.

sample_noise(n: int = 1) → ndarray

Sample noise from the configured noise model.

Parameters:

n (int, default=1) – Number of independent noise samples to generate.

Returns:

Array of noise samples with shape (n,). Returns zeros if no noise model is configured.

Return type:

np.ndarray

Raises:

RuntimeError – If noise model is configured but RNG is not available.

Examples

>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> def gaussian_noise(rng, n):
...     return rng.normal(0, 0.1, n)
>>>
>>> rng = np.random.default_rng(seed=42)
>>> var = Variable("signal", Interval(-1, 1),
...                noise=gaussian_noise, rng=rng)
>>>
>>> # Single noise sample
>>> var.sample_noise()
array([0.03...])
>>>
>>> # Multiple samples
>>> print(var.sample_noise(5))
[-0.10...  0.07...  0.09... -0.19... -0.13...]
>>>
>>> # Variable without noise model
>>> deterministic_var = Variable("constant", Interval(0, 1), 0.5)
>>> deterministic_var.sample_noise(3)  # Returns zeros
array([0., 0., 0.])

Notes

This method samples noise independently of the variable’s current value. The noise samples can be applied to any value using addition:

>>> value = 5.0
>>> noise = var.sample_noise()
>>> noisy_value = value + float(noise[0])

For variables without noise models, this method returns zeros, allowing uniform handling of both deterministic and stochastic variables in algorithms.

The RNG state advances with each call, ensuring proper stochastic behavior in sequential sampling.

set_source_weight(source: str, weight: float) → None

Set or update the weight of a source.

Parameters:

• source (str) – Source identifier.

• weight (non-negative float) – New weight for the source.

Raises:

• ValueError – If weight is negative.

• KeyError – If source is not registered.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("process1", weight=1.0)
Variable('output', value=None, sources=1)
>>> var.set_source_weight("process1", 3.0)
>>> var.get_source_weight("process1")
3.0

Notes

Changing source weights affects future value resolution but doesn’t change the variable’s current value. The new weight takes effect on the next value computation.

To immediately update the variable’s value with new weights, call register_output again or access the value property.

property sources: List[str]

Get list of registered sources.

Returns:

List of source identifiers in registration order.

Return type:

List[str]

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("temperature", Interval(0, 100))
>>> var.add_source("indoor_sensor")
Variable('temperature', value=None, sources=1)
>>> var.add_source("outdoor_sensor")
Variable('temperature', value=None, sources=2)
>>> var.sources
['indoor_sensor', 'outdoor_sensor']

Notes

The order of sources in this list reflects registration order, which may be relevant for “latest” resolution strategy or for understanding the variable’s dependency structure.

to_dict() → Dict

Convert variable to detailed dictionary representation.

Returns:

Dictionary containing variable metadata, source information, deterministic value, and noise configuration.

Return type:

Dict

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("proc1", weight=2.0)
Variable('output', value=None, sources=1)
>>> var.add_source("proc2", weight=1.0)
Variable('output', value=None, sources=2)
>>> var.register_output("proc1", 75.0)
>>> var.to_dict()
{
    'name': 'output',
    'type': 'continuous',
    'value': 75.0,
    'units': None,
    'domain': 'Interval(0, 100)',
    'sources': ['proc1', 'proc2'],
    'weights': {'proc1': 2.0, 'proc2': 1.0},
    'source_values': {'proc1': 75.0},
    'has_noise': False
}

Notes

This representation is suitable for serialization (JSON, YAML), persistence, and configuration. It captures all necessary information to reconstruct the variable’s state, though noise models and RNGs may require special handling for complete serialization.

The dictionary includes both structural information (sources, weights) and current state (value, source_values).

property value: Any

Get the deterministic value of the variable.

Returns:

Deterministic value resolved from all registered source values using the default “weighted” strategy.

Return type:

Any

Notes

This property returns the pure deterministic value without any stochastic perturbations. It represents the variable’s “true” value in the causal model, free from measurement error or process noise.

The value is recomputed on each access to reflect the current state of all source values. This ensures consistency in dynamic systems.

For variables with noise models, this property does NOT apply noise. Use apply_noise() or sample_noise() for stochastic values.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("concentration", Interval(0, 100))
>>> var.register_output("source1", 30.0)
>>> var.register_output("source2", 40.0)
>>> var.value  # Weighted average: (30 + 40) / 2
35.0

>>> # Even with noise model, value is deterministic
>>> noisy_var = Variable("measurement", Interval(0, 10),
...                      noise=..., rng=...)
>>> noisy_var.register_output("sensor", 5.0)
>>> noisy_var.value  # Still 5.0, not 5.0 + noise
5.0

class causaloop.VariableType(*values)

Bases: Enum

Enumeration of variable data types in the causal modeling system.

These types determine how variables are validated, transformed, and interpreted during simulation and analysis.

CONTINUOUS

Real-valued variables with potentially infinite precision. Examples: temperature, pressure, concentration.

Type:

str

DISCRETE

Integer-valued or count variables with distinct, separate values. Examples: population count, inventory levels, event counts.

Type:

str

CATEGORICAL

Variables taking values from a finite set of unordered categories. Examples: species type, disease status, product category.

Type:

str

ORDINAL

Variables with ordered categories but unknown intervals between them. Examples: severity level (mild/moderate/severe), Likert scale responses.

Type:

str

BOOLEAN

Binary true/false or 0/1 variables. Examples: switch status, diagnostic test result, presence/absence.

Type:

str

BOOLEAN = 'boolean'

CATEGORICAL = 'categorical'

CONTINUOUS = 'continuous'

DISCRETE = 'discrete'

ORDINAL = 'ordinal'

causaloop.create_noise(noise_type: str = 'gaussian', **params) → NoiseModel

Create noise function.

Parameters:

• noise_type (str, default="gaussian") – Type of noise: “gaussian”, “uniform”

• **params – Parameters for noise model.

Returns:

Configured noise model.

Return type:

NoiseModel

Example

>>> from causaloop import create_noise
>>> import numpy as np
>>> noise = create_noise(noise_type="gaussian", **{'mean': 0.0, 'std': 1.0})
>>> rng = np.random.default_rng(seed=42)
>>> noise(rng, 100).shape
(100,)

Core Module

class causaloop.core.ARNoise(alpha: float, sigma: float = 1.0)

Autoregressive noise model for temporal correlations.

Parameters:

• alpha (float) – Autoregressive coefficient (0 < alpha < 1).

• sigma (float, default=1.0) – Innovation standard deviation.

Examples

>>> from causaloop import ARNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = ARNoise(alpha=0.8, sigma=0.1)
>>> # First sample
>>> noise(rng)
array([0.03...])
>>> # Second sample (correlated with first)
>>> noise(rng)
array([-0.07...])

alpha: float

reset() → None

Reset AR process state.

sigma: float = 1.0

class causaloop.core.BaseVariable(name: str, domain: Domain, value: Any | None = None, units: str | None = None)

Minimal variable abstraction without causal features.

BaseVariable provides the foundation for all variables in the CausaLoop system. It encapsulates a named value within a defined domain, with automatic validation and type inference. This class does not include causal tracking, multi-source resolution, or temporal logic - those features are added in subclasses.

Parameters:

• name (str) – Unique identifier for the variable within its context. Should be descriptive and follow naming conventions of the domain.

• domain (Domain) – The domain defining valid values for this variable. Determines validation rules and variable type.

• value (Any, optional) – Initial value for the variable. Must be valid according to the domain’s validation rules. If None, variable starts uninitialized.

• units (str, optional) – Physical or logical units for the variable’s value. Used for dimensional analysis and unit conversion. Examples: “mg/dL”, “seconds”, “USD”, “count”.

name

Variable identifier.

Type:

str

domain

Domain defining valid values.

Type:

Domain

units

Measurement units, if applicable.

Type:

Optional[str]

_value

Internal storage for the current value.

Type:

Optional[Any]

Raises:

ValueError – If the provided initial value fails domain validation.

Examples

>>> from causaloop import BaseVariable, Interval
>>> # Valid construction
>>> temp = BaseVariable("temperature", Interval(-273.15, 1000), 25.0, "°C")
>>> temp.value
25.0
>>> # Invalid construction - value outside domain
>>> try:
...     var = BaseVariable("var", Interval(0, 1), 1.1, "g/mol")
... except ValueError as e:
...     print(e)
Value 1.1 invalid for domain Interval(0, 1)
>>> # Construction with None value
>>> uninitialized = BaseVariable("pressure", Interval(0, 100))
>>> uninitialized.value is None
True

See also

Variable

Extended variable with multi-source resolution.

CausalVariable

Variable with causal tracking and temporal logic.

Domain

Base class for value domains.

clear() → None

Clear the variable’s value, setting it to None.

Examples

>>> from causaloop import BaseVariable, Interval
>>> var = BaseVariable("counter", Interval(0, 100), 42)
>>> var.value
42
>>> var.clear()
>>> var.value is None
True

Notes

Clearing a variable sets its value to None, representing an uninitialized state. This is different from setting a value that happens to be None within the domain (e.g., for optional categorical variables).

Use this method when you need to reset a variable without destroying the object.

is_valid(value: Any) → bool

Check if a value would be valid for the variable’s domain.

Parameters:

value (Any) – Value to check against the variable’s domain.

Returns:

True if the value is valid for this variable’s domain, False otherwise.

Return type:

bool

Examples

>>> from causaloop import BaseVariable, Interval
>>> var = BaseVariable("age", Interval(0, 120))
>>> var.is_valid(25)
True
>>> var.is_valid(-5)
False
>>> var.is_valid("twenty")  # Non-numeric string
False

Notes

This method performs the same validation as the value setter but without actually setting the value. Useful for pre-checking values before assignment.

See also

value

Property setter that uses the same validation.

to_dict() → dict

Convert variable to dictionary representation.

Returns:

Dictionary containing variable metadata and current state.

Return type:

dict

Examples

>>> from causaloop import BaseVariable, Interval
>>> var = BaseVariable("pressure", Interval(0, 100), 85.5, "kPa")
>>> var.to_dict()
{
    'name': 'pressure',
    'type': 'continuous',
    'value': 85.5,
    'units': 'kPa',
    'domain': 'Interval(0, 100)'
}

Notes

This representation is suitable for serialization (JSON, YAML) and persistence. The domain is represented as its string representation for compactness.

property type: VariableType

Infer the variable’s data type from its domain.

Returns:

The variable type inferred from the domain class.

Return type:

VariableType

Notes

Type inference is based on the domain class hierarchy: - Interval → CONTINUOUS - Categorical → CATEGORICAL - Boolean → BOOLEAN - All others → DISCRETE

This inference is used for serialization, visualization, and automatic solver selection.

Examples

>>> from causaloop import BaseVariable, Interval, Boolean, Categorical
>>> var = BaseVariable("temp", Interval(0, 100), 25)
>>> var.type
<VariableType.CONTINUOUS: 'continuous'>
>>> var = BaseVariable("flag", Boolean(), True)
>>> var.type
<VariableType.BOOLEAN: 'boolean'>
>>> var = BaseVariable("category", Categorical(["A", "B"]), "A")
>>> var.type
<VariableType.CATEGORICAL: 'categorical'>

property value: Any | None

Get the current value of the variable.

Returns:

The current value, or None if the variable is uninitialized.

Return type:

Optional[Any]

Examples

>>> from causaloop import BaseVariable, Interval, Categorical
>>> var = BaseVariable("count", Interval(0, 100), 42.)
>>> var.value
42.0
>>> var = BaseVariable("status", Categorical(["on", "off"]))
>>> var.value is not None
False

class causaloop.core.Boolean

Binary true/false domain.

Represents variables that can take only boolean values (True/False) or their integer equivalents (1/0).

Examples

>>> from causaloop import Boolean
>>> domain = Boolean()
>>> domain.validate(True)
True
>>> domain.validate(1)
True
>>> domain.validate(0)
True
>>> domain.validate(False)
True
>>> domain.validate(2)
False
>>> domain.validate("true")
False

See also

Domain

Base class for all domains.

validate(value: Any) → bool

Check if a value is a valid boolean representation.

Parameters:

value (Any) – The value to validate as boolean.

Returns:

True if value is boolean (True/False) or integer (0/1), False otherwise.

Return type:

bool

Notes

Only accepts Python bool type or integers 0/1. Does not accept string representations like “true” or “false”.

Examples

>>> from causaloop import Boolean
>>> domain = Boolean()
>>> domain.validate(True)
True
>>> domain.validate(1)
True
>>> domain.validate(0.0)
False
>>> domain.validate(2)
False
>>> domain.validate(None)
False

class causaloop.core.Categorical(categories: List[Any])

Domain of unordered categorical values.

Represents variables that can take values from a finite set of discrete categories. Categories can be of any hashable type.

Parameters:

categories (List[Any]) – List of valid categorical values. Values should be hashable.

categories

The list of valid categorical values.

Type:

List[Any]

Examples

>>> from causaloop import Categorical
>>> domain = Categorical(["cat", "dog", "bird"])
>>> domain.validate("cat")
True
>>> domain.validate("fish")
False
>>> domain = Categorical([1, 2, 3])
>>> domain.validate(2)
True

See also

Domain

Base class for all domains.

validate(value: Any) → bool

Check if a value is in the list of valid categories.

Parameters:

value (Any) – The value to check for category membership.

Returns:

True if value is in the categories list, False otherwise.

Return type:

bool

Notes

Uses exact equality comparison (==), not type conversion. Case-sensitive for strings.

Examples

>>> from causaloop import Categorical
>>> domain = Categorical(["red", "green", "blue"])
>>> domain.validate("red")
True
>>> domain.validate("Red")  # Case sensitive
False
>>> domain.validate("yellow")
False

exception causaloop.core.CausalError

Exception raised for causal consistency violations.

This exception indicates that a causal operation would violate temporal consistency, create causal loops, or otherwise break the causal integrity of the system.

Examples

>>> from causaloop import Interval, CausalVariable, CausalError
>>> var = CausalVariable("temp", Interval(0, 100))
>>> var.update("heater", 25.0, time=10.0)
'update_id_1'
>>> try:
...     var.update("heater", 30.0, time=9.0)  # Earlier time!
... except CausalError as e:
...     print(e)
Temporal violation: heater at 9.0 (last update at 10.0)

class causaloop.core.CausalUpdate(source: str, value: ~typing.Any, timestamp: float, id: str = '', deps: ~typing.List[str] = <factory>, status: ~causaloop.core.variable.types.TemporalStatus = TemporalStatus.CONSEQUENT, confidence: float = 1.0, meta: ~typing.Dict[str, ~typing.Any] = <factory>)

Record of a causal update with full temporal and dependency metadata.

A CausalUpdate represents a single causal event where a variable’s value changes due to a specific mechanism. It captures not just the new value, but the complete causal context: when it happened, what caused it, and with what confidence.

Parameters:

• source (str) – Identifier of the mechanism or process that caused this update (e.g., “metabolism”, “insulin_injection”, “sensor_reading”).

• value (Any) – The new value assigned by this update.

• timestamp (float) – Simulation time when this update occurred. Must be monotonically non-decreasing for each source to maintain causal consistency.

• id (str, optional) – Unique identifier for this update. Auto-generated if not provided.

• deps (List[str], optional) – List of causal IDs that this update depends on. Forms the causal dependency graph when combined across updates.

• status (TemporalStatus, default=TemporalStatus.CONSEQUENT) – Temporal status indicating the update’s position in causal ordering.

• confidence (float, default=1.0) – Confidence level (0.0 to 1.0) in this update’s value. Used in weighted resolution strategies.

• meta (Dict[str, Any], optional) – Additional metadata for domain-specific information or analysis.

source

Source mechanism identifier.

Type:

str

value

The update value.

Type:

Any

timestamp

Update timestamp.

Type:

float

id

Unique causal identifier.

Type:

str

deps

Causal dependencies.

Type:

List[str]

status

Temporal status.

Type:

TemporalStatus

confidence

Update confidence.

Type:

float

meta

Additional metadata.

Type:

Dict[str, Any]

Examples

>>> from causaloop import CausalUpdate
>>> update = CausalUpdate(
...     source="metabolism",
...     value=95.0,
...     timestamp=10.5,
...     deps=["prev_update_id"],
...     confidence=0.95
... )
>>> update.id  # Auto-generated
'a1b2c3d4e5f6g7h8'
>>> update.timestamp
10.5

Notes

CausalUpdates form the building blocks of causal reasoning. By tracking dependencies between updates, the system can reconstruct causal pathways and detect inconsistencies.

The auto-generated ID uses SHA-256 hashing for uniqueness while keeping the identifier reasonably short. For extremely high-frequency systems, consider providing custom IDs to avoid collisions.

confidence: float = 1.0

deps: List[str]

id: str = ''

meta: Dict[str, Any]

source: str

status: TemporalStatus = 2

timestamp: float

value: Any

class causaloop.core.CausalVariable(name: str, domain: Domain, value: Any | None = None, units: str | None = None, noise: Callable[[Generator, int], ndarray] | None = None, rng: Generator | None = None, max_history: int = 1000)

Variable with causal-temporal tracking and counterfactual reasoning.

CausalVariable extends Variable with the ability to track the complete causal history of value changes, enforce temporal consistency, and support counterfactual reasoning. It maintains a causal graph of updates and provides methods for analyzing causal pathways.

Key features: 1. Temporal consistency: Prevents retroactive causation and causal loops 2. Causal history: Complete record of all updates with metadata 3. Counterfactual reasoning: Alternative causal scenarios for what-if analysis 4. Causal resolution: Value determination considering temporal context 5. Dependency tracking: Explicit causal dependencies between updates

Parameters:

• name (str) – Variable identifier.

• domain (Domain) – Value domain with validation rules.

• value (Any, optional) – Initial deterministic value.

• units (str, optional) – Measurement units.

• noise (Distribution, optional) – Noise model for stochastic perturbations.

• rng (np.random.Generator, optional) – Random number generator for noise (required if noise specified).

• max_history (int, default=1000) – Maximum number of causal updates to retain in history. Older updates are discarded when this limit is exceeded.

max_history

Maximum causal history length.

Type:

int

_history

Complete causal history in chronological order.

Type:

List[CausalUpdate]

_constraints

Temporal consistency constraints.

Type:

List[Callable]

_pending

Updates pending causal resolution.

Type:

Dict[str, CausalUpdate]

_last_time

Timestamp of most recent update.

Type:

float

_branches

Counterfactual causal branches.

Type:

Dict[str, List[CausalUpdate]]

Examples

>>> import numpy as np
>>> from causaloop import CausalVariable, Interval

>>> # Create causal variable
>>> var = CausalVariable("glucose", Interval(50, 300), 95.0, "mg/dL")

>>> # Record causal updates
>>> update_id = var.update(
...     source="metabolism",
...     value=95.0,
...     time=10.0,
...     deps=["previous_event"],
...     confidence=0.95
... )

>>> # Query causal history
>>> history = var.history(start=0, end=20)
>>> len(history)
1

>>> # Create counterfactual scenario
>>> var.branch(
...     branch_id="insulin_intervention",
...     time=10.0,
...     value=85.0,
...     source="intervention"
... )

>>> # Query counterfactual value
>>> cf_value = var.get_branch_value("insulin_intervention", time=11.0)

See also

Variable

Parent class with deterministic resolution and noise handling.

CausalUpdate

Individual causal event record.

add_constraint(constraint: Callable[[CausalVariable, str, float], bool]) → None

Add a temporal consistency constraint.

Parameters:

constraint (Callable) – Constraint function with signature: constraint(variable: CausalVariable, source: str, time: float) -> bool Should return True if the update is allowed, False otherwise.

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("pressure", Interval(0, 100))

>>> # Constraint: updates from "valve" can only occur at integer times
>>> def integer_time_constraint(var, source, time):
...     if source == "valve":
...         return time.is_integer()
...     return True

>>> var.add_constraint(integer_time_constraint)

>>> # This will work
>>> var.update("valve", 50.0, time=10.0)
'update_id'

>>> # This will fail the constraint
>>> try:
...     var.update("valve", 60.0, time=10.5)
... except CausalError:
...     print("Constraint violation")
Constraint violation

Notes

Constraints allow domain-specific temporal logic to be enforced. Common uses include: - Minimum time between updates from certain sources - Synchronization requirements between variables - Domain-specific timing patterns - Resource availability constraints

Constraints are checked in registration order. The first constraint that returns False prevents the update.

For performance, keep constraint functions simple and fast.

branch(branch_id: str, time: float, value: Any, source: str = 'counterfactual') → None

Create a counterfactual causal branch.

Parameters:

• branch_id (str) – Unique identifier for this counterfactual branch.

• time (float) – Intervention time where the counterfactual diverges from actual history.

• value (Any) – Counterfactual intervention value at the divergence point.

• source (str, default="counterfactual") – Source identifier for the counterfactual intervention.

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("glucose", Interval(50, 300), 95.0, "mg/dL")
>>> var.update("metabolism", 95.0, time=9.0)
'id1'
>>> var.update("meal", 150.0, time=10.0)
'id2'

>>> # Create counterfactual: what if insulin was given instead of meal?
>>> var.branch(
...     branch_id="insulin_intervention",
...     time=10.0,  # Same time as the meal
...     value=85.0,  # Insulin would lower glucose
...     source="insulin_injection"
... )

>>> # Query counterfactual value at later time
>>> cf_value = var.get_branch_value("insulin_intervention", time=11.0)

Notes

Counterfactual branches allow what-if analysis by creating alternative causal histories. Each branch: 1. Copies all actual history up to the intervention time 2. Adds the counterfactual intervention as a new update 3. Can be extended with additional hypothetical updates

Branches are stored separately from actual history and don’t affect the variable’s current value or normal operations.

Multiple branches can be created for different intervention scenarios. Branch IDs must be unique within a variable.

clear_pending() → None

Clear pending updates without affecting history.

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("output", Interval(0, 100))
>>> var.update("process", 50.0, time=10.0)
'id1'
>>> len(var._pending)
1

>>> var.clear_pending()
>>> len(var._pending)
0
>>> len(var._history)  # History preserved
1

Notes

Pending updates are used for current value resolution but can be cleared to reset the variable’s “current state” while preserving historical records. This is useful between simulation runs or analysis phases.

Clearing pending updates does not affect the causal history or counterfactual branches.

get_branch_value(branch_id: str, time: float, strategy: str = 'causal') → Any | None

Get value in a counterfactual branch at specified time.

Parameters:

• branch_id (str) – Counterfactual branch identifier.

• time (float) – Time to query in the counterfactual scenario.

• strategy (str, default="causal") – Resolution strategy for determining the value. See _resolve_causal for available strategies.

Returns:

Value in the counterfactual scenario at the specified time, or None if the branch doesn’t exist or has no relevant updates.

Return type:

Optional[Any]

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100), 20.0)
>>> var.update("heater", 25.0, time=10.0)
'id1'

>>> # Create counterfactual: what if heater was stronger?
>>> var.branch("stronger_heater", time=10.0, value=30.0)

>>> # Query counterfactual value
>>> cf_value = var.get_branch_value("stronger_heater", time=11.0)
>>> cf_value
30.0

Notes

This method reconstructs the counterfactual state by: 1. Finding all updates in the branch up to the query time 2. Applying the specified resolution strategy to those updates

The resolution considers only updates within the branch, ignoring actual history after the intervention point.

For complex counterfactuals with multiple hypothetical updates after the intervention, additional updates would need to be added to the branch manually.

get_causal_path(start_time: float, end_time: float) → List[CausalUpdate]

Get the causal pathway between two times.

Parameters:

• start_time (float) – Start time for causal pathway.

• end_time (float) – End time for causal pathway.

Returns:

Updates in the causal pathway, in chronological order.

Return type:

List[CausalUpdate]

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("signal", Interval(0, 10))
>>> var.update("A", 1.0, time=1.0, deps=[])
'id1'
>>> var.update("B", 2.0, time=2.0, deps=["id1"])
'id2'
>>> var.update("C", 3.0, time=3.0, deps=["id2"])
'id3'

>>> # Get causal pathway
>>> path = var.get_causal_path(start_time=1.0, end_time=3.0)
>>> [u.source for u in path]
['A', 'B', 'C']

Notes

A causal pathway is a sequence of updates where each update (after the first) depends on the previous one. This method reconstructs such pathways by following dependency links.

If multiple dependency paths exist, this method returns one valid path (not necessarily the shortest or only one).

Circular dependencies will cause infinite recursion.

get_most_recent(source: str | None = None) → CausalUpdate | None

Get the most recent causal update.

Parameters:

source (str, optional) – If provided, returns most recent update from this specific source.

Returns:

Most recent update matching criteria, or None if no updates exist.

Return type:

Optional[CausalUpdate]

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100))
>>> var.update("sensor1", 25.0, time=10.0)
'id1'
>>> var.update("sensor2", 26.0, time=11.0)
'id2'

>>> # Most recent overall
>>> recent = var.get_most_recent()
>>> recent.source
'sensor2'
>>> recent.timestamp
11.0

>>> # Most recent from specific source
>>> sensor1_recent = var.get_most_recent(source="sensor1")
>>> sensor1_recent.value
25.0

Notes

This method is optimized for frequent access by tracking the most recent update time internally. For source-specific queries, it performs a reverse linear search through history.

For variables with very long histories, consider maintaining source-specific pointers for O(1) access.

history(start: float = 0.0, end: float | None = None, source: str | None = None) → List[CausalUpdate]

Retrieve causal history with optional filtering.

Parameters:

• start (float, default=0.0) – Start time (inclusive) for history query.

• end (float, optional) – End time (inclusive) for history query. If None, includes all updates from start onward.

• source (str, optional) – If provided, only returns updates from this source.

Returns:

List of causal updates matching the filter criteria, in chronological order.

Return type:

List[CausalUpdate]

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("pressure", Interval(0, 100))
>>> var.update("valve", 50.0, time=10.0)
'id1'
>>> var.update("pump", 60.0, time=12.0)
'id2'
>>> var.update("valve", 55.0, time=15.0)
'id3'

>>> # All history
>>> all_history = var.history()
>>> len(all_history)
3

>>> # Time window
>>> window = var.history(start=11.0, end=14.0)
>>> len(window)
1
>>> window[0].source
'pump'

>>> # Source-specific
>>> valve_history = var.history(source="valve")
>>> len(valve_history)
2

Notes

The history is stored in chronological order, so this method performs linear filtering. For large histories with frequent queries, consider maintaining additional indexing structures.

If max_history is set and old updates have been discarded, queries for times before the oldest retained update will return empty results.

reset() → None

Reset all causal tracking (history, branches, pending updates).

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("signal", Interval(0, 10))
>>> var.update("source", 5.0, time=1.0)
'id1'
>>> var.branch("counterfactual", time=1.0, value=6.0)

>>> len(var._history)
1
>>> len(var._branches)
1

>>> var.reset()
>>> len(var._history)
0
>>> len(var._branches)
0
>>> var._last_time
0.0

Notes

This method completely resets causal tracking while preserving the variable’s basic configuration (domain, units, noise model).

Use this method to start fresh causal tracking, such as when beginning a new simulation or analysis from time zero.

Source registrations and weights from the parent class are preserved. Only causal-specific state is reset.

resolve_causal(strategy: str = 'causal') → Any

Get value using specified causal resolution strategy.

Parameters:

strategy (str, default="causal") – Resolution strategy (see _resolve_causal for options).

Returns:

Resolved value according to specified strategy.

Return type:

Any

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("signal", Interval(0, 10))
>>> var.update("source1", 5.0, time=1.0, confidence=0.8)
'id1'
>>> var.update("source2", 7.0, time=2.0, confidence=0.9)
'id2'

>>> # Different resolution strategies
>>> var.resolve_causal("causal")  # Weighted by time and confidence
np.float64(6.10...)
>>> var.resolve_causal("most_recent")  # Most recent update
7.0
>>> var.resolve_causal("most_confident")  # Highest confidence
7.0

Notes

This method provides explicit control over causal resolution, which can be important for different analysis purposes:

• “causal”: For simulation where recent, confident updates dominate

• “most_recent”: For real-time monitoring or state tracking

• “most_confident”: When reliability is more important than recency

to_dict() → Dict

Convert variable to detailed dictionary representation.

Returns:

Dictionary containing variable metadata, source information, causal history summary, and configuration.

Return type:

Dict

Notes

The full causal history can be large, so this method only includes summary statistics. For complete serialization of causal state, additional methods would be needed to export/import the full history and branches.

Noise models and RNGs require special handling for serialization and are not included in the basic dictionary representation.

update(source: str, value: Any, time: float, deps: List[str] | None = None, confidence: float = 1.0, meta: Dict[str, Any] | None = None) → str

Record a causal update with temporal consistency checking.

Parameters:

• source (str) – Source mechanism causing this update.

• value (Any) – New deterministic value (must satisfy domain constraints).

• time (float) – Simulation time when this update occurs.

• deps (List[str], optional) – List of causal IDs that this update depends on.

• confidence (float, default=1.0) – Confidence in this update (0.0 to 1.0).

• meta (Dict[str, Any], optional) – Additional metadata for analysis or domain-specific information.

Returns:

Unique causal ID of the created update.

Return type:

str

Raises:

• CausalError – If the update violates temporal consistency constraints.

• ValueError – If the value violates domain constraints.

Examples

>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100))

>>> # Simple update
>>> update_id = var.update("heater", 25.0, time=10.0)

>>> # Update with dependencies
>>> update_id2 = var.update(
...     source="thermostat",
...     value=26.0,
...     time=11.0,
...     deps=[update_id],  # Depends on previous update
...     confidence=0.9,
...     meta={"room": "living_room"}
... )

>>> # Invalid: retroactive update
>>> try:
...     var.update("heater", 30.0, time=9.0)
... except CausalError as e:
...     print(e)
Temporal violation: heater at 9.0 (last update at 10.0)

Notes

This method enforces causal consistency by: 1. Checking that time >= last update time for this source 2. Validating against all registered temporal constraints 3. Recording dependencies to build the causal graph

The update is stored in causal history and also registered as a normal source output for deterministic value resolution.

Dependencies should reference valid causal IDs from previous updates. Circular dependencies are not automatically detected but will cause issues in causal analysis.

property value: Any

Get current value using causal-aware resolution.

Returns:

Current deterministic value resolved using causal strategy.

Return type:

Any

Notes

The causal resolution strategy considers: 1. Temporal recency: More recent updates have higher weight 2. Confidence: Updates with higher confidence have higher weight 3. Source-specific weighting: As configured via add_source()

The resolution uses exponential decay for temporal weighting:

weight = confidence * exp(-decay_rate * time_since_update)

This gives a smooth transition between updates while respecting causal ordering.

For the pure deterministic value without causal weighting, use the parent class’s resolution directly:

>>> det_value = super(CausalVariable, var).value

Or access the most recent update:

>>> recent = var.get_most_recent()
>>> det_value = recent.value if recent else None

class causaloop.core.CompositeNoise(models: list = <factory>, weights: list | None = None)

Composite noise model combining multiple noise sources.

Parameters:

• models (list) – List of noise models to combine.

• weights (list, optional) – Relative weights for each model. If None, equal weights.

Examples

>>> from causaloop import GaussianNoise, UniformNoise, CompositeNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> models = [
...     GaussianNoise(mean=0, std=0.1),
...     UniformNoise(low=-0.05, high=0.05)
... ]
>>> noise = CompositeNoise(models=models, weights=[0.7, 0.3])
>>> noise(rng, 3)
array([ 0.02..., -0.08...,  0.06...])

models: list

weights: list | None = None

class causaloop.core.Domain

Abstract base class for variable domains.

A Domain defines the set of valid values a variable can take and provides validation logic. Concrete subclasses implement specific types of domains (numeric intervals, categorical sets, etc.).

Subclasses must implement the validate method and may override __repr__ for better debugging output.

See also

Interval

Continuous numeric range domain.

Categorical

Finite set of unordered categories.

Boolean

Binary true/false domain.

abstractmethod validate(value: Any) → bool

Check if a value is valid for this domain.

Parameters:

value (Any) – The value to check for domain membership.

Returns:

True if the value is within the domain’s definition, False otherwise.

Return type:

bool

Notes

Implementations should handle type conversion gracefully where appropriate. For example, numeric domains should attempt to convert string representations to numbers before validation.

Raises:

NotImplementedError – When called directly on the abstract base class.

class causaloop.core.DomainProtocol(*args, **kwargs)

Protocol defining the interface for variable domain implementations.

A Domain defines the valid value space for a variable and provides validation logic. This protocol ensures all domain implementations provide consistent validation and representation.

validate(value: Any) → bool

Check if a value is within the domain’s definition.

__repr__() → str

Return a string representation of the domain.

See also

Domain

Abstract base class implementing this protocol.

Interval

Continuous numeric domain implementation.

Categorical

Discrete categorical domain implementation.

Boolean

Binary domain implementation.

validate(value: Any) → bool

Validate that a value belongs to this domain.

Parameters:

value (Any) – The value to validate against the domain’s constraints.

Returns:

True if the value is valid for this domain, False otherwise.

Return type:

bool

Notes

Validation should be permissive where possible, attempting type conversions for numeric domains. For example, an Interval domain should accept string representations of numbers.

class causaloop.core.GaussianNoise(mean: float = 0.0, std: float = 1.0)

Gaussian (normal) noise model.

Parameters:

• mean (float, default=0.0) – Mean of the Gaussian distribution.

• std (float, default=1.0) – Standard deviation of the Gaussian distribution.

Examples

>>> from causaloop import GaussianNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = GaussianNoise(mean=0, std=0.1)
>>> noise(rng)
array([0.030...])

mean: float = 0.0

std: float = 1.0

class causaloop.core.Interval(low: float = -inf, high: float = inf)

Continuous numeric domain defined by lower and upper bounds.

Represents variables that can take any real value within a specified range [low, high]. The bounds can be infinite for unbounded domains.

Parameters:

• low (float, optional) – Lower bound of the interval (inclusive). Default is -inf.

• high (float, optional) – Upper bound of the interval (inclusive). Default is +inf.

low

Lower bound of the interval.

Type:

float

high

Upper bound of the interval.

Type:

float

Examples

>>> from causaloop import Interval
>>> domain = Interval(0, 100)
>>> domain.validate(50)
True
>>> domain.validate(-10)
False
>>> domain.validate("75")  # String representation
True

See also

Domain

Base class for all domains.

validate(value: Any) → bool

Check if a value is within the interval bounds.

Parameters:

value (Any) – The value to validate. Can be numeric or string representation.

Returns:

True if value can be converted to float and lies within [low, high] (inclusive), False otherwise.

Return type:

bool

Notes

Attempts to convert the value to float before comparison. Returns False if conversion fails or value is outside bounds.

Examples

>>> from causaloop import Interval
>>> domain = Interval(0, 10)
>>> domain.validate(5)
True
>>> domain.validate(15)
False
>>> domain.validate("7.5")
True
>>> domain.validate("not a number")
False

class causaloop.core.LogNormalNoise(mean: float = 0.0, sigma: float = 1.0)

Log-normal noise model for positive-valued variables.

Parameters:

• mean (float, default=0.0) – Mean of the underlying normal distribution.

• sigma (float, default=1.0) – Standard deviation of the underlying normal distribution.

Examples

>>> from causaloop import LogNormalNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = LogNormalNoise(mean=0, sigma=0.1)
>>> noise(rng)
array([1.03...])

mean: float = 0.0

sigma: float = 1.0

class causaloop.core.PoissonNoise(lam: float = 1.0)

Poisson noise for count variables.

Parameters:

lam (float, default=1.0) – Expected value (lambda) of Poisson distribution.

Examples

>>> from causaloop import PoissonNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = PoissonNoise(lam=5.0)
>>> noise(rng)
array([8])

lam: float = 1.0

class causaloop.core.TemporalStatus(*values)

Enumeration of temporal positions in causal ordering.

These statuses track where a variable update sits in the causal-temporal sequence, crucial for maintaining causal consistency and enabling counterfactual reasoning.

ANTECEDENT

Update occurs before a causal event (in the causal past). Used for establishing baseline conditions.

Type:

auto

CONSEQUENT

Update occurs after a causal event (in the causal future). Represents effects or outcomes of causal mechanisms.

Type:

auto

PENDING

Update is scheduled but not yet resolved in the causal order. Used for asynchronous or concurrent process execution.

Type:

auto

VIOLATED

Update violates causal consistency constraints. Indicates a causal paradox or constraint violation.

Type:

auto

COUNTERFACTUAL

Update exists in a hypothetical counterfactual scenario. Used for what-if analysis and intervention testing.

Type:

auto

ANTECEDENT = 1

CONSEQUENT = 2

COUNTERFACTUAL = 5

PENDING = 3

VIOLATED = 4

class causaloop.core.UniformNoise(low: float = -1.0, high: float = 1.0)

Uniform noise model.

Parameters:

• low (float, default=-1.0) – Lower bound of uniform distribution.

• high (float, default=1.0) – Upper bound of uniform distribution.

Examples

>>> from causaloop import UniformNoise
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42)
>>> noise = UniformNoise(low=-0.5, high=0.5)
>>> noise(rng, 4)
array([ 0.27..., -0.06...,  0.35...,  0.19...])

high: float = 1.0

low: float = -1.0

class causaloop.core.Variable(name: str, domain: Domain, value: Any | None = None, units: str | None = None, noise: Callable[[Generator, int], ndarray] | None = None, rng: Generator | None = None)

Multi-source variable with deterministic resolution and optional stochastic noise.

Variable extends BaseVariable with the ability to receive values from multiple sources (e.g., mechanisms, processes, measurements) and provides configurable strategies for resolving conflicts. Following scientific convention, deterministic values and stochastic noise are clearly separated:

1. Deterministic value: Computed from source values using weighted averaging

2. Noise model: Optional stochastic perturbation applied separately

3. RNG: Explicit random number generator for reproducible noise

This separation ensures causal integrity while enabling flexible stochastic modeling. Users can choose when and how to apply noise based on their specific requirements.

Parameters:

• name (str) – Unique identifier for the variable (e.g., “temperature”, “glucose_level”).

• domain (Domain) – Mathematical domain defining valid values (e.g., Interval, Categorical).

• value (Any, optional) – Initial deterministic value. Must satisfy domain constraints if provided.

• units (str, optional) – Physical or logical units (e.g., “°C”, “mg/dL”, “count”).

• noise (Distribution, optional) – Noise model implementing the Distribution protocol. Signature: noise(rng: np.random.Generator, n: int) -> np.ndarray

• rng (np.random.Generator, optional) – Random number generator for reproducible stochasticity. Required if noise is specified.

noise

Noise model for stochastic perturbations.

Type:

Optional[Distribution]

rng

Random number generator for noise sampling.

Type:

Optional[np.random.Generator]

_sources

List of registered source identifiers.

Type:

List[str]

_weights

Source weights for weighted averaging.

Type:

Dict[str, float]

_source_values

Current values provided by each source.

Type:

Dict[str, Any]

Raises:

ValueError – If initial value violates domain constraints, or if noise is specified without providing an RNG.

Examples

>>> from causaloop import Variable, Interval
>>> import numpy as np

>>> # Create deterministic variable
>>> var = Variable("temperature", Interval(0, 100), 25.0, "°C")
>>> var.value  # Deterministic value
25.0

>>> # Create variable with noise model
>>> def gaussian_noise(rng: np.random.Generator, n: int) -> np.ndarray:
...     return rng.normal(loc=0.0, scale=0.1, size=n)
>>>
>>> rng = np.random.default_rng(seed=42)  # For reproducibility
>>> noisy_var = Variable("signal", Interval(-1, 1), 0.0,
...                      noise=gaussian_noise, rng=rng)
>>>
>>> # Deterministic value (always the same)
>>> noisy_var.value
0.0
>>>
>>> # Apply noise (different each call due to RNG state)
>>> noisy_var.apply_noise()
(0.0, 0.03..)
>>>
>>> # Sample noise separately
>>> noise_samples = noisy_var.sample_noise(n=5)
>>> noise_samples.shape
(5,)

See also

BaseVariable

Parent class with domain validation.

CausalVariable

Extension with temporal causal tracking.

property active_sources: List[str]

Get list of sources that have provided current values.

Returns:

List of source identifiers that have non-None values.

Return type:

List[str]

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("score", Interval(0, 100))
>>> var.add_source("judge1")
Variable('score', value=None, sources=1)
>>> var.add_source("judge2")
Variable('score', value=None, sources=2)
>>> var.register_output("judge1", 85)
>>> var.active_sources
['judge1']

Notes

This property helps distinguish between potential influences (all registered sources) and actual current influences (active sources).

Inactive sources (registered but not providing values) don’t affect the variable’s current deterministic value.

add_source(source: str, weight: float = 1.0) → Variable

Register a potential source for this variable.

Parameters:

• source (str) – Unique identifier for the source (e.g., “metabolism”, “sensor_A”, “process_1”). Should be descriptive and consistent.

• weight (non-negative float, default=1.0) – Relative weight of this source in weighted averaging. Higher weights give the source more influence. Weights are relative, not normalized.

Returns:

Self for method chaining.

Return type:

Variable

Raises:

ValueError – If weight is negative.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("process_A", weight=2.0)
Variable('output', value=None, sources=1)
>>> var.add_source("process_B", weight=1.0)
Variable('output', value=None, sources=2)
>>> var.get_source_weight("process_A")
2.0

Notes

Sources can be registered before they provide values, allowing pre-configuration of the variable’s dependency structure. This is useful for declaring expected influences in complex systems.

Weights are used in the default “weighted” resolution strategy:

weighted_sum = Σ(weight_i * value_i) total_weight = Σ(weight_i) result = weighted_sum / total_weight

Weight values are relative: a source with weight 2.0 has twice the influence of a source with weight 1.0.

apply_noise(value: Any | None = None) → Tuple[Any, Any]

Apply noise to a value, returning both deterministic and noisy results.

Parameters:

value (Any, optional) – Value to apply noise to. If None, uses the variable’s current deterministic value.

Returns:

Tuple containing (deterministic_value, noisy_value). If no noise model is configured, returns (value, value).

Return type:

Tuple[Any, Any]

Raises:

ValueError – If the provided value violates domain constraints.

Examples

>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> def uniform_noise(rng, n):
...     return rng.uniform(-0.1, 0.1, n)
>>>
>>> rng = np.random.default_rng(seed=42)
>>> var = Variable("measurement", Interval(0, 10),
...                noise=uniform_noise, rng=rng)
>>> var.register_output("sensor", 5.0)
>>>
>>> # Apply noise to current value
>>> deterministic, noisy = var.apply_noise()
>>> deterministic
5.0
>>> noisy  # 5.0 + noise
5.05...
>>>
>>> # Apply noise to specific value
>>> var.apply_noise(7.5)
(7.5, 7.48...)
>>>
>>> # Deterministic variable
>>> det_var = Variable("constant", Interval(0, 1), 0.5)
>>> det_var.apply_noise()  # No noise model
(0.5, 0.5)

Notes

This method follows the scientific convention of clearly separating deterministic values from stochastic perturbations. It always returns both components, allowing users to:

1. Track the underlying deterministic signal

2. Apply domain-specific handling of noisy values

3. Compute error statistics (noise = noisy - deterministic)

4. Implement custom noise handling (clipping, rejection, etc.)

The method validates that the input value satisfies domain constraints but does NOT validate the noisy result. Users are responsible for handling cases where noise pushes values outside valid domains.

This design supports flexible modeling strategies: - Measurement error: deterministic = true value, noisy = observed - Process noise: deterministic = expected, noisy = realized - Stochastic processes: deterministic = mean, noisy = sample

clear_sources() → None

Clear all source values while preserving source registrations.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("temperature", Interval(0, 100))
>>> var.register_output("sensor1", 25.0)
>>> var.register_output("sensor2", 26.0)
>>> var.value
25.5
>>> var.clear_sources()
>>> var.value is None
True
>>> var._sources  # Sources still registered
['sensor1', 'sensor2']
>>> var._source_values  # Values cleared
{}

Notes

This method clears only the current values from sources, not the source registrations themselves. This is useful for:

• Resetting variable state between simulation runs

• Clearing temporary measurements while keeping configurations

• Reusing variable structures with different input data

Source weights and noise models remain unchanged. To completely reset a variable, create a new instance.

get_deterministic_and_noisy() → Tuple[Any, Any]

Get both deterministic value and noisy version in one call.

Returns:

Tuple containing (deterministic_value, noisy_value). Convenience method equivalent to apply_noise(None).

Return type:

Tuple[Any, Any]

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("signal", Interval(-1, 1), noise=..., rng=...)
>>> var.register_output("source", 0.5)
>>> det, noisy = var.get_deterministic_and_noisy()
>>> det  # Deterministic value
0.5

Notes

This is a convenience method for the common pattern of wanting both the deterministic value and a noisy version. It’s equivalent to:

>>> det, noisy = var.apply_noise(var.value)

The method always returns a tuple, even for deterministic variables (where both elements are identical).

get_source_value(source: str) → Any | None

Get the current value from a specific source.

Parameters:

source (str) – Source identifier.

Returns:

Current deterministic value from the specified source, or None if the source hasn’t provided a value.

Return type:

Optional[Any]

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("pressure", Interval(0, 100))
>>> var.register_output("sensor_A", 85.5)
>>> var.get_source_value("sensor_A")
85.5
>>> var.get_source_value("sensor_B") is None  # Not provided
True

Notes

This method retrieves the raw value provided by a specific source, before any resolution or noise application. It’s useful for:

• Debugging source contributions

• Implementing custom resolution logic

• Monitoring individual component outputs

• Validating source behavior

get_source_weight(source: str) → float

Get the weight of a registered source.

Parameters:

source (str) – Source identifier.

Returns:

Current weight of the source. Returns 0.0 if source is not registered.

Return type:

float

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("process1", weight=2.0)
Variable('output', value=None, sources=1)
>>> var.get_source_weight("process1")
2.0
>>> var.get_source_weight("unknown_process")  # Not registered
0.0

Notes

Source weights determine influence in weighted averaging. A weight of 0.0 effectively disables a source’s contribution (unless using other strategy).

Weights are relative, not normalized. Changing one source’s weight doesn’t automatically adjust others.

property has_noise: bool

Check if the variable has a configured noise model.

Returns:

True if a noise model and RNG are configured, False otherwise.

Return type:

bool

Examples

>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> var = Variable("signal", Interval(-1, 1))
>>> var.has_noise
False

>>> noisy_var = Variable("noisy_signal", Interval(-1, 1),
...                      noise=lambda rng, n: rng.normal(0, 0.1, n),
...                      rng=np.random.default_rng())
>>> noisy_var.has_noise
True

Notes

This property checks both the presence of a noise model and a valid RNG. A variable with a noise model but no RNG is considered invalid and will raise an error during initialization.

register_output(source: str, value: Any) → None

Register a deterministic value from a specific source.

Parameters:

• source (str) – Source identifier that generated this value.

• value (Any) – The deterministic value to register. Must satisfy all domain constraints.

Raises:

ValueError – If the value violates domain constraints.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("glucose", Interval(50, 300))
>>> var.add_source("metabolism", weight=1.0)
Variable('glucose', value=None, sources=1)
>>> var.register_output("metabolism", 95.0)
>>> var.value  # Deterministic value
95.0

>>> # Multiple sources with different weights
>>> var.add_source("diet", weight=0.5)
Variable('glucose', value=95.0, sources=2)
>>> var.register_output("diet", 105.0)
>>> var.value  # Weighted average: (1.0*95 + 0.5*105) / 1.5
98.333...

Notes

This method updates the variable’s deterministic value immediately based on all registered source values. The computation is purely deterministic and does not involve any stochastic components.

The method performs full domain validation to ensure mathematical consistency. If validation fails, the variable’s state remains unchanged.

Sources are automatically registered if not already present (with default weight 1.0). This allows flexible usage patterns.

resolve(strategy: str = 'weighted') → Any

Get deterministic value using specified resolution strategy.

Parameters:

strategy (str, default="weighted") – Resolution strategy for handling multiple source values: - “weighted”: Weighted average (recommended for most applications) - “latest”: Most recently registered value - “priority”: Value from source with highest weight

Returns:

Deterministic value according to the specified strategy.

Return type:

Any

Raises:

ValueError – If an unknown resolution strategy is specified.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("score", Interval(0, 100))
>>> var.register_output("judge1", 85)
>>> var.register_output("judge2", 90)
>>> var.add_source("judge1", weight=2.0)
Variable('score', value=86.66..., sources=2)

>>> # Different resolution strategies
>>> var.resolve("weighted")  # (2*85 + 1*90) / 3
86.666...
>>> var.resolve("latest")    # Most recent value
90
>>> var.resolve("priority")  # Judge1 has higher weight
85

Notes

This method provides explicit control over the resolution strategy, which can be important in specific modeling contexts:

• “weighted”: Appropriate when sources have different reliabilities

• “latest”: Useful for temporal sequences or state updates

• “priority”: When certain sources should dominate others

All strategies produce deterministic results. For stochastic perturbations, use apply_noise() on the resolved value.

sample_noise(n: int = 1) → ndarray

Sample noise from the configured noise model.

Parameters:

n (int, default=1) – Number of independent noise samples to generate.

Returns:

Array of noise samples with shape (n,). Returns zeros if no noise model is configured.

Return type:

np.ndarray

Raises:

RuntimeError – If noise model is configured but RNG is not available.

Examples

>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> def gaussian_noise(rng, n):
...     return rng.normal(0, 0.1, n)
>>>
>>> rng = np.random.default_rng(seed=42)
>>> var = Variable("signal", Interval(-1, 1),
...                noise=gaussian_noise, rng=rng)
>>>
>>> # Single noise sample
>>> var.sample_noise()
array([0.03...])
>>>
>>> # Multiple samples
>>> print(var.sample_noise(5))
[-0.10...  0.07...  0.09... -0.19... -0.13...]
>>>
>>> # Variable without noise model
>>> deterministic_var = Variable("constant", Interval(0, 1), 0.5)
>>> deterministic_var.sample_noise(3)  # Returns zeros
array([0., 0., 0.])

Notes

This method samples noise independently of the variable’s current value. The noise samples can be applied to any value using addition:

>>> value = 5.0
>>> noise = var.sample_noise()
>>> noisy_value = value + float(noise[0])

For variables without noise models, this method returns zeros, allowing uniform handling of both deterministic and stochastic variables in algorithms.

The RNG state advances with each call, ensuring proper stochastic behavior in sequential sampling.

set_source_weight(source: str, weight: float) → None

Set or update the weight of a source.

Parameters:

• source (str) – Source identifier.

• weight (non-negative float) – New weight for the source.

Raises:

• ValueError – If weight is negative.

• KeyError – If source is not registered.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("process1", weight=1.0)
Variable('output', value=None, sources=1)
>>> var.set_source_weight("process1", 3.0)
>>> var.get_source_weight("process1")
3.0

Notes

Changing source weights affects future value resolution but doesn’t change the variable’s current value. The new weight takes effect on the next value computation.

To immediately update the variable’s value with new weights, call register_output again or access the value property.

property sources: List[str]

Get list of registered sources.

Returns:

List of source identifiers in registration order.

Return type:

List[str]

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("temperature", Interval(0, 100))
>>> var.add_source("indoor_sensor")
Variable('temperature', value=None, sources=1)
>>> var.add_source("outdoor_sensor")
Variable('temperature', value=None, sources=2)
>>> var.sources
['indoor_sensor', 'outdoor_sensor']

Notes

The order of sources in this list reflects registration order, which may be relevant for “latest” resolution strategy or for understanding the variable’s dependency structure.

to_dict() → Dict

Convert variable to detailed dictionary representation.

Returns:

Dictionary containing variable metadata, source information, deterministic value, and noise configuration.

Return type:

Dict

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("proc1", weight=2.0)
Variable('output', value=None, sources=1)
>>> var.add_source("proc2", weight=1.0)
Variable('output', value=None, sources=2)
>>> var.register_output("proc1", 75.0)
>>> var.to_dict()
{
    'name': 'output',
    'type': 'continuous',
    'value': 75.0,
    'units': None,
    'domain': 'Interval(0, 100)',
    'sources': ['proc1', 'proc2'],
    'weights': {'proc1': 2.0, 'proc2': 1.0},
    'source_values': {'proc1': 75.0},
    'has_noise': False
}

Notes

This representation is suitable for serialization (JSON, YAML), persistence, and configuration. It captures all necessary information to reconstruct the variable’s state, though noise models and RNGs may require special handling for complete serialization.

The dictionary includes both structural information (sources, weights) and current state (value, source_values).

property value: Any

Get the deterministic value of the variable.

Returns:

Deterministic value resolved from all registered source values using the default “weighted” strategy.

Return type:

Any

Notes

This property returns the pure deterministic value without any stochastic perturbations. It represents the variable’s “true” value in the causal model, free from measurement error or process noise.

The value is recomputed on each access to reflect the current state of all source values. This ensures consistency in dynamic systems.

For variables with noise models, this property does NOT apply noise. Use apply_noise() or sample_noise() for stochastic values.

Examples

>>> from causaloop import Variable, Interval
>>> var = Variable("concentration", Interval(0, 100))
>>> var.register_output("source1", 30.0)
>>> var.register_output("source2", 40.0)
>>> var.value  # Weighted average: (30 + 40) / 2
35.0

>>> # Even with noise model, value is deterministic
>>> noisy_var = Variable("measurement", Interval(0, 10),
...                      noise=..., rng=...)
>>> noisy_var.register_output("sensor", 5.0)
>>> noisy_var.value  # Still 5.0, not 5.0 + noise
5.0

class causaloop.core.VariableType(*values)

Enumeration of variable data types in the causal modeling system.

These types determine how variables are validated, transformed, and interpreted during simulation and analysis.

CONTINUOUS

Real-valued variables with potentially infinite precision. Examples: temperature, pressure, concentration.

Type:

str

DISCRETE

Integer-valued or count variables with distinct, separate values. Examples: population count, inventory levels, event counts.

Type:

str

CATEGORICAL

Variables taking values from a finite set of unordered categories. Examples: species type, disease status, product category.

Type:

str

ORDINAL

Variables with ordered categories but unknown intervals between them. Examples: severity level (mild/moderate/severe), Likert scale responses.

Type:

str

BOOLEAN

Binary true/false or 0/1 variables. Examples: switch status, diagnostic test result, presence/absence.

Type:

str

BOOLEAN = 'boolean'

CATEGORICAL = 'categorical'

CONTINUOUS = 'continuous'

DISCRETE = 'discrete'

ORDINAL = 'ordinal'

causaloop.core.create_noise(noise_type: str = 'gaussian', **params) → NoiseModel

Create noise function.

Parameters:

• noise_type (str, default="gaussian") – Type of noise: “gaussian”, “uniform”

• **params – Parameters for noise model.

Returns:

Configured noise model.

Return type:

NoiseModel

Example

>>> from causaloop import create_noise
>>> import numpy as np
>>> noise = create_noise(noise_type="gaussian", **{'mean': 0.0, 'std': 1.0})
>>> rng = np.random.default_rng(seed=42)
>>> noise(rng, 100).shape
(100,)

Variables