community
/
idea141-openimpact-privacy-preserving
34
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

build(agent): semicolon#54de0b iteration

This commit is contained in:
agent-54de0bcc6a17828b 2026-04-24 18:51:39 +02:00
parent 2830598920
commit e59a52d8a2
16 changed files with 779 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
.gitignore vendored Normal file
View File

@ -0,0 +1,21 @@
node_modules/
.npmrc
.env
.env.*
__tests__/
coverage/
.nyc_output/
dist/
build/
.cache/
*.log
.DS_Store
tmp/
.tmp/
__pycache__/
*.pyc
.venv/
venv/
*.egg-info/
.pytest_cache/
READY_TO_PUBLISH

32
AGENTS.md Normal file
View File

@ -0,0 +1,32 @@
# OpenImpact Agent Guide
## Architecture
- `src/openimpact/contracts.py`: shared data contracts for requests, plan deltas, and venue batches.
- `src/openimpact/model.py`: deterministic temporary-impact regression model.
- `src/openimpact/federated.py`: local venue training, secure aggregation, and federated orchestration.
- `src/openimpact/synthetic.py`: deterministic synthetic market generator for safe testing.
- `src/openimpact/replay.py`: replay engine with deterministic latency jitter.
- `src/openimpact/ledger.py`: SQLite governance ledger for model updates.
- `src/openimpact/evaluation.py`: RMSE, latency, and leakage-style metrics.
- `src/openimpact/pipeline.py`: end-to-end demo runner.
## Tech Stack
- Python 3.11+
- `numpy` and `pandas` for numeric and tabular work.
- `sqlite3` from the standard library for durable audit logging.
- `pytest` for tests.
## Rules
- Keep the synthetic generator deterministic for a fixed seed.
- Prefer the smallest correct change that preserves the model/evaluation contract.
- Update tests whenever behavior changes.
- Keep public APIs importable from `openimpact`.
- `test.sh` must keep passing `pytest` and `python3 -m build`.
## Verification
- `bash test.sh`
- `python -m openimpact`

41
README.md
View File

@ -1,3 +1,40 @@
# idea141-openimpact-privacy-preserving
# OpenImpact
Source logic for Idea #141
OpenImpact is a privacy-preserving market-impact modeling stack for multi-venue execution research.
It provides:
- deterministic synthetic venue generation for safe stress testing
- a temporary-impact regression model
- federated training across venues with masked aggregation
- a SQLite governance ledger for model-update audit trails
- deterministic replay with venue-specific latency
- an evaluation path for RMSE, latency, and a simple leakage bound
## What This Repo Ships
- `TemporaryImpactModel`: fits a linear temporary-impact model from local order requests.
- `FederatedTrainer`: trains per-venue models and aggregates coefficients without exposing raw requests.
- `GovernanceLedger`: stores update metadata in SQLite.
- `DeterministicReplayEngine`: replays the same request stream reproducibly.
- `SyntheticMarketConfig` / `generate_synthetic_market`: builds repeatable venue datasets.
- `evaluate_federated_setup`: returns benchmark metrics against a pooled baseline.
## Quick Start
```bash
python3 -m pip install -e ".[dev]"
pytest
python3 -m openimpact
```
## Package Metadata
This project is published as `idea141-openimpact-privacy-preserving` and uses this `README.md` as its long description.
## Repository Rules
- Keep the synthetic market deterministic by seed.
- Preserve the public API exported from `openimpact/__init__.py`.
- Update tests for every behavior change.
- Keep `test.sh` working; it must run tests and `python3 -m build`.

26
pyproject.toml Normal file
View File

@ -0,0 +1,26 @@
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "idea141-openimpact-privacy-preserving"
version = "0.1.0"
description = "Privacy-preserving federated market-impact modeling and execution simulation across exchanges"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"numpy>=1.26",
"pandas>=2.1",
]
[project.optional-dependencies]
dev = [
"build>=1.2",
"pytest>=8.0",
]
[tool.setuptools]
package-dir = {"" = "src"}
[tool.setuptools.packages.find]
where = ["src"]

29
src/openimpact/__init__.py Normal file
View File

