41 lines
1.2 KiB
Python
41 lines
1.2 KiB
Python
from __future__ import annotations
|
|
from typing import List, Dict
|
|
|
|
|
|
class SecureAggregator:
|
|
"""Placeholder secure-aggregation primitive.
|
|
In a real system this would perform cryptographic aggregation (e.g., via MPC).
|
|
Here we simply sum numeric signals and average others defensively.
|
|
"""
|
|
|
|
@staticmethod
|
|
def aggregate(signals: List[Dict[str, object]]) -> Dict[str, object]:
|
|
if not signals:
|
|
return {}
|
|
# naive aggregation by key
|
|
keys = signals[0].keys()
|
|
result: Dict[str, object] = {}
|
|
for k in keys:
|
|
try:
|
|
vals = [float(s.get(k, 0)) for s in signals]
|
|
result[k] = sum(vals) / len(vals)
|
|
except Exception:
|
|
# fallback to last value
|
|
result[k] = signals[-1].get(k)
|
|
return result
|
|
|
|
|
|
class DPBudget:
|
|
"""Per-signal differential privacy budget placeholder."""
|
|
|
|
def __init__(self, total_budget: float = 1.0) -> None:
|
|
self.total_budget = float(total_budget)
|
|
self.used = 0.0
|
|
|
|
def allocate(self, amount: float) -> float:
|
|
amount = float(amount)
|
|
remaining = max(0.0, self.total_budget - self.used)
|
|
take = min(remaining, amount)
|
|
self.used += take
|
|
return take
|