Skip to content

API Documentation

Core Components

Base Simulation

sim_lab.core.BaseSimulation

Bases: ABC

Base class for all SimLab simulations.

This abstract class defines the common interface and utility methods for all simulation types in the SimLab package.

Attributes:

Name Type Description
random_seed Optional[int]

Seed for random number generation to ensure reproducible results
Source code in src/sim_lab/core/base_simulation.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class BaseSimulation(ABC):
    """Base class for all SimLab simulations.

    This abstract class defines the common interface and utility methods
    for all simulation types in the SimLab package.

    Attributes:
        random_seed (Optional[int]): Seed for random number generation to ensure reproducible results
    """

    def __init__(self, days: int, random_seed: Optional[int] = None, **kwargs):
        """Initialize the base simulation.

        Args:
            days (int): The duration of the simulation in days/steps.
            random_seed (Optional[int]): Seed for random number generation. If None, random results will vary.
            **kwargs: Additional parameters for specific simulation types.
        """
        self.days = days
        self.random_seed = random_seed
        self._initialize_random_generators()

    def _initialize_random_generators(self) -> None:
        """Initialize random number generators with the seed."""
        if self.random_seed is not None:
            random.seed(self.random_seed)
            np.random.seed(self.random_seed)

    @abstractmethod
    def run_simulation(self) -> List[Union[float, int]]:
        """Run the simulation and return results.

        This method must be implemented by all simulation subclasses.

        Returns:
            A list of values representing the simulation results over time.
        """
        pass

    def reset(self) -> None:
        """Reset the simulation to its initial state.

        This allows a simulation instance to be re-run with the same parameters.
        """
        self._initialize_random_generators()

    @classmethod
    def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
        """Get information about the parameters required by this simulation.

        Returns:
            A dictionary mapping parameter names to their metadata (type, description, default, etc.)
        """
        return {
            'days': {
                'type': 'int',
                'description': 'The duration of the simulation in days/steps',
                'required': True
            },
            'random_seed': {
                'type': 'int',
                'description': 'Seed for random number generation to ensure reproducible results',
                'required': False,
                'default': None
            }
        }

__init__(days, random_seed=None, **kwargs)

Initialize the base simulation.

Parameters:

Name Type Description Default
days int

The duration of the simulation in days/steps.

 required
random_seed Optional[int]

Seed for random number generation. If None, random results will vary.

 None
**kwargs

Additional parameters for specific simulation types.

 {}
Source code in src/sim_lab/core/base_simulation.py 
19
20
21
22
23
24
25
26
27
28
29
def __init__(self, days: int, random_seed: Optional[int] = None, **kwargs):
    """Initialize the base simulation.

    Args:
        days (int): The duration of the simulation in days/steps.
        random_seed (Optional[int]): Seed for random number generation. If None, random results will vary.
        **kwargs: Additional parameters for specific simulation types.
    """
    self.days = days
    self.random_seed = random_seed
    self._initialize_random_generators()

get_parameters_info() classmethod

Get information about the parameters required by this simulation.

Returns:

Type Description
Dict[str, Dict[str, Any]]

A dictionary mapping parameter names to their metadata (type, description, default, etc.)
Source code in src/sim_lab/core/base_simulation.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@classmethod
def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
    """Get information about the parameters required by this simulation.

    Returns:
        A dictionary mapping parameter names to their metadata (type, description, default, etc.)
    """
    return {
        'days': {
            'type': 'int',
            'description': 'The duration of the simulation in days/steps',
            'required': True
        },
        'random_seed': {
            'type': 'int',
            'description': 'Seed for random number generation to ensure reproducible results',
            'required': False,
            'default': None
        }
    }

reset()

Reset the simulation to its initial state.

This allows a simulation instance to be re-run with the same parameters.

Source code in src/sim_lab/core/base_simulation.py 
48
49
50
51
52
53
def reset(self) -> None:
    """Reset the simulation to its initial state.

    This allows a simulation instance to be re-run with the same parameters.
    """
    self._initialize_random_generators()

run_simulation() abstractmethod

Run the simulation and return results.

This method must be implemented by all simulation subclasses.

Returns:

Type Description
List[Union[float, int]]

A list of values representing the simulation results over time.
Source code in src/sim_lab/core/base_simulation.py 
37
38
39
40
41
42
43
44
45
46
@abstractmethod
def run_simulation(self) -> List[Union[float, int]]:
    """Run the simulation and return results.

    This method must be implemented by all simulation subclasses.

    Returns:
        A list of values representing the simulation results over time.
    """
    pass

Simulator Registry

sim_lab.core.SimulatorRegistry

Registry for simulation models.

This class maintains a registry of all available simulation models, allowing for dynamic discovery and instantiation of simulations.

Source code in src/sim_lab/core/registry.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class SimulatorRegistry:
    """Registry for simulation models.

    This class maintains a registry of all available simulation models,
    allowing for dynamic discovery and instantiation of simulations.
    """

    _registry: Dict[str, Type[BaseSimulation]] = {}

    @classmethod
    def register(cls, name: Optional[str] = None) -> callable:
        """Decorator to register a simulation class.

        Args:
            name: The name to register the simulation under. If None, 
                 the class name will be used.

        Returns:
            A decorator function that registers the class.
        """
        def decorator(sim_class: Type[T]) -> Type[T]:
            if not inspect.isclass(sim_class) or not issubclass(sim_class, BaseSimulation):
                raise TypeError(f"Class {sim_class.__name__} must be a subclass of BaseSimulation")

            sim_name = name if name is not None else sim_class.__name__
            cls._registry[sim_name] = sim_class
            return sim_class

        return decorator

    @classmethod
    def unregister(cls, name: str) -> None:
        """Remove a simulation from the registry.

        Args:
            name: The name of the simulation to remove.

        Raises:
            KeyError: If the simulation is not registered.
        """
        if name in cls._registry:
            del cls._registry[name]
        else:
            raise KeyError(f"Simulation '{name}' is not registered")

    @classmethod
    def get(cls, name: str) -> Type[BaseSimulation]:
        """Get a simulation class by name.

        Args:
            name: The name of the simulation to get.

        Returns:
            The simulation class.

        Raises:
            KeyError: If the simulation is not registered.
        """
        if name in cls._registry:
            return cls._registry[name]
        else:
            raise KeyError(f"Simulation '{name}' is not registered")

    @classmethod
    def list_simulators(cls) -> List[str]:
        """List all registered simulations.

        Returns:
            A list of simulation names.
        """
        return list(cls._registry.keys())

    @classmethod
    def create(cls, name: str, **kwargs: Any) -> BaseSimulation:
        """Create an instance of a simulation.

        Args:
            name: The name of the simulation to create.
            **kwargs: Parameters to pass to the simulation constructor.

        Returns:
            A new instance of the requested simulation.

        Raises:
            KeyError: If the simulation is not registered.
        """
        sim_class = cls.get(name)
        return sim_class(**kwargs)

    @classmethod
    def load_simulator_from_path(cls, module_path: str, class_name: str, 
                                register_as: Optional[str] = None) -> Type[BaseSimulation]:
        """Load a simulator class from a module path and register it.

        Args:
            module_path: The dotted path to the module (e.g. 'sim_lab.custom.my_simulation').
            class_name: The name of the class to load.
            register_as: The name to register the simulation under. If None,
                         the class name will be used.

        Returns:
            The loaded simulation class.

        Raises:
            ImportError: If the module or class cannot be loaded.
            TypeError: If the class is not a subclass of BaseSimulation.
        """
        try:
            module = importlib.import_module(module_path)
            sim_class = getattr(module, class_name)

            if not inspect.isclass(sim_class) or not issubclass(sim_class, BaseSimulation):
                raise TypeError(f"Class {class_name} must be a subclass of BaseSimulation")

            reg_name = register_as if register_as is not None else class_name
            cls._registry[reg_name] = sim_class
            return sim_class

        except ImportError:
            raise ImportError(f"Could not import module '{module_path}'")
        except AttributeError:
            raise ImportError(f"Module '{module_path}' has no class named '{class_name}'")

create(name, **kwargs) classmethod

Create an instance of a simulation.

Parameters:

Name Type Description Default
name str

The name of the simulation to create.

 required
**kwargs Any

Parameters to pass to the simulation constructor.

 {}

Returns:

Type Description
BaseSimulation

A new instance of the requested simulation.

Raises:

Type Description
KeyError

If the simulation is not registered.
Source code in src/sim_lab/core/registry.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@classmethod
def create(cls, name: str, **kwargs: Any) -> BaseSimulation:
    """Create an instance of a simulation.

    Args:
        name: The name of the simulation to create.
        **kwargs: Parameters to pass to the simulation constructor.

    Returns:
        A new instance of the requested simulation.

    Raises:
        KeyError: If the simulation is not registered.
    """
    sim_class = cls.get(name)
    return sim_class(**kwargs)

get(name) classmethod

Get a simulation class by name.

Parameters:

Name Type Description Default
name str

The name of the simulation to get.

 required

Returns:

Type Description
Type[BaseSimulation]

The simulation class.

Raises:

Type Description
KeyError

If the simulation is not registered.
Source code in src/sim_lab/core/registry.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@classmethod
def get(cls, name: str) -> Type[BaseSimulation]:
    """Get a simulation class by name.

    Args:
        name: The name of the simulation to get.

    Returns:
        The simulation class.

    Raises:
        KeyError: If the simulation is not registered.
    """
    if name in cls._registry:
        return cls._registry[name]
    else:
        raise KeyError(f"Simulation '{name}' is not registered")

list_simulators() classmethod

List all registered simulations.

Returns:

Type Description
List[str]

A list of simulation names.
Source code in src/sim_lab/core/registry.py 
75
76
77
78
79
80
81
82
@classmethod
def list_simulators(cls) -> List[str]:
    """List all registered simulations.

    Returns:
        A list of simulation names.
    """
    return list(cls._registry.keys())

load_simulator_from_path(module_path, class_name, register_as=None) classmethod

Load a simulator class from a module path and register it.

Parameters:

Name Type Description Default
module_path str

The dotted path to the module (e.g. 'sim_lab.custom.my_simulation').

 required
class_name str

The name of the class to load.

 required
register_as Optional[str]

The name to register the simulation under. If None, the class name will be used.

 None

Returns:

Type Description
Type[BaseSimulation]

The loaded simulation class.

Raises:

Type Description
ImportError

If the module or class cannot be loaded.
TypeError

