Skip to content

API Reference

This reference exposes the primary research classes and protocol surfaces. The command-line interfaces remain the recommended path for reproducible experiments because they capture configuration and evidence more consistently than ad-hoc interactive calls.

Models

vertimosaic.models.VFLLogisticRegression dataclass

Reference first-principles vertical logistic regression protocol.

Setting both l1 and l2 to non-zero values gives an elastic-net objective. Mini-batches are entity-aligned across every party and are shuffled deterministically from seed. Passive-party logit contributions and residual signals cross the simulated Message transport boundary; raw party feature matrices remain local.

require_entity_ids enables protocol-boundary verification of exact ordered entity alignment. Official VertiMosaic research/reproduction entry points enable it. The low-level class keeps False as a backwards-compatibility bridge for callers that have not yet bound identifiers with bind_entity_ids.

missing_party_policy='error' is the default and rejects incomplete inference. zero_contribution is an explicit research fallback that treats absent passive parties as contributing zero logits; the active party may never be omitted.

residual_dp_backend can apply the executable clipped-Gaussian message-level release mechanism to residual messages sent to passive parties. This protects that release only and is not an end-to-end VFL privacy claim.

Source code in src/vertimosaic/models/vfl_logistic.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
@dataclass
class VFLLogisticRegression:
    """Reference first-principles vertical logistic regression protocol.

    Setting both ``l1`` and ``l2`` to non-zero values gives an elastic-net
    objective. Mini-batches are entity-aligned across every party and are
    shuffled deterministically from ``seed``. Passive-party logit contributions
    and residual signals cross the simulated ``Message`` transport boundary;
    raw party feature matrices remain local.

    ``require_entity_ids`` enables protocol-boundary verification of exact ordered
    entity alignment. Official VertiMosaic research/reproduction entry points enable
    it. The low-level class keeps ``False`` as a backwards-compatibility bridge for
    callers that have not yet bound identifiers with ``bind_entity_ids``.

    ``missing_party_policy='error'`` is the default and rejects incomplete inference.
    ``zero_contribution`` is an explicit research fallback that treats absent passive
    parties as contributing zero logits; the active party may never be omitted.

    ``residual_dp_backend`` can apply the executable clipped-Gaussian message-level
    release mechanism to residual messages sent to passive parties. This protects that
    release only and is not an end-to-end VFL privacy claim.
    """

    learning_rate: float = 0.1
    max_iter: int = 500
    l2: float = 0.0
    l1: float = 0.0
    tolerance: float = 1e-7
    gradient_clip: float | None = 10.0
    class_weight: str | dict[int, float] | None = None
    batch_size: int | None = None
    learning_rate_schedule: str = "constant"
    early_stopping_rounds: int | None = None
    warm_start: bool = False
    residual_noise_std: float = 0.0
    residual_dp_backend: ClippedGaussianDPBackend | None = None
    require_entity_ids: bool = False
    missing_party_policy: str = "error"
    seed: int = 42
    transport: InMemoryTransport = field(default_factory=InMemoryTransport)
    weights_: dict[str, np.ndarray] = field(default_factory=dict, init=False)
    intercept_: float = field(default=0.0, init=False)
    loss_history_: list[float] = field(default_factory=list, init=False)
    validation_loss_history_: list[float] = field(default_factory=list, init=False)
    n_iter_: int = field(default=0, init=False)
    converged_: bool = field(default=False, init=False)
    best_iteration_: int | None = field(default=None, init=False)
    trained_party_names_: tuple[str, ...] = field(default=(), init=False)
    active_party_name_: str | None = field(default=None, init=False)

    def _validate_hyperparameters(self) -> None:
        if self.learning_rate <= 0:
            raise ValueError("learning_rate must be positive")
        if self.max_iter <= 0:
            raise ValueError("max_iter must be positive")
        if self.l1 < 0 or self.l2 < 0:
            raise ValueError("l1 and l2 must be non-negative")
        if self.gradient_clip is not None and self.gradient_clip <= 0:
            raise ValueError("gradient_clip must be positive when supplied")
        if self.batch_size is not None and self.batch_size <= 0:
            raise ValueError("batch_size must be positive when supplied")
        if self.learning_rate_schedule not in {"constant", "inverse_sqrt", "linear_decay"}:
            raise ValueError(
                "learning_rate_schedule must be constant, inverse_sqrt or linear_decay"
            )
        if self.early_stopping_rounds is not None and self.early_stopping_rounds <= 0:
            raise ValueError("early_stopping_rounds must be positive when supplied")
        if not np.isfinite(self.residual_noise_std) or self.residual_noise_std < 0:
            raise ValueError("residual_noise_std must be finite and non-negative")
        if self.residual_noise_std > 0.0 and self.residual_dp_backend is not None:
            raise ValueError("residual_noise_std and residual_dp_backend are mutually exclusive")
        if self.missing_party_policy not in {"error", "zero_contribution"}:
            raise ValueError("missing_party_policy must be error or zero_contribution")
        if isinstance(self.class_weight, str) and self.class_weight != "balanced":
            raise ValueError("class_weight string must be 'balanced'")
        if isinstance(self.class_weight, dict) and any(
            float(value) < 0.0 for value in self.class_weight.values()
        ):
            raise ValueError("class weights must be non-negative")

    def _sample_weights(self, y: np.ndarray) -> np.ndarray:
        n = len(y)
        if self.class_weight == "balanced":
            pos = max(float(y.sum()), 1.0)
            neg = max(float((1.0 - y).sum()), 1.0)
            return np.where(y == 1.0, n / (2.0 * pos), n / (2.0 * neg))
        if isinstance(self.class_weight, dict):
            return np.where(
                y == 1.0,
                float(self.class_weight.get(1, 1.0)),
                float(self.class_weight.get(0, 1.0)),
            )
        return np.ones(n, dtype=float)

    def _epoch_learning_rate(self, epoch: int) -> float:
        if self.learning_rate_schedule == "inverse_sqrt":
            return self.learning_rate / np.sqrt(epoch + 1.0)
        if self.learning_rate_schedule == "linear_decay":
            fraction = max(0.05, 1.0 - epoch / max(self.max_iter, 1))
            return self.learning_rate * fraction
        return self.learning_rate

    def _validate_party_collection(
        self,
        parties: list[PassiveParty],
        *,
        context: str,
        require_entity_ids: bool | None = None,
    ) -> None:
        if not parties:
            raise ValueError(f"{context}: at least one party is required")
        names = [party.name for party in parties]
        if len(set(names)) != len(names):
            raise ValueError(f"{context}: party names must be unique")
        n = parties[0].n_rows
        if any(party.n_rows != n for party in parties):
            raise ValueError(f"{context}: all VFL parties must align to the same row count")
        validate_exact_entity_alignment(
            parties,
            context=context,
            require_bound_ids=self.require_entity_ids
            if require_entity_ids is None
            else require_entity_ids,
        )

    def _initialize(self, parties: list[PassiveParty]) -> None:
        expected = {party.name: party.n_features for party in parties}
        can_reuse = self.warm_start and set(self.weights_) == set(expected)
        if can_reuse:
            can_reuse = all(len(self.weights_[name]) == width for name, width in expected.items())
        if not can_reuse:
            self.weights_ = {
                party.name: np.zeros(party.n_features, dtype=float) for party in parties
            }
            self.intercept_ = 0.0

    def _penalty(self) -> float:
        return float(
            sum(
                0.5 * self.l2 * float(weights @ weights) + self.l1 * float(np.abs(weights).sum())
                for weights in self.weights_.values()
            )
        )

    def _loss(
        self,
        y: np.ndarray,
        probabilities: np.ndarray,
        sample_weight: np.ndarray,
    ) -> float:
        eps = 1e-12
        data_loss = -np.average(
            y * np.log(probabilities + eps) + (1.0 - y) * np.log(1.0 - probabilities + eps),
            weights=sample_weight,
        )
        return float(data_loss + self._penalty())

    def _logits(
        self,
        parties: list[PassiveParty],
        indices: np.ndarray | None = None,
        *,
        stage: str = "epoch",
        step: int | None = None,
    ) -> np.ndarray:
        n = parties[0].n_rows if indices is None else len(indices)
        logits = np.full(n, self.intercept_, dtype=float)
        active_name = parties[0].name
        for party in parties:
            local = party.local_logits(self.weights_[party.name], indices)
            if party.name == active_name:
                logits += local
            else:
                delivered = self.transport.send(
                    local,
                    message_type="local_logits",
                    sender_role=party.name,
                    receiver_role=active_name,
                    direction="forward",
                    stage=stage,
                    step=step,
                )
                logits += np.asarray(delivered, dtype=float)
        return logits

    def fit(
        self,
        active: ActiveParty,
        passive: list[PassiveParty],
        validation_active: ActiveParty | None = None,
        validation_passive: list[PassiveParty] | None = None,
    ) -> VFLLogisticRegression:
        self._validate_hyperparameters()
        parties: list[PassiveParty] = [active, *passive]
        self._validate_party_collection(parties, context="training")
        self.active_party_name_ = active.name
        self.trained_party_names_ = tuple(party.name for party in parties)
        n = active.n_rows
        if self.early_stopping_rounds is not None and validation_active is None:
            raise ValueError("validation data are required when early stopping is enabled")
        self._initialize(parties)
        self.loss_history_.clear()
        self.validation_loss_history_.clear()
        self.n_iter_ = 0
        self.converged_ = False
        self.best_iteration_ = None
        y = active.labels
        sample_weight = self._sample_weights(y)
        if not np.isfinite(sample_weight).all() or float(sample_weight.sum()) <= 0.0:
            raise ValueError("sample weights must be finite with positive total weight")
        rng = np.random.default_rng(self.seed)
        noise_rng = np.random.default_rng(self.seed + 104729)
        batch_size = min(self.batch_size or n, n)

        validation_parties: list[PassiveParty] | None = None
        validation_labels: np.ndarray | None = None
        validation_weights: np.ndarray | None = None
        if validation_active is not None:
            validation_passive = validation_passive or []
            validation_parties = [validation_active, *validation_passive]
            self._validate_party_collection(validation_parties, context="validation")
            if tuple(party.name for party in validation_parties) != self.trained_party_names_:
                raise ValueError(
                    "validation data must provide the same VFL parties "
                    "in the same order as training"
                )
            validation_labels = validation_active.labels
            validation_weights = self._sample_weights(validation_labels)
            if not np.isfinite(validation_weights).all() or float(validation_weights.sum()) <= 0.0:
                raise ValueError("validation weights must be finite with positive total weight")

        best_validation_loss = np.inf
        best_weights: dict[str, np.ndarray] | None = None
        best_intercept = self.intercept_
        rounds_without_improvement = 0
        previous = np.inf
        for epoch in range(self.max_iter):
            logits = self._logits(parties, stage="epoch", step=epoch)
            probs = _sigmoid(logits)
            loss = self._loss(y, probs, sample_weight)
            self.loss_history_.append(loss)
            self.n_iter_ = epoch + 1
            if abs(previous - loss) < self.tolerance:
                self.converged_ = True
                break
            previous = loss

            order = np.arange(n, dtype=int) if batch_size == n else rng.permutation(n)
            rate = self._epoch_learning_rate(epoch)
            for start in range(0, n, batch_size):
                batch = order[start : start + batch_size]
                batch_logits = self._logits(parties, batch, stage="epoch", step=epoch)
                batch_probs = _sigmoid(batch_logits)
                batch_weights = sample_weight[batch]
                batch_weight_sum = float(batch_weights.sum())
                if batch_weight_sum <= 0.0:
                    raise ValueError("every optimization batch must have positive total weight")
                residual = (batch_probs - y[batch]) * batch_weights
                residual *= len(batch) / batch_weight_sum
                delivered_residuals: dict[str, np.ndarray] = {active.name: residual}
                for party in passive:
                    party_residual = residual
                    if self.residual_dp_backend is not None:
                        party_residual = self.residual_dp_backend.release(residual)
                    elif self.residual_noise_std > 0.0:
                        party_residual = residual + noise_rng.normal(
                            scale=self.residual_noise_std,
                            size=residual.shape,
                        )
                    delivered_residuals[party.name] = np.asarray(
                        self.transport.send(
                            party_residual,
                            message_type="residual_signal",
                            sender_role=active.name,
                            receiver_role=party.name,
                            direction="backward",
                            stage="epoch",
                            step=epoch,
                        ),
                        dtype=float,
                    )
                for party in parties:
                    grad = party.local_gradient(delivered_residuals[party.name], batch)
                    grad += self.l2 * self.weights_[party.name]
                    if self.gradient_clip is not None:
                        norm = float(np.linalg.norm(grad))
                        if norm > self.gradient_clip:
                            grad *= self.gradient_clip / max(norm, 1e-12)
                    weights = self.weights_[party.name] - rate * grad
                    if self.l1 > 0:
                        shrink = rate * self.l1
                        weights = np.sign(weights) * np.maximum(np.abs(weights) - shrink, 0.0)
                    self.weights_[party.name] = weights
                self.intercept_ -= rate * float(residual.mean())

            if (
                validation_parties is not None
                and validation_labels is not None
                and validation_weights is not None
            ):
                validation_probability = _sigmoid(
                    self._logits(validation_parties, stage="validation_epoch", step=epoch)
                )
                validation_loss = self._loss(
                    validation_labels,
                    validation_probability,
                    validation_weights,
                )
                self.validation_loss_history_.append(validation_loss)
                if validation_loss < best_validation_loss - 1e-12:
                    best_validation_loss = validation_loss
                    best_weights = {name: weights.copy() for name, weights in self.weights_.items()}
                    best_intercept = self.intercept_
                    self.best_iteration_ = epoch
                    rounds_without_improvement = 0
                else:
                    rounds_without_improvement += 1
                if (
                    self.early_stopping_rounds is not None
                    and rounds_without_improvement >= self.early_stopping_rounds
                ):
                    self.converged_ = True
                    break

        if self.early_stopping_rounds is not None and best_weights is not None:
            self.weights_ = best_weights
            self.intercept_ = best_intercept
        elif self.best_iteration_ is None and self.n_iter_:
            self.best_iteration_ = self.n_iter_ - 1

        final_probability = _sigmoid(self._logits(parties, stage="final_training"))
        final_loss = self._loss(y, final_probability, sample_weight)
        if self.loss_history_:
            self.loss_history_[-1] = final_loss
        else:
            self.loss_history_.append(final_loss)
        return self

    def _inference_parties(self, parties: list[PassiveParty]) -> list[PassiveParty]:
        if not self.weights_ or not self.trained_party_names_ or self.active_party_name_ is None:
            raise RuntimeError("model is not fitted")
        if not parties:
            raise ValueError("at least one party is required")
        n_rows = parties[0].n_rows
        if any(party.n_rows != n_rows for party in parties):
            raise ValueError("inference parties must have equal row counts")
        by_name = {party.name: party for party in parties}
        if len(by_name) != len(parties):
            raise ValueError("inference party names must be unique")
        unknown = set(by_name) - set(self.trained_party_names_)
        if unknown:
            if len(unknown) == 1:
                raise ValueError(f"unknown inference party: {next(iter(unknown))}")
            raise ValueError(f"unknown inference parties: {sorted(unknown)}")
        if self.active_party_name_ not in by_name:
            raise ValueError("the active party cannot be omitted at inference")
        missing = set(self.trained_party_names_) - set(by_name)
        if missing and self.missing_party_policy == "error":
            raise ValueError(
                "missing inference parties: " + ", ".join(sorted(missing)) + "; "
                "set missing_party_policy='zero_contribution' only for an explicitly "
                "evaluated fallback configuration"
            )
        ordered = [by_name[name] for name in self.trained_party_names_ if name in by_name]
        self._validate_party_collection(ordered, context="inference")
        return ordered

    def decision_function(self, parties: list[PassiveParty]) -> np.ndarray:
        ordered = self._inference_parties(parties)
        return self._logits(ordered, stage="inference")

    def predict_proba(self, parties: list[PassiveParty]) -> np.ndarray:
        probability = _sigmoid(self.decision_function(parties))
        return np.column_stack([1.0 - probability, probability])

    def predict(self, parties: list[PassiveParty], threshold: float = 0.5) -> np.ndarray:
        return (self.predict_proba(parties)[:, 1] >= threshold).astype(int)

    def privacy_report(self, *, delta: float) -> dict[str, float | int | str] | None:
        """Return accounting for the optional residual message mechanism, if enabled."""
        if self.residual_dp_backend is None:
            return None
        report = self.residual_dp_backend.privacy_report(delta=delta)
        report["protected_message"] = "active-to-passive residual_signal"
        report["non_guarantee"] = (
            "other VFL messages are outside this mechanism; this is not end-to-end VFL DP"
        )
        return report

learning_rate class-attribute instance-attribute 

learning_rate = 0.1

max_iter class-attribute instance-attribute 

max_iter = 500

l2 class-attribute instance-attribute 

l2 = 0.0

l1 class-attribute instance-attribute 

l1 = 0.0

tolerance class-attribute instance-attribute 

tolerance = 1e-07

gradient_clip class-attribute instance-attribute 

gradient_clip = 10.0

class_weight class-attribute instance-attribute 

class_weight = None

batch_size class-attribute instance-attribute 

batch_size = None

learning_rate_schedule class-attribute instance-attribute 

learning_rate_schedule = 'constant'

early_stopping_rounds class-attribute instance-attribute 

early_stopping_rounds = None

warm_start class-attribute instance-attribute 

warm_start = False

residual_noise_std class-attribute instance-attribute 

residual_noise_std = 0.0

residual_dp_backend class-attribute instance-attribute 

residual_dp_backend = None

require_entity_ids class-attribute instance-attribute 

require_entity_ids = False

missing_party_policy class-attribute instance-attribute 

missing_party_policy = 'error'

seed class-attribute instance-attribute 

seed = 42

transport class-attribute instance-attribute 

transport = field(default_factory=InMemoryTransport)

weights_ class-attribute instance-attribute 

weights_ = field(default_factory=dict, init=False)

intercept_ class-attribute instance-attribute 

intercept_ = field(default=0.0, init=False)

loss_history_ class-attribute instance-attribute 

loss_history_ = field(default_factory=list, init=False)

validation_loss_history_ class-attribute instance-attribute 

validation_loss_history_ = field(default_factory=list, init=False)

n_iter_ class-attribute instance-attribute 

n_iter_ = field(default=0, init=False)

converged_ class-attribute instance-attribute 

converged_ = field(default=False, init=False)

best_iteration_ class-attribute instance-attribute 

best_iteration_ = field(default=None, init=False)

trained_party_names_ class-attribute instance-attribute 

trained_party_names_ = field(default=(), init=False)

active_party_name_ class-attribute instance-attribute 

active_party_name_ = field(default=None, init=False)

_validate_hyperparameters 

_validate_hyperparameters()
Source code in src/vertimosaic/models/vfl_logistic.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def _validate_hyperparameters(self) -> None:
    if self.learning_rate <= 0:
        raise ValueError("learning_rate must be positive")
    if self.max_iter <= 0:
        raise ValueError("max_iter must be positive")
    if self.l1 < 0 or self.l2 < 0:
        raise ValueError("l1 and l2 must be non-negative")
    if self.gradient_clip is not None and self.gradient_clip <= 0:
        raise ValueError("gradient_clip must be positive when supplied")
    if self.batch_size is not None and self.batch_size <= 0:
        raise ValueError("batch_size must be positive when supplied")
    if self.learning_rate_schedule not in {"constant", "inverse_sqrt", "linear_decay"}:
        raise ValueError(
            "learning_rate_schedule must be constant, inverse_sqrt or linear_decay"
        )
    if self.early_stopping_rounds is not None and self.early_stopping_rounds <= 0:
        raise ValueError("early_stopping_rounds must be positive when supplied")
    if not np.isfinite(self.residual_noise_std) or self.residual_noise_std < 0:
        raise ValueError("residual_noise_std must be finite and non-negative")
    if self.residual_noise_std > 0.0 and self.residual_dp_backend is not None:
        raise ValueError("residual_noise_std and residual_dp_backend are mutually exclusive")
    if self.missing_party_policy not in {"error", "zero_contribution"}:
        raise ValueError("missing_party_policy must be error or zero_contribution")
    if isinstance(self.class_weight, str) and self.class_weight != "balanced":
        raise ValueError("class_weight string must be 'balanced'")
    if isinstance(self.class_weight, dict) and any(
        float(value) < 0.0 for value in self.class_weight.values()
    ):
        raise ValueError("class weights must be non-negative")

_sample_weights 

_sample_weights(y)
Source code in src/vertimosaic/models/vfl_logistic.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
def _sample_weights(self, y: np.ndarray) -> np.ndarray:
    n = len(y)
    if self.class_weight == "balanced":
        pos = max(float(y.sum()), 1.0)
        neg = max(float((1.0 - y).sum()), 1.0)
        return np.where(y == 1.0, n / (2.0 * pos), n / (2.0 * neg))
    if isinstance(self.class_weight, dict):
        return np.where(
            y == 1.0,
            float(self.class_weight.get(1, 1.0)),
            float(self.class_weight.get(0, 1.0)),
        )
    return np.ones(n, dtype=float)

_epoch_learning_rate 

_epoch_learning_rate(epoch)
Source code in src/vertimosaic/models/vfl_logistic.py 
116
117
118
119
120
121
122
def _epoch_learning_rate(self, epoch: int) -> float:
    if self.learning_rate_schedule == "inverse_sqrt":
        return self.learning_rate / np.sqrt(epoch + 1.0)
    if self.learning_rate_schedule == "linear_decay":
        fraction = max(0.05, 1.0 - epoch / max(self.max_iter, 1))
        return self.learning_rate * fraction
    return self.learning_rate

_validate_party_collection 

