80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
"""OpenGrowth DSL sketch for CatOpt-inspired federation primitives.
|
|
|
|
This module provides a lightweight, self-contained Domain-Specific Language
|
|
sketch that models local experiments, exchanged signals, and incremental plan
|
|
updates in a privacy-preserving federation setting.
|
|
|
|
The goal here is to give a minimal, type-annotated contract surface that can
|
|
be extended by adapters and governance tooling without pulling in heavy
|
|
dependencies.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, Any
|
|
|
|
|
|
@dataclass
|
|
class LocalExperiment:
|
|
"""Represents a locally-defined experiment on a single participant.
|
|
|
|
Attributes:
|
|
id: Unique experiment identifier within the federation.
|
|
variables: Local experiment parameters (e.g., price, funnel step).
|
|
objectives: Desired outcome measures (e.g., maximize LTV, minimize CPA).
|
|
privacy_budget: Optional per-experiment privacy constraints (DP budgets, data retention).
|
|
"""
|
|
|
|
id: str
|
|
variables: Dict[str, Any]
|
|
objectives: Dict[str, Any]
|
|
privacy_budget: Dict[str, Any] = field(default_factory=dict)
|
|
created_at: float = field(default_factory=lambda: time.time())
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"variables": self.variables,
|
|
"objectives": self.objectives,
|
|
"privacy_budget": self.privacy_budget,
|
|
"created_at": self.created_at,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class SharedSignal:
|
|
"""Represents aggregated or shared signals exchanged between participants."""
|
|
|
|
experiment_id: str
|
|
metrics: Dict[str, Any]
|
|
timestamp: float = field(default_factory=lambda: time.time())
|
|
metadata: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"experiment_id": self.experiment_id,
|
|
"metrics": self.metrics,
|
|
"timestamp": self.timestamp,
|
|
"metadata": self.metadata,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
"""Incremental update to an local plan or experiment outcome."""
|
|
|
|
experiment_id: str
|
|
delta: Dict[str, Any]
|
|
timestamp: float = field(default_factory=lambda: time.time())
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"experiment_id": self.experiment_id,
|
|
"delta": self.delta,
|
|
"timestamp": self.timestamp,
|
|
}
|
|
|
|
|
|
__all__ = ["LocalExperiment", "SharedSignal", "PlanDelta"]
|