If the class is not a subclass of BaseSimulation.
Source code in src/sim_lab/core/registry.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@classmethod
def load_simulator_from_path(cls, module_path: str, class_name: str, 
                            register_as: Optional[str] = None) -> Type[BaseSimulation]:
    """Load a simulator class from a module path and register it.

    Args:
        module_path: The dotted path to the module (e.g. 'sim_lab.custom.my_simulation').
        class_name: The name of the class to load.
        register_as: The name to register the simulation under. If None,
                     the class name will be used.

    Returns:
        The loaded simulation class.

    Raises:
        ImportError: If the module or class cannot be loaded.
        TypeError: If the class is not a subclass of BaseSimulation.
    """
    try:
        module = importlib.import_module(module_path)
        sim_class = getattr(module, class_name)

        if not inspect.isclass(sim_class) or not issubclass(sim_class, BaseSimulation):
            raise TypeError(f"Class {class_name} must be a subclass of BaseSimulation")

        reg_name = register_as if register_as is not None else class_name
        cls._registry[reg_name] = sim_class
        return sim_class

    except ImportError:
        raise ImportError(f"Could not import module '{module_path}'")
    except AttributeError:
        raise ImportError(f"Module '{module_path}' has no class named '{class_name}'")

register(name=None) classmethod

Decorator to register a simulation class.

Parameters:

Name Type Description Default
name Optional[str]

The name to register the simulation under. If None, the class name will be used.

 None

Returns:

Type Description
callable

A decorator function that registers the class.
Source code in src/sim_lab/core/registry.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@classmethod
def register(cls, name: Optional[str] = None) -> callable:
    """Decorator to register a simulation class.

    Args:
        name: The name to register the simulation under. If None, 
             the class name will be used.

    Returns:
        A decorator function that registers the class.
    """
    def decorator(sim_class: Type[T]) -> Type[T]:
        if not inspect.isclass(sim_class) or not issubclass(sim_class, BaseSimulation):
            raise TypeError(f"Class {sim_class.__name__} must be a subclass of BaseSimulation")

        sim_name = name if name is not None else sim_class.__name__
        cls._registry[sim_name] = sim_class
        return sim_class

    return decorator

unregister(name) classmethod

Remove a simulation from the registry.

Parameters:

Name Type Description Default
name str

The name of the simulation to remove.

 required

Raises:

Type Description
KeyError

If the simulation is not registered.
Source code in src/sim_lab/core/registry.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@classmethod
def unregister(cls, name: str) -> None:
    """Remove a simulation from the registry.

    Args:
        name: The name of the simulation to remove.

    Raises:
        KeyError: If the simulation is not registered.
    """
    if name in cls._registry:
        del cls._registry[name]
    else:
        raise KeyError(f"Simulation '{name}' is not registered")

Basic Simulations

Product Popularity Simulation Class

sim_lab.core.ProductPopularitySimulation

Bases: BaseSimulation

A simulation class to model the dynamics of product popularity over time, incorporating factors like natural growth, marketing impact, and promotional campaigns.

Attributes:

Name Type Description
start_demand int

Initial demand for the product.
days int

Duration of the simulation in days.
growth_rate float

Natural growth rate of product demand.
marketing_impact float

Impact of ongoing marketing efforts on demand.
promotion_day Optional[int]

Day on which a major marketing campaign starts (default is None).
promotion_effectiveness float

Effectiveness of the marketing campaign.
random_seed Optional[int]

The seed for the random number generator to ensure reproducibility (default is None).

Methods:

Name Description
run_simulation

Runs the simulation and returns a list of demand values over time.
Source code in src/sim_lab/core/product_popularity_simulation.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@SimulatorRegistry.register("ProductPopularity")
class ProductPopularitySimulation(BaseSimulation):
    """
    A simulation class to model the dynamics of product popularity over time,
    incorporating factors like natural growth, marketing impact, and promotional campaigns.

    Attributes:
        start_demand (int): Initial demand for the product.
        days (int): Duration of the simulation in days.
        growth_rate (float): Natural growth rate of product demand.
        marketing_impact (float): Impact of ongoing marketing efforts on demand.
        promotion_day (Optional[int]): Day on which a major marketing campaign starts (default is None).
        promotion_effectiveness (float): Effectiveness of the marketing campaign.
        random_seed (Optional[int]): The seed for the random number generator to ensure reproducibility (default is None).

    Methods:
        run_simulation(): Runs the simulation and returns a list of demand values over time.
    """

    def __init__(
        self, start_demand: float, days: int, growth_rate: float, marketing_impact: float,
        promotion_day: Optional[int] = None, promotion_effectiveness: float = 0,
        random_seed: Optional[int] = None
    ) -> None:
        """
        Initializes the ProductPopularitySimulation with all necessary parameters.

        Parameters:
            start_demand (int): The initial level of demand for the product.
            days (int): The total number of days to simulate.
            growth_rate (float): The natural daily growth rate of demand, as a decimal.
            marketing_impact (float): Daily impact of marketing on demand, as a decimal.
            promotion_day (Optional[int]): The specific day on which a promotional event occurs (defaults to None).
            promotion_effectiveness (float): Multiplicative impact of the promotion on demand.
            random_seed (Optional[int]): Seed for the random number generator to ensure reproducible results (defaults to None).
        """
        super().__init__(days=days, random_seed=random_seed)
        self.start_demand = start_demand
        self.growth_rate = growth_rate
        self.marketing_impact = marketing_impact
        self.promotion_day = promotion_day
        self.promotion_effectiveness = promotion_effectiveness

    def run_simulation(self) -> List[float]:
        """
        Simulates the demand for a product over a specified number of days based on the initial settings.

        Returns:
            List[int]: A list containing the demand for the product for each day of the simulation.
        """
        # Base class handles random seed initialization
        self.reset()

        demand = [self.start_demand]
        for day in range(1, self.days):
            previous_demand = demand[-1]
            natural_growth = previous_demand * (1 + self.growth_rate)
            marketing_influence = previous_demand * self.marketing_impact

            new_demand = natural_growth + marketing_influence

            if day == self.promotion_day:
                new_demand = (natural_growth + marketing_influence) * (1 + self.promotion_effectiveness)

            demand.append(new_demand)

        return demand

    @classmethod
    def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
        """Get information about the parameters required by this simulation.

        Returns:
            A dictionary mapping parameter names to their metadata (type, description, default, etc.)
        """
        # Get base parameters from parent class
        params = super().get_parameters_info()

        # Add class-specific parameters
        params.update({
            'start_demand': {
                'type': 'float',
                'description': 'The initial level of demand for the product',
                'required': True
            },
            'growth_rate': {
                'type': 'float',
                'description': 'The natural daily growth rate of demand, as a decimal',
                'required': True
            },
            'marketing_impact': {
                'type': 'float',
                'description': 'Daily impact of marketing on demand, as a decimal',
                'required': True
            },
            'promotion_day': {
                'type': 'int',
                'description': 'The specific day on which a promotional event occurs',
                'required': False,
                'default': None
            },
            'promotion_effectiveness': {
                'type': 'float',
                'description': 'Multiplicative impact of the promotion on demand',
                'required': False,
                'default': 0
            }
        })

        return params

__init__(start_demand, days, growth_rate, marketing_impact, promotion_day=None, promotion_effectiveness=0, random_seed=None)

Initializes the ProductPopularitySimulation with all necessary parameters.

Parameters:

Name Type Description Default
start_demand int

The initial level of demand for the product.

 required
days int

The total number of days to simulate.

 required
growth_rate float

The natural daily growth rate of demand, as a decimal.

 required
marketing_impact float

Daily impact of marketing on demand, as a decimal.

 required
promotion_day Optional[int]

The specific day on which a promotional event occurs (defaults to None).

 None
promotion_effectiveness float

Multiplicative impact of the promotion on demand.

 0
random_seed Optional[int]

Seed for the random number generator to ensure reproducible results (defaults to None).

 None
Source code in src/sim_lab/core/product_popularity_simulation.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(
    self, start_demand: float, days: int, growth_rate: float, marketing_impact: float,
    promotion_day: Optional[int] = None, promotion_effectiveness: float = 0,
    random_seed: Optional[int] = None
) -> None:
    """
    Initializes the ProductPopularitySimulation with all necessary parameters.

    Parameters:
        start_demand (int): The initial level of demand for the product.
        days (int): The total number of days to simulate.
        growth_rate (float): The natural daily growth rate of demand, as a decimal.
        marketing_impact (float): Daily impact of marketing on demand, as a decimal.
        promotion_day (Optional[int]): The specific day on which a promotional event occurs (defaults to None).
        promotion_effectiveness (float): Multiplicative impact of the promotion on demand.
        random_seed (Optional[int]): Seed for the random number generator to ensure reproducible results (defaults to None).
    """
    super().__init__(days=days, random_seed=random_seed)
    self.start_demand = start_demand
    self.growth_rate = growth_rate
    self.marketing_impact = marketing_impact
    self.promotion_day = promotion_day
    self.promotion_effectiveness = promotion_effectiveness

get_parameters_info() classmethod

Get information about the parameters required by this simulation.

Returns:

Type Description
Dict[str, Dict[str, Any]]

A dictionary mapping parameter names to their metadata (type, description, default, etc.)
Source code in src/sim_lab/core/product_popularity_simulation.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@classmethod
def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
    """Get information about the parameters required by this simulation.

    Returns:
        A dictionary mapping parameter names to their metadata (type, description, default, etc.)
    """
    # Get base parameters from parent class
    params = super().get_parameters_info()

    # Add class-specific parameters
    params.update({
        'start_demand': {
            'type': 'float',
            'description': 'The initial level of demand for the product',
            'required': True
        },
        'growth_rate': {
            'type': 'float',
            'description': 'The natural daily growth rate of demand, as a decimal',
            'required': True
        },
        'marketing_impact': {
            'type': 'float',
            'description': 'Daily impact of marketing on demand, as a decimal',
            'required': True
        },
        'promotion_day': {
            'type': 'int',
            'description': 'The specific day on which a promotional event occurs',
            'required': False,
            'default': None
        },
        'promotion_effectiveness': {
            'type': 'float',
            'description': 'Multiplicative impact of the promotion on demand',
            'required': False,
            'default': 0
        }
    })

    return params

run_simulation()

Simulates the demand for a product over a specified number of days based on the initial settings.

Returns:

Type Description
List[float]

List[int]: A list containing the demand for the product for each day of the simulation.
Source code in src/sim_lab/core/product_popularity_simulation.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def run_simulation(self) -> List[float]:
    """
    Simulates the demand for a product over a specified number of days based on the initial settings.

    Returns:
        List[int]: A list containing the demand for the product for each day of the simulation.
    """
    # Base class handles random seed initialization
    self.reset()

    demand = [self.start_demand]
    for day in range(1, self.days):
        previous_demand = demand[-1]
        natural_growth = previous_demand * (1 + self.growth_rate)
        marketing_influence = previous_demand * self.marketing_impact

        new_demand = natural_growth + marketing_influence

        if day == self.promotion_day:
            new_demand = (natural_growth + marketing_influence) * (1 + self.promotion_effectiveness)

        demand.append(new_demand)

    return demand

Resource Fluctuation Simulation Class

sim_lab.core.ResourceFluctuationsSimulation

Bases: BaseSimulation

A simulation class to model the fluctuations of resource prices over time, considering factors like volatility, market trends (drift), and supply disruptions.

Attributes:

Name Type Description
start_price float

The initial price of the resource.
days int

The duration of the simulation in days.
volatility float

