Simulation#

class bamengine.Simulation(ec, config, rng, n_firms, n_households, n_banks, prod, wrk, emp, bor, lend, con, sh, pipeline, lb, n_periods, t, _role_instances=<factory>, extra_params=<factory>)[source]#

Bases: object

Main simulation facade for BAM Engine.

The Simulation class is the primary interface for running BAM (Bottom-Up Adaptive Macroeconomics) simulations. It manages the economy state, agent roles, event pipeline, and provides methods for stepping through periods.

Variables:

ec (Economy) – Global economy state (prices, wages, histories).
config (Config) – Configuration parameters for the simulation.
rng (numpy.random.Generator) – Random number generator for deterministic simulations.
n_firms (int) – Number of firms in the economy.
n_households (int) – Number of households in the economy.
n_banks (int) – Number of banks in the economy.
prod (Producer) – Producer role (firm production state).
wrk (Worker) – Worker role (household employment state).
emp (Employer) – Employer role (firm hiring state).
bor (Borrower) – Borrower role (firm financial state).
lend (Lender) – Lender role (bank state).
con (Consumer) – Consumer role (household consumption state).
sh (Shareholder) – Shareholder role (household per-period dividend income).
pipeline (Pipeline) – Event pipeline controlling simulation execution order.
lb (LoanBook) – Loan relationship between borrowers and lenders.
n_periods (int) – Default run length for run() method.
t (int) – Current period (starts at 0).
extra_params (dict[str, Any]) – Dictionary of extension-specific parameters passed to init() that are not core configuration fields. Access directly as attributes: sim.param_name.

Examples

Basic usage with default configuration:

>>> import bamengine as bam
>>> sim = bam.Simulation.init(seed=42)
>>> sim.run(n_periods=100)
>>> unemployment = np.mean(~sim.wrk.employed)
>>> print(f"Final unemployment: {unemployment:.2%}")
Final unemployment: 0.04%

Override configuration parameters:

>>> sim = bam.Simulation.init(n_firms=200, n_households=1000, n_banks=15, seed=42)
>>> sim.step()  # Single period
>>> sim.t
1

Use custom configuration file:

>>> sim = bam.Simulation.init(config="my_config.yml", seed=42)
>>> sim.run(n_periods=50)

Access roles and inspect state:

>>> sim = bam.Simulation.init(seed=42)
>>> sim.step()
>>> prod = sim.get_role("Producer")
>>> avg_price = prod.price.mean()
>>> avg_price > 0
True

Custom pipeline:

>>> sim = bam.Simulation.init(pipeline_path="custom_pipeline.yml", seed=42)
>>> sim.run(n_periods=100)

Notes

All simulations are deterministic when seed is specified
State is mutated in-place during step() and run()
Agent roles share NumPy arrays for memory efficiency
Pipeline execution order can be customized via YAML files

See also

init: Class method to create Simulation instances
step: Execute one simulation period
run: Execute multiple periods
get_role: Access role instances
get_event: Access event instances
Pipeline: Event pipeline configuration

ec#: Economy-wide state container (prices, wages, histories).

config#: Immutable configuration parameters for this simulation.

rng#: Random number generator for deterministic simulations.

n_firms#: Number of firms in the economy.

n_households#: Number of households in the economy.

n_banks#: Number of banks in the economy.

prod#: Producer role storing firm production state.

wrk#: Worker role storing household employment state.

emp#: Employer role storing firm hiring state.

bor#: Borrower role storing firm financial state.

lend#: Lender role storing bank state.

con#: Consumer role storing household consumption state.

sh#: Shareholder role tracking per-period dividend income.

pipeline#: Event pipeline controlling simulation execution order.

lb#: Loan relationship between borrowers and lenders.

n_periods#: Default run length for run().

t#: Current simulation period (starts at 0).

extra_params#: Extension-specific parameters not part of core Config.

property h_rho#: Max production-growth shock.

property h_xi#: Max wage-growth shock.

property h_phi#: Max bank operational costs shock.

property h_eta#: Max price-growth shock.

property max_M#: Max job applications per unemployed worker.

property max_H#: Max loan applications per firm.

property max_Z#: Max firm visits per consumer.

property labor_productivity#: Labor productivity (goods per worker).

property theta#: Job contract length θ.

property beta#: Propensity to consume exponent β.

