From deed2a69dfa83aff9fa00b5569314456774804bd Mon Sep 17 00:00:00 2001 From: agent-14fd4b738639d573 Date: Sat, 25 Apr 2026 21:22:55 +0200 Subject: [PATCH] build(agent): melter#14fd4b iteration --- deltax_forge_cross/core/delta_sync.py | 146 +++++++++++++++++--------- tests/test_delta_sync.py | 31 ++++++ 2 files changed, 128 insertions(+), 49 deletions(-) create mode 100644 tests/test_delta_sync.py diff --git a/deltax_forge_cross/core/delta_sync.py b/deltax_forge_cross/core/delta_sync.py index 4ad9d07..ad11a04 100644 --- a/deltax_forge_cross/core/delta_sync.py +++ b/deltax_forge_cross/core/delta_sync.py @@ -1,68 +1,116 @@ from __future__ import annotations -"""Delta Synchronization helpers (lightweight CRDT-inspired). +from dataclasses import dataclass, field +from typing import Dict, Optional, List, Tuple +import json -This module provides a tiny, production-safe primitive to track per-venue -local deltas and produce a deterministic merged view across venues. It is -intentionally simple: at merge time we compute an element-wise maximum delta -per asset across all known venues. This mirrors a conservative reconstruction -of the cross-venue delta plan when connectivity is partitioned and reconciled. -Rationale -- Enables offline/partitioned operation with deterministic re-assembly. -- Minimal surface area to avoid interfering with existing tests and code. -""" +@dataclass +class DeltaAction: + venue: str + asset: str + delta: float + version: Dict[str, int] = field(default_factory=dict) + note: Optional[str] = None -from typing import Dict, List + def to_dict(self) -> Dict: + return { + "venue": self.venue, + "asset": self.asset, + "delta": self.delta, + "version": dict(self.version), + "note": self.note, + } class DeltaCRDT: - """Lightweight cross-venue delta store with deterministic merge. + """A tiny CRDT-inspired delta synchronizer. - Attributes: - venue_deltas: mapping from venue_id to a dict of asset_symbol -> delta + - Each action carries a version vector (per-venue counters). + - deterministic_merge accepts multiple per-venue action lists and + produces a single, deterministic conservative merge. + + Conflict resolution policy (deterministic & conservative): + - If multiple deltas exist for the same asset, prefer the one with the + smallest absolute magnitude (safety bias). If there's a tie, break + deterministically by lexicographic venue id. + + This is intentionally minimal for the MVP and provides a clear, auditable + merge rule suitable for replay and testing. """ - def __init__(self, venue_ids: List[str] | None = None): - self.venue_ids: List[str] = list(venue_ids or []) - self.venue_deltas: Dict[str, Dict[str, float]] = {} + def __init__(self, venue_id: str): + self.venue_id = venue_id + # local counter for version vector + self.counter = 0 + # local actions by asset (latest action) + self.local_actions: Dict[str, DeltaAction] = {} - def set_venue_delta(self, venue_id: str, delta: Dict[str, float]) -> None: - """Record the local delta from a venue.""" - self.venue_ids = sorted(set(self.venue_ids) | {venue_id}) - self.venue_deltas[venue_id] = dict(delta) + def gen_action(self, asset: str, delta: float, note: Optional[str] = None) -> DeltaAction: + self.counter += 1 + action = DeltaAction(venue=self.venue_id, asset=asset, delta=delta, version={self.venue_id: self.counter}, note=note) + self.local_actions[asset] = action + return action - def clear(self) -> None: - self.venue_deltas.clear() + def apply_remote_action(self, action: DeltaAction) -> None: + # Merge remote action into local_actions using deterministic_merge policy + existing = self.local_actions.get(action.asset) + merged = deterministic_merge([existing, action]) if existing is not None else action + if isinstance(merged, DeltaAction): + self.local_actions[action.asset] = merged + else: + # deterministic_merge may return a new DeltaAction + self.local_actions[action.asset] = merged - def merged(self) -> Dict[str, float]: - """Return a deterministic merged delta across all venues. - - We use an element-wise maximum across all venue deltas to stay conservative - and avoid under-hedging when reconciling. - """ - if not self.venue_deltas: - return {} - keys = set() - for d in self.venue_deltas.values(): - keys.update(d.keys()) - result: Dict[str, float] = {} - for k in keys: - max_val = max((d.get(k, 0.0) for d in self.venue_deltas.values()), default=0.0) - result[k] = max_val - return result + def snapshot(self) -> List[DeltaAction]: + return list(self.local_actions.values()) -def deterministic_merge(deltas: List[Dict[str, float]]) -> Dict[str, float]: - """Pure function helper to deterministically merge a list of deltas. +def deterministic_merge(actions: List[Optional[DeltaAction]]) -> Optional[DeltaAction]: + """Deterministically merge a list of DeltaAction (some may be None). - Given a list of per-venue delta dictionaries, return an element-wise maximum - across all dictionaries. If no deltas are provided, returns an empty dict. + Rules: + - Ignore None entries. + - If only one action present, return it. + - If multiple actions for same asset: choose the one with smallest absolute + delta (conservative). Tie-break by venue id lexicographically. + - The returned action's version vector is the element-wise max of inputs. """ - if not deltas: - return {} - keys = set().union(*deltas) - merged: Dict[str, float] = {} - for k in keys: - merged[k] = max(d.get(k, 0.0) for d in deltas) + filtered = [a for a in actions if a is not None] + if not filtered: + return None + + # Ensure all actions are for the same asset + assets = set(a.asset for a in filtered) + if len(assets) != 1: + raise ValueError("deterministic_merge expects actions for the same asset") + asset = filtered[0].asset + + # Choose candidate by smallest absolute delta, tie-break by venue id + filtered.sort(key=lambda a: (abs(a.delta), a.venue)) + winner = filtered[0] + + # Merge version vectors: take max per key + merged_version: Dict[str, int] = {} + for a in filtered: + for v, c in a.version.items(): + merged_version[v] = max(merged_version.get(v, 0), int(c)) + + # Return a new action representing the merged view + merged_action = DeltaAction(venue=winner.venue, asset=asset, delta=winner.delta, version=merged_version, note=f"merged_from:{json.dumps([a.to_dict() for a in filtered])}") + return merged_action + + +def merge_snapshots(snapshots: List[List[DeltaAction]]) -> Dict[str, DeltaAction]: + """Merge multiple venue snapshots into a deterministic map asset->DeltaAction.""" + by_asset: Dict[str, List[DeltaAction]] = {} + for snap in snapshots: + for a in snap: + by_asset.setdefault(a.asset, []).append(a) + + merged: Dict[str, DeltaAction] = {} + for asset, actions in by_asset.items(): + merged_action = deterministic_merge(actions) + if merged_action is not None: + merged[asset] = merged_action return merged diff --git a/tests/test_delta_sync.py b/tests/test_delta_sync.py new file mode 100644 index 0000000..180f559 --- /dev/null +++ b/tests/test_delta_sync.py @@ -0,0 +1,31 @@ +from deltax_forge_cross.core.delta_sync import DeltaCRDT, DeltaAction, deterministic_merge, merge_snapshots + + +def test_gen_and_snapshot_merge(): + a = DeltaCRDT("VENUE_A") + b = DeltaCRDT("VENUE_B") + + act_a = a.gen_action(asset="AAPL", delta=5.0) + act_b = b.gen_action(asset="AAPL", delta=3.0) + + # Local snapshots + snap_a = a.snapshot() + snap_b = b.snapshot() + + merged = merge_snapshots([snap_a, snap_b]) + assert "AAPL" in merged + # Conservative policy selects smaller absolute delta (3.0) + assert merged["AAPL"].delta == 3.0 + # Merged version should include both counters + assert "VENUE_A" in merged["AAPL"].version and "VENUE_B" in merged["AAPL"].version + + +def test_deterministic_merge_tie_break(): + # Two equal magnitude deltas: tie breaks by venue id lexicographically + a1 = DeltaAction(venue="B", asset="MSFT", delta=2.0, version={"B": 1}) + a2 = DeltaAction(venue="A", asset="MSFT", delta=2.0, version={"A": 1}) + merged = deterministic_merge([a1, a2]) + assert merged is not None + # tie-break picks venue 'A' because ('abs', 'venue') sorts to ('2.0','A') first + assert merged.venue == "A" + assert merged.delta == 2.0