Creating Tasks

This guide shows how to define new problem types (tasks) for the DP Testing Platform.

When to Create a New Task

Create a new task when:

• You have a new problem type (e.g., clustering, quantile estimation)
• The input/output format differs from existing tasks
• You need specialized metrics for evaluation

Task Structure

A task consists of:

1. Types: Metadata, Input, and Output dataclasses
2. Task definition: Registration with the platform
3. Metrics: Evaluation functions
4. Algorithms: Implementations that solve the task
5. Instance generators: Data generators for benchmarking

Quick Start: Using the Generator

The fastest way to create a task is using the CLI:

dp-testing-platform generate-task --task-name my_task

This creates a complete task scaffold in ./my_task/:

my_task/
├── __init__.py
├── task.py                   # Task definition
├── types.py                  # Metadata, Input, Output
├── metrics.py                # Evaluation metrics
├── algorithms/
│   ├── __init__.py
│   ├── template_algorithm.py
│   └── lifted.py             # Algorithms lifted from other tasks (optional)
└── instance_generators/
    ├── __init__.py
    └── template_instance.py

Step-by-Step Guide

Imports inside your task package

Use explicit relative imports between the modules of your own task package (from .task import MyTask, from ..types import MyTaskInput). They keep working no matter how the package is imported — added to sys.path directly, vendored inside another project, or loaded from a notebook — whereas absolute self-imports (from my_task.task import …) break as soon as the package is reachable under a different root. Import the library itself absolutely (from dp_testing_platform.core import …). Single-module code such as a dataset converters.py has no package of its own and simply imports the library absolutely.

1. Define Types

Create types.py with frozen dataclasses:

# my_task/types.py
from dataclasses import dataclass
import numpy as np

from dp_testing_platform.core import Domain


@dataclass(frozen=True)
class MyTaskMetadata:
    """Metadata describing the problem instance."""
    n_samples: int
    dim: int
    domain: Domain | None = None  # Optional bounded domain


@dataclass(frozen=True)
class MyTaskInput:
    """Input data for the task."""
    X: np.ndarray  # Shape: (n_samples, dim)


@dataclass(frozen=True)
class MyTaskOutput:
    """Output produced by algorithms."""
    result: np.ndarray  # Shape: (dim,)

Frozen Dataclasses Required

Metadata, Input, and Output must be frozen dataclasses for immutability and hashing.

2. Create the Task

Create task.py:

# my_task/task.py
from dp_testing_platform.core import Task
from dp_testing_platform.tasks import register_task

from .types import MyTaskInput, MyTaskMetadata, MyTaskOutput

MyTask = register_task(
    Task(
        id="my_task",                    # Unique identifier
        name="My Custom Task",           # Display name
        description="Description of what this task does.",
        metadata_cls=MyTaskMetadata,
        input_cls=MyTaskInput,
        output_cls=MyTaskOutput,
    )
)

3. Define Metrics

Create metrics.py:

# my_task/metrics.py
import numpy as np

from dp_testing_platform.core.metrics import MetricSpec

from .task import MyTask
from .types import MyTaskOutput


def l2_error_to_reference(output: MyTaskOutput, reference: np.ndarray) -> float:
    """L2 distance between output and reference."""
    return float(np.linalg.norm(output.result - reference))


def relative_error(output: MyTaskOutput, reference: np.ndarray) -> float:
    """Relative L2 error."""
    ref_norm = np.linalg.norm(reference)
    if ref_norm == 0:
        return float(np.linalg.norm(output.result))
    return float(np.linalg.norm(output.result - reference) / ref_norm)


# Register metrics with the task
MyTask.common_eval_metrics = {
    "l2_error": MetricSpec(
        task=MyTask,
        name="l2_error",
        label="L2 Error",
        description="L2 distance to reference",
        fn=l2_error_to_reference,
        arg_names=["reference"],
        allow_custom_fn=True,
    ),
    "relative_error": MetricSpec(
        task=MyTask,
        name="relative_error",
        label="Relative Error",
        description="Relative L2 error",
        fn=relative_error,
        arg_names=["reference"],
        allow_custom_fn=True,
    ),
}

