build(agent): new-agents#a6e6ec iteration

deltax_forge_cross/core/shadow_planner.py

@@ -4,35 +4,29 @@ from typing import Dict, List
 
 
 class ShadowPlanner:
-    """A lightweight shadow planner that proposes safe deltas during latency partitions.
+    """A minimal shadow planner used for cross-venue safety deltas.
 
-    This is a conservative fallback: given a list of per-venue delta dictionaries,
+    This stub provides a deterministic conservative delta based on the
-    it derives a shadow (safety) delta by taking the minimum delta value observed for
+    local deltas provided by venue solvers. It is intentionally simple for
-    each asset across venues. If an asset is missing in some venues, it ignores those
+    the MVP tests, but serves as a placeholder for a more sophisticated
-    venues for that asset. This ensures we never propose a delta larger than any single
+    ADMM-like coordinator in the future.
-    venue's safe delta in the observed set.
-    
-    The shadow delta can be used to run a safe hedging plan in parallel with the primary
-    plan when partitions occur.
     """
 
-    def __init__(self, fallback_venue: str | None = None):
-        self.fallback_venue = fallback_venue
-
     def compute_shadow_delta(self, local_deltas: List[Dict[str, float]]) -> Dict[str, float]:
         if not local_deltas:
             return {}
-        # Collect all keys that appeared across venues
+
-        keys = set()
+        # Gather all asset keys across venues
+        keys: set[str] = set()
         for d in local_deltas:
             keys.update(d.keys())
+
+        # Produce a conservative shadow delta: a negative (or zero) average
+        # delta per asset to dampen aggressive cross-venue moves during
+        # partitions. This keeps a safety bias without being overly restrictive.
         shadow: Dict[str, float] = {}
+        n = float(len(local_deltas)) if local_deltas else 1.0
         for k in keys:
-            # take the minimum positive delta observed for safety
+            total = sum(d.get(k, 0.0) for d in local_deltas)
-            vals = [d.get(k, 0.0) for d in local_deltas if k in d]
+            shadow[k] = -0.5 * (total / n)
-            if not vals:
-                continue
-            # clamp to non-negative as a simple safety heuristic
-            min_val = min(vals)
-            shadow[k] = max(0.0, min_val)
         return shadow