property delta#: Dividend payout ratio δ (DPR).

property r_bar#: Baseline interest rate r̄.

property v#: Bank capital requirement coefficient v.

property cap_factor#: Breakeven price cap factor.

__getattr__(name)[source]#

Allow access to extra_params via attribute syntax.

Parameters:: name (str) – Attribute name to look up.
Returns:: Value from extra_params if found.
Return type:: Any
Raises:: AttributeError – If attribute not found in extra_params.

Examples

>>> sim = bam.Simulation.init(seed=42, sigma_min=0.0, sigma_max=0.1)
>>> sim.sigma_min
0.0
>>> sim.sigma_max
0.1

__init__(ec, config, rng, n_firms, n_households, n_banks, prod, wrk, emp, bor, lend, con, sh, pipeline, lb, n_periods, t, _role_instances=<factory>, extra_params=<factory>)#

classmethod init(config=None, **overrides)[source]#

Create a new Simulation instance with validated configuration.

Configuration parameters are merged from three sources (later overrides earlier): 1. Package defaults (bamengine/config/defaults.yml) 2. User config (YAML file path, dict, or None) 3. Keyword arguments (highest priority)

Parameters:

config (str, Path, Mapping, or None, optional) – Configuration source: - str/Path: Path to YAML configuration file - Mapping: Dictionary of configuration parameters - None: Use package defaults only
**overrides (Any) – Configuration parameters to override (highest precedence). Common parameters: - n_firms : int (default: 100) - n_households : int (default: 500) - n_banks : int (default: 10) - seed : int or None (default: 0) - pipeline_path : str or None (default: None) - log_level : str, e.g. “WARNING” (simple global log level) - logging : dict (advanced per-event/file configuration) See config/defaults.yml for all parameters.

Returns:

Initialized simulation ready to run.

Return type:

Simulation

Raises:

ValueError – If configuration validation fails (invalid ranges, types, etc.).
FileNotFoundError – If config file path does not exist.

Examples

Use default configuration:

>>> import bamengine as bam
>>> sim = bam.Simulation.init(seed=42)
>>> sim.n_firms, sim.n_households, sim.n_banks
(100, 500, 10)

Override population sizes:

>>> sim = bam.Simulation.init(
...     n_firms=200, n_households=1000, n_banks=15, seed=42
... )
>>> sim.n_firms
200

Load configuration from file:

>>> sim = bam.Simulation.init(config="my_config.yml")

Combine file config with overrides:

>>> sim = bam.Simulation.init(
...     config="base_config.yml", seed=42, n_firms=150
... )

Custom pipeline:

>>> sim = bam.Simulation.init(
...     pipeline_path="custom_pipeline.yml", seed=42
... )

Configure log level:

>>> sim = bam.Simulation.init(log_level="WARNING", seed=42)

Advanced logging (per-event levels, file output):

>>> log_config = {
...     "default_level": "DEBUG",
...     "events": {
...         "firms_adjust_price": "INFO",
...         "labor_market_round": "WARNING",
...     },
... }
>>> sim = bam.Simulation.init(logging=log_config, seed=42)

Notes

All configuration is validated before initialization
Invalid parameters raise ValueError with clear error messages
Vector parameters (price_init, net_worth_init, etc.) accept scalars (broadcast to all agents) or 1D arrays of appropriate length

See also

Config: Configuration dataclass
ConfigValidator: Centralized validation logic
Pipeline: Event pipeline configuration

run(n_periods=None, collect=True, progress=False)[source]#

Run the simulation for multiple periods.

Executes the event pipeline for a specified number of periods, advancing the economy state incrementally. If no argument is provided, uses the n_periods value from initialization.

Parameters:

n_periods (int, optional) – Number of periods to simulate. If None (default), uses the n_periods value passed at initialization via Simulation.init().
collect (bool, list, or dict, default True) –
Controls data collection during the simulation.
- True: Collect all roles and economy metrics with unaggregated per-agent data (2D arrays of shape (n_periods, n_agents)). Relationships are opt-in only and not included by default.
- False: No collection, returns None. Use this for benchmarks or when only final state is needed.
- list[str]: Collect specified roles with all their variables. Economy metrics are always included. Example: ["Producer", "Worker"]
- dict: Specify variables per role:
  - Keys: role names ("Producer", "Worker", etc.)
  - Values: True for all variables, or list of variable names
  - Optional "aggregate" key: "mean", "median", "sum", "std", or None (default: None, full per-agent data)
  - Economy metrics are always collected regardless of dict contents