4. Create an Instance Generator

Create instance_generators/simple.py:

# my_task/instance_generators/simple.py
import numpy as np

from dp_testing_platform.core import InstanceGenerator, Metric

from ..task import MyTask
from ..types import MyTaskInput, MyTaskMetadata


@MyTask.register_instance
class SimpleGenerator(InstanceGenerator):
    """Generates simple test instances."""

    PARAM_INFO = {
        "n_samples": {
            "type": int,
            "description": "Number of samples",
            "default": 100,
            "default_range": [100, 1000],
        },
        "dim": {
            "type": int,
            "description": "Dimensionality",
            "default": 10,
            "default_range": [2, 10, 50],
        },
    }

    AVAILABLE_METRICS = ["l2_error"]

    def __init__(
        self,
        n_samples: int | None = None,
        dim: int | None = None,
        metric: str = "l2_error",
    ):
        self.n_samples = (
            n_samples if n_samples is not None else self.PARAM_INFO["n_samples"]["default"]
        )
        self.dim = dim if dim is not None else self.PARAM_INFO["dim"]["default"]
        self.metric_name = metric

    def generate(self, seed: int) -> tuple[MyTaskMetadata, MyTaskInput, Metric]:
        rng = np.random.default_rng(seed)

        # Generate synthetic data
        X = rng.normal(0, 1, size=(self.n_samples, self.dim))

        metadata = MyTaskMetadata(n_samples=self.n_samples, dim=self.dim)
        input_data = MyTaskInput(X=X)

        # Create reference for metric
        reference = np.mean(X, axis=0)
        metric_spec = MyTask.common_eval_metrics[self.metric_name]
        metric = metric_spec.create_metric(reference=reference)

        return metadata, input_data, metric

Instance generator conventions

• The metric parameter should always be the last argument in __init__.
• Validate the requested metric against AVAILABLE_METRICS in __init__ (if metric not in self.AVAILABLE_METRICS: raise ValueError(...)).

PARAM_INFO

Instance generators declare their parameters in PARAM_INFO. Each entry gives the parameter's type, a description, a default (used when the argument is omitted), and an optional default_range — the discrete values that define the sweep grid for that parameter. A parameter with no default_range falls back to [default] (a single-value grid), so PARAM_INFO fully subsumes a plain discrete grid.

PARAM_INFO = {
    "n_samples": {
        "type": int,
        "description": "Number of samples",
        "default": 1000,
        "default_range": [100, 1000, 10000],
    },
    "dimension": {
        "type": int,
        "description": "Dimensionality",
        "default": 10,
        "default_range": [2, 10, 100],
    },
}

The engine builds the default sweep grid via get_default_param_grid() (the default_range of each parameter), and params / the registry UUID are derived from PARAM_INFO.keys() — so every swept parameter must appear in PARAM_INFO, otherwise distinct configs would serialize identical params and collide on one UUID.

5. Create a Baseline Algorithm

Create algorithms/baseline.py:

# my_task/algorithms/baseline.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 ..task import MyTask
from ..types import MyTaskInput, MyTaskMetadata, MyTaskOutput


@MyTask.register_algorithm
class BaselineAlgorithm(BaseAlgorithm):
    """Non-private baseline algorithm."""

    def run(
        self,
        metadata: MyTaskMetadata,
        input: MyTaskInput,
        budget: DPBudget,
        seed: int,
    ) -> MyTaskOutput:
        # Simple non-private implementation
        result = np.mean(input.X, axis=0)
        return MyTaskOutput(result=result)

6. Wire Everything Together

Create __init__.py:

# my_task/__init__.py
from . import algorithms, instance_generators, metrics
from .task import MyTask
from .types import MyTaskInput, MyTaskMetadata, MyTaskOutput

__all__ = [
    "MyTask",
    "MyTaskInput",
    "MyTaskMetadata",
    "MyTaskOutput",
    "algorithms",
    "instance_generators",
    "metrics",
]

And algorithms/__init__.py:

