ReformSimulation

A class used to build and simulate reform data.

This class inherits from EmpiricalSimulator and provides methods to build and simulate reform data. It includes methods to build simulation data, simulate reforms, and iterate over multiple reform simulations.

Attributes:

Name	Type	Description
log_filename	PathLike	The path to the log file for the simulation. Default is a path in the logs directory.

Source code in bozio_wasmer_simulations/simulation/empirical/base.py

class ReformSimulation(EmpiricalSimulator):
    """
    A class used to build and simulate reform data.

    This class inherits from EmpiricalSimulator and provides methods to build and simulate reform data.
    It includes methods to build simulation data, simulate reforms, and iterate over multiple reform simulations.

    Attributes:
        log_filename (os.PathLike, optional):
            The path to the log file for the simulation. Default is a path in the logs directory.

    """

    # Initialisation
    def __init__(
        self,
        log_filename: Optional[os.PathLike] = os.path.join(
            FILE_PATH.parents[3], "logs/reform_simulation.log"
        ),
    ) -> None:
        """
        Constructs all the necessary attributes for the ReformSimulation object.

        Args:
            log_filename (os.PathLike, optional): The path to the log file. Defaults to os.path.join(FILE_PATH.parents[3], 'logs/reform_simulation.log').

        Returns:
            None
        """
        # Initialisation du simulateur
        super().__init__(log_filename=log_filename)

    # Fonction auxilaire de construction des données contenant les résultats d'une simulation à partir de laquelle dériver une réforme
    def build_data_simulation(
        self,
        data: Optional[Union[pd.DataFrame, None]] = None,
        path: Optional[Union[os.PathLike, None]] = None,
    ) -> None:
        """
        Builds the simulation data.

        If data is provided, it uses _build_data_simulation_from_dataframe to build the simulation data.
        If not, it reads the data from the provided path.

        Args:
            data (pd.DataFrame, optional):
                The data to use for building the simulation data. If None, the data is read from the path.
            path (os.PathLike, optional):
                The path to the data file. Used if data is None.

        Returns:
            None
        """
        if data is not None:
            self._build_data_simulation_from_dataframe(data=data)
        else:
            self.data_simulation = pd.read_csv(path)

    # Fonction auxiliaire de construction des données de simulation à partir d'un DataFrame
    def _build_data_simulation_from_dataframe(self, data: pd.DataFrame) -> None:
        """
        Builds the simulation data from a DataFrame.

        Checks if the DataFrame contains all the required variables. If it does, it sets the data as the simulation data.
        If it doesn't, it logs an error and raises a ValueError.

        Args:
            data (pd.DataFrame):
                The DataFrame to use for building the simulation data.

        Returns:
            None
        """
        # Vérification que l'ensemble des variables attendues sont dans le jeu de données
        # Variables manquantes
        missing_variables = np.setdiff1d(
            params["VARIABLES"], data.columns.tolist()
        ).tolist()
        if missing_variables == []:
            self.data_simulation = data
            # Logging
            self.logger.info("Successfully build data_simulation")
        else:
            # Logging
            self.logger.error(
                f"Given DataFrame should contain {missing_variables} as columns"
            )
            # Erreur
            raise ValueError(
                f"Given DataFrame should contain {missing_variables} as columns"
            )

    # Fonction auxiliaire d'itération d'une simulation de réforme
    def simulate_reform(
        self,
        name: str,
        reform_params: dict,
        year: int,
        taux_bascule_vm: Optional[Union[float, None]] = None,
    ) -> None:
        """
        Simulates a reform.

        Applies the reform to the simulation data and calculates new variables. The new variables are added to the simulation data.

        Args:
            name (str):
                The name of the reform.
            reform_params (dict):
                The parameters of the reform.
            year (int):
                The year for the simulation.
            taux_bascule_vm (float, optional):
                The rate of the "versement mobilité" (VM) switch. If provided, new variables are calculated for the VM switch.

        Returns:
            None
        """
        # Simulation du SMIC proratisé
        if "smic_proratise" not in self.data_simulation.columns:
            self.data_simulation = self.simulate_smic_proratise(
                data=self.data_simulation, year=year, list_var_exclude=[], inplace=True
            )

        # Initialisation des paramètres du système sociofiscal
        tax_benefit_system = FranceTaxBenefitSystem()

        # Application de la réforme
        reformed_tax_benefit_system = create_and_apply_structural_reform_ag(
            tax_benefit_system=tax_benefit_system, dict_params=reform_params
        )

        # Extraction du type de la réforme
        reform_type = reform_params["TYPE"]
        # Itération de la simulation
        self.data_simulation = self.iterate_simulation(
            data=self.data_simulation,
            tax_benefit_system=reformed_tax_benefit_system,
            year=year,
            list_var_simul=[f"new_allegement_{reform_type}"],
            list_var_exclude=params["REFORM"]["VAR_EXCLUDE"],
            inplace=True,
        )

        # Renomination de la variable simulée pour correspondre au nom du scénario
        self.data_simulation.rename(
            {f"new_allegement_{reform_type}": f"new_allegement_{name.lower()}"},
            axis=1,
            inplace=True,
        )
        # Somme des exonérations et allègements
        self.data_simulation[f"exonerations_et_allegements_{name.lower()}"] = (
            self.data_simulation[
                ["exonerations", f"new_allegement_{name.lower()}"]
            ].sum(axis=1)
        )
        # Calcul du salaire brut avec la réforme
        self.data_simulation[f"salaire_super_brut_{name.lower()}"] = (
            self.data_simulation["salaire_super_brut_hors_allegements"]
            - self.data_simulation[f"exonerations_et_allegements_{name.lower()}"]
            + self.data_simulation["prime_partage_valeur_exoneree"]
        )
        # Construction des variables de variation du coût du travail
        self.data_simulation = build_data_evol_ct(
            data_source=self.data_simulation,
            col_new_ct=f"salaire_super_brut_{name.lower()}",
            col_ct="salaire_super_brut",
            to_concat=True,
        )
        # Logging
        self.logger.info(
            f"Successfully built variables related to labor costs variations inducted by the reform"
        )

        # Construction d'un nouveau coût du travail associé à la bascule d'une fraction du versement transport
        if taux_bascule_vm is not None:
            self.data_simulation[f"salaire_super_brut_vm_{name.lower()}"] = (
                self.data_simulation["salaire_super_brut_hors_allegements"]
                + self.data_simulation["versement_transport"] * (1 - taux_bascule_vm)
                - self.data_simulation[f"exonerations_et_allegements_{name.lower()}"]
                + self.data_simulation["prime_partage_valeur_exoneree"]
            )
            # Construction des variables de variation du coût du travail
            self.data_simulation = build_data_evol_ct(
                data_source=self.data_simulation,
                col_new_ct=f"salaire_super_brut_vm_{name.lower()}",
                col_ct="salaire_super_brut",
                to_concat=True,
            )
            # Logging
            self.logger.info(
                f"Successfully built variables related to labor costs variations inducted by the reform and the vm"
            )

    # Fonction auxiliaire de d'itération de l'ensemble des simulations de réformes
    def iterate_reform_simulations(
        self,
        scenarios: dict,
        year: int,
        taux_bascule_vm: Optional[Union[float, None]] = None,
    ) -> None:
        """
        Iterates over multiple reform simulations.

        Simulates each reform in the provided scenarios. The results are added to the simulation data.

        Args:
            scenarios (dict):
                The scenarios to simulate. Each scenario is a dictionary of reform parameters.
            year (int):
                The year for the simulation.
            taux_bascule_vm (float, optional):
                The rate of the value-added tax (VM) switch. If provided, new variables are calculated for the VM switch.

        Returns:
            None
        """
        # Itération sur les scénarii référencés dans le jeu de données de paramètres
        for scenario in tqdm(scenarios.keys()):
            # Itération des réformes
            self.simulate_reform(
                name=scenario,
                reform_params=scenarios[scenario],
                year=year,
                taux_bascule_vm=taux_bascule_vm,
            )

    # Fonction de construction du jeu de données
    def build(
        self,
        scenarios: dict,
        year: int,
        taux_bascule_vm: Optional[Union[float, None]] = None,
        data: Optional[Union[pd.DataFrame, None]] = None,
        path: Optional[Union[os.PathLike, None]] = None,
    ) -> pd.DataFrame:
        """
        Builds the reform simulation data.

        Calls build_data_simulation and iterate_reform_simulations to build the simulation data.

        Args:
            scenarios (dict):
                The scenarios to simulate. Each scenario is a dictionary of reform parameters.
            year (int):
                The year for the simulation.
            taux_bascule_vm (float, optional):
                The rate of the value-added tax (VM) switch. If provided, new variables are calculated for the VM switch.
            data (pd.DataFrame, optional):
                The data to use for building the simulation data. If None, the data is read from the path.
            path (os.PathLike, optional):
                The path to the data file. Used if data is None.

        Returns:
            (pd.DataFrame): The simulation data.
        """
        # Ajout du jeu de données de simulations
        self.build_data_simulation(data=data, path=path)
        # Itération des simulations de réformes
        self.iterate_reform_simulations(
            scenarios=scenarios, year=year, taux_bascule_vm=taux_bascule_vm
        )

        return self.data_simulation

