Federated Imputation Execution Environment
Environment
fedimpute.execution_environment.env_fedimp.FedImputeEnv
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
fedimpute.execution_environment.loaders.register.Register.register_imputer
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
fedimpute.execution_environment.loaders.register.Register.register_workflow
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
fedimpute.execution_environment.loaders.register.Register.register_strategy
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
fedimpute.execution_environment.client.client.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
fedimpute.execution_environment.client.client.Client.initial_impute
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'
fedimpute.execution_environment.client.client.Client.fit_local_imp_model
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[Union[dict, torch.nn.Module], dict] — model parameters and fitting results dictionary
fedimpute.execution_environment.client.client.Client.local_imputation
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
fedimpute.execution_environment.server.server.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
fedimpute.execution_environment.workflows.workflow.BaseWorkflow
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
fedimpute.execution_environment.workflows.workflow.BaseWorkflow.run_fed_imp
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] — ListClient - 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