Federated Imputation Execution Environment

Environment

class FedImputeEnv(debug_mode: bool = False)

Methods

  • configuration

  • setup_from_data

  • setup_from_scenario_builder

  • clear_env

  • run_fed_imputation

  • show_env_info

  • save_env

  • load_env

  • reset_env

  • get_data

method Register.register_imputer(imputer_name: str, imputer_class: Union[BaseMLImputer, BaseNNImputer], workflow_name: str, supported_fed_strategies: List[str])

Register a new imputer to the environment

Parameters

  • imputer_name : str — str, name of the imputer

  • imputer_class : Union[BaseMLImputer, BaseNNImputer] — Union[BaseMLImputer, BaseNNImputer], class of the imputer

  • workflow_name : str — str, name of the workflow

  • supported_fed_strategies : List[str] — List[str], list of supported federated strategies

Note

This function will register the imputer to the environment.

Raises

  • ValueError

method Register.register_workflow(workflow_name: str, workflow_class: BaseWorkflow)

Register a new workflow to the environment

Parameters

  • workflow_name : str — str, name of the workflow

  • workflow_class : BaseWorkflow — BaseWorkflow, class of the workflow

Note

This function will register the workflow to the environment.

Raises

  • ValueError

method Register.register_strategy(strategy_name: str, strategy_client: StrategyBaseClient, strategy_server: Union[RawBaseStrategyServer, NNStrategyBaseServer])

Register a new strategy to the environment

Parameters

  • strategy_name : str — str, name of the strategy

  • strategy_client : StrategyBaseClient — StrategyBaseClient, class of the strategy client

  • strategy_server : Union[RawBaseStrategyServer, NNStrategyBaseServer] — Union[RawBaseStrategyServer, NNStrategyBaseServer], class of the strategy server

Note

This function will register the strategy to the environment.

Raises

  • ValueError

Client

class Client(client_id: int, train_data: np.ndarray, test_data: np.ndarray, X_train_ms: np.ndarray, data_config: dict, imp_model_name, imp_model_params, fed_strategy: str, fed_strategy_params: dict, client_config: dict, columns: List[str], register: Register, seed=0)

Client class presenting a client in the federated imputation execution environment, it contains the training and testing data, missing data, imputed data, imputation model class, and federated strategy class.

Attributes

  • client_id : int — client id

  • X_train : np.ndarray — training data

  • y_train : np.ndarray — training labels

  • X_test : np.ndarray — testing data

  • y_test : np.ndarray — testing labels

  • X_train_ms : np.ndarray — missing data

  • X_train_mask : np.ndarray — missing data mask

  • X_train_imp : np.ndarray — imputed data

  • data_utils : dict — data statistics

  • imputer : BaseImputer — imputation model

  • fed_strategy : BaseFedStrategy — federated strategy

  • seed : int — seed

  • client_config : dict — client configuration

Methods

  • initial_impute — Initial imputation

  • fit_local_imp_model — Fit a local imputation model

  • update_local_imp_model — Fit a local imputation model

  • local_imputation — Perform local imputation

  • save_imp_model — Save imputation model

  • load_imp_model — Save imputation model

  • calculate_data_utils — Calculate data statistic

  • profile

method Client.initial_impute(imp_values: np.ndarray, col_type: str = 'num') → None

Initial imputation

Parameters

  • imp_values : np.ndarray — imputation values

  • col_type : str — column type, 'num' or 'cat'

method Client.fit_local_imp_model(params: dict) → Tuple[Union[dict, torch.nn.Module], dict]

Fit a local imputation model

Parameters

  • params : dict — instructions for fitting the imputation model

Returns

  • Tuple[dict, dict] — model parameters and fitting results dictionary

method Client.local_imputation(params: dict) → Union[None, np.ndarray]

Perform local imputation

Parameters

  • params : dict — instructions for imputation - e.g temp_imp for temporary imputation

Returns

  • Union[None, np.ndarray] — imputed data or None

Server

class Server(fed_strategy_name: str, fed_strategy_params: dict, imputer_name: str, imputer_params: dict, global_test: np.ndarray, data_config: dict, server_config: Dict[str, Union[str, int, float]], seed: int = 21, columns: List[str] = None, register: Register = None)

Server class to be used in the federated imputation environment

Attributes

  • fed_strategy — str - name of the federated strategy

  • fed_strategy_params — dict - parameters of the federated strategy

  • server_config — dict - configuration of the server

  • X_test_global — np.ndarray - global test data

Methods

  • initial_impute — Initial imputation

  • local_imputation — Perform local imputation

  • calculate_data_utils

  • profile

Workflow

class BaseWorkflow(name: str)

Bases : ABC

Abstract class for the workflow to be used in the federated imputation environment

Methods

  • fed_imp_sequential — Sequential federated imputation workflow

  • fed_imp_parallel — Parallel federated imputation workflow

  • run_fed_imp — Run the federated imputation workflow based on the

  • eval_and_track

  • eval_and_track_parallel

method BaseWorkflow.run_fed_imp(clients: List[Client], server: Server, evaluator, tracker: Tracker, run_type: str, verbose: int) → Tracker

Run the federated imputation workflow based on the

Parameters

  • clients : List[Client] — List[Client] - list of clients

  • server : Server — Server - server

  • evaluator — Evaluator - evaluator

  • tracker : Tracker — Tracker - tracker to tracking results

  • run_type : str — str - type of the workflow run (sequential or parallel)

Returns

  • Tracker — Tracker - tracker with tracked results

Raises

  • ValueError