_validate_party_collection(parties, *, context, require_entity_ids=None)
Source code in src/vertimosaic/models/vfl_logistic.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def _validate_party_collection(
    self,
    parties: list[PassiveParty],
    *,
    context: str,
    require_entity_ids: bool | None = None,
) -> None:
    if not parties:
        raise ValueError(f"{context}: at least one party is required")
    names = [party.name for party in parties]
    if len(set(names)) != len(names):
        raise ValueError(f"{context}: party names must be unique")
    n = parties[0].n_rows
    if any(party.n_rows != n for party in parties):
        raise ValueError(f"{context}: all VFL parties must align to the same row count")
    validate_exact_entity_alignment(
        parties,
        context=context,
        require_bound_ids=self.require_entity_ids
        if require_entity_ids is None
        else require_entity_ids,
    )

_initialize 

_initialize(parties)
Source code in src/vertimosaic/models/vfl_logistic.py 
147
148
149
150
151
152
153
154
155
156
def _initialize(self, parties: list[PassiveParty]) -> None:
    expected = {party.name: party.n_features for party in parties}
    can_reuse = self.warm_start and set(self.weights_) == set(expected)
    if can_reuse:
        can_reuse = all(len(self.weights_[name]) == width for name, width in expected.items())
    if not can_reuse:
        self.weights_ = {
            party.name: np.zeros(party.n_features, dtype=float) for party in parties
        }
        self.intercept_ = 0.0

_penalty 

_penalty()
Source code in src/vertimosaic/models/vfl_logistic.py 
158
159
160
161
162
163
164
def _penalty(self) -> float:
    return float(
        sum(
            0.5 * self.l2 * float(weights @ weights) + self.l1 * float(np.abs(weights).sum())
            for weights in self.weights_.values()
        )
    )

_loss 

_loss(y, probabilities, sample_weight)
Source code in src/vertimosaic/models/vfl_logistic.py 
166
167
168
169
170
171
172
173
174
175
176
177
def _loss(
    self,
    y: np.ndarray,
    probabilities: np.ndarray,
    sample_weight: np.ndarray,
) -> float:
    eps = 1e-12
    data_loss = -np.average(
        y * np.log(probabilities + eps) + (1.0 - y) * np.log(1.0 - probabilities + eps),
        weights=sample_weight,
    )
    return float(data_loss + self._penalty())

_logits 

_logits(parties, indices=None, *, stage='epoch', step=None)
Source code in src/vertimosaic/models/vfl_logistic.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def _logits(
    self,
    parties: list[PassiveParty],
    indices: np.ndarray | None = None,
    *,
    stage: str = "epoch",
    step: int | None = None,
) -> np.ndarray:
    n = parties[0].n_rows if indices is None else len(indices)
    logits = np.full(n, self.intercept_, dtype=float)
    active_name = parties[0].name
    for party in parties:
        local = party.local_logits(self.weights_[party.name], indices)
        if party.name == active_name:
            logits += local
        else:
            delivered = self.transport.send(
                local,
                message_type="local_logits",
                sender_role=party.name,
                receiver_role=active_name,
                direction="forward",
                stage=stage,
                step=step,
            )
            logits += np.asarray(delivered, dtype=float)
    return logits

fit 

fit(active, passive, validation_active=None, validation_passive=None)
Source code in src/vertimosaic/models/vfl_logistic.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
def fit(
    self,
    active: ActiveParty,
    passive: list[PassiveParty],
    validation_active: ActiveParty | None = None,
    validation_passive: list[PassiveParty] | None = None,
) -> VFLLogisticRegression:
    self._validate_hyperparameters()
    parties: list[PassiveParty] = [active, *passive]
    self._validate_party_collection(parties, context="training")
    self.active_party_name_ = active.name
    self.trained_party_names_ = tuple(party.name for party in parties)
    n = active.n_rows
    if self.early_stopping_rounds is not None and validation_active is None:
        raise ValueError("validation data are required when early stopping is enabled")
    self._initialize(parties)
    self.loss_history_.clear()
    self.validation_loss_history_.clear()
    self.n_iter_ = 0
    self.converged_ = False
    self.best_iteration_ = None
    y = active.labels
    sample_weight = self._sample_weights(y)
    if not np.isfinite(sample_weight).all() or float(sample_weight.sum()) <= 0.0:
        raise ValueError("sample weights must be finite with positive total weight")
    rng = np.random.default_rng(self.seed)
    noise_rng = np.random.default_rng(self.seed + 104729)
    batch_size = min(self.batch_size or n, n)

    validation_parties: list[PassiveParty] | None = None
    validation_labels: np.ndarray | None = None
    validation_weights: np.ndarray | None = None
    if validation_active is not None:
        validation_passive = validation_passive or []
        validation_parties = [validation_active, *validation_passive]
        self._validate_party_collection(validation_parties, context="validation")
        if tuple(party.name for party in validation_parties) != self.trained_party_names_:
            raise ValueError(
                "validation data must provide the same VFL parties "
                "in the same order as training"
            )
        validation_labels = validation_active.labels
        validation_weights = self._sample_weights(validation_labels)
        if not np.isfinite(validation_weights).all() or float(validation_weights.sum()) <= 0.0:
            raise ValueError("validation weights must be finite with positive total weight")

    best_validation_loss = np.inf
    best_weights: dict[str, np.ndarray] | None = None
    best_intercept = self.intercept_
    rounds_without_improvement = 0
    previous = np.inf
    for epoch in range(self.max_iter):
        logits = self._logits(parties, stage="epoch", step=epoch)
        probs = _sigmoid(logits)
        loss = self._loss(y, probs, sample_weight)
        self.loss_history_.append(loss)
        self.n_iter_ = epoch + 1
        if abs(previous - loss) < self.tolerance:
            self.converged_ = True
            break
        previous = loss

        order = np.arange(n, dtype=int) if batch_size == n else rng.permutation(n)
        rate = self._epoch_learning_rate(epoch)
        for start in range(0, n, batch_size):
            batch = order[start : start + batch_size]
            batch_logits = self._logits(parties, batch, stage="epoch", step=epoch)
            batch_probs = _sigmoid(batch_logits)
            batch_weights = sample_weight[batch]
            batch_weight_sum = float(batch_weights.sum())
            if batch_weight_sum <= 0.0:
                raise ValueError("every optimization batch must have positive total weight")
            residual = (batch_probs - y[batch]) * batch_weights
            residual *= len(batch) / batch_weight_sum
            delivered_residuals: dict[str, np.ndarray] = {active.name: residual}
            for party in passive:
                party_residual = residual
                if self.residual_dp_backend is not None:
                    party_residual = self.residual_dp_backend.release(residual)
                elif self.residual_noise_std > 0.0:
                    party_residual = residual + noise_rng.normal(
                        scale=self.residual_noise_std,
                        size=residual.shape,
                    )
                delivered_residuals[party.name] = np.asarray(
                    self.transport.send(
                        party_residual,
                        message_type="residual_signal",
                        sender_role=active.name,
                        receiver_role=party.name,
                        direction="backward",
                        stage="epoch",
                        step=epoch,
                    ),
                    dtype=float,
                )
            for party in parties:
                grad = party.local_gradient(delivered_residuals[party.name], batch)
                grad += self.l2 * self.weights_[party.name]
                if self.gradient_clip is not None:
                    norm = float(np.linalg.norm(grad))
                    if norm > self.gradient_clip:
                        grad *= self.gradient_clip / max(norm, 1e-12)
                weights = self.weights_[party.name] - rate * grad
                if self.l1 > 0:
                    shrink = rate * self.l1
                    weights = np.sign(weights) * np.maximum(np.abs(weights) - shrink, 0.0)
                self.weights_[party.name] = weights
            self.intercept_ -= rate * float(residual.mean())

        if (
            validation_parties is not None
            and validation_labels is not None
            and validation_weights is not None
        ):
            validation_probability = _sigmoid(
                self._logits(validation_parties, stage="validation_epoch", step=epoch)
            )
            validation_loss = self._loss(
                validation_labels,
                validation_probability,
                validation_weights,
            )
            self.validation_loss_history_.append(validation_loss)
            if validation_loss < best_validation_loss - 1e-12:
                best_validation_loss = validation_loss
                best_weights = {name: weights.copy() for name, weights in self.weights_.items()}
                best_intercept = self.intercept_
                self.best_iteration_ = epoch
                rounds_without_improvement = 0
            else:
                rounds_without_improvement += 1
            if (
                self.early_stopping_rounds is not None
                and rounds_without_improvement >= self.early_stopping_rounds
            ):
                self.converged_ = True
                break

    if self.early_stopping_rounds is not None and best_weights is not None:
        self.weights_ = best_weights
        self.intercept_ = best_intercept
    elif self.best_iteration_ is None and self.n_iter_:
        self.best_iteration_ = self.n_iter_ - 1

    final_probability = _sigmoid(self._logits(parties, stage="final_training"))
    final_loss = self._loss(y, final_probability, sample_weight)
    if self.loss_history_:
        self.loss_history_[-1] = final_loss
    else:
        self.loss_history_.append(final_loss)
    return self

_inference_parties 

_inference_parties(parties)
Source code in src/vertimosaic/models/vfl_logistic.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
def _inference_parties(self, parties: list[PassiveParty]) -> list[PassiveParty]:
    if not self.weights_ or not self.trained_party_names_ or self.active_party_name_ is None:
        raise RuntimeError("model is not fitted")
    if not parties:
        raise ValueError("at least one party is required")
    n_rows = parties[0].n_rows
    if any(party.n_rows != n_rows for party in parties):
        raise ValueError("inference parties must have equal row counts")
    by_name = {party.name: party for party in parties}
    if len(by_name) != len(parties):
        raise ValueError("inference party names must be unique")
    unknown = set(by_name) - set(self.trained_party_names_)
    if unknown:
        if len(unknown) == 1:
            raise ValueError(f"unknown inference party: {next(iter(unknown))}")
        raise ValueError(f"unknown inference parties: {sorted(unknown)}")
    if self.active_party_name_ not in by_name:
        raise ValueError("the active party cannot be omitted at inference")
    missing = set(self.trained_party_names_) - set(by_name)
    if missing and self.missing_party_policy == "error":
        raise ValueError(
            "missing inference parties: " + ", ".join(sorted(missing)) + "; "
            "set missing_party_policy='zero_contribution' only for an explicitly "
            "evaluated fallback configuration"
        )
    ordered = [by_name[name] for name in self.trained_party_names_ if name in by_name]
    self._validate_party_collection(ordered, context="inference")
    return ordered

decision_function 

decision_function(parties)
Source code in src/vertimosaic/models/vfl_logistic.py 
389
390
391
def decision_function(self, parties: list[PassiveParty]) -> np.ndarray:
    ordered = self._inference_parties(parties)
    return self._logits(ordered, stage="inference")

predict_proba 

predict_proba(parties)
Source code in src/vertimosaic/models/vfl_logistic.py 
393
394
395
def predict_proba(self, parties: list[PassiveParty]) -> np.ndarray:
    probability = _sigmoid(self.decision_function(parties))
    return np.column_stack([1.0 - probability, probability])

predict 

predict(parties, threshold=0.5)
Source code in src/vertimosaic/models/vfl_logistic.py 
397
398
def predict(self, parties: list[PassiveParty], threshold: float = 0.5) -> np.ndarray:
    return (self.predict_proba(parties)[:, 1] >= threshold).astype(int)

privacy_report 

privacy_report(*, delta)

Return accounting for the optional residual message mechanism, if enabled.

Source code in src/vertimosaic/models/vfl_logistic.py 
400
401
402
403
404
405
406
407
408
409
def privacy_report(self, *, delta: float) -> dict[str, float | int | str] | None:
    """Return accounting for the optional residual message mechanism, if enabled."""
    if self.residual_dp_backend is None:
        return None
    report = self.residual_dp_backend.privacy_report(delta=delta)
    report["protected_message"] = "active-to-passive residual_signal"
    report["non_guarantee"] = (
        "other VFL messages are outside this mechanism; this is not end-to-end VFL DP"
    )
    return report

__init__ 

__init__(learning_rate=0.1, max_iter=500, l2=0.0, l1=0.0, tolerance=1e-07, gradient_clip=10.0, class_weight=None, batch_size=None, learning_rate_schedule='constant', early_stopping_rounds=None, warm_start=False, residual_noise_std=0.0, residual_dp_backend=None, require_entity_ids=False, missing_party_policy='error', seed=42, transport=InMemoryTransport())

vertimosaic.models.VFLHistGBDT dataclass

CPU vertical histogram gradient boosting research implementation.

Passive parties receive target-derived gradient/Hessian signals through the simulated transport, build histograms from retained local bins, and send only aggregate split statistics plus opaque feature/bin references back through Message objects. The active party sends node membership and selected opaque references through the same transport; the split-owning party resolves its private threshold state, performs routing locally, and returns only aligned partition indices. The model retains only token-only routing handles, never numeric thresholds.

This in-process simulator is not cryptographically secure: gradients, Hessians, node membership, opaque references, and routing information can leak information. See the threat-model documentation.

Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
@dataclass
class VFLHistGBDT:
    """CPU vertical histogram gradient boosting research implementation.

    Passive parties receive target-derived gradient/Hessian signals through the
    simulated transport, build histograms from retained local bins, and send only
    aggregate split statistics plus opaque feature/bin references back through
    ``Message`` objects. The active party sends node membership and selected opaque
    references through the same transport; the split-owning party resolves its private
    threshold state, performs routing locally, and returns only aligned partition
    indices. The model retains only token-only routing handles, never numeric thresholds.

    This in-process simulator is not cryptographically secure: gradients, Hessians,
    node membership, opaque references, and routing information can leak information.
    See the threat-model documentation.
    """

    n_estimators: int = 20
    learning_rate: float = 0.1
    max_depth: int = 3
    max_leaves: int | None = None
    min_samples_leaf: int = 10
    min_child_weight: float = 1e-3
    l2_leaf_reg: float = 1.0
    max_bins: int = 16
    subsample: float = 1.0
    feature_subsample: float = 1.0
    early_stopping_rounds: int | None = None
    missing_party_policy: str = "error"
    seed: int = 42
    transport: InMemoryTransport = field(default_factory=InMemoryTransport)
    trees_: list[TreeNode] = field(default_factory=list, init=False)
    base_score_: float = field(default=0.0, init=False)
    party_names_: list[str] = field(default_factory=list, init=False)
    training_loss_history_: list[float] = field(default_factory=list, init=False)
    validation_loss_history_: list[float] = field(default_factory=list, init=False)
    best_iteration_: int | None = field(default=None, init=False)
    _routing_states: dict[str, HistogramRoutingState] = field(
        default_factory=dict,
        init=False,
        repr=False,
    )

    def _validate_hyperparameters(self) -> None:
        if self.n_estimators <= 0:
            raise ValueError("n_estimators must be positive")
        if self.learning_rate <= 0:
            raise ValueError("learning_rate must be positive")
        if self.max_depth < 0:
            raise ValueError("max_depth must be non-negative")
        if self.max_leaves is not None and self.max_leaves < 2:
            raise ValueError("max_leaves must be at least 2 when supplied")
        if self.min_samples_leaf <= 0:
            raise ValueError("min_samples_leaf must be positive")
        if self.min_child_weight < 0 or self.l2_leaf_reg < 0:
            raise ValueError("child and leaf regularization values must be non-negative")
        if self.max_bins < 2:
            raise ValueError("max_bins must be at least 2")
        if not 0.0 < self.subsample <= 1.0:
            raise ValueError("subsample must be in (0, 1]")
        if not 0.0 < self.feature_subsample <= 1.0:
            raise ValueError("feature_subsample must be in (0, 1]")
        if self.early_stopping_rounds is not None and self.early_stopping_rounds <= 0:
            raise ValueError("early_stopping_rounds must be positive when supplied")
        if self.missing_party_policy not in {"error", "zero_contribution"}:
            raise ValueError("missing_party_policy must be error or zero_contribution")

    @staticmethod
    def _gain(g: float, h: float, reg: float) -> float:
        return (g * g) / (h + reg)

    def _split_gain(self, cand: HistogramCandidate) -> float:
        gl = cand["g_left"]
        hl = cand["h_left"]
        gr = cand["g_right"]
        hr = cand["h_right"]
        if hl < self.min_child_weight or hr < self.min_child_weight:
            return -np.inf
        return 0.5 * (
            self._gain(gl, hl, self.l2_leaf_reg)
            + self._gain(gr, hr, self.l2_leaf_reg)
            - self._gain(gl + gr, hl + hr, self.l2_leaf_reg)
        )

    def _leaf_value(self, gradients: np.ndarray, hessians: np.ndarray, idx: np.ndarray) -> float:
        return -float(gradients[idx].sum()) / (float(hessians[idx].sum()) + self.l2_leaf_reg)

    def _feature_indices(self, party: PassiveParty, rng: np.random.Generator) -> np.ndarray:
        if party.n_features == 0:
            return np.empty(0, dtype=int)
        if self.feature_subsample >= 1.0:
            return np.arange(party.n_features, dtype=int)
        count = max(1, int(np.ceil(party.n_features * self.feature_subsample)))
        return np.sort(rng.choice(party.n_features, size=count, replace=False)).astype(int)

    @staticmethod
    def _candidate_payload(candidates: list[HistogramCandidate]) -> StructuredPayload:
        # Per candidate: 2 opaque integer refs + 4 gradient/Hessian sums + 2 counts.
        scalar_count = len(candidates) * 8
        return StructuredPayload(
            value=candidates,
            shape=(len(candidates), 8),
            scalar_count=scalar_count,
            estimated_bytes=scalar_count * 8,
        )

    @staticmethod
    def _routing_payload(
        left_idx: np.ndarray,
        right_idx: np.ndarray,
    ) -> StructuredPayload:
        scalar_count = int(left_idx.size + right_idx.size)
        return StructuredPayload(
            value=(left_idx, right_idx),
            shape=(scalar_count,),
            scalar_count=scalar_count,
            estimated_bytes=int(left_idx.nbytes + right_idx.nbytes),
        )

    @staticmethod
    def _split_selection_payload(
        split_ref: OpaqueSplitReference,
        indices: np.ndarray,
    ) -> StructuredPayload:
        return StructuredPayload(
            value=(split_ref, indices),
            shape=(len(indices),),
            scalar_count=int(len(indices) + 2),
            estimated_bytes=int(indices.nbytes + 16),
        )

    def _route_selected_split(
        self,
        *,
        party: PassiveParty,
        indices: np.ndarray,
        split_ref: OpaqueSplitReference,
        active_name: str,
        stage: str,
        step: int,
    ) -> tuple[np.ndarray, np.ndarray]:
        routing_state = self._routing_states.get(party.name)
        if routing_state is None:
            raise RuntimeError(f"missing party-local routing handle for {party.name}")

        route_indices = indices
        route_ref = split_ref
        if party.name != active_name:
            delivered = self.transport.send(
                self._split_selection_payload(split_ref, indices),
                message_type="split_selection",
                sender_role=active_name,
                receiver_role=party.name,
                direction="backward",
                stage=stage,
                step=step,
            )
            route_ref, route_indices = delivered
            route_indices = np.asarray(route_indices, dtype=int)

        left_idx, right_idx = party.route_split(route_indices, route_ref, routing_state)
        if party.name != active_name:
            delivered_routing = self.transport.send(
                self._routing_payload(left_idx, right_idx),
                message_type="partition_routing_indices",
                sender_role=party.name,
                receiver_role=active_name,
                direction="forward",
                stage=stage,
                step=step,
            )
            left_idx, right_idx = delivered_routing
        return np.asarray(left_idx, dtype=int), np.asarray(right_idx, dtype=int)

    def _build_node(
        self,
        parties: dict[str, PassiveParty],
        gradients: np.ndarray,
        hessians: np.ndarray,
        gradient_signals: dict[str, np.ndarray],
        hessian_signals: dict[str, np.ndarray],
        indices: np.ndarray,
        depth: int,
        rng: np.random.Generator,
        leaf_count: list[int],
        tree_index: int,
    ) -> TreeNode:
        node = TreeNode(indices=indices.copy(), depth=depth)
        node.value = self._leaf_value(gradients, hessians, indices)
        leaf_limit_reached = self.max_leaves is not None and leaf_count[0] >= self.max_leaves
        if (
            depth >= self.max_depth
            or len(indices) < 2 * self.min_samples_leaf
            or leaf_limit_reached
        ):
            return node

        active_name = next(iter(parties))
        best_gain = 0.0
        best: tuple[PassiveParty, HistogramCandidate] | None = None
        for party in parties.values():
            feature_indices = self._feature_indices(party, rng)
            party_indices = indices
            party_feature_indices = feature_indices
            if party.name != active_name:
                party_indices = np.asarray(
                    self.transport.send(
                        indices,
                        message_type="node_membership",
                        sender_role=active_name,
                        receiver_role=party.name,
                        direction="backward",
                        stage="tree",
                        step=tree_index,
                    ),
                    dtype=int,
                )
                party_feature_indices = np.asarray(
                    self.transport.send(
                        feature_indices,
                        message_type="feature_subsample_refs",
                        sender_role=active_name,
                        receiver_role=party.name,
                        direction="backward",
                        stage="tree",
                        step=tree_index,
                    ),
                    dtype=int,
                )

            local_candidates = party.candidate_histograms(
                gradient_signals[party.name],
                hessian_signals[party.name],
                party_indices,
                self.max_bins,
                self.min_samples_leaf,
                party_feature_indices,
            )
            candidates = local_candidates
            if party.name != active_name:
                candidates = self.transport.send(
                    self._candidate_payload(local_candidates),
                    message_type="candidate_histogram_statistics",
                    sender_role=party.name,
                    receiver_role=active_name,
                    direction="forward",
                    stage="tree",
                    step=tree_index,
                )

            for cand in candidates:
                gain = self._split_gain(cand)
                if gain > best_gain:
                    best_gain = gain
                    best = (party, cand)

        if best is None:
            return node

        party, cand = best
        split_ref = cand["split_ref"]
        left_idx, right_idx = self._route_selected_split(
            party=party,
            indices=indices,
            split_ref=split_ref,
            active_name=active_name,
            stage="tree",
            step=tree_index,
        )
        if len(left_idx) < self.min_samples_leaf or len(right_idx) < self.min_samples_leaf:
            return node

        node.party = party.name
        node.split_ref = split_ref
        node.gain = float(best_gain)
        leaf_count[0] += 1
        node.left = self._build_node(
            parties,
            gradients,
            hessians,
            gradient_signals,
            hessian_signals,
            left_idx,
            depth + 1,
            rng,
            leaf_count,
            tree_index,
        )
        node.right = self._build_node(
            parties,
            gradients,
            hessians,
            gradient_signals,
            hessian_signals,
            right_idx,
            depth + 1,
            rng,
            leaf_count,
            tree_index,
        )
        return node

    def _predict_tree(
        self,
        tree: TreeNode,
        parties: dict[str, PassiveParty],
        n_rows: int,
        *,
        tree_index: int,
        stage: str,
    ) -> np.ndarray:
        out = np.zeros(n_rows, dtype=float)
        active_name = self.party_names_[0]

        def walk(node: TreeNode, idx: np.ndarray) -> None:
            if node.is_leaf:
                out[idx] = node.value
                return
            if node.party is None or node.split_ref is None:
                raise RuntimeError("non-leaf node is missing split metadata")
            if node.party not in parties:
                if self.missing_party_policy == "zero_contribution":
                    out[idx] = 0.0
                    return
                raise ValueError(f"missing split-owning party for inference: {node.party}")
            left_idx, right_idx = self._route_selected_split(
                party=parties[node.party],
                indices=idx,
                split_ref=node.split_ref,
                active_name=active_name,
                stage=stage,
                step=tree_index,
            )
            if node.left is None or node.right is None:
                raise RuntimeError("non-leaf node is missing child nodes")
            walk(node.left, left_idx)
            walk(node.right, right_idx)

        walk(tree, np.arange(n_rows, dtype=int))
        return out

    @staticmethod
    def _party_mapping(active: ActiveParty, passive: list[PassiveParty]) -> dict[str, PassiveParty]:
        return {party.name: party for party in [active, *passive]}

    def fit(
        self,
        active: ActiveParty,
        passive: list[PassiveParty],
        validation_active: ActiveParty | None = None,
        validation_passive: list[PassiveParty] | None = None,
    ) -> VFLHistGBDT:
        self._validate_hyperparameters()
        party_list: list[PassiveParty] = [active, *passive]
        n = active.n_rows
        if any(party.n_rows != n for party in party_list):
            raise ValueError("all VFL parties must align to the same row count")
        if self.early_stopping_rounds is not None and validation_active is None:
            raise ValueError("validation data are required when early stopping is enabled")

        parties = {party.name: party for party in party_list}
        self.party_names_ = list(parties)
        for party in party_list:
            party.prepare_histogram_bins(self.max_bins)
        self._routing_states = {
            party.name: party.export_histogram_routing_state() for party in party_list
        }

        y = active.labels
        prevalence = np.clip(y.mean(), 1e-6, 1.0 - 1e-6)
        self.base_score_ = float(np.log(prevalence / (1.0 - prevalence)))
        raw = np.full(n, self.base_score_, dtype=float)
        self.trees_.clear()
        self.training_loss_history_.clear()
        self.validation_loss_history_.clear()
        self.best_iteration_ = None
        rng = np.random.default_rng(self.seed)

        validation_mapping: dict[str, PassiveParty] | None = None
        validation_raw: np.ndarray | None = None
        validation_labels: np.ndarray | None = None
        if validation_active is not None:
            validation_passive = validation_passive or []
            validation_mapping = self._party_mapping(validation_active, validation_passive)
            if set(validation_mapping) != set(parties):
                raise ValueError("validation data must provide the same VFL parties as training")
            validation_n = validation_active.n_rows
            if any(party.n_rows != validation_n for party in validation_mapping.values()):
                raise ValueError("validation parties must align to the same row count")
            for name, source_party in parties.items():
                source_party.share_histogram_routing_state_with(validation_mapping[name])
            validation_raw = np.full(validation_n, self.base_score_, dtype=float)
            validation_labels = validation_active.labels

        best_loss = np.inf
        rounds_without_improvement = 0
        best_tree_count = 0
        for tree_index in range(self.n_estimators):
            probability = _sigmoid(raw)
            gradients = probability - y
            hessians = np.maximum(probability * (1.0 - probability), 1e-8)
            gradient_signals = {active.name: gradients}
            hessian_signals = {active.name: hessians}
            for party in passive:
                gradient_signals[party.name] = np.asarray(
                    self.transport.send(
                        gradients,
                        message_type="gradients",
                        sender_role=active.name,
                        receiver_role=party.name,
                        direction="backward",
                        stage="tree",
                        step=tree_index,
                    ),
                    dtype=float,
                )
                hessian_signals[party.name] = np.asarray(
                    self.transport.send(
                        hessians,
                        message_type="hessians",
                        sender_role=active.name,
                        receiver_role=party.name,
                        direction="backward",
                        stage="tree",
                        step=tree_index,
                    ),
                    dtype=float,
                )

            if self.subsample >= 1.0:
                tree_indices = np.arange(n, dtype=int)
            else:
                sample_size = max(
                    2 * self.min_samples_leaf,
                    int(np.ceil(n * self.subsample)),
                )
                sample_size = min(sample_size, n)
                tree_indices = np.sort(rng.choice(n, size=sample_size, replace=False)).astype(int)

            tree = self._build_node(
                parties,
                gradients,
                hessians,
                gradient_signals,
                hessian_signals,
                tree_indices,
                0,
                rng,
                [1],
                tree_index,
            )
            self.trees_.append(tree)
            raw += self.learning_rate * self._predict_tree(
                tree,
                parties,
                n,
                tree_index=tree_index,
                stage="training_routing",
            )
            self.training_loss_history_.append(_binary_log_loss(y, _sigmoid(raw)))

            if (
                validation_mapping is not None
                and validation_raw is not None
                and validation_labels is not None
            ):
                validation_raw += self.learning_rate * self._predict_tree(
                    tree,
                    validation_mapping,
                    len(validation_raw),
                    tree_index=tree_index,
                    stage="validation_routing",
                )
                validation_loss = _binary_log_loss(
                    validation_labels,
                    _sigmoid(validation_raw),
                )
                self.validation_loss_history_.append(validation_loss)
                if validation_loss < best_loss - 1e-12:
                    best_loss = validation_loss
                    best_tree_count = len(self.trees_)
                    self.best_iteration_ = best_tree_count - 1
                    rounds_without_improvement = 0
                else:
                    rounds_without_improvement += 1
                if (
                    self.early_stopping_rounds is not None
                    and rounds_without_improvement >= self.early_stopping_rounds
                ):
                    break

        if self.early_stopping_rounds is not None and best_tree_count:
            self.trees_ = self.trees_[:best_tree_count]
        elif self.trees_:
            self.best_iteration_ = len(self.trees_) - 1
        return self

    def decision_function(self, parties: list[PassiveParty]) -> np.ndarray:
        if not self.trees_:
            raise RuntimeError("model is not fitted")
        if not parties:
            raise ValueError("at least one party is required")
        mapping = {party.name: party for party in parties}
        missing = set(self.party_names_) - set(mapping)
        if missing and self.missing_party_policy == "error":
            raise ValueError(f"missing parties for inference: {sorted(missing)}")
        n = parties[0].n_rows
        if any(party.n_rows != n for party in parties):
            raise ValueError("inference parties must have equal row counts")
        raw = np.full(n, self.base_score_, dtype=float)
        for tree_index, tree in enumerate(self.trees_):
            raw += self.learning_rate * self._predict_tree(
                tree,
                mapping,
                n,
                tree_index=tree_index,
                stage="inference_routing",
            )
        return raw

    def predict_proba(self, parties: list[PassiveParty]) -> np.ndarray:
        probability = _sigmoid(self.decision_function(parties))
        return np.column_stack([1.0 - probability, probability])