build

build(scenarios: dict, year: int, taux_bascule_vm: Optional[Union[float, None]] = None, data: Optional[Union[DataFrame, None]] = None, path: Optional[Union[PathLike, None]] = None) -> DataFrame

Builds the reform simulation data.

Calls build_data_simulation and iterate_reform_simulations to build the simulation data.

Parameters:

Name	Type	Description	Default
scenarios	dict	The scenarios to simulate. Each scenario is a dictionary of reform parameters.	required
year	int	The year for the simulation.	required
taux_bascule_vm	float	The rate of the value-added tax (VM) switch. If provided, new variables are calculated for the VM switch.	None
data	DataFrame	The data to use for building the simulation data. If None, the data is read from the path.	None
path	PathLike	The path to the data file. Used if data is None.	None

Returns:

Type	Description
DataFrame	The simulation data.

Source code in bozio_wasmer_simulations/simulation/empirical/base.py

def build(
    self,
    scenarios: dict,
    year: int,
    taux_bascule_vm: Optional[Union[float, None]] = None,
    data: Optional[Union[pd.DataFrame, None]] = None,
    path: Optional[Union[os.PathLike, None]] = None,
) -> pd.DataFrame:
    """
    Builds the reform simulation data.

    Calls build_data_simulation and iterate_reform_simulations to build the simulation data.

    Args:
        scenarios (dict):
            The scenarios to simulate. Each scenario is a dictionary of reform parameters.
        year (int):
            The year for the simulation.
        taux_bascule_vm (float, optional):
            The rate of the value-added tax (VM) switch. If provided, new variables are calculated for the VM switch.
        data (pd.DataFrame, optional):
            The data to use for building the simulation data. If None, the data is read from the path.
        path (os.PathLike, optional):
            The path to the data file. Used if data is None.

    Returns:
        (pd.DataFrame): The simulation data.
    """
    # Ajout du jeu de données de simulations
    self.build_data_simulation(data=data, path=path)
    # Itération des simulations de réformes
    self.iterate_reform_simulations(
        scenarios=scenarios, year=year, taux_bascule_vm=taux_bascule_vm
    )

    return self.data_simulation

