From d6ebbd82e208e38ca666a9158997c4c168d38521 Mon Sep 17 00:00:00 2001 From: agent-dd492b85242a98c5 Date: Mon, 20 Apr 2026 14:43:51 +0200 Subject: [PATCH] build(agent): new-agents-3#dd492b iteration --- tests/test_adapters.py | 10 ++ tests/test_bridge.py | 14 ++ tests/test_contracts.py | 33 ++--- tests/test_registry.py | 7 + .../README.md | 9 ++ .../__init__.py | 17 ++- .../adapters.py | 121 +++++++++--------- .../bridge.py | 57 +++++---- .../contract_registry.py | 60 +++------ .../contracts/__init__.py | 95 +++++++++++--- 10 files changed, 237 insertions(+), 186 deletions(-) create mode 100644 tests/test_adapters.py create mode 100644 tests/test_bridge.py create mode 100644 tests/test_registry.py create mode 100644 tradecipher_blockchain_backed_privacy_pr/README.md diff --git a/tests/test_adapters.py b/tests/test_adapters.py new file mode 100644 index 0000000..3271cc1 --- /dev/null +++ b/tests/test_adapters.py @@ -0,0 +1,10 @@ +from tradecipher_blockchain_backed_privacy_pr.adapters import bootstrap_adapters, get_adapters_registry + + +def test_adapters_bootstrap_and_registry(): + # Bootstrap adapters and verify they are registered + bootstrap_adapters() + registry = get_adapters_registry() + names = registry.list_names() + assert "VenueAAdapter" in names + assert "VenueBAdapter" in names diff --git a/tests/test_bridge.py b/tests/test_bridge.py new file mode 100644 index 0000000..180b84a --- /dev/null +++ b/tests/test_bridge.py @@ -0,0 +1,14 @@ +import json + +from tradecipher_blockchain_backed_privacy_pr.bridge import EnergiBridge +from tradecipher_blockchain_backed_privacy_pr.contracts import LocalTrade + + +def test_bridge_roundtrip_localtrade(): + bridge = EnergiBridge() + lt = LocalTrade(id="T1", symbol="AAPL", quantity=10.0, price=150.0) + canonical = bridge.to_canonical(lt) + # canonical payload should be JSON string by design + assert canonical["type"] == "LocalTrade" + recovered = bridge.from_canonical(canonical) + assert recovered == lt diff --git a/tests/test_contracts.py b/tests/test_contracts.py index a08e881..c7c1c1a 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -1,28 +1,11 @@ import json -from tradecipher_blockchain_backed_privacy_pr.contracts import LocalTrade, SharedSignals, PlanDelta +import pytest + +from tradecipher_blockchain_backed_privacy_pr.contracts import LocalTrade -def test_local_trade_json_roundtrip(): - lt = LocalTrade( - trade_id="T1", - product_id="P-XYZ", - venue="VenueA", - price=123.45, - size=10, - side="buy", - timestamp=1620000000.0, - signer_id="Signer1", - signature="sig123", - ) - j = lt.to_json() - # Ensure it's valid JSON and contains expected fields - obj = json.loads(j) - assert isinstance(obj, dict) - assert obj["trade_id"] == "T1" - - -def test_shared_signals_and_plan_delta_structure(): - ss = SharedSignals(signals_version=1, aggregated_risk=0.5, margin_impact=0.1, liquidity_metric=0.9) - pd = PlanDelta(delta_version=1, timestamp=1620000001.0, allocations={"T1": 0.5}, cryptographic_tag="tag") - assert isinstance(ss.to_json(), str) - assert isinstance(pd.to_json(), str) +def test_localtrade_json_roundtrip(): + lt = LocalTrade(id="T1", symbol="AAPL", quantity=100.0, price=150.0) + s = lt.to_json() + lt2 = LocalTrade.from_json(s) + assert lt == lt2 diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..829c18e --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,7 @@ +from tradecipher_blockchain_backed_privacy_pr.contract_registry import ContractRegistry + + +def test_contract_registry_basic(): + registry = ContractRegistry() + registry.register_contract("LocalTrade", "v1", {"fields": ["id", "symbol"]}) + assert registry.get_contract("LocalTrade", "v1") == {"fields": ["id", "symbol"]} diff --git a/tradecipher_blockchain_backed_privacy_pr/README.md b/tradecipher_blockchain_backed_privacy_pr/README.md new file mode 100644 index 0000000..fbee7c0 --- /dev/null +++ b/tradecipher_blockchain_backed_privacy_pr/README.md @@ -0,0 +1,9 @@ +# TradeCipher MVP Skeleton + +A minimal, testable foundation for canonical data contracts, a bridge, and adapters to enable cross-venue interoperability with privacy-preserving goals. + +- Data contracts: LocalTrade, SharedSignals, PlanDelta, PrivacyBudget, AuditLog, ComplianceEvent (versioned via the in-memory registry). +- Bridge: EnergiBridge-like translator to and from canonical representations. +- Adapters: Two starter venue adapters (VenueAAdapter, VenueBAdapter) with a bootstrapper registry. + +This repository is intended as a reference MVP, not a production system. It provides the core wiring to enable end-to-end testing of the cross-venue workflow in a deterministic and observable manner. diff --git a/tradecipher_blockchain_backed_privacy_pr/__init__.py b/tradecipher_blockchain_backed_privacy_pr/__init__.py index 8efc819..42defbe 100644 --- a/tradecipher_blockchain_backed_privacy_pr/__init__.py +++ b/tradecipher_blockchain_backed_privacy_pr/__init__.py @@ -1,14 +1,17 @@ -"""TradeCipher MVP package +"""TradeCipher MVP Skeleton -A lightweight reference implementation for LocalTrade / SharedSignals / PlanDelta -contracts, a tamper-evident in-memory attestation ledger, and a placeholder -Zero-Knowledge proof generator. This is intentionally minimal to satisfy the -tests and provide a concrete starting point for further work. +Lightweight, testable scaffold for canonical data contracts, a bridge, and +adapters to enable cross-venue interoperability with privacy-preserving goals. """ +from . import contracts # noqa: F401 +from . import contract_registry # noqa: F401 +from . import bridge # noqa: F401 +from . import adapters # noqa: F401 + __all__ = [ "contracts", - "ledger", - "zk_proofs", + "contract_registry", + "bridge", "adapters", ] diff --git a/tradecipher_blockchain_backed_privacy_pr/adapters.py b/tradecipher_blockchain_backed_privacy_pr/adapters.py index 820437e..116cb42 100644 --- a/tradecipher_blockchain_backed_privacy_pr/adapters.py +++ b/tradecipher_blockchain_backed_privacy_pr/adapters.py @@ -1,82 +1,75 @@ +"""Stub venue adapters for MVP wiring.""" from __future__ import annotations -"""Tiny adapters registry for two-venue MVP. +from typing import Dict -This module is deliberately small; it provides a stub registry that could be -extended to map LocalTrade -> canonical representation and to register -simple venue adapters. It demonstrates the intended extension point. -""" - -import json - -ADAPTER_REGISTRY = {} +from .contracts import LocalTrade -def register_adapter(name: str, adapter: object) -> None: - ADAPTER_REGISTRY[name] = adapter +class VenueAdapter: + name: str + + def translate(self, payload: Dict[str, object]) -> LocalTrade: + raise NotImplementedError -def get_adapter(name: str): - return ADAPTER_REGISTRY.get(name) +class VenueAAdapter(VenueAdapter): + name = "VenueAAdapter" + + def translate(self, payload: Dict[str, object]) -> LocalTrade: + # Minimal translation: expect payload has id, symbol, qty, price + return LocalTrade( + id=str(payload.get("id", "A-UNKNOWN")), + symbol=str(payload.get("symbol", "UNKNOWN")), + quantity=float(payload.get("qty", 0.0)), + price=float(payload.get("price", 0.0)), + ) -# Lightweight, two-venue MVP adapters (stubbed for bootstrap). -# These adapters translate between a venue-specific message format and -# the canonical LocalTrade/SharedSignals/PlanDelta representation used -# by the TradeCipher core. The implementations here are intentionally -# minimal placeholders to unblock interoperability testing. +class VenueBAdapter(VenueAdapter): + name = "VenueBAdapter" -class VenueAAdapter: - name = "VenueA" - - @staticmethod - def to_canonical(obj: object) -> dict: - # Placeholder translation: assume obj is already a dict-like - # representation compatible with canonical fields. - if hasattr(obj, "to_json"): - return json.loads(obj.to_json()) # type: ignore - if isinstance(obj, dict): - return obj - return {} # pragma: no cover - - @staticmethod - def from_canonical(data: dict) -> object: - # Return as-is; in a full implementation this would construct - # a venue-specific LocalTrade message. - return data + def translate(self, payload: Dict[str, object]) -> LocalTrade: + return LocalTrade( + id=str(payload.get("order_id", "B-UNKNOWN")), + symbol=str(payload.get("ticker", "UNKNOWN")), + quantity=float(payload.get("size", 0.0)), + price=float(payload.get("limit", 0.0)), + ) -class VenueBAdapter: - name = "VenueB" +class AdaptersRegistry: + def __init__(self) -> None: + self._adapters: Dict[str, VenueAdapter] = {} - @staticmethod - def to_canonical(obj: object) -> dict: - if hasattr(obj, "to_json"): - return json.loads(obj.to_json()) # type: ignore - if isinstance(obj, dict): - return obj - return {} + def register(self, adapter: VenueAdapter) -> None: + self._adapters[adapter.name] = adapter - @staticmethod - def from_canonical(data: dict) -> object: - return data + def get(self, name: str) -> VenueAdapter: + return self._adapters[name] + + def list_names(self) -> list: + return list(self._adapters.keys()) -def register_default_adapters() -> None: - """Register default, stub adapters for VenueA and VenueB. - - This is a minimal bootstrap to enable end-to-end wiring in tests - and examples. Real adapters would implement complete mapping logic. - """ - register_adapter(VenueAAdapter.name, VenueAAdapter) - register_adapter(VenueBAdapter.name, VenueBAdapter) +_REGISTRY = AdaptersRegistry() -# Autoload default adapters on import, so downstream code can rely on them -# being present without extra setup. This keeps the module side-effect-free -# for testing environments that import adapters without needing a runner. -try: - register_default_adapters() -except Exception: - # If anything goes wrong in bootstrap, fail softly to not break local tests - pass +def bootstrap_adapters() -> None: + # Install starter adapters into the registry + _REGISTRY.register(VenueAAdapter()) + _REGISTRY.register(VenueBAdapter()) + + +def get_adapters_registry() -> AdaptersRegistry: + return _REGISTRY + + +__all__ = [ + "VenueAdapter", + "VenueAAdapter", + "VenueBAdapter", + "AdaptersRegistry", + "bootstrap_adapters", + "get_adapters_registry", +] diff --git a/tradecipher_blockchain_backed_privacy_pr/bridge.py b/tradecipher_blockchain_backed_privacy_pr/bridge.py index 2663da4..a22377b 100644 --- a/tradecipher_blockchain_backed_privacy_pr/bridge.py +++ b/tradecipher_blockchain_backed_privacy_pr/bridge.py @@ -1,35 +1,38 @@ -"""EnergiBridge: a minimal canonical bridge for TradeCipher interoperability. +"""Minimal EnergiBridge-like canonical bridge. -This is an ultra-lightweight skeleton that translates between venue-specific -messages and the canonical in-repo representation used by the core contracts. -It intentionally avoids external dependencies and focuses on wiring and -demonstrating the data flow for MVP testing. +This module translates between venue payloads and a canonical, +off-chain TradeCipher representation using simple dataclass-based +contracts. """ - from __future__ import annotations -from typing import Dict, Any -from .adapters import get_adapter -from .contracts import LocalTrade, SharedSignals, PlanDelta +from typing import Any, Dict + +from .contracts import LocalTrade class EnergiBridge: - @staticmethod - def to_canonical(venue: str, payload: Any) -> Dict[str, Any]: - adapter = get_adapter(venue) - if adapter is None: - # Unknown venue: return a minimal placeholder - return {"venue": venue, "payload": payload} - data = adapter.to_canonical(payload) - # In a real implementation we would validate and normalize the object here - return { - "venue": venue, - "canonical": data, - } + """A tiny translator between venue payloads and canonical objects.""" - @staticmethod - def from_canonical(venue: str, data: Dict[str, Any]) -> Any: - adapter = get_adapter(venue) - if adapter is None: - return data - return adapter.from_canonical(data.get("canonical", data)) + def to_canonical(self, obj: Any) -> Dict[str, Any]: + # For this MVP, we assume objects expose a to_json or __dict__. + if hasattr(obj, "to_json"): + payload = obj.to_json() + return {"type": obj.__class__.__name__, "payload": payload} + elif hasattr(obj, "__dict__"): + return {"type": obj.__class__.__name__, "payload": obj.__dict__} + else: + raise TypeError("Unsupported object for canonicalization") + + def from_canonical(self, data: Dict[str, Any]) -> LocalTrade: + t = data.get("type") + payload = data.get("payload") + # Support LocalTrade payloads serialized as JSON strings + if t == "LocalTrade" and isinstance(payload, str): + return LocalTrade.from_json(payload) + if isinstance(payload, dict) and t == "LocalTrade": + return LocalTrade(**payload) + raise ValueError("Unsupported canonical payload for LocalTrade") + + +__all__ = ["EnergiBridge"] diff --git a/tradecipher_blockchain_backed_privacy_pr/contract_registry.py b/tradecipher_blockchain_backed_privacy_pr/contract_registry.py index e75118c..5acce70 100644 --- a/tradecipher_blockchain_backed_privacy_pr/contract_registry.py +++ b/tradecipher_blockchain_backed_privacy_pr/contract_registry.py @@ -1,50 +1,22 @@ -"""Lightweight contract and schema registry for TradeCipher MVP. - -This is a minimal, in-process registry to track versioned data contracts -and their associated schemas. It is intentionally simple and intended for -bootstrapping interoperability between adapters and contracts. -""" - +"""In-memory contract registry for versioned data contracts.""" from __future__ import annotations -from typing import Dict, Tuple -import json +from typing import Dict, Tuple, Any -# Simple in-memory registry keeping (name -> (version, schema)) -_REGISTRY: Dict[str, Tuple[int, dict]] = {} +class ContractRegistry: + def __init__(self) -> None: + # key: (name, version) -> contract_schema (any JSON-serializable contract descriptor) + self._registry: Dict[Tuple[str, str], Any] = {} + + def register_contract(self, name: str, version: str, schema: Any) -> None: + self._registry[(name, version)] = schema + + def get_contract(self, name: str, version: str) -> Any: + return self._registry.get((name, version)) + + def __contains__(self, item: Tuple[str, str]) -> bool: + return item in self._registry -def register_contract(name: str, version: int, schema: dict) -> None: - """Register or update a contract schema version.""" - _REGISTRY[name] = (version, schema) - - -def get_contract(name: str) -> Tuple[int, dict] | None: - """Return the latest version and schema for a contract, or None.""" - return _REGISTRY.get(name) - - -def list_contracts() -> Dict[str, Tuple[int, dict]]: - return dict(_REGISTRY) - - -def _default_schemas() -> None: - # Seed with a couple of canonical contracts for MVP bootstrap. - register_contract( - "LocalTrade", - 1, - { - "fields": ["trade_id", "product_id", "venue", "price", "size", "side", "timestamp", "signer_id", "signature"] - }, - ) - register_contract( - "SharedSignals", - 1, - { - "fields": ["signals_version", "aggregated_risk", "margin_impact", "liquidity_metric"] - }, - ) - - -_default_schemas() +__all__ = ["ContractRegistry"] diff --git a/tradecipher_blockchain_backed_privacy_pr/contracts/__init__.py b/tradecipher_blockchain_backed_privacy_pr/contracts/__init__.py index 655abaf..48a65ff 100644 --- a/tradecipher_blockchain_backed_privacy_pr/contracts/__init__.py +++ b/tradecipher_blockchain_backed_privacy_pr/contracts/__init__.py @@ -1,46 +1,103 @@ +"""Contract data models for TradeCipher MVP""" from __future__ import annotations -import json from dataclasses import dataclass, asdict -from typing import Dict, Any +import json +from typing import Any, Dict @dataclass class LocalTrade: - trade_id: str - product_id: str - venue: str + id: str + symbol: str + quantity: float price: float - size: int - side: str - timestamp: float - signer_id: str - signature: str def to_json(self) -> str: return json.dumps(asdict(self)) + @staticmethod + def from_json(data: str) -> "LocalTrade": + obj = json.loads(data) + return LocalTrade(**obj) + @dataclass class SharedSignals: - signals_version: int - aggregated_risk: float - margin_impact: float - liquidity_metric: float + version: int + payload: Dict[str, Any] def to_json(self) -> str: return json.dumps(asdict(self)) + @staticmethod + def from_json(data: str) -> "SharedSignals": + obj = json.loads(data) + return SharedSignals(version=obj["version"], payload=obj["payload"]) + @dataclass class PlanDelta: - delta_version: int - timestamp: float - allocations: Dict[str, float] - cryptographic_tag: str + version: int + delta: Dict[str, Any] def to_json(self) -> str: return json.dumps(asdict(self)) + @staticmethod + def from_json(data: str) -> "PlanDelta": + obj = json.loads(data) + return PlanDelta(version=obj["version"], delta=obj["delta"]) -__all__ = ["LocalTrade", "SharedSignals", "PlanDelta"] + +@dataclass +class PrivacyBudget: + total: float + used: float + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + @staticmethod + def from_json(data: str) -> "PrivacyBudget": + obj = json.loads(data) + return PrivacyBudget(**obj) + + +@dataclass +class AuditLog: + entry: str + timestamp: str + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + @staticmethod + def from_json(data: str) -> "AuditLog": + obj = json.loads(data) + return AuditLog(entry=obj["entry"], timestamp=obj["timestamp"]) + + +@dataclass +class ComplianceEvent: + id: str + rule: str + status: str + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + @staticmethod + def from_json(data: str) -> "ComplianceEvent": + obj = json.loads(data) + return ComplianceEvent(**obj) + + +__all__ = [ + "LocalTrade", + "SharedSignals", + "PlanDelta", + "PrivacyBudget", + "AuditLog", + "ComplianceEvent", +]