PPVReintegrationSimulator

A class for simulating the reintegration of the PPV in the base for exemptions.

This class inherits from TheoreticalSimulator and provides methods for initializing a case with the PPV exempted or reintegrated, simulating variables of interest, simulating a reform, simulating multiple reforms, calculating the implicit tax rate, and building the dataset.

Attributes:

Name	Type	Description
logger	Logger	A logger for logging messages.

Source code in bozio_wasmer_simulations/simulation/theoretical/ppv.py

class PPVReintegrationSimulator(TheoreticalSimulator):
    """
    A class for simulating the reintegration of the PPV in the base for exemptions.

    This class inherits from TheoreticalSimulator and provides methods for initializing a case with the PPV exempted or reintegrated,
    simulating variables of interest, simulating a reform, simulating multiple reforms, calculating the implicit tax rate,
    and building the dataset.

    Attributes:
        logger (logging.Logger):
            A logger for logging messages.

    """

    # Initialisation
    def __init__(
        self,
        log_filename: os.PathLike = os.path.join(
            FILE_PATH.parents[3], "logs/ppv_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/ppv_simulation.log').

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

    # Fonction auxiliaire d'extraction des noms de scénarios
    def scenarios_names(self, scenarios: Union[str, List[str], dict]) -> List[str]:
        """
        Extracts the names of the scenarios.

        Args:
            scenarios (Union[str, List[str], dict]):
                The scenarios to extract the names from.

        Returns:
            (List[str]): The names of the scenarios.

        Raises:
            ValueError: If the type of scenarios is not 'dict', 'list', or 'str'.
        """
        # Extraction des noms des scenarios
        if isinstance(scenarios, str):
            scenarios_names = [scenarios]
        elif isinstance(scenarios, dict):
            scenarios_names = [name.lower() for name in scenarios.keys()]
        elif isinstance(scenarios, list):
            scenarios_names = [name.lower() for name in scenarios]
        else:
            raise ValueError(
                f"Invalid type for scenarios : {type(scenarios)}. Should be in 'dict', 'list' or 'str'"
            )

        return scenarios_names

    # Méthode auxiliaire de construction du cas de base dans le cadre exonéré
    def init_case_ppv_exoneree(
        self,
        year: int,
        simulation_step_smic: float,
        simulation_max_smic: float,
        ppv: float,
    ) -> None:
        """
        Initializes a case with the PPV exempted.

        Args:
            year (int):
                The year for which the simulation is being performed.
            simulation_step_smic (float):
                The step size for the simulation, as a multiple of the SMIC value.
            simulation_max_smic (float):
                The maximum value for the simulation, as a multiple of the SMIC value.
            ppv (float):
                The amount of the PPV.

        Returns:
            None
        """
        # Initialisation du cas de base
        self.init_base_case(
            year=year,
            simulation_step_smic=simulation_step_smic,
            simulation_max_smic=simulation_max_smic,
        )
        # Modification des paramètres de PPV
        self.base_case["individus"]["individu_1"]["prime_partage_valeur_exoneree"][
            2022
        ] = ppv
        self.base_case["individus"]["individu_1"]["prime_partage_valeur_non_exoneree"][
            2022
        ] = 0

    # Méthode auxiliaire de construction du cas de base dans le cadre réintégré
    # /!\
    # 'axes' = [[
    #         {'count' : simulation_count, 'name' : 'salaire_de_base', 'min' : value_smic, 'max' : simulation_max, 'period' : 2022},
    #         {'count' : simulation_count, 'name' : 'assiette_allegement', 'min' : value_smic+montant_ppv, 'max' : simulation_max+montant_ppv, 'period' : 2022}
    #     ]]
    # N'est peut-être pas équivalent pour l'année 2022 à mettre 'prime_partage_valeur_non_exoneree' = ppv car elle n'est intégrée aux primes non exonérées qu'à compter du 1er juillet
    def init_case_ppv_reintegree(
        self,
        year: int,
        simulation_step_smic: float,
        simulation_max_smic: float,
        ppv: float,
    ) -> None:
        """
        Initializes a case with the PPV reintegrated.

        Args:
            year (int):
                The year for which the simulation is being performed.
            simulation_step_smic (float):
                The step size for the simulation, as a multiple of the SMIC value.
            simulation_max_smic (float):
                The maximum value for the simulation, as a multiple of the SMIC value.
            ppv (float):
                The amount of the PPV.
        """
        # Initialisation du cas de base
        self.init_base_case(
            year=year,
            simulation_step_smic=simulation_step_smic,
            simulation_max_smic=simulation_max_smic,
        )
        # Modification des paramètres de PPV
        self.base_case["individus"]["individu_1"]["prime_partage_valeur_exoneree"][
            2022
        ] = 0
        self.base_case["individus"]["individu_1"]["prime_partage_valeur_non_exoneree"][
            2022
        ] = ppv

    # Fonction auxiliaire de preprocessing du salaire de base
    def _preprocess_salaire_de_base(
        self, data: pd.DataFrame, year: int
    ) -> pd.DataFrame:
        """
        Preprocesses the gross salary.

        Expresses the gross salary as a proportion of the SMIC.

        Args:
            data (pd.DataFrame):
                The input data.
            year (int):
                The year for which the data is being processed.

        Returns:
            (pd.DataFrame): The preprocessed data.
        """
        # Expression du salaire en proportion du SMIC
        data["salaire_de_base_prop_smic"] = data["salaire_de_base"] / self.value_smic(
            year=year
        )

        return data

    # Fonction auxiliaire de preprocessing du salaire super brut
    def _preprocess_salaire_super_brut(
        self,
        data: pd.DataFrame,
        scenarios: Union[str, List[str], dict],
        simulation_case: str,
    ) -> pd.DataFrame:
        """
        Preprocesses the gross salary.

        Calculates the gross salary for each scenario and drops the unnecessary columns.

        Args:
            data (pd.DataFrame):
                The input data.
            scenarios (Union[str, List[str], dict]):
                The scenarios to simulate.
            simulation_case (str):
                The simulation case ('exoneree' or 'reintegree').

        Returns:
            (pd.DataFrame): The preprocessed data.
        """
        # Parcours des scénarios
        for scenario in self.scenarios_names(scenarios=scenarios):
            # Création du salaire super_brut
            data[f"{simulation_case}_salaire_super_brut_{scenario}"] = (
                data[f"{simulation_case}_salaire_super_brut"]
                + data[f"{simulation_case}_allegement_general"]
                - data[f"{simulation_case}_new_allegement_{scenario}"]
            )
            # Suppression du nouvel allègement
            data.drop(
                f"{simulation_case}_new_allegement_{scenario}", axis=1, inplace=True
            )
        # Suppression de l'allègement général
        data.drop(f"{simulation_case}_allegement_general", axis=1, inplace=True)

        return data

    # Fonction auxiliaire de simulation des variables d'intérêt
    def core_simulation(
        self,
        year: int,
        simulation_step_smic: float,
        simulation_max_smic: float,
        simulation_case: str,
        ppv: float,
    ) -> pd.DataFrame:
        """
        Simulates the variables of interest.

        Initializes the simulation case, initializes the tax-benefit system,
        simulates the variables, postprocesses the simulated variables,
        and preprocesses the gross salary.

        Args:
            year (int):
                The year for which the simulation is being performed.
            simulation_step_smic (float):
                The step size for the simulation, as a multiple of the SMIC value.
            simulation_max_smic (float):
                The maximum value for the simulation, as a multiple of the SMIC value.
            simulation_case (str):
                The simulation case ('exoneree' or 'reintegree').
            ppv (float):
                The amount of the PPV.

        Returns:
            (pd.DataFrame): The simulated data.

        Raises:
            ValueError: If the simulation case is not 'exoneree' or 'reintegree'.
        """
        # Initialisation du cas de simulation
        if simulation_case == "exoneree":
            self.init_case_ppv_exoneree(
                year=year,
                simulation_step_smic=simulation_step_smic,
                simulation_max_smic=simulation_max_smic,
                ppv=ppv,
            )
        elif simulation_case == "reintegree":
            self.init_case_ppv_reintegree(
                year=year,
                simulation_step_smic=simulation_step_smic,
                simulation_max_smic=simulation_max_smic,
                ppv=ppv,
            )
        else:
            raise ValueError(
                f"Invalid value for 'simulation_case' : {simulation_case}. Should be in ['exoneree', 'reintegree']"
            )

        # Initialisation du système socio-fiscal
        tax_benefit_system = FranceTaxBenefitSystem()
        # Extraction des variables à simuler
        list_var_simul = params["VARIABLES_PPV"]
        # Simulation
        data_simul = self.base_case_simulation(
            tax_benefit_system=tax_benefit_system,
            year=year,
            list_var_simul=list_var_simul,
        )
        # Post-processing des variables simulées
        # Ajout de la prime partage de la valeur exonérée
        data_simul["prime_partage_valeur_exoneree"] = self.base_case["individus"][
            "individu_1"
        ]["prime_partage_valeur_exoneree"][year]
        # Postprocessing
        data_simul = preprocess_simulated_variables(data=data_simul)

        # Retraitement du salaire de base
        data_simul = self._preprocess_salaire_de_base(data=data_simul, year=year)
        # Logging
        self.logger.info("Successfully preprocessed simulated variables")

        return data_simul[
            [
                "salaire_de_base",
                "salaire_de_base_prop_smic",
                "salaire_super_brut",
                "allegement_general",
                "salaire_net",
            ]
        ].add_prefix(prefix=f"{simulation_case}_")

    # Fonction auxiliaire de simulation théorique des réformes
    def simulate_reform(
        self,
        name: str,
        reform_params: dict,
        year: int,
        simulation_step_smic: float,
        simulation_max_smic: float,
        simulation_case: str,
        ppv: float,
    ) -> pd.DataFrame:
        """
        Simulates a reform.

        Initializes the simulation case, initializes the tax-benefit system,
        applies the reform, simulates the variables, and preprocesses the gross salary.

        Args:
            name (str):
                The name of the reform.
            reform_params (dict):
                The parameters of the reform.
            year (int):
                The year for which the simulation is being performed.
            simulation_step_smic (float):
                The step size for the simulation, as a multiple of the SMIC value.
            simulation_max_smic (float):
                The maximum value for the simulation, as a multiple of the SMIC value.
            simulation_case (str):
                The simulation case ('exoneree' or 'reintegree').
            ppv (float):
                The amount of the PPV.

        Returns:
            (pd.DataFrame): The simulated data.

        Raises:
            ValueError: If the simulation case is not 'exoneree' or 'reintegree'.
        """
        # Initialisation du cas de simulation
        if simulation_case == "exoneree":
            self.init_case_ppv_exoneree(
                year=year,
                simulation_step_smic=simulation_step_smic,
                simulation_max_smic=simulation_max_smic,
                ppv=ppv,
            )
        elif simulation_case == "reintegree":
            self.init_case_ppv_reintegree(
                year=year,
                simulation_step_smic=simulation_step_smic,
                simulation_max_smic=simulation_max_smic,
                ppv=ppv,
            )
        else:
            raise ValueError(
                f"Invalid value for 'simulation_case' : {simulation_case}. Should be in ['exoneree', 'reintegree']"
            )

        # 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
        )

        # Logging
        self.logger.info("Successfully updated the tax-benefit system")

        # Extraction du type de la réforme
        reform_type = reform_params["TYPE"]

        # Itération de la simulation
        data_simul = self.base_case_simulation(
            tax_benefit_system=reformed_tax_benefit_system,
            year=year,
            list_var_simul=["salaire_de_base", f"new_allegement_{reform_type}"],
        )

        # Renomination de la variable simulée pour correspondre au nom du scénario
        data_simul.rename(
            {f"new_allegement_{reform_type}": f"new_allegement_{name.lower()}"},
            axis=1,
            inplace=True,
        )

        # Retraitement du salaire de base
        data_simul = self._preprocess_salaire_de_base(data=data_simul, year=year)

        return data_simul.add_prefix(prefix=f"{simulation_case}_")

    # Fonction auxiliaire de simulation de plusieurs réformes théoriques
    def iterate_reform_simulations(
        self,
        scenarios: Union[str, List[str], dict],
        year: int,
        simulation_step_smic: float,
        simulation_max_smic: float,
        simulation_case: str,
        ppv: float,
    ) -> pd.DataFrame:
        """
        Simulates multiple reforms.

        Iterates over the scenarios and simulates each reform.
        Concatenates the simulated data for all reforms.

        Args:
            scenarios (Union[str, List[str], dict]):
                The scenarios to simulate.
            year (int):
                The year for which the simulation is being performed.
            simulation_step_smic (float):
                The step size for the simulation, as a multiple of the SMIC value.
            simulation_max_smic (float):
                The maximum value for the simulation, as a multiple of the SMIC value.
            simulation_case (str):
                The simulation case ('exoneree' or 'reintegree').
            ppv (float):
                The amount of the PPV.

        Returns:
            (pd.DataFrame): The simulated data for all reforms.
        """
        # Initialisation de la liste résultat
        list_data_simul = []
        # Itération sur les scénarii référencés dans le jeu de données de paramètres
        for i, scenario in tqdm(enumerate(self.scenarios_names(scenarios=scenarios))):
            # Itération des réformes
            data_simul = self.simulate_reform(
                name=scenario,
                reform_params=scenarios[scenario.upper()],
                year=year,
                simulation_step_smic=simulation_step_smic,
                simulation_max_smic=simulation_max_smic,
                simulation_case=simulation_case,
                ppv=ppv,
            )
            # Ajout à la liste résultat
            if i > 0:
                list_data_simul.append(
                    data_simul.drop(
                        [
                            f"{simulation_case}_salaire_de_base",
                            f"{simulation_case}_salaire_de_base_prop_smic",
                        ],
                        axis=1,
                    )
                )
            else:
                list_data_simul.append(data_simul)
        # Concaténation
        data_simul = pd.concat(list_data_simul, axis=1, join="outer")

        return data_simul

    # Fonction auxiliaire de calcul du taux de cotisation implicite
    def calculate_taux_cotisation_implicite(
        self,
        data: pd.DataFrame,
        scenarios: Union[str, List[str], dict],
        simulation_case: str,
    ) -> pd.DataFrame:
        """
        Calculates the implicit tax rate.

        Calculates the implicit tax rate for each scenario and drops the unnecessary columns.

        Args:
            data (pd.DataFrame):
                The input data.
            scenarios (Union[str, List[str], dict]):
                The scenarios to simulate.
            simulation_case (str):
                The simulation case ('exoneree' or 'reintegree').

        Returns:
            (pd.DataFrame): The data with the implicit tax rate calculated.
        """
        # Parcours des scénarios
        for scenario in self.scenarios_names(scenarios=scenarios):
            # Calcul du taux de cotisations implicite
            data[f"{simulation_case}_taux_cotisation_implicite_{scenario}"] = (
                data[f"{simulation_case}_salaire_super_brut_{scenario}"]
                - data[f"{simulation_case}_salaire_net"]
            ) / data[f"{simulation_case}_salaire_super_brut_{scenario}"]
            # Suppression du salaire super_brut
            data.drop(
                f"{simulation_case}_salaire_super_brut_{scenario}", axis=1, inplace=True
            )

        return data

    # Fonction de construction du jeu de données
    def build(
        self,
        scenarios: dict,
        year: int,
        simulation_step_smic: float,
        simulation_max_smic: float,
        simulation_case: str,
        ppv: float,
    ) -> pd.DataFrame:
        """
        Builds the dataset.

        Simulates the core variables and the reform variables, preprocesses the gross salary,
        calculates the implicit tax rate, and returns the dataset.

        Args:
            scenarios (dict):
                The scenarios to simulate.
            year (int):
                The year for which the simulation is being performed.
            simulation_step_smic (float):
                The step size for the simulation, as a multiple of the SMIC value.
            simulation_max_smic (float):
                The maximum value for the simulation, as a multiple of the SMIC value.
            simulation_case (str):
                The simulation case ('exoneree' or 'reintegree').
            ppv (float):
                The amount of the PPV.

        Returns:
            (pd.DataFrame): The dataset.
        """
        # Construction du scénario de base et concaténration des simulations des scénarios de réforme
        data_simul = pd.concat(
            [
                self.core_simulation(
                    year=year,
                    simulation_step_smic=simulation_step_smic,
                    simulation_max_smic=simulation_max_smic,
                    simulation_case=simulation_case,
                    ppv=ppv,
                ),
                self.iterate_reform_simulations(
                    scenarios=scenarios,
                    year=year,
                    simulation_step_smic=simulation_step_smic,
                    simulation_max_smic=simulation_max_smic,
                    simulation_case=simulation_case,
                    ppv=ppv,
                ).drop(
                    [
                        f"{simulation_case}_salaire_de_base",
                        f"{simulation_case}_salaire_de_base_prop_smic",
                    ],
                    axis=1,
                ),
            ],
            axis=1,
            join="outer",
        )
        # Construction des salaires super_bruts associés à chaque scénario
        data_simul = self._preprocess_salaire_super_brut(
            data=data_simul, scenarios=scenarios, simulation_case=simulation_case
        )

        # Calcul du taux de cotisations implicite
        # Pour le cas de base
        data_simul[f"{simulation_case}_taux_cotisation_implicite"] = (
            data_simul[f"{simulation_case}_salaire_super_brut"]
            - data_simul[f"{simulation_case}_salaire_net"]
        ) / data_simul[f"{simulation_case}_salaire_super_brut"]
        # Suppression du salaire super brut
        data_simul.drop(f"{simulation_case}_salaire_super_brut", axis=1, inplace=True)
        # Pour les réformes
        data_simul = self.calculate_taux_cotisation_implicite(
            data=data_simul, scenarios=scenarios, simulation_case=simulation_case
        )
        # Suppression du salaire net
        data_simul.drop(f"{simulation_case}_salaire_net", axis=1, inplace=True)

        return data_simul

base_case_simulation

base_case_simulation(tax_benefit_system: TaxBenefitSystem, year: int, list_var_simul: List[str]) -> DataFrame

Performs a simulation on the base case.

Parameters:

Name	Type	Description	Default
tax_benefit_system	TaxBenefitSystem	The tax-benefit system to use for the simulation.	required
year	int	The year for which the simulation is performed.	required
list_var_simul	List[str]	A list of variables to simulate.	required

Returns:

Type	Description
DataFrame	A dataframe containing the results of the simulation.

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

def base_case_simulation(
    self, tax_benefit_system: TaxBenefitSystem, year: int, list_var_simul: List[str]
) -> pd.DataFrame:
    """
    Performs a simulation on the base case.

    Args:
        tax_benefit_system (TaxBenefitSystem):
            The tax-benefit system to use for the simulation.
        year (int):
            The year for which the simulation is performed.
        list_var_simul (List[str]):
            A list of variables to simulate.

    Returns:
        (pd.DataFrame): A dataframe containing the results of the simulation.
    """
    # Initialisation des paramètres de la simulation
    simulation_builder = SimulationBuilder()
    simulation = simulation_builder.build_from_entities(
        tax_benefit_system, self.base_case
    )
    # Initialisation du dictionnaire résultat
    dict_simul = {}
    # Itération sur la liste des variables à simuler
    for variable in list_var_simul:
        dict_simul[variable] = simulation.calculate_add(variable, year)
        # Logging
        self.logger.info(f"Successfully simulated {variable} for period {year}")
    # Conversion en dataFrame
    data_simul = pd.DataFrame(dict_simul)

    return data_simul

build

build(scenarios: dict, year: int, simulation_step_smic: float, simulation_max_smic: float, simulation_case: str, ppv: float) -> DataFrame

Builds the dataset.

Simulates the core variables and the reform variables, preprocesses the gross salary, calculates the implicit tax rate, and returns the dataset.

Parameters:

Name	Type	Description	Default
scenarios	dict	The scenarios to simulate.	required
year	int	The year for which the simulation is being performed.	required
simulation_step_smic	float	The step size for the simulation, as a multiple of the SMIC value.	required
simulation_max_smic	float	The maximum value for the simulation, as a multiple of the SMIC value.	required
simulation_case	str	The simulation case ('exoneree' or 'reintegree').	required
ppv	float	The amount of the PPV.	required

Returns:

Type	Description
DataFrame	The dataset.

Source code in bozio_wasmer_simulations/simulation/theoretical/ppv.py

def build(
    self,
    scenarios: dict,
    year: int,
    simulation_step_smic: float,
    simulation_max_smic: float,
    simulation_case: str,
    ppv: float,
) -> pd.DataFrame:
    """
    Builds the dataset.

    Simulates the core variables and the reform variables, preprocesses the gross salary,
    calculates the implicit tax rate, and returns the dataset.

    Args:
        scenarios (dict):
            The scenarios to simulate.
        year (int):
            The year for which the simulation is being performed.
        simulation_step_smic (float):
            The step size for the simulation, as a multiple of the SMIC value.
        simulation_max_smic (float):
            The maximum value for the simulation, as a multiple of the SMIC value.
        simulation_case (str):
            The simulation case ('exoneree' or 'reintegree').
        ppv (float):
            The amount of the PPV.

    Returns:
        (pd.DataFrame): The dataset.
    """
    # Construction du scénario de base et concaténration des simulations des scénarios de réforme
    data_simul = pd.concat(
        [
            self.core_simulation(
                year=year,
                simulation_step_smic=simulation_step_smic,
                simulation_max_smic=simulation_max_smic,
                simulation_case=simulation_case,
                ppv=ppv,
            ),
            self.iterate_reform_simulations(
                scenarios=scenarios,
                year=year,
                simulation_step_smic=simulation_step_smic,
                simulation_max_smic=simulation_max_smic,
                simulation_case=simulation_case,
                ppv=ppv,
            ).drop(
                [
                    f"{simulation_case}_salaire_de_base",
                    f"{simulation_case}_salaire_de_base_prop_smic",
                ],
                axis=1,
            ),
        ],
        axis=1,
        join="outer",
    )
    # Construction des salaires super_bruts associés à chaque scénario
    data_simul = self._preprocess_salaire_super_brut(
        data=data_simul, scenarios=scenarios, simulation_case=simulation_case
    )

    # Calcul du taux de cotisations implicite
    # Pour le cas de base
    data_simul[f"{simulation_case}_taux_cotisation_implicite"] = (
        data_simul[f"{simulation_case}_salaire_super_brut"]
        - data_simul[f"{simulation_case}_salaire_net"]
    ) / data_simul[f"{simulation_case}_salaire_super_brut"]
    # Suppression du salaire super brut
    data_simul.drop(f"{simulation_case}_salaire_super_brut", axis=1, inplace=True)
    # Pour les réformes
    data_simul = self.calculate_taux_cotisation_implicite(
        data=data_simul, scenarios=scenarios, simulation_case=simulation_case
    )
    # Suppression du salaire net
    data_simul.drop(f"{simulation_case}_salaire_net", axis=1, inplace=True)

    return data_simul

calculate_taux_cotisation_implicite

calculate_taux_cotisation_implicite(data: DataFrame, scenarios: Union[str, List[str], dict], simulation_case: str) -> DataFrame

Calculates the implicit tax rate.

Calculates the implicit tax rate for each scenario and drops the unnecessary columns.

Parameters:

Name	Type	Description	Default
data	DataFrame	The input data.	required
scenarios	Union[str, List[str], dict]	The scenarios to simulate.	required
simulation_case	str	The simulation case ('exoneree' or 'reintegree').	required

Returns:

Type	Description
DataFrame	The data with the implicit tax rate calculated.

Source code in bozio_wasmer_simulations/simulation/theoretical/ppv.py

def calculate_taux_cotisation_implicite(
    self,
    data: pd.DataFrame,
    scenarios: Union[str, List[str], dict],
    simulation_case: str,
) -> pd.DataFrame:
    """
    Calculates the implicit tax rate.

    Calculates the implicit tax rate for each scenario and drops the unnecessary columns.

    Args:
        data (pd.DataFrame):
            The input data.
        scenarios (Union[str, List[str], dict]):
            The scenarios to simulate.
        simulation_case (str):
            The simulation case ('exoneree' or 'reintegree').

    Returns:
        (pd.DataFrame): The data with the implicit tax rate calculated.
    """
    # Parcours des scénarios
    for scenario in self.scenarios_names(scenarios=scenarios):
        # Calcul du taux de cotisations implicite
        data[f"{simulation_case}_taux_cotisation_implicite_{scenario}"] = (
            data[f"{simulation_case}_salaire_super_brut_{scenario}"]
            - data[f"{simulation_case}_salaire_net"]
        ) / data[f"{simulation_case}_salaire_super_brut_{scenario}"]
        # Suppression du salaire super_brut
        data.drop(
            f"{simulation_case}_salaire_super_brut_{scenario}", axis=1, inplace=True
        )

    return data

core_simulation

core_simulation(year: int, simulation_step_smic: float, simulation_max_smic: float, simulation_case: str, ppv: float) -> DataFrame

Simulates the variables of interest.

Initializes the simulation case, initializes the tax-benefit system, simulates the variables, postprocesses the simulated variables, and preprocesses the gross salary.

Parameters:

Name	Type	Description	Default
year	int	The year for which the simulation is being performed.	required
simulation_step_smic	float	The step size for the simulation, as a multiple of the SMIC value.	required
simulation_max_smic	float	The maximum value for the simulation, as a multiple of the SMIC value.	required
simulation_case	str	The simulation case ('exoneree' or 'reintegree').	required
ppv	float	The amount of the PPV.	required

Returns:

Type	Description
DataFrame	The simulated data.

Raises:

Type	Description
ValueError	If the simulation case is not 'exoneree' or 'reintegree'.

Source code in bozio_wasmer_simulations/simulation/theoretical/ppv.py

def core_simulation(
    self,
    year: int,
    simulation_step_smic: float,
    simulation_max_smic: float,
    simulation_case: str,
    ppv: float,
) -> pd.DataFrame:
    """
    Simulates the variables of interest.

    Initializes the simulation case, initializes the tax-benefit system,
    simulates the variables, postprocesses the simulated variables,
    and preprocesses the gross salary.

    Args:
        year (int):
            The year for which the simulation is being performed.
        simulation_step_smic (float):
            The step size for the simulation, as a multiple of the SMIC value.
        simulation_max_smic (float):
            The maximum value for the simulation, as a multiple of the SMIC value.
        simulation_case (str):
            The simulation case ('exoneree' or 'reintegree').
        ppv (float):
            The amount of the PPV.

    Returns:
        (pd.DataFrame): The simulated data.

    Raises:
        ValueError: If the simulation case is not 'exoneree' or 'reintegree'.
    """
    # Initialisation du cas de simulation
    if simulation_case == "exoneree":
        self.init_case_ppv_exoneree(
            year=year,
            simulation_step_smic=simulation_step_smic,
            simulation_max_smic=simulation_max_smic,
            ppv=ppv,
        )
    elif simulation_case == "reintegree":
        self.init_case_ppv_reintegree(
            year=year,
            simulation_step_smic=simulation_step_smic,
            simulation_max_smic=simulation_max_smic,
            ppv=ppv,
        )
    else:
        raise ValueError(
            f"Invalid value for 'simulation_case' : {simulation_case}. Should be in ['exoneree', 'reintegree']"
        )

    # Initialisation du système socio-fiscal
    tax_benefit_system = FranceTaxBenefitSystem()
    # Extraction des variables à simuler
    list_var_simul = params["VARIABLES_PPV"]
    # Simulation
    data_simul = self.base_case_simulation(
        tax_benefit_system=tax_benefit_system,
        year=year,
        list_var_simul=list_var_simul,
    )
    # Post-processing des variables simulées
    # Ajout de la prime partage de la valeur exonérée
    data_simul["prime_partage_valeur_exoneree"] = self.base_case["individus"][
        "individu_1"
    ]["prime_partage_valeur_exoneree"][year]
    # Postprocessing
    data_simul = preprocess_simulated_variables(data=data_simul)

    # Retraitement du salaire de base
    data_simul = self._preprocess_salaire_de_base(data=data_simul, year=year)
    # Logging
    self.logger.info("Successfully preprocessed simulated variables")

    return data_simul[
        [
            "salaire_de_base",
            "salaire_de_base_prop_smic",
            "salaire_super_brut",
            "allegement_general",
            "salaire_net",
        ]
    ].add_prefix(prefix=f"{simulation_case}_")

init_base_case

init_base_case(year: int, simulation_step_smic: float, simulation_max_smic: float) -> None

Initializes a base case for simulation.

Parameters:

Name	Type	Description	Default
year	int	The year for which the simulation is performed.	required
simulation_step_smic	float	The step size for the simulation, as a multiple of the SMIC value.	required
simulation_max_smic	float	The maximum value for the simulation, as a multiple of the SMIC value.	required

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

def init_base_case(
    self, year: int, simulation_step_smic: float, simulation_max_smic: float
) -> None:
    """
    Initializes a base case for simulation.

    Args:
        year (int):
            The year for which the simulation is performed.
        simulation_step_smic (float):
            The step size for the simulation, as a multiple of the SMIC value.
        simulation_max_smic (float):
            The maximum value for the simulation, as a multiple of the SMIC value.
    """
    # Initialisation du système socio-fiscal contenant les valeurs de SMIC en paramètres
    tax_benefit_system = FranceTaxBenefitSystem()
    # Extraction de la valeur moyenne de SMIC sur l'année
    value_smic = self.value_smic(year=year)
    # Calcul de la valeur maximale de la simulation et de la valeur du pas
    simulation_max = simulation_max_smic * value_smic
    simulation_step = simulation_step_smic * value_smic
    # Calcul du nombre d'observations dans la simulation entre le min (1 SMIC) et le max avec le pas spécifié
    simulation_count = ceil((simulation_max - value_smic) / simulation_step) + 1
    # Définition des caractéristiques de l'individu
    self.base_case = {
        "individus": {
            "individu_1": {
                "effectif_entreprise": {year: 200},
                "depcom_entreprise": {year: "93001"},
                "contrat_de_travail_debut": {year: "2009-03-16"},
                "heures_remunerees_volume": {year: 1820},
                "prime_exceptionnelle_pouvoir_achat": {year: 0},
                "quotite_de_travail": {year: 12},
                "prime_partage_valeur_exoneree": {year: 0},
                "prime_partage_valeur_non_exoneree": {year: 0},
                "age": {year: 40},
                "secteur_activite_employeur": {
                    year: "non_agricole"
                },  # {year : TypesSecteurActivite.non_agricole},
                "exoneration_cotisations_employeur_tode_eligibilite": {year: False},
                "choix_exoneration_cotisations_employeur_agricole": {year: False},
                "travailleur_occasionnel_agricole": {year: False},
                "zone_restructuration_defense": {year: False},
                "zone_revitalisation_rurale": {year: False},
                "categorie_salarie": {
                    year: "prive_non_cadre"
                },  # {year : TypesCategorieSalarie.prive_non_cadre},
                "contrat_de_travail": {
                    year: "temps_plein"
                },  # {year : TypesContratDeTravail.temps_plein},
                "contrat_de_travail_fin": {year: "2099-12-31"},
                "contrat_de_travail_type": {
                    year: "cdi"
                },  # {year : TypesContrat.cdi},
                "salarie_regime_alsace_moselle": {year: False},
                #'salaire_de_base'
                "remuneration_apprenti": {year: 0},
                "apprentissage_contrat_debut": {year: "1970-01-01"},
                "apprenti": {year: False},
                "stage_duree_heures": {year: 0},
                "stage_gratification": {year: 0},
                "taux_versement_transport": {year: 0.032},
                "taux_accident_travail": {year: 0.0212},
            }
        },
        "menages": {
            "menage_1": {
                "personne_de_reference": ["individu_1"],
                "depcom": {year: "93001"},
            },
        },
        "familles": {"famille_1": {"parents": ["individu_1"]}},
        "foyers_fiscaux": {"foyer_fiscal_1": {"declarants": ["individu_1"]}},
        "axes": [
            [
                {
                    "count": simulation_count,
                    "name": "salaire_de_base",
                    "min": value_smic,
                    "max": simulation_max,
                    "period": year,
                }
            ]
        ],
    }

    # Logging
    self.logger.info("Successfully initialized a test case")

init_case_ppv_exoneree

init_case_ppv_exoneree(year: int, simulation_step_smic: float, simulation_max_smic: float, ppv: float) -> None

Initializes a case with the PPV exempted.

Parameters:

Name	Type	Description	Default
year	int	The year for which the simulation is being performed.	required
simulation_step_smic	float	The step size for the simulation, as a multiple of the SMIC value.	required
simulation_max_smic	float	The maximum value for the simulation, as a multiple of the SMIC value.	required
ppv	float	The amount of the PPV.	required

Returns:

Type	Description
None	None

Source code in bozio_wasmer_simulations/simulation/theoretical/ppv.py

def init_case_ppv_exoneree(
    self,
    year: int,
    simulation_step_smic: float,
    simulation_max_smic: float,
    ppv: float,
) -> None:
    """
    Initializes a case with the PPV exempted.

    Args:
        year (int):
            The year for which the simulation is being performed.
        simulation_step_smic (float):
            The step size for the simulation, as a multiple of the SMIC value.
        simulation_max_smic (float):
            The maximum value for the simulation, as a multiple of the SMIC value.
        ppv (float):
            The amount of the PPV.

    Returns:
        None
    """
    # Initialisation du cas de base
    self.init_base_case(
        year=year,
        simulation_step_smic=simulation_step_smic,
        simulation_max_smic=simulation_max_smic,
    )
    # Modification des paramètres de PPV
    self.base_case["individus"]["individu_1"]["prime_partage_valeur_exoneree"][
        2022
    ] = ppv
    self.base_case["individus"]["individu_1"]["prime_partage_valeur_non_exoneree"][
        2022
    ] = 0

init_case_ppv_reintegree

init_case_ppv_reintegree(year: int, simulation_step_smic: float, simulation_max_smic: float, ppv: float) -> None

Initializes a case with the PPV reintegrated.

Parameters:

Name	Type	Description	Default
year	int	The year for which the simulation is being performed.	required
simulation_step_smic	float	The step size for the simulation, as a multiple of the SMIC value.	required
simulation_max_smic	float	The maximum value for the simulation, as a multiple of the SMIC value.	required
ppv	float	The amount of the PPV.	required

Source code in bozio_wasmer_simulations/simulation/theoretical/ppv.py

def init_case_ppv_reintegree(
    self,
    year: int,
    simulation_step_smic: float,
    simulation_max_smic: float,
    ppv: float,
) -> None:
    """
    Initializes a case with the PPV reintegrated.

    Args:
        year (int):
            The year for which the simulation is being performed.
        simulation_step_smic (float):
            The step size for the simulation, as a multiple of the SMIC value.
        simulation_max_smic (float):
            The maximum value for the simulation, as a multiple of the SMIC value.
        ppv (float):
            The amount of the PPV.
    """
    # Initialisation du cas de base
    self.init_base_case(
        year=year,
        simulation_step_smic=simulation_step_smic,
        simulation_max_smic=simulation_max_smic,
    )
    # Modification des paramètres de PPV
    self.base_case["individus"]["individu_1"]["prime_partage_valeur_exoneree"][
        2022
    ] = 0
    self.base_case["individus"]["individu_1"]["prime_partage_valeur_non_exoneree"][
        2022
    ] = ppv

iterate_reform_simulations

iterate_reform_simulations(scenarios: Union[str, List[str], dict], year: int, simulation_step_smic: float, simulation_max_smic: float, simulation_case: str, ppv: float) -> DataFrame

Simulates multiple reforms.

Iterates over the scenarios and simulates each reform. Concatenates the simulated data for all reforms.

Parameters:

Name	Type	Description	Default
scenarios	Union[str, List[str], dict]	The scenarios to simulate.	required
year	int	The year for which the simulation is being performed.	required
simulation_step_smic	float	The step size for the simulation, as a multiple of the SMIC value.	required
simulation_max_smic	float	The maximum value for the simulation, as a multiple of the SMIC value.	required
simulation_case	str	The simulation case ('exoneree' or 'reintegree').	required
ppv	float	The amount of the PPV.	required

Returns:

Type	Description
DataFrame	The simulated data for all reforms.

Source code in bozio_wasmer_simulations/simulation/theoretical/ppv.py

def iterate_reform_simulations(
    self,
    scenarios: Union[str, List[str], dict],
    year: int,
    simulation_step_smic: float,
    simulation_max_smic: float,
    simulation_case: str,
    ppv: float,
) -> pd.DataFrame:
    """
    Simulates multiple reforms.

    Iterates over the scenarios and simulates each reform.
    Concatenates the simulated data for all reforms.

    Args:
        scenarios (Union[str, List[str], dict]):
            The scenarios to simulate.
        year (int):
            The year for which the simulation is being performed.
        simulation_step_smic (float):
            The step size for the simulation, as a multiple of the SMIC value.
        simulation_max_smic (float):
            The maximum value for the simulation, as a multiple of the SMIC value.
        simulation_case (str):
            The simulation case ('exoneree' or 'reintegree').
        ppv (float):
            The amount of the PPV.

    Returns:
        (pd.DataFrame): The simulated data for all reforms.
    """
    # Initialisation de la liste résultat
    list_data_simul = []
    # Itération sur les scénarii référencés dans le jeu de données de paramètres
    for i, scenario in tqdm(enumerate(self.scenarios_names(scenarios=scenarios))):
        # Itération des réformes
        data_simul = self.simulate_reform(
            name=scenario,
            reform_params=scenarios[scenario.upper()],
            year=year,
            simulation_step_smic=simulation_step_smic,
            simulation_max_smic=simulation_max_smic,
            simulation_case=simulation_case,
            ppv=ppv,
        )
        # Ajout à la liste résultat
        if i > 0:
            list_data_simul.append(
                data_simul.drop(
                    [
                        f"{simulation_case}_salaire_de_base",
                        f"{simulation_case}_salaire_de_base_prop_smic",
                    ],
                    axis=1,
                )
            )
        else:
            list_data_simul.append(data_simul)
    # Concaténation
    data_simul = pd.concat(list_data_simul, axis=1, join="outer")

    return data_simul

plot

plot(data: DataFrame, x: str, hue: Union[str, List[str]], x_label: Optional[Union[str, None]] = None, y_label: Optional[Union[str, None]] = None, hue_label: Optional[Union[str, None]] = None, labels: Optional[Dict[str, str]] = {}, export_key: Optional[Union[PathLike, None]] = None, show: Optional[bool] = True) -> None

Plots the results of the simulation.

Parameters:

Name	Type	Description	Default
data	DataFrame	The data to plot.	required
x	str	The variable to use for the x-axis.	required
hue	Union[str, List[str]]	The variable(s) to use for the hue.	required
x_label	Optional[Union[str, None]]	The label for the x-axis. Defaults to None.	None
y_label	Optional[Union[str, None]]	The label for the y-axis. Defaults to None.	None
hue_label	Optional[Union[str, None]]	The label for the hue. Defaults to None.	None
labels	Optional[Dict[str, str]]	A dictionary of labels to apply to the data. Defaults to {}.	{}
export_key	Optional[Union[PathLike, None]]	The path to save the plot to. Defaults to None.	None
show	Optional[bool]	Whether to display the plot. Defaults to True.	True

Returns:

Type	Description
None	None

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

def plot(
    self,
    data: pd.DataFrame,
    x: str,
    hue: Union[str, List[str]],
    x_label: Optional[Union[str, None]] = None,
    y_label: Optional[Union[str, None]] = None,
    hue_label: Optional[Union[str, None]] = None,
    labels: Optional[Dict[str, str]] = {},
    export_key: Optional[Union[os.PathLike, None]] = None,
    show: Optional[bool] = True,
) -> None:
    """
    Plots the results of the simulation.

    Args:
        data (pd.DataFrame):
            The data to plot.
        x (str):
            The variable to use for the x-axis.
        hue (Union[str, List[str]]):
            The variable(s) to use for the hue.
        x_label (Optional[Union[str, None]], optional):
            The label for the x-axis. Defaults to None.
        y_label (Optional[Union[str, None]], optional):
            The label for the y-axis. Defaults to None.
        hue_label (Optional[Union[str, None]], optional):
            The label for the hue. Defaults to None.
        labels (Optional[Dict[str, str]], optional):
            A dictionary of labels to apply to the data. Defaults to {}.
        export_key (Optional[Union[os.PathLike, None]], optional):
            The path to save the plot to. Defaults to None.
        show (Optional[bool], optional):
            Whether to display the plot. Defaults to True.

    Returns:
        None
    """
    # Conversion des arguments en liste
    if isinstance(hue, str):
        hue = [hue]

    # Création des noms à partir des labels
    id_name = x_label if (x_label is not None) else x
    var_name = hue_label if (hue_label is not None) else "Variable"
    value_name = y_label if (y_label is not None) else "Valeur"

    # Réorganisation du jeu de données
    data_graph = pd.melt(
        frame=data,
        id_vars=x,
        value_vars=hue,
        var_name=var_name,
        value_name=value_name,
    ).rename({x: id_name}, axis=1)
    # Application des labels
    data_graph[var_name] = (
        data_graph[var_name].map(labels).fillna(data_graph[var_name])
    )

    # Initialisation de la figure
    fig, ax = plt.subplots()
    # Construction du graphique
    sns.lineplot(data=data_graph, x=id_name, y=value_name, hue=var_name)
    # Formattage de l'axe des ordonnées
    if all(["_prop_" in var_hue for var_hue in hue]):
        ax.yaxis.set_major_formatter(PercentFormatter(xmax=1))
    # Exportation
    if export_key is not None:
        plt.savefig(export_key, bbox_inches="tight")

    # Logging
    self.logger.info(f"Successfully build graph")

    if show:
        plt.show()
    else:
        plt.close("all")

scenarios_names

scenarios_names(scenarios: Union[str, List[str], dict]) -> List[str]

Extracts the names of the scenarios.

Parameters:

Name	Type	Description	Default
scenarios	Union[str, List[str], dict]	The scenarios to extract the names from.	required

Returns:

Type	Description
List[str]	The names of the scenarios.

Raises:

Type	Description
ValueError	If the type of scenarios is not 'dict', 'list', or 'str'.

Source code in bozio_wasmer_simulations/simulation/theoretical/ppv.py

def scenarios_names(self, scenarios: Union[str, List[str], dict]) -> List[str]:
    """
    Extracts the names of the scenarios.

    Args:
        scenarios (Union[str, List[str], dict]):
            The scenarios to extract the names from.

    Returns:
        (List[str]): The names of the scenarios.

    Raises:
        ValueError: If the type of scenarios is not 'dict', 'list', or 'str'.
    """
    # Extraction des noms des scenarios
    if isinstance(scenarios, str):
        scenarios_names = [scenarios]
    elif isinstance(scenarios, dict):
        scenarios_names = [name.lower() for name in scenarios.keys()]
    elif isinstance(scenarios, list):
        scenarios_names = [name.lower() for name in scenarios]
    else:
        raise ValueError(
            f"Invalid type for scenarios : {type(scenarios)}. Should be in 'dict', 'list' or 'str'"
        )

    return scenarios_names

simulate_reform

simulate_reform(name: str, reform_params: dict, year: int, simulation_step_smic: float, simulation_max_smic: float, simulation_case: str, ppv: float) -> DataFrame

Simulates a reform.

Initializes the simulation case, initializes the tax-benefit system, applies the reform, simulates the variables, and preprocesses the gross salary.

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 which the simulation is being performed.	required
simulation_step_smic	float	The step size for the simulation, as a multiple of the SMIC value.	required
simulation_max_smic	float	The maximum value for the simulation, as a multiple of the SMIC value.	required
simulation_case	str	The simulation case ('exoneree' or 'reintegree').	required
ppv	float	The amount of the PPV.	required

Returns:

Type	Description
DataFrame	The simulated data.

Raises:

Type	Description
ValueError	If the simulation case is not 'exoneree' or 'reintegree'.

Source code in bozio_wasmer_simulations/simulation/theoretical/ppv.py

def simulate_reform(
    self,
    name: str,
    reform_params: dict,
    year: int,
    simulation_step_smic: float,
    simulation_max_smic: float,
    simulation_case: str,
    ppv: float,
) -> pd.DataFrame:
    """
    Simulates a reform.

    Initializes the simulation case, initializes the tax-benefit system,
    applies the reform, simulates the variables, and preprocesses the gross salary.

    Args:
        name (str):
            The name of the reform.
        reform_params (dict):
            The parameters of the reform.
        year (int):
            The year for which the simulation is being performed.
        simulation_step_smic (float):
            The step size for the simulation, as a multiple of the SMIC value.
        simulation_max_smic (float):
            The maximum value for the simulation, as a multiple of the SMIC value.
        simulation_case (str):
            The simulation case ('exoneree' or 'reintegree').
        ppv (float):
            The amount of the PPV.

    Returns:
        (pd.DataFrame): The simulated data.

    Raises:
        ValueError: If the simulation case is not 'exoneree' or 'reintegree'.
    """
    # Initialisation du cas de simulation
    if simulation_case == "exoneree":
        self.init_case_ppv_exoneree(
            year=year,
            simulation_step_smic=simulation_step_smic,
            simulation_max_smic=simulation_max_smic,
            ppv=ppv,
        )
    elif simulation_case == "reintegree":
        self.init_case_ppv_reintegree(
            year=year,
            simulation_step_smic=simulation_step_smic,
            simulation_max_smic=simulation_max_smic,
            ppv=ppv,
        )
    else:
        raise ValueError(
            f"Invalid value for 'simulation_case' : {simulation_case}. Should be in ['exoneree', 'reintegree']"
        )

    # 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
    )

    # Logging
    self.logger.info("Successfully updated the tax-benefit system")

    # Extraction du type de la réforme
    reform_type = reform_params["TYPE"]

    # Itération de la simulation
    data_simul = self.base_case_simulation(
        tax_benefit_system=reformed_tax_benefit_system,
        year=year,
        list_var_simul=["salaire_de_base", f"new_allegement_{reform_type}"],
    )

    # Renomination de la variable simulée pour correspondre au nom du scénario
    data_simul.rename(
        {f"new_allegement_{reform_type}": f"new_allegement_{name.lower()}"},
        axis=1,
        inplace=True,
    )

    # Retraitement du salaire de base
    data_simul = self._preprocess_salaire_de_base(data=data_simul, year=year)

    return data_simul.add_prefix(prefix=f"{simulation_case}_")

value_smic

value_smic(year: int) -> float

Calculates the value of the SMIC for the given year.

Parameters:

Name	Type	Description	Default
year	int	The year for which the SMIC value is calculated.	required

Returns:

Type	Description
float	The value of the SMIC for the given year.

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

def value_smic(self, year: int) -> float:
    """
    Calculates the value of the SMIC for the given year.

    Args:
        year (int):
            The year for which the SMIC value is calculated.

    Returns:
        (float): The value of the SMIC for the given year.
    """
    # Initialisation du système socio-fiscal contenant les valeurs de SMIC en paramètres
    tax_benefit_system = FranceTaxBenefitSystem()
    value_smic = sum(
        [
            tax_benefit_system.get_parameters_at_instant(
                instant=f"{year}-{month}"
            ).marche_travail.salaire_minimum.smic.smic_b_mensuel
            for month in [str(m).zfill(2) for m in range(1, 13)]
        ]
    )
    # Logging
    self.logger.info(f"The SMIC value computed for {year} is {value_smic} €")

    return value_smic