The volatility of price changes, representing day-to-day variability.
drift float

The average daily price change, indicating the trend over time.
supply_disruption_day Optional[int]

The specific day a supply disruption occurs (default is None).
disruption_severity float

The magnitude of the disruption's impact on price (default is 0).
random_seed Optional[int]

The seed for the random number generator to ensure reproducibility (default is None).
Source code in src/sim_lab/core/resource_fluctuations_simulation.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@SimulatorRegistry.register("ResourceFluctuations")
class ResourceFluctuationsSimulation(BaseSimulation):
    """
    A simulation class to model the fluctuations of resource prices over time,
    considering factors like volatility, market trends (drift), and supply disruptions.

    Attributes:
        start_price (float): The initial price of the resource.
        days (int): The duration of the simulation in days.
        volatility (float): The volatility of price changes, representing day-to-day variability.
        drift (float): The average daily price change, indicating the trend over time.
        supply_disruption_day (Optional[int]): The specific day a supply disruption occurs (default is None).
        disruption_severity (float): The magnitude of the disruption's impact on price (default is 0).
        random_seed (Optional[int]): The seed for the random number generator to ensure reproducibility (default is None).
    """
    def __init__(self, start_price: float, days: int, volatility: float, drift: float,
                 supply_disruption_day: Optional[int] = None, disruption_severity: float = 0,
                 random_seed: Optional[int] = None) -> None:
        """
        Initializes the ResourceSimulation with all necessary parameters.

        Parameters:
            start_price (float): The initial price of the resource.
            days (int): The total number of days to simulate.
            volatility (float): The volatility of the resource price, representing the randomness of day-to-day price changes.
            drift (float): The expected daily percentage change in price, which can be positive or negative.
            supply_disruption_day (Optional[int]): Day on which a supply disruption occurs (defaults to None).
            disruption_severity (float): The severity of the supply disruption, affecting prices multiplicatively.
            random_seed (Optional[int]): Seed for the random number generator to ensure reproducible results (defaults to None).
        """
        super().__init__(days=days, random_seed=random_seed)
        self.start_price = start_price
        self.volatility = volatility
        self.drift = drift
        self.supply_disruption_day = supply_disruption_day
        self.disruption_severity = disruption_severity

    def run_simulation(self) -> List[float]:
        """
        Simulates the price of the resource over a specified number of days based on the initial settings.

        Returns:
            List[float]: A list containing the price of the resource for each day of the simulation.
        """
        # Base class handles random seed initialization
        self.reset()

        prices = [self.start_price]
        for day in range(1, self.days):
            previous_price = prices[-1]
            random_change = np.random.normal(self.drift, self.volatility)
            new_price = previous_price * (1 + random_change)

            if day == self.supply_disruption_day:
                new_price = previous_price * (1 + self.disruption_severity)

            prices.append(new_price)

        return prices

    @classmethod
    def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
        """Get information about the parameters required by this simulation.

        Returns:
            A dictionary mapping parameter names to their metadata (type, description, default, etc.)
        """
        # Get base parameters from parent class
        params = super().get_parameters_info()

        # Add class-specific parameters
        params.update({
            'start_price': {
                'type': 'float',
                'description': 'The initial price of the resource',
                'required': True
            },
            'volatility': {
                'type': 'float',
                'description': 'The volatility of resource price changes, representing day-to-day variability',
                'required': True
            },
            'drift': {
                'type': 'float',
                'description': 'The average daily price change, indicating the trend over time',
                'required': True
            },
            'supply_disruption_day': {
                'type': 'int',
                'description': 'The specific day a supply disruption occurs',
                'required': False,
                'default': None
            },
            'disruption_severity': {
                'type': 'float',
                'description': 'The magnitude of the disruption\'s impact on prices',
                'required': False,
                'default': 0
            }
        })

        return params

__init__(start_price, days, volatility, drift, supply_disruption_day=None, disruption_severity=0, random_seed=None)

Initializes the ResourceSimulation with all necessary parameters.

Parameters:

Name Type Description Default
start_price float

The initial price of the resource.

 required
days int

The total number of days to simulate.

 required
volatility float

The volatility of the resource price, representing the randomness of day-to-day price changes.

 required
drift float

The expected daily percentage change in price, which can be positive or negative.

 required
supply_disruption_day Optional[int]

Day on which a supply disruption occurs (defaults to None).

 None
disruption_severity float

The severity of the supply disruption, affecting prices multiplicatively.

 0
random_seed Optional[int]

Seed for the random number generator to ensure reproducible results (defaults to None).

 None
Source code in src/sim_lab/core/resource_fluctuations_simulation.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, start_price: float, days: int, volatility: float, drift: float,
             supply_disruption_day: Optional[int] = None, disruption_severity: float = 0,
             random_seed: Optional[int] = None) -> None:
    """
    Initializes the ResourceSimulation with all necessary parameters.

    Parameters:
        start_price (float): The initial price of the resource.
        days (int): The total number of days to simulate.
        volatility (float): The volatility of the resource price, representing the randomness of day-to-day price changes.
        drift (float): The expected daily percentage change in price, which can be positive or negative.
        supply_disruption_day (Optional[int]): Day on which a supply disruption occurs (defaults to None).
        disruption_severity (float): The severity of the supply disruption, affecting prices multiplicatively.
        random_seed (Optional[int]): Seed for the random number generator to ensure reproducible results (defaults to None).
    """
    super().__init__(days=days, random_seed=random_seed)
    self.start_price = start_price
    self.volatility = volatility
    self.drift = drift
    self.supply_disruption_day = supply_disruption_day
    self.disruption_severity = disruption_severity

get_parameters_info() classmethod

Get information about the parameters required by this simulation.

Returns:

Type Description
Dict[str, Dict[str, Any]]

A dictionary mapping parameter names to their metadata (type, description, default, etc.)
Source code in src/sim_lab/core/resource_fluctuations_simulation.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@classmethod
def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
    """Get information about the parameters required by this simulation.

    Returns:
        A dictionary mapping parameter names to their metadata (type, description, default, etc.)
    """
    # Get base parameters from parent class
    params = super().get_parameters_info()

    # Add class-specific parameters
    params.update({
        'start_price': {
            'type': 'float',
            'description': 'The initial price of the resource',
            'required': True
        },
        'volatility': {
            'type': 'float',
            'description': 'The volatility of resource price changes, representing day-to-day variability',
            'required': True
        },
        'drift': {
            'type': 'float',
            'description': 'The average daily price change, indicating the trend over time',
            'required': True
        },
        'supply_disruption_day': {
            'type': 'int',
            'description': 'The specific day a supply disruption occurs',
            'required': False,
            'default': None
        },
        'disruption_severity': {
            'type': 'float',
            'description': 'The magnitude of the disruption\'s impact on prices',
            'required': False,
            'default': 0
        }
    })

    return params

run_simulation()

Simulates the price of the resource over a specified number of days based on the initial settings.

Returns:

Type Description
List[float]

List[float]: A list containing the price of the resource for each day of the simulation.
Source code in src/sim_lab/core/resource_fluctuations_simulation.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def run_simulation(self) -> List[float]:
    """
    Simulates the price of the resource over a specified number of days based on the initial settings.

    Returns:
        List[float]: A list containing the price of the resource for each day of the simulation.
    """
    # Base class handles random seed initialization
    self.reset()

    prices = [self.start_price]
    for day in range(1, self.days):
        previous_price = prices[-1]
        random_change = np.random.normal(self.drift, self.volatility)
        new_price = previous_price * (1 + random_change)

        if day == self.supply_disruption_day:
            new_price = previous_price * (1 + self.disruption_severity)

        prices.append(new_price)

    return prices

Stock Market Simulation Class

sim_lab.core.StockMarketSimulation

Bases: BaseSimulation

A simulation class to model the fluctuations of stock prices over time, accounting for volatility, general market trends (drift), and specific market events.

Attributes:

Name Type Description
start_price float

The initial price of the stock.
days int

The duration of the simulation in days.
volatility float

The volatility of stock price changes, representing day-to-day variability.
drift float

The average daily price change, indicating the trend over time.
event_day Optional[int]

The specific day a major market event occurs (default is None).
event_impact float

The magnitude of the event's impact on stock prices (default is 0).
random_seed Optional[int]

The seed for the random number generator to ensure reproducibility (default is None).

Methods:

Name Description
run_simulation

Runs the simulation and returns a list of stock prices over the simulation period.
Source code in src/sim_lab/core/stock_market_simulation.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@SimulatorRegistry.register("StockMarket")
class StockMarketSimulation(BaseSimulation):
    """
    A simulation class to model the fluctuations of stock prices over time, accounting for volatility,
    general market trends (drift), and specific market events.

    Attributes:
        start_price (float): The initial price of the stock.
        days (int): The duration of the simulation in days.
        volatility (float): The volatility of stock price changes, representing day-to-day variability.
        drift (float): The average daily price change, indicating the trend over time.
        event_day (Optional[int]): The specific day a major market event occurs (default is None).
        event_impact (float): The magnitude of the event's impact on stock prices (default is 0).
        random_seed (Optional[int]): The seed for the random number generator to ensure reproducibility (default is None).

    Methods:
        run_simulation(): Runs the simulation and returns a list of stock prices over the simulation period.
    """

    def __init__(
        self, start_price: float, days: int, volatility: float, drift: float,
        event_day: Optional[int] = None, event_impact: float = 0,
        random_seed: Optional[int] = None
    ) -> None:
        """
        Initializes the StockMarketSimulation with all necessary parameters.

        Parameters:
            start_price (float): The initial stock price.
            days (int): The total number of days to simulate.
            volatility (float): The volatility of the stock price, representing the randomness of day-to-day price changes.
            drift (float): The expected daily percentage change in price, which can be positive or negative.
            event_day (Optional[int]): Day on which a major market event occurs (defaults to None).
            event_impact (float): The severity of the market event, affecting prices multiplicatively.
            random_seed (Optional[int]): Seed for the random number generator to ensure reproducible results (defaults to None).
        """
        super().__init__(days=days, random_seed=random_seed)
        self.start_price = start_price
        self.volatility = volatility
        self.drift = drift
        self.event_day = event_day
        self.event_impact = event_impact

    def run_simulation(self) -> List[float]:
        """
        Simulates the stock price over a specified number of days based on the initial settings.

        Returns:
            List[float]: A list containing the stock prices for each day of the simulation.
        """
        # Base class handles random seed initialization
        self.reset()

        prices = [self.start_price]
        for day in range(1, self.days):
            previous_price = prices[-1]
            random_change = np.random.normal(self.drift, self.volatility)
            new_price = previous_price * (1 + random_change)

            if day == self.event_day:
                new_price = previous_price * (1 + self.event_impact)

            prices.append(new_price)

        return prices

    @classmethod
    def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
        """Get information about the parameters required by this simulation.

        Returns:
            A dictionary mapping parameter names to their metadata (type, description, default, etc.)
        """
        # Get base parameters from parent class
        params = super().get_parameters_info()

        # Add class-specific parameters
        params.update({
            'start_price': {
                'type': 'float',
                'description': 'The initial price of the stock',
                'required': True
            },
            'volatility': {
                'type': 'float',
                'description': 'The volatility of stock price changes, representing day-to-day variability',
                'required': True
            },
            'drift': {
                'type': 'float',
                'description': 'The average daily price change, indicating the trend over time',
                'required': True
            },
            'event_day': {
                'type': 'int',
                'description': 'The specific day a major market event occurs',
                'required': False,
                'default': None
            },
            'event_impact': {
                'type': 'float',
                'description': 'The magnitude of the event\'s impact on stock prices',
                'required': False,
                'default': 0
            }
        })

        return params

