Enable federated XGBoost using bootstrap aggregation in Task Runner

openfl-workspace/workspace/plan/defaults/aggregator.yaml

@@ -1,4 +1,4 @@
 template : openfl.component.Aggregator
 settings :
-    db_store_rounds   : 2
-    write_logs : true
+    db_store_rounds : 2
+    write_logs      : true

openfl-workspace/workspace/plan/defaults/tasks_xgb.yaml

@@ -0,0 +1,21 @@
+aggregated_model_validation:
+  function : validate_task
+  kwargs :
+    apply : global
+    metrics :
+      - acc
+
+locally_tuned_model_validation:
+  function : validate_task
+  kwargs :
+    apply : local
+    metrics :
+      - acc
+
+train:
+  function : train_task
+  kwargs :
+    metrics :
+    - loss
+  aggregation_type :
+      template : openfl.interface.aggregation_functions.FedBaggingXGBoost  

openfl-workspace/xgb_higgs/plan/cols.yaml

@@ -0,0 +1,5 @@
+# Copyright (C) 2024 Intel Corporation
+# Licensed subject to the terms of the separately executed evaluation license agreement between Intel Corporation and you.
+
+# DO NOT EDIT: This file lists the collaborators associated with the federation. The list will be auto-populated during collaborator creation.
+collaborators:

openfl-workspace/xgb_higgs/plan/data.yaml

@@ -0,0 +1,5 @@
+# Copyright (C) 2024 Intel Corporation
+# Licensed subject to the terms of the separately executed evaluation license agreement between Intel Corporation and you.
+
+# DO NOT EDIT: This file specifies the local data directory associated with the respective collaborator. This will be auto-populated during collaborator creation
+# collaborator_name,data_directory_path

openfl-workspace/xgb_higgs/plan/plan.yaml

@@ -0,0 +1,51 @@
+# Copyright (C) 2024 Intel Corporation
+# Licensed subject to the terms of the separately executed evaluation license agreement between Intel Corporation and you.
+
+aggregator :
+  defaults : plan/defaults/aggregator.yaml
+  template : openfl.component.aggregator.Aggregator
+  settings :
+    init_state_path : save/init.pbuf
+    best_state_path : save/best.pbuf
+    last_state_path : save/last.pbuf
+    rounds_to_train : 10
+    write_logs      : false
+    use_delta_updates : false
+
+collaborator :
+  defaults : plan/defaults/collaborator.yaml
+  template : openfl.component.collaborator.Collaborator
+  settings :
+    delta_updates    : false
+    opt_treatment    : RESET
+
+data_loader :
+  defaults : plan/defaults/data_loader.yaml
+  template : src.dataloader.HiggsDataLoader
+  settings :
+    input_shape : 28
+
+task_runner :
+  defaults : plan/defaults/task_runner.yaml
+  template : src.taskrunner.XGBoostRunner
+  settings :
+    params :
+      objective: binary:logistic
+      eval_metric: logloss
+      max_depth: 6
+      eta: 0.3
+      num_parallel_tree: 1
+
+network :
+  defaults : plan/defaults/network.yaml
+  settings :
+    {}
+
+assigner :
+  defaults : plan/defaults/assigner.yaml
+
+tasks :
+  defaults : plan/defaults/tasks_xgb.yaml
+
+compression_pipeline :
+  defaults : plan/defaults/compression_pipeline.yaml

openfl-workspace/xgb_higgs/requirements.txt

@@ -0,0 +1,3 @@
+modin[all]
+scikit-learn
+xgboost

openfl-workspace/xgb_higgs/src/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0

openfl-workspace/xgb_higgs/src/dataloader.py