@ -0,0 +1,29 @@
from .contracts import LocalRequest, PlanDelta, VenueBatch
from .evaluation import EvaluationMetrics, evaluate_federated_setup
from .federated import FederatedTrainer, FederatedTrainingResult, SecureAggregator, VenueUpdate
from .ledger import GovernanceLedger
from .model import TemporaryImpactModel
from .pipeline import OpenImpactPlatform, PipelineReport
from .replay import DeterministicReplayEngine, ReplayStep
from .synthetic import SyntheticMarket, SyntheticMarketConfig, generate_synthetic_market
__all__ = [
"LocalRequest",
"PlanDelta",
"VenueBatch",
"EvaluationMetrics",
"evaluate_federated_setup",
"FederatedTrainer",
"FederatedTrainingResult",
"SecureAggregator",
"VenueUpdate",
"GovernanceLedger",
"TemporaryImpactModel",
"OpenImpactPlatform",
"PipelineReport",
"DeterministicReplayEngine",
"ReplayStep",
"SyntheticMarket",
"SyntheticMarketConfig",
"generate_synthetic_market",
]

14
src/openimpact/__main__.py Normal file
View File

@ -0,0 +1,14 @@
from __future__ import annotations
from .pipeline import OpenImpactPlatform
from .synthetic import SyntheticMarketConfig, generate_synthetic_market
def main() -> None:
market = generate_synthetic_market(SyntheticMarketConfig())
report = OpenImpactPlatform().run(market)
print(report)
if __name__ == "__main__":
main()

88
src/openimpact/contracts.py Normal file
View File

@ -0,0 +1,88 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Iterable
import numpy as np
import pandas as pd
@dataclass(frozen=True)
class LocalRequest:
venue_id: str
instrument: str
side: str
quantity: float
mid_price: float
volatility: float
spread_bps: float
timestamp_ns: int
order_imbalance: float = 0.0
queue_position: float = 0.0
def to_features(self) -> np.ndarray:
side_sign = 1.0 if self.side.lower() == "buy" else -1.0
signed_quantity = side_sign * self.quantity
return np.array(
[
1.0,
signed_quantity,
np.log1p(self.quantity),
self.mid_price,
self.volatility,
self.spread_bps,
self.order_imbalance,
self.queue_position,
signed_quantity * self.volatility,
],
dtype=float,
)
@dataclass(frozen=True)
class PlanDelta:
venue_id: str
instrument: str
side: str
quantity: float
limit_price: float
expected_impact: float
confidence: float
rationale: str = ""
@dataclass(frozen=True)
class VenueBatch:
venue_id: str
requests: tuple[LocalRequest, ...]
realized_impact: np.ndarray
metadata: dict[str, float] = field(default_factory=dict)
def to_frame(self) -> pd.DataFrame:
rows = []
for request, impact in zip(self.requests, self.realized_impact, strict=True):
rows.append(
{
"venue_id": request.venue_id,
"instrument": request.instrument,
"side": request.side,
"quantity": request.quantity,
"mid_price": request.mid_price,
"volatility": request.volatility,
"spread_bps": request.spread_bps,
"timestamp_ns": request.timestamp_ns,
"order_imbalance": request.order_imbalance,
"queue_position": request.queue_position,
"realized_impact": float(impact),
}
)
return pd.DataFrame(rows)
def flatten_batches(batches: Iterable[VenueBatch]) -> tuple[list[LocalRequest], np.ndarray]:
requests: list[LocalRequest] = []
impacts: list[float] = []
for batch in batches:
requests.extend(batch.requests)
impacts.extend(float(v) for v in batch.realized_impact)
return requests, np.asarray(impacts, dtype=float)

40
src/openimpact/evaluation.py Normal file
View File

@ -0,0 +1,40 @@
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
from .contracts import flatten_batches
from .federated import FederatedTrainer
from .model import TemporaryImpactModel
from .replay import DeterministicReplayEngine
from .synthetic import SyntheticMarket
@dataclass(frozen=True)
class EvaluationMetrics:
rmse: float
latency_ms: float
privacy_budget: float
leakage_bound: float
def _rmse(prediction: np.ndarray, target: np.ndarray) -> float:
return float(np.sqrt(np.mean((prediction - target) ** 2)))
def evaluate_federated_setup(market: SyntheticMarket, trainer: FederatedTrainer | None = None) -> tuple[EvaluationMetrics, TemporaryImpactModel, TemporaryImpactModel]:
trainer = trainer or FederatedTrainer()
flat_requests, flat_impacts = flatten_batches(market.batches)
baseline = TemporaryImpactModel().fit(flat_requests, flat_impacts)
result = trainer.fit(market.batches)
replay = DeterministicReplayEngine(seed=trainer.seed)
steps = replay.replay(flat_requests, result.global_model, flat_impacts)
latency_ms = float(np.mean([(step.delivery_ns - step.timestamp_ns) / 1_000_000.0 for step in steps]))
predictions = result.global_model.predict(flat_requests)
rmse = _rmse(predictions, flat_impacts)
privacy_budget = float(sum(update.privacy_budget for update in result.updates))
leakage_bound = float(np.max(np.abs(predictions - baseline.predict(flat_requests))))
return EvaluationMetrics(rmse=rmse, latency_ms=latency_ms, privacy_budget=privacy_budget, leakage_bound=leakage_bound), result.global_model, baseline