progress (bool, default False) – If True, log period progress regardless of log level.

Returns:

Results object containing collected data, or None if collect=False.

Return type:

SimulationResults or None

Examples

Run and collect results (the default):

>>> import bamengine as be
>>> sim = be.Simulation.init(seed=42)
>>> results = sim.run(n_periods=100)

Access data via bracket syntax or attribute access:

>>> results["Economy.unemployment_rate"]
array([0.052, 0.048, ...])
>>> results.Producer.price  # attribute-style access
array([[1.02, 0.98, ...], ...])

Discover what was collected:

>>> results.available()
['Economy.avg_price', 'Economy.inflation', 'Producer.price', ...]

Skip collection for performance-sensitive runs:

>>> sim.run(n_periods=100, collect=False)

Use n_periods from initialization:

>>> sim = be.Simulation.init(n_periods=50, seed=42)
>>> results = sim.run()  # Runs for 50 periods

Collect specific roles with all their variables:

>>> results = sim.run(
...     n_periods=100,
...     collect=["Producer", "Worker"],
... )

Custom data collection with specific variables:

>>> results = sim.run(
...     n_periods=100,
...     collect={
...         "Producer": ["price", "inventory"],
...         "Worker": True,
...     },
... )

Notes

Each period corresponds to one execution of the full event pipeline
State is mutated in-place regardless of the collect parameter
Simulation halts early if the economy collapses (all firms/banks bankrupt); results.metadata["actual_periods"] records how many periods actually ran
For step-by-step execution with custom logic, use step() instead
Use sim.collectables() to list all variables available for collection before running

See also

step: Execute a single simulation period
init: Initialize simulation with configuration
collectables: List all collectable variables
SimulationResults: Container for collected simulation data

step()[source]#

Execute one simulation period through the event pipeline.

Advances the economy by exactly one period, executing all events in the pipeline in the order specified. This is the core execution method called by run(). Users can call this directly for fine-grained control between periods.

Returns:: State is mutated in-place. The period counter (sim.t) is incremented by 1. If the economy is collapsed (all firms/banks bankrupt), execution halts immediately.
Return type:: None

Examples

Step through simulation manually with intermediate analysis:

>>> import bamengine as be
>>> sim = be.Simulation.init(seed=42)
>>> for period in range(100):
...     sim.step()
...     if period % 10 == 0:
...         unemployment = np.mean(~sim.wrk.employed)
...         print(f"Period {period}: Unemployment = {unemployment:.2%}")
Period 0: Unemployment = 8.40%
Period 10: Unemployment = 5.20%
Period 20: Unemployment = 4.80%
...

Modify pipeline before stepping:

>>> sim = be.Simulation.init(seed=42)
>>> # Remove a specific event from the pipeline
>>> sim.pipeline.remove("firms_pay_dividends")
>>> sim.step()  # Executes modified pipeline

Conditional execution based on economy state:

>>> sim = be.Simulation.init(seed=42)
>>> while sim.t < 100 and not sim.ec.collapsed:
...     sim.step()
...     avg_price = sim.ec.avg_mkt_price
...     if avg_price > 2.0:
...         print(f"High prices detected at period {sim.t}")
...         break

Notes

Executes all events in sim.pipeline in order
Increments period counter (sim.t) before pipeline execution
Halts immediately if sim.ec.collapsed is True (economy collapse)
For bulk execution over many periods, use run() instead
Pipeline can be modified between calls to step()

See also

run: Execute multiple periods
pipeline: Event pipeline (can be modified before stepping)

get_role(name)[source]#

Get role instance by name.

Parameters:: name (str) – Role name (case-insensitive): ‘Producer’, ‘Worker’, ‘Employer’, ‘Borrower’, ‘Lender’, ‘Consumer’, or any custom role name.
Returns:: Role instance from simulation.
Return type:: Role
Raises:: KeyError – If role name not found.

Examples

>>> sim = Simulation.init()
>>> prod = sim.get_role("Producer")
>>> assert prod is sim.prod

use(ext)[source]#

Apply an extension bundle to the simulation.