@@ -0,0 +1,60 @@
+# Copyright (C) 2024 Intel Corporation
+# Licensed subject to the terms of the separately executed evaluation license agreement between
+# Intel Corporation and you.
+
+from openfl.federated import XGBoostDataLoader
+import os
+import modin.pandas as pd
+
+class HiggsDataLoader(XGBoostDataLoader):
+    """
+    DataLoader for the Higgs dataset.
+
+    This class inherits from XGBoostDataLoader and is responsible for loading
+    the Higgs dataset for training and validation.
+
+    Attributes:
+        X_train (numpy.ndarray): Training features.
+        y_train (numpy.ndarray): Training labels.
+        X_valid (numpy.ndarray): Validation features.
+        y_valid (numpy.ndarray): Validation labels.
+    """
+    def __init__(self, data_path, **kwargs):
+        super().__init__(**kwargs)
+        X_train, y_train, X_valid, y_valid = load_Higgs(
+            data_path, **kwargs
+        )
+        self.X_train = X_train
+        self.y_train = y_train
+        self.X_valid = X_valid
+        self.y_valid = y_valid
+
+
+def load_Higgs(data_path, **kwargs):
+    """
+    Load the Higgs dataset from CSV files.
+
+    The dataset is expected to be in two CSV files: 'train.csv' and 'valid.csv'.
+    The first column in each file represents the labels, and the remaining
+    columns represent the features.
+
+    Args:
+        data_path (str): The directory path where the CSV files are located.
+        **kwargs: Additional keyword arguments.
+
+    Returns:
+        tuple: A tuple containing four elements:
+            - X_train (numpy.ndarray): Training features.
+            - y_train (numpy.ndarray): Training labels.
+            - X_valid (numpy.ndarray): Validation features.
+            - y_valid (numpy.ndarray): Validation labels.
+    """
+    train_data = pd.read_csv(os.path.join(data_path, 'train.csv'), header=None)
+    X_train = train_data.iloc[:, 1:].values
+    y_train = train_data.iloc[:, 0].values
+
+    valid_data = pd.read_csv(os.path.join(data_path, 'valid.csv'), header=None)
+    X_valid = valid_data.iloc[:, 1:].values
+    y_valid = valid_data.iloc[:, 0].values
+
+    return X_train, y_train, X_valid, y_valid

openfl-workspace/xgb_higgs/src/setup_data.py