93
src/openimpact/federated.py Normal file
View File

@ -0,0 +1,93 @@
from __future__ import annotations
from dataclasses import dataclass
from hashlib import sha256
from typing import Sequence
import numpy as np
from .contracts import VenueBatch, flatten_batches
from .model import FEATURE_NAMES, TemporaryImpactModel
@dataclass(frozen=True)
class VenueUpdate:
venue_id: str
coefficients: np.ndarray
sample_count: int
privacy_budget: float
@dataclass(frozen=True)
class FederatedTrainingResult:
global_model: TemporaryImpactModel
local_models: dict[str, TemporaryImpactModel]
updates: tuple[VenueUpdate, ...]
aggregate_checksum: str
class SecureAggregator:
def __init__(self, seed: int = 7):
self.seed = seed
def _pairwise_mask(self, venue_ids: Sequence[str], shape: tuple[int, ...]) -> dict[str, np.ndarray]:
masks = {venue_id: np.zeros(shape, dtype=float) for venue_id in venue_ids}
sorted_ids = sorted(venue_ids)
for left_index, left in enumerate(sorted_ids):
for right in sorted_ids[left_index + 1 :]:
digest = sha256(f"{self.seed}:{left}:{right}".encode("utf-8")).digest()
rng = np.random.default_rng(int.from_bytes(digest[:8], "big", signed=False))
noise = rng.normal(loc=0.0, scale=1e-4, size=shape)
masks[left] += noise
masks[right] -= noise
return masks
def aggregate(self, updates: Sequence[VenueUpdate]) -> np.ndarray:
if not updates:
raise ValueError("cannot aggregate empty update set")
venue_ids = [update.venue_id for update in updates]
masks = self._pairwise_mask(venue_ids, updates[0].coefficients.shape)
total_samples = sum(update.sample_count for update in updates)
aggregated = np.zeros_like(updates[0].coefficients, dtype=float)
for update in updates:
weighted = update.coefficients * update.sample_count
masked = weighted + masks[update.venue_id]
aggregated += masked
return aggregated / float(total_samples)
class FederatedTrainer:
def __init__(self, seed: int = 7, regularization: float = 1e-3):
self.seed = seed
self.regularization = regularization
self.aggregator = SecureAggregator(seed=seed)
def fit(self, batches: Sequence[VenueBatch]) -> FederatedTrainingResult:
if not batches:
raise ValueError("at least one venue batch is required")
local_models: dict[str, TemporaryImpactModel] = {}
updates: list[VenueUpdate] = []
for batch in batches:
model = TemporaryImpactModel(regularization=self.regularization).fit(batch.requests, batch.realized_impact)
local_models[batch.venue_id] = model
sample_count = len(batch.requests)
privacy_budget = 1.0 / max(sample_count, 1)
updates.append(
VenueUpdate(
venue_id=batch.venue_id,
coefficients=model.coefficients_.copy(),
sample_count=sample_count,
privacy_budget=privacy_budget,
)
)
aggregated = self.aggregator.aggregate(updates)
global_model = TemporaryImpactModel(regularization=self.regularization).clone_with_coefficients(aggregated)
checksum = sha256(np.asarray(aggregated, dtype=float).tobytes()).hexdigest()
return FederatedTrainingResult(
global_model=global_model,
local_models=local_models,
updates=tuple(updates),
aggregate_checksum=checksum,
)

93
src/openimpact/ledger.py Normal file
View File

