From fcdd3d0e6436475fdf222acf2c3b8828d95b0e64 Mon Sep 17 00:00:00 2001 From: agent-7e3bbc424e07835b Date: Fri, 24 Apr 2026 17:49:31 +0200 Subject: [PATCH] build(agent): new-agents-2#7e3bbc iteration --- README.md | 10 ++--- crossvenue_arbx/__init__.py | 2 + crossvenue_arbx/adapters.py | 30 ++++++++------ crossvenue_arbx/bridge.py | 28 ++++++------- crossvenue_arbx/coordinator.py | 42 +++++++++++++------- crossvenue_arbx/core.py | 72 ++++++++++++++++++++++++++++++---- crossvenue_arbx/governance.py | 13 ++++-- tests/test_bridge.py | 24 +++++++++++- tests/test_core.py | 31 ++++++++++++++- tests/test_end_to_end.py | 20 ++++++++++ 10 files changed, 212 insertions(+), 60 deletions(-) diff --git a/README.md b/README.md index 56fb808..112f331 100644 --- a/README.md +++ b/README.md @@ -3,23 +3,23 @@ CrossVenueArbX is a lightweight, privacy-aware framework for federated discovery and execution of cross-venue equity arbitrage signals. - Core IR: LocalArbProblem, SharedSignals, PlanDelta, DualVariables, AuditLog, PrivacyBudget. -- Lightweight coordination: Async ADMM-lite with deterministic replay and bounded staleness. +- Deterministic coordination: replay-stable plan signatures, stable merges, and reconnect reconciliation. - Privacy by design: secure aggregation stubs, optional local DP budgets, cryptographic attestation of adapters. -- Adapters and data feeds: starter price-feed adapters and a minimal broker adapter for execution. -- Governance and provenance: tamper-evident audit logs and a Graph-of-Contracts registry. +- Adapters and data feeds: deterministic starter price-feed adapters and a broker adapter for execution. +- Governance and provenance: tamper-evident audit logs, contract fingerprints, and a Graph-of-Contracts registry. - MVP pipeline: Phase 0 skeleton with 2 starter adapters; Phase 1 governance ledger; Phase 2 cross-venue demo; Phase 3 production-like testing and SDK. Getting Started - Run tests and build: `bash test.sh`. - Run the demo locally: `python -m crossvenue_arbx.demo --iterations 3`. - View coverage and governance artefacts in `crossvenue_arbx/governance.py` and `crossvenue_arbx/core.py`. -- Interoperability: a lightweight EnergiBridge (crossvenue_arbx.bridge) translates CrossVenueArbX primitives into a canonical CatOpt-IR for integration with external ecosystems. +- Interoperability: a lightweight EnergiBridge (`crossvenue_arbx.bridge`) translates CrossVenueArbX primitives into a canonical CatOpt-IR for integration with external ecosystems. Architecture Summary - Local signal discovery per venue; aggregated signals managed by a central coordinator. - PlanDelta encodes actionable cross-venue arb legs. - Graph-of-Contracts provides versioning for adapters and data contracts. -- CRDT-like delta merges enable deterministic replay after reconnects. +- CRDT-like delta merges enable deterministic replay after reconnects, with canonical fingerprints for governance and audit trails. Usage and API surface - LocalArbProblem: per-asset, per-venue problem statement. diff --git a/crossvenue_arbx/__init__.py b/crossvenue_arbx/__init__.py index 8bf6b47..5d61ac6 100644 --- a/crossvenue_arbx/__init__.py +++ b/crossvenue_arbx/__init__.py @@ -4,6 +4,7 @@ from .core import LocalArbProblem, SharedSignals, PlanDelta, DualVariables, Audi from .adapters import PriceFeedAdapter, BrokerAdapter from .coordinator import CentralCoordinator from .governance import GraphOfContracts +from .bridge import EnergiBridge __all__ = [ "LocalArbProblem", @@ -16,4 +17,5 @@ __all__ = [ "BrokerAdapter", "CentralCoordinator", "GraphOfContracts", + "EnergiBridge", ] diff --git a/crossvenue_arbx/adapters.py b/crossvenue_arbx/adapters.py index 43cf376..6db9268 100644 --- a/crossvenue_arbx/adapters.py +++ b/crossvenue_arbx/adapters.py @@ -5,6 +5,11 @@ from typing import Dict, Any from .core import LocalArbProblem, SharedSignals, PlanDelta +def _deterministic_delta(seed: str) -> float: + basis = sum(ord(ch) for ch in seed) + return 0.0002 * (basis % 5) + + class PriceFeedAdapter: """Adapter A: emits LocalArbProblem + SharedSignals on a fixed cadence.""" @@ -14,18 +19,17 @@ class PriceFeedAdapter: self.version = 1 def step(self) -> tuple[LocalArbProblem, SharedSignals]: - # Create a simple local arb problem and some market deltas prob = LocalArbProblem( id=f"{self.venue}-p1", venue=self.venue, assets=self.assets, - target_misprice=0.001, # placeholder target + target_misprice=0.001, max_exposure=100000.0, latency_budget=0.1, ) signals = SharedSignals( version=self.version, - price_delta_by_asset={a: 0.0001 * (hash(a) % 5) for a in self.assets}, + price_delta_by_asset={a: _deterministic_delta(f"{self.venue}:{a}") for a in self.assets}, cross_corr={(a1, a2): 0.1 for a1 in self.assets for a2 in self.assets if a1 != a2}, liquidity_estimates={a: 1.0 for a in self.assets}, ) @@ -40,19 +44,23 @@ class BrokerAdapter: self.venue = venue def execute(self, plan: PlanDelta) -> Dict[str, Any]: - # Simulate a fill with deterministic outcome based on delta_actions fills = [] for action in plan.delta_actions: - fills.append({ - "asset": action.get("asset"), - "size": action.get("size"), - "from": action.get("from_venue"), - "to": action.get("to_venue"), - "status": "filled", - }) + fills.append( + { + "asset": action.get("asset"), + "size": action.get("size"), + "from": action.get("from_venue"), + "to": action.get("to_venue"), + "status": "filled", + "fee_bps": 1.0, + } + ) return { "venue": self.venue, "timestamp": plan.timestamp, "fills": fills, "ack": True, + "contract_id": plan.contract_id, + "signature": plan.signature, } diff --git a/crossvenue_arbx/bridge.py b/crossvenue_arbx/bridge.py index 80f345b..cd2774c 100644 --- a/crossvenue_arbx/bridge.py +++ b/crossvenue_arbx/bridge.py @@ -12,7 +12,7 @@ Rationale: - Supports deterministic replay semantics by producing stable, versioned IR blocks """ -from typing import Any, Dict +from typing import Any, Dict, Optional from .core import LocalArbProblem, SharedSignals, PlanDelta @@ -21,7 +21,11 @@ class EnergiBridge: """Bridge utility to convert CrossVenueArbX primitives to a CatOpt-like IR.""" @staticmethod - def to_catopt(problem: LocalArbProblem, signals: SharedSignals) -> Dict[str, Any]: + def to_catopt( + problem: LocalArbProblem, + signals: SharedSignals, + plan_delta: Optional[PlanDelta] = None, + ) -> Dict[str, Any]: """Convert a LocalArbProblem and its associated SharedSignals into a canonical, vendor-agnostic representation. @@ -31,26 +35,19 @@ class EnergiBridge: the MVP lean while still providing a predictable schema. """ catopt: Dict[str, Any] = { - "version": max(getattr(problem, "version", 0), getattr(signals, "version", 0)) - if hasattr(problem, "version") or hasattr(signals, "version") - else 1, + "version": max(getattr(problem, "version", 0), signals.version, 1), + "contract_id": problem.fingerprint(), "objects": { - "LocalArbProblem": problem.to_dict(), - "SharedSignals": { - "version": signals.version, - "price_delta_by_asset": dict(signals.price_delta_by_asset), - "cross_corr": dict(signals.cross_corr), - "liquidity_estimates": dict(signals.liquidity_estimates), - }, + "LocalArbProblem": {**problem.to_dict(), "fingerprint": problem.fingerprint()}, + "SharedSignals": {**signals.to_dict(), "fingerprint": signals.fingerprint()}, }, "morphisms": { "SharedSignals": { "version": signals.version, - # shallow representation; real implementations may include attestations - "connections": list(signals.cross_corr.keys()), + "connections": sorted((repr(key) for key in signals.cross_corr.keys())), } }, - "plan_delta_ref": None, # linked PlanDelta would be populated at runtime + "plan_delta_ref": plan_delta.to_dict() if plan_delta is not None else None, } return catopt @@ -64,4 +61,5 @@ class EnergiBridge: return { "objects": catopt.get("objects", {}), "morphisms": catopt.get("morphisms", {}), + "plan_delta_ref": catopt.get("plan_delta_ref"), } diff --git a/crossvenue_arbx/coordinator.py b/crossvenue_arbx/coordinator.py index 7fceff1..040677b 100644 --- a/crossvenue_arbx/coordinator.py +++ b/crossvenue_arbx/coordinator.py @@ -1,6 +1,5 @@ from __future__ import annotations -import time from typing import Dict, List, Any from .core import LocalArbProblem, SharedSignals, PlanDelta @@ -25,6 +24,16 @@ class CentralCoordinator: # Optional bridge for canonical IR interoperability (can be used by adapters) self.bridge = EnergiBridge() self.last_catopt: Dict[str, Any] | None = None + self.replay_log: List[str] = [] + + def _other_venue(self, venue: str) -> str: + venues = sorted(self.shared_signals_by_venue) + if len(venues) < 2: + return "VenueB" if venue != "VenueB" else "VenueA" + for candidate in venues: + if candidate != venue: + return candidate + return venue def ingest_local(self, risk_source: LocalArbProblem, signals: SharedSignals) -> PlanDelta | None: self.version += 1 @@ -35,46 +44,49 @@ class CentralCoordinator: except Exception: # Non-critical for MVP; retain existing behavior on bridge failure self.last_catopt = None - # Naive cross-venue decision: if any price_delta_by_asset exceeds threshold, propose cross-venue move delta = [] - for asset, delta_price in signals.price_delta_by_asset.items(): + for asset in sorted(signals.price_delta_by_asset): + delta_price = signals.price_delta_by_asset[asset] if abs(delta_price) > 0.0005: - # simple dummy cross-venue action: move asset from venue A to venue B (names inferred) action = { "from_venue": risk_source.venue, - "to_venue": "VenueB" if risk_source.venue != "VenueB" else "VenueA", + "to_venue": self._other_venue(risk_source.venue), "asset": asset, - "size": 10.0 * abs(delta_price), - "time": time.time(), + "size": min(risk_source.max_exposure, 10_000.0 * abs(delta_price)), + "time": float(self.version), } delta.append(action) if delta: + contract_id = risk_source.fingerprint() plan = PlanDelta( delta_actions=delta, - timestamp=time.time(), - contract_id="contract-1", - signature=f"sig-{self.version}", + timestamp=float(self.version), + contract_id=contract_id, ) self.last_plan = plan self.pending_delta_actions.append(delta) + self.contracts.register(contract_id, adapter_version=str(signals.version), schema_hash=plan.signature) + self.replay_log.append(plan.signature) return plan return None def reconcile(self) -> PlanDelta | None: - # Deterministic delta reconciliation on reconnect: merge all pending deltas if not self.pending_delta_actions: return None - # Flatten actions and sort by a stable key merged = [] for d in self.pending_delta_actions: merged.extend(d) merged.sort(key=lambda a: (a.get("from_venue"), a.get("to_venue"), a.get("asset"))) + contract_id = "merged:" + ",".join( + sorted({action.get("asset", "") for action in merged}) + ) plan = PlanDelta( delta_actions=merged, - timestamp=time.time(), - contract_id="contract-merged", - signature=f"sig-reconciled-{len(self.pending_delta_actions)}", + timestamp=float(self.version), + contract_id=contract_id, ) self.pending_delta_actions.clear() self.last_plan = plan + self.contracts.register(contract_id, adapter_version="reconciled", schema_hash=plan.signature) + self.replay_log.append(plan.signature) return plan diff --git a/crossvenue_arbx/core.py b/crossvenue_arbx/core.py index f660660..43dc039 100644 --- a/crossvenue_arbx/core.py +++ b/crossvenue_arbx/core.py @@ -1,9 +1,26 @@ from __future__ import annotations +import hashlib +import json from dataclasses import dataclass, field from typing import Dict, List, Any +def _stable_json(payload: Dict[str, Any]) -> str: + def normalize(value: Any) -> Any: + if isinstance(value, dict): + return {str(key): normalize(val) for key, val in sorted(value.items(), key=lambda item: _stable_key(item[0]))} + if isinstance(value, (list, tuple)): + return [normalize(item) for item in value] + return value + + return json.dumps(normalize(payload), sort_keys=True, separators=(",", ":"), default=str) + + +def _stable_key(item: Any) -> str: + return repr(item) + + @dataclass class LocalArbProblem: id: str @@ -23,27 +40,38 @@ class LocalArbProblem: "latency_budget": self.latency_budget, } + def fingerprint(self) -> str: + return hashlib.sha256(_stable_json(self.to_dict()).encode("utf-8")).hexdigest() + @dataclass class SharedSignals: version: int price_delta_by_asset: Dict[str, float] = field(default_factory=dict) - cross_corr: Dict[str, float] = field(default_factory=dict) + cross_corr: Dict[Any, float] = field(default_factory=dict) liquidity_estimates: Dict[str, float] = field(default_factory=dict) + def to_dict(self) -> Dict[str, Any]: + return { + "version": self.version, + "price_delta_by_asset": dict(self.price_delta_by_asset), + "cross_corr": dict(self.cross_corr), + "liquidity_estimates": dict(self.liquidity_estimates), + } + + def fingerprint(self) -> str: + return hashlib.sha256(_stable_json(self.to_dict()).encode("utf-8")).hexdigest() + def merge(self, other: "SharedSignals") -> "SharedSignals": - # Simple deterministic merge: take max delta per asset and average correlations merged = SharedSignals(version=max(self.version, other.version)) - # merge price deltas by taking the maximum absolute delta for stability - keys = set(self.price_delta_by_asset) | set(other.price_delta_by_asset) + keys = sorted(set(self.price_delta_by_asset) | set(other.price_delta_by_asset), key=_stable_key) for k in keys: a = self.price_delta_by_asset.get(k, 0.0) b = other.price_delta_by_asset.get(k, 0.0) merged.price_delta_by_asset[k] = a if abs(a) >= abs(b) else b - # simple average for correlations and liquidity estimates - for k in set(self.cross_corr) | set(other.cross_corr): + for k in sorted(set(self.cross_corr) | set(other.cross_corr), key=_stable_key): merged.cross_corr[k] = (self.cross_corr.get(k, 0.0) + other.cross_corr.get(k, 0.0)) / 2.0 - for k in set(self.liquidity_estimates) | set(other.liquidity_estimates): + for k in sorted(set(self.liquidity_estimates) | set(other.liquidity_estimates), key=_stable_key): merged.liquidity_estimates[k] = ( self.liquidity_estimates.get(k, 0.0) + other.liquidity_estimates.get(k, 0.0) ) / 2.0 @@ -55,7 +83,11 @@ class PlanDelta: delta_actions: List[Dict[str, Any]] timestamp: float contract_id: str - signature: str + signature: str = "" + + def __post_init__(self) -> None: + if not self.signature: + self.signature = self.fingerprint() def to_dict(self) -> Dict[str, Any]: return { @@ -65,23 +97,47 @@ class PlanDelta: "signature": self.signature, } + def fingerprint(self) -> str: + payload = { + "contract_id": self.contract_id, + "delta_actions": self.delta_actions, + "timestamp": self.timestamp, + } + return hashlib.sha256(_stable_json(payload).encode("utf-8")).hexdigest() + @dataclass class DualVariables: shadow_price: float = 0.0 + def to_dict(self) -> Dict[str, Any]: + return {"shadow_price": self.shadow_price} + @dataclass class AuditLog: entries: List[str] = field(default_factory=list) + hashes: List[str] = field(default_factory=list) def log(self, msg: str) -> None: + prev = self.hashes[-1] if self.hashes else "" + entry_hash = hashlib.sha256(f"{prev}|{msg}".encode("utf-8")).hexdigest() self.entries.append(msg) + self.hashes.append(entry_hash) + + def root_hash(self) -> str: + return self.hashes[-1] if self.hashes else "" @dataclass class PrivacyBudget: leakage_budget: float + def remaining(self) -> float: + return self.leakage_budget + + def can_consume(self, amount: float) -> bool: + return amount <= self.leakage_budget + def consume(self, amount: float) -> None: self.leakage_budget = max(0.0, self.leakage_budget - amount) diff --git a/crossvenue_arbx/governance.py b/crossvenue_arbx/governance.py index b52c52d..dbe7ae7 100644 --- a/crossvenue_arbx/governance.py +++ b/crossvenue_arbx/governance.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Dict, List +from typing import Dict, List, Optional @dataclass @@ -9,8 +9,13 @@ class GraphOfContracts: """A simple in-memory registry for contract adapters and versions.""" contracts: Dict[str, Dict[str, str]] = field(default_factory=dict) - def register(self, contract_id: str, adapter_version: str) -> None: - self.contracts[contract_id] = {"adapter_version": adapter_version} + def register(self, contract_id: str, adapter_version: str, schema_hash: str = "") -> Dict[str, str]: + record = {"adapter_version": adapter_version, "schema_hash": schema_hash} + self.contracts[contract_id] = record + return record - def get_version(self, contract_id: str) -> str | None: + def get_version(self, contract_id: str) -> Optional[str]: return self.contracts.get(contract_id, {}).get("adapter_version") + + def get_record(self, contract_id: str) -> Dict[str, str]: + return dict(self.contracts.get(contract_id, {})) diff --git a/tests/test_bridge.py b/tests/test_bridge.py index 0f17a8d..d20d94c 100644 --- a/tests/test_bridge.py +++ b/tests/test_bridge.py @@ -1,5 +1,5 @@ from crossvenue_arbx.bridge import EnergiBridge -from crossvenue_arbx.core import LocalArbProblem, SharedSignals +from crossvenue_arbx.core import LocalArbProblem, SharedSignals, PlanDelta def test_to_catopt_basic_structure(): @@ -31,3 +31,25 @@ def test_to_catopt_basic_structure(): assert objs["LocalArbProblem"]["venue"] == "Venue1" assert "SharedSignals" in objs assert objs["SharedSignals"]["version"] == 1 + + +def test_to_catopt_with_plan_delta_round_trip(): + problem = LocalArbProblem( + id="venue1-p1", + venue="Venue1", + assets=["AAPL"], + target_misprice=0.001, + max_exposure=100000.0, + latency_budget=0.1, + ) + signals = SharedSignals(version=2, price_delta_by_asset={"AAPL": 0.001}) + plan = PlanDelta( + delta_actions=[{"from_venue": "Venue1", "to_venue": "Venue2", "asset": "AAPL", "size": 1.0, "time": 2.0}], + timestamp=2.0, + contract_id=problem.fingerprint(), + ) + + catopt = EnergiBridge.to_catopt(problem, signals, plan) + decoded = EnergiBridge.from_catopt(catopt) + + assert decoded["plan_delta_ref"]["contract_id"] == problem.fingerprint() diff --git a/tests/test_core.py b/tests/test_core.py index c666041..9d192b3 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -4,7 +4,8 @@ import sys BASE = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if BASE not in sys.path: sys.path.insert(0, BASE) -from crossvenue_arbx.core import LocalArbProblem, SharedSignals, PlanDelta +from crossvenue_arbx.core import LocalArbProblem, SharedSignals, PlanDelta, AuditLog, PrivacyBudget +from crossvenue_arbx.governance import GraphOfContracts def test_local_arb_problem_serialization(): @@ -29,3 +30,31 @@ def test_shared_signals_merge_basic(): assert merged.version >= 2 assert "AAPL" in merged.price_delta_by_asset assert "MSFT" in merged.price_delta_by_asset + + +def test_plan_delta_signature_is_stable(): + plan = PlanDelta( + delta_actions=[{"from_venue": "VenueA", "to_venue": "VenueB", "asset": "AAPL", "size": 1.0, "time": 1.0}], + timestamp=1.0, + contract_id="contract-x", + ) + assert plan.signature + assert plan.signature == plan.fingerprint() + + +def test_audit_log_hash_chain_and_privacy_budget(): + log = AuditLog() + log.log("first") + log.log("second") + assert log.root_hash() + budget = PrivacyBudget(leakage_budget=1.0) + assert budget.can_consume(0.25) + budget.consume(0.25) + assert budget.remaining() == 0.75 + + +def test_governance_registry_tracks_metadata(): + registry = GraphOfContracts() + registry.register("contract-x", "1.2.3", "schema-abc") + assert registry.get_version("contract-x") == "1.2.3" + assert registry.get_record("contract-x")["schema_hash"] == "schema-abc" diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py index 241cbb9..805861c 100644 --- a/tests/test_end_to_end.py +++ b/tests/test_end_to_end.py @@ -41,3 +41,23 @@ def test_end_to_end_two_venues_basic_reconciliation(): actions = recon.delta_actions sorted_actions = sorted(actions, key=lambda x: (x.get("from_venue"), x.get("to_venue"), x.get("asset"))) assert actions == sorted_actions + assert recon.signature + + +def test_price_feed_and_execution_are_deterministic(): + ad1 = PriceFeedAdapter("VenueA", ["AAPL", "MSFT"]) + ad2 = PriceFeedAdapter("VenueA", ["AAPL", "MSFT"]) + p1, s1 = ad1.step() + p2, s2 = ad2.step() + + assert p1.fingerprint() == p2.fingerprint() + assert s1.price_delta_by_asset == s2.price_delta_by_asset + + coord1 = CentralCoordinator() + coord2 = CentralCoordinator() + plan1 = coord1.ingest_local(p1, s1) + plan2 = coord2.ingest_local(p2, s2) + + assert plan1 is not None and plan2 is not None + assert plan1.signature == plan2.signature + assert plan1.to_dict() == plan2.to_dict()