Skip to content

Adding Algorithms

This guide shows how to implement your own DP algorithm and benchmark it against built-in baselines.

Algorithm Structure

All algorithms inherit from BaseAlgorithm and must:

  1. Be decorated with @Task.register_algorithm
  2. Implement the run() method (the single abstract method on BaseAlgorithm — see Implementing run() below)
  3. Return the appropriate output type for the task

Quick Example

Here's a simple algorithm for mean estimation:

from typing import Any

import numpy as np

from dp_testing_platform.core import BaseAlgorithm, accounting
from dp_testing_platform.core.dp_budget import DPBudget
from dp_testing_platform.tasks.mean_estimation import MeanEstimationTask


@MeanEstimationTask.register_algorithm
class MyDPMean(BaseAlgorithm):
    """My differentially private mean estimator."""

    def __init__(
        self,
        clip_norm: float = 1.0,
        name: str | None = None,
        description: str | None = None,
        metadata: dict[str, Any] | None = None,
    ):
        super().__init__(name=name, description=description, metadata=metadata)
        self.clip_norm = clip_norm

    def run(
        self,
        metadata,  # MeanEstimationMetadata
        input,     # MeanEstimationInput
        budget: DPBudget,
        seed: int,
    ):
        rng = np.random.default_rng(seed)

        X = input.X
        n = metadata.n_samples

        # Clip each row to bounded L2 norm
        norms = np.linalg.norm(X, axis=1, keepdims=True)
        X_clipped = X * np.minimum(1, self.clip_norm / norms)

        # Compute mean
        mean_est = np.mean(X_clipped, axis=0)

        # Add Gaussian noise for privacy
        sensitivity = 2 * self.clip_norm / n
        sigma = accounting.calibrate_gaussian_sigma(budget, sensitivity)
        mean_est += rng.normal(0, sigma, size=mean_est.shape)

        return self.TASK.Output(mean_est=mean_est)

Step-by-Step Guide

1. Choose a Task

First, identify which task your algorithm solves. Available tasks:

from dp_testing_platform.tasks import ALL_TASKS

for task_id, task in ALL_TASKS.items():
    print(f"{task_id}: {task.name}")

2. Create Your Algorithm File

Create a new Python file (e.g., my_algorithm.py):

# my_algorithm.py
from typing import Any
import numpy as np

from dp_testing_platform.core import BaseAlgorithm
from dp_testing_platform.core.dp_budget import DPBudget
from dp_testing_platform.tasks.linear_regression import LinearRegression


@LinearRegression.register_algorithm
class MyDPLinearRegression(BaseAlgorithm):
    """My DP linear regression algorithm."""

    # Define hyperparameters with defaults and ranges
    HYPERPARAM_INFO = {
        "clip_norm": {
            "default": 1.0,
            "default_range": [0.1, 1.0, 10.0],
            "description": "Gradient clipping norm",
        },
        "learning_rate": {
            "default": 0.01,
            "default_range": [0.001, 0.01, 0.1],
            "description": "Learning rate for gradient descent",
        },
    }

    def __init__(
        self,
        clip_norm: float = 1.0,
        learning_rate: float = 0.01,
        name: str | None = None,
        description: str | None = None,
        metadata: dict[str, Any] | None = None,
    ):
        super().__init__(name=name, description=description, metadata=metadata)
        self.clip_norm = clip_norm
        self.learning_rate = learning_rate

    def run(
        self,
        metadata,
        input,
        budget: DPBudget,
        seed: int,
    ):
        rng = np.random.default_rng(seed)

        X, y = input.X, input.y
        n, d = X.shape

        # Your DP algorithm implementation here
        beta = np.zeros(d)

        # ... gradient descent with noise ...

        return self.TASK.Output(beta=beta)

3. Register Your Algorithm

Import your algorithm file to register it with the task:

# In your script or config
import my_algorithm  # This triggers the @register_algorithm decorator

Or use the CLI:

dp-testing-platform register-algorithms --path my_algorithm.py

4. Run Benchmarks

Evaluate your algorithm:

dp-testing-platform evaluate \
  --task-name linear_regression \
  --epsilons 0.1 1.0 10.0 \
  --deltas 1e-5 \
  --alg-seeds 0 1 2 3 4 \
  --instance-seeds 0 1 2