build_data_simulation

build_data_simulation(data: Optional[Union[DataFrame, None]] = None, path: Optional[Union[PathLike, None]] = None) -> None

Builds the simulation data.

If data is provided, it uses _build_data_simulation_from_dataframe to build the simulation data. If not, it reads the data from the provided path.

Parameters:

Name	Type	Description	Default
data	DataFrame	The data to use for building the simulation data. If None, the data is read from the path.	None
path	PathLike	The path to the data file. Used if data is None.	None

Returns:

Type	Description
None	None

Source code in bozio_wasmer_simulations/simulation/empirical/base.py

def build_data_simulation(
    self,
    data: Optional[Union[pd.DataFrame, None]] = None,
    path: Optional[Union[os.PathLike, None]] = None,
) -> None:
    """
    Builds the simulation data.

    If data is provided, it uses _build_data_simulation_from_dataframe to build the simulation data.
    If not, it reads the data from the provided path.

    Args:
        data (pd.DataFrame, optional):
            The data to use for building the simulation data. If None, the data is read from the path.
        path (os.PathLike, optional):
            The path to the data file. Used if data is None.

    Returns:
        None
    """
    if data is not None:
        self._build_data_simulation_from_dataframe(data=data)
    else:
        self.data_simulation = pd.read_csv(path)