@@ -0,0 +1,95 @@
+import sys
+import os
+import shutil
+from logging import getLogger
+from urllib.request import urlretrieve
+from hashlib import sha384
+from os import path, makedirs
+from tqdm import tqdm
+import modin.pandas as pd
+import gzip
+from sklearn.model_selection import train_test_split
+import numpy as np
+
+logger = getLogger(__name__)
+
+"""HIGGS Dataset."""
+
+URL = "https://archive.ics.uci.edu/ml/machine-learning-databases/00280/HIGGS.csv.gz"
+FILENAME = "HIGGS.csv.gz"
+CSV_FILENAME = "HIGGS.csv"
+CSV_SHA384 = 'b8b82e11a78b81601381420878ad42ba557291f394a88dc5293e4077c8363c87429639b120e299a2a9939c1f943b6a63'
+DEFAULT_PATH = path.join(os.getcwd(), 'data')
+
+pbar = tqdm(total=None)
+
+def report_hook(count, block_size, total_size):
+    """Update progressbar."""
+    if pbar.total is None and total_size:
+        pbar.total = total_size
+    progress_bytes = count * block_size
+    pbar.update(progress_bytes - pbar.n)
+
+def verify_sha384(file_path, expected_hash):
+    """Verify the SHA-384 hash of a file."""
+    sha384_hash = sha384()
+    with open(file_path, 'rb') as f:
+        for byte_block in iter(lambda: f.read(4096), b""):
+            sha384_hash.update(byte_block)
+    computed_hash = sha384_hash.hexdigest()
+    if computed_hash != expected_hash:
+        raise ValueError(f"SHA-384 hash mismatch: expected {expected_hash}, got {computed_hash}")
+    print(f"SHA-384 hash verified: {computed_hash}")
+
+def setup_data(root: str = DEFAULT_PATH, **kwargs):
+    """Initialize."""
+    makedirs(root, exist_ok=True)
+    filepath = path.join(root, FILENAME)
+    csv_filepath = path.join(root, CSV_FILENAME)
+    if not path.exists(filepath):
+        urlretrieve(URL, filepath, report_hook)  # nosec
+        verify_sha384(filepath, CSV_SHA384)
+        # Extract the CSV file from the gzip file
+        with gzip.open(filepath, 'rb') as f_in:
+            with open(csv_filepath, 'wb') as f_out:
+                shutil.copyfileobj(f_in, f_out)
+
+def main():
+    if len(sys.argv) < 2:
+        raise ValueError("Provide the number of collaborators")
+    src = 'higgs_data'
+    if os.path.exists(src):
+        shutil.rmtree(src)
+    setup_data(src)
+    collaborators = int(sys.argv[1])
+    print("Creating splits for {} collaborators".format(collaborators))
+
+    # Load the dataset
+    higgs_data = pd.read_csv(path.join(src, CSV_FILENAME), header=None)
+
+    # Split the dataset into features and labels
+    X = higgs_data.iloc[:, 1:].values
+    y = higgs_data.iloc[:, 0].values
+
+    # Split the dataset into training and testing sets
+    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
+
+    # Combine X and y for train and test sets
+    train_data = pd.DataFrame(data=np.column_stack((y_train, X_train)))
+    test_data = pd.DataFrame(data=np.column_stack((y_test, X_test)))
+
+    # Split the training data into parts for each collaborator
+    for i in range(collaborators):
+        dst = f'data/{i+1}'
+        makedirs(dst, exist_ok=True)
+
+        # Split the training data for the current collaborator
+        split_train_data = train_data.iloc[i::collaborators]
+        split_train_data.to_csv(path.join(dst, 'train.csv'), index=False, header=False)
+
+        # Split the test data for the current collaborator
+        split_test_data = test_data.iloc[i::collaborators]
+        split_test_data.to_csv(path.join(dst, 'valid.csv'), index=False, header=False)
+
+if __name__ == '__main__':
+    main()

openfl-workspace/xgb_higgs/src/taskrunner.py

@@ -0,0 +1,77 @@
+# Copyright (C) 2024 Intel Corporation
+# Licensed subject to the terms of the separately executed evaluation license agreement between
+# Intel Corporation and you.
+
+"""You may copy this file as the starting point of your own model."""
+import numpy as np
+import xgboost as xgb
+
+from openfl.federated import XGBoostTaskRunner
+from openfl.utilities import Metric
+from sklearn.metrics import accuracy_score
+
+
+class XGBoostRunner(XGBoostTaskRunner):
+    """
+    A class to run XGBoost training and validation tasks.
+
+    This class inherits from XGBoostTaskRunner and provides methods to train and validate
+    an XGBoost model using federated learning.
+
+    Attributes:
+        bst (xgb.Booster): The XGBoost model.
+        params (dict): Parameters for the XGBoost model.
+        num_rounds (int): Number of boosting rounds.
+    """
+    def __init__(self, params=None, num_rounds=1, **kwargs):
+        """
+        Initialize the XGBoostRunner.
+
+        Args:
+            params (dict, optional): Parameters for the XGBoost model. Defaults to None.
+            num_rounds (int, optional): Number of boosting rounds. Defaults to 1.
+            **kwargs: Additional arguments to pass to the function.
+        """
+        super().__init__(**kwargs)
+
+        self.bst = None
+        self.params = params
+        self.num_rounds = num_rounds
+
+    def train_(self, data) -> Metric:
+        """
+        Train the XGBoost model.
+
+        Args:
+            train_dataloader (dict): A dictionary containing the training data with keys 'dmatrix'.
+
+        Returns:
+            Metric: A Metric object containing the training loss.
+        """
+        dtrain = data['dmatrix']
+        evals = [(dtrain, 'train')]
+        evals_result = {}
+
+        self.bst = xgb.train(self.params, dtrain, self.num_rounds, xgb_model=self.bst,
+                             evals=evals, evals_result=evals_result, verbose_eval=False)
+
+        loss = evals_result['train']['logloss'][-1]
+        return Metric(name=self.params['eval_metric'], value=np.array(loss))
+
+    def validate_(self, data) -> Metric:
+        """
+        Validate the XGBoost model.
+
+        Args:
+            validation_dataloader (dict): A dictionary containing the validation data with keys 'dmatrix' and 'labels'.
+
+        Returns:
+            Metric: A Metric object containing the validation accuracy.
+        """
+        dtest = data['dmatrix']
+        y_test = data['labels']
+        preds = self.bst.predict(dtest)
+        y_pred_binary = np.where(preds > 0.5, 1, 0)
+        acc = accuracy_score(y_test, y_pred_binary)
+
+        return Metric(name="accuracy", value=np.array(acc))