Convenience method that activates all roles, events, relationships, and default configuration from an Extension in a single call.

Parameters:: ext (Extension) – Extension bundle to apply.

Examples

>>> from extensions.rnd import RND
>>> sim = bam.Simulation.init(seed=42)
>>> sim.use(RND)

use_relationship(rel_cls)[source]#

Register a custom relationship class.

Note

Placeholder for future use. Currently a no-op.

Parameters:: rel_cls (type) – Relationship class to register.

use_role(role_cls, *, n_agents=None)[source]#

Instantiate and attach a custom role to the simulation.

Creates a role instance with zeroed arrays and stores it in the simulation. If the role is already attached, returns the existing instance.

Parameters:

role_cls (type) – Role class to instantiate (decorated with @role).
n_agents (int, optional) – Number of agents (array size). Defaults to n_firms. Use sim.n_households for household-level roles, sim.n_banks for bank-level roles.

Returns:

The newly created or existing instance, attached to simulation.

Return type:

Role instance

Examples

Firm-level role (default):

>>> from bamengine import role, Float
>>> @role
... class RnD:
...     sigma: Float
...     rnd_intensity: Float
>>> sim = bam.Simulation.init(n_firms=100, seed=42)
>>> rnd = sim.use_role(RnD)
>>> rnd.sigma.shape
(100,)

Household-level role:

>>> @role
... class BufferStock:
...     prev_income: Float
...     propensity: Float
>>> buf = sim.use_role(BufferStock, n_agents=sim.n_households)
>>> buf.prev_income.shape
(500,)

use_events(*event_classes)[source]#

Apply event hooks to the simulation pipeline.

For each event class with hook metadata (@event(after=..., before=..., replace=...)), applies the hook to the current pipeline. Classes without hook metadata are silently skipped.

Parameters:: *event_classes (type) – Event classes decorated with @event(after=..., before=..., or replace=...).

Examples

>>> from extensions.rnd import RND_EVENTS
>>> sim.use_events(*RND_EVENTS)

>>> from extensions.buffer_stock import BUFFER_STOCK_EVENTS
>>> sim.use_events(*BUFFER_STOCK_EVENTS)

use_config(config)[source]#

Apply extension default configuration.

Merges into extra_params with “don’t overwrite” semantics: parameters already set via Simulation.init(**kwargs) or earlier use_config() calls take precedence.

Parameters:: config (dict[str, Any]) – Extension default parameters to apply.

Examples

>>> from extensions.rnd import RND_CONFIG
>>> sim.use_config(RND_CONFIG)

>>> from extensions.buffer_stock import BUFFER_STOCK_CONFIG
>>> sim.use_config(BUFFER_STOCK_CONFIG)

collectables()[source]#

List all variables available for collection.

Returns a sorted list of ‘Name.variable’ strings based on currently active roles, registered relationships, and economy metrics. Extension roles appear automatically after sim.use().

Returns:: Available collection keys.
Return type:: list[str]

Examples

>>> sim = Simulation.init(seed=42)
>>> "Producer.production" in sim.collectables()
True

get_event(name)[source]#

Get event instance from pipeline by name.

Parameters:: name (str) – Event name (e.g., ‘firms_adjust_price’).
Returns:: Event instance from current pipeline.
Return type:: Event
Raises:: KeyError – If event not found in pipeline.

Examples

>>> sim = Simulation.init()
>>> pricing_event = sim.get_event("firms_adjust_price")

get_relationship(name)[source]#

Get relationship instance by name.

Parameters:: name (str) – Relationship name (case-insensitive): ‘LoanBook’.
Returns:: Relationship instance from simulation.
Return type:: Relationship
Raises:: KeyError – If relationship name not found.

Examples

>>> sim = Simulation.init()
>>> lb = sim.get_relationship("LoanBook")
>>> assert lb is sim.lb

get(name)[source]#

Get role, event or relationship by name.

Parameters:: name (str) – Role, event or relationship name.
Returns:: Role, event or relationship instance from simulation.
Return type:: Role | Event | Relationship
Raises:: ValueError – If name not found in simulation.

Note

Searches roles first, then events, then relationships.

Examples

>>> sim = Simulation.init()
>>> prod = sim.get("Producer")
>>> event = sim.get("firms_adjust_price")

Simulation#

This Page