__init__(start_price, days, volatility, drift, event_day=None, event_impact=0, random_seed=None)

Initializes the StockMarketSimulation with all necessary parameters.

Parameters:

Name Type Description Default
start_price float

The initial stock price.

 required
days int

The total number of days to simulate.

 required
volatility float

The volatility of the stock price, representing the randomness of day-to-day price changes.

 required
drift float

The expected daily percentage change in price, which can be positive or negative.

 required
event_day Optional[int]

Day on which a major market event occurs (defaults to None).

 None
event_impact float

The severity of the market event, affecting prices multiplicatively.

 0
random_seed Optional[int]

Seed for the random number generator to ensure reproducible results (defaults to None).

 None
Source code in src/sim_lab/core/stock_market_simulation.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(
    self, start_price: float, days: int, volatility: float, drift: float,
    event_day: Optional[int] = None, event_impact: float = 0,
    random_seed: Optional[int] = None
) -> None:
    """
    Initializes the StockMarketSimulation with all necessary parameters.

    Parameters:
        start_price (float): The initial stock price.
        days (int): The total number of days to simulate.
        volatility (float): The volatility of the stock price, representing the randomness of day-to-day price changes.
        drift (float): The expected daily percentage change in price, which can be positive or negative.
        event_day (Optional[int]): Day on which a major market event occurs (defaults to None).
        event_impact (float): The severity of the market event, affecting prices multiplicatively.
        random_seed (Optional[int]): Seed for the random number generator to ensure reproducible results (defaults to None).
    """
    super().__init__(days=days, random_seed=random_seed)
    self.start_price = start_price
    self.volatility = volatility
    self.drift = drift
    self.event_day = event_day
    self.event_impact = event_impact

get_parameters_info() classmethod

Get information about the parameters required by this simulation.

Returns:

Type Description
Dict[str, Dict[str, Any]]

A dictionary mapping parameter names to their metadata (type, description, default, etc.)
Source code in src/sim_lab/core/stock_market_simulation.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@classmethod
def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
    """Get information about the parameters required by this simulation.

    Returns:
        A dictionary mapping parameter names to their metadata (type, description, default, etc.)
    """
    # Get base parameters from parent class
    params = super().get_parameters_info()

    # Add class-specific parameters
    params.update({
        'start_price': {
            'type': 'float',
            'description': 'The initial price of the stock',
            'required': True
        },
        'volatility': {
            'type': 'float',
            'description': 'The volatility of stock price changes, representing day-to-day variability',
            'required': True
        },
        'drift': {
            'type': 'float',
            'description': 'The average daily price change, indicating the trend over time',
            'required': True
        },
        'event_day': {
            'type': 'int',
            'description': 'The specific day a major market event occurs',
            'required': False,
            'default': None
        },
        'event_impact': {
            'type': 'float',
            'description': 'The magnitude of the event\'s impact on stock prices',
            'required': False,
            'default': 0
        }
    })

    return params

run_simulation()

Simulates the stock price over a specified number of days based on the initial settings.

Returns:

Type Description
List[float]

List[float]: A list containing the stock prices for each day of the simulation.
Source code in src/sim_lab/core/stock_market_simulation.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def run_simulation(self) -> List[float]:
    """
    Simulates the stock price over a specified number of days based on the initial settings.

    Returns:
        List[float]: A list containing the stock prices for each day of the simulation.
    """
    # Base class handles random seed initialization
    self.reset()

    prices = [self.start_price]
    for day in range(1, self.days):
        previous_price = prices[-1]
        random_change = np.random.normal(self.drift, self.volatility)
        new_price = previous_price * (1 + random_change)

        if day == self.event_day:
            new_price = previous_price * (1 + self.event_impact)

        prices.append(new_price)

    return prices

Discrete Event Simulations

Discrete Event Simulation

sim_lab.core.DiscreteEventSimulation

Bases: BaseSimulation

A simulation class for discrete event simulations.

This simulation processes events in chronological order, with each event potentially generating new events. The simulation runs until a specified end time or until there are no more events to process.

Attributes:

Name Type Description
max_time float

The maximum simulation time.
days int

Used for compatibility with other simulations (days = max_time).
current_time float

The current simulation time.
event_queue List[Event]

The priority queue of pending events.
state Dict[str, Any]

The current state of the simulation.
results List[float]

The results of the simulation at each time step.
random_seed Optional[int]

Seed for random number generation.
Source code in src/sim_lab/core/discrete_event_simulation.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@SimulatorRegistry.register("DiscreteEvent")
class DiscreteEventSimulation(BaseSimulation):
    """A simulation class for discrete event simulations.

    This simulation processes events in chronological order, with each event potentially
    generating new events. The simulation runs until a specified end time or until
    there are no more events to process.

    Attributes:
        max_time (float): The maximum simulation time.
        days (int): Used for compatibility with other simulations (days = max_time).
        current_time (float): The current simulation time.
        event_queue (List[Event]): The priority queue of pending events.
        state (Dict[str, Any]): The current state of the simulation.
        results (List[float]): The results of the simulation at each time step.
        random_seed (Optional[int]): Seed for random number generation.
    """

    def __init__(
        self, max_time: float, initial_events: List[Tuple[float, Callable, Any]] = None,
        time_step: float = 1.0, random_seed: Optional[int] = None
    ) -> None:
        """Initialize the discrete event simulation.

        Args:
            max_time: The maximum simulation time.
            initial_events: List of (time, action, data) tuples to initialize the event queue.
            time_step: The time step for recording results (default: 1.0).
            random_seed: Seed for random number generation.
        """
        # Convert max_time to days for BaseSimulation compatibility
        days = int(max_time)
        super().__init__(days=days, random_seed=random_seed)

        self.max_time = max_time
        self.current_time = 0.0
        self.event_queue = []
        self.state = {"value": 0.0}  # Default state with a value field
        self.results = [0.0]  # Start with initial value
        self.time_step = time_step

        # Schedule initial events
        if initial_events:
            for time, action, data in initial_events:
                self.schedule_event(time, action, data=data)

    def schedule_event(self, time: float, action: Callable, priority: int = 0, data: Any = None) -> None:
        """Schedule a new event to occur at the specified time.

        Args:
            time: The absolute time at which the event should occur.
            action: The function to execute when the event occurs.
            priority: The priority of the event (lower is higher priority).
            data: Additional data associated with the event.
        """
        event = Event(time, action, priority, data)
        heapq.heappush(self.event_queue, event)

    def run_simulation(self) -> List[float]:
        """Run the simulation until max_time or until there are no more events.

        The simulation processes events in chronological order, with each event potentially
        generating new events by calling schedule_event().

        Returns:
            A list of values representing the simulation state at regular intervals.
        """
        # Reset the simulation
        self.reset()

        next_recording_time = self.time_step

        # Process events until max_time or until there are no more events
        while self.event_queue and self.current_time < self.max_time:
            # Get the next event
            event = heapq.heappop(self.event_queue)

            # Update the current time
            self.current_time = event.time

            # Record results at regular intervals
            while next_recording_time <= self.current_time and next_recording_time <= self.max_time:
                self.results.append(self.state["value"])
                next_recording_time += self.time_step

            # Process the event if we haven't exceeded max_time
            if self.current_time <= self.max_time:
                event.action(self, event.data)

        # Make sure we have results for all time steps
        while len(self.results) <= self.days:
            self.results.append(self.state["value"])

        return self.results

    def reset(self) -> None:
        """Reset the simulation to its initial state."""
        super().reset()
        self.current_time = 0.0
        self.state = {"value": 0.0}
        self.results = [0.0]

    @classmethod
    def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
        """Get information about the parameters required by this simulation.

        Returns:
            A dictionary mapping parameter names to their metadata.
        """
        # Get base parameters from parent class
        params = super().get_parameters_info()

        # Replace 'days' with 'max_time'
        del params['days']

        # Add class-specific parameters
        params.update({
            'max_time': {
                'type': 'float',
                'description': 'The maximum simulation time',
                'required': True
            },
            'initial_events': {
                'type': 'List[Tuple[float, Callable, Any]]',
                'description': 'List of (time, action, data) tuples to initialize the event queue',
                'required': False,
                'default': []
            },
            'time_step': {
                'type': 'float',
                'description': 'The time step for recording results',
                'required': False,
                'default': 1.0
            }
        })

        return params

__init__(max_time, initial_events=None, time_step=1.0, random_seed=None)

Initialize the discrete event simulation.

Parameters:

Name Type Description Default
max_time float

The maximum simulation time.

 required
initial_events List[Tuple[float, Callable, Any]]

List of (time, action, data) tuples to initialize the event queue.

 None
time_step float

The time step for recording results (default: 1.0).

 1.0
random_seed Optional[int]

Seed for random number generation.

 None
Source code in src/sim_lab/core/discrete_event_simulation.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(
    self, max_time: float, initial_events: List[Tuple[float, Callable, Any]] = None,
    time_step: float = 1.0, random_seed: Optional[int] = None
) -> None:
    """Initialize the discrete event simulation.

    Args:
        max_time: The maximum simulation time.
        initial_events: List of (time, action, data) tuples to initialize the event queue.
        time_step: The time step for recording results (default: 1.0).
        random_seed: Seed for random number generation.
    """
    # Convert max_time to days for BaseSimulation compatibility
    days = int(max_time)
    super().__init__(days=days, random_seed=random_seed)

    self.max_time = max_time
    self.current_time = 0.0
    self.event_queue = []
    self.state = {"value": 0.0}  # Default state with a value field
    self.results = [0.0]  # Start with initial value
    self.time_step = time_step

    # Schedule initial events
    if initial_events:
        for time, action, data in initial_events:
            self.schedule_event(time, action, data=data)

get_parameters_info() classmethod

Get information about the parameters required by this simulation.

Returns:

Type Description
Dict[str, Dict[str, Any]]

A dictionary mapping parameter names to their metadata.
Source code in src/sim_lab/core/discrete_event_simulation.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@classmethod
def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
    """Get information about the parameters required by this simulation.

    Returns:
        A dictionary mapping parameter names to their metadata.
    """
    # Get base parameters from parent class
    params = super().get_parameters_info()

    # Replace 'days' with 'max_time'
    del params['days']

    # Add class-specific parameters
    params.update({
        'max_time': {
            'type': 'float',
            'description': 'The maximum simulation time',
            'required': True
        },
        'initial_events': {
            'type': 'List[Tuple[float, Callable, Any]]',
            'description': 'List of (time, action, data) tuples to initialize the event queue',
            'required': False,
            'default': []
        },
        'time_step': {
            'type': 'float',
            'description': 'The time step for recording results',
            'required': False,
            'default': 1.0
        }
    })

    return params