# my_task/algorithms/__init__.py
from .baseline import BaselineAlgorithm

__all__ = ["BaselineAlgorithm"]

And instance_generators/__init__.py:

# my_task/instance_generators/__init__.py
from .simple import SimpleGenerator

__all__ = ["SimpleGenerator"]

7. Register with the Platform

Add your task to dp_testing_platform/tasks/__init__.py:

# In dp_testing_platform/tasks/__init__.py
from my_task import MyTask  # Add this import

Or import it in your evaluation script:

import my_task  # This triggers registration
from dp_testing_platform.tasks import ALL_TASKS

print(ALL_TASKS["my_task"])  # Your task is now available

Reusing Algorithms Across Tasks (Lifted Classes)

An algorithm native to one task is reused on another by presenting it as a native lifted class of the target task. make_lifted_algorithm (core/lifted.py) wraps a donor algorithm class with three maps and returns a BaseAlgorithm subclass you register like any other algorithm. Put these in the task's algorithms/lifted.py.

The three maps:

• lift_metadata — map this task's public problem spec to the donor's.
• lift_input — map this task's private data to the donor's. This must be per-row: row i of the result depends only on row i of the input. That is the privacy invariant — it makes the donor's (eps, delta)-DP guarantee transfer verbatim (dropping one input row drops exactly one donor row).
• push_output — project the donor's output back to this task's output. This is post-processing (privacy-free); it may read the target-task metadata (passed as an optional second argument) to, e.g., select a coefficient.

Example: reuse MEstimation algorithms on LinearRegression

# In linear_regression/algorithms/lifted.py
import numpy as np

from dp_testing_platform.core.lifted import make_lifted_algorithm
from dp_testing_platform.tasks.m_estimation.algorithms.dp_gradient_descent import (
    DPGradientDescent,
)
from dp_testing_platform.tasks.m_estimation.types import (
    MEstimationInput,
    MEstimationMetadata,
    MEstimationOutput,
)

from ..task import LinearRegression
from ..types import LinearRegressionInput, LinearRegressionMetadata, LinearRegressionOutput


def squared_residual_loss(z, theta):
    ...  # loss(data_row, theta); vectorized over rows


def grad_squared_residual(z, theta):
    ...  # grad(data_row, theta); vectorized over rows


def _lift_metadata(metadata: LinearRegressionMetadata) -> MEstimationMetadata:
    return MEstimationMetadata(
        dim=metadata.dim,
        n_samples=metadata.n_samples,
        loss=squared_residual_loss,
        grad_loss=grad_squared_residual,
    )


def _lift_input(input: LinearRegressionInput) -> MEstimationInput:
    # Per-row: row i -> row i (the privacy invariant).
    return MEstimationInput(data=np.column_stack([input.X, input.y]))


def _push_output(
    output: MEstimationOutput,
    _child_metadata: LinearRegressionMetadata | None = None,
) -> LinearRegressionOutput:
    return LinearRegressionOutput(beta=output.theta)


DPGradientDescentLR = LinearRegression.register_algorithm(
    make_lifted_algorithm(
        DPGradientDescent,
        task=LinearRegression,
        lift_metadata=_lift_metadata,
        lift_input=_lift_input,
        push_output=_push_output,
        name="DPGradientDescentLR",
    )
)

Import the task's algorithms/lifted module from the task package so a clean import registers the lifted classes. A lifted class forwards the donor's HYPERPARAM_INFO/default_search_grid, marks the donor task on LIFTED_FROM, and serializes to an importable path like any other algorithm.

Lifted classes compose. A lifted class can wrap an already-lifted class — e.g. single_parameter_estimation lifts from linear_regression, which itself lifts from m_estimation — so the donor here can be another task's lifted class.

Generalized Donors: The M-Estimation Pattern

m_estimation is a generalized task: its loss and gradient functions are part of the problem specification (Metadata), not the private data (Input):

@dataclass(frozen=True)
class MEstimationMetadata:
    dim: int
    n_samples: int
    loss: Callable[[ArrayND, Array1D], float]      # loss(data, theta)
    grad_loss: Callable[[ArrayND, Array1D], ArrayND]  # grad(data, theta)

