build(agent): melter#14fd4b iteration

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
+@dataclass
-- Enables offline/partitioned operation with deterministic re-assembly.
+class DeltaAction:
-- Minimal surface area to avoid interfering with existing tests and code.
+    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:
+    - Each action carries a version vector (per-venue counters).
-        venue_deltas: mapping from venue_id to a dict of asset_symbol -> delta
+    - 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):
+    def __init__(self, venue_id: str):
-        self.venue_ids: List[str] = list(venue_ids or [])
+        self.venue_id = venue_id
-        self.venue_deltas: Dict[str, Dict[str, float]] = {}
+        # 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:
+    def gen_action(self, asset: str, delta: float, note: Optional[str] = None) -> DeltaAction:
-        """Record the local delta from a venue."""
+        self.counter += 1
-        self.venue_ids = sorted(set(self.venue_ids) | {venue_id})
+        action = DeltaAction(venue=self.venue_id, asset=asset, delta=delta, version={self.venue_id: self.counter}, note=note)
-        self.venue_deltas[venue_id] = dict(delta)
+        self.local_actions[asset] = action
+        return action
 
-    def clear(self) -> None:
+    def apply_remote_action(self, action: DeltaAction) -> None:
-        self.venue_deltas.clear()
+        # 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]:
+    def snapshot(self) -> List[DeltaAction]:
-        """Return a deterministic merged delta across all venues.
+        return list(self.local_actions.values())
-
-        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 deterministic_merge(deltas: List[Dict[str, float]]) -> Dict[str, float]:
+def deterministic_merge(actions: List[Optional[DeltaAction]]) -> Optional[DeltaAction]:
-    """Pure function helper to deterministically merge a list of deltas.
+    """Deterministically merge a list of DeltaAction (some may be None).
 
-    Given a list of per-venue delta dictionaries, return an element-wise maximum
+    Rules:
-    across all dictionaries. If no deltas are provided, returns an empty dict.
+    - 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:
+    filtered = [a for a in actions if a is not None]
-        return {}
+    if not filtered:
-    keys = set().union(*deltas)
+        return None
-    merged: Dict[str, float] = {}
+
-    for k in keys:
+    # Ensure all actions are for the same asset
-        merged[k] = max(d.get(k, 0.0) for d in deltas)
+    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

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