reset()

Reset the simulation to its initial state.

Source code in src/sim_lab/core/discrete_event_simulation.py 
137
138
139
140
141
142
def reset(self) -> None:
    """Reset the simulation to its initial state."""
    super().reset()
    self.current_time = 0.0
    self.state = {"value": 0.0}
    self.results = [0.0]

run_simulation()

Run the simulation until max_time or until there are no more events.

The simulation processes events in chronological order, with each event potentially generating new events by calling schedule_event().

Returns:

Type Description
List[float]

A list of values representing the simulation state at regular intervals.
Source code in src/sim_lab/core/discrete_event_simulation.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def run_simulation(self) -> List[float]:
    """Run the simulation until max_time or until there are no more events.

    The simulation processes events in chronological order, with each event potentially
    generating new events by calling schedule_event().

    Returns:
        A list of values representing the simulation state at regular intervals.
    """
    # Reset the simulation
    self.reset()

    next_recording_time = self.time_step

    # Process events until max_time or until there are no more events
    while self.event_queue and self.current_time < self.max_time:
        # Get the next event
        event = heapq.heappop(self.event_queue)

        # Update the current time
        self.current_time = event.time

        # Record results at regular intervals
        while next_recording_time <= self.current_time and next_recording_time <= self.max_time:
            self.results.append(self.state["value"])
            next_recording_time += self.time_step

        # Process the event if we haven't exceeded max_time
        if self.current_time <= self.max_time:
            event.action(self, event.data)

    # Make sure we have results for all time steps
    while len(self.results) <= self.days:
        self.results.append(self.state["value"])

    return self.results

schedule_event(time, action, priority=0, data=None)

Schedule a new event to occur at the specified time.

Parameters:

Name Type Description Default
time float

The absolute time at which the event should occur.

 required
action Callable

The function to execute when the event occurs.

 required
priority int

The priority of the event (lower is higher priority).

 0
data Any

Additional data associated with the event.

 None
Source code in src/sim_lab/core/discrete_event_simulation.py 
88
89
90
91
92
93
94
95
96
97
98
def schedule_event(self, time: float, action: Callable, priority: int = 0, data: Any = None) -> None:
    """Schedule a new event to occur at the specified time.

    Args:
        time: The absolute time at which the event should occur.
        action: The function to execute when the event occurs.
        priority: The priority of the event (lower is higher priority).
        data: Additional data associated with the event.
    """
    event = Event(time, action, priority, data)
    heapq.heappush(self.event_queue, event)

Advanced Simulation Types

Agent-Based Simulation

sim_lab.core.AgentBasedSimulation

Bases: BaseSimulation

A simulation class for agent-based modeling.

This simulation models complex systems by simulating the actions and interactions of autonomous agents, allowing for emergent behavior to be observed.

Attributes:

Name Type Description
agents List[Agent]

List of agents in the simulation.
environment Environment

The environment in which agents operate.
days int

Number of steps to simulate.
neighborhood_radius float

Radius for determining agent neighbors.
random_seed Optional[int]

Seed for random number generation.
Source code in src/sim_lab/core/agent_based_simulation.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
@SimulatorRegistry.register("AgentBased")
class AgentBasedSimulation(BaseSimulation):
    """A simulation class for agent-based modeling.

    This simulation models complex systems by simulating the actions and interactions
    of autonomous agents, allowing for emergent behavior to be observed.

    Attributes:
        agents (List[Agent]): List of agents in the simulation.
        environment (Environment): The environment in which agents operate.
        days (int): Number of steps to simulate.
        neighborhood_radius (float): Radius for determining agent neighbors.
        random_seed (Optional[int]): Seed for random number generation.
    """

    def __init__(
        self,
        agent_factory: Callable[[int], Agent],
        num_agents: int,
        environment: Optional[Environment] = None,
        days: int = 100,
        neighborhood_radius: float = 10.0,
        save_history: bool = False,
        random_seed: Optional[int] = None
    ) -> None:
        """Initialize the agent-based simulation.

        Args:
            agent_factory: Function that creates new agents with given IDs.
            num_agents: Number of agents to create.
            environment: The environment in which agents operate. If None, a default environment is created.
            days: Number of steps to simulate.
            neighborhood_radius: Radius for determining agent neighbors.
            save_history: Whether to save agent and environment history.
            random_seed: Seed for random number generation.
        """
        super().__init__(days=days, random_seed=random_seed)

        self.environment = environment or Environment()
        self.neighborhood_radius = neighborhood_radius
        self.save_history = save_history

        # Create agents
        self.agents = [agent_factory(i) for i in range(num_agents)]

        # Initialize tracking variables
        self.metrics = []

    def get_agent_neighbors(self, agent: Agent) -> List[Agent]:
        """Get the neighbors of an agent based on proximity.

        Args:
            agent: The agent whose neighbors to find.

        Returns:
            List of neighboring agents within the neighborhood radius.
        """
        if agent.position is None:
            return []  # No position, no neighbors

        neighbors = []
        for other in self.agents:
            if other.agent_id == agent.agent_id or other.position is None:
                continue  # Skip self and agents without position

            # Calculate Euclidean distance
            dx = agent.position[0] - other.position[0]
            dy = agent.position[1] - other.position[1]
            distance = (dx**2 + dy**2)**0.5

            if distance <= self.neighborhood_radius:
                neighbors.append(other)

        return neighbors

    def calculate_metrics(self) -> Dict[str, Any]:
        """Calculate metrics for the current simulation state.

        Override this method to define specific metrics for your simulation.

        Returns:
            Dictionary of metrics derived from agent and environment states.
        """
        # Default implementation: count agents in different states
        state_counts = {}

        for agent in self.agents:
            for key, value in agent.state.items():
                if isinstance(value, (bool, int, str, float)):
                    state_key = f"{key}_{value}"
                    state_counts[state_key] = state_counts.get(state_key, 0) + 1

        return state_counts

    def run_simulation(self) -> List[Dict[str, Any]]:
        """Run the agent-based simulation.

        Returns:
            A list of metrics dictionaries for each time step.
        """
        self.reset()

        # Initialize history if tracking
        if self.save_history:
            for agent in self.agents:
                agent.reset()
                agent.save_history()
            self.environment.reset()
            self.environment.save_history()

        # Calculate initial metrics
        self.metrics = [self.calculate_metrics()]

        # Run for specified number of days
        for _ in range(1, self.days):
            # Update agents
            for agent in self.agents:
                neighbors = self.get_agent_neighbors(agent)
                agent.update(self.environment, neighbors)

                if self.save_history:
                    agent.save_history()

            # Update environment
            self.environment.update(self.agents)

            if self.save_history:
                self.environment.save_history()

            # Calculate metrics
            self.metrics.append(self.calculate_metrics())

        return self.metrics

    def get_agent_history(self, agent_id: int) -> List[Dict[str, Any]]:
        """Get the state history for a specific agent.

        Args:
            agent_id: The ID of the agent.

        Returns:
            List of state dictionaries representing the agent's history.
        """
        if not self.save_history:
            raise ValueError("Agent history was not saved. Set save_history=True when creating the simulation.")

        for agent in self.agents:
            if agent.agent_id == agent_id:
                return agent.history

        raise ValueError(f"No agent with ID {agent_id} found")

    def get_environment_history(self) -> List[Dict[str, Any]]:
        """Get the environment state history.

        Returns:
            List of state dictionaries representing the environment's history.
        """
        if not self.save_history:
            raise ValueError("Environment history was not saved. Set save_history=True when creating the simulation.")

        return self.environment.history

    def get_metric_history(self, metric_name: str) -> List[Any]:
        """Get the history of a specific metric.

        Args:
            metric_name: The name of the metric to retrieve.

        Returns:
            List of values for the specified metric over time.
        """
        if not self.metrics:
            raise ValueError("No simulation results available. Run the simulation first.")

        try:
            return [metrics[metric_name] for metrics in self.metrics]
        except KeyError:
            raise ValueError(f"Metric '{metric_name}' not found in simulation results")

    def reset(self) -> None:
        """Reset the simulation to its initial state."""
        super().reset()
        self.metrics = []

    @classmethod
    def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
        """Get information about the parameters required by this simulation.

        Returns:
            A dictionary mapping parameter names to their metadata.
        """
        # Get base parameters from parent class
        params = super().get_parameters_info()

        # Add class-specific parameters
        params.update({
            'agent_factory': {
                'type': 'Callable[[int], Agent]',
                'description': 'Function that creates new agents with given IDs',
                'required': True
            },
            'num_agents': {
                'type': 'int',
                'description': 'Number of agents to create',
                'required': True
            },
            'environment': {
                'type': 'Environment',
                'description': 'The environment in which agents operate',
                'required': False,
                'default': 'Default Environment'
            },
            'neighborhood_radius': {
                'type': 'float',
                'description': 'Radius for determining agent neighbors',
                'required': False,
                'default': 10.0
            },
            'save_history': {
                'type': 'bool',
                'description': 'Whether to save agent and environment history',
                'required': False,
                'default': False
            }
        })

        return params

__init__(agent_factory, num_agents, environment=None, days=100, neighborhood_radius=10.0, save_history=False, random_seed=None)

Initialize the agent-based simulation.

Parameters:

Name Type Description Default
agent_factory Callable[[int], Agent]

Function that creates new agents with given IDs.

 required
num_agents int

Number of agents to create.

 required
environment Optional[Environment]

The environment in which agents operate. If None, a default environment is created.

 None
days int

Number of steps to simulate.

 100
neighborhood_radius float

Radius for determining agent neighbors.

 10.0
save_history bool

Whether to save agent and environment history.

 False
random_seed Optional[int]

Seed for random number generation.

 None
Source code in src/sim_lab/core/agent_based_simulation.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def __init__(
    self,
    agent_factory: Callable[[int], Agent],
    num_agents: int,
    environment: Optional[Environment] = None,
    days: int = 100,
    neighborhood_radius: float = 10.0,
    save_history: bool = False,
    random_seed: Optional[int] = None
) -> None:
    """Initialize the agent-based simulation.

    Args:
        agent_factory: Function that creates new agents with given IDs.
        num_agents: Number of agents to create.
        environment: The environment in which agents operate. If None, a default environment is created.
        days: Number of steps to simulate.
        neighborhood_radius: Radius for determining agent neighbors.
        save_history: Whether to save agent and environment history.
        random_seed: Seed for random number generation.
    """
    super().__init__(days=days, random_seed=random_seed)

    self.environment = environment or Environment()
    self.neighborhood_radius = neighborhood_radius
    self.save_history = save_history

    # Create agents
    self.agents = [agent_factory(i) for i in range(num_agents)]

    # Initialize tracking variables
    self.metrics = []