Your algorithm will be included alongside the built-in baselines. The CLI flags construct concrete approximate-DP budgets here; they are convenience syntax over the engine's single typed budget axis.

Implementing run()

run() is the single abstract method on BaseAlgorithm: subclasses implement it directly.

  • An algorithm runs on instances of its own task (self.TASK format); run() performs no adaptation of its own.
  • Data passed to run() is therefore guaranteed to be in self.TASK format.
  • To reuse an algorithm on a different task, present it as a native lifted class in that task's algorithms/lifted.py (via make_lifted_algorithm) — see docs/custom-tasks.md § "Reusing Algorithms Across Tasks". run_experiment rejects an algorithm whose TASK is not the sweep's task (the error names the donor and points at the lifted class).

The run() method receives:

Parameter Description
metadata Task-specific metadata (e.g., dimensions, sample count, domain bounds)
input The actual data (e.g., X matrix, y vector)
budget One concrete PureDPBudget, ApproxDPBudget, or ZCDPBudget allowance
seed Required integer seed for reproducibility

The method must return an instance of self.TASK.Output.

Type Hints

For better IDE support, use explicit types:

from dp_testing_platform.tasks.mean_estimation.types import (
    MeanEstimationInput,
    MeanEstimationMetadata,
    MeanEstimationOutput,
)
from dp_testing_platform.core.dp_budget import DPBudget

def run(
    self,
    metadata: MeanEstimationMetadata,
    input: MeanEstimationInput,
    budget: DPBudget,
    seed: int,
) -> MeanEstimationOutput:
    ...

Failure Semantics

An algorithm has two distinct ways to not return a result, and the distinction is a privacy boundary:

NoRelease = releasing ⊥ ("no result") as a mechanism outcome. Raising it is a release, not a crash: the failure event must be covered by the algorithm's charged budget, because consumers are allowed to observe and act on it — the engine records an honest no_release status row, and the DP selection tuners score a failed candidate +inf and keep selecting. Two kinds of raiser satisfy the contract:

  • Data-independent infeasibility — the failure is a deterministic function of public quantities (metadata, hyperparameters, and the typed budget), so the failure bit carries no information about the private data. Example: the requested offered budget is below the mechanism's feasibility floor.
  • Accounted data-dependent failure — the failure probability depends on the private data and the algorithm's privacy analysis covers the joint release over outputs-plus-⊥. Example: a Propose-Test-Release check failure. State the argument in a comment at the raise site.

Ordinary exceptions (e.g. RuntimeError) = execution errors. Anything the privacy analysis does not account for — a solver that could not certify its tolerance, non-convergence, numerical breakdown — must abort with an ordinary exception. The engine records these as error rows, and the tuners re-raise them rather than folding them into selection (a bug is not a bad candidate).

Worked example: the AMP objective-perturbation paper assumes the solver reaches its certified gradient-norm tolerance (guaranteed in exact arithmetic for the strongly convex objective). When certification fails at runtime anyway, the implementation raises RuntimeError, not NoRelease: whether the solver certifies depends on the private data, and no budget is charged for that event — so it must surface as a crash outside the accounted release, never as a ⊥ that DP selection could observe.

Hyperparameter Specification

Define HYPERPARAM_INFO for tunable parameters:

HYPERPARAM_INFO = {
    "param_name": {
        "type": float,                     # Parameter type
        "default": 1.0,                    # Default value
        "default_range": [0.01, 1000.0],   # [min, max] the tuner searches between
        "scale": "log",                    # "log" => geometric grid, else "linear"
        "description": "What this does",   # Documentation
    },
}

Recognized HYPERPARAM_INFO fields:

  • type — the parameter's Python type (float / int / str).
  • default — value used when the __init__ argument is omitted.
  • default_range — [min, max] numeric endpoints the tuner searches between. default_search_grid() discretizes this into a grid of n_points values (non-numeric values are enumerated verbatim, but declare categorical candidates with default_grid instead).
  • default_grid — an explicit discrete list of candidate values; takes precedence over default_range in default_search_grid(). The required form for categorical (e.g. str-valued) parameters.
  • scale — "log" makes default_search_grid() space the grid geometrically (use for magnitude-spanning clip multipliers and learning-rate-like params); "linear" (the default when absent) spaces it evenly.
  • tunable — False excludes the parameter from default_search_grid() (e.g. a parameter whose off-default values are not DP-sound). Defaults to True.