iterate_reform_simulations

iterate_reform_simulations(scenarios: dict, year: int, taux_bascule_vm: Optional[Union[float, None]] = None) -> None

Iterates over multiple reform simulations.

Simulates each reform in the provided scenarios. The results are added to the simulation data.

Parameters:

Name	Type	Description	Default
scenarios	dict	The scenarios to simulate. Each scenario is a dictionary of reform parameters.	required
year	int	The year for the simulation.	required
taux_bascule_vm	float	The rate of the value-added tax (VM) switch. If provided, new variables are calculated for the VM switch.	None

Returns:

Type	Description
None	None

Source code in bozio_wasmer_simulations/simulation/empirical/base.py

def iterate_reform_simulations(
    self,
    scenarios: dict,
    year: int,
    taux_bascule_vm: Optional[Union[float, None]] = None,
) -> None:
    """
    Iterates over multiple reform simulations.

    Simulates each reform in the provided scenarios. The results are added to the simulation data.

    Args:
        scenarios (dict):
            The scenarios to simulate. Each scenario is a dictionary of reform parameters.
        year (int):
            The year for the simulation.
        taux_bascule_vm (float, optional):
            The rate of the value-added tax (VM) switch. If provided, new variables are calculated for the VM switch.

    Returns:
        None
    """
    # Itération sur les scénarii référencés dans le jeu de données de paramètres
    for scenario in tqdm(scenarios.keys()):
        # Itération des réformes
        self.simulate_reform(
            name=scenario,
            reform_params=scenarios[scenario],
            year=year,
            taux_bascule_vm=taux_bascule_vm,
        )

iterate_simulation

iterate_simulation(data: DataFrame, tax_benefit_system: TaxBenefitSystem, year: int, list_var_simul: List[str], list_var_exclude: Optional[List[str]] = [], inplace: Optional[bool] = True) -> DataFrame

Iterates a simulation.

Parameters:

Name	Type	Description	Default
data	DataFrame	The data to simulate.	required
tax_benefit_system	FranceTaxBenefitSystem	The tax benefit system.	required
year	int	The year of the simulation.	required
list_var_simul	List[str]	The list of variables to simulate.	required
list_var_exclude	Optional[List[str]]	The list of variables to exclude. Defaults to [].	[]
inplace	Optional[bool]	Whether to perform the simulation in place. Defaults to True.	True

Returns:

Type	Description
DataFrame	The simulated data.

Source code in bozio_wasmer_simulations/simulation/empirical/base.py