calculate_metrics()

Calculate metrics for the current simulation state.

Override this method to define specific metrics for your simulation.

Returns:

Type Description
Dict[str, Any]

Dictionary of metrics derived from agent and environment states.
Source code in src/sim_lab/core/agent_based_simulation.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def calculate_metrics(self) -> Dict[str, Any]:
    """Calculate metrics for the current simulation state.

    Override this method to define specific metrics for your simulation.

    Returns:
        Dictionary of metrics derived from agent and environment states.
    """
    # Default implementation: count agents in different states
    state_counts = {}

    for agent in self.agents:
        for key, value in agent.state.items():
            if isinstance(value, (bool, int, str, float)):
                state_key = f"{key}_{value}"
                state_counts[state_key] = state_counts.get(state_key, 0) + 1

    return state_counts

get_agent_history(agent_id)

Get the state history for a specific agent.

Parameters:

Name Type Description Default
agent_id int

The ID of the agent.

 required

Returns:

Type Description
List[Dict[str, Any]]

List of state dictionaries representing the agent's history.
Source code in src/sim_lab/core/agent_based_simulation.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def get_agent_history(self, agent_id: int) -> List[Dict[str, Any]]:
    """Get the state history for a specific agent.

    Args:
        agent_id: The ID of the agent.

    Returns:
        List of state dictionaries representing the agent's history.
    """
    if not self.save_history:
        raise ValueError("Agent history was not saved. Set save_history=True when creating the simulation.")

    for agent in self.agents:
        if agent.agent_id == agent_id:
            return agent.history

    raise ValueError(f"No agent with ID {agent_id} found")

get_agent_neighbors(agent)

Get the neighbors of an agent based on proximity.

Parameters:

Name Type Description Default
agent Agent

The agent whose neighbors to find.

 required

Returns:

Type Description
List[Agent]

List of neighboring agents within the neighborhood radius.
Source code in src/sim_lab/core/agent_based_simulation.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def get_agent_neighbors(self, agent: Agent) -> List[Agent]:
    """Get the neighbors of an agent based on proximity.

    Args:
        agent: The agent whose neighbors to find.

    Returns:
        List of neighboring agents within the neighborhood radius.
    """
    if agent.position is None:
        return []  # No position, no neighbors

    neighbors = []
    for other in self.agents:
        if other.agent_id == agent.agent_id or other.position is None:
            continue  # Skip self and agents without position

        # Calculate Euclidean distance
        dx = agent.position[0] - other.position[0]
        dy = agent.position[1] - other.position[1]
        distance = (dx**2 + dy**2)**0.5

        if distance <= self.neighborhood_radius:
            neighbors.append(other)

    return neighbors

get_environment_history()

Get the environment state history.

Returns:

Type Description
List[Dict[str, Any]]

List of state dictionaries representing the environment's history.
Source code in src/sim_lab/core/agent_based_simulation.py 
279
280
281
282
283
284
285
286
287
288
def get_environment_history(self) -> List[Dict[str, Any]]:
    """Get the environment state history.

    Returns:
        List of state dictionaries representing the environment's history.
    """
    if not self.save_history:
        raise ValueError("Environment history was not saved. Set save_history=True when creating the simulation.")

    return self.environment.history

get_metric_history(metric_name)

Get the history of a specific metric.

Parameters:

Name Type Description Default
metric_name str

The name of the metric to retrieve.

 required

Returns:

Type Description
List[Any]

List of values for the specified metric over time.
Source code in src/sim_lab/core/agent_based_simulation.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def get_metric_history(self, metric_name: str) -> List[Any]:
    """Get the history of a specific metric.

    Args:
        metric_name: The name of the metric to retrieve.

    Returns:
        List of values for the specified metric over time.
    """
    if not self.metrics:
        raise ValueError("No simulation results available. Run the simulation first.")

    try:
        return [metrics[metric_name] for metrics in self.metrics]
    except KeyError:
        raise ValueError(f"Metric '{metric_name}' not found in simulation results")

get_parameters_info() classmethod

Get information about the parameters required by this simulation.

Returns:

Type Description
Dict[str, Dict[str, Any]]

A dictionary mapping parameter names to their metadata.
Source code in src/sim_lab/core/agent_based_simulation.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
@classmethod
def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
    """Get information about the parameters required by this simulation.

    Returns:
        A dictionary mapping parameter names to their metadata.
    """
    # Get base parameters from parent class
    params = super().get_parameters_info()

    # Add class-specific parameters
    params.update({
        'agent_factory': {
            'type': 'Callable[[int], Agent]',
            'description': 'Function that creates new agents with given IDs',
            'required': True
        },
        'num_agents': {
            'type': 'int',
            'description': 'Number of agents to create',
            'required': True
        },
        'environment': {
            'type': 'Environment',
            'description': 'The environment in which agents operate',
            'required': False,
            'default': 'Default Environment'
        },
        'neighborhood_radius': {
            'type': 'float',
            'description': 'Radius for determining agent neighbors',
            'required': False,
            'default': 10.0
        },
        'save_history': {
            'type': 'bool',
            'description': 'Whether to save agent and environment history',
            'required': False,
            'default': False
        }
    })

    return params

reset()

Reset the simulation to its initial state.

Source code in src/sim_lab/core/agent_based_simulation.py 
307
308
309
310
def reset(self) -> None:
    """Reset the simulation to its initial state."""
    super().reset()
    self.metrics = []

run_simulation()

Run the agent-based simulation.

Returns:

Type Description
List[Dict[str, Any]]

A list of metrics dictionaries for each time step.
Source code in src/sim_lab/core/agent_based_simulation.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def run_simulation(self) -> List[Dict[str, Any]]:
    """Run the agent-based simulation.

    Returns:
        A list of metrics dictionaries for each time step.
    """
    self.reset()

    # Initialize history if tracking
    if self.save_history:
        for agent in self.agents:
            agent.reset()
            agent.save_history()
        self.environment.reset()
        self.environment.save_history()

    # Calculate initial metrics
    self.metrics = [self.calculate_metrics()]

    # Run for specified number of days
    for _ in range(1, self.days):
        # Update agents
        for agent in self.agents:
            neighbors = self.get_agent_neighbors(agent)
            agent.update(self.environment, neighbors)

            if self.save_history:
                agent.save_history()

        # Update environment
        self.environment.update(self.agents)

        if self.save_history:
            self.environment.save_history()

        # Calculate metrics
        self.metrics.append(self.calculate_metrics())

    return self.metrics

Network Simulation

sim_lab.core.NetworkSimulation

Bases: BaseSimulation

A simulation class for network/graph dynamics.

This simulation models the evolution of a network over time, allowing for changes in node and edge attributes, as well as network structure.

Attributes:

Name Type Description
nodes Dict[Any, Node]

Dictionary of nodes in the network.
edges List[Edge]

List of edges in the network.
days int

Number of steps to simulate.
update_function Callable

Function to update the network at each time step.
save_history bool

Whether to save node and edge history.
random_seed Optional[int]