n_estimators class-attribute instance-attribute 

n_estimators = 20

learning_rate class-attribute instance-attribute 

learning_rate = 0.1

max_depth class-attribute instance-attribute 

max_depth = 3

max_leaves class-attribute instance-attribute 

max_leaves = None

min_samples_leaf class-attribute instance-attribute 

min_samples_leaf = 10

min_child_weight class-attribute instance-attribute 

min_child_weight = 0.001

l2_leaf_reg class-attribute instance-attribute 

l2_leaf_reg = 1.0

max_bins class-attribute instance-attribute 

max_bins = 16

subsample class-attribute instance-attribute 

subsample = 1.0

feature_subsample class-attribute instance-attribute 

feature_subsample = 1.0

early_stopping_rounds class-attribute instance-attribute 

early_stopping_rounds = None

missing_party_policy class-attribute instance-attribute 

missing_party_policy = 'error'

seed class-attribute instance-attribute 

seed = 42

transport class-attribute instance-attribute 

transport = field(default_factory=InMemoryTransport)

trees_ class-attribute instance-attribute 

trees_ = field(default_factory=list, init=False)

base_score_ class-attribute instance-attribute 

base_score_ = field(default=0.0, init=False)

party_names_ class-attribute instance-attribute 

party_names_ = field(default_factory=list, init=False)

training_loss_history_ class-attribute instance-attribute 

training_loss_history_ = field(default_factory=list, init=False)

validation_loss_history_ class-attribute instance-attribute 

validation_loss_history_ = field(default_factory=list, init=False)

best_iteration_ class-attribute instance-attribute 

best_iteration_ = field(default=None, init=False)

_routing_states class-attribute instance-attribute 

_routing_states = field(default_factory=dict, init=False, repr=False)

_validate_hyperparameters 

_validate_hyperparameters()
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def _validate_hyperparameters(self) -> None:
    if self.n_estimators <= 0:
        raise ValueError("n_estimators must be positive")
    if self.learning_rate <= 0:
        raise ValueError("learning_rate must be positive")
    if self.max_depth < 0:
        raise ValueError("max_depth must be non-negative")
    if self.max_leaves is not None and self.max_leaves < 2:
        raise ValueError("max_leaves must be at least 2 when supplied")
    if self.min_samples_leaf <= 0:
        raise ValueError("min_samples_leaf must be positive")
    if self.min_child_weight < 0 or self.l2_leaf_reg < 0:
        raise ValueError("child and leaf regularization values must be non-negative")
    if self.max_bins < 2:
        raise ValueError("max_bins must be at least 2")
    if not 0.0 < self.subsample <= 1.0:
        raise ValueError("subsample must be in (0, 1]")
    if not 0.0 < self.feature_subsample <= 1.0:
        raise ValueError("feature_subsample must be in (0, 1]")
    if self.early_stopping_rounds is not None and self.early_stopping_rounds <= 0:
        raise ValueError("early_stopping_rounds must be positive when supplied")
    if self.missing_party_policy not in {"error", "zero_contribution"}:
        raise ValueError("missing_party_policy must be error or zero_contribution")

_gain staticmethod 

_gain(g, h, reg)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
106
107
108
@staticmethod
def _gain(g: float, h: float, reg: float) -> float:
    return (g * g) / (h + reg)

_split_gain 

_split_gain(cand)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
110
111
112
113
114
115
116
117
118
119
120
121
def _split_gain(self, cand: HistogramCandidate) -> float:
    gl = cand["g_left"]
    hl = cand["h_left"]
    gr = cand["g_right"]
    hr = cand["h_right"]
    if hl < self.min_child_weight or hr < self.min_child_weight:
        return -np.inf
    return 0.5 * (
        self._gain(gl, hl, self.l2_leaf_reg)
        + self._gain(gr, hr, self.l2_leaf_reg)
        - self._gain(gl + gr, hl + hr, self.l2_leaf_reg)
    )

_leaf_value 

_leaf_value(gradients, hessians, idx)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
123
124
def _leaf_value(self, gradients: np.ndarray, hessians: np.ndarray, idx: np.ndarray) -> float:
    return -float(gradients[idx].sum()) / (float(hessians[idx].sum()) + self.l2_leaf_reg)

_feature_indices 

_feature_indices(party, rng)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
126
127
128
129
130
131
132
def _feature_indices(self, party: PassiveParty, rng: np.random.Generator) -> np.ndarray:
    if party.n_features == 0:
        return np.empty(0, dtype=int)
    if self.feature_subsample >= 1.0:
        return np.arange(party.n_features, dtype=int)
    count = max(1, int(np.ceil(party.n_features * self.feature_subsample)))
    return np.sort(rng.choice(party.n_features, size=count, replace=False)).astype(int)

_candidate_payload staticmethod 

_candidate_payload(candidates)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
134
135
136
137
138
139
140
141
142
143
@staticmethod
def _candidate_payload(candidates: list[HistogramCandidate]) -> StructuredPayload:
    # Per candidate: 2 opaque integer refs + 4 gradient/Hessian sums + 2 counts.
    scalar_count = len(candidates) * 8
    return StructuredPayload(
        value=candidates,
        shape=(len(candidates), 8),
        scalar_count=scalar_count,
        estimated_bytes=scalar_count * 8,
    )

_routing_payload staticmethod 

_routing_payload(left_idx, right_idx)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
145
146
147
148
149
150
151
152
153
154
155
156
@staticmethod
def _routing_payload(
    left_idx: np.ndarray,
    right_idx: np.ndarray,
) -> StructuredPayload:
    scalar_count = int(left_idx.size + right_idx.size)
    return StructuredPayload(
        value=(left_idx, right_idx),
        shape=(scalar_count,),
        scalar_count=scalar_count,
        estimated_bytes=int(left_idx.nbytes + right_idx.nbytes),
    )

_split_selection_payload staticmethod 

_split_selection_payload(split_ref, indices)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
158
159
160
161
162
163
164
165
166
167
168
@staticmethod
def _split_selection_payload(
    split_ref: OpaqueSplitReference,
    indices: np.ndarray,
) -> StructuredPayload:
    return StructuredPayload(
        value=(split_ref, indices),
        shape=(len(indices),),
        scalar_count=int(len(indices) + 2),
        estimated_bytes=int(indices.nbytes + 16),
    )

_route_selected_split 

_route_selected_split(*, party, indices, split_ref, active_name, stage, step)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def _route_selected_split(
    self,
    *,
    party: PassiveParty,
    indices: np.ndarray,
    split_ref: OpaqueSplitReference,
    active_name: str,
    stage: str,
    step: int,
) -> tuple[np.ndarray, np.ndarray]:
    routing_state = self._routing_states.get(party.name)
    if routing_state is None:
        raise RuntimeError(f"missing party-local routing handle for {party.name}")

    route_indices = indices
    route_ref = split_ref
    if party.name != active_name:
        delivered = self.transport.send(
            self._split_selection_payload(split_ref, indices),
            message_type="split_selection",
            sender_role=active_name,
            receiver_role=party.name,
            direction="backward",
            stage=stage,
            step=step,
        )
        route_ref, route_indices = delivered
        route_indices = np.asarray(route_indices, dtype=int)

    left_idx, right_idx = party.route_split(route_indices, route_ref, routing_state)
    if party.name != active_name:
        delivered_routing = self.transport.send(
            self._routing_payload(left_idx, right_idx),
            message_type="partition_routing_indices",
            sender_role=party.name,
            receiver_role=active_name,
            direction="forward",
            stage=stage,
            step=step,
        )
        left_idx, right_idx = delivered_routing
    return np.asarray(left_idx, dtype=int), np.asarray(right_idx, dtype=int)

_build_node 

_build_node(parties, gradients, hessians, gradient_signals, hessian_signals, indices, depth, rng, leaf_count, tree_index)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
def _build_node(
    self,
    parties: dict[str, PassiveParty],
    gradients: np.ndarray,
    hessians: np.ndarray,
    gradient_signals: dict[str, np.ndarray],
    hessian_signals: dict[str, np.ndarray],
    indices: np.ndarray,
    depth: int,
    rng: np.random.Generator,
    leaf_count: list[int],
    tree_index: int,
) -> TreeNode:
    node = TreeNode(indices=indices.copy(), depth=depth)
    node.value = self._leaf_value(gradients, hessians, indices)
    leaf_limit_reached = self.max_leaves is not None and leaf_count[0] >= self.max_leaves
    if (
        depth >= self.max_depth
        or len(indices) < 2 * self.min_samples_leaf
        or leaf_limit_reached
    ):
        return node

    active_name = next(iter(parties))
    best_gain = 0.0
    best: tuple[PassiveParty, HistogramCandidate] | None = None
    for party in parties.values():
        feature_indices = self._feature_indices(party, rng)
        party_indices = indices
        party_feature_indices = feature_indices
        if party.name != active_name:
            party_indices = np.asarray(
                self.transport.send(
                    indices,
                    message_type="node_membership",
                    sender_role=active_name,
                    receiver_role=party.name,
                    direction="backward",
                    stage="tree",
                    step=tree_index,
                ),
                dtype=int,
            )
            party_feature_indices = np.asarray(
                self.transport.send(
                    feature_indices,
                    message_type="feature_subsample_refs",
                    sender_role=active_name,
                    receiver_role=party.name,
                    direction="backward",
                    stage="tree",
                    step=tree_index,
                ),
                dtype=int,
            )

        local_candidates = party.candidate_histograms(
            gradient_signals[party.name],
            hessian_signals[party.name],
            party_indices,
            self.max_bins,
            self.min_samples_leaf,
            party_feature_indices,
        )
        candidates = local_candidates
        if party.name != active_name:
            candidates = self.transport.send(
                self._candidate_payload(local_candidates),
                message_type="candidate_histogram_statistics",
                sender_role=party.name,
                receiver_role=active_name,
                direction="forward",
                stage="tree",
                step=tree_index,
            )

        for cand in candidates:
            gain = self._split_gain(cand)
            if gain > best_gain:
                best_gain = gain
                best = (party, cand)

    if best is None:
        return node

    party, cand = best
    split_ref = cand["split_ref"]
    left_idx, right_idx = self._route_selected_split(
        party=party,
        indices=indices,
        split_ref=split_ref,
        active_name=active_name,
        stage="tree",
        step=tree_index,
    )
    if len(left_idx) < self.min_samples_leaf or len(right_idx) < self.min_samples_leaf:
        return node

    node.party = party.name
    node.split_ref = split_ref
    node.gain = float(best_gain)
    leaf_count[0] += 1
    node.left = self._build_node(
        parties,
        gradients,
        hessians,
        gradient_signals,
        hessian_signals,
        left_idx,
        depth + 1,
        rng,
        leaf_count,
        tree_index,
    )
    node.right = self._build_node(
        parties,
        gradients,
        hessians,
        gradient_signals,
        hessian_signals,
        right_idx,
        depth + 1,
        rng,
        leaf_count,
        tree_index,
    )
    return node

_predict_tree 

_predict_tree(tree, parties, n_rows, *, tree_index, stage)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
def _predict_tree(
    self,
    tree: TreeNode,
    parties: dict[str, PassiveParty],
    n_rows: int,
    *,
    tree_index: int,
    stage: str,
) -> np.ndarray:
    out = np.zeros(n_rows, dtype=float)
    active_name = self.party_names_[0]

    def walk(node: TreeNode, idx: np.ndarray) -> None:
        if node.is_leaf:
            out[idx] = node.value
            return
        if node.party is None or node.split_ref is None:
            raise RuntimeError("non-leaf node is missing split metadata")
        if node.party not in parties:
            if self.missing_party_policy == "zero_contribution":
                out[idx] = 0.0
                return
            raise ValueError(f"missing split-owning party for inference: {node.party}")
        left_idx, right_idx = self._route_selected_split(
            party=parties[node.party],
            indices=idx,
            split_ref=node.split_ref,
            active_name=active_name,
            stage=stage,
            step=tree_index,
        )
        if node.left is None or node.right is None:
            raise RuntimeError("non-leaf node is missing child nodes")
        walk(node.left, left_idx)
        walk(node.right, right_idx)

    walk(tree, np.arange(n_rows, dtype=int))
    return out

_party_mapping staticmethod 

_party_mapping(active, passive)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
380
381
382
@staticmethod
def _party_mapping(active: ActiveParty, passive: list[PassiveParty]) -> dict[str, PassiveParty]:
    return {party.name: party for party in [active, *passive]}

fit 