BaseAlgorithm.default_search_grid(n_points=7) reads these fields (no data input) to build the discrete grid the DP private-selection tuners consume as alg_hyperparam_grid.

Access hyperparameters:

alg = MyAlgorithm(clip_norm=5.0)
print(alg.hyperparams)  # {"clip_norm": 5.0, "learning_rate": 0.01}
print(MyAlgorithm.get_default_hyperparams())  # {"clip_norm": 1.0, "learning_rate": 0.01}

Defaulting Hyperparameters in __init__

In the library's built-in algorithms, hyperparameter arguments default to None and fall back to HYPERPARAM_INFO[...]["default"] only when the argument is omitted:

@TaskClass.register_algorithm
class MyAlgorithm(BaseAlgorithm):
    HYPERPARAM_INFO = {
        "learning_rate": {
            "type": float,
            "description": "Step size for gradient descent",
            "default": 0.1,
            "default_range": [0.01, 1.0],
        },
        "num_iters": {
            "type": int,
            "description": "Number of iterations",
            "default": 100,
        },
    }

    def __init__(
        self,
        learning_rate: float | None = None,
        num_iters: int | None = None,
    ):
        # Fall back to defaults from HYPERPARAM_INFO only when the argument is
        # omitted (None). Never use `x or default` — it silently replaces
        # valid falsy values like 0/0.0 with the default.
        self.learning_rate = (
            learning_rate
            if learning_rate is not None
            else self.HYPERPARAM_INFO["learning_rate"]["default"]
        )
        self.num_iters = (
            num_iters
            if num_iters is not None
            else self.HYPERPARAM_INFO["num_iters"]["default"]
        )

This keeps HYPERPARAM_INFO the single source of truth for defaults.

Hyperparameter Design Principle

Important

Hyperparameters must be problem-agnostic. A single hyperparameter value should work well across different problem instances (varying dimensions, sample sizes, domain bounds, etc.).

If the optimal parameter depends on problem characteristics (dimension d, sample size n, domain diameter), factor out the data-dependent part:

Approach Example Verdict
Wrong clip = 10.0 (assumes specific dimension) ❌
Right clip_multiplier = 1.0, compute clip = multiplier * sqrt(d) in run() ✅

This ensures:

  1. Users can tune hyperparameters on one problem and transfer to others
  2. Default values are meaningful across problem scales
  3. Hyperparameter search spaces don't depend on problem size

Example: Problem-Agnostic Clipping

@MeanEstimationTask.register_algorithm
class MyDPMean(BaseAlgorithm):
    # Good: clip_multiplier is problem-agnostic
    HYPERPARAM_INFO = {
        "clip_multiplier": {
            "default": 1.0,
            "default_range": [0.1, 1.0, 10.0],
            "description": "Multiplier for dimension-scaled clipping",
        },
    }

    def __init__(self, clip_multiplier: float = 1.0, **kwargs):
        super().__init__(**kwargs)
        self.clip_multiplier = clip_multiplier

    def run(self, metadata, input, budget: DPBudget, seed: int):
        # Compute actual clip from metadata
        d = metadata.dim
        clip = self.clip_multiplier * np.sqrt(d)

        # Now use clip in the algorithm...

Reading Problem Characteristics

Access these from metadata in run():

  • metadata.dim - Dimension of the problem
  • metadata.n_samples - Number of samples
  • metadata.domain - Domain bounds (if bounded)
  • metadata.domain.l2_diameter() - L2 diameter of domain
  • metadata.domain_lower_bound / domain_upper_bound - Scalar bounds (for 1D tasks)

Using Domain Bounds

Many DP algorithms require bounded data. Access bounds from metadata:

def run(self, metadata, input, budget: DPBudget, seed: int):
    if metadata.domain is None:
        raise ValueError("This algorithm requires bounded domain")

    l2_diameter = metadata.domain.l2_diameter()
    sensitivity = l2_diameter / metadata.n_samples

    # Use sensitivity in your noise calibration
    ...

Example: Laplace Mechanism for Mean Estimation