Seed for random number generation.
Source code in src/sim_lab/core/network_simulation.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
@SimulatorRegistry.register("Network")
class NetworkSimulation(BaseSimulation):
    """A simulation class for network/graph dynamics.

    This simulation models the evolution of a network over time, allowing for changes 
    in node and edge attributes, as well as network structure.

    Attributes:
        nodes (Dict[Any, Node]): Dictionary of nodes in the network.
        edges (List[Edge]): List of edges in the network.
        days (int): Number of steps to simulate.
        update_function (Callable): Function to update the network at each time step.
        save_history (bool): Whether to save node and edge history.
        random_seed (Optional[int]): Seed for random number generation.
    """

    def __init__(
        self,
        initial_nodes: Optional[Dict[Any, Dict[str, Any]]] = None,
        initial_edges: Optional[List[Tuple[Any, Any, Dict[str, Any]]]] = None,
        update_function: Optional[Callable] = None,
        directed: bool = False,
        days: int = 100,
        save_history: bool = False,
        random_seed: Optional[int] = None
    ) -> None:
        """Initialize the network simulation.

        Args:
            initial_nodes: Dictionary mapping node IDs to attribute dictionaries.
            initial_edges: List of (source, target, attributes) tuples.
            update_function: Function that updates the network at each time step.
            directed: Whether the network is directed.
            days: Number of steps to simulate.
            save_history: Whether to save node and edge history.
            random_seed: Seed for random number generation.
        """
        super().__init__(days=days, random_seed=random_seed)

        self.directed = directed
        self.save_history = save_history
        self.update_function = update_function or (lambda network, day: None)

        # Initialize nodes
        self.nodes = {}
        if initial_nodes:
            for node_id, attributes in initial_nodes.items():
                self.add_node(node_id, attributes)

        # Initialize edges
        self.edges = []
        if initial_edges:
            for source, target, attributes in initial_edges:
                weight = attributes.pop('weight', 1.0) if attributes else 1.0
                self.add_edge(source, target, directed, weight, attributes)

        # Initialize metrics tracking
        self.metrics = {}

    def add_node(self, node_id: Any, attributes: Optional[Dict[str, Any]] = None) -> Node:
        """Add a node to the network.

        Args:
            node_id: Unique identifier for the node.
            attributes: Dictionary of node attributes.

        Returns:
            The created node.

        Raises:
            ValueError: If a node with the given ID already exists.
        """
        if node_id in self.nodes:
            raise ValueError(f"Node with ID {node_id} already exists")

        node = Node(node_id, attributes)
        self.nodes[node_id] = node
        return node

    def remove_node(self, node_id: Any) -> None:
        """Remove a node from the network.

        Args:
            node_id: The ID of the node to remove.

        Raises:
            ValueError: If the node doesn't exist.
        """
        if node_id not in self.nodes:
            raise ValueError(f"Node with ID {node_id} doesn't exist")

        # Remove edges connected to this node
        self.edges = [edge for edge in self.edges if edge.source != node_id and edge.target != node_id]

        # Remove node from neighbor lists
        for node in self.nodes.values():
            if node_id in node.neighbors:
                node.neighbors.remove(node_id)

        # Remove the node
        del self.nodes[node_id]

    def add_edge(
        self,
        source: Any,
        target: Any,
        directed: Optional[bool] = None,
        weight: float = 1.0,
        attributes: Optional[Dict[str, Any]] = None
    ) -> Edge:
        """Add an edge to the network.

        Args:
            source: Source node ID.
            target: Target node ID.
            directed: Whether the edge is directed (defaults to network's directed attribute).
            weight: Edge weight.
            attributes: Dictionary of edge attributes.

        Returns:
            The created edge.

        Raises:
            ValueError: If the source or target node doesn't exist.
        """
        if source not in self.nodes:
            raise ValueError(f"Source node with ID {source} doesn't exist")
        if target not in self.nodes:
            raise ValueError(f"Target node with ID {target} doesn't exist")

        # Use network's directed attribute if not specified
        if directed is None:
            directed = self.directed

        # Create the edge
        edge = Edge(source, target, directed, weight, attributes)
        self.edges.append(edge)

        # Update node neighbor lists
        self.nodes[source].add_neighbor(target)
        if not directed:
            self.nodes[target].add_neighbor(source)

        return edge

    def remove_edge(self, source: Any, target: Any) -> None:
        """Remove an edge from the network.

        Args:
            source: Source node ID.
            target: Target node ID.

        Raises:
            ValueError: If the edge doesn't exist.
        """
        # Find the edge
        for i, edge in enumerate(self.edges):
            if edge.source == source and edge.target == target:
                # Remove from neighbor lists
                self.nodes[source].remove_neighbor(target)
                if not edge.directed:
                    self.nodes[target].remove_neighbor(source)

                # Remove the edge
                self.edges.pop(i)
                return

        # Check for undirected edge in reverse direction
        if not self.directed:
            for i, edge in enumerate(self.edges):
                if edge.source == target and edge.target == source:
                    # Remove from neighbor lists
                    self.nodes[target].remove_neighbor(source)
                    self.nodes[source].remove_neighbor(target)

                    # Remove the edge
                    self.edges.pop(i)
                    return

        raise ValueError(f"Edge from {source} to {target} doesn't exist")

    def get_adjacency_matrix(self) -> np.ndarray:
        """Get the adjacency matrix of the network.

        Returns:
            A NumPy array representing the adjacency matrix, with weights if applicable.
        """
        # Create a mapping from node IDs to indices
        node_ids = list(self.nodes.keys())
        node_to_index = {node_id: i for i, node_id in enumerate(node_ids)}

        # Initialize the adjacency matrix
        n = len(node_ids)
        adj_matrix = np.zeros((n, n))

        # Fill in the adjacency matrix
        for edge in self.edges:
            i = node_to_index[edge.source]
            j = node_to_index[edge.target]
            adj_matrix[i, j] = edge.weight
            if not edge.directed:
                adj_matrix[j, i] = edge.weight

        return adj_matrix

    def get_degree_distribution(self) -> Dict[int, int]:
        """Get the degree distribution of the network.

        Returns:
            A dictionary mapping degrees to the number of nodes with that degree.
        """
        degrees = [len(node.neighbors) for node in self.nodes.values()]
        degree_counts = {}
        for degree in degrees:
            degree_counts[degree] = degree_counts.get(degree, 0) + 1
        return degree_counts

    def calculate_metrics(self) -> Dict[str, Any]:
        """Calculate metrics for the current network state.

        Returns:
            A dictionary of network metrics.
        """
        # Number of nodes and edges
        num_nodes = len(self.nodes)
        num_edges = len(self.edges)

        # Average degree
        total_degree = sum(len(node.neighbors) for node in self.nodes.values())
        avg_degree = total_degree / num_nodes if num_nodes > 0 else 0

        # Density (ratio of actual to possible edges)
        possible_edges = num_nodes * (num_nodes - 1)
        if self.directed:
            density = num_edges / possible_edges if possible_edges > 0 else 0
        else:
            density = 2 * num_edges / possible_edges if possible_edges > 0 else 0

        # Result dictionary
        metrics = {
            'num_nodes': num_nodes,
            'num_edges': num_edges,
            'avg_degree': avg_degree,
            'density': density
        }

        return metrics

    def run_simulation(self) -> List[Dict[str, Any]]:
        """Run the network simulation.

        In each step, the network is updated according to the update function.

        Returns:
            A list of dictionaries containing network metrics for each time step.
        """
        self.reset()

        # Initialize history if tracking
        if self.save_history:
            for node in self.nodes.values():
                node.history = [node.attributes.copy()]

            for edge in self.edges:
                state = edge.attributes.copy()
                state['weight'] = edge.weight
                edge.history = [state]

        # Calculate initial metrics
        self.metrics = [self.calculate_metrics()]

        # Run for specified number of days
        for day in range(1, self.days):
            # Update the network
            self.update_function(self, day)

            # Save history if tracking
            if self.save_history:
                for node in self.nodes.values():
                    node.save_history()

                for edge in self.edges:
                    edge.save_history()

            # Calculate metrics
            self.metrics.append(self.calculate_metrics())

        return self.metrics

    def get_node_attribute_history(self, node_id: Any, attribute: str) -> List[Any]:
        """Get the history of a specific node attribute.

        Args:
            node_id: The ID of the node.
            attribute: The name of the attribute.

        Returns:
            List of values for the attribute over time.

        Raises:
            ValueError: If the node doesn't exist or history wasn't saved.
        """
        if not self.save_history:
            raise ValueError("Node history wasn't saved. Set save_history=True when creating the simulation.")

        if node_id not in self.nodes:
            raise ValueError(f"Node with ID {node_id} doesn't exist")

        return self.nodes[node_id].get_attribute_history(attribute)

    def get_edge_attribute_history(self, source: Any, target: Any, attribute: str) -> List[Any]:
        """Get the history of a specific edge attribute.

        Args:
            source: Source node ID.
            target: Target node ID.
            attribute: The name of the attribute.

        Returns:
            List of values for the attribute over time.

        Raises:
            ValueError: If the edge doesn't exist or history wasn't saved.
        """
        if not self.save_history:
            raise ValueError("Edge history wasn't saved. Set save_history=True when creating the simulation.")

        # Find the edge
        for edge in self.edges:
            if edge.source == source and edge.target == target:
                return edge.get_attribute_history(attribute)

        # Check for undirected edge in reverse direction
        if not self.directed:
            for edge in self.edges:
                if edge.source == target and edge.target == source:
                    return edge.get_attribute_history(attribute)

        raise ValueError(f"Edge from {source} to {target} doesn't exist")

    def reset(self) -> None:
        """Reset the simulation to its initial state."""
        super().reset()

        # Clear metrics
        self.metrics = []

        # Clear node and edge history
        if self.save_history:
            for node in self.nodes.values():
                node.history = [node.attributes.copy()]

            for edge in self.edges:
                state = edge.attributes.copy()
                state['weight'] = edge.weight
                edge.history = [state]

    @classmethod
    def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
        """Get information about the parameters required by this simulation.

        Returns:
            A dictionary mapping parameter names to their metadata.
        """
        # Get base parameters from parent class
        params = super().get_parameters_info()

        # Add class-specific parameters
        params.update({
            'initial_nodes': {
                'type': 'Dict[Any, Dict[str, Any]]',
                'description': 'Dictionary mapping node IDs to attribute dictionaries',
                'required': False,
                'default': '{}'
            },
            'initial_edges': {
                'type': 'List[Tuple[Any, Any, Dict[str, Any]]]',
                'description': 'List of (source, target, attributes) tuples',
                'required': False,
                'default': '[]'
            },
            'update_function': {
                'type': 'Callable',
                'description': 'Function that updates the network at each time step',
                'required': False,
                'default': 'None'
            },
            'directed': {
                'type': 'bool',
                'description': 'Whether the network is directed',
                'required': False,
                'default': 'False'
            },
            'save_history': {
                'type': 'bool',
                'description': 'Whether to save node and edge history',
                'required': False,
                'default': 'False'
            }
        })

        return params

__init__(initial_nodes=None, initial_edges=None, update_function=None, directed=False, days=100, save_history=False, random_seed=None)

Initialize the network simulation.

Parameters:

Name Type Description Default
initial_nodes Optional[Dict[Any, Dict[str, Any]]]

Dictionary mapping node IDs to attribute dictionaries.

 None
initial_edges Optional[List[Tuple[Any, Any, Dict[str, Any]]]]

List of (source, target, attributes) tuples.

 None
update_function Optional[Callable]

Function that updates the network at each time step.

 None
directed bool

Whether the network is directed.

 False
days int

Number of steps to simulate.

 100
save_history bool

Whether to save node and edge history.

 False
random_seed Optional[int]

Seed for random number generation.

 None
Source code in src/sim_lab/core/network_simulation.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def __init__(
    self,
    initial_nodes: Optional[Dict[Any, Dict[str, Any]]] = None,
    initial_edges: Optional[List[Tuple[Any, Any, Dict[str, Any]]]] = None,
    update_function: Optional[Callable] = None,
    directed: bool = False,
    days: int = 100,
    save_history: bool = False,
    random_seed: Optional[int] = None
) -> None:
    """Initialize the network simulation.

    Args:
        initial_nodes: Dictionary mapping node IDs to attribute dictionaries.
        initial_edges: List of (source, target, attributes) tuples.
        update_function: Function that updates the network at each time step.
        directed: Whether the network is directed.
        days: Number of steps to simulate.
        save_history: Whether to save node and edge history.
        random_seed: Seed for random number generation.
    """
    super().__init__(days=days, random_seed=random_seed)

    self.directed = directed
    self.save_history = save_history
    self.update_function = update_function or (lambda network, day: None)

    # Initialize nodes
    self.nodes = {}
    if initial_nodes:
        for node_id, attributes in initial_nodes.items():
            self.add_node(node_id, attributes)

    # Initialize edges
    self.edges = []
    if initial_edges:
        for source, target, attributes in initial_edges:
            weight = attributes.pop('weight', 1.0) if attributes else 1.0
            self.add_edge(source, target, directed, weight, attributes)

    # Initialize metrics tracking
    self.metrics = {}

add_edge(source, target, directed=None, weight=1.0, attributes=None)

Add an edge to the network.

Parameters:

Name Type Description Default
source Any

Source node ID.

 required
target Any

Target node ID.

 required
directed Optional[bool]