@ -0,0 +1,93 @@
from __future__ import annotations
from dataclasses import dataclass
from hashlib import sha256
from pathlib import Path
import sqlite3
from typing import Iterable
from .federated import VenueUpdate
@dataclass(frozen=True)
class LedgerEntry:
id: int
venue_id: str
round_id: int
model_version: str
checksum: str
sample_count: int
privacy_budget: float
created_at: str
class GovernanceLedger:
def __init__(self, path: str | Path):
self.path = Path(path)
self.path.parent.mkdir(parents=True, exist_ok=True)
self._ensure_schema()
def _connect(self) -> sqlite3.Connection:
connection = sqlite3.connect(self.path)
connection.row_factory = sqlite3.Row
return connection
def _ensure_schema(self) -> None:
with self._connect() as connection:
connection.execute(
"""
CREATE TABLE IF NOT EXISTS governance_ledger (
id INTEGER PRIMARY KEY AUTOINCREMENT,
venue_id TEXT NOT NULL,
round_id INTEGER NOT NULL,
model_version TEXT NOT NULL,
checksum TEXT NOT NULL,
sample_count INTEGER NOT NULL,
privacy_budget REAL NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)
"""
)
connection.commit()
def record_update(self, round_id: int, model_version: str, update: VenueUpdate) -> LedgerEntry:
checksum = sha256(update.coefficients.tobytes()).hexdigest()
with self._connect() as connection:
cursor = connection.execute(
"""
INSERT INTO governance_ledger (
venue_id, round_id, model_version, checksum, sample_count, privacy_budget
) VALUES (?, ?, ?, ?, ?, ?)
""",
(update.venue_id, round_id, model_version, checksum, update.sample_count, update.privacy_budget),
)
connection.commit()
row_id = cursor.lastrowid
row = connection.execute("SELECT * FROM governance_ledger WHERE id = ?", (row_id,)).fetchone()
return LedgerEntry(
id=row["id"],
venue_id=row["venue_id"],
round_id=row["round_id"],
model_version=row["model_version"],
checksum=row["checksum"],
sample_count=row["sample_count"],
privacy_budget=row["privacy_budget"],
created_at=row["created_at"],
)
def list_entries(self) -> list[LedgerEntry]:
with self._connect() as connection:
rows = connection.execute("SELECT * FROM governance_ledger ORDER BY id ASC").fetchall()
return [
LedgerEntry(
id=row["id"],
venue_id=row["venue_id"],
round_id=row["round_id"],
model_version=row["model_version"],
checksum=row["checksum"],
sample_count=row["sample_count"],
privacy_budget=row["privacy_budget"],
created_at=row["created_at"],
)
for row in rows
]

88
src/openimpact/model.py Normal file
View File

@ -0,0 +1,88 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Iterable, Sequence
import numpy as np
from .contracts import LocalRequest, PlanDelta
FEATURE_NAMES = (
"intercept",
"signed_quantity",
"log_quantity",
"mid_price",
"volatility",
"spread_bps",
"order_imbalance",
"queue_position",
"signed_quantity_x_volatility",
)
def build_design_matrix(requests: Sequence[LocalRequest]) -> np.ndarray:
return np.vstack([request.to_features() for request in requests]) if requests else np.zeros((0, len(FEATURE_NAMES)))
@dataclass
class TemporaryImpactModel:
regularization: float = 1e-3
coefficients_: np.ndarray = field(default_factory=lambda: np.zeros(len(FEATURE_NAMES), dtype=float))
def fit(
self,
requests: Sequence[LocalRequest],
realized_impact: Sequence[float],
sample_weight: Sequence[float] | None = None,
) -> "TemporaryImpactModel":
x = build_design_matrix(requests)
y = np.asarray(realized_impact, dtype=float)
if x.size == 0:
raise ValueError("cannot fit empty impact dataset")
if sample_weight is None:
w = np.ones(len(y), dtype=float)
else:
w = np.asarray(sample_weight, dtype=float)
if len(w) != len(y):
raise ValueError("sample_weight must match realized_impact length")
sqrt_w = np.sqrt(w)[:, None]
xw = x * sqrt_w
yw = y * np.sqrt(w)
reg = self.regularization * np.eye(x.shape[1], dtype=float)
reg[0, 0] = 0.0
lhs = xw.T @ xw + reg
rhs = xw.T @ yw
self.coefficients_ = np.linalg.solve(lhs, rhs)
return self
def predict(self, requests: Sequence[LocalRequest]) -> np.ndarray:
x = build_design_matrix(requests)
if x.size == 0:
return np.asarray([], dtype=float)
return x @ self.coefficients_
def clone_with_coefficients(self, coefficients: Sequence[float]) -> "TemporaryImpactModel":
model = TemporaryImpactModel(regularization=self.regularization)
model.coefficients_ = np.asarray(coefficients, dtype=float)
return model
def to_plan_deltas(self, requests: Sequence[LocalRequest], confidence: float = 0.8) -> list:
predictions = self.predict(requests)
plan_deltas = []
for request, impact in zip(requests, predictions, strict=True):
limit_price = request.mid_price + impact if request.side.lower() == "buy" else request.mid_price - impact
plan_deltas.append(
PlanDelta(
venue_id=request.venue_id,
instrument=request.instrument,
side=request.side,
quantity=request.quantity,
limit_price=float(limit_price),
expected_impact=float(impact),
confidence=confidence,
rationale="temporary-impact estimate",
)
)
return plan_deltas