def iterate_simulation(
    self,
    data: pd.DataFrame,
    tax_benefit_system: TaxBenefitSystem,
    year: int,
    list_var_simul: List[str],
    list_var_exclude: Optional[List[str]] = [],
    inplace: Optional[bool] = True,
) -> pd.DataFrame:
    """
    Iterates a simulation.

    Args:
        data (pd.DataFrame): The data to simulate.
        tax_benefit_system (FranceTaxBenefitSystem): The tax benefit system.
        year (int): The year of the simulation.
        list_var_simul (List[str]): The list of variables to simulate.
        list_var_exclude (Optional[List[str]], optional): The list of variables to exclude. Defaults to [].
        inplace (Optional[bool], optional): Whether to perform the simulation in place. Defaults to True.

    Returns:
        (pd.DataFrame): The simulated data.
    """
    # Disjonction de cas suivant la nécessité de réaliser une copie indépendante du jeu de données
    if inplace:
        data_res = data
    else:
        data_res = data.copy()

    # Initialisation des paramètres de la simulation
    simulation = SimulationBuilder().build_default_simulation(
        tax_benefit_system, len(data_res)
    )
    # Ajout de l'ensemble des données
    # /!\ On ajout 'smic_proratisé' aux variables à exclure de l'imputation pour contourner l'écueil de la mauvaise transition entre valeurs mensuelles et annuelles # + ['smic_proratise']
    # Finalement retiré car les rému restent divisées par 12 et ne sont pas intersectées avec la durée du contrat
    # Il s'agit sans doute d'un point à améliorer dans le package
    for caracteristic in np.setdiff1d(data_res.columns, list_var_exclude):
        try:  # if not (caracteristic in ['id', 'siren']) :
            simulation.set_input(
                caracteristic, year, data_res[caracteristic].to_numpy()
            )
            # logging
            self.logger.info(
                f"Successfully initialized {caracteristic} in the french tax benefit system"
            )
        except Exception as e:
            # Logging
            self.logger.warning(
                f"Cannot initialize {caracteristic} in the french tax benefit system : {e}"
            )
            pass
    # Ajout des cotisations et des allègements généraux
    for var_simul in tqdm(list_var_simul):
        data_res[var_simul] = simulation.calculate_add(var_simul, year)
        # Logging
        self.logger.info(f"Successfully simulated {var_simul} for period {year}")

    return data_res

simulate_reform

simulate_reform(name: str, reform_params: dict, year: int, taux_bascule_vm: Optional[Union[float, None]] = None) -> None

Simulates a reform.

Applies the reform to the simulation data and calculates new variables. The new variables are added to the simulation data.

Parameters:

Name	Type	Description	Default
name	str	The name of the reform.	required
reform_params	dict	The parameters of the reform.	required
year	int	The year for the simulation.	required
taux_bascule_vm	float	The rate of the "versement mobilité" (VM) switch. If provided, new variables are calculated for the VM switch.	None

Returns:

Type	Description
None	None

Source code in bozio_wasmer_simulations/simulation/empirical/base.py