@dataclass(frozen=True)
class MEstimationInput:
    data: Array2D  # Only the private data

A lifting task's lift_metadata supplies the loss/gradient functions (e.g. squared residual loss for linear regression) and lift_input transforms the data format. Omit dim in the child's metadata where it is implied and supply dim=1 in lift_metadata for scalar tasks. Loss/gradient functions should handle both single examples and entire datasets (vectorized).

Domain Constraints

Tasks use optional domain fields in metadata rather than creating separate task variants. This avoids task proliferation while allowing algorithms to require bounded domains.

Domain types (in dp_testing_platform/core/domain.py):

• BoxDomain(bounds) - Axis-aligned box, bounds is (d, 2) array
• LpBallDomain(p, radius, dim, center=None) - Lp ball with optional center

Usage in metadata:

from dp_testing_platform.core.domain import Domain, LpBallDomain

@dataclass(frozen=True)
class MeanEstimationMetadata:
    dim: int
    n_samples: int
    domain: Domain | None = None  # None = unbounded

Algorithms that require bounded domains raise ValueError when metadata.domain is None and read e.g. metadata.domain.l2_diameter() for sensitivity calculations — see Using Domain Bounds.

Future extensibility:

If task variants become complex, consider:

1. Task inheritance - BoundedMeanEstimation(MeanEstimation)
2. Protocol-based constraints - Algorithms declare required metadata traits
3. Capability system - Tasks advertise capabilities, algorithms declare requirements

The current approach (optional fields) works well for 2-3 variants per task.

Metric Types

Evaluation Metrics

Used to measure algorithm performance:

MyTask.common_eval_metrics = {
    "error": MetricSpec(
        task=MyTask,
        name="error",
        label="Error",
        description="Error measure",
        fn=error_fn,            # Function to compute error
        arg_names=["reference"], # Arguments beyond output
        allow_custom_fn=True,
    ),
}

DP Quality Metrics

Metrics that require DP noise (for hyperparameter selection). A DP quality metric is a cost: lower is better, since the DP selectors take the argmin over the private scores. Expose a utility-like quantity as its cost complement (e.g. error rate, not accuracy).

from dp_testing_platform.core import accounting
from dp_testing_platform.core.dp_budget import DPBudget

def dp_quality(
    output,
    evaluation_metadata,
    evaluation_input,
    budget: DPBudget,
    seed: int,
):
    """DP-safe quality metric."""
    eps = accounting.spend_as_pure(budget).eps
    rng = np.random.default_rng(seed)
    # Add appropriate noise for privacy
    ...

MyTask.dp_quality_metrics = {
    "dp_quality": MetricSpec(
        task=MyTask,
        name="dp_quality",
        label="DP Quality",
        description="DP-safe quality metric",
        fn=dp_quality,
        allow_custom_fn=False,
    ),
}

Testing Your Task

import numpy as np
from dp_testing_platform import PureDPBudget
from my_task import MyTask

# Test types
metadata = MyTask.Metadata(n_samples=100, dim=5)
input_data = MyTask.Input(X=np.random.randn(100, 5))

# Test algorithm
alg = MyTask.get_algorithms()["BaselineAlgorithm"]()
output = alg.run(metadata, input_data, PureDPBudget(1.0), seed=42)
assert output.result.shape == (5,)

# Test instance generator
gen = MyTask.get_instance_generators()[0](n_samples=100, dim=5)
meta, inp, metric = gen.generate(seed=42)
error = metric(output)
print(f"Error: {error}")

Best Practices

1. Keep types simple: Metadata should only contain problem parameters, not data
2. Frozen dataclasses: Always use @dataclass(frozen=True)
3. Document metrics: Clear descriptions help users understand what's measured
4. Include baselines: Add non-private algorithms for comparison
5. Test generators: Verify they produce valid instances

Next Steps

• Adding Algorithms - Add algorithms to your task
• Adding Datasets - Use real data with converters
• Running Evaluations - Benchmark on your task