Update Features.py

Features.py

@@ -494,71 +494,243 @@ TIMEFRAMES = {
 
 
 # ============================================================================
-# FEATURE CONTRACT (EXACTLY 60 FEATURES)
+# FEATURE CONTRACT — SINGLE SOURCE OF TRUTH (60 FEATURES)
+# ----------------------------------------------------------------------------
+#
+# ENGINEERING NOTE — why this looks the way it does:
+#
+#   Previously the contract was spread across three top-level sets
+#   (REQUIRED_FEATURES, METADATA_FIELDS, BINARY_FEATURES, ...), with no
+#   runtime check that they were mutually consistent. A drift where a
+#   single key ('price') landed in BOTH the "required features" list AND
+#   the "metadata to strip before validating" set caused the validator to
+#   report {'price'} missing on every tick, which silently shut down
+#   publishing for the entire pipeline.
+#
+#   The fix is structural: ONE FeatureContract object owns the full schema
+#   and checks its own invariants at import time. Any future drift crashes
+#   the module on load with a named offender, instead of corrupting the
+#   wire format at 60Hz for hours.
+#
+#   The old module-level names (REQUIRED_FEATURES, METADATA_FIELDS, etc.)
+#   are kept as PROJECTIONS of the contract for call-site back-compat — the
+#   rest of Features.py can import them exactly as before.
 # ============================================================================
 
-REQUIRED_FEATURES = [
+from dataclasses import dataclass, field
+from typing import FrozenSet, Mapping, Any
+
+CONTRACT_VERSION = "feat-v1.0.0"
+EXPECTED_FEATURE_COUNT = 60
+
+# ---- Feature keys (60) — values fed into model inference ------------------
+_FEATURES: FrozenSet[str] = frozenset({
     # Core Technical (19)
     'log_return', 'rolling_mean_5', 'rolling_std_5', 'zscore_5',
     'rsi_14', 'macd', 'macd_signal', 'macd_hist', 'atr',
     'cdf_value', 'cdf_slope', 'cdf_diff',
     'volatility_quantile_90', 'volatility_ratio', 'entropy_50',
     'autocorr_3', 'momentum_10', 'volume_change_rate', 'volume_zscore',
-    
     # Derivatives (15)
     'price_vel', 'price_acc', 'price_jrk',
     'price_vel_mean', 'price_vel_std', 'price_vel_skew', 'price_vel_kurtosis',
     'price_acc_mean', 'price_acc_std', 'price_acc_skew', 'price_acc_kurtosis',
     'price_jrk_mean', 'price_jrk_std', 'price_jrk_skew', 'price_jrk_kurtosis',
-    
     # Additional Technical (7)
     'ma10', 'ma20', 'std20',
     'bollinger_upper', 'bollinger_lower', 'bollinger_width', 'bollinger_position',
-    
     # Candlestick (9)
     'gravestone_doji', 'four_price_doji', 'doji', 'spinning_top',
     'bullish_candle', 'bearish_candle', 'dragonfly_candle',
     'spinning_top_bearish_followup', 'bullish_then_dragonfly',
-    
-    # Support/Resistance (7)
+    # Support / Resistance (7)
     'distance_to_nearest_support', 'distance_to_nearest_resistance',
     'near_support', 'near_resistance', 'distance_to_stop_loss',
     'support_strength', 'resistance_strength',
-    
-    # Price Variants (3)
-    'price', 'close_scaled', 'close_price'
-]
-
-METADATA_FIELDS = {'timestamp', 'tick_count', 'tick_index', 'timeframe', 'agent', 'feature_count', 'price'}
-
-BINARY_FEATURES = {
-    'near_support', 'near_resistance',
-    'gravestone_doji', 'four_price_doji', 'doji', 'spinning_top',
-    'bullish_candle', 'bearish_candle', 'dragonfly_candle',
-    'spinning_top_bearish_followup', 'bullish_then_dragonfly'
-}
-
-NORMALIZATION_EXCLUSIONS = {
-    # Binary flags
+    # Price Variants (3) — models consume these for absolute-scale context
+    'price', 'close_scaled', 'close_price',
+})
+
+# ---- Envelope keys — wire metadata, NEVER fed to a model -------------------
+# Disjoint from _FEATURES by invariant (checked below in __post_init__).
+_ENVELOPE: FrozenSet[str] = frozenset({
+    'agent',             # routing
+    'timeframe',         # routing
+    'timestamp',         # wall-clock ISO-8601 at publish
+    'tick_index',        # monotonic producer tick counter
+    'tick_count',        # legacy alias, kept for back-compat
+    'feature_count',     # integrity check: len(features)
+    'contract_version',  # schema version string
+    'features',          # nested payload key
+})
+
+# ---- Typed subsets of _FEATURES (validated as subsets at import time) ------
+_BINARY: FrozenSet[str] = frozenset({
     'near_support', 'near_resistance',
     'gravestone_doji', 'four_price_doji', 'doji', 'spinning_top',
     'bullish_candle', 'bearish_candle', 'dragonfly_candle',
     'spinning_top_bearish_followup', 'bullish_then_dragonfly',
-    
-    # Raw price (keep absolute scale)
+})
+_PRICE_SCALE: FrozenSet[str] = frozenset({
     'price', 'close_scaled', 'close_price',
-    
-    # Price-unit features (interpretability)
     'ma10', 'ma20', 'bollinger_upper', 'bollinger_lower',
-    
-    # Raw derivatives (normalize their stats only)
-    'price_vel', 'price_acc', 'price_jrk'
-}
+})
+_NON_NORMALISED: FrozenSet[str] = _BINARY | _PRICE_SCALE | frozenset({
+    'price_vel', 'price_acc', 'price_jrk',
+})
+
+
+@dataclass(frozen=True)
+class ValidationResult:
+    """Structured validation outcome with three distinct failure modes."""
+    ok: bool
+    missing: FrozenSet[str]           # required features absent from dict
+    leaked_envelope: FrozenSet[str]   # envelope keys found inside features dict
+    unexpected: FrozenSet[str]        # keys that belong to neither set
+
+    def as_error_lines(self):
+        lines = []
+        if self.missing:
+            lines.append(f"missing features: {sorted(self.missing)}")
+        if self.leaked_envelope:
+            lines.append(f"envelope keys inside features dict: "
+                         f"{sorted(self.leaked_envelope)}")
+        if self.unexpected:
+            lines.append(f"unknown keys: {sorted(self.unexpected)}")
+        return lines
+
+
+@dataclass(frozen=True)
+class FeatureContract:
+    """
+    The schema for a single timeframe's feature payload.
 
-PRICE_FEATURES = {
-    'price', 'close_scaled', 'close_price', 
-    'ma10', 'ma20', 'bollinger_upper', 'bollinger_lower'
-}
+    Invariants (all checked in __post_init__ — module fails to import if
+    any are violated):
+
+        (1) features ∩ envelope = ∅
+            No key is allowed to be "both a feature and envelope". This
+            was the original bug — 'price' was in both sets, and the
+            validator silently rejected every tick.
+
+        (2) |features| == EXPECTED_FEATURE_COUNT
+            The contract declares an exact 60-feature shape. Drift here
+            would corrupt downstream tensor shapes.
+
+        (3) binary, price_scale, non_normalised are all ⊆ features
+            A typed subset cannot contain a key that isn't a feature at
+            all. This catches stale references after a feature rename.
+    """
+    version:         str          = CONTRACT_VERSION
+    features:        FrozenSet[str] = field(default_factory=lambda: _FEATURES)
+    envelope:        FrozenSet[str] = field(default_factory=lambda: _ENVELOPE)
+    binary:          FrozenSet[str] = field(default_factory=lambda: _BINARY)
+    price_scale:     FrozenSet[str] = field(default_factory=lambda: _PRICE_SCALE)
+    non_normalised:  FrozenSet[str] = field(default_factory=lambda: _NON_NORMALISED)
+
+    def __post_init__(self):
+        # (1) disjointness
+        overlap = self.features & self.envelope
+        if overlap:
+            raise RuntimeError(
+                f"[FeatureContract] BROKEN INVARIANT: keys in BOTH features "
+                f"and envelope: {sorted(overlap)}. Remove from one set — the "
+                f"validator cannot distinguish feature-vs-envelope for these "
+                f"keys, so every tick will be rejected."
+            )
+        # (2) cardinality
+        if len(self.features) != EXPECTED_FEATURE_COUNT:
+            raise RuntimeError(
+                f"[FeatureContract] BROKEN INVARIANT: expected "
+                f"{EXPECTED_FEATURE_COUNT} features, got {len(self.features)}. "
+                f"Update EXPECTED_FEATURE_COUNT or fix the feature list."
+            )
+        # (3) subsets
+        for name, subset in (
+            ('binary', self.binary),
+            ('price_scale', self.price_scale),
+            ('non_normalised', self.non_normalised),
+        ):
+            stray = subset - self.features
+            if stray:
+                raise RuntimeError(
+                    f"[FeatureContract] BROKEN INVARIANT: '{name}' contains "
+                    f"non-feature keys: {sorted(stray)}"
+                )
+
+    # ---- public API -------------------------------------------------------
+
+    def validate(self, features_dict: Mapping[str, Any]) -> ValidationResult:
+        """
+        Validate the INNER features dict only — envelope keys should NOT
+        be present here; if they are, they're reported as leaked_envelope,
+        not stripped and hidden.
+        """
+        actual = set(features_dict.keys())
+        return ValidationResult(
+            ok              = (actual == self.features),
+            missing         = frozenset(self.features - actual),
+            leaked_envelope = frozenset(actual & self.envelope),
+            unexpected      = frozenset(actual - self.features - self.envelope),
+        )
+
+    def build_payload(
+        self,
+        agent_name:    str,
+        features_dict: Mapping[str, float],
+        tick_index,
+        timestamp_iso: str,
+    ) -> dict:
+        """
+        Construct the wire payload with envelope / feature separation
+        enforced structurally. Envelope fields live at the top level;
+        features live ONLY inside payload['features'].
+        """
+        return {
+            'agent':            agent_name,
+            'timestamp':        timestamp_iso,
+            'tick_index':       tick_index,
+            'feature_count':    len(features_dict),
+            'contract_version': self.version,
+            'features':         dict(features_dict),
+        }
+
+    def extract_features(self, payload: Mapping[str, Any]) -> dict:
+        """
+        Consumer-side: pull the inner features dict and verify envelope
+        version. Raises ValueError on schema drift so the consumer can
+        log-and-drop rather than silently accept malformed payloads.
+        """
+        got_ver = payload.get('contract_version')
+        if got_ver is not None and got_ver != self.version:
+            raise ValueError(
+                f"contract version mismatch: payload={got_ver!r} "
+                f"expected={self.version!r}"
+            )
+        feats = payload.get('features')
+        if not isinstance(feats, dict):
+            raise ValueError(
+                f"payload.features missing or wrong type: {type(feats).__name__}"
+            )
+        return feats
+
+
+# Singleton — import this, don't construct your own.
+# Module import will FAIL LOUDLY here if any invariant is violated.
+FEATURE_CONTRACT = FeatureContract()
+
+
+# ---------------------------------------------------------------------------
+# Back-compat aliases — projections of FEATURE_CONTRACT. Existing call sites
+# keep working unchanged; only the source of truth moved. Deleting any of
+# these will break older code paths that haven't been migrated to use
+# FEATURE_CONTRACT directly.
+# ---------------------------------------------------------------------------
+REQUIRED_FEATURES        = tuple(FEATURE_CONTRACT.features)    # order-agnostic
+METADATA_FIELDS          = FEATURE_CONTRACT.envelope
+BINARY_FEATURES          = FEATURE_CONTRACT.binary
+PRICE_FEATURES           = FEATURE_CONTRACT.price_scale
+NORMALIZATION_EXCLUSIONS = FEATURE_CONTRACT.non_normalised
 
 # ============================================================================
 # REGIME DETECTION PARAMETERS
@@ -1021,8 +1193,19 @@ class IntegratedFeatureEnhancer:
         self.features_lock = threading.Lock()
         
         logger.info(f"Regime-Adaptive Feature Enhancer initialized")
-        logger.info(f"Expected features: {len(REQUIRED_FEATURES)}")
-        assert len(REQUIRED_FEATURES) == 60, f"Feature count mismatch: {len(REQUIRED_FEATURES)}"
+        logger.info(
+            f"Contract: version={FEATURE_CONTRACT.version} "
+            f"features={len(FEATURE_CONTRACT.features)} "
+            f"envelope={len(FEATURE_CONTRACT.envelope)} "
+            f"(invariants enforced at import time)"
+        )
+        # Defensive re-check at instantiation. The contract's __post_init__
+        # already verified this at import, but a runtime assert catches
+        # anyone monkey-patching FEATURE_CONTRACT.features before first use.
+        assert len(FEATURE_CONTRACT.features) == EXPECTED_FEATURE_COUNT, (
+            f"Feature count mismatch at runtime: "
+            f"{len(FEATURE_CONTRACT.features)} != {EXPECTED_FEATURE_COUNT}"
+        )
 
     def compute_core_technical_features(self, df):
         """Compute 19 core technical indicators with robust edge case handling"""
@@ -1218,15 +1401,18 @@ class IntegratedFeatureEnhancer:
         return df
     
     def _validate_feature_contract(self, features_dict):
-        """Enforce strict 60-feature contract"""
-        expected = set(REQUIRED_FEATURES)
-        actual = set(features_dict.keys()) - METADATA_FIELDS
-        
-        missing = expected - actual
-        extra = actual - expected
-        
-        is_valid = (len(missing) == 0 and len(extra) == 0)
-        return is_valid, missing, extra
+        """
+        Delegate to FEATURE_CONTRACT.validate() and return a legacy
+        3-tuple (is_valid, missing, extra) for call-site back-compat.
+
+        `extra` in the legacy contract conflated two distinct failure
+        modes — envelope leakage and unknown keys. We preserve the
+        3-tuple shape but keep them merged; richer diagnostics are
+        available by calling FEATURE_CONTRACT.validate() directly.
+        """
+        result = FEATURE_CONTRACT.validate(features_dict)
+        extra = result.leaked_envelope | result.unexpected
+        return result.ok, result.missing, extra
     
     def compute_all_features(self, df):
         """
@@ -1381,17 +1567,28 @@ class IntegratedFeatureEnhancer:
                     logger.warning(f"[{agent_name}] Missing feature: {feature}")
                     latest_features[feature] = 0.0
             
-            # ENFORCE CONTRACT
-            is_valid, missing, extra = self._validate_feature_contract(latest_features)
-            
-            if not is_valid:
+            # ENFORCE CONTRACT — use the rich ValidationResult directly so we
+            # log three distinct failure modes separately instead of collapsing
+            # them into a single ambiguous "Missing / Extra" pair.
+            validation = FEATURE_CONTRACT.validate(latest_features)
+
+            if not validation.ok:
                 logger.error("=" * 80)
-                logger.error(f"❌ [{agent_name}] FEATURE CONTRACT VIOLATION")
-                if missing:
-                    logger.error(f"Missing: {missing}")
-                if extra:
-                    logger.error(f"Extra: {extra}")
+                logger.error(
+                    f"❌ [{agent_name}] FEATURE CONTRACT VIOLATION "
+                    f"(contract={FEATURE_CONTRACT.version})"
+                )
+                for line in validation.as_error_lines():
+                    logger.error(f"    {line}")
                 logger.error("=" * 80)
+
+                # Bookkeeping counter — lets ops tell the difference between
+                # "feed is dry" and "feed is arriving but contract is broken".
+                if not hasattr(self, '_contract_violation_counts'):
+                    self._contract_violation_counts = {}
+                self._contract_violation_counts[agent_name] = (
+                    self._contract_violation_counts.get(agent_name, 0) + 1
+                )
                 return
             
             with self.features_lock:
@@ -1401,25 +1598,52 @@ class IntegratedFeatureEnhancer:
             logger.error(f"[{agent_name}] Feature enhancement failed: {e}")
     
     async def publish_features(self, agent_name, features_dict, tick_index=None):
-        """Publish 60 features"""
+        """
+        Publish 60 features on the wire. Payload shape is enforced by
+        FEATURE_CONTRACT.build_payload() — envelope keys live at the
+        top level, feature keys live ONLY inside payload['features'],
+        and a contract_version string accompanies every message so the
+        consumer can detect schema drift.
+        """
         try:
+            # Defensive re-validation at the publish boundary. Zero cost on
+            # the happy path; catches any mutation between compute and
+            # publish (e.g. a caller accidentally injecting envelope keys
+            # into the features dict).
+            validation = FEATURE_CONTRACT.validate(features_dict)
+            if not validation.ok:
+                logger.error(
+                    f"[{agent_name}] publish BLOCKED — contract violation at "
+                    f"publish boundary: {validation.as_error_lines()}"
+                )
+                return
+
+            # Coerce numpy scalars to native floats so the JSON serialiser
+            # doesn't choke. Done on the features-only dict, inside the
+            # contract shape.
             clean_features = {
                 k: float(v) if isinstance(v, (np.floating, np.integer)) else v
                 for k, v in features_dict.items()
             }
 
-            # Resolve tick_index: caller may pass it, or it may be embedded in the dict
+            # Resolve tick_index: caller may pass it explicitly, or it may
+            # be embedded in the dict (legacy path). Envelope keys should
+            # NOT be inside features_dict after the validation above, so
+            # these .get() calls will normally return None — kept for
+            # defensive back-compat.
             resolved_tick = tick_index
             if resolved_tick is None:
-                resolved_tick = features_dict.get('tick_count') or features_dict.get('tick_index')
-
-            payload = {
-                'agent': agent_name,
-                'features': clean_features,
-                'timestamp': datetime.now(UTC).isoformat(),
-                'feature_count': len(clean_features),
-                'tick_index': resolved_tick
-            }
+                resolved_tick = (
+                    features_dict.get('tick_count')
+                    or features_dict.get('tick_index')
+                )
+
+            payload = FEATURE_CONTRACT.build_payload(
+                agent_name    = agent_name,
+                features_dict = clean_features,
+                tick_index    = resolved_tick,
+                timestamp_iso = datetime.now(UTC).isoformat(),
+            )
 
             await self.features_channel.publish("integrated-features", payload)
 
@@ -1499,7 +1723,11 @@ class IntegratedFeatureEnhancer:
             sample_agent = list(self.latest_computed_features.keys())[0]
             features = self.latest_computed_features[sample_agent]
             
-            actual_count = len([k for k in features.keys() if k not in METADATA_FIELDS])
+            # Count only keys that are actually declared features in the
+            # contract. This is set-intersection, not set-difference — so
+            # it's correct regardless of whether envelope keys have leaked
+            # into the features dict or not.
+            actual_count = len(set(features.keys()) & FEATURE_CONTRACT.features)
             
             summary = f"REGIME-ADAPTIVE FEATURE ENHANCER\n"
             summary += "=" * 60 + "\n\n"