CovidContagionCalibrator

This example demonstrates how to use Melodie’s Calibrator module. It extends the base covid_contagion model by using a genetic algorithm (GA) to find the value of the infection_prob parameter that results in a target final infected ratio.

Calibrator: Project Structure

examples/covid_contagion_calibrator
│   ├── core/
│   │   ├── calibrator.py
│   │   ├── scenario.py
│   │   ├── data_collector.py
│   │   └── model.py
│   ├── data/
│   │   ├── CalibratorScenarios.csv
│   │   └── CalibratorParamsScenarios.csv
│   └── output/
└── main.py

Calibrator: Key Changes

The primary changes for the calibrator example are:

• A ``calibrator.py`` Module: This new file defines the CovidCalibrator class, which inherits from Melodie.Calibrator.

• Calibrator Scenarios: The input data now includes CalibratorScenarios.csv and CalibratorParamsScenarios.csv to control the calibration process. CalibratorScenarios.csv is similar to SimulatorScenarios but does not use a run_num column.

• Path Modification in ``main.py``: A special modification to sys.path is included. This is necessary because the Calibrator uses multiprocessing for parallel execution. Worker processes are spawned in a new environment and need to be able to find and import the project’s modules (e.g., from examples.covid_contagion_calibrator.core…). This path modification is a robust way to ensure modules are discoverable when running examples directly from the Melodie repository, which is not a standard installed package.

Calibrator: GA Concepts

The primary goal of the Calibrator is to automatically tune one or more model parameters to match observed, real-world data. It treats this as an optimization problem: finding the set of parameters that minimizes the “distance” between the model’s output and a specified target.

Core Idea: Minimizing Distance

• Target: You define a target outcome. In this example, the target is for the final proportion of the population that has been infected to be 80%. This target is hard-coded in the distance function.

• Parameters: You specify which scenario parameter(s) the Calibrator should adjust. Here, it’s the infection_prob.

• Distance Function: You must implement a distance(model) method. This function is the core of the calibration. It runs after a simulation is complete, calculates a metric from the model’s final state, and returns a single float value representing how far off that metric is from the target. The Calibrator’s goal is to make this value as close to zero as possible.

Genetic Algorithm (GA) for Optimization

Melodie uses a genetic algorithm to search the parameter space efficiently.

• Chromosome: A single set of the parameters being calibrated (in this case, just a single value for infection_prob) is treated as a “chromosome.”

• Population: The GA starts with a “population” of these chromosomes, i.e., many different random values for infection_prob.

• Fitness: For each chromosome, the model is run, and the distance function is calculated. This distance is the fitness score (where lower is better).

• Evolution: The GA then proceeds through “generations.” In each generation, it selects the best-performing chromosomes (those that produced the smallest distance), “breeds” them (crossover), and introduces random changes (mutation) to create a new population of parameter sets to test. This process iteratively converges toward the parameter set that produces the best fit to the target.

Parameter Encoding: From Float to Binary

A key aspect of the genetic algorithm is how it represents continuous parameters like infection_prob (a float between 0.0 and 1.0). The GA does not work with the float values directly; instead, it converts them into a binary string of a fixed length. This process is called discretization.

• Encoding: The range of a parameter (e.g., from infection_prob_min to infection_prob_max) is mapped to a sequence of integers, which are then represented as binary numbers. For instance, a float value is converted into a binary string like 01101.

• GA Operations: All genetic operations, such as crossover (swapping parts of two binary strings) and mutation (flipping a random bit, e.g., 01101 -> 01111), are performed on these binary representations.

• Decoding: When a simulation needs to be run, the binary string is decoded back into its corresponding float value. This is done in three steps:

  1. The binary string is converted to its decimal integer representation.

  2. This integer is normalized to a float between 0.0 and 1.0 by dividing it by the maximum possible integer for that bit length (which is 2^length - 1).

  3. This normalized float is linearly mapped to the parameter’s defined range (e.g., from infection_prob_min to infection_prob_max).

• Precision: The strategy_param_code_length parameter in the .csv file determines the length of this binary string. A longer string allows for a more fine-grained representation of the parameter space (more steps between the min and max bounds), thus offering higher precision at the cost of a larger search space.

Parameter Configuration (`CalibratorParamsScenarios.csv`)

This file controls the behavior of the genetic algorithm:

• id: Unique identifier for a set of GA parameters.

• path_num: How many independent calibration processes (paths) to run. Each path is a complete run of the GA from a random start, which helps ensure the result is robust and not just a local minimum.

• generation_num: The number of generations the GA will run for in each path.

• strategy_population: The size of the population in each generation (how many different parameter sets are tested).