def simulate_reform(
    self,
    name: str,
    reform_params: dict,
    year: int,
    taux_bascule_vm: Optional[Union[float, None]] = None,
) -> None:
    """
    Simulates a reform.

    Applies the reform to the simulation data and calculates new variables. The new variables are added to the simulation data.

    Args:
        name (str):
            The name of the reform.
        reform_params (dict):
            The parameters of the reform.
        year (int):
            The year for the simulation.
        taux_bascule_vm (float, optional):
            The rate of the "versement mobilité" (VM) switch. If provided, new variables are calculated for the VM switch.

    Returns:
        None
    """
    # Simulation du SMIC proratisé
    if "smic_proratise" not in self.data_simulation.columns:
        self.data_simulation = self.simulate_smic_proratise(
            data=self.data_simulation, year=year, list_var_exclude=[], inplace=True
        )

    # Initialisation des paramètres du système sociofiscal
    tax_benefit_system = FranceTaxBenefitSystem()

    # Application de la réforme
    reformed_tax_benefit_system = create_and_apply_structural_reform_ag(
        tax_benefit_system=tax_benefit_system, dict_params=reform_params
    )

    # Extraction du type de la réforme
    reform_type = reform_params["TYPE"]
    # Itération de la simulation
    self.data_simulation = self.iterate_simulation(
        data=self.data_simulation,
        tax_benefit_system=reformed_tax_benefit_system,
        year=year,
        list_var_simul=[f"new_allegement_{reform_type}"],
        list_var_exclude=params["REFORM"]["VAR_EXCLUDE"],
        inplace=True,
    )

    # Renomination de la variable simulée pour correspondre au nom du scénario
    self.data_simulation.rename(
        {f"new_allegement_{reform_type}": f"new_allegement_{name.lower()}"},
        axis=1,
        inplace=True,
    )
    # Somme des exonérations et allègements
    self.data_simulation[f"exonerations_et_allegements_{name.lower()}"] = (
        self.data_simulation[
            ["exonerations", f"new_allegement_{name.lower()}"]
        ].sum(axis=1)
    )
    # Calcul du salaire brut avec la réforme
    self.data_simulation[f"salaire_super_brut_{name.lower()}"] = (
        self.data_simulation["salaire_super_brut_hors_allegements"]
        - self.data_simulation[f"exonerations_et_allegements_{name.lower()}"]
        + self.data_simulation["prime_partage_valeur_exoneree"]
    )
    # Construction des variables de variation du coût du travail
    self.data_simulation = build_data_evol_ct(
        data_source=self.data_simulation,
        col_new_ct=f"salaire_super_brut_{name.lower()}",
        col_ct="salaire_super_brut",
        to_concat=True,
    )
    # Logging
    self.logger.info(
        f"Successfully built variables related to labor costs variations inducted by the reform"
    )

    # Construction d'un nouveau coût du travail associé à la bascule d'une fraction du versement transport
    if taux_bascule_vm is not None:
        self.data_simulation[f"salaire_super_brut_vm_{name.lower()}"] = (
            self.data_simulation["salaire_super_brut_hors_allegements"]
            + self.data_simulation["versement_transport"] * (1 - taux_bascule_vm)
            - self.data_simulation[f"exonerations_et_allegements_{name.lower()}"]
            + self.data_simulation["prime_partage_valeur_exoneree"]
        )
        # Construction des variables de variation du coût du travail
        self.data_simulation = build_data_evol_ct(
            data_source=self.data_simulation,
            col_new_ct=f"salaire_super_brut_vm_{name.lower()}",
            col_ct="salaire_super_brut",
            to_concat=True,
        )
        # Logging
        self.logger.info(
            f"Successfully built variables related to labor costs variations inducted by the reform and the vm"
        )

simulate_smic_proratise

simulate_smic_proratise(data: DataFrame, year: int, list_var_exclude: Optional[List[str]] = [], inplace: Optional[bool] = True) -> DataFrame

Simulates the prorated minimum wage.

Parameters:

Name	Type	Description	Default
data	DataFrame	The data to simulate.	required
year	int	The year of the simulation.	required
list_var_exclude	Optional[List[str]]	The list of variables to exclude. Defaults to [].	[]
inplace	Optional[bool]	Whether to perform the simulation in place. Defaults to True.	True

Returns:

Type	Description
DataFrame	The simulated data.

Source code in bozio_wasmer_simulations/simulation/empirical/base.py

def simulate_smic_proratise(
    self,
    data: pd.DataFrame,
    year: int,
    list_var_exclude: Optional[List[str]] = [],
    inplace: Optional[bool] = True,
) -> pd.DataFrame:
    """
    Simulates the prorated minimum wage.

    Args:
        data (pd.DataFrame): The data to simulate.
        year (int): The year of the simulation.
        list_var_exclude (Optional[List[str]], optional): The list of variables to exclude. Defaults to [].
        inplace (Optional[bool], optional): Whether to perform the simulation in place. Defaults to True.

    Returns:
        (pd.DataFrame): The simulated data.
    """
    # Initialisation des paramètres du système sociofiscal français
    tax_benefit_system = FranceTaxBenefitSystem()

    # Simulation du SMIC proratisé pour l'année des données
    data = self.iterate_simulation(
        data=data,
        tax_benefit_system=tax_benefit_system,
        year=year,
        list_var_simul=["smic_proratise"],
        list_var_exclude=list_var_exclude,
        inplace=inplace,
    )

    return data