29
src/openimpact/pipeline.py Normal file
View File

@ -0,0 +1,29 @@
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from .evaluation import EvaluationMetrics, evaluate_federated_setup
from .federated import FederatedTrainer
from .ledger import GovernanceLedger
from .synthetic import SyntheticMarket
@dataclass(frozen=True)
class PipelineReport:
metrics: EvaluationMetrics
aggregate_checksum: str
ledger_rows: int
class OpenImpactPlatform:
def __init__(self, ledger_path: str | Path = "openimpact-ledger.sqlite3", seed: int = 7):
self.ledger = GovernanceLedger(ledger_path)
self.trainer = FederatedTrainer(seed=seed)
def run(self, market: SyntheticMarket) -> PipelineReport:
metrics, _, _ = evaluate_federated_setup(market, self.trainer)
result = self.trainer.fit(market.batches)
for round_id, update in enumerate(result.updates, start=1):
self.ledger.record_update(round_id=round_id, model_version="v1", update=update)
return PipelineReport(metrics=metrics, aggregate_checksum=result.aggregate_checksum, ledger_rows=len(self.ledger.list_entries()))

48
src/openimpact/replay.py Normal file
View File

@ -0,0 +1,48 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Sequence
import numpy as np
from .contracts import LocalRequest
from .model import TemporaryImpactModel
@dataclass(frozen=True)
class ReplayStep:
venue_id: str
timestamp_ns: int
delivery_ns: int
predicted_impact: float
realized_impact: float | None
class DeterministicReplayEngine:
def __init__(self, base_latency_ms: dict[str, float] | None = None, seed: int = 7):
self.base_latency_ms = base_latency_ms or {}
self.seed = seed
def replay(
self,
requests: Sequence[LocalRequest],
model: TemporaryImpactModel,
realized_impact: Sequence[float] | None = None,
) -> list[ReplayStep]:
predictions = model.predict(requests)
realized = list(realized_impact) if realized_impact is not None else [None] * len(requests)
steps: list[ReplayStep] = []
for index, (request, prediction, actual) in enumerate(zip(requests, predictions, realized, strict=True)):
digest = np.frombuffer(f"{self.seed}:{request.venue_id}:{index}".encode("utf-8"), dtype=np.uint8)
jitter_ms = float(digest.sum() % 11) / 10.0
latency_ms = self.base_latency_ms.get(request.venue_id, 1.0) + jitter_ms
steps.append(
ReplayStep(
venue_id=request.venue_id,
timestamp_ns=request.timestamp_ns,
delivery_ns=request.timestamp_ns + int(latency_ms * 1_000_000),
predicted_impact=float(prediction),
realized_impact=None if actual is None else float(actual),
)
)
return steps

80
src/openimpact/synthetic.py Normal file
View File