fit(active, passive, validation_active=None, validation_passive=None)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
def fit(
    self,
    active: ActiveParty,
    passive: list[PassiveParty],
    validation_active: ActiveParty | None = None,
    validation_passive: list[PassiveParty] | None = None,
) -> VFLHistGBDT:
    self._validate_hyperparameters()
    party_list: list[PassiveParty] = [active, *passive]
    n = active.n_rows
    if any(party.n_rows != n for party in party_list):
        raise ValueError("all VFL parties must align to the same row count")
    if self.early_stopping_rounds is not None and validation_active is None:
        raise ValueError("validation data are required when early stopping is enabled")

    parties = {party.name: party for party in party_list}
    self.party_names_ = list(parties)
    for party in party_list:
        party.prepare_histogram_bins(self.max_bins)
    self._routing_states = {
        party.name: party.export_histogram_routing_state() for party in party_list
    }

    y = active.labels
    prevalence = np.clip(y.mean(), 1e-6, 1.0 - 1e-6)
    self.base_score_ = float(np.log(prevalence / (1.0 - prevalence)))
    raw = np.full(n, self.base_score_, dtype=float)
    self.trees_.clear()
    self.training_loss_history_.clear()
    self.validation_loss_history_.clear()
    self.best_iteration_ = None
    rng = np.random.default_rng(self.seed)

    validation_mapping: dict[str, PassiveParty] | None = None
    validation_raw: np.ndarray | None = None
    validation_labels: np.ndarray | None = None
    if validation_active is not None:
        validation_passive = validation_passive or []
        validation_mapping = self._party_mapping(validation_active, validation_passive)
        if set(validation_mapping) != set(parties):
            raise ValueError("validation data must provide the same VFL parties as training")
        validation_n = validation_active.n_rows
        if any(party.n_rows != validation_n for party in validation_mapping.values()):
            raise ValueError("validation parties must align to the same row count")
        for name, source_party in parties.items():
            source_party.share_histogram_routing_state_with(validation_mapping[name])
        validation_raw = np.full(validation_n, self.base_score_, dtype=float)
        validation_labels = validation_active.labels

    best_loss = np.inf
    rounds_without_improvement = 0
    best_tree_count = 0
    for tree_index in range(self.n_estimators):
        probability = _sigmoid(raw)
        gradients = probability - y
        hessians = np.maximum(probability * (1.0 - probability), 1e-8)
        gradient_signals = {active.name: gradients}
        hessian_signals = {active.name: hessians}
        for party in passive:
            gradient_signals[party.name] = np.asarray(
                self.transport.send(
                    gradients,
                    message_type="gradients",
                    sender_role=active.name,
                    receiver_role=party.name,
                    direction="backward",
                    stage="tree",
                    step=tree_index,
                ),
                dtype=float,
            )
            hessian_signals[party.name] = np.asarray(
                self.transport.send(
                    hessians,
                    message_type="hessians",
                    sender_role=active.name,
                    receiver_role=party.name,
                    direction="backward",
                    stage="tree",
                    step=tree_index,
                ),
                dtype=float,
            )

        if self.subsample >= 1.0:
            tree_indices = np.arange(n, dtype=int)
        else:
            sample_size = max(
                2 * self.min_samples_leaf,
                int(np.ceil(n * self.subsample)),
            )
            sample_size = min(sample_size, n)
            tree_indices = np.sort(rng.choice(n, size=sample_size, replace=False)).astype(int)

        tree = self._build_node(
            parties,
            gradients,
            hessians,
            gradient_signals,
            hessian_signals,
            tree_indices,
            0,
            rng,
            [1],
            tree_index,
        )
        self.trees_.append(tree)
        raw += self.learning_rate * self._predict_tree(
            tree,
            parties,
            n,
            tree_index=tree_index,
            stage="training_routing",
        )
        self.training_loss_history_.append(_binary_log_loss(y, _sigmoid(raw)))

        if (
            validation_mapping is not None
            and validation_raw is not None
            and validation_labels is not None
        ):
            validation_raw += self.learning_rate * self._predict_tree(
                tree,
                validation_mapping,
                len(validation_raw),
                tree_index=tree_index,
                stage="validation_routing",
            )
            validation_loss = _binary_log_loss(
                validation_labels,
                _sigmoid(validation_raw),
            )
            self.validation_loss_history_.append(validation_loss)
            if validation_loss < best_loss - 1e-12:
                best_loss = validation_loss
                best_tree_count = len(self.trees_)
                self.best_iteration_ = best_tree_count - 1
                rounds_without_improvement = 0
            else:
                rounds_without_improvement += 1
            if (
                self.early_stopping_rounds is not None
                and rounds_without_improvement >= self.early_stopping_rounds
            ):
                break

    if self.early_stopping_rounds is not None and best_tree_count:
        self.trees_ = self.trees_[:best_tree_count]
    elif self.trees_:
        self.best_iteration_ = len(self.trees_) - 1
    return self

decision_function 

decision_function(parties)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
def decision_function(self, parties: list[PassiveParty]) -> np.ndarray:
    if not self.trees_:
        raise RuntimeError("model is not fitted")
    if not parties:
        raise ValueError("at least one party is required")
    mapping = {party.name: party for party in parties}
    missing = set(self.party_names_) - set(mapping)
    if missing and self.missing_party_policy == "error":
        raise ValueError(f"missing parties for inference: {sorted(missing)}")
    n = parties[0].n_rows
    if any(party.n_rows != n for party in parties):
        raise ValueError("inference parties must have equal row counts")
    raw = np.full(n, self.base_score_, dtype=float)
    for tree_index, tree in enumerate(self.trees_):
        raw += self.learning_rate * self._predict_tree(
            tree,
            mapping,
            n,
            tree_index=tree_index,
            stage="inference_routing",
        )
    return raw

predict_proba 

predict_proba(parties)
Source code in src/vertimosaic/models/vfl_hist_gbdt.py 
559
560
561
def predict_proba(self, parties: list[PassiveParty]) -> np.ndarray:
    probability = _sigmoid(self.decision_function(parties))
    return np.column_stack([1.0 - probability, probability])

__init__ 

__init__(n_estimators=20, learning_rate=0.1, max_depth=3, max_leaves=None, min_samples_leaf=10, min_child_weight=0.001, l2_leaf_reg=1.0, max_bins=16, subsample=1.0, feature_subsample=1.0, early_stopping_rounds=None, missing_party_policy='error', seed=42, transport=InMemoryTransport())

Parties

vertimosaic.parties.ActiveParty dataclass

Bases: PassiveParty

Source code in src/vertimosaic/parties/core.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
@dataclass
class ActiveParty(PassiveParty):
    _y: np.ndarray

    def __post_init__(self) -> None:
        super().__post_init__()
        y = np.asarray(self._y, dtype=float).reshape(-1)
        if len(y) != self.n_rows:
            raise ValueError("label vector must match feature rows")
        if not np.all(np.isin(y, [0.0, 1.0])):
            raise ValueError("binary target must contain only 0/1")
        self._y = y

    @property
    def labels(self) -> np.ndarray:
        return self._y

_y instance-attribute 

_y

labels property 

labels

__post_init__ 

__post_init__()
Source code in src/vertimosaic/parties/core.py 
347
348
349
350
351
352
353
354
def __post_init__(self) -> None:
    super().__post_init__()
    y = np.asarray(self._y, dtype=float).reshape(-1)
    if len(y) != self.n_rows:
        raise ValueError("label vector must match feature rows")
    if not np.all(np.isin(y, [0.0, 1.0])):
        raise ValueError("binary target must contain only 0/1")
    self._y = y

__init__ 

__init__(name, _x, _y)

vertimosaic.parties.PassiveParty dataclass

Bases: Party

Source code in src/vertimosaic/parties/core.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
@dataclass
class PassiveParty(Party):
    _x: np.ndarray
    _histogram_bins: np.ndarray | None = field(default=None, init=False, repr=False)
    _histogram_thresholds: dict[int, np.ndarray] = field(
        default_factory=dict,
        init=False,
        repr=False,
    )
    _histogram_max_bins: int | None = field(default=None, init=False, repr=False)
    _histogram_routing_state: HistogramRoutingState | None = field(
        default=None,
        init=False,
        repr=False,
    )
    _histogram_threshold_state: _HistogramThresholdState | None = field(
        default=None,
        init=False,
        repr=False,
    )

    def __post_init__(self) -> None:
        x = np.asarray(self._x, dtype=float)
        if x.ndim != 2:
            raise ValueError("party features must be 2D")
        self._x = x

    @property
    def n_rows(self) -> int:
        return self._x.shape[0]

    @property
    def n_features(self) -> int:
        return self._x.shape[1]

    @property
    def histogram_bins_ready(self) -> bool:
        """Whether party-local histogram bins have been fitted for this matrix."""
        return self._histogram_bins is not None and self._histogram_max_bins is not None

    def local_logits(self, weights: np.ndarray, indices: np.ndarray | None = None) -> np.ndarray:
        x = self._x if indices is None else self._x[indices]
        return x @ weights

    def local_gradient(self, residual: np.ndarray, indices: np.ndarray | None = None) -> np.ndarray:
        x = self._x if indices is None else self._x[indices]
        return x.T @ residual / x.shape[0]

    def prepare_histogram_bins(self, max_bins: int) -> None:
        """Fit and retain party-local quantile bins once for histogram tree training."""
        if max_bins < 2:
            raise ValueError("max_bins must be at least 2")
        binned = np.zeros(self._x.shape, dtype=np.int32)
        thresholds: dict[int, np.ndarray] = {}
        quantiles = np.linspace(0.0, 1.0, max_bins + 1)[1:-1]
        for feature_idx in range(self.n_features):
            values = self._x[:, feature_idx]
            finite = values[np.isfinite(values)]
            feature_thresholds = (
                np.unique(np.quantile(finite, quantiles)).astype(float)
                if finite.size and quantiles.size
                else np.empty(0, dtype=float)
            )
            thresholds[feature_idx] = feature_thresholds
            codes = np.searchsorted(feature_thresholds, values, side="left").astype(np.int32)
            if np.any(~np.isfinite(values)):
                codes[~np.isfinite(values)] = len(feature_thresholds)
            binned[:, feature_idx] = codes

        private_state = _HistogramThresholdState(
            tuple(thresholds[index] for index in range(self.n_features)),
            max_bins=max_bins,
        )
        handle = HistogramRoutingState(
            party_name=self.name,
            state_ref=uuid4().hex,
            n_features=self.n_features,
            max_bins=max_bins,
        )
        self._histogram_bins = binned
        self._histogram_thresholds = thresholds
        self._histogram_max_bins = max_bins
        self._histogram_threshold_state = private_state
        self._histogram_routing_state = handle

    def _ensure_histogram_bins(self, max_bins: int) -> None:
        if self._histogram_bins is None or self._histogram_max_bins != max_bins:
            self.prepare_histogram_bins(max_bins)

    def export_histogram_routing_state(self) -> HistogramRoutingState:
        """Return only the opaque handle; numeric thresholds stay party-local."""
        if self._histogram_routing_state is None:
            raise RuntimeError("party-local histogram bins were not prepared")
        return self._histogram_routing_state

    def share_histogram_routing_state_with(self, other: PassiveParty) -> None:
        """Explicitly attach train-derived routing state to another same-party partition.

        The immutable numeric threshold state moves only between two objects representing
        the same owning organization. The coordinating model receives only the opaque
        ``HistogramRoutingState`` handle. There is intentionally no module-global registry
        or implicit name-based lookup.
        """
        if other.name != self.name:
            raise ValueError("histogram routing state can only be shared within one party")
        if other.n_features != self.n_features:
            raise ValueError("histogram routing state requires matching feature width")
        if self._histogram_routing_state is None or self._histogram_threshold_state is None:
            raise RuntimeError("party-local histogram bins were not prepared")
        other._histogram_routing_state = self._histogram_routing_state
        other._histogram_threshold_state = self._histogram_threshold_state

    def _private_routing_state(
        self,
        handle: HistogramRoutingState,
    ) -> _HistogramThresholdState:
        if handle.party_name != self.name:
            raise ValueError("routing handle belongs to a different party")
        if handle.n_features != self.n_features:
            raise ValueError("routing state feature width does not match this party")
        if self._histogram_routing_state is None or self._histogram_threshold_state is None:
            raise ValueError("routing handle does not match this party's local tree state")
        if handle.state_ref != self._histogram_routing_state.state_ref:
            raise ValueError("routing handle does not match this party's local tree state")
        state = self._histogram_threshold_state
        if state.n_features != self.n_features or state.max_bins != handle.max_bins:
            raise ValueError("private routing state metadata does not match the handle")
        return state

    def candidate_histograms(
        self,
        gradients: np.ndarray,
        hessians: np.ndarray,
        indices: np.ndarray,
        max_bins: int,
        min_samples_leaf: int,
        feature_indices: np.ndarray | None = None,
    ) -> list[HistogramCandidate]:
        """Compute local split statistics from retained bins without exposing thresholds."""
        out: list[HistogramCandidate] = []
        if len(indices) < 2 * min_samples_leaf:
            return out
        self._ensure_histogram_bins(max_bins)
        if self._histogram_bins is None:
            raise RuntimeError("party-local histogram bins were not prepared")
        if feature_indices is None:
            features = np.arange(self.n_features, dtype=int)
        else:
            features = np.asarray(feature_indices, dtype=int).reshape(-1)
            if np.any(features < 0) or np.any(features >= self.n_features):
                raise ValueError("feature_indices contain an out-of-range feature")

        node_gradients = np.asarray(gradients, dtype=float)[indices]
        node_hessians = np.asarray(hessians, dtype=float)[indices]
        total_gradient = float(node_gradients.sum())
        total_hessian = float(node_hessians.sum())
        for feature_idx in features:
            feature_thresholds = self._histogram_thresholds[int(feature_idx)]
            if feature_thresholds.size == 0:
                continue
            bin_codes = self._histogram_bins[indices, feature_idx]
            bin_count = len(feature_thresholds) + 1
            counts = np.bincount(bin_codes, minlength=bin_count)
            gradient_sums = np.bincount(
                bin_codes,
                weights=node_gradients,
                minlength=bin_count,
            )
            hessian_sums = np.bincount(
                bin_codes,
                weights=node_hessians,
                minlength=bin_count,
            )
            cumulative_counts = np.cumsum(counts)
            cumulative_gradients = np.cumsum(gradient_sums)
            cumulative_hessians = np.cumsum(hessian_sums)
            for threshold_idx in range(len(feature_thresholds)):
                n_left = int(cumulative_counts[threshold_idx])
                n_right = len(indices) - n_left
                if n_left < min_samples_leaf or n_right < min_samples_leaf:
                    continue
                g_left = float(cumulative_gradients[threshold_idx])
                h_left = float(cumulative_hessians[threshold_idx])
                out.append(
                    {
                        "split_ref": OpaqueSplitReference(
                            feature_ref=int(feature_idx),
                            bin_ref=int(threshold_idx),
                        ),
                        "g_left": g_left,
                        "h_left": h_left,
                        "g_right": total_gradient - g_left,
                        "h_right": total_hessian - h_left,
                        "n_left": n_left,
                        "n_right": n_right,
                    }
                )
        return out

    def aggregate_local_split_importance(
        self,
        records: list[tuple[OpaqueSplitReference, float]],
    ) -> dict[int, dict[str, float | int]]:
        """Aggregate split usage locally by opaque party feature reference."""
        gains_by_feature: dict[int, list[float]] = {}
        for split_ref, gain in records:
            feature_ref = split_ref.feature_ref
            if feature_ref < 0 or feature_ref >= self.n_features:
                raise ValueError("split feature reference is out of range for this party")
            gains_by_feature.setdefault(feature_ref, []).append(float(gain))
        output: dict[int, dict[str, float | int]] = {}
        for feature_ref, gains in gains_by_feature.items():
            gain_sum = float(np.sum(gains))
            output[feature_ref] = {
                "split_count": len(gains),
                "gain_sum": gain_sum,
                "gain_mean": gain_sum / len(gains),
            }
        return output

    def _route_with_threshold(
        self, indices: np.ndarray, feature_idx: int, threshold: float
    ) -> tuple[np.ndarray, np.ndarray]:
        if feature_idx < 0 or feature_idx >= self.n_features:
            raise ValueError("split feature reference is out of range for this party")
        values = self._x[indices, feature_idx]
        left_mask = values <= threshold
        return indices[left_mask], indices[~left_mask]

    def route_split(
        self,
        indices: np.ndarray,
        split_ref: OpaqueSplitReference,
        routing_state: HistogramRoutingState | None = None,
    ) -> tuple[np.ndarray, np.ndarray]:
        """Apply an opaque split using party-owned training-derived threshold state."""
        handle = routing_state or self._histogram_routing_state
        if handle is None:
            raise RuntimeError("party-local histogram routing handle is unavailable")
        private_state = self._private_routing_state(handle)
        threshold = private_state.threshold_for(split_ref)
        return self._route_with_threshold(indices, split_ref.feature_ref, threshold)

    def route(
        self, indices: np.ndarray, feature_idx: int, threshold: float
    ) -> tuple[np.ndarray, np.ndarray]:
        """Direct local routing helper retained for controlled tests and diagnostics."""
        return self._route_with_threshold(indices, feature_idx, threshold)

_x instance-attribute 

_x

_histogram_bins class-attribute instance-attribute 

_histogram_bins = field(default=None, init=False, repr=False)

_histogram_thresholds class-attribute instance-attribute 

_histogram_thresholds = field(default_factory=dict, init=False, repr=False)

_histogram_max_bins class-attribute instance-attribute 

_histogram_max_bins = field(default=None, init=False, repr=False)

_histogram_routing_state class-attribute instance-attribute 

_histogram_routing_state = field(default=None, init=False, repr=False)

_histogram_threshold_state class-attribute instance-attribute 

_histogram_threshold_state = field(default=None, init=False, repr=False)

n_rows property 

n_rows

n_features property 

n_features

histogram_bins_ready property 

histogram_bins_ready

Whether party-local histogram bins have been fitted for this matrix.

__post_init__ 

__post_init__()
Source code in src/vertimosaic/parties/core.py 
114
115
116
117
118
def __post_init__(self) -> None:
    x = np.asarray(self._x, dtype=float)
    if x.ndim != 2:
        raise ValueError("party features must be 2D")
    self._x = x

local_logits 

local_logits(weights, indices=None)
Source code in src/vertimosaic/parties/core.py 
133
134
135
def local_logits(self, weights: np.ndarray, indices: np.ndarray | None = None) -> np.ndarray:
    x = self._x if indices is None else self._x[indices]
    return x @ weights

local_gradient 

local_gradient(residual, indices=None)
Source code in src/vertimosaic/parties/core.py 
137
138
139
def local_gradient(self, residual: np.ndarray, indices: np.ndarray | None = None) -> np.ndarray:
    x = self._x if indices is None else self._x[indices]
    return x.T @ residual / x.shape[0]

prepare_histogram_bins 

prepare_histogram_bins(max_bins)

Fit and retain party-local quantile bins once for histogram tree training.

Source code in src/vertimosaic/parties/core.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def prepare_histogram_bins(self, max_bins: int) -> None:
    """Fit and retain party-local quantile bins once for histogram tree training."""
    if max_bins < 2:
        raise ValueError("max_bins must be at least 2")
    binned = np.zeros(self._x.shape, dtype=np.int32)
    thresholds: dict[int, np.ndarray] = {}
    quantiles = np.linspace(0.0, 1.0, max_bins + 1)[1:-1]
    for feature_idx in range(self.n_features):
        values = self._x[:, feature_idx]
        finite = values[np.isfinite(values)]
        feature_thresholds = (
            np.unique(np.quantile(finite, quantiles)).astype(float)
            if finite.size and quantiles.size
            else np.empty(0, dtype=float)
        )
        thresholds[feature_idx] = feature_thresholds
        codes = np.searchsorted(feature_thresholds, values, side="left").astype(np.int32)
        if np.any(~np.isfinite(values)):
            codes[~np.isfinite(values)] = len(feature_thresholds)
        binned[:, feature_idx] = codes

    private_state = _HistogramThresholdState(
        tuple(thresholds[index] for index in range(self.n_features)),
        max_bins=max_bins,
    )
    handle = HistogramRoutingState(
        party_name=self.name,
        state_ref=uuid4().hex,
        n_features=self.n_features,
        max_bins=max_bins,
    )
    self._histogram_bins = binned
    self._histogram_thresholds = thresholds
    self._histogram_max_bins = max_bins
    self._histogram_threshold_state = private_state
    self._histogram_routing_state = handle

_ensure_histogram_bins 

_ensure_histogram_bins(max_bins)
Source code in src/vertimosaic/parties/core.py 
178
179
180
def _ensure_histogram_bins(self, max_bins: int) -> None:
    if self._histogram_bins is None or self._histogram_max_bins != max_bins:
        self.prepare_histogram_bins(max_bins)

export_histogram_routing_state 

export_histogram_routing_state()

Return only the opaque handle; numeric thresholds stay party-local.

Source code in src/vertimosaic/parties/core.py 
182
183
184
185
186
def export_histogram_routing_state(self) -> HistogramRoutingState:
    """Return only the opaque handle; numeric thresholds stay party-local."""
    if self._histogram_routing_state is None:
        raise RuntimeError("party-local histogram bins were not prepared")
    return self._histogram_routing_state

share_histogram_routing_state_with 

share_histogram_routing_state_with(other)

Explicitly attach train-derived routing state to another same-party partition.

The immutable numeric threshold state moves only between two objects representing the same owning organization. The coordinating model receives only the opaque HistogramRoutingState handle. There is intentionally no module-global registry or implicit name-based lookup.

Source code in src/vertimosaic/parties/core.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def share_histogram_routing_state_with(self, other: PassiveParty) -> None:
    """Explicitly attach train-derived routing state to another same-party partition.

    The immutable numeric threshold state moves only between two objects representing
    the same owning organization. The coordinating model receives only the opaque
    ``HistogramRoutingState`` handle. There is intentionally no module-global registry
    or implicit name-based lookup.
    """
    if other.name != self.name:
        raise ValueError("histogram routing state can only be shared within one party")
    if other.n_features != self.n_features:
        raise ValueError("histogram routing state requires matching feature width")
    if self._histogram_routing_state is None or self._histogram_threshold_state is None:
        raise RuntimeError("party-local histogram bins were not prepared")
    other._histogram_routing_state = self._histogram_routing_state
    other._histogram_threshold_state = self._histogram_threshold_state

_private_routing_state 

_private_routing_state(handle)
Source code in src/vertimosaic/parties/core.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def _private_routing_state(
    self,
    handle: HistogramRoutingState,
) -> _HistogramThresholdState:
    if handle.party_name != self.name:
        raise ValueError("routing handle belongs to a different party")
    if handle.n_features != self.n_features:
        raise ValueError("routing state feature width does not match this party")
    if self._histogram_routing_state is None or self._histogram_threshold_state is None:
        raise ValueError("routing handle does not match this party's local tree state")
    if handle.state_ref != self._histogram_routing_state.state_ref:
        raise ValueError("routing handle does not match this party's local tree state")
    state = self._histogram_threshold_state
    if state.n_features != self.n_features or state.max_bins != handle.max_bins:
        raise ValueError("private routing state metadata does not match the handle")
    return state