• mutation_prob: The probability of a random mutation occurring during breeding.

• strategy_param_code_length: The precision of the parameters, defined by the bit-length of the chromosome.

• infection_prob_min, infection_prob_max: The lower and upper bounds for the infection_prob parameter search space.

Calibrator: Input Data

In this example, the distance is calculated based on the model’s state at the end of the simulation. If you need to use data from all periods to calculate the target metric, you can compute and store the required value in an environment property throughout the simulation, ensuring it holds the final desired value at the last period.

Calibrator: Running the Model

You can run the calibrator using the main script:

python examples/covid_contagion_calibrator/main.py

This will execute the genetic algorithm, running multiple simulations in parallel to find the optimal infection_prob. The results, including the progression of parameters and distances across generations, are saved to the data/output folder.

Parallel Execution Mode

The Calibrator supports two parallelization modes, controlled by the parallel_mode parameter when creating the calibrator instance:

• ``parallel_mode=”process”`` (default): Uses subprocess-based parallelism via multiprocessing. This is the traditional approach and works on all Python versions. It is recommended for most use cases.

• ``parallel_mode=”thread”``: Uses thread-based parallelism via ThreadPoolExecutor. This mode is recommended for Python 3.13+ (free-threaded/No-GIL builds) as it can provide better performance by avoiding the overhead of process creation and data serialization. In older Python versions, this mode will still run but may be limited by the Global Interpreter Lock (GIL).

You can specify the mode when creating the calibrator:

calibrator = CovidCalibrator(
    config=cfg,
    model_cls=CovidModel,
    scenario_cls=CovidScenario,
    processors=8,
    parallel_mode="thread",  # or "process" (default)
)

Calibrator: Code

This section shows the key code implementation for the calibrator model. Files that are identical to the base covid_contagion model are noted.

Calibrator Definition

Defined in core/calibrator.py.

import os
from typing import List

import pandas as pd
from Melodie import Calibrator

from examples.covid_contagion_calibrator.core.model import CovidModel
from examples.covid_contagion_calibrator.core.scenario import CovidScenario


class CovidCalibrator(Calibrator):
    """
    Simple calibrator that tunes `infection_prob` so the final infected ratio matches a target.
    
    In this example, the target infected ratio is hardcoded to 0.8 inside the `distance` method.
    The calibrator minimizes the squared difference between the actual infected ratio and this target.
    """

    scenario_cls: type[CovidScenario]
    model_cls: type[CovidModel]

    def setup(self) -> None:
        """
        Setup the calibrator by defining which properties to tune and which to record.
        """
        # Calibrate the `infection_prob` parameter in the scenario.
        # This tells the Genetic Algorithm (GA) to optimize this specific attribute.
        self.add_scenario_calibrating_property("infection_prob")
        
        # Record environment-level data for analysis.
        # 'num_susceptible' will be recorded in `Result_Calibrator_Environment.csv`
        # allowing us to verify the calibration result.
        self.add_environment_property("num_susceptible")

    def distance(self, model: CovidModel) -> float:
        """
        Calculates the distance (error) between the model's result and the target.
        The GA minimizes this value.
        """
        env = model.environment
        
        # Calculate the ratio of agents who were infected (including recovered)
        # 1 - (susceptible / total)
        infected_ratio = 1 - env.num_susceptible / env.scenario.agent_num
        
        # Return squared error. 
        # Target is hardcoded as 0.8 (80% infection rate).
        return (infected_ratio - 0.8) ** 2

Scenario Definition

Defined in core/scenario.py.

from Melodie import Scenario


class CovidScenario(Scenario):
    """
    Defines the parameters for each simulation run of the virus contagion model.
    It inherits from the base `Melodie.Scenario`.

    Special Melodie attributes (automatically populated from `SimulatorScenarios`):
    - `id`: A unique identifier for the scenario.
    - `run_num`: How many times to repeat this scenario (for stochastic models). Defaults to 1.
    - `period_num`: How many simulation steps (periods) for each run. Defaults to 0.
    """

    def setup(self) -> None:
        """
        Declares custom scenario parameters for type hinting and clarity.
        These are automatically populated from columns in `SimulatorScenarios`.
        """
        self.agent_num: int = 0
        self.initial_infected_percentage: float = 0.0
        self.infection_prob: float = 0.0
        self.recovery_prob: float = 0.0

    def load_data(self) -> None:
        """
        This method is automatically called by Melodie after scenario parameters are set.
        It is the recommended place to load all static input dataframes, making them
        accessible via `self.scenario.*` from `Model`, `Agent`, and `Environment`.
        """
        self.health_states = self.load_dataframe("ID_HealthState.csv")