@ -0,0 +1,80 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Sequence
import numpy as np
from .contracts import LocalRequest, VenueBatch, flatten_batches
@dataclass(frozen=True)
class SyntheticMarketConfig:
venue_count: int = 2
samples_per_venue: int = 128
seed: int = 7
@dataclass(frozen=True)
class SyntheticMarket:
batches: tuple[VenueBatch, ...]
@property
def requests(self) -> list[LocalRequest]:
requests, _ = flatten_batches(self.batches)
return requests
@property
def impacts(self) -> np.ndarray:
_, impacts = flatten_batches(self.batches)
return impacts
def _venue_request(rng: np.random.Generator, venue_id: str, index: int) -> LocalRequest:
side = "buy" if rng.random() > 0.5 else "sell"
quantity = float(rng.lognormal(mean=2.8, sigma=0.35))
mid_price = float(90.0 + rng.normal(0.0, 4.0))
volatility = float(abs(rng.normal(0.02, 0.007)))
spread_bps = float(abs(rng.normal(4.0, 1.2)))
order_imbalance = float(rng.uniform(-1.0, 1.0))
queue_position = float(rng.uniform(0.0, 1.0))
return LocalRequest(
venue_id=venue_id,
instrument="OPEN",
side=side,
quantity=quantity,
mid_price=mid_price,
volatility=volatility,
spread_bps=spread_bps,
timestamp_ns=1_000_000_000 + index * 1_000_000,
order_imbalance=order_imbalance,
queue_position=queue_position,
)
def generate_synthetic_market(config: SyntheticMarketConfig) -> SyntheticMarket:
rng = np.random.default_rng(config.seed)
batches: list[VenueBatch] = []
beta = np.array([0.01, 0.015, 0.02, 0.0, 0.2, 0.08, 0.05, 0.015, 0.09], dtype=float)
for venue_index in range(config.venue_count):
venue_id = f"venue-{venue_index + 1}"
requests: list[LocalRequest] = []
impacts: list[float] = []
for sample_index in range(config.samples_per_venue):
request = _venue_request(rng, venue_id, sample_index + venue_index * config.samples_per_venue)
features = request.to_features()
venue_noise = rng.normal(0.0, 0.01)
impact = float(features @ beta + venue_noise)
requests.append(request)
impacts.append(impact)
batches.append(
VenueBatch(
venue_id=venue_id,
requests=tuple(requests),
realized_impact=np.asarray(impacts, dtype=float),
metadata={"seed": float(config.seed), "venue_index": float(venue_index)},
)
)
return SyntheticMarket(batches=tuple(batches))

6
test.sh Executable file
View File

@ -0,0 +1,6 @@
#!/usr/bin/env bash
set -euo pipefail
python3 -m pip install -e ".[dev]"
pytest
python3 -m build

53
tests/test_openimpact.py Normal file
View File

@ -0,0 +1,53 @@
from __future__ import annotations
from pathlib import Path
import numpy as np
from openimpact.evaluation import evaluate_federated_setup
from openimpact.federated import FederatedTrainer
from openimpact.ledger import GovernanceLedger
from openimpact.model import TemporaryImpactModel
from openimpact.replay import DeterministicReplayEngine
from openimpact.synthetic import SyntheticMarketConfig, generate_synthetic_market
def test_temporary_impact_model_fits_synthetic_market():
market = generate_synthetic_market(SyntheticMarketConfig(seed=11, venue_count=2, samples_per_venue=64))
requests = market.requests
impacts = market.impacts
model = TemporaryImpactModel().fit(requests, impacts)
predictions = model.predict(requests)
assert predictions.shape == impacts.shape
assert float(np.sqrt(np.mean((predictions - impacts) ** 2))) < 0.05
def test_federated_training_and_evaluation_are_deterministic():
market = generate_synthetic_market(SyntheticMarketConfig(seed=3, venue_count=3, samples_per_venue=32))
trainer = FederatedTrainer(seed=3)
metrics_a, model_a, baseline_a = evaluate_federated_setup(market, trainer)
metrics_b, model_b, baseline_b = evaluate_federated_setup(market, trainer)
assert metrics_a == metrics_b
assert np.allclose(model_a.coefficients_, model_b.coefficients_)
assert np.allclose(baseline_a.coefficients_, baseline_b.coefficients_)
assert metrics_a.rmse < 0.08
def test_replay_and_ledger_work_end_to_end(tmp_path: Path):
market = generate_synthetic_market(SyntheticMarketConfig(seed=5, venue_count=2, samples_per_venue=16))
trainer = FederatedTrainer(seed=5)
result = trainer.fit(market.batches)
replay = DeterministicReplayEngine(seed=5, base_latency_ms={"venue-1": 2.5, "venue-2": 3.5})
steps = replay.replay(market.requests, result.global_model, market.impacts)
ledger = GovernanceLedger(tmp_path / "ledger.sqlite3")
for round_id, update in enumerate(result.updates, start=1):
ledger.record_update(round_id=round_id, model_version="v1", update=update)
entries = ledger.list_entries()
assert len(entries) == 2
assert len(steps) == len(market.requests)
assert steps[0].delivery_ns > steps[0].timestamp_ns