candidate_histograms 

candidate_histograms(gradients, hessians, indices, max_bins, min_samples_leaf, feature_indices=None)

Compute local split statistics from retained bins without exposing thresholds.

Source code in src/vertimosaic/parties/core.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def candidate_histograms(
    self,
    gradients: np.ndarray,
    hessians: np.ndarray,
    indices: np.ndarray,
    max_bins: int,
    min_samples_leaf: int,
    feature_indices: np.ndarray | None = None,
) -> list[HistogramCandidate]:
    """Compute local split statistics from retained bins without exposing thresholds."""
    out: list[HistogramCandidate] = []
    if len(indices) < 2 * min_samples_leaf:
        return out
    self._ensure_histogram_bins(max_bins)
    if self._histogram_bins is None:
        raise RuntimeError("party-local histogram bins were not prepared")
    if feature_indices is None:
        features = np.arange(self.n_features, dtype=int)
    else:
        features = np.asarray(feature_indices, dtype=int).reshape(-1)
        if np.any(features < 0) or np.any(features >= self.n_features):
            raise ValueError("feature_indices contain an out-of-range feature")

    node_gradients = np.asarray(gradients, dtype=float)[indices]
    node_hessians = np.asarray(hessians, dtype=float)[indices]
    total_gradient = float(node_gradients.sum())
    total_hessian = float(node_hessians.sum())
    for feature_idx in features:
        feature_thresholds = self._histogram_thresholds[int(feature_idx)]
        if feature_thresholds.size == 0:
            continue
        bin_codes = self._histogram_bins[indices, feature_idx]
        bin_count = len(feature_thresholds) + 1
        counts = np.bincount(bin_codes, minlength=bin_count)
        gradient_sums = np.bincount(
            bin_codes,
            weights=node_gradients,
            minlength=bin_count,
        )
        hessian_sums = np.bincount(
            bin_codes,
            weights=node_hessians,
            minlength=bin_count,
        )
        cumulative_counts = np.cumsum(counts)
        cumulative_gradients = np.cumsum(gradient_sums)
        cumulative_hessians = np.cumsum(hessian_sums)
        for threshold_idx in range(len(feature_thresholds)):
            n_left = int(cumulative_counts[threshold_idx])
            n_right = len(indices) - n_left
            if n_left < min_samples_leaf or n_right < min_samples_leaf:
                continue
            g_left = float(cumulative_gradients[threshold_idx])
            h_left = float(cumulative_hessians[threshold_idx])
            out.append(
                {
                    "split_ref": OpaqueSplitReference(
                        feature_ref=int(feature_idx),
                        bin_ref=int(threshold_idx),
                    ),
                    "g_left": g_left,
                    "h_left": h_left,
                    "g_right": total_gradient - g_left,
                    "h_right": total_hessian - h_left,
                    "n_left": n_left,
                    "n_right": n_right,
                }
            )
    return out

aggregate_local_split_importance 

aggregate_local_split_importance(records)

Aggregate split usage locally by opaque party feature reference.

Source code in src/vertimosaic/parties/core.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def aggregate_local_split_importance(
    self,
    records: list[tuple[OpaqueSplitReference, float]],
) -> dict[int, dict[str, float | int]]:
    """Aggregate split usage locally by opaque party feature reference."""
    gains_by_feature: dict[int, list[float]] = {}
    for split_ref, gain in records:
        feature_ref = split_ref.feature_ref
        if feature_ref < 0 or feature_ref >= self.n_features:
            raise ValueError("split feature reference is out of range for this party")
        gains_by_feature.setdefault(feature_ref, []).append(float(gain))
    output: dict[int, dict[str, float | int]] = {}
    for feature_ref, gains in gains_by_feature.items():
        gain_sum = float(np.sum(gains))
        output[feature_ref] = {
            "split_count": len(gains),
            "gain_sum": gain_sum,
            "gain_mean": gain_sum / len(gains),
        }
    return output

_route_with_threshold 

_route_with_threshold(indices, feature_idx, threshold)
Source code in src/vertimosaic/parties/core.py 
313
314
315
316
317
318
319
320
def _route_with_threshold(
    self, indices: np.ndarray, feature_idx: int, threshold: float
) -> tuple[np.ndarray, np.ndarray]:
    if feature_idx < 0 or feature_idx >= self.n_features:
        raise ValueError("split feature reference is out of range for this party")
    values = self._x[indices, feature_idx]
    left_mask = values <= threshold
    return indices[left_mask], indices[~left_mask]

route_split 

route_split(indices, split_ref, routing_state=None)

Apply an opaque split using party-owned training-derived threshold state.

Source code in src/vertimosaic/parties/core.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
def route_split(
    self,
    indices: np.ndarray,
    split_ref: OpaqueSplitReference,
    routing_state: HistogramRoutingState | None = None,
) -> tuple[np.ndarray, np.ndarray]:
    """Apply an opaque split using party-owned training-derived threshold state."""
    handle = routing_state or self._histogram_routing_state
    if handle is None:
        raise RuntimeError("party-local histogram routing handle is unavailable")
    private_state = self._private_routing_state(handle)
    threshold = private_state.threshold_for(split_ref)
    return self._route_with_threshold(indices, split_ref.feature_ref, threshold)

route 

route(indices, feature_idx, threshold)

Direct local routing helper retained for controlled tests and diagnostics.

Source code in src/vertimosaic/parties/core.py 
336
337
338
339
340
def route(
    self, indices: np.ndarray, feature_idx: int, threshold: float
) -> tuple[np.ndarray, np.ndarray]:
    """Direct local routing helper retained for controlled tests and diagnostics."""
    return self._route_with_threshold(indices, feature_idx, threshold)

__init__ 

__init__(name, _x)

vertimosaic.parties.RemotePassiveParty dataclass

Metadata-only proxy for a passive party whose raw features are remote.

Source code in src/vertimosaic/parties/remote.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
@dataclass
class RemotePassiveParty:
    """Metadata-only proxy for a passive party whose raw features are remote."""

    name: str
    n_rows: int
    n_features: int
    transport: RemoteHTTPTransport
    coordinator_role: str = "bank"
    partition: str = "train"
    _histogram_routing_state: HistogramRoutingState | None = field(
        default=None,
        init=False,
        repr=False,
    )
    _active_signal_ref: str | None = field(default=None, init=False, repr=False)

    @classmethod
    def discover(
        cls,
        name: str,
        transport: RemoteHTTPTransport,
        *,
        coordinator_role: str = "bank",
        partition: str = "train",
    ) -> RemotePassiveParty:
        metadata = transport.send(
            {"partition": partition},
            message_type=_METADATA,
            sender_role=coordinator_role,
            receiver_role=name,
            direction="rpc",
            stage="discovery",
        )
        if not isinstance(metadata, dict):
            raise RuntimeError("remote party metadata response must be a mapping")
        if metadata.get("name") != name:
            raise RuntimeError("remote party metadata name does not match requested role")
        if metadata.get("partition") != partition:
            raise RuntimeError("remote party metadata partition does not match request")
        return cls(
            name=name,
            n_rows=int(metadata["n_rows"]),
            n_features=int(metadata["n_features"]),
            transport=transport,
            coordinator_role=coordinator_role,
            partition=partition,
        )

    def _rpc(self, message_type: str, payload: dict[str, Any], *, stage: str) -> Any:
        request = {"partition": self.partition, **payload}
        return self.transport.send(
            request,
            message_type=message_type,
            sender_role=self.coordinator_role,
            receiver_role=self.name,
            direction="rpc",
            stage=stage,
        )

    @property
    def histogram_bins_ready(self) -> bool:
        return self._histogram_routing_state is not None

    def local_logits(
        self,
        weights: np.ndarray,
        indices: np.ndarray | None = None,
    ) -> np.ndarray:
        delivered = self._rpc(
            _LOCAL_LOGITS,
            {"weights": np.asarray(weights, dtype=float), "indices": indices},
            stage="party_compute",
        )
        return np.asarray(delivered, dtype=float)

    def local_gradient(
        self,
        residual: np.ndarray,
        indices: np.ndarray | None = None,
    ) -> np.ndarray:
        delivered = self._rpc(
            _LOCAL_GRADIENT,
            {"residual": np.asarray(residual, dtype=float), "indices": indices},
            stage="party_compute",
        )
        return np.asarray(delivered, dtype=float)

    def prepare_histogram_bins(self, max_bins: int) -> None:
        state = self._rpc(
            _HIST_PREPARE,
            {"max_bins": int(max_bins)},
            stage="histogram_prepare",
        )
        if not isinstance(state, HistogramRoutingState):
            raise RuntimeError("remote histogram prepare did not return a routing handle")
        self._histogram_routing_state = state

    def export_histogram_routing_state(self) -> HistogramRoutingState:
        if self._histogram_routing_state is None:
            state = self._rpc(_HIST_EXPORT, {}, stage="histogram_state")
            if not isinstance(state, HistogramRoutingState):
                raise RuntimeError("remote histogram state response must be an opaque handle")
            self._histogram_routing_state = state
        return self._histogram_routing_state

    def share_histogram_routing_state_with(self, other: Any) -> None:
        if not isinstance(other, RemotePassiveParty):
            raise ValueError("remote routing state can only be shared with a remote partition")
        if other.name != self.name or other.n_features != self.n_features:
            raise ValueError(
                "remote routing state sharing requires the same party and feature width"
            )
        source_endpoint = self.transport.endpoints.get(self.name)
        target_endpoint = other.transport.endpoints.get(other.name)
        if source_endpoint != target_endpoint:
            raise ValueError("remote routing state sharing requires partitions on the same service")
        state = self._rpc(
            _HIST_SHARE,
            {"target_partition": other.partition},
            stage="histogram_state",
        )
        if not isinstance(state, HistogramRoutingState):
            raise RuntimeError("remote routing state sharing did not return an opaque handle")
        other._histogram_routing_state = state

    def set_gradient_hessian(
        self,
        gradients: np.ndarray,
        hessians: np.ndarray,
        *,
        signal_ref: str,
    ) -> None:
        response = self._rpc(
            _HIST_SIGNALS,
            {
                "signal_ref": signal_ref,
                "gradients": np.asarray(gradients, dtype=float),
                "hessians": np.asarray(hessians, dtype=float),
            },
            stage="histogram_signals",
        )
        if not isinstance(response, dict) or response.get("signal_ref") != signal_ref:
            raise RuntimeError("remote gradient/Hessian signal registration failed")
        self._active_signal_ref = signal_ref

    def clear_gradient_hessian(self, signal_ref: str | None = None) -> None:
        ref = signal_ref if signal_ref is not None else self._active_signal_ref
        self._rpc(
            _HIST_CLEAR_SIGNALS,
            {"signal_ref": ref},
            stage="histogram_signals",
        )
        if ref == self._active_signal_ref:
            self._active_signal_ref = None

    @staticmethod
    def _signal_reference(gradients: np.ndarray, hessians: np.ndarray) -> str:
        digest = hashlib.sha256()
        gradient_values = np.ascontiguousarray(gradients, dtype=float)
        hessian_values = np.ascontiguousarray(hessians, dtype=float)
        digest.update(gradient_values.shape[0].to_bytes(8, "big", signed=False))
        digest.update(gradient_values.tobytes())
        digest.update(hessian_values.tobytes())
        return f"round-{digest.hexdigest()[:24]}"

    def candidate_histograms(
        self,
        gradients: np.ndarray,
        hessians: np.ndarray,
        indices: np.ndarray,
        max_bins: int,
        min_samples_leaf: int,
        feature_indices: np.ndarray | None = None,
    ) -> list[HistogramCandidate]:
        gradient_values = np.asarray(gradients, dtype=float).reshape(-1)
        hessian_values = np.asarray(hessians, dtype=float).reshape(-1)
        signal_ref = self._signal_reference(gradient_values, hessian_values)
        if self._active_signal_ref != signal_ref:
            self.set_gradient_hessian(
                gradient_values,
                hessian_values,
                signal_ref=signal_ref,
            )
        delivered = self._rpc(
            _HIST_CANDIDATES,
            {
                "signal_ref": signal_ref,
                "indices": np.asarray(indices, dtype=int),
                "max_bins": int(max_bins),
                "min_samples_leaf": int(min_samples_leaf),
                "feature_indices": feature_indices,
            },
            stage="histogram_candidates",
        )
        if not isinstance(delivered, list):
            raise RuntimeError("remote histogram candidate response must be a list")
        return delivered

    def route_split(
        self,
        indices: np.ndarray,
        split_ref: OpaqueSplitReference,
        routing_state: HistogramRoutingState | None = None,
    ) -> tuple[np.ndarray, np.ndarray]:
        state = routing_state or self.export_histogram_routing_state()
        delivered = self._rpc(
            _HIST_ROUTE,
            {
                "indices": np.asarray(indices, dtype=int),
                "split_ref": split_ref,
                "routing_state": state,
            },
            stage="histogram_routing",
        )
        if not isinstance(delivered, tuple) or len(delivered) != 2:
            raise RuntimeError("remote histogram routing response must contain two partitions")
        left, right = delivered
        return np.asarray(left, dtype=int), np.asarray(right, dtype=int)

    def aggregate_local_split_importance(
        self,
        records: list[tuple[OpaqueSplitReference, float]],
    ) -> dict[int, dict[str, float | int]]:
        delivered = self._rpc(
            _HIST_IMPORTANCE,
            {"records": records},
            stage="histogram_importance",
        )
        if not isinstance(delivered, dict):
            raise RuntimeError("remote histogram importance response must be a mapping")
        return {int(key): value for key, value in delivered.items()}

name instance-attribute 

name

n_rows instance-attribute 

n_rows

n_features instance-attribute 

n_features

transport instance-attribute 

transport

coordinator_role class-attribute instance-attribute 

coordinator_role = 'bank'

partition class-attribute instance-attribute 

partition = 'train'

_histogram_routing_state class-attribute instance-attribute 

_histogram_routing_state = field(default=None, init=False, repr=False)

_active_signal_ref class-attribute instance-attribute 

_active_signal_ref = field(default=None, init=False, repr=False)

histogram_bins_ready property 

histogram_bins_ready

discover classmethod 

discover(name, transport, *, coordinator_role='bank', partition='train')
Source code in src/vertimosaic/parties/remote.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
@classmethod
def discover(
    cls,
    name: str,
    transport: RemoteHTTPTransport,
    *,
    coordinator_role: str = "bank",
    partition: str = "train",
) -> RemotePassiveParty:
    metadata = transport.send(
        {"partition": partition},
        message_type=_METADATA,
        sender_role=coordinator_role,
        receiver_role=name,
        direction="rpc",
        stage="discovery",
    )
    if not isinstance(metadata, dict):
        raise RuntimeError("remote party metadata response must be a mapping")
    if metadata.get("name") != name:
        raise RuntimeError("remote party metadata name does not match requested role")
    if metadata.get("partition") != partition:
        raise RuntimeError("remote party metadata partition does not match request")
    return cls(
        name=name,
        n_rows=int(metadata["n_rows"]),
        n_features=int(metadata["n_features"]),
        transport=transport,
        coordinator_role=coordinator_role,
        partition=partition,
    )

_rpc 

_rpc(message_type, payload, *, stage)
Source code in src/vertimosaic/parties/remote.py 
281
282
283
284
285
286
287
288
289
290
def _rpc(self, message_type: str, payload: dict[str, Any], *, stage: str) -> Any:
    request = {"partition": self.partition, **payload}
    return self.transport.send(
        request,
        message_type=message_type,
        sender_role=self.coordinator_role,
        receiver_role=self.name,
        direction="rpc",
        stage=stage,
    )

local_logits 

local_logits(weights, indices=None)
Source code in src/vertimosaic/parties/remote.py 
296
297
298
299
300
301
302
303
304
305
306
def local_logits(
    self,
    weights: np.ndarray,
    indices: np.ndarray | None = None,
) -> np.ndarray:
    delivered = self._rpc(
        _LOCAL_LOGITS,
        {"weights": np.asarray(weights, dtype=float), "indices": indices},
        stage="party_compute",
    )
    return np.asarray(delivered, dtype=float)

local_gradient 

local_gradient(residual, indices=None)
Source code in src/vertimosaic/parties/remote.py 
308
309
310
311
312
313
314
315
316
317
318
def local_gradient(
    self,
    residual: np.ndarray,
    indices: np.ndarray | None = None,
) -> np.ndarray:
    delivered = self._rpc(
        _LOCAL_GRADIENT,
        {"residual": np.asarray(residual, dtype=float), "indices": indices},
        stage="party_compute",
    )
    return np.asarray(delivered, dtype=float)

prepare_histogram_bins 

prepare_histogram_bins(max_bins)
Source code in src/vertimosaic/parties/remote.py 
320
321
322
323
324
325
326
327
328
def prepare_histogram_bins(self, max_bins: int) -> None:
    state = self._rpc(
        _HIST_PREPARE,
        {"max_bins": int(max_bins)},
        stage="histogram_prepare",
    )
    if not isinstance(state, HistogramRoutingState):
        raise RuntimeError("remote histogram prepare did not return a routing handle")
    self._histogram_routing_state = state

export_histogram_routing_state 

export_histogram_routing_state()
Source code in src/vertimosaic/parties/remote.py 
330
331
332
333
334
335
336
def export_histogram_routing_state(self) -> HistogramRoutingState:
    if self._histogram_routing_state is None:
        state = self._rpc(_HIST_EXPORT, {}, stage="histogram_state")
        if not isinstance(state, HistogramRoutingState):
            raise RuntimeError("remote histogram state response must be an opaque handle")
        self._histogram_routing_state = state
    return self._histogram_routing_state

share_histogram_routing_state_with 

share_histogram_routing_state_with(other)
Source code in src/vertimosaic/parties/remote.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def share_histogram_routing_state_with(self, other: Any) -> None:
    if not isinstance(other, RemotePassiveParty):
        raise ValueError("remote routing state can only be shared with a remote partition")
    if other.name != self.name or other.n_features != self.n_features:
        raise ValueError(
            "remote routing state sharing requires the same party and feature width"
        )
    source_endpoint = self.transport.endpoints.get(self.name)
    target_endpoint = other.transport.endpoints.get(other.name)
    if source_endpoint != target_endpoint:
        raise ValueError("remote routing state sharing requires partitions on the same service")
    state = self._rpc(
        _HIST_SHARE,
        {"target_partition": other.partition},
        stage="histogram_state",
    )
    if not isinstance(state, HistogramRoutingState):
        raise RuntimeError("remote routing state sharing did not return an opaque handle")
    other._histogram_routing_state = state

set_gradient_hessian 

set_gradient_hessian(gradients, hessians, *, signal_ref)
Source code in src/vertimosaic/parties/remote.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def set_gradient_hessian(
    self,
    gradients: np.ndarray,
    hessians: np.ndarray,
    *,
    signal_ref: str,
) -> None:
    response = self._rpc(
        _HIST_SIGNALS,
        {
            "signal_ref": signal_ref,
            "gradients": np.asarray(gradients, dtype=float),
            "hessians": np.asarray(hessians, dtype=float),
        },
        stage="histogram_signals",
    )
    if not isinstance(response, dict) or response.get("signal_ref") != signal_ref:
        raise RuntimeError("remote gradient/Hessian signal registration failed")
    self._active_signal_ref = signal_ref

clear_gradient_hessian 

clear_gradient_hessian(signal_ref=None)
Source code in src/vertimosaic/parties/remote.py 
378
379
380
381
382
383
384
385
386
def clear_gradient_hessian(self, signal_ref: str | None = None) -> None:
    ref = signal_ref if signal_ref is not None else self._active_signal_ref
    self._rpc(
        _HIST_CLEAR_SIGNALS,
        {"signal_ref": ref},
        stage="histogram_signals",
    )
    if ref == self._active_signal_ref:
        self._active_signal_ref = None

_signal_reference staticmethod 

_signal_reference(gradients, hessians)
Source code in src/vertimosaic/parties/remote.py 
388
389
390
391
392
393
394
395
396
@staticmethod
def _signal_reference(gradients: np.ndarray, hessians: np.ndarray) -> str:
    digest = hashlib.sha256()
    gradient_values = np.ascontiguousarray(gradients, dtype=float)
    hessian_values = np.ascontiguousarray(hessians, dtype=float)
    digest.update(gradient_values.shape[0].to_bytes(8, "big", signed=False))
    digest.update(gradient_values.tobytes())
    digest.update(hessian_values.tobytes())
    return f"round-{digest.hexdigest()[:24]}"

candidate_histograms 

candidate_histograms(gradients, hessians, indices, max_bins, min_samples_leaf, feature_indices=None)
Source code in src/vertimosaic/parties/remote.py 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
def candidate_histograms(
    self,
    gradients: np.ndarray,
    hessians: np.ndarray,
    indices: np.ndarray,
    max_bins: int,
    min_samples_leaf: int,
    feature_indices: np.ndarray | None = None,
) -> list[HistogramCandidate]:
    gradient_values = np.asarray(gradients, dtype=float).reshape(-1)
    hessian_values = np.asarray(hessians, dtype=float).reshape(-1)
    signal_ref = self._signal_reference(gradient_values, hessian_values)
    if self._active_signal_ref != signal_ref:
        self.set_gradient_hessian(
            gradient_values,
            hessian_values,
            signal_ref=signal_ref,
        )
    delivered = self._rpc(
        _HIST_CANDIDATES,
        {
            "signal_ref": signal_ref,
            "indices": np.asarray(indices, dtype=int),
            "max_bins": int(max_bins),
            "min_samples_leaf": int(min_samples_leaf),
            "feature_indices": feature_indices,
        },
        stage="histogram_candidates",
    )
    if not isinstance(delivered, list):
        raise RuntimeError("remote histogram candidate response must be a list")
    return delivered

route_split 