@MeanEstimationTask.register_algorithm
class LaplaceMeanEstimation(BaseAlgorithm):
    """Mean estimation using the Laplace mechanism."""

    def run(
        self,
        metadata,
        input,
        budget: DPBudget,
        seed: int,
    ):
        if metadata.domain is None:
            raise ValueError("Laplace mechanism requires bounded domain")

        eps = accounting.spend_as_pure(budget).eps
        rng = np.random.default_rng(seed)

        X = input.X
        n = metadata.n_samples
        d = metadata.dim

        # Compute non-private mean
        mean_est = np.mean(X, axis=0)

        # Add Laplace noise after an explicit pure-DP spend.
        l1_diameter = metadata.domain.l1_diameter()
        sensitivity = l1_diameter / n
        scale = sensitivity / eps

        mean_est += rng.laplace(0, scale, size=d)

        return self.TASK.Output(mean_est=mean_est)

Testing Your Algorithm

Write unit tests to verify correctness:

import numpy as np
from dp_testing_platform import ApproxDPBudget
from dp_testing_platform.tasks import ALL_TASKS

def test_my_algorithm():
    task = ALL_TASKS["mean_estimation"]

    # Create test data
    X = np.random.randn(100, 5)
    metadata = task.Metadata(dim=5, n_samples=100)
    input_data = task.Input(X=X)

    # Import to register
    import my_algorithm

    # Run algorithm
    alg = my_algorithm.MyDPMean(clip_norm=1.0)
    output = alg.run(metadata, input_data, ApproxDPBudget(1.0, 1e-5), seed=42)

    assert output.mean_est.shape == (5,)
    assert np.all(np.isfinite(output.mean_est))

Common Patterns

Non-Private Baseline

@SomeTask.register_algorithm
class NonPrivateBaseline(BaseAlgorithm):
    """Non-private baseline for comparison."""

    def run(self, metadata, input, budget: DPBudget, seed: int):
        # A non-private baseline ignores the offered budget.
        result = compute_optimal_solution(input)
        return self.TASK.Output(...)

Algorithm Requiring Specific Bounds

def run(self, metadata, input, budget: DPBudget, seed: int):
    domain = metadata.domain
    if domain is None:
        raise ValueError("Algorithm requires bounded domain")

    # For mean estimation with L2 ball domain
    from dp_testing_platform.core import LpBallDomain
    if not isinstance(domain, LpBallDomain):
        raise ValueError("Algorithm requires LpBallDomain")

    radius = domain.radius
    ...

Documenting Your Algorithm

Algorithm class docstrings include a Hyperparameters: section listing each hyperparameter's name, description, and default. Paper implementations cite the paper (title + arXiv/DOI link) in the class docstring:

@LinearRegression.register_algorithm
class ObjectivePerturbation(BaseAlgorithm):
    """Objective perturbation for linear regression (RKW 2023 Thm 3.1, Gaussian).

    Adds a Gaussian linear term to a regularized half-Huber objective and
    returns the minimizer.

    The default two-stage configuration (Approximate Minima Perturbation,
    Iyengar et al. 2019) is solver-robust. The "exact" no-output-stage
    configuration (`ObjectivePerturbation.exact()`, output_budget_frac=0) carries
    a HIDDEN ASSUMPTION (gap 2): its (eps, delta)-DP guarantee holds only for the
    EXACT minimizer of the perturbed objective (Redberg-Koskela-Wang 2023), and
    the numerical solver returns an approximate minimizer, so that configuration's
    released coefficients are NOT certifiably (eps, delta)-DP.

    Hyperparameters:
        C_x_clip: Clipping constant for X. Clips rows to norm C_x_clip * sqrt(d). Default: 1.0.
        huber_L: Half-Huber loss parameter (strictly positive). Default: 1.0.
        lambda_mult: Dimensionless L2 regularization multiplier; the actual
            regularizer is lambda_mult * R^2 with R = C_x_clip * sqrt(d). Must
            exceed 1 for RKW Thm 3.1 feasibility (lambda > beta = R^2). Default: 5.0.
    """

The lambda_mult multiplier is a worked example of the hyperparameter design principle: the regularizer scales with the data radius R = C_x_clip * sqrt(d) inside run(), so a single dimensionless default transfers across problem sizes rather than baking in a fixed lambda.

Note how a privacy-relevant caveat (here, the exact-minimizer assumption) belongs in the docstring and in the registered description so it travels with results.csv.

Next Steps