Whether the edge is directed (defaults to network's directed attribute).

 None
weight float

Edge weight.

 1.0
attributes Optional[Dict[str, Any]]

Dictionary of edge attributes.

 None

Returns:

Type Description
Edge

The created edge.

Raises:

Type Description
ValueError

If the source or target node doesn't exist.
Source code in src/sim_lab/core/network_simulation.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def add_edge(
    self,
    source: Any,
    target: Any,
    directed: Optional[bool] = None,
    weight: float = 1.0,
    attributes: Optional[Dict[str, Any]] = None
) -> Edge:
    """Add an edge to the network.

    Args:
        source: Source node ID.
        target: Target node ID.
        directed: Whether the edge is directed (defaults to network's directed attribute).
        weight: Edge weight.
        attributes: Dictionary of edge attributes.

    Returns:
        The created edge.

    Raises:
        ValueError: If the source or target node doesn't exist.
    """
    if source not in self.nodes:
        raise ValueError(f"Source node with ID {source} doesn't exist")
    if target not in self.nodes:
        raise ValueError(f"Target node with ID {target} doesn't exist")

    # Use network's directed attribute if not specified
    if directed is None:
        directed = self.directed

    # Create the edge
    edge = Edge(source, target, directed, weight, attributes)
    self.edges.append(edge)

    # Update node neighbor lists
    self.nodes[source].add_neighbor(target)
    if not directed:
        self.nodes[target].add_neighbor(source)

    return edge

add_node(node_id, attributes=None)

Add a node to the network.

Parameters:

Name Type Description Default
node_id Any

Unique identifier for the node.

 required
attributes Optional[Dict[str, Any]]

Dictionary of node attributes.

 None

Returns:

Type Description
Node

The created node.

Raises:

Type Description
ValueError

If a node with the given ID already exists.
Source code in src/sim_lab/core/network_simulation.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def add_node(self, node_id: Any, attributes: Optional[Dict[str, Any]] = None) -> Node:
    """Add a node to the network.

    Args:
        node_id: Unique identifier for the node.
        attributes: Dictionary of node attributes.

    Returns:
        The created node.

    Raises:
        ValueError: If a node with the given ID already exists.
    """
    if node_id in self.nodes:
        raise ValueError(f"Node with ID {node_id} already exists")

    node = Node(node_id, attributes)
    self.nodes[node_id] = node
    return node

calculate_metrics()

Calculate metrics for the current network state.

Returns:

Type Description
Dict[str, Any]

A dictionary of network metrics.
Source code in src/sim_lab/core/network_simulation.py 
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def calculate_metrics(self) -> Dict[str, Any]:
    """Calculate metrics for the current network state.

    Returns:
        A dictionary of network metrics.
    """
    # Number of nodes and edges
    num_nodes = len(self.nodes)
    num_edges = len(self.edges)

    # Average degree
    total_degree = sum(len(node.neighbors) for node in self.nodes.values())
    avg_degree = total_degree / num_nodes if num_nodes > 0 else 0

    # Density (ratio of actual to possible edges)
    possible_edges = num_nodes * (num_nodes - 1)
    if self.directed:
        density = num_edges / possible_edges if possible_edges > 0 else 0
    else:
        density = 2 * num_edges / possible_edges if possible_edges > 0 else 0

    # Result dictionary
    metrics = {
        'num_nodes': num_nodes,
        'num_edges': num_edges,
        'avg_degree': avg_degree,
        'density': density
    }

    return metrics

get_adjacency_matrix()

Get the adjacency matrix of the network.

Returns:

Type Description
ndarray

A NumPy array representing the adjacency matrix, with weights if applicable.
Source code in src/sim_lab/core/network_simulation.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def get_adjacency_matrix(self) -> np.ndarray:
    """Get the adjacency matrix of the network.

    Returns:
        A NumPy array representing the adjacency matrix, with weights if applicable.
    """
    # Create a mapping from node IDs to indices
    node_ids = list(self.nodes.keys())
    node_to_index = {node_id: i for i, node_id in enumerate(node_ids)}

    # Initialize the adjacency matrix
    n = len(node_ids)
    adj_matrix = np.zeros((n, n))

    # Fill in the adjacency matrix
    for edge in self.edges:
        i = node_to_index[edge.source]
        j = node_to_index[edge.target]
        adj_matrix[i, j] = edge.weight
        if not edge.directed:
            adj_matrix[j, i] = edge.weight

    return adj_matrix

get_degree_distribution()

Get the degree distribution of the network.

Returns:

Type Description
Dict[int, int]

A dictionary mapping degrees to the number of nodes with that degree.
Source code in src/sim_lab/core/network_simulation.py 
352
353
354
355
356
357
358
359
360
361
362
def get_degree_distribution(self) -> Dict[int, int]:
    """Get the degree distribution of the network.

    Returns:
        A dictionary mapping degrees to the number of nodes with that degree.
    """
    degrees = [len(node.neighbors) for node in self.nodes.values()]
    degree_counts = {}
    for degree in degrees:
        degree_counts[degree] = degree_counts.get(degree, 0) + 1
    return degree_counts

get_edge_attribute_history(source, target, attribute)

Get the history of a specific edge attribute.

Parameters:

Name Type Description Default
source Any

Source node ID.

 required
target Any

Target node ID.

 required
attribute str

The name of the attribute.

 required

Returns:

Type Description
List[Any]

List of values for the attribute over time.

Raises:

Type Description
ValueError

If the edge doesn't exist or history wasn't saved.
Source code in src/sim_lab/core/network_simulation.py 
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
def get_edge_attribute_history(self, source: Any, target: Any, attribute: str) -> List[Any]:
    """Get the history of a specific edge attribute.

    Args:
        source: Source node ID.
        target: Target node ID.
        attribute: The name of the attribute.

    Returns:
        List of values for the attribute over time.

    Raises:
        ValueError: If the edge doesn't exist or history wasn't saved.
    """
    if not self.save_history:
        raise ValueError("Edge history wasn't saved. Set save_history=True when creating the simulation.")

    # Find the edge
    for edge in self.edges:
        if edge.source == source and edge.target == target:
            return edge.get_attribute_history(attribute)

    # Check for undirected edge in reverse direction
    if not self.directed:
        for edge in self.edges:
            if edge.source == target and edge.target == source:
                return edge.get_attribute_history(attribute)

    raise ValueError(f"Edge from {source} to {target} doesn't exist")

get_node_attribute_history(node_id, attribute)

Get the history of a specific node attribute.

Parameters:

Name Type Description Default
node_id Any

The ID of the node.

 required
attribute str

The name of the attribute.

 required

Returns:

Type Description
List[Any]

List of values for the attribute over time.

Raises:

Type Description
ValueError

If the node doesn't exist or history wasn't saved.
Source code in src/sim_lab/core/network_simulation.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
def get_node_attribute_history(self, node_id: Any, attribute: str) -> List[Any]:
    """Get the history of a specific node attribute.

    Args:
        node_id: The ID of the node.
        attribute: The name of the attribute.

    Returns:
        List of values for the attribute over time.

    Raises:
        ValueError: If the node doesn't exist or history wasn't saved.
    """
    if not self.save_history:
        raise ValueError("Node history wasn't saved. Set save_history=True when creating the simulation.")

    if node_id not in self.nodes:
        raise ValueError(f"Node with ID {node_id} doesn't exist")

    return self.nodes[node_id].get_attribute_history(attribute)

get_parameters_info() classmethod

Get information about the parameters required by this simulation.

Returns:

Type Description
Dict[str, Dict[str, Any]]

A dictionary mapping parameter names to their metadata.
Source code in src/sim_lab/core/network_simulation.py 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
@classmethod
def get_parameters_info(cls) -> Dict[str, Dict[str, Any]]:
    """Get information about the parameters required by this simulation.

    Returns:
        A dictionary mapping parameter names to their metadata.
    """
    # Get base parameters from parent class
    params = super().get_parameters_info()

    # Add class-specific parameters
    params.update({
        'initial_nodes': {
            'type': 'Dict[Any, Dict[str, Any]]',
            'description': 'Dictionary mapping node IDs to attribute dictionaries',
            'required': False,
            'default': '{}'
        },
        'initial_edges': {
            'type': 'List[Tuple[Any, Any, Dict[str, Any]]]',
            'description': 'List of (source, target, attributes) tuples',
            'required': False,
            'default': '[]'
        },
        'update_function': {
            'type': 'Callable',
            'description': 'Function that updates the network at each time step',
            'required': False,
            'default': 'None'
        },
        'directed': {
            'type': 'bool',
            'description': 'Whether the network is directed',
            'required': False,
            'default': 'False'
        },
        'save_history': {
            'type': 'bool',
            'description': 'Whether to save node and edge history',
            'required': False,
            'default': 'False'
        }
    })

    return params

remove_edge(source, target)

Remove an edge from the network.

Parameters:

Name Type Description Default
source Any

Source node ID.

 required
target Any

Target node ID.

 required

Raises:

Type Description
ValueError

If the edge doesn't exist.
Source code in src/sim_lab/core/network_simulation.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
def remove_edge(self, source: Any, target: Any) -> None:
    """Remove an edge from the network.

    Args:
        source: Source node ID.
        target: Target node ID.

    Raises:
        ValueError: If the edge doesn't exist.
    """
    # Find the edge
    for i, edge in enumerate(self.edges):
        if edge.source == source and edge.target == target:
            # Remove from neighbor lists
            self.nodes[source].remove_neighbor(target)
            if not edge.directed:
                self.nodes[target].remove_neighbor(source)

            # Remove the edge
            self.edges.pop(i)
            return

    # Check for undirected edge in reverse direction
    if not self.directed:
        for i, edge in enumerate(self.edges):
            if edge.source == target and edge.target == source:
                # Remove from neighbor lists
                self.nodes[target].remove_neighbor(source)
                self.nodes[source].remove_neighbor(target)

                # Remove the edge
                self.edges.pop(i)
                return

    raise ValueError(f"Edge from {source} to {target} doesn't exist")

remove_node(node_id)

Remove a node from the network.

Parameters:

Name Type Description Default
node_id Any

The ID of the node to remove.

 required

Raises:

Type Description
ValueError

If the node doesn't exist.
Source code in src/sim_lab/core/network_simulation.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def remove_node(self, node_id: Any) -> None:
    """Remove a node from the network.

    Args:
        node_id: The ID of the node to remove.

    Raises:
        ValueError: If the node doesn't exist.
    """
    if node_id not in self.nodes:
        raise ValueError(f"Node with ID {node_id} doesn't exist")

    # Remove edges connected to this node
    self.edges = [edge for edge in self.edges if edge.source != node_id and edge.target != node_id]

    # Remove node from neighbor lists
    for node in self.nodes.values():
        if node_id in node.neighbors:
            node.neighbors.remove(node_id)

    # Remove the node
    del self.nodes[node_id]

reset()

Reset the simulation to its initial state.

Source code in src/sim_lab/core/network_simulation.py 
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
def reset(self) -> None:
    """Reset the simulation to its initial state."""
    super().reset()

    # Clear metrics
    self.metrics = []

    # Clear node and edge history
    if self.save_history:
        for node in self.nodes.values():
            node.history = [node.attributes.copy()]

        for edge in self.edges:
            state = edge.attributes.copy()
            state['weight'] = edge.weight
            edge.history = [state]

run_simulation()

Run the network simulation.

In each step, the network is updated according to the update function.

Returns:

Type Description
List[Dict[str, Any]]

A list of dictionaries containing network metrics for each time step.
Source code in src/sim_lab/core/network_simulation.py 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
def run_simulation(self) -> List[Dict[str, Any]]:
    """Run the network simulation.

    In each step, the network is updated according to the update function.

    Returns:
        A list of dictionaries containing network metrics for each time step.
    """
    self.reset()

    # Initialize history if tracking
    if self.save_history:
        for node in self.nodes.values():
            node.history = [node.attributes.copy()]

        for edge in self.edges:
            state = edge.attributes.copy()
            state['weight'] = edge.weight
            edge.history = [state]

    # Calculate initial metrics
    self.metrics = [self.calculate_metrics()]

    # Run for specified number of days
    for day in range(1, self.days):
        # Update the network
        self.update_function(self, day)

        # Save history if tracking
        if self.save_history:
            for node in self.nodes.values():
                node.save_history()

            for edge in self.edges:
                edge.save_history()

        # Calculate metrics
        self.metrics.append(self.calculate_metrics())

    return self.metrics