Model Structure

Identical to the base model. Defined in core/model.py.

from typing import TYPE_CHECKING

from Melodie import Model
from .agent import CovidAgent
from .environment import CovidEnvironment
from .data_collector import CovidDataCollector
from .scenario import CovidScenario

if TYPE_CHECKING:
    from Melodie import AgentList


class CovidModel(Model):
    scenario: "CovidScenario"
    agents: "AgentList[CovidAgent]"
    environment: CovidEnvironment
    data_collector: CovidDataCollector

    def create(self) -> None:
        """
        This method is called once at the beginning of the simulation
        to create the core components of the model.
        """
        self.agents = self.create_agent_list(CovidAgent)
        self.environment = self.create_environment(CovidEnvironment)
        self.data_collector = self.create_data_collector(CovidDataCollector)

    def setup(self) -> None:
        """
        This method is called once after `create` to set up the initial state
        of the model components.
        """
        self.agents.setup_agents(self.scenario.agent_num)
        self.environment.setup_infection(self.agents)

    def run(self) -> None:
        """
        This method defines the main simulation loop that runs for `scenario.period_num` periods.
        """
        for t in self.iterator(self.scenario.period_num):
            self.environment.agents_interaction(self.agents)
            self.environment.agents_recover(self.agents)
            self.environment.update_population_stats(self.agents)
            self.data_collector.collect(t)
        self.data_collector.save()

Environment Logic

Identical to the base model. Defined in core/environment.py.

import random
from typing import TYPE_CHECKING

from Melodie import Environment
from .agent import CovidAgent
from .scenario import CovidScenario

if TYPE_CHECKING:
    from Melodie import AgentList


class CovidEnvironment(Environment):
    scenario: "CovidScenario"

    def setup(self) -> None:
        # Macro-level counters for population statistics.
        # These are updated each period by `update_population_stats` and recorded by the DataCollector.
        self.num_susceptible: int = 0
        self.num_infected: int = 0
        self.num_recovered: int = 0

    def setup_infection(self, agents: "AgentList[CovidAgent]") -> None:
        # Sets the initial percentage of infected agents based on scenario parameters.
        for agent in agents:
            if (
                agent.health_state == 0
                and random.random() < self.scenario.initial_infected_percentage
            ):
                agent.health_state = 1

    def agents_interaction(self, agents: "AgentList[CovidAgent]") -> None:
        # Simulates random interactions where infected agents can spread the virus.
        for agent in agents:
            if agent.health_state == 1:
                # Randomly meet another agent
                # Note: Melodie's AgentList.random_sample returns a list
                other_agent = agents.random_sample(1)[0]
                if other_agent.health_state == 0:
                    if random.random() < self.scenario.infection_prob:
                        other_agent.health_state = 1
    
    def agents_recover(self, agents: "AgentList[CovidAgent]") -> None:
        # Triggers the recovery process for all infected agents.
        for agent in agents:
            agent.health_state_update(self.scenario.recovery_prob)

    def update_population_stats(self, agents: "AgentList[CovidAgent]") -> None:
        # Aggregates agent states into environment-level population counts.
        self.num_susceptible = 0
        self.num_infected = 0
        self.num_recovered = 0

        for agent in agents:
            if agent.health_state == 0:
                self.num_susceptible += 1
            elif agent.health_state == 1:
                self.num_infected += 1
            elif agent.health_state == 2:
                self.num_recovered += 1

Agent Behavior

Identical to the base model. Defined in core/agent.py.

import random
from Melodie import Agent


class CovidAgent(Agent):
    def setup(self) -> None:
        """
        Initializes the agent's state.
        `health_state`: 0 = susceptible, 1 = infected, 2 = recovered.
        """
        self.health_state: int = 0  # All agents start as susceptible.

    def health_state_update(self, recovery_prob: float) -> None:
        """
        Agent-level logic for transitioning from infected to recovered.
        """
        if self.health_state == 1 and random.random() < recovery_prob:
            self.health_state = 2

Data Collection Setup

Identical to the base model. Defined in core/data_collector.py.

from Melodie import DataCollector


class CovidDataCollector(DataCollector):
    def setup(self) -> None:
        """
        Registers which properties of agents and the environment should be recorded.

        In this minimal example we record:
        - micro-level results: each agent's ``health_state``;
        - macro-level results: population counts on the environment
          (``num_susceptible``, ``num_infected``, ``num_recovered``).
        """
        self.add_agent_property("agents", "health_state")
        self.add_environment_property("num_susceptible")
        self.add_environment_property("num_infected")
        self.add_environment_property("num_recovered")