diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bd5590b --- /dev/null +++ b/.gitignore @@ -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 diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..7f83b50 --- /dev/null +++ b/AGENTS.md @@ -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` diff --git a/README.md b/README.md index 8aa3189..4ca155a 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,40 @@ -# idea141-openimpact-privacy-preserving +# OpenImpact -Source logic for Idea #141 \ No newline at end of file +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`. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f5f0d20 --- /dev/null +++ b/pyproject.toml @@ -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"] diff --git a/src/openimpact/__init__.py b/src/openimpact/__init__.py new file mode 100644 index 0000000..0a49587 --- /dev/null +++ b/src/openimpact/__init__.py @@ -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", +] diff --git a/src/openimpact/__main__.py b/src/openimpact/__main__.py new file mode 100644 index 0000000..51f78a5 --- /dev/null +++ b/src/openimpact/__main__.py @@ -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() diff --git a/src/openimpact/contracts.py b/src/openimpact/contracts.py new file mode 100644 index 0000000..2c18fbf --- /dev/null +++ b/src/openimpact/contracts.py @@ -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) diff --git a/src/openimpact/evaluation.py b/src/openimpact/evaluation.py new file mode 100644 index 0000000..67785f4 --- /dev/null +++ b/src/openimpact/evaluation.py @@ -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 diff --git a/src/openimpact/federated.py b/src/openimpact/federated.py new file mode 100644 index 0000000..687d651 --- /dev/null +++ b/src/openimpact/federated.py @@ -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, + ) diff --git a/src/openimpact/ledger.py b/src/openimpact/ledger.py new file mode 100644 index 0000000..6df9900 --- /dev/null +++ b/src/openimpact/ledger.py @@ -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 + ] diff --git a/src/openimpact/model.py b/src/openimpact/model.py new file mode 100644 index 0000000..3306e38 --- /dev/null +++ b/src/openimpact/model.py @@ -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 diff --git a/src/openimpact/pipeline.py b/src/openimpact/pipeline.py new file mode 100644 index 0000000..92919a2 --- /dev/null +++ b/src/openimpact/pipeline.py @@ -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())) diff --git a/src/openimpact/replay.py b/src/openimpact/replay.py new file mode 100644 index 0000000..d2fe42f --- /dev/null +++ b/src/openimpact/replay.py @@ -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 diff --git a/src/openimpact/synthetic.py b/src/openimpact/synthetic.py new file mode 100644 index 0000000..e0669f9 --- /dev/null +++ b/src/openimpact/synthetic.py @@ -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)) diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..fbc1778 --- /dev/null +++ b/test.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +set -euo pipefail + +python3 -m pip install -e ".[dev]" +pytest +python3 -m build diff --git a/tests/test_openimpact.py b/tests/test_openimpact.py new file mode 100644 index 0000000..0a0b02c --- /dev/null +++ b/tests/test_openimpact.py @@ -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