route_split(indices, split_ref, routing_state=None)
Source code in src/vertimosaic/parties/remote.py 
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
def route_split(
    self,
    indices: np.ndarray,
    split_ref: OpaqueSplitReference,
    routing_state: HistogramRoutingState | None = None,
) -> tuple[np.ndarray, np.ndarray]:
    state = routing_state or self.export_histogram_routing_state()
    delivered = self._rpc(
        _HIST_ROUTE,
        {
            "indices": np.asarray(indices, dtype=int),
            "split_ref": split_ref,
            "routing_state": state,
        },
        stage="histogram_routing",
    )
    if not isinstance(delivered, tuple) or len(delivered) != 2:
        raise RuntimeError("remote histogram routing response must contain two partitions")
    left, right = delivered
    return np.asarray(left, dtype=int), np.asarray(right, dtype=int)

aggregate_local_split_importance 

aggregate_local_split_importance(records)
Source code in src/vertimosaic/parties/remote.py 
452
453
454
455
456
457
458
459
460
461
462
463
def aggregate_local_split_importance(
    self,
    records: list[tuple[OpaqueSplitReference, float]],
) -> dict[int, dict[str, float | int]]:
    delivered = self._rpc(
        _HIST_IMPORTANCE,
        {"records": records},
        stage="histogram_importance",
    )
    if not isinstance(delivered, dict):
        raise RuntimeError("remote histogram importance response must be a mapping")
    return {int(key): value for key, value in delivered.items()}

__init__ 

__init__(name, n_rows, n_features, transport, coordinator_role='bank', partition='train')

vertimosaic.parties.RemotePartyService dataclass

Expose party-local VFL computations without moving raw feature matrices.

One service can own multiple row partitions for the same organization, typically train and validation. Histogram thresholds, binned matrices, gradient/ Hessian round state, and routing decisions stay inside this service boundary. The coordinator receives only metadata, aggregate split statistics, opaque split references, opaque routing handles, and routed row indices.

Source code in src/vertimosaic/parties/remote.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
@dataclass
class RemotePartyService:
    """Expose party-local VFL computations without moving raw feature matrices.

    One service can own multiple row partitions for the same organization, typically
    ``train`` and ``validation``. Histogram thresholds, binned matrices, gradient/
    Hessian round state, and routing decisions stay inside this service boundary.
    The coordinator receives only metadata, aggregate split statistics, opaque split
    references, opaque routing handles, and routed row indices.
    """

    party: PassiveParty
    allowed_senders: set[str] | None = None
    partitions: dict[str, PassiveParty] = field(default_factory=dict)
    _signal_cache: dict[tuple[str, str], tuple[np.ndarray, np.ndarray]] = field(
        default_factory=dict,
        init=False,
        repr=False,
    )

    def __post_init__(self) -> None:
        normalized = dict(self.partitions)
        normalized.setdefault("train", self.party)
        for partition, candidate in normalized.items():
            if not partition:
                raise ValueError("remote party partition names must be non-empty")
            if candidate.name != self.party.name:
                raise ValueError("all remote service partitions must belong to the same party")
            if candidate.n_features != self.party.n_features:
                raise ValueError("all remote service partitions must have equal feature width")
        self.partitions = normalized

    def _party_for(self, payload: Any) -> tuple[str, PassiveParty]:
        partition = "train"
        if isinstance(payload, dict) and "partition" in payload:
            partition = str(payload["partition"])
        party = self.partitions.get(partition)
        if party is None:
            raise ValueError(f"unknown remote party partition: {partition}")
        return partition, party

    @staticmethod
    def _mapping(payload: Any, operation: str) -> dict[str, Any]:
        if not isinstance(payload, dict):
            raise ValueError(f"{operation} expects a mapping payload")
        return payload

    @staticmethod
    def _pair(payload: Any, operation: str) -> tuple[Any, Any]:
        if not isinstance(payload, tuple) or len(payload) != 2:
            raise ValueError(f"{operation} expects a two-item tuple payload")
        return payload[0], payload[1]

    @staticmethod
    def _indices(value: Any) -> np.ndarray | None:
        if value is None:
            return None
        return np.asarray(value, dtype=int).reshape(-1)

    def handle(self, message_type: str, sender_role: str, payload: Any) -> Any:
        if self.allowed_senders is not None and sender_role not in self.allowed_senders:
            raise ValueError(f"sender role {sender_role!r} is not authorized for this party")
        partition, party = self._party_for(payload)

        if message_type == _METADATA:
            return {
                "name": party.name,
                "partition": partition,
                "n_rows": party.n_rows,
                "n_features": party.n_features,
            }
        if message_type == _LOCAL_LOGITS:
            if isinstance(payload, dict):
                request = self._mapping(payload, _LOCAL_LOGITS)
                weights = request.get("weights")
                indices = request.get("indices")
            else:
                weights, indices = self._pair(payload, _LOCAL_LOGITS)
            return party.local_logits(
                np.asarray(weights, dtype=float),
                self._indices(indices),
            )
        if message_type == _LOCAL_GRADIENT:
            if isinstance(payload, dict):
                request = self._mapping(payload, _LOCAL_GRADIENT)
                residual = request.get("residual")
                indices = request.get("indices")
            else:
                residual, indices = self._pair(payload, _LOCAL_GRADIENT)
            return party.local_gradient(
                np.asarray(residual, dtype=float),
                self._indices(indices),
            )
        if message_type == _HIST_PREPARE:
            request = self._mapping(payload, _HIST_PREPARE)
            max_bins = int(request["max_bins"])
            party.prepare_histogram_bins(max_bins)
            return party.export_histogram_routing_state()
        if message_type == _HIST_EXPORT:
            return party.export_histogram_routing_state()
        if message_type == _HIST_SIGNALS:
            request = self._mapping(payload, _HIST_SIGNALS)
            registered_signal_ref = str(request["signal_ref"])
            gradients = np.asarray(request["gradients"], dtype=float).reshape(-1)
            hessians = np.asarray(request["hessians"], dtype=float).reshape(-1)
            if gradients.shape != hessians.shape or gradients.size != party.n_rows:
                raise ValueError("gradient/Hessian signals must match the remote partition rows")
            if not np.isfinite(gradients).all() or not np.isfinite(hessians).all():
                raise ValueError("gradient/Hessian signals must be finite")
            for key in [key for key in self._signal_cache if key[0] == partition]:
                self._signal_cache.pop(key, None)
            self._signal_cache[(partition, registered_signal_ref)] = (
                gradients.copy(),
                hessians.copy(),
            )
            return {"signal_ref": registered_signal_ref, "rows": int(gradients.size)}
        if message_type == _HIST_CLEAR_SIGNALS:
            request = self._mapping(payload, _HIST_CLEAR_SIGNALS)
            clear_signal_ref = request.get("signal_ref")
            if clear_signal_ref is None:
                keys = [key for key in self._signal_cache if key[0] == partition]
                for key in keys:
                    self._signal_cache.pop(key, None)
            else:
                self._signal_cache.pop((partition, str(clear_signal_ref)), None)
            return None
        if message_type == _HIST_CANDIDATES:
            request = self._mapping(payload, _HIST_CANDIDATES)
            candidate_signal_ref = request.get("signal_ref")
            if candidate_signal_ref is not None:
                cached = self._signal_cache.get((partition, str(candidate_signal_ref)))
                if cached is None:
                    raise ValueError("unknown or expired gradient/Hessian signal reference")
                gradients, hessians = cached
            else:
                gradients = np.asarray(request["gradients"], dtype=float).reshape(-1)
                hessians = np.asarray(request["hessians"], dtype=float).reshape(-1)
            feature_indices = request.get("feature_indices")
            return party.candidate_histograms(
                gradients,
                hessians,
                np.asarray(request["indices"], dtype=int).reshape(-1),
                int(request["max_bins"]),
                int(request["min_samples_leaf"]),
                None
                if feature_indices is None
                else np.asarray(feature_indices, dtype=int).reshape(-1),
            )
        if message_type == _HIST_ROUTE:
            request = self._mapping(payload, _HIST_ROUTE)
            split_ref = request.get("split_ref")
            routing_state = request.get("routing_state")
            if not isinstance(split_ref, OpaqueSplitReference):
                raise ValueError("histogram routing requires an opaque split reference")
            if routing_state is not None and not isinstance(routing_state, HistogramRoutingState):
                raise ValueError("histogram routing state must be an opaque routing handle")
            return party.route_split(
                np.asarray(request["indices"], dtype=int).reshape(-1),
                split_ref,
                routing_state,
            )
        if message_type == _HIST_SHARE:
            request = self._mapping(payload, _HIST_SHARE)
            target_partition = str(request["target_partition"])
            target = self.partitions.get(target_partition)
            if target is None:
                raise ValueError(f"unknown target remote party partition: {target_partition}")
            party.share_histogram_routing_state_with(target)
            return target.export_histogram_routing_state()
        if message_type == _HIST_IMPORTANCE:
            request = self._mapping(payload, _HIST_IMPORTANCE)
            records_raw = request.get("records")
            if not isinstance(records_raw, list):
                raise ValueError("histogram importance records must be a list")
            records: list[tuple[OpaqueSplitReference, float]] = []
            for item in records_raw:
                if not isinstance(item, tuple) or len(item) != 2:
                    raise ValueError("histogram importance records must be two-item tuples")
                split_ref, gain = item
                if not isinstance(split_ref, OpaqueSplitReference):
                    raise ValueError("histogram importance requires opaque split references")
                records.append((split_ref, float(gain)))
            return party.aggregate_local_split_importance(records)
        raise ValueError(f"unsupported remote party operation: {message_type}")

    def make_server(
        self,
        address: tuple[str, int],
        *,
        bearer_token: str | None = None,
        path: str = "/v1/messages",
    ) -> ReferenceRelayServer:
        return ReferenceRelayServer(
            address,
            receiver_role=self.party.name,
            bearer_token=bearer_token,
            path=path,
            request_handler=self.handle,
        )

party instance-attribute 

party

allowed_senders class-attribute instance-attribute 

allowed_senders = None

partitions class-attribute instance-attribute 

partitions = field(default_factory=dict)

_signal_cache class-attribute instance-attribute 

_signal_cache = field(default_factory=dict, init=False, repr=False)

__post_init__ 

__post_init__()
Source code in src/vertimosaic/parties/remote.py 
51
52
53
54
55
56
57
58
59
60
61
def __post_init__(self) -> None:
    normalized = dict(self.partitions)
    normalized.setdefault("train", self.party)
    for partition, candidate in normalized.items():
        if not partition:
            raise ValueError("remote party partition names must be non-empty")
        if candidate.name != self.party.name:
            raise ValueError("all remote service partitions must belong to the same party")
        if candidate.n_features != self.party.n_features:
            raise ValueError("all remote service partitions must have equal feature width")
    self.partitions = normalized

_party_for 

_party_for(payload)
Source code in src/vertimosaic/parties/remote.py 
63
64
65
66
67
68
69
70
def _party_for(self, payload: Any) -> tuple[str, PassiveParty]:
    partition = "train"
    if isinstance(payload, dict) and "partition" in payload:
        partition = str(payload["partition"])
    party = self.partitions.get(partition)
    if party is None:
        raise ValueError(f"unknown remote party partition: {partition}")
    return partition, party

_mapping staticmethod 

_mapping(payload, operation)
Source code in src/vertimosaic/parties/remote.py 
72
73
74
75
76
@staticmethod
def _mapping(payload: Any, operation: str) -> dict[str, Any]:
    if not isinstance(payload, dict):
        raise ValueError(f"{operation} expects a mapping payload")
    return payload

_pair staticmethod 

_pair(payload, operation)
Source code in src/vertimosaic/parties/remote.py 
78
79
80
81
82
@staticmethod
def _pair(payload: Any, operation: str) -> tuple[Any, Any]:
    if not isinstance(payload, tuple) or len(payload) != 2:
        raise ValueError(f"{operation} expects a two-item tuple payload")
    return payload[0], payload[1]

_indices staticmethod 

_indices(value)
Source code in src/vertimosaic/parties/remote.py 
84
85
86
87
88
@staticmethod
def _indices(value: Any) -> np.ndarray | None:
    if value is None:
        return None
    return np.asarray(value, dtype=int).reshape(-1)

handle 

handle(message_type, sender_role, payload)
Source code in src/vertimosaic/parties/remote.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def handle(self, message_type: str, sender_role: str, payload: Any) -> Any:
    if self.allowed_senders is not None and sender_role not in self.allowed_senders:
        raise ValueError(f"sender role {sender_role!r} is not authorized for this party")
    partition, party = self._party_for(payload)

    if message_type == _METADATA:
        return {
            "name": party.name,
            "partition": partition,
            "n_rows": party.n_rows,
            "n_features": party.n_features,
        }
    if message_type == _LOCAL_LOGITS:
        if isinstance(payload, dict):
            request = self._mapping(payload, _LOCAL_LOGITS)
            weights = request.get("weights")
            indices = request.get("indices")
        else:
            weights, indices = self._pair(payload, _LOCAL_LOGITS)
        return party.local_logits(
            np.asarray(weights, dtype=float),
            self._indices(indices),
        )
    if message_type == _LOCAL_GRADIENT:
        if isinstance(payload, dict):
            request = self._mapping(payload, _LOCAL_GRADIENT)
            residual = request.get("residual")
            indices = request.get("indices")
        else:
            residual, indices = self._pair(payload, _LOCAL_GRADIENT)
        return party.local_gradient(
            np.asarray(residual, dtype=float),
            self._indices(indices),
        )
    if message_type == _HIST_PREPARE:
        request = self._mapping(payload, _HIST_PREPARE)
        max_bins = int(request["max_bins"])
        party.prepare_histogram_bins(max_bins)
        return party.export_histogram_routing_state()
    if message_type == _HIST_EXPORT:
        return party.export_histogram_routing_state()
    if message_type == _HIST_SIGNALS:
        request = self._mapping(payload, _HIST_SIGNALS)
        registered_signal_ref = str(request["signal_ref"])
        gradients = np.asarray(request["gradients"], dtype=float).reshape(-1)
        hessians = np.asarray(request["hessians"], dtype=float).reshape(-1)
        if gradients.shape != hessians.shape or gradients.size != party.n_rows:
            raise ValueError("gradient/Hessian signals must match the remote partition rows")
        if not np.isfinite(gradients).all() or not np.isfinite(hessians).all():
            raise ValueError("gradient/Hessian signals must be finite")
        for key in [key for key in self._signal_cache if key[0] == partition]:
            self._signal_cache.pop(key, None)
        self._signal_cache[(partition, registered_signal_ref)] = (
            gradients.copy(),
            hessians.copy(),
        )
        return {"signal_ref": registered_signal_ref, "rows": int(gradients.size)}
    if message_type == _HIST_CLEAR_SIGNALS:
        request = self._mapping(payload, _HIST_CLEAR_SIGNALS)
        clear_signal_ref = request.get("signal_ref")
        if clear_signal_ref is None:
            keys = [key for key in self._signal_cache if key[0] == partition]
            for key in keys:
                self._signal_cache.pop(key, None)
        else:
            self._signal_cache.pop((partition, str(clear_signal_ref)), None)
        return None
    if message_type == _HIST_CANDIDATES:
        request = self._mapping(payload, _HIST_CANDIDATES)
        candidate_signal_ref = request.get("signal_ref")
        if candidate_signal_ref is not None:
            cached = self._signal_cache.get((partition, str(candidate_signal_ref)))
            if cached is None:
                raise ValueError("unknown or expired gradient/Hessian signal reference")
            gradients, hessians = cached
        else:
            gradients = np.asarray(request["gradients"], dtype=float).reshape(-1)
            hessians = np.asarray(request["hessians"], dtype=float).reshape(-1)
        feature_indices = request.get("feature_indices")
        return party.candidate_histograms(
            gradients,
            hessians,
            np.asarray(request["indices"], dtype=int).reshape(-1),
            int(request["max_bins"]),
            int(request["min_samples_leaf"]),
            None
            if feature_indices is None
            else np.asarray(feature_indices, dtype=int).reshape(-1),
        )
    if message_type == _HIST_ROUTE:
        request = self._mapping(payload, _HIST_ROUTE)
        split_ref = request.get("split_ref")
        routing_state = request.get("routing_state")
        if not isinstance(split_ref, OpaqueSplitReference):
            raise ValueError("histogram routing requires an opaque split reference")
        if routing_state is not None and not isinstance(routing_state, HistogramRoutingState):
            raise ValueError("histogram routing state must be an opaque routing handle")
        return party.route_split(
            np.asarray(request["indices"], dtype=int).reshape(-1),
            split_ref,
            routing_state,
        )
    if message_type == _HIST_SHARE:
        request = self._mapping(payload, _HIST_SHARE)
        target_partition = str(request["target_partition"])
        target = self.partitions.get(target_partition)
        if target is None:
            raise ValueError(f"unknown target remote party partition: {target_partition}")
        party.share_histogram_routing_state_with(target)
        return target.export_histogram_routing_state()
    if message_type == _HIST_IMPORTANCE:
        request = self._mapping(payload, _HIST_IMPORTANCE)
        records_raw = request.get("records")
        if not isinstance(records_raw, list):
            raise ValueError("histogram importance records must be a list")
        records: list[tuple[OpaqueSplitReference, float]] = []
        for item in records_raw:
            if not isinstance(item, tuple) or len(item) != 2:
                raise ValueError("histogram importance records must be two-item tuples")
            split_ref, gain = item
            if not isinstance(split_ref, OpaqueSplitReference):
                raise ValueError("histogram importance requires opaque split references")
            records.append((split_ref, float(gain)))
        return party.aggregate_local_split_importance(records)
    raise ValueError(f"unsupported remote party operation: {message_type}")

make_server 

make_server(address, *, bearer_token=None, path='/v1/messages')
Source code in src/vertimosaic/parties/remote.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def make_server(
    self,
    address: tuple[str, int],
    *,
    bearer_token: str | None = None,
    path: str = "/v1/messages",
) -> ReferenceRelayServer:
    return ReferenceRelayServer(
        address,
        receiver_role=self.party.name,
        bearer_token=bearer_token,
        path=path,
        request_handler=self.handle,
    )

__init__ 

__init__(party, allowed_senders=None, partitions=dict())

Transport

vertimosaic.transport.InMemoryTransport dataclass

Ephemeral Message transport with metadata-only persistent auditing.

Every protocol send creates a :class:Message carrying the ephemeral payload. The transport returns that message payload to the protocol caller but persists only an :class:AuditEvent, so row-level values are never retained in the audit log.

Source code in src/vertimosaic/transport/core.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@dataclass
class InMemoryTransport:
    """Ephemeral Message transport with metadata-only persistent auditing.

    Every protocol ``send`` creates a :class:`Message` carrying the ephemeral payload.
    The transport returns that message payload to the protocol caller but persists only
    an :class:`AuditEvent`, so row-level values are never retained in the audit log.
    """

    audit_log: list[AuditEvent] = field(default_factory=list)

    def send(
        self,
        payload: Any,
        *,
        message_type: str,
        sender_role: str,
        receiver_role: str,
        direction: str | None = None,
        stage: str | None = None,
        step: int | None = None,
    ) -> Any:
        if isinstance(payload, StructuredPayload):
            wire_value = payload.value
            shape = payload.shape
            count = payload.scalar_count
            size = payload.estimated_bytes
        else:
            wire_value = payload
            shape, count, size = self._metadata(payload)
        message = Message(
            message_type,
            sender_role,
            receiver_role,
            shape,
            count,
            size,
            direction,
            stage,
            step,
            wire_value,
        )
        self.audit_log.append(AuditEvent.from_message(message))
        return message.payload

    @staticmethod
    def _metadata(payload: Any) -> tuple[tuple[int, ...] | None, int, int]:
        if isinstance(payload, np.ndarray):
            return tuple(payload.shape), int(payload.size), int(payload.nbytes)
        if np.isscalar(payload):
            array = np.asarray(payload)
            return tuple(array.shape), 1, int(array.nbytes)
        if isinstance(payload, (tuple, list)) and all(np.isscalar(value) for value in payload):
            array = np.asarray(payload)
            return tuple(array.shape), int(array.size), int(array.nbytes)
        return None, 0, 0

    @property
    def estimated_payload_bytes(self) -> int:
        return sum(event.estimated_bytes for event in self.audit_log)

audit_log class-attribute instance-attribute 

audit_log = field(default_factory=list)

estimated_payload_bytes property 

estimated_payload_bytes

send 

send(payload, *, message_type, sender_role, receiver_role, direction=None, stage=None, step=None)
Source code in src/vertimosaic/transport/core.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def send(
    self,
    payload: Any,
    *,
    message_type: str,
    sender_role: str,
    receiver_role: str,
    direction: str | None = None,
    stage: str | None = None,
    step: int | None = None,
) -> Any:
    if isinstance(payload, StructuredPayload):
        wire_value = payload.value
        shape = payload.shape
        count = payload.scalar_count
        size = payload.estimated_bytes
    else:
        wire_value = payload
        shape, count, size = self._metadata(payload)
    message = Message(
        message_type,
        sender_role,
        receiver_role,
        shape,
        count,
        size,
        direction,
        stage,
        step,
        wire_value,
    )
    self.audit_log.append(AuditEvent.from_message(message))
    return message.payload

_metadata staticmethod 

_metadata(payload)
Source code in src/vertimosaic/transport/core.py 
139
140
141
142
143
144
145
146
147
148
149
@staticmethod
def _metadata(payload: Any) -> tuple[tuple[int, ...] | None, int, int]:
    if isinstance(payload, np.ndarray):
        return tuple(payload.shape), int(payload.size), int(payload.nbytes)
    if np.isscalar(payload):
        array = np.asarray(payload)
        return tuple(array.shape), 1, int(array.nbytes)
    if isinstance(payload, (tuple, list)) and all(np.isscalar(value) for value in payload):
        array = np.asarray(payload)
        return tuple(array.shape), int(array.size), int(array.nbytes)
    return None, 0, 0