openfl/component/aggregator/aggregator.py

@@ -69,6 +69,7 @@ def __init__(
         best_state_path,
         last_state_path,
         assigner,
+        use_delta_updates=True,
         straggler_handling_policy=None,
         rounds_to_train=256,
         single_col_cert_common_name=None,
@@ -186,6 +187,8 @@ def __init__(
         # Initialize a lock for thread safety
         self.lock = Lock()
 
+        self.use_delta_updates = use_delta_updates
+
     def _load_initial_tensors(self):
         """Load all of the tensors required to begin federated learning.
 
@@ -801,7 +804,7 @@ def _prepare_trained(self, tensor_name, origin, round_number, report, agg_result
         # Create delta and save it in TensorDB
         base_model_tk = TensorKey(tensor_name, origin, round_number, report, ("model",))
         base_model_nparray = self.tensor_db.get_tensor_from_cache(base_model_tk)
-        if base_model_nparray is not None:
+        if base_model_nparray is not None and self.use_delta_updates:
             delta_tk, delta_nparray = self.tensor_codec.generate_delta(
                 agg_tag_tk, agg_results, base_model_nparray
             )
@@ -830,7 +833,7 @@ def _prepare_trained(self, tensor_name, origin, round_number, report, agg_result
         self.tensor_db.cache_tensor({decompressed_delta_tk: decompressed_delta_nparray})
 
         # Apply delta (unless delta couldn't be created)
-        if base_model_nparray is not None:
+        if base_model_nparray is not None and self.use_delta_updates:
             self.logger.debug("Applying delta for layer %s", decompressed_delta_tk[0])
             new_model_tk, new_model_nparray = self.tensor_codec.apply_delta(
                 decompressed_delta_tk,

openfl/federated/__init__.py

@@ -20,6 +20,11 @@
     from openfl.federated.data import PyTorchDataLoader
     from openfl.federated.task import FederatedModel  # NOQA
     from openfl.federated.task import PyTorchTaskRunner
+if importlib.util.find_spec("xgboost") is not None:
+    from openfl.federated.data import FederatedDataSet  # NOQA
+    from openfl.federated.data import XGBoostDataLoader
+    from openfl.federated.task import FederatedModel  # NOQA
+    from openfl.federated.task import XGBoostTaskRunner
 
 __all__ = [
     "Plan",

openfl/federated/data/__init__.py

@@ -23,3 +23,7 @@
 if importlib.util.find_spec("torch") is not None:
     from openfl.federated.data.federated_data import FederatedDataSet  # NOQA
     from openfl.federated.data.loader_pt import PyTorchDataLoader  # NOQA
+
+if importlib.util.find_spec("xgboost") is not None:
+    from openfl.federated.data.federated_data import FederatedDataSet  # NOQA
+    from openfl.federated.data.loader_xgb import XGBoostDataLoader  # NOQA