__init__ 

__init__(audit_log=list())

vertimosaic.transport.RemoteHTTPTransport dataclass

Bases: InMemoryTransport

Reference network transport with bounded replay and application integrity controls.

HTTPS remains required by default. Optional HMAC signing authenticates the exact serialized request/response body and supports key rotation through explicit key IDs. These controls improve transport integrity/availability; they do not make residuals, gradients, Hessians or routing messages cryptographically private.

Source code in src/vertimosaic/transport/remote.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
@dataclass
class RemoteHTTPTransport(InMemoryTransport):
    """Reference network transport with bounded replay and application integrity controls.

    HTTPS remains required by default. Optional HMAC signing authenticates the exact
    serialized request/response body and supports key rotation through explicit key IDs.
    These controls improve transport integrity/availability; they do not make residuals,
    gradients, Hessians or routing messages cryptographically private.
    """

    endpoints: dict[str, str] = field(default_factory=dict)
    bearer_tokens: dict[str, str] = field(default_factory=dict, repr=False)
    signing_keys: dict[str, tuple[str, bytes]] = field(default_factory=dict, repr=False)
    ca_file: str | None = None
    client_cert: str | None = None
    client_key: str | None = None
    timeout_seconds: float = 10.0
    max_retries: int = 2
    backoff_seconds: float = 0.25
    allow_insecure_http: bool = False
    array_codec: str = "json"
    network_audit_log: list[RemoteNetworkEvent] = field(default_factory=list)

    def __post_init__(self) -> None:
        if self.timeout_seconds <= 0:
            raise ValueError("timeout_seconds must be positive")
        if self.max_retries < 0:
            raise ValueError("max_retries must be non-negative")
        if self.backoff_seconds < 0:
            raise ValueError("backoff_seconds must be non-negative")
        if self.array_codec not in _ARRAY_CODECS:
            raise ValueError(f"array_codec must be one of {sorted(_ARRAY_CODECS)}")
        for role, endpoint in self.endpoints.items():
            if endpoint.startswith("https://"):
                continue
            if endpoint.startswith("http://") and self.allow_insecure_http:
                continue
            raise ValueError(
                f"remote endpoint for {role} must use HTTPS; "
                "allow_insecure_http is intended only for local tests"
            )
        for role, (key_id, secret) in self.signing_keys.items():
            if not key_id or len(secret) < 16:
                raise ValueError(f"signing key for {role} needs a key ID and >=16-byte secret")

    def _ssl_context(self) -> ssl.SSLContext | None:
        if not any(endpoint.startswith("https://") for endpoint in self.endpoints.values()):
            return None
        context = ssl.create_default_context(cafile=self.ca_file)
        if self.client_cert is not None:
            context.load_cert_chain(self.client_cert, keyfile=self.client_key)
        return context

    def network_totals(self) -> dict[str, float | int | bool | str]:
        events = self.network_audit_log
        latencies = np.asarray([event.round_trip_seconds for event in events], dtype=float)
        return {
            "message_count": len(events),
            "request_body_bytes": sum(event.request_body_bytes for event in events),
            "request_header_bytes": sum(event.request_header_bytes for event in events),
            "response_body_bytes": sum(event.response_body_bytes for event in events),
            "response_header_bytes": sum(event.response_header_bytes for event in events),
            "measured_application_bytes": sum(event.measured_application_bytes for event in events),
            "round_trip_p50_seconds": (
                float(np.percentile(latencies, 50)) if len(latencies) else 0.0
            ),
            "round_trip_p95_seconds": (
                float(np.percentile(latencies, 95)) if len(latencies) else 0.0
            ),
            "tls_record_bytes_measured": False,
            "scope": (
                "serialized HTTP application bytes and round-trip latency; excludes TLS record, "
                "TCP/IP framing and retransmissions not exposed by urllib"
            ),
        }

    def send(
        self,
        payload: Any,
        *,
        message_type: str,
        sender_role: str,
        receiver_role: str,
        direction: str | None = None,
        stage: str | None = None,
        step: int | None = None,
    ) -> Any:
        endpoint = self.endpoints.get(receiver_role)
        if endpoint is None:
            raise ValueError(f"no remote endpoint configured for receiver role {receiver_role}")

        if isinstance(payload, StructuredPayload):
            wire_value = payload.value
            shape = payload.shape
            count = payload.scalar_count
            size = payload.estimated_bytes
        else:
            wire_value = payload
            shape, count, size = self._metadata(payload)

        message_id = uuid4().hex
        envelope = {
            "schema_version": SCHEMA_VERSION,
            "message_id": message_id,
            "message_type": message_type,
            "sender_role": sender_role,
            "receiver_role": receiver_role,
            "direction": direction,
            "stage": stage,
            "step": step,
            "sent_at_unix": time.time(),
            "nonce": uuid4().hex,
            "array_codec": self.array_codec,
            "payload": _encode_value(wire_value, array_codec=self.array_codec),
        }
        body = json.dumps(envelope, separators=(",", ":")).encode()
        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "X-VertiMosaic-Schema": str(SCHEMA_VERSION),
            "X-VertiMosaic-Message-ID": message_id,
            "X-VertiMosaic-Sender": sender_role,
        }
        token = self.bearer_tokens.get(receiver_role)
        if token:
            headers["Authorization"] = f"Bearer {token}"
        signing = self.signing_keys.get(receiver_role)
        if signing is not None:
            key_id, secret = signing
            headers["X-VertiMosaic-Key-ID"] = key_id
            headers["X-VertiMosaic-Signature"] = _sign(secret, body)

        context = self._ssl_context()
        last_error: Exception | None = None
        response_payload: Any = None
        for attempt in range(self.max_retries + 1):
            request = Request(endpoint, data=body, headers=headers, method="POST")
            started = time.perf_counter()
            try:
                with urlopen(  # nosec B310 - endpoint schemes are restricted above
                    request,
                    timeout=self.timeout_seconds,
                    context=context,
                ) as response:
                    response_body = response.read()
                    elapsed = time.perf_counter() - started
                    response_headers = [(str(k), str(v)) for k, v in response.headers.items()]
                    request_headers = [(str(k), str(v)) for k, v in request.header_items()]
                    self.network_audit_log.append(
                        RemoteNetworkEvent(
                            message_id=message_id,
                            request_body_bytes=len(body),
                            request_header_bytes=_header_bytes(request_headers),
                            response_body_bytes=len(response_body),
                            response_header_bytes=_header_bytes(
                                response_headers,
                                status_line_bytes=17,
                            ),
                            round_trip_seconds=elapsed,
                            status_code=int(response.status),
                            attempt=attempt + 1,
                        )
                    )
                    if signing is not None:
                        key_id, secret = signing
                        response_key_id = response.headers.get("X-VertiMosaic-Key-ID")
                        response_signature = response.headers.get("X-VertiMosaic-Signature")
                        if response_key_id != key_id or response_signature is None:
                            raise RuntimeError(
                                "signed request received an unsigned remote response"
                            )
                        if not hmac.compare_digest(
                            response_signature,
                            _sign(secret, response_body),
                        ):
                            raise RuntimeError("remote response signature mismatch")
                    decoded = json.loads(response_body.decode())
                    if decoded.get("schema_version") != SCHEMA_VERSION:
                        raise RuntimeError("remote response schema version mismatch")
                    if decoded.get("message_id") != message_id:
                        raise RuntimeError("remote response message ID mismatch")
                    if "payload" not in decoded:
                        raise RuntimeError("remote response is missing payload")
                    response_payload = _decode_value(decoded["payload"])
                    break
            except (HTTPError, URLError, TimeoutError, OSError, json.JSONDecodeError) as exc:
                last_error = exc
                if attempt >= self.max_retries:
                    raise RuntimeError(
                        f"remote delivery to {receiver_role} failed after {attempt + 1} attempts"
                    ) from exc
                time.sleep(self.backoff_seconds * (2**attempt))
        else:
            raise RuntimeError(
                "remote transport retry loop terminated unexpectedly"
            ) from last_error

        message = Message(
            message_type=message_type,
            sender_role=sender_role,
            receiver_role=receiver_role,
            shape=shape,
            scalar_count=count,
            estimated_bytes=size,
            direction=direction,
            stage=stage,
            step=step,
            payload=None,
        )
        self.audit_log.append(AuditEvent.from_message(message))
        return response_payload

endpoints class-attribute instance-attribute 

endpoints = field(default_factory=dict)

bearer_tokens class-attribute instance-attribute 

bearer_tokens = field(default_factory=dict, repr=False)

signing_keys class-attribute instance-attribute 

signing_keys = field(default_factory=dict, repr=False)

ca_file class-attribute instance-attribute 

ca_file = None

client_cert class-attribute instance-attribute 

client_cert = None

client_key class-attribute instance-attribute 

client_key = None

timeout_seconds class-attribute instance-attribute 

timeout_seconds = 10.0

max_retries class-attribute instance-attribute 

max_retries = 2

backoff_seconds class-attribute instance-attribute 

backoff_seconds = 0.25

allow_insecure_http class-attribute instance-attribute 

allow_insecure_http = False

array_codec class-attribute instance-attribute 

array_codec = 'json'

network_audit_log class-attribute instance-attribute 

network_audit_log = field(default_factory=list)

__post_init__ 

__post_init__()
Source code in src/vertimosaic/transport/remote.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def __post_init__(self) -> None:
    if self.timeout_seconds <= 0:
        raise ValueError("timeout_seconds must be positive")
    if self.max_retries < 0:
        raise ValueError("max_retries must be non-negative")
    if self.backoff_seconds < 0:
        raise ValueError("backoff_seconds must be non-negative")
    if self.array_codec not in _ARRAY_CODECS:
        raise ValueError(f"array_codec must be one of {sorted(_ARRAY_CODECS)}")
    for role, endpoint in self.endpoints.items():
        if endpoint.startswith("https://"):
            continue
        if endpoint.startswith("http://") and self.allow_insecure_http:
            continue
        raise ValueError(
            f"remote endpoint for {role} must use HTTPS; "
            "allow_insecure_http is intended only for local tests"
        )
    for role, (key_id, secret) in self.signing_keys.items():
        if not key_id or len(secret) < 16:
            raise ValueError(f"signing key for {role} needs a key ID and >=16-byte secret")

_ssl_context 

_ssl_context()
Source code in src/vertimosaic/transport/remote.py 
239
240
241
242
243
244
245
def _ssl_context(self) -> ssl.SSLContext | None:
    if not any(endpoint.startswith("https://") for endpoint in self.endpoints.values()):
        return None
    context = ssl.create_default_context(cafile=self.ca_file)
    if self.client_cert is not None:
        context.load_cert_chain(self.client_cert, keyfile=self.client_key)
    return context

network_totals 

network_totals()
Source code in src/vertimosaic/transport/remote.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def network_totals(self) -> dict[str, float | int | bool | str]:
    events = self.network_audit_log
    latencies = np.asarray([event.round_trip_seconds for event in events], dtype=float)
    return {
        "message_count": len(events),
        "request_body_bytes": sum(event.request_body_bytes for event in events),
        "request_header_bytes": sum(event.request_header_bytes for event in events),
        "response_body_bytes": sum(event.response_body_bytes for event in events),
        "response_header_bytes": sum(event.response_header_bytes for event in events),
        "measured_application_bytes": sum(event.measured_application_bytes for event in events),
        "round_trip_p50_seconds": (
            float(np.percentile(latencies, 50)) if len(latencies) else 0.0
        ),
        "round_trip_p95_seconds": (
            float(np.percentile(latencies, 95)) if len(latencies) else 0.0
        ),
        "tls_record_bytes_measured": False,
        "scope": (
            "serialized HTTP application bytes and round-trip latency; excludes TLS record, "
            "TCP/IP framing and retransmissions not exposed by urllib"
        ),
    }

send 

send(payload, *, message_type, sender_role, receiver_role, direction=None, stage=None, step=None)
Source code in src/vertimosaic/transport/remote.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
def send(
    self,
    payload: Any,
    *,
    message_type: str,
    sender_role: str,
    receiver_role: str,
    direction: str | None = None,
    stage: str | None = None,
    step: int | None = None,
) -> Any:
    endpoint = self.endpoints.get(receiver_role)
    if endpoint is None:
        raise ValueError(f"no remote endpoint configured for receiver role {receiver_role}")

    if isinstance(payload, StructuredPayload):
        wire_value = payload.value
        shape = payload.shape
        count = payload.scalar_count
        size = payload.estimated_bytes
    else:
        wire_value = payload
        shape, count, size = self._metadata(payload)

    message_id = uuid4().hex
    envelope = {
        "schema_version": SCHEMA_VERSION,
        "message_id": message_id,
        "message_type": message_type,
        "sender_role": sender_role,
        "receiver_role": receiver_role,
        "direction": direction,
        "stage": stage,
        "step": step,
        "sent_at_unix": time.time(),
        "nonce": uuid4().hex,
        "array_codec": self.array_codec,
        "payload": _encode_value(wire_value, array_codec=self.array_codec),
    }
    body = json.dumps(envelope, separators=(",", ":")).encode()
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "X-VertiMosaic-Schema": str(SCHEMA_VERSION),
        "X-VertiMosaic-Message-ID": message_id,
        "X-VertiMosaic-Sender": sender_role,
    }
    token = self.bearer_tokens.get(receiver_role)
    if token:
        headers["Authorization"] = f"Bearer {token}"
    signing = self.signing_keys.get(receiver_role)
    if signing is not None:
        key_id, secret = signing
        headers["X-VertiMosaic-Key-ID"] = key_id
        headers["X-VertiMosaic-Signature"] = _sign(secret, body)

    context = self._ssl_context()
    last_error: Exception | None = None
    response_payload: Any = None
    for attempt in range(self.max_retries + 1):
        request = Request(endpoint, data=body, headers=headers, method="POST")
        started = time.perf_counter()
        try:
            with urlopen(  # nosec B310 - endpoint schemes are restricted above
                request,
                timeout=self.timeout_seconds,
                context=context,
            ) as response:
                response_body = response.read()
                elapsed = time.perf_counter() - started
                response_headers = [(str(k), str(v)) for k, v in response.headers.items()]
                request_headers = [(str(k), str(v)) for k, v in request.header_items()]
                self.network_audit_log.append(
                    RemoteNetworkEvent(
                        message_id=message_id,
                        request_body_bytes=len(body),
                        request_header_bytes=_header_bytes(request_headers),
                        response_body_bytes=len(response_body),
                        response_header_bytes=_header_bytes(
                            response_headers,
                            status_line_bytes=17,
                        ),
                        round_trip_seconds=elapsed,
                        status_code=int(response.status),
                        attempt=attempt + 1,
                    )
                )
                if signing is not None:
                    key_id, secret = signing
                    response_key_id = response.headers.get("X-VertiMosaic-Key-ID")
                    response_signature = response.headers.get("X-VertiMosaic-Signature")
                    if response_key_id != key_id or response_signature is None:
                        raise RuntimeError(
                            "signed request received an unsigned remote response"
                        )
                    if not hmac.compare_digest(
                        response_signature,
                        _sign(secret, response_body),
                    ):
                        raise RuntimeError("remote response signature mismatch")
                decoded = json.loads(response_body.decode())
                if decoded.get("schema_version") != SCHEMA_VERSION:
                    raise RuntimeError("remote response schema version mismatch")
                if decoded.get("message_id") != message_id:
                    raise RuntimeError("remote response message ID mismatch")
                if "payload" not in decoded:
                    raise RuntimeError("remote response is missing payload")
                response_payload = _decode_value(decoded["payload"])
                break
        except (HTTPError, URLError, TimeoutError, OSError, json.JSONDecodeError) as exc:
            last_error = exc
            if attempt >= self.max_retries:
                raise RuntimeError(
                    f"remote delivery to {receiver_role} failed after {attempt + 1} attempts"
                ) from exc
            time.sleep(self.backoff_seconds * (2**attempt))
    else:
        raise RuntimeError(
            "remote transport retry loop terminated unexpectedly"
        ) from last_error

    message = Message(
        message_type=message_type,
        sender_role=sender_role,
        receiver_role=receiver_role,
        shape=shape,
        scalar_count=count,
        estimated_bytes=size,
        direction=direction,
        stage=stage,
        step=step,
        payload=None,
    )
    self.audit_log.append(AuditEvent.from_message(message))
    return response_payload

__init__ 

__init__(audit_log=list(), endpoints=dict(), bearer_tokens=dict(), signing_keys=dict(), ca_file=None, client_cert=None, client_key=None, timeout_seconds=10.0, max_retries=2, backoff_seconds=0.25, allow_insecure_http=False, array_codec='json', network_audit_log=list())

Privacy research backends

vertimosaic.privacy.GaussianDPBackend dataclass

Gaussian release with a caller-enforced L2-sensitivity contract.

This legacy research backend is retained for experiments where sensitivity is established by protocol-specific reasoning outside this class. Prefer :class:ClippedGaussianDPBackend when the release can be bounded by clipping.

Source code in src/vertimosaic/privacy/backends.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@dataclass
class GaussianDPBackend:
    """Gaussian release with a caller-enforced L2-sensitivity contract.

    This legacy research backend is retained for experiments where sensitivity is
    established by protocol-specific reasoning outside this class. Prefer
    :class:`ClippedGaussianDPBackend` when the release can be bounded by clipping.
    """

    l2_sensitivity: float
    noise_multiplier: float
    seed: int = 42
    accountant: GaussianZCDPAccountant = field(init=False)
    _rng: np.random.Generator = field(init=False, repr=False)

    def __post_init__(self) -> None:
        if not np.isfinite(self.l2_sensitivity) or self.l2_sensitivity <= 0:
            raise ValueError("l2_sensitivity must be finite and positive")
        self.accountant = GaussianZCDPAccountant(self.noise_multiplier)
        self._rng = np.random.default_rng(self.seed)

    @property
    def noise_std(self) -> float:
        return float(self.l2_sensitivity * self.noise_multiplier)

    def release(self, value: np.ndarray) -> np.ndarray:
        array = np.asarray(value, dtype=float)
        if not np.isfinite(array).all():
            raise ValueError("DP release values must be finite")
        noise = self._rng.normal(0.0, self.noise_std, size=array.shape)
        self.accountant.step()
        return array + noise

    def privacy_report(self, *, delta: float) -> dict[str, float | int | str]:
        report = self.accountant.summary(delta=delta)
        report["l2_sensitivity"] = float(self.l2_sensitivity)
        report["noise_std"] = self.noise_std
        report["scope"] = (
            "release-level accounting only; caller must establish the sensitivity bound; "
            "end-to-end VFL DP requires every sensitive release to be covered"
        )
        return report

l2_sensitivity instance-attribute 

l2_sensitivity

noise_multiplier instance-attribute 

noise_multiplier

seed class-attribute instance-attribute 

seed = 42

accountant class-attribute instance-attribute 

accountant = field(init=False)

_rng class-attribute instance-attribute 

_rng = field(init=False, repr=False)

noise_std property 

noise_std

__post_init__ 

__post_init__()
Source code in src/vertimosaic/privacy/backends.py 
67
68
69
70
71
def __post_init__(self) -> None:
    if not np.isfinite(self.l2_sensitivity) or self.l2_sensitivity <= 0:
        raise ValueError("l2_sensitivity must be finite and positive")
    self.accountant = GaussianZCDPAccountant(self.noise_multiplier)
    self._rng = np.random.default_rng(self.seed)

release 

release(value)
Source code in src/vertimosaic/privacy/backends.py 
77
78
79
80
81
82
83
def release(self, value: np.ndarray) -> np.ndarray:
    array = np.asarray(value, dtype=float)
    if not np.isfinite(array).all():
        raise ValueError("DP release values must be finite")
    noise = self._rng.normal(0.0, self.noise_std, size=array.shape)
    self.accountant.step()
    return array + noise

privacy_report 

privacy_report(*, delta)
Source code in src/vertimosaic/privacy/backends.py 
85
86
87
88
89
90
91
92
93
def privacy_report(self, *, delta: float) -> dict[str, float | int | str]:
    report = self.accountant.summary(delta=delta)
    report["l2_sensitivity"] = float(self.l2_sensitivity)
    report["noise_std"] = self.noise_std
    report["scope"] = (
        "release-level accounting only; caller must establish the sensitivity bound; "
        "end-to-end VFL DP requires every sensitive release to be covered"
    )
    return report

__init__ 

__init__(l2_sensitivity, noise_multiplier, seed=42)

vertimosaic.privacy.ClippedGaussianDPBackend dataclass

Sensitivity-enforcing Gaussian release for one bounded vector message.

The input vector is clipped to clip_l2_norm before noise is added. Under replace_one message adjacency, two clipped vectors can differ by at most 2 * clip_l2_norm in L2 norm. Under add_remove adjacency the bound is clip_l2_norm. This makes the sensitivity contract executable instead of caller-supplied, while remaining a message-level mechanism. It does not turn a complete VFL protocol into end-to-end differential privacy unless all sensitive releases and the relevant neighboring-dataset relation are covered.

Source code in src/vertimosaic/privacy/backends.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
@dataclass
class ClippedGaussianDPBackend:
    """Sensitivity-enforcing Gaussian release for one bounded vector message.

    The input vector is clipped to ``clip_l2_norm`` before noise is added. Under
    ``replace_one`` message adjacency, two clipped vectors can differ by at most
    ``2 * clip_l2_norm`` in L2 norm. Under ``add_remove`` adjacency the bound is
    ``clip_l2_norm``. This makes the sensitivity contract executable instead of
    caller-supplied, while remaining a *message-level* mechanism. It does not turn
    a complete VFL protocol into end-to-end differential privacy unless all
    sensitive releases and the relevant neighboring-dataset relation are covered.
    """

    clip_l2_norm: float
    noise_multiplier: float
    adjacency: str = "replace_one"
    seed: int = 42
    accountant: GaussianZCDPAccountant = field(init=False)
    _rng: np.random.Generator = field(init=False, repr=False)

    def __post_init__(self) -> None:
        if not np.isfinite(self.clip_l2_norm) or self.clip_l2_norm <= 0:
            raise ValueError("clip_l2_norm must be finite and positive")
        if self.adjacency not in {"replace_one", "add_remove"}:
            raise ValueError("adjacency must be replace_one or add_remove")
        self.accountant = GaussianZCDPAccountant(self.noise_multiplier)
        self._rng = np.random.default_rng(self.seed)

    @property
    def l2_sensitivity(self) -> float:
        factor = 2.0 if self.adjacency == "replace_one" else 1.0
        return float(factor * self.clip_l2_norm)

    @property
    def noise_std(self) -> float:
        return float(self.noise_multiplier * self.l2_sensitivity)

    def clip(self, value: np.ndarray) -> tuple[np.ndarray, float]:
        array = np.asarray(value, dtype=float)
        if not np.isfinite(array).all():
            raise ValueError("DP release values must be finite")
        norm = float(np.linalg.norm(array.reshape(-1), ord=2))
        if norm <= self.clip_l2_norm or norm == 0.0:
            return array.copy(), norm
        return array * (self.clip_l2_norm / norm), norm

    def release(self, value: np.ndarray) -> np.ndarray:
        clipped, _ = self.clip(value)
        noise = self._rng.normal(0.0, self.noise_std, size=clipped.shape)
        self.accountant.step()
        return clipped + noise

    def privacy_report(self, *, delta: float) -> dict[str, float | int | str]:
        report = self.accountant.summary(delta=delta)
        report.update(
            {
                "clip_l2_norm": float(self.clip_l2_norm),
                "adjacency": self.adjacency,
                "l2_sensitivity": self.l2_sensitivity,
                "noise_std": self.noise_std,
                "sensitivity_enforcement": "L2 clipping before every release",
                "scope": (
                    "message-level clipped Gaussian mechanism; end-to-end VFL DP requires "
                    "all sensitive protocol releases and the dataset adjacency to be covered"
                ),
            }
        )
        return report

clip_l2_norm instance-attribute 

clip_l2_norm

noise_multiplier instance-attribute 

noise_multiplier

adjacency class-attribute instance-attribute 

adjacency = 'replace_one'

seed class-attribute instance-attribute 

seed = 42

accountant class-attribute instance-attribute 

accountant = field(init=False)

_rng class-attribute instance-attribute 

_rng = field(init=False, repr=False)

l2_sensitivity property 

l2_sensitivity

noise_std property 

noise_std

__post_init__ 

__post_init__()
Source code in src/vertimosaic/privacy/backends.py 
116
117
118
119
120
121
122
def __post_init__(self) -> None:
    if not np.isfinite(self.clip_l2_norm) or self.clip_l2_norm <= 0:
        raise ValueError("clip_l2_norm must be finite and positive")
    if self.adjacency not in {"replace_one", "add_remove"}:
        raise ValueError("adjacency must be replace_one or add_remove")
    self.accountant = GaussianZCDPAccountant(self.noise_multiplier)
    self._rng = np.random.default_rng(self.seed)

clip 

clip(value)
Source code in src/vertimosaic/privacy/backends.py 
133
134
135
136
137
138
139
140
def clip(self, value: np.ndarray) -> tuple[np.ndarray, float]:
    array = np.asarray(value, dtype=float)
    if not np.isfinite(array).all():
        raise ValueError("DP release values must be finite")
    norm = float(np.linalg.norm(array.reshape(-1), ord=2))
    if norm <= self.clip_l2_norm or norm == 0.0:
        return array.copy(), norm
    return array * (self.clip_l2_norm / norm), norm

release 

release(value)
Source code in src/vertimosaic/privacy/backends.py 
142
143
144
145
146
def release(self, value: np.ndarray) -> np.ndarray:
    clipped, _ = self.clip(value)
    noise = self._rng.normal(0.0, self.noise_std, size=clipped.shape)
    self.accountant.step()
    return clipped + noise

privacy_report 

privacy_report(*, delta)
Source code in src/vertimosaic/privacy/backends.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def privacy_report(self, *, delta: float) -> dict[str, float | int | str]:
    report = self.accountant.summary(delta=delta)
    report.update(
        {
            "clip_l2_norm": float(self.clip_l2_norm),
            "adjacency": self.adjacency,
            "l2_sensitivity": self.l2_sensitivity,
            "noise_std": self.noise_std,
            "sensitivity_enforcement": "L2 clipping before every release",
            "scope": (
                "message-level clipped Gaussian mechanism; end-to-end VFL DP requires "
                "all sensitive protocol releases and the dataset adjacency to be covered"
            ),
        }
    )
    return report

__init__ 

__init__(clip_l2_norm, noise_multiplier, adjacency='replace_one', seed=42)

vertimosaic.privacy.PairwiseMaskSecureAggregation dataclass

Reference pairwise-mask secure sum for integer vectors.

Every pair of parties must receive the same out-of-band secret without exposing it to the aggregator. Pairwise masks cancel in the aggregate. This models the core honest-but-curious secure-aggregation idea, but intentionally does not provide dropout recovery, malicious-party security, or collusion resistance.

Source code in src/vertimosaic/privacy/crypto.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@dataclass(frozen=True)
class PairwiseMaskSecureAggregation:
    """Reference pairwise-mask secure sum for integer vectors.

    Every pair of parties must receive the same out-of-band secret without exposing
    it to the aggregator. Pairwise masks cancel in the aggregate. This models the
    core honest-but-curious secure-aggregation idea, but intentionally does not
    provide dropout recovery, malicious-party security, or collusion resistance.
    """

    prime: int = _FIELD_PRIME

    def __post_init__(self) -> None:
        if self.prime <= 3:
            raise ValueError("prime must be greater than 3")

    @staticmethod
    def pair_key(left: str, right: str) -> tuple[str, str]:
        if left == right:
            raise ValueError("a pairwise secret requires two distinct parties")
        return (left, right) if left < right else (right, left)

    def mask_update(
        self,
        *,
        party_id: str,
        update: np.ndarray,
        party_ids: tuple[str, ...],
        pairwise_secrets: dict[tuple[str, str], bytes],
        round_id: str,
    ) -> np.ndarray:
        if party_id not in party_ids:
            raise ValueError("party_id must be included in party_ids")
        if len(set(party_ids)) != len(party_ids):
            raise ValueError("party_ids must be unique")
        encoded = np.mod(_as_int_vector(update), self.prime)
        masked = encoded.copy()
        for peer in party_ids:
            if peer == party_id:
                continue
            key = self.pair_key(party_id, peer)
            secret = pairwise_secrets.get(key)
            if secret is None or len(secret) < 16:
                raise ValueError(f"missing or weak pairwise secret for {key}")
            mask = _pairwise_mask(
                secret,
                round_id=round_id,
                shape=tuple(encoded.shape),
                prime=self.prime,
            )
            if party_id < peer:
                masked = np.mod(masked + mask, self.prime)
            else:
                masked = np.mod(masked - mask, self.prime)
        return np.asarray(masked, dtype=object)

    def aggregate(self, masked_updates: list[np.ndarray]) -> np.ndarray:
        if not masked_updates:
            raise ValueError("at least one masked update is required")
        shape = np.asarray(masked_updates[0]).shape
        total = np.zeros(shape, dtype=object)
        for update in masked_updates:
            array = _as_int_vector(update)
            if array.shape != shape:
                raise ValueError("all masked updates must have equal shape")
            total = np.mod(total + array, self.prime)
        return _signed_from_field(total, self.prime)

    @property
    def security_scope(self) -> str:
        return (
            "honest-but-curious aggregator reference; pairwise secrets stay out of band; "
            "no dropout recovery, malicious-party security, or collusion resistance"
        )

prime class-attribute instance-attribute 

prime = _FIELD_PRIME

security_scope property 

security_scope

__post_init__ 

__post_init__()
Source code in src/vertimosaic/privacy/crypto.py 
60
61
62
def __post_init__(self) -> None:
    if self.prime <= 3:
        raise ValueError("prime must be greater than 3")

pair_key staticmethod 

pair_key(left, right)
Source code in src/vertimosaic/privacy/crypto.py 
64
65
66
67
68
@staticmethod
def pair_key(left: str, right: str) -> tuple[str, str]:
    if left == right:
        raise ValueError("a pairwise secret requires two distinct parties")
    return (left, right) if left < right else (right, left)

mask_update 

mask_update(*, party_id, update, party_ids, pairwise_secrets, round_id)
Source code in src/vertimosaic/privacy/crypto.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def mask_update(
    self,
    *,
    party_id: str,
    update: np.ndarray,
    party_ids: tuple[str, ...],
    pairwise_secrets: dict[tuple[str, str], bytes],
    round_id: str,
) -> np.ndarray:
    if party_id not in party_ids:
        raise ValueError("party_id must be included in party_ids")
    if len(set(party_ids)) != len(party_ids):
        raise ValueError("party_ids must be unique")
    encoded = np.mod(_as_int_vector(update), self.prime)
    masked = encoded.copy()
    for peer in party_ids:
        if peer == party_id:
            continue
        key = self.pair_key(party_id, peer)
        secret = pairwise_secrets.get(key)
        if secret is None or len(secret) < 16:
            raise ValueError(f"missing or weak pairwise secret for {key}")
        mask = _pairwise_mask(
            secret,
            round_id=round_id,
            shape=tuple(encoded.shape),
            prime=self.prime,
        )
        if party_id < peer:
            masked = np.mod(masked + mask, self.prime)
        else:
            masked = np.mod(masked - mask, self.prime)
    return np.asarray(masked, dtype=object)

aggregate 

aggregate(masked_updates)
Source code in src/vertimosaic/privacy/crypto.py 
104
105
106
107
108
109
110
111
112
113
114
def aggregate(self, masked_updates: list[np.ndarray]) -> np.ndarray:
    if not masked_updates:
        raise ValueError("at least one masked update is required")
    shape = np.asarray(masked_updates[0]).shape
    total = np.zeros(shape, dtype=object)
    for update in masked_updates:
        array = _as_int_vector(update)
        if array.shape != shape:
            raise ValueError("all masked updates must have equal shape")
        total = np.mod(total + array, self.prime)
    return _signed_from_field(total, self.prime)

__init__ 

__init__(prime=_FIELD_PRIME)

vertimosaic.privacy.AdditiveSecretSharingSum dataclass

Reference additive secret-sharing sum over a prime field.

This is a narrow MPC primitive for integer-vector summation, not a general MPC runtime. Shares are sampled with Python's cryptographic secrets module.

Source code in src/vertimosaic/privacy/crypto.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@dataclass(frozen=True)
class AdditiveSecretSharingSum:
    """Reference additive secret-sharing sum over a prime field.

    This is a narrow MPC primitive for integer-vector summation, not a general MPC
    runtime. Shares are sampled with Python's cryptographic ``secrets`` module.
    """

    share_count: int
    prime: int = _FIELD_PRIME

    def __post_init__(self) -> None:
        if self.share_count < 2:
            raise ValueError("share_count must be at least 2")
        if self.prime <= 3:
            raise ValueError("prime must be greater than 3")

    def split(self, value: np.ndarray) -> list[np.ndarray]:
        encoded = np.mod(_as_int_vector(value), self.prime)
        shares: list[np.ndarray] = []
        running = np.zeros(encoded.shape, dtype=object)
        for _ in range(self.share_count - 1):
            random_values = [secrets.randbelow(self.prime) for _ in range(encoded.size)]
            share = np.asarray(random_values, dtype=object).reshape(encoded.shape)
            shares.append(share)
            running = np.mod(running + share, self.prime)
        shares.append(np.mod(encoded - running, self.prime))
        return shares

    def reconstruct(self, shares: list[np.ndarray]) -> np.ndarray:
        if len(shares) != self.share_count:
            raise ValueError("incorrect number of shares")
        shape = np.asarray(shares[0]).shape
        total = np.zeros(shape, dtype=object)
        for share in shares:
            array = _as_int_vector(share)
            if array.shape != shape:
                raise ValueError("all shares must have equal shape")
            total = np.mod(total + array, self.prime)
        return _signed_from_field(total, self.prime)

    @property
    def security_scope(self) -> str:
        return (
            "additive secret sharing for integer-vector sums only; not a general MPC "
            "runtime and no malicious-party verification"
        )

share_count instance-attribute 

share_count

prime class-attribute instance-attribute 

prime = _FIELD_PRIME

security_scope property 

security_scope

__post_init__ 

__post_init__()
Source code in src/vertimosaic/privacy/crypto.py 
135
136
137
138
139
def __post_init__(self) -> None:
    if self.share_count < 2:
        raise ValueError("share_count must be at least 2")
    if self.prime <= 3:
        raise ValueError("prime must be greater than 3")

split 

split(value)
Source code in src/vertimosaic/privacy/crypto.py 
141
142
143
144
145
146
147
148
149
150
151
def split(self, value: np.ndarray) -> list[np.ndarray]:
    encoded = np.mod(_as_int_vector(value), self.prime)
    shares: list[np.ndarray] = []
    running = np.zeros(encoded.shape, dtype=object)
    for _ in range(self.share_count - 1):
        random_values = [secrets.randbelow(self.prime) for _ in range(encoded.size)]
        share = np.asarray(random_values, dtype=object).reshape(encoded.shape)
        shares.append(share)
        running = np.mod(running + share, self.prime)
    shares.append(np.mod(encoded - running, self.prime))
    return shares

reconstruct 

reconstruct(shares)
Source code in src/vertimosaic/privacy/crypto.py 
153
154
155
156
157
158
159
160
161
162
163
def reconstruct(self, shares: list[np.ndarray]) -> np.ndarray:
    if len(shares) != self.share_count:
        raise ValueError("incorrect number of shares")
    shape = np.asarray(shares[0]).shape
    total = np.zeros(shape, dtype=object)
    for share in shares:
        array = _as_int_vector(share)
        if array.shape != shape:
            raise ValueError("all shares must have equal shape")
        total = np.mod(total + array, self.prime)
    return _signed_from_field(total, self.prime)

__init__ 

__init__(share_count, prime=_FIELD_PRIME)

vertimosaic.privacy.OpenMinedPSIBackend dataclass

Optional adapter for OpenMined's ECDH-based PSI implementation.

Source code in src/vertimosaic/privacy/crypto.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
@dataclass(frozen=True)
class OpenMinedPSIBackend:
    """Optional adapter for OpenMined's ECDH-based PSI implementation."""

    false_positive_rate: float = 1e-9

    def __post_init__(self) -> None:
        if not 0.0 < self.false_positive_rate < 1.0:
            raise ValueError("false_positive_rate must be in (0, 1)")

    @staticmethod
    def _module() -> Any:
        candidates = ("openmined_psi", "private_set_intersection.python")
        for name in candidates:
            try:
                return import_module(name)
            except ImportError:
                continue
        raise RuntimeError(
            "OpenMined PSI is optional; install VertiMosaic with the 'privacy-crypto' extra"
        )

    def intersection(self, client_ids: list[str], server_ids: list[str]) -> list[str]:
        module = self._module()
        client = module.client.CreateWithNewKey(True)
        server = module.server.CreateWithNewKey(True)
        data_structure = getattr(getattr(module, "DataStructure", object), "RAW", None)
        if data_structure is None:
            setup = server.CreateSetupMessage(
                self.false_positive_rate,
                len(client_ids),
                server_ids,
            )
        else:
            setup = server.CreateSetupMessage(
                self.false_positive_rate,
                len(client_ids),
                server_ids,
                data_structure,
            )
        request = client.CreateRequest(client_ids)
        response = server.ProcessRequest(request)
        indices = client.GetIntersection(setup, response)
        return [client_ids[int(index)] for index in indices]

    @property
    def security_scope(self) -> str:
        return (
            "optional OpenMined ECDH PSI adapter; PSI protects set-intersection protocol "
            "inputs subject to the upstream implementation's threat model, not VFL "
            "training messages"
        )

false_positive_rate class-attribute instance-attribute 

false_positive_rate = 1e-09

security_scope property 

security_scope

__post_init__ 

__post_init__()
Source code in src/vertimosaic/privacy/crypto.py 
179
180
181
def __post_init__(self) -> None:
    if not 0.0 < self.false_positive_rate < 1.0:
        raise ValueError("false_positive_rate must be in (0, 1)")

_module staticmethod 

_module()
Source code in src/vertimosaic/privacy/crypto.py 
183
184
185
186
187
188
189
190
191
192
193
@staticmethod
def _module() -> Any:
    candidates = ("openmined_psi", "private_set_intersection.python")
    for name in candidates:
        try:
            return import_module(name)
        except ImportError:
            continue
    raise RuntimeError(
        "OpenMined PSI is optional; install VertiMosaic with the 'privacy-crypto' extra"
    )

intersection 

intersection(client_ids, server_ids)
Source code in src/vertimosaic/privacy/crypto.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def intersection(self, client_ids: list[str], server_ids: list[str]) -> list[str]:
    module = self._module()
    client = module.client.CreateWithNewKey(True)
    server = module.server.CreateWithNewKey(True)
    data_structure = getattr(getattr(module, "DataStructure", object), "RAW", None)
    if data_structure is None:
        setup = server.CreateSetupMessage(
            self.false_positive_rate,
            len(client_ids),
            server_ids,
        )
    else:
        setup = server.CreateSetupMessage(
            self.false_positive_rate,
            len(client_ids),
            server_ids,
            data_structure,
        )
    request = client.CreateRequest(client_ids)
    response = server.ProcessRequest(request)
    indices = client.GetIntersection(setup, response)
    return [client_ids[int(index)] for index in indices]

__init__ 

__init__(false_positive_rate=1e-09)

vertimosaic.privacy.PaillierHomomorphicSum dataclass

Optional Paillier adapter for bounded signed integer-vector summation.

Source code in src/vertimosaic/privacy/crypto.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
@dataclass(frozen=True)
class PaillierHomomorphicSum:
    """Optional Paillier adapter for bounded signed integer-vector summation."""

    key_bits: int = 1024

    def __post_init__(self) -> None:
        if self.key_bits < 512:
            raise ValueError("key_bits must be at least 512 for research use")

    @staticmethod
    def _module() -> Any:
        try:
            return import_module("pailliers")
        except ImportError as exc:
            raise RuntimeError(
                "Paillier support is optional; install VertiMosaic with the 'privacy-crypto' extra"
            ) from exc

    def sum(self, vectors: list[np.ndarray]) -> np.ndarray:
        if not vectors:
            raise ValueError("at least one vector is required")
        arrays = [_as_int_vector(vector) for vector in vectors]
        shape = arrays[0].shape
        if any(array.shape != shape for array in arrays):
            raise ValueError("all vectors must have equal shape")
        module = self._module()
        secret_key = module.secret(self.key_bits)
        public_key = module.public(secret_key)
        modulus = int(public_key[0])
        true_sum_bound = sum(int(np.max(np.abs(array.astype(object)))) for array in arrays)
        if true_sum_bound >= modulus // 3:
            raise OverflowError("plaintext sum is too large for the generated Paillier modulus")
        result: list[int] = []
        for index in range(arrays[0].size):
            encrypted = [module.encrypt(public_key, int(array.flat[index])) for array in arrays]
            total_cipher = sum(encrypted)
            residue = int(module.decrypt(secret_key, total_cipher))
            result.append(residue - modulus if residue > modulus // 2 else residue)
        return np.asarray(result, dtype=object).reshape(shape)

    @property
    def security_scope(self) -> str:
        return (
            "optional Paillier additive-homomorphic integer sum; does not by itself provide "
            "secure VFL, key management, malicious-party security, or private comparisons"
        )

key_bits class-attribute instance-attribute 

key_bits = 1024

security_scope property 

security_scope

__post_init__ 

__post_init__()
Source code in src/vertimosaic/privacy/crypto.py 
233
234
235
def __post_init__(self) -> None:
    if self.key_bits < 512:
        raise ValueError("key_bits must be at least 512 for research use")

_module staticmethod 

_module()
Source code in src/vertimosaic/privacy/crypto.py 
237
238
239
240
241
242
243
244
@staticmethod
def _module() -> Any:
    try:
        return import_module("pailliers")
    except ImportError as exc:
        raise RuntimeError(
            "Paillier support is optional; install VertiMosaic with the 'privacy-crypto' extra"
        ) from exc

sum 

sum(vectors)
Source code in src/vertimosaic/privacy/crypto.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def sum(self, vectors: list[np.ndarray]) -> np.ndarray:
    if not vectors:
        raise ValueError("at least one vector is required")
    arrays = [_as_int_vector(vector) for vector in vectors]
    shape = arrays[0].shape
    if any(array.shape != shape for array in arrays):
        raise ValueError("all vectors must have equal shape")
    module = self._module()
    secret_key = module.secret(self.key_bits)
    public_key = module.public(secret_key)
    modulus = int(public_key[0])
    true_sum_bound = sum(int(np.max(np.abs(array.astype(object)))) for array in arrays)
    if true_sum_bound >= modulus // 3:
        raise OverflowError("plaintext sum is too large for the generated Paillier modulus")
    result: list[int] = []
    for index in range(arrays[0].size):
        encrypted = [module.encrypt(public_key, int(array.flat[index])) for array in arrays]
        total_cipher = sum(encrypted)
        residue = int(module.decrypt(secret_key, total_cipher))
        result.append(residue - modulus if residue > modulus // 2 else residue)
    return np.asarray(result, dtype=object).reshape(shape)

__init__ 

__init__(key_bits=1024)

Scope the guarantee

The presence of a privacy primitive in the API does not imply that unrelated protocol messages inherit its guarantee. Use Privacy Boundaries and Threat Model when describing an experiment.