#!/usr/bin/env python3 """ ╔══════════════════════════════════════════════════════════════════════════════╗ ║ MAYTHOS — Multi-Architecture Yield-Tuned Hybrid Oscillation System ║ ║ Version : 1.1 ║ ║ Build : 1.1.0 ║ ║ Spec : 1.1 ║ ║ Single-file — all engines embedded, no runtime deps except numpy ║ ╚══════════════════════════════════════════════════════════════════════════════╝ Architecture (10 layers): Layer 1 — Data Integrity Engine Layer 2 — Adaptive Market State + Asset Profile + Session Intelligence Layer 3 — Multi-Timeframe Fusion Engine Layer 4 — Real-Time Event Detection Engine Layer 5 — Liquidity + Pressure Engine Layer 6 — Technical Confirmation Stack (EMA200/RSI/MACD/BB/S&R) Layer 7 — Adaptive Scoring Engine Layer 8 — Warm-up Controller Layer 9 — Signal Lifecycle Engine Layer 10 — Output Formatter + Validator Usage: engine = MAYTHOS() output = engine.tick(candle) """ # ============================================================================== # PART A CONSTANTS UTILS BUFFERS # ============================================================================== from __future__ import annotations import math import time from collections import deque from dataclasses import dataclass, field from typing import Any, Deque, Dict, List, Optional, Tuple import numpy as np # ============================================================================== # VERSIONING # ============================================================================== SPEC_VERSION = "1.1" ENGINE_VERSION = "1.1.0" OUTPUT_VERSION = "1.1.0" # ============================================================================== # NUMERICAL CONSTANTS # ============================================================================== EPSILON = 1e-10 # guard against div-by-zero CLAMP_MIN = 0.0 CLAMP_MAX = 1.0 SCORE_CLAMP = (0.0, 1.0) # ============================================================================== # MARKET STATE LABELS (predefined, append-only) # ============================================================================== MARKET_STATES = frozenset({ "trend", "range", "compression", "expansion", "manipulation", "sweep", "absorption", "reversal", "continuation", "unstable", "noisy", "undefined", }) REGIME_LABELS = frozenset({ "bullish_trend", "bearish_trend", "bullish_range", "bearish_range", "breakout_attempt", "breakout_confirmed", "exhaustion", "distribution", "accumulation", "squeeze", "trap", "recovery", "transition", "neutral_dirty", }) MARKET_RECOGNITION_LABELS = frozenset({ "clean_trend", "dirty_trend", "clean_range", "dirty_range", "compression_before_expansion", "exhaustion_after_impulse", "manipulation_sweep", "absorption_zone", "reversal_attempt", "continuation_attempt", "unstable_noise_cluster", }) ASSET_MODES = frozenset({"crypto", "forex", "OTC", "hybrid", "unknown"}) EXECUTION_SUITABILITY = frozenset({"blocked", "weak", "moderate", "strong"}) SIGNAL_DIRECTIONS = frozenset({"BUY", "SELL"}) OPERATIONAL_MODES = frozenset({ "cold_start", "normal", "cautious", "degraded", "blocked", "recovery", "hot_response", }) SIGNAL_LIFECYCLE_STATES = frozenset({ "idle", "forming", "candidate", "confirmed", "cooling", "expired", "invalidated", }) REASON_CODES = frozenset({ "DATA_OK", "DATA_DEGRADED", "TF_ALIGNED", "TF_CONFLICT", "LIQUIDITY_SWEEP", "MANIP_HIGH", "MOMENTUM_OK", "TIMING_BAD", "STALE_DECAY", "SESSION_SUPPORT", "SESSION_RISK", }) EXPIRY_BUCKETS = frozenset({"ultra_short", "short", "medium"}) EVENT_SEVERITIES = frozenset({"minor", "moderate", "strong", "extreme"}) MARKET_TEMPERATURE_STATES = frozenset({"cold", "warm", "overheated", "unstable_explosive"}) SESSION_LABELS = frozenset({ "london", "ny", "asia", "ny_london_overlap", "asia_london_overlap", "rollover", "off_hours", }) TIMEFRAME_LABELS = frozenset({ "5s", "30s", "1m", "2m", "5m", "10m", "15m", "higher", }) # ============================================================================== # CONFIGURATION DEFAULTS # ============================================================================== DEFAULT_BUFFER_SIZE = 300 # candles (covers ~5h on 1m) MIN_WARM_CANDLES = 30 # cold-start gate FULL_WARM_CANDLES = 50 # full confidence unlock (FIX-3: halved from 100 → 50 to cut cold-start delay) MAX_CACHE_ENTRIES = 64 # bounded memory-behavior caches AGE_DECAY_TICKS = 50 # TIMING-FIX-5: ticks before cache entry ages out (used by BoundedCache if instantiated) SIGNAL_COOLDOWN_TICKS = 5 # TIMING-FIX-5: was 3 → 5 ticks minimum between same-direction signals # RSI periods and defaults RSI_PERIOD = 14 MACD_FAST = 12 MACD_SLOW = 26 MACD_SIGNAL = 9 BB_PERIOD = 20 BB_STD_DEV = 2.0 EMA200_PERIOD = 200 RSI_OB_CRYPTO = 80.0 RSI_OS_CRYPTO = 20.0 RSI_OB_DEFAULT = 70.0 RSI_OS_DEFAULT = 30.0 # Confidence bands (Section 17) CONF_BLOCKED_MAX = 0.24 CONF_WEAK_MAX = 0.44 CONF_MODERATE_MAX = 0.69 # above 0.69 → strong # Sudden candle expansion multiplier SUDDEN_EXPANSION_MULT = 3.0 # ============================================================================== # NUMERIC UTILITY FUNCTIONS # ============================================================================== def clamp(value: float, lo: float = 0.0, hi: float = 1.0) -> float: """Clamp value to [lo, hi]. Handles NaN → lo.""" if not math.isfinite(value): return lo return max(lo, min(hi, value)) def safe_div(numerator: float, denominator: float, fallback: float = 0.0) -> float: """Division protected against zero/NaN/inf. Returns fallback on error.""" if not math.isfinite(numerator) or not math.isfinite(denominator): return fallback if abs(denominator) < EPSILON: return fallback result = numerator / denominator return result if math.isfinite(result) else fallback def safe_sqrt(value: float, fallback: float = 0.0) -> float: """Square root protected against negative and non-finite inputs.""" if not math.isfinite(value) or value < 0.0: return fallback return math.sqrt(value) def is_finite_positive(value: float) -> bool: return math.isfinite(value) and value > 0.0 def normalize_01(value: float, min_val: float, max_val: float, fallback: float = 0.5) -> float: """Normalize value to [0, 1] given range. Clamped. Returns fallback if range is zero.""" rng = max_val - min_val if abs(rng) < EPSILON: return fallback return clamp(safe_div(value - min_val, rng)) def weighted_mean(values: List[float], weights: List[float]) -> float: """Weighted average. Returns 0.0 if all weights are zero. FIX-5: Was returning 0.5 (neutral), which leaked a false mid-range confidence into blocked/cold outputs. 0.0 is correct: no weight → no evidence → no score.""" total_w = sum(weights) if total_w < EPSILON: return 0.0 return clamp(sum(v * w for v, w in zip(values, weights)) / total_w) def nan_safe(value: float, fallback: float = 0.0) -> float: """Replace NaN/inf with fallback.""" return value if math.isfinite(value) else fallback def _validate_candle_fields(o: float, h: float, l: float, c: float) -> bool: """Return True if OHLC relationships are self-consistent.""" if not all(math.isfinite(x) for x in (o, h, l, c)): return False if h < l - EPSILON: return False if c > h + EPSILON or c < l - EPSILON: return False if o > h + EPSILON or o < l - EPSILON: return False return True # ============================================================================== # ENGINE DIAGNOSTICS (AUDIT FIX — observability for the silent-exception path) # ============================================================================== # # Every engine below is deliberately fault-tolerant: its public entry point # wraps all real logic in try/except and falls back to a safe default on any # exception, so a single bad tick (or a genuine bug) can never crash the # pipeline or take signal generation down. That is the correct behavior for # 24/7 uptime -- but as originally written, those except blocks were # completely silent: nothing was ever counted, logged, or exposed anywhere. # A real, persistent logic bug in any engine would degrade that engine's # output to its safe-default forever with zero way to detect it was # happening. This lightweight, bounded recorder closes that gap without # changing any existing fallback behavior — it is purely additive and never # itself raises. Process-wide (shared across however many MAYTHOS instances # run in one process, e.g. one per symbol/feed), since "is something actually # broken" is most useful as a process-level health signal. Inspect via # MAYTHOS.engine_health() or the module-level ENGINE_DIAGNOSTICS singleton — # e.g. expose it on a /health endpoint in your HF Space. class _EngineDiagnostics: _MAX_RECENT = 200 def __init__(self) -> None: self.counts: Dict[str, int] = {} self.recent: "deque" = deque(maxlen=self._MAX_RECENT) def record(self, engine: str, exc: BaseException) -> None: try: self.counts[engine] = self.counts.get(engine, 0) + 1 self.recent.append({ "engine" : engine, "type" : type(exc).__name__, "message": str(exc)[:200], "time" : time.time(), }) except Exception: pass # diagnostics must never themselves raise (no recursive self-record) def snapshot(self) -> Dict: return { "error_counts" : dict(self.counts), "total_errors" : sum(self.counts.values()), "recent_errors": list(self.recent), } def reset(self) -> None: self.counts.clear() self.recent.clear() ENGINE_DIAGNOSTICS = _EngineDiagnostics() # ============================================================================== # ROLLING FIXED-SIZE BUFFERS # ============================================================================== class RollingBuffer: """ Fixed-size FIFO ring buffer backed by a pre-allocated numpy array. FIX-1 (memory): Eliminates the per-tick deque→np.array() conversion that created a new Python object on every mean/std/min/max call. The backing array is allocated once at construction and reused forever. FIX-2 (CPU): Uses pure-Python arithmetic for arrays with fewer than 10 elements, where numpy dispatch overhead exceeds computation cost on 2-core HuggingFace CPUs. Interface is fully backward-compatible with the old deque version. """ __slots__ = ("maxlen", "_data", "_head", "_count") def __init__(self, maxlen: int) -> None: assert maxlen > 0, "maxlen must be positive" self.maxlen: int = maxlen self._data: np.ndarray = np.zeros(maxlen, dtype=np.float64) self._head: int = 0 # next write slot (= oldest slot when buffer full) self._count: int = 0 # number of valid entries ≤ maxlen def push(self, value: float) -> None: self._data[self._head] = nan_safe(value) self._head = (self._head + 1) % self.maxlen if self._count < self.maxlen: self._count += 1 # ------------------------------------------------------------------ # _view(): zero-copy slice of active (unordered) data. # Use ONLY for commutative operations (mean/std/min/max). # ------------------------------------------------------------------ def _view(self) -> np.ndarray: if self._count < self.maxlen: return self._data[:self._count] return self._data # full ring: all slots valid, order irrelevant for stats def as_array(self) -> np.ndarray: """Ordered (oldest→newest) copy. Used for indexed slicing / diffs.""" n = self._count if n == 0: return np.empty(0, dtype=np.float64) if n < self.maxlen: return self._data[:n].copy() if self._head == 0: return self._data.copy() # Unroll ring so index 0 is the oldest written entry return np.concatenate([self._data[self._head:], self._data[:self._head]]) def last(self, fallback: float = 0.0) -> float: if self._count == 0: return fallback return float(self._data[(self._head - 1) % self.maxlen]) def __len__(self) -> int: return self._count @property def full(self) -> bool: return self._count == self.maxlen def is_ready(self, min_len: int) -> bool: return self._count >= min_len # ------------------------------------------------------------------ # Stats — pure Python for n < 10, numpy for larger arrays. # ------------------------------------------------------------------ def mean(self) -> float: n = self._count if n == 0: return 0.0 v = self._view() if n < 10: s = 0.0 for x in v: s += float(x) return s / n return float(v.mean()) def std(self) -> float: n = self._count if n < 2: return 0.0 v = self._view() if n < 10: s = 0.0 for x in v: s += float(x) m = s / n ss = 0.0 for x in v: d = float(x) - m ss += d * d return math.sqrt(ss / n) return float(v.std()) def min(self) -> float: n = self._count if n == 0: return 0.0 v = self._view() if n < 10: mn = math.inf for x in v: fx = float(x) if fx < mn: mn = fx return mn return float(v.min()) def max(self) -> float: n = self._count if n == 0: return 0.0 v = self._view() if n < 10: mx = -math.inf for x in v: fx = float(x) if fx > mx: mx = fx return mx return float(v.max()) class RollingObjectBuffer: """Fixed-size FIFO ring buffer for arbitrary Python objects (candles, events).""" __slots__ = ("_buf", "maxlen") def __init__(self, maxlen: int) -> None: assert maxlen > 0 self.maxlen = maxlen self._buf: Deque[Any] = deque(maxlen=maxlen) def push(self, obj: Any) -> None: self._buf.append(obj) def last(self, fallback: Any = None) -> Any: return self._buf[-1] if self._buf else fallback def to_list(self) -> List[Any]: return list(self._buf) def __len__(self) -> int: return len(self._buf) def is_ready(self, min_len: int) -> bool: return len(self._buf) >= min_len class BoundedCache: """ Bounded key-value cache with age-based eviction. Maximum MAX_CACHE_ENTRIES entries. """ def __init__(self, maxlen: int = MAX_CACHE_ENTRIES) -> None: self.maxlen = maxlen self._data: Dict[str, Tuple[Any, int]] = {} # key → (value, insert_tick) self._tick: int = 0 def tick(self) -> None: self._tick += 1 def set(self, key: str, value: Any) -> None: if len(self._data) >= self.maxlen and key not in self._data: self._evict_oldest() self._data[key] = (value, self._tick) def get(self, key: str, fallback: Any = None) -> Any: entry = self._data.get(key) return entry[0] if entry is not None else fallback def evict_stale(self, max_age: int = AGE_DECAY_TICKS) -> None: stale_keys = [k for k, (_, t) in self._data.items() if self._tick - t > max_age] for k in stale_keys: del self._data[k] def _evict_oldest(self) -> None: if not self._data: return oldest_key = min(self._data, key=lambda k: self._data[k][1]) del self._data[oldest_key] def __contains__(self, key: str) -> bool: return key in self._data def __len__(self) -> int: return len(self._data) # ============================================================================== # CANDLE DATACLASS — minimal validated representation # ============================================================================== @dataclass class Candle: """ Validated OHLCV candle. Optional fields default to safe sentinels. Immutable once constructed. """ timestamp : float open : float high : float low : float close : float volume : float = 0.0 # 0 = absent, triggers proxy logic spread : float = 0.0 bid : float = 0.0 ask : float = 0.0 source_id : str = "default" session_label: str = "unknown" is_closed : bool = True # False = intrabar/live tick def __post_init__(self) -> None: # Clamp float fields against non-finite for attr in ("open", "high", "low", "close", "volume", "spread", "bid", "ask"): v = getattr(self, attr) if not math.isfinite(v): object.__setattr__(self, attr, 0.0) # Ensure OHLC consistency after sanitization; if broken, mark invalid object.__setattr__(self, "_valid", _validate_candle_fields( self.open, self.high, self.low, self.close)) @property def valid(self) -> bool: return self._valid # type: ignore[attr-defined] @property def body_size(self) -> float: return abs(self.close - self.open) @property def candle_range(self) -> float: return self.high - self.low @property def upper_wick(self) -> float: return self.high - max(self.open, self.close) @property def lower_wick(self) -> float: return min(self.open, self.close) - self.low @property def is_bullish(self) -> bool: return self.close >= self.open @property def body_ratio(self) -> float: r = self.candle_range return safe_div(self.body_size, r) if r > EPSILON else 0.5 @property def close_position(self) -> float: """Where the close sits within the full range [0=bottom, 1=top].""" r = self.candle_range return safe_div(self.close - self.low, r) if r > EPSILON else 0.5 # ============================================================================== # SAFE OUTPUT DEFAULTS (used when subsystems are unavailable/cold-start) # ============================================================================== def _default_output() -> Dict[str, Any]: """Return a fully-populated safe output dict in cold/blocked state.""" return { # Core "direction" : "BUY", "confidence" : 0.0, "internal_trust" : 0.0, "execution_suitability" : "blocked", "market_state" : "undefined", "regime_label" : "transition", "asset_mode" : "unknown", "timeframe_alignment" : 0.0, "manipulation_probability": 0.0, "liquidity_score" : 0.5, "volatility_score" : 0.5, "pressure_score" : 0.5, "momentum_score" : 0.5, "timing_score" : 0.0, "signal_freshness" : 0.0, "spread_health" : 1.0, "data_quality" : 1.0, "readability_score" : 0.5, "stale_signal_flag" : False, "degraded_mode_flag" : False, "blocked_flag" : True, "reason_summary" : "cold_start: insufficient history", "reason_codes" : ["DATA_OK"], # Optional "call_bias" : 0.5, "put_bias" : 0.5, "expiry_suitability_ultra_short" : 0.0, "expiry_suitability_short" : 0.0, "expiry_suitability_medium" : 0.0, "debug_trace" : None, # Warm-up state "warm_up_fraction" : 0.0, "operational_mode" : "cold_start", # Version "spec_version" : SPEC_VERSION, "engine_version" : ENGINE_VERSION, "output_version" : OUTPUT_VERSION, } # ============================================================================== # PART B DATA INTEGRITY # ============================================================================== import math import time from collections import deque from typing import Dict, Optional, Tuple # ============================================================================== # DATA INTEGRITY ENGINE # ============================================================================== class DataIntegrityEngine: """ Input contract: receives a Candle and processes it. Output contract: DataIntegrityResult dict. Responsibilities (Section 9): - Tick validation and deduplication. - Timestamp ordering and gap detection. - Spread / latency monitoring. - Tick structure analysis. - Noise control. - Source confidence scoring. """ # Configurable thresholds (adaptive, not static) _SPREAD_WINDOW = 30 _LATENCY_WINDOW = 30 _NOISE_WINDOW = 20 _GAP_SIGMA_THRESHOLD = 4.0 # z-score to flag timestamp gap _SPREAD_SIGMA_THRESHOLD = 3.5 _WICK_SIGMA_THRESHOLD = 3.0 _SYNTHETIC_TICK_BODY_RATIO_MIN = 0.97 # FIX 2: was 0.90 — 90% body ratio fires on real trend candles (7.6% false positive rate on live BTC). 0.97 limits to truly pathological cases. _CLUSTER_TICK_WINDOW = 5 # candles for cluster detection _DUPLICATE_TOLERANCE = 0.5 # seconds _MAX_SOURCES = 64 # AUDIT FIX: cap on distinct source_id entries tracked def __init__(self) -> None: # --- timestamp tracking --- self._last_ts : float = -1.0 self._ts_gaps : RollingBuffer = RollingBuffer(self._LATENCY_WINDOW) self._ts_intervals : RollingBuffer = RollingBuffer(50) # for gap z-score # --- spread tracking --- self._spread_buf : RollingBuffer = RollingBuffer(self._SPREAD_WINDOW) self._spread_events : int = 0 # --- latency / websocket --- self._receive_ts_buf : RollingBuffer = RollingBuffer(self._LATENCY_WINDOW) self._latency_buf : RollingBuffer = RollingBuffer(self._LATENCY_WINDOW) # --- range / wick --- self._range_buf : RollingBuffer = RollingBuffer(self._NOISE_WINDOW) self._upper_wick_buf : RollingBuffer = RollingBuffer(self._NOISE_WINDOW) self._lower_wick_buf : RollingBuffer = RollingBuffer(self._NOISE_WINDOW) self._body_buf : RollingBuffer = RollingBuffer(self._NOISE_WINDOW) # --- duplicate / ordering detection --- self._seen_ts_window : deque = deque(maxlen=10) # --- source confidence --- # AUDIT FIX (memory-leak risk): _source_scores is keyed by whatever # source_id the caller sets on each Candle. In normal use this is a # small fixed set (e.g. one per exchange/feed), but nothing previously # bounded its cardinality -- if source_id ever carried high-cardinality # values (a bug upstream, a per-tick id, a rotating feed id, etc.) on # a 24/7 deployment, this dict would grow without limit for the life # of the process. _MAX_SOURCES + _source_last_seen below cap it with # simple least-recently-seen eviction, mirroring the pattern already # used by BoundedCache elsewhere in this file. self._source_scores : Dict[str, RollingBuffer] = {} self._source_last_seen: Dict[str, int] = {} # --- noise counting --- self._noise_count : int = 0 self._tick_count : int = 0 self._invalid_count : int = 0 self._gap_count : int = 0 # --- instability level --- self._instability : float = 0.0 # 0=stable, 1=very unstable self._source_confidence: float = 1.0 # ------------------------------------------------------------------ # PUBLIC: process a new candle # ------------------------------------------------------------------ def process(self, candle: Candle, receive_time: Optional[float] = None) -> Dict: """ Main entry. Call on every new candle/tick. Returns DataIntegrityResult dict. Always returns a result; never raises. """ try: return self._process_inner(candle, receive_time) except Exception as exc: ENGINE_DIAGNOSTICS.record("DataIntegrityEngine", exc) # Total fail-safe: return a fully safe degraded result return self._safe_fallback_result() # ------------------------------------------------------------------ # INTERNAL PROCESSING # ------------------------------------------------------------------ def _process_inner(self, candle: Candle, receive_time: Optional[float]) -> Dict: self._tick_count += 1 result: Dict = { "valid" : True, "duplicate" : False, "out_of_order" : False, "gap_detected" : False, "gap_severity" : 0.0, "spread_anomaly" : False, "spread_expansion" : False, "wick_anomaly" : False, "synthetic_tick" : False, "noise_burst" : False, "candle_distortion" : False, "latency_spike" : False, "latency_drift" : False, "source_confidence" : 1.0, "data_quality" : 1.0, "instability_level" : 0.0, "suppress_signal" : False, "has_volume" : candle.volume > EPSILON, "has_bid_ask" : candle.bid > EPSILON and candle.ask > EPSILON, } # 1. Basic candle validity if not candle.valid: result["valid"] = False result["suppress_signal"] = True self._invalid_count += 1 result["data_quality"] = clamp(1.0 - safe_div(self._invalid_count, max(1, self._tick_count))) return self._finalize(result, candle) # 2. Duplicate detection ts = candle.timestamp if self._is_duplicate(ts): result["duplicate"] = True result["suppress_signal"] = True return self._finalize(result, candle) # 3. Timestamp ordering and gap analysis if self._last_ts > 0.0: interval = ts - self._last_ts if interval < -self._DUPLICATE_TOLERANCE: result["out_of_order"] = True result["suppress_signal"] = True return self._finalize(result, candle) if interval > EPSILON: self._ts_intervals.push(interval) # Gap detection using z-score of intervals gap_severity = self._compute_gap_severity(interval) if gap_severity > 0.5: result["gap_detected"] = True result["gap_severity"] = gap_severity self._gap_count += 1 self._last_ts = ts self._seen_ts_window.append(ts) # 4. Spread analysis spread = candle.spread if spread > EPSILON: self._spread_buf.push(spread) s_anomaly, s_expansion = self._analyze_spread(spread) result["spread_anomaly"] = s_anomaly result["spread_expansion"] = s_expansion if s_anomaly: self._spread_events += 1 # 5. Range / wick / body updates cr = candle.candle_range uw = candle.upper_wick lw = candle.lower_wick body = candle.body_size if cr > EPSILON: self._range_buf.push(cr) if uw >= 0: self._upper_wick_buf.push(uw) if lw >= 0: self._lower_wick_buf.push(lw) if body >= 0: self._body_buf.push(body) # 6. Wick anomaly result["wick_anomaly"] = self._detect_wick_anomaly(candle) # 7. Synthetic tick detection (near-perfect body, near-zero wicks) result["synthetic_tick"] = self._detect_synthetic(candle) # 8. Candle distortion (extreme ratio) result["candle_distortion"] = self._detect_distortion(candle) # 9. Noise burst detection result["noise_burst"] = self._detect_noise_burst(candle) if result["noise_burst"]: self._noise_count += 1 # 10. Latency analysis (if receive_time provided) if receive_time is not None and math.isfinite(receive_time): lat = receive_time - ts if lat >= 0: self._latency_buf.push(lat) result["latency_spike"] = self._detect_latency_spike(lat) result["latency_drift"] = self._detect_latency_drift() # 11. Source confidence scoring sc = self._update_source_confidence(candle.source_id, result) result["source_confidence"] = sc self._source_confidence = sc return self._finalize(result, candle) # ------------------------------------------------------------------ # FINALIZE: compute aggregate data_quality and instability # ------------------------------------------------------------------ def _finalize(self, result: Dict, candle: Candle) -> Dict: penalties = 0.0 if not result["valid"]: penalties += 0.5 if result["duplicate"] or result["out_of_order"]: penalties += 0.3 if result["gap_detected"]: penalties += result["gap_severity"] * 0.2 if result["spread_anomaly"]: penalties += 0.1 if result["spread_expansion"]: penalties += 0.15 if result["wick_anomaly"]: penalties += 0.1 if result["synthetic_tick"]: penalties += 0.2 if result["noise_burst"]: penalties += 0.1 if result["candle_distortion"]: penalties += 0.15 if result["latency_spike"]: penalties += 0.1 if result["latency_drift"]: penalties += 0.1 dq = clamp(1.0 - penalties) # Blend with source confidence dq = clamp(0.7 * dq + 0.3 * result["source_confidence"]) # Instability: exponential moving average of penalty alpha = 0.15 self._instability = clamp(alpha * penalties + (1 - alpha) * self._instability) result["data_quality"] = dq result["instability_level"] = self._instability result["suppress_signal"] = result.get("suppress_signal", False) or dq < 0.25 return result # ------------------------------------------------------------------ # COMPONENT DETECTORS # ------------------------------------------------------------------ def _is_duplicate(self, ts: float) -> bool: if not self._seen_ts_window: return False return any(abs(ts - prev) < self._DUPLICATE_TOLERANCE for prev in self._seen_ts_window) def _compute_gap_severity(self, interval: float) -> float: """Z-score of the interval relative to rolling mean/std of intervals.""" if not self._ts_intervals.is_ready(5): return 0.0 mean = self._ts_intervals.mean() std = self._ts_intervals.std() if std < EPSILON: return 0.0 z = abs(safe_div(interval - mean, std)) # Normalize to [0,1]: z>=GAP_SIGMA_THRESHOLD → 1.0 return clamp(safe_div(z, self._GAP_SIGMA_THRESHOLD)) def _analyze_spread(self, spread: float) -> Tuple[bool, bool]: """Returns (anomaly, expansion).""" if not self._spread_buf.is_ready(5): return False, False mean = self._spread_buf.mean() std = self._spread_buf.std() if std < EPSILON or mean < EPSILON: return False, False z = safe_div(spread - mean, std) anomaly = z > self._SPREAD_SIGMA_THRESHOLD expansion = z > self._SPREAD_SIGMA_THRESHOLD * 0.6 return anomaly, expansion def _detect_wick_anomaly(self, candle: Candle) -> bool: if not self._range_buf.is_ready(5): return False mean_range = self._range_buf.mean() std_range = self._range_buf.std() if mean_range < EPSILON: return False max_wick = max(candle.upper_wick, candle.lower_wick) if std_range < EPSILON: return max_wick > mean_range * (self._WICK_SIGMA_THRESHOLD + 1) z = safe_div(max_wick - mean_range, std_range + EPSILON) return z > self._WICK_SIGMA_THRESHOLD def _detect_synthetic(self, candle: Candle) -> bool: """Suspiciously perfect body → synthetic candle flag.""" cr = candle.candle_range if cr < EPSILON: return False body_r = safe_div(candle.body_size, cr) return body_r >= self._SYNTHETIC_TICK_BODY_RATIO_MIN def _detect_distortion(self, candle: Candle) -> bool: """Candle that is massively larger than recent average.""" if not self._range_buf.is_ready(5): return False mean_r = self._range_buf.mean() if mean_r < EPSILON: return False ratio = safe_div(candle.candle_range, mean_r) return ratio > 5.0 def _detect_noise_burst(self, candle: Candle) -> bool: """Tiny random candle in a stable environment.""" if not self._range_buf.is_ready(5): return False mean_r = self._range_buf.mean() if mean_r < EPSILON: return False ratio = safe_div(candle.candle_range, mean_r) # Very tiny candle compared to mean → possible noise tick return ratio < 0.05 def _detect_latency_spike(self, latency: float) -> bool: if not self._latency_buf.is_ready(5): return False mean_lat = self._latency_buf.mean() std_lat = self._latency_buf.std() if std_lat < EPSILON: return latency > mean_lat * 3.0 z = safe_div(latency - mean_lat, std_lat) return z > 3.0 def _detect_latency_drift(self) -> bool: """Detect steady creep in latency over recent window.""" if not self._latency_buf.is_ready(10): return False arr = self._latency_buf.as_array() # Simple slope: compare first half mean vs second half mean half = len(arr) // 2 first_half = float(arr[:half].mean()) second_half = float(arr[half:].mean()) if first_half < EPSILON: return False return safe_div(second_half - first_half, first_half) > 0.5 def _update_source_confidence(self, source_id: str, result: Dict) -> float: """Rolling exponential confidence score per source.""" if source_id not in self._source_scores: # AUDIT FIX (memory-leak risk): evict the least-recently-seen # source before adding a new one once at capacity, so this dict # can never grow past _MAX_SOURCES entries regardless of how many # distinct source_id values are ever observed over the engine's # 24/7 lifetime. if len(self._source_scores) >= self._MAX_SOURCES: oldest = min(self._source_last_seen, key=self._source_last_seen.get) del self._source_scores[oldest] del self._source_last_seen[oldest] self._source_scores[source_id] = RollingBuffer(30) self._source_last_seen[source_id] = self._tick_count buf = self._source_scores[source_id] # Quality this tick: start at 1.0, deduct for bad flags quality = 1.0 if not result.get("valid", True): quality -= 0.5 if result.get("spread_anomaly"): quality -= 0.1 if result.get("wick_anomaly"): quality -= 0.1 if result.get("noise_burst"): quality -= 0.1 if result.get("gap_detected"): quality -= result.get("gap_severity", 0.0) * 0.2 quality = clamp(quality) buf.push(quality) return buf.mean() if len(buf) > 0 else 1.0 # ------------------------------------------------------------------ # FALLBACK # ------------------------------------------------------------------ @staticmethod def _safe_fallback_result() -> Dict: return { "valid": False, "duplicate": False, "out_of_order": False, "gap_detected": False, "gap_severity": 0.0, "spread_anomaly": False, "spread_expansion": False, "wick_anomaly": False, "synthetic_tick": False, "noise_burst": False, "candle_distortion": False, "latency_spike": False, "latency_drift": False, "source_confidence": 0.5, "data_quality": 0.3, "instability_level": 0.5, "suppress_signal": True, "has_volume": False, "has_bid_ask": False, } # ------------------------------------------------------------------ # READ-ONLY PROPERTIES (for downstream engines) # ------------------------------------------------------------------ @property def instability(self) -> float: return self._instability @property def source_confidence(self) -> float: return self._source_confidence @property def tick_count(self) -> int: return self._tick_count # ============================================================================== # PART C MARKET STATE ASSET SESSION # ============================================================================== import math import time from typing import Dict, List, Optional, Tuple # ============================================================================== # PART C1 — ADAPTIVE MARKET STATE ENGINE (Layer 2) # ============================================================================== class MarketStateEngine: """ Classifies market state and regime adaptively from rolling candle data. Input contract: update(candle, di_result) → dict Output contract: { "market_state": str (from MARKET_STATES), "regime_label": str (from REGIME_LABELS), "market_recognition": str (from MARKET_RECOGNITION_LABELS), "trend_direction": float, # -1=bear, 0=none, 1=bull "exhaustion_prob": float, "continuation_prob": float, "regime_shift_prob": float, "fake_trend_flag": bool, "readability_hint": float, # [0,1] partial local readability } No engine modifies this engine's state. No circular deps. """ _VOL_WINDOW = 20 _HH_HL_WINDOW = 10 # candles for higher-high / lower-low detection _MOMENTUM_WINDOW = 5 _COMPRESSION_THRESHOLD = 0.4 # BB bandwidth z-score to flag compression def __init__(self) -> None: self._closes : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._highs : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._lows : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._ranges : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._bodies : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._directions: RollingBuffer = RollingBuffer(50) # +1 bull / -1 bear self._tick_count: int = 0 def update(self, candle: Candle, di_result: Dict) -> Dict: try: return self._update_inner(candle, di_result) except Exception as exc: ENGINE_DIAGNOSTICS.record("MarketStateEngine", exc) return self._safe_default() def _update_inner(self, candle: Candle, di_result: Dict) -> Dict: if not candle.valid: return self._safe_default() self._tick_count += 1 self._closes.push(candle.close) self._highs.push(candle.high) self._lows.push(candle.low) self._ranges.push(candle.candle_range) self._bodies.push(candle.body_size) self._directions.push(1.0 if candle.is_bullish else -1.0) if not self._closes.is_ready(MIN_WARM_CANDLES): return self._cold_default() vol_score, mean_range, std_range = self._compute_volatility() momentum = self._compute_momentum() hh_hl = self._detect_hh_hl() lh_ll = self._detect_lh_ll() compression = self._detect_compression(mean_range, std_range) fake_trend = self._detect_fake_trend(momentum, mean_range, std_range) market_state, regime, recognition = self._classify( candle, momentum, hh_hl, lh_ll, compression, fake_trend, vol_score, di_result) trend_dir = self._resolve_trend_dir(hh_hl, lh_ll, momentum) exhaustion = self._compute_exhaustion(momentum, vol_score) continuation = self._compute_continuation(momentum, hh_hl, lh_ll, compression) regime_shift = self._compute_regime_shift(market_state, momentum, vol_score) return { "market_state" : market_state, "regime_label" : regime, "market_recognition": recognition, "trend_direction" : trend_dir, "exhaustion_prob" : exhaustion, "continuation_prob" : continuation, "regime_shift_prob" : regime_shift, "fake_trend_flag" : fake_trend, "readability_hint" : self._local_readability(compression, vol_score, di_result), } # --- helpers --- def _compute_volatility(self) -> Tuple[float, float, float]: """Returns (vol_score, mean_range, std_range).""" arr = self._ranges.as_array() mean_r = float(arr.mean()) if len(arr) > 0 else 0.0 std_r = float(arr.std()) if len(arr) > 1 else 0.0 # Normalize volatility using z-score of current vs history vol_score = clamp(safe_div(mean_r, mean_r + std_r + EPSILON)) return vol_score, mean_r, std_r def _compute_momentum(self) -> float: """EMA-like directional momentum in [-1, 1].""" if not self._directions.is_ready(self._MOMENTUM_WINDOW): return 0.0 arr = self._directions.as_array()[-self._MOMENTUM_WINDOW:] # Weight recent candles more weights = [1.5 ** i for i in range(len(arr))] total_w = sum(weights) momentum = sum(d * w for d, w in zip(arr, weights)) / (total_w + EPSILON) return clamp(momentum, -1.0, 1.0) def _detect_hh_hl(self) -> bool: """True if recent highs and lows are ascending (uptrend structure).""" if not self._highs.is_ready(self._HH_HL_WINDOW): return False h = self._highs.as_array()[-self._HH_HL_WINDOW:] l = self._lows.as_array()[-self._HH_HL_WINDOW:] hh = all(h[i] >= h[i-1] - EPSILON for i in range(1, len(h))) hl = all(l[i] >= l[i-1] - EPSILON for i in range(1, len(l))) return hh and hl def _detect_lh_ll(self) -> bool: """True if recent highs and lows are descending (downtrend structure).""" if not self._highs.is_ready(self._HH_HL_WINDOW): return False h = self._highs.as_array()[-self._HH_HL_WINDOW:] l = self._lows.as_array()[-self._HH_HL_WINDOW:] lh = all(h[i] <= h[i-1] + EPSILON for i in range(1, len(h))) ll = all(l[i] <= l[i-1] + EPSILON for i in range(1, len(l))) return lh and ll def _detect_compression(self, mean_range: float, std_range: float) -> bool: """Price compressed: range shrinking relative to its own mean. FIX 3: Require min 10 candles for recent window (was 5) and use threshold 0.5 (was 0.6) to avoid false positives on noisy 1m data.""" if not self._ranges.is_ready(self._VOL_WINDOW): return False arr = self._ranges.as_array() # Need at least 10 candles for a stable recent window if len(arr) < 10: return False recent = arr[-10:].mean() if mean_range < EPSILON: return False ratio = safe_div(recent, mean_range) return ratio < 0.5 def _detect_fake_trend(self, momentum: float, mean_range: float, std_range: float) -> bool: """One-candle burst does not qualify as trend.""" if not self._ranges.is_ready(5): return False recent_body = self._bodies.as_array()[-1] if len(self._bodies) > 0 else 0.0 # Large single-candle body but weak directional momentum overall → fake body_z = safe_div(recent_body - mean_range, std_range + EPSILON) return bool(body_z > 2.5 and abs(momentum) < 0.4) def _classify(self, candle: Candle, momentum: float, hh_hl: bool, lh_ll: bool, compression: bool, fake_trend: bool, vol_score: float, di_result: Dict) -> Tuple[str, str, str]: """Return (market_state, regime_label, market_recognition).""" dq = di_result.get("data_quality", 1.0) manip_hint = di_result.get("synthetic_tick", False) or di_result.get("wick_anomaly", False) # --- Market state --- if manip_hint: state = "manipulation" elif compression: state = "compression" elif di_result.get("instability_level", 0.0) > 0.5: state = "unstable" elif di_result.get("noise_burst", False): state = "noisy" elif hh_hl and abs(momentum) > 0.5 and not fake_trend: state = "trend" elif lh_ll and abs(momentum) > 0.5 and not fake_trend: state = "trend" elif di_result.get("wick_anomaly", False): state = "sweep" elif abs(momentum) < 0.2: state = "range" else: state = "continuation" # --- Regime --- if compression: regime = "squeeze" elif manip_hint: regime = "trap" elif hh_hl and momentum > 0.4 and not fake_trend: regime = "bullish_trend" elif lh_ll and momentum < -0.4 and not fake_trend: regime = "bearish_trend" elif hh_hl and momentum > 0.1: regime = "bullish_range" elif lh_ll and momentum < -0.1: regime = "bearish_range" elif fake_trend and momentum > 0: regime = "breakout_attempt" elif fake_trend and momentum < 0: regime = "breakout_attempt" elif abs(momentum) < 0.15: regime = "neutral_dirty" elif vol_score > 0.7 and abs(momentum) > 0.5: regime = "breakout_confirmed" else: regime = "transition" # --- Market recognition --- if compression: recognition = "compression_before_expansion" elif manip_hint: recognition = "manipulation_sweep" elif hh_hl and momentum > 0.6: recognition = "clean_trend" elif lh_ll and momentum < -0.6: recognition = "clean_trend" elif (hh_hl or lh_ll) and 0.2 < abs(momentum) <= 0.6: recognition = "dirty_trend" elif state == "range" and dq > 0.7: recognition = "clean_range" elif state == "range": recognition = "dirty_range" elif state == "sweep": recognition = "manipulation_sweep" elif fake_trend: recognition = "continuation_attempt" elif di_result.get("noise_burst"): recognition = "unstable_noise_cluster" else: recognition = "continuation_attempt" return state, regime, recognition def _resolve_trend_dir(self, hh_hl: bool, lh_ll: bool, momentum: float) -> float: if hh_hl and momentum > 0.3: return 1.0 if lh_ll and momentum < -0.3: return -1.0 return 0.0 def _compute_exhaustion(self, momentum: float, vol_score: float) -> float: """Higher exhaustion when momentum extreme + volatility waning.""" mom_extreme = abs(momentum) vol_waning = clamp(1.0 - vol_score) return clamp(mom_extreme * 0.5 + vol_waning * 0.5) def _compute_continuation(self, momentum: float, hh_hl: bool, lh_ll: bool, compression: bool) -> float: base = abs(momentum) * 0.6 if hh_hl or lh_ll: base += 0.3 if compression: base -= 0.2 return clamp(base) def _compute_regime_shift(self, state: str, momentum: float, vol_score: float) -> float: """Probability that market regime is about to shift.""" if state in ("compression", "squeeze"): return clamp(0.4 + vol_score * 0.3) if abs(momentum) > 0.8: return clamp(0.2 + (abs(momentum) - 0.8) * 1.5) return clamp(abs(momentum) * 0.2) def _local_readability(self, compression: bool, vol_score: float, di_result: Dict) -> float: score = 1.0 if compression: score -= 0.15 if di_result.get("noise_burst"): score -= 0.2 if di_result.get("instability_level", 0.0) > 0.3: score -= 0.2 if di_result.get("synthetic_tick"): score -= 0.15 return clamp(score) @staticmethod def _cold_default() -> Dict: return { "market_state": "undefined", "regime_label": "transition", "market_recognition": "unstable_noise_cluster", "trend_direction": 0.0, "exhaustion_prob": 0.5, "continuation_prob": 0.3, "regime_shift_prob": 0.5, "fake_trend_flag": False, "readability_hint": 0.3, } @staticmethod def _safe_default() -> Dict: return { "market_state": "undefined", "regime_label": "transition", "market_recognition": "unstable_noise_cluster", "trend_direction": 0.0, "exhaustion_prob": 0.5, "continuation_prob": 0.0, "regime_shift_prob": 0.5, "fake_trend_flag": False, "readability_hint": 0.2, } # ============================================================================== # PART C2 — ASSET PROFILE ENGINE # ============================================================================== class AssetProfileEngine: """ Detects asset type (crypto/forex/OTC) from candle characteristics. Outputs RSI thresholds, volatility profile, OTC flags. Input contract: update(candle, di_result) → dict Output contract: { "asset_mode": str, "rsi_overbought": float, "rsi_oversold": float, "otc_probability": float, "crypto_probability": float, "forex_probability": float, "volatility_profile": str, # "low"|"medium"|"high"|"extreme" "broker_controlled_prob": float, } No shared state with other engines. No circular deps. """ _VOL_WINDOW = 50 _OTC_SMOOTH_WINDOW = 20 _CRYPTO_VOL_RATIO = 0.015 # average daily move ratio for crypto baseline def __init__(self) -> None: self._return_buf : RollingBuffer = RollingBuffer(self._VOL_WINDOW) self._spread_buf : RollingBuffer = RollingBuffer(self._VOL_WINDOW) self._range_buf : RollingBuffer = RollingBuffer(self._VOL_WINDOW) self._body_ratio_buf: RollingBuffer = RollingBuffer(self._OTC_SMOOTH_WINDOW) self._tick_count : int = 0 # AUDIT FIX (return-buffer bug): tracks the previous close so the # return calc below no longer has to (mis)use _return_buf for that. self._last_close : float = 0.0 # Asset probabilities (exponential moving) self._crypto_prob : float = 0.33 self._forex_prob : float = 0.33 self._otc_prob : float = 0.33 def update(self, candle: Candle, di_result: Dict) -> Dict: try: return self._update_inner(candle, di_result) except Exception as exc: ENGINE_DIAGNOSTICS.record("AssetProfileEngine", exc) return self._safe_default() def _update_inner(self, candle: Candle, di_result: Dict) -> Dict: if not candle.valid: return self._safe_default() self._tick_count += 1 # Compute return (pct change vs previous close) # AUDIT FIX (return-buffer bug): this previously computed `ret` and # then discarded it, pushing candle.close (a raw price) into # _return_buf instead. Since _return_buf.std() further down is used # as "volatility", that meant volatility was actually std-dev of raw # price LEVEL, not of returns -- harmless near price~1.0 (forex) by # coincidence, but completely broken for any other price magnitude # (e.g. always reads "extreme" for BTC-like prices, since std-dev of # raw prices vastly exceeds the 0.002-0.02 fractional thresholds # _classify_vol() expects). Fix: actually push the return, and track # prev_close separately instead of overloading _return_buf for it. if self._last_close > EPSILON: ret = abs(safe_div(candle.close - self._last_close, self._last_close)) else: ret = 0.0 self._return_buf.push(ret) self._last_close = candle.close self._range_buf.push(candle.candle_range) self._body_ratio_buf.push(candle.body_ratio) if candle.spread > EPSILON: self._spread_buf.push(candle.spread) # Need some warmup if not self._return_buf.is_ready(10): return self._safe_default() vol = self._return_buf.std() avg_range = self._range_buf.mean() price = candle.close # Crypto signal: high volatility, price often large absolute value crypto_vol_signal = clamp(safe_div(vol, self._CRYPTO_VOL_RATIO)) # OTC signal: synthetic smoothness (very high body ratios) avg_body_ratio = self._body_ratio_buf.mean() otc_smooth_signal = clamp((avg_body_ratio - 0.5) * 2.0) otc_synthetic = float(di_result.get("synthetic_tick", False)) otc_signal = clamp(0.6 * otc_smooth_signal + 0.4 * otc_synthetic) # Forex signal: small spread relative to price, small absolute range if price > EPSILON and self._spread_buf.is_ready(5): spread_ratio = safe_div(self._spread_buf.mean(), price) forex_spread_signal = clamp(1.0 - spread_ratio * 1000.0) # pips else: forex_spread_signal = 0.3 # Update EMA probabilities alpha = 0.1 crypto_new = clamp(0.5 * crypto_vol_signal + 0.5 * (1.0 - otc_signal)) otc_new = otc_signal forex_new = clamp(forex_spread_signal * (1.0 - otc_signal) * 0.7) # Normalize to sum to 1 total = crypto_new + otc_new + forex_new + EPSILON crypto_new /= total otc_new /= total forex_new /= total self._crypto_prob = clamp(alpha * crypto_new + (1 - alpha) * self._crypto_prob) self._otc_prob = clamp(alpha * otc_new + (1 - alpha) * self._otc_prob) self._forex_prob = clamp(alpha * forex_new + (1 - alpha) * self._forex_prob) # Normalize again total = self._crypto_prob + self._otc_prob + self._forex_prob + EPSILON self._crypto_prob /= total self._otc_prob /= total self._forex_prob /= total # Determine dominant mode if max(self._crypto_prob, self._otc_prob, self._forex_prob) < 0.4: mode = "hybrid" elif self._crypto_prob > self._otc_prob and self._crypto_prob > self._forex_prob: mode = "crypto" elif self._otc_prob > self._forex_prob: mode = "OTC" else: mode = "forex" # RSI thresholds per mode if mode in ("crypto",): rsi_ob, rsi_os = RSI_OB_CRYPTO, RSI_OS_CRYPTO else: rsi_ob, rsi_os = RSI_OB_DEFAULT, RSI_OS_DEFAULT # Volatility profile vol_profile = self._classify_vol(vol) return { "asset_mode" : mode, "rsi_overbought" : rsi_ob, "rsi_oversold" : rsi_os, "otc_probability" : self._otc_prob, "crypto_probability" : self._crypto_prob, "forex_probability" : self._forex_prob, "volatility_profile" : vol_profile, "broker_controlled_prob": clamp(self._otc_prob * 0.8 + otc_smooth_signal * 0.2), } def _classify_vol(self, vol: float) -> str: if vol < 0.002: return "low" elif vol < 0.008: return "medium" elif vol < 0.020: return "high" else: return "extreme" @staticmethod def _safe_default() -> Dict: return { "asset_mode": "unknown", "rsi_overbought": RSI_OB_DEFAULT, "rsi_oversold": RSI_OS_DEFAULT, "otc_probability": 0.33, "crypto_probability": 0.33, "forex_probability": 0.33, "volatility_profile": "medium", "broker_controlled_prob": 0.33, } # ============================================================================== # PART C3 — SESSION INTELLIGENCE ENGINE # ============================================================================== class SessionIntelligenceEngine: """ Detects current forex/crypto trading session from UTC timestamp. Computes session-based timing penalty/boost and liquidity context. Input contract: update(candle) → dict Output contract: { "session_label": str, "session_liquidity": float, # [0,1] "timing_penalty": float, # [0,1] → higher = worse timing "weekend_flag": bool, "rollover_flag": bool, "session_transition_flag": bool, "session_score": float, # [0,1] timing quality } """ # UTC hour ranges (start, end) — intentionally simple, non-DST _SESSIONS = { "asia" : (0, 8), "london" : (7, 12), "ny_london_overlap": (12, 17), "ny" : (12, 21), "asia_london_overlap": (7, 9), "rollover" : (21, 23), "off_hours" : (23, 24), } # Session liquidity score _SESSION_LIQUIDITY = { "ny_london_overlap" : 1.0, "london" : 0.85, "ny" : 0.80, "asia_london_overlap": 0.70, "asia" : 0.55, "rollover" : 0.30, "off_hours" : 0.20, } def __init__(self) -> None: self._last_session : str = "unknown" self._session_tick : int = 0 def update(self, candle: Candle) -> Dict: try: return self._update_inner(candle) except Exception as exc: ENGINE_DIAGNOSTICS.record("SessionIntelligenceEngine", exc) return self._safe_default() def _update_inner(self, candle: Candle) -> Dict: self._session_tick += 1 ts = candle.timestamp if not math.isfinite(ts) or ts <= 0: return self._safe_default() import datetime try: dt = datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc).replace(tzinfo=None) except (OSError, OverflowError, ValueError): # FIX-6: OS timezone broken — reuse the last known session rather than # returning an off_hours generic default which wrongly penalises timing. if self._last_session not in ("unknown", "off_hours"): liq = self._SESSION_LIQUIDITY.get(self._last_session, 0.4) tp = clamp(1.0 - liq) return { "session_label" : self._last_session, "session_liquidity" : liq, "timing_penalty" : tp, "weekend_flag" : False, "rollover_flag" : False, "session_transition_flag" : False, "session_score" : clamp(1.0 - tp), } return self._safe_default() hour = dt.hour weekday = dt.weekday() # 0=Mon … 6=Sun weekend = weekday >= 5 # Sat, Sun # Rollover window rollover = 20 <= hour < 22 # Determine session session = self._detect_session(hour, weekend) # Transition transition = session != self._last_session and self._last_session != "unknown" self._last_session = session liquidity = self._SESSION_LIQUIDITY.get(session, 0.4) if weekend: liquidity *= 0.3 # Timing penalty: inverse of liquidity, plus penalties timing_penalty = clamp(1.0 - liquidity) if rollover: timing_penalty = clamp(timing_penalty + 0.2) if transition: timing_penalty = clamp(timing_penalty + 0.1) if weekend: timing_penalty = clamp(timing_penalty + 0.4) session_score = clamp(1.0 - timing_penalty) return { "session_label" : session, "session_liquidity" : liquidity, "timing_penalty" : timing_penalty, "weekend_flag" : weekend, "rollover_flag" : rollover, "session_transition_flag" : transition, "session_score" : session_score, } def _detect_session(self, hour: int, weekend: bool) -> str: if weekend: return "off_hours" if 20 <= hour < 22: return "rollover" if hour == 23: return "off_hours" if 7 <= hour < 9: return "asia_london_overlap" if 12 <= hour < 17: return "ny_london_overlap" if 7 <= hour < 12: return "london" if 12 <= hour < 21: return "ny" if 0 <= hour < 8: return "asia" return "off_hours" @staticmethod def _safe_default() -> Dict: return { "session_label": "off_hours", "session_liquidity": 0.4, "timing_penalty": 0.3, "weekend_flag": False, "rollover_flag": False, "session_transition_flag": False, "session_score": 0.5, } # ============================================================================== # PART D MTF EVENTS LIQUIDITY # ============================================================================== import math from collections import deque from typing import Dict, List, Optional, Tuple # ============================================================================== # PART D1 — MULTI-TIMEFRAME FUSION ENGINE (Layer 3) # ============================================================================== class MultiTimeframeFusionEngine: """ Synthesizes three logical timeframe layers from a single OHLC stream. Low TF = short rolling window (recent momentum). Mid TF = medium rolling window (structural continuation). High TF = long rolling window (macro bias). Input contract: update(candle, ms_result) → dict Output contract: { "tf_alignment": float, # [0,1] all layers aligned "tf_conflict_score": float, # [0,1] how much layers disagree "lower_bias": float, # [-1,1] "mid_bias": float, # [-1,1] "higher_bias": float, # [-1,1] "trend_persistence": float, # [0,1] "reversal_probability": float, # [0,1] "timeframe_label": str, # detected base TF } """ _LOW_TF_WIN = 5 _MID_TF_WIN = 20 _HIGH_TF_WIN = 60 def __init__(self) -> None: self._closes : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._highs : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._lows : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._ts_buf : RollingBuffer = RollingBuffer(50) self._detected_tf_seconds: float = 60.0 def update(self, candle: Candle, ms_result: Dict) -> Dict: try: return self._update_inner(candle, ms_result) except Exception as exc: ENGINE_DIAGNOSTICS.record("MultiTimeframeFusionEngine", exc) return self._safe_default() def _update_inner(self, candle: Candle, ms_result: Dict) -> Dict: if not candle.valid: return self._safe_default() self._closes.push(candle.close) self._highs.push(candle.high) self._lows.push(candle.low) self._ts_buf.push(candle.timestamp) # Auto-detect timeframe from timestamp intervals if self._ts_buf.is_ready(3): arr = self._ts_buf.as_array() diffs = arr[1:] - arr[:-1] diffs = diffs[diffs > 0] if len(diffs) > 0: self._detected_tf_seconds = float(diffs.mean()) if not self._closes.is_ready(self._LOW_TF_WIN + 1): return self._safe_default() low_bias = self._compute_bias(self._LOW_TF_WIN) mid_bias = self._compute_bias(self._MID_TF_WIN) high_bias = self._compute_bias(self._HIGH_TF_WIN) # Alignment: product of normalized biases; max when all point same direction alignment = self._compute_alignment(low_bias, mid_bias, high_bias) conflict = clamp(1.0 - alignment) persistence = self._compute_persistence(ms_result) reversal_prob = self._compute_reversal(low_bias, high_bias, ms_result) tf_label = self._label_tf(self._detected_tf_seconds) return { "tf_alignment" : alignment, "tf_conflict_score" : conflict, "lower_bias" : low_bias, "mid_bias" : mid_bias, "higher_bias" : high_bias, "trend_persistence" : persistence, "reversal_probability": reversal_prob, "timeframe_label" : tf_label, } def _compute_bias(self, window: int) -> float: """EMA slope as directional bias in [-1,1].""" n = min(window, len(self._closes)) if n < 2: return 0.0 arr = self._closes.as_array()[-n:] if len(arr) < 2: return 0.0 # Bias: (last - first) / max_range rng = float(arr.max() - arr.min()) if rng < EPSILON: return 0.0 raw_bias = safe_div(float(arr[-1]) - float(arr[0]), rng) return clamp(raw_bias, -1.0, 1.0) def _compute_alignment(self, low: float, mid: float, high: float) -> float: """Alignment: all three biases same sign and magnitude.""" # Sign agreement signs = [math.copysign(1, x) if abs(x) > 0.1 else 0 for x in (low, mid, high)] nonzero_signs = [s for s in signs if s != 0] if not nonzero_signs: return 0.5 sign_agree = len(set(nonzero_signs)) == 1 if not sign_agree: return clamp(0.3 - abs(low + mid + high) * 0.1) # Magnitude alignment: average of absolute biases avg_mag = (abs(low) + abs(mid) + abs(high)) / 3.0 return clamp(0.5 + avg_mag * 0.5) def _compute_persistence(self, ms_result: Dict) -> float: cont = ms_result.get("continuation_prob", 0.5) exh = ms_result.get("exhaustion_prob", 0.5) return clamp(cont - exh * 0.3) def _compute_reversal(self, low: float, high: float, ms_result: Dict) -> float: """Reversal probability: low TF opposes high TF bias.""" diverge = abs(low - high) * 0.5 exh = ms_result.get("exhaustion_prob", 0.0) * 0.5 return clamp(diverge + exh) @staticmethod def _label_tf(seconds: float) -> str: if seconds <= 6: return "5s" if seconds <= 35: return "30s" if seconds <= 75: return "1m" if seconds <= 130: return "2m" if seconds <= 310: return "5m" if seconds <= 610: return "10m" if seconds <= 910: return "15m" return "higher" @property def detected_tf_seconds(self) -> float: """TIMING-FIX-1: Expose auto-detected candle interval for downstream timeframe-aware stale-signal and cooling computations.""" return self._detected_tf_seconds @staticmethod def _safe_default() -> Dict: return { "tf_alignment": 0.5, "tf_conflict_score": 0.5, "lower_bias": 0.0, "mid_bias": 0.0, "higher_bias": 0.0, "trend_persistence": 0.3, "reversal_probability": 0.3, "timeframe_label": "1m", } # ============================================================================== # PART D2 — REAL-TIME EVENT DETECTION ENGINE (Layer 4) # ============================================================================== _EVENT_FIELDS = ( "intensity", "confidence", "continuation_prob", "reversal_prob", "liquidity_impact", "pressure_impact", "manip_prob", ) def _make_event(name: str, severity: str, **kwargs) -> Dict: """Build a validated event dict.""" evt: Dict = { "name" : name, "severity" : severity if severity in EVENT_SEVERITIES else "minor", "intensity" : clamp(kwargs.get("intensity", 0.5)), "confidence" : clamp(kwargs.get("confidence", 0.5)), "continuation_prob": clamp(kwargs.get("continuation_prob", 0.5)), "reversal_prob" : clamp(kwargs.get("reversal_prob", 0.5)), "liquidity_impact" : clamp(kwargs.get("liquidity_impact", 0.3)), "pressure_impact" : clamp(kwargs.get("pressure_impact", 0.3)), "manip_prob" : clamp(kwargs.get("manip_prob", 0.1)), } return evt class EventDetectionEngine: """ Detects 14 real-time market events from candle structure. Each event carries full scoring metadata (Section 12). Input contract: update(candle, di_result, ms_result) → dict Output contract: { "events": [event_dict, ...], # active events this tick "dominant_event": str, # strongest event name "event_severity": str, # overall severity "event_cluster": bool, # multiple simultaneous events "direction_influence": float, # [-1,1] net directional push "manipulation_flag": bool, } """ _RANGE_WINDOW = 20 _MOMENTUM_WINDOW = 5 _CLUSTER_THRESHOLD = 2 # min events to be a cluster _DEDUP_WINDOW = 5 # FIX-8: suppress same event if seen within this many ticks def __init__(self) -> None: self._range_buf : RollingBuffer = RollingBuffer(self._RANGE_WINDOW) self._close_buf : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._high_buf : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._low_buf : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._body_buf : RollingBuffer = RollingBuffer(self._RANGE_WINDOW) self._direction_buf: RollingBuffer = RollingBuffer(self._MOMENTUM_WINDOW) self._wick_zones : deque = deque(maxlen=10) # (high, low) memory self._event_history: RollingObjectBuffer = RollingObjectBuffer(20) # FIX-8: deduplication state — track last tick each event type was emitted self._tick_count : int = 0 self._last_event_tick : Dict[str, int] = {} def update(self, candle: Candle, di_result: Dict, ms_result: Dict) -> Dict: try: return self._update_inner(candle, di_result, ms_result) except Exception as exc: ENGINE_DIAGNOSTICS.record("EventDetectionEngine", exc) return self._safe_default() def _update_inner(self, candle: Candle, di_result: Dict, ms_result: Dict) -> Dict: if not candle.valid: return self._safe_default() self._tick_count += 1 # FIX-8: track absolute tick for dedup window self._range_buf.push(candle.candle_range) self._close_buf.push(candle.close) self._high_buf.push(candle.high) self._low_buf.push(candle.low) self._body_buf.push(candle.body_size) self._direction_buf.push(1.0 if candle.is_bullish else -1.0) self._wick_zones.append((candle.high, candle.low)) if not self._range_buf.is_ready(5): return self._safe_default() events: List[Dict] = [] mean_range = self._range_buf.mean() std_range = self._range_buf.std() mean_body = self._body_buf.mean() # AUDIT FIX (body_z statistical inconsistency): body_z previously # normalized by std_range (the RANGE buffer's spread) while its # numerator came from the BODY buffer's mean -- mixing two different # distributions. range_z, just below, correctly pairs mean_range with # std_range from the same buffer; body_z should likewise be a proper # z-score against the body buffer's own spread. This only changes the # sensitivity of how readily body_z crosses the 1.5/-0.5 thresholds # used below (events 1 and 14); it does not change what's being # measured conceptually. std_body = self._body_buf.std() # Precompute shared values cr = candle.candle_range body = candle.body_size uw = candle.upper_wick lw = candle.lower_wick body_z = safe_div(body - mean_body, std_body + EPSILON) range_z = safe_div(cr - mean_range, std_range + EPSILON) momentum = self._direction_buf.mean() if len(self._direction_buf) > 1 else 0.0 # 1. Sudden aggressive move if range_z > 2.5 and body_z > 1.5: events.append(_make_event("sudden_aggressive_move", "strong", intensity=clamp(range_z / 4.0), confidence=0.7, continuation_prob=0.55, reversal_prob=0.45, pressure_impact=0.8, liquidity_impact=0.6)) # 2. Fast rejection (large wick, small body) if cr > EPSILON: wick_dom = safe_div(max(uw, lw), cr) if wick_dom > 0.6 and range_z > 1.0: dir_rev = -1.0 if uw > lw else 1.0 # reject up → sell pressure events.append(_make_event("fast_rejection", "moderate", intensity=wick_dom, confidence=0.65, continuation_prob=0.35, reversal_prob=0.65, liquidity_impact=0.5, manip_prob=0.2)) # 3. Liquidity sweep (spike beyond then retrace) if self._close_buf.is_ready(3): closes = self._close_buf.as_array()[-3:] highs = self._high_buf.as_array()[-3:] lows = self._low_buf.as_array()[-3:] if highs[-1] > highs[-2] and candle.close < closes[-2]: events.append(_make_event("liquidity_sweep", "strong", intensity=0.8, confidence=0.75, continuation_prob=0.3, reversal_prob=0.7, liquidity_impact=0.9, manip_prob=0.5)) elif lows[-1] < lows[-2] and candle.close > closes[-2]: events.append(_make_event("liquidity_sweep", "strong", intensity=0.8, confidence=0.75, continuation_prob=0.3, reversal_prob=0.7, liquidity_impact=0.9, manip_prob=0.5)) # 4. Stop hunt (fast spike beyond key level + immediate reversal) if di_result.get("wick_anomaly", False) and range_z > 1.5: events.append(_make_event("stop_hunt", "strong", intensity=0.75, confidence=0.6, continuation_prob=0.25, reversal_prob=0.75, manip_prob=0.7)) # 5. Fake breakout (strong range but weak close) if range_z > 1.5 and candle.close_position < 0.3 and candle.is_bullish is False: events.append(_make_event("fake_breakout", "moderate", intensity=0.65, confidence=0.6, continuation_prob=0.2, reversal_prob=0.8, manip_prob=0.4)) elif range_z > 1.5 and candle.close_position > 0.7 and candle.is_bullish is True: # potentially fake if momentum was already exhausted if ms_result.get("exhaustion_prob", 0.0) > 0.6: events.append(_make_event("fake_breakout", "moderate", intensity=0.6, confidence=0.55, continuation_prob=0.25, reversal_prob=0.75, manip_prob=0.35)) # 6. Trap move if di_result.get("synthetic_tick", False) and range_z > 1.0: events.append(_make_event("trap_move", "moderate", intensity=0.7, confidence=0.55, continuation_prob=0.2, reversal_prob=0.8, manip_prob=0.75)) # 7. Spread spike if di_result.get("spread_expansion", False): events.append(_make_event("spread_spike", "moderate", intensity=0.6, confidence=0.8, continuation_prob=0.4, reversal_prob=0.4, manip_prob=0.3)) # 8. Unusual volatility burst if range_z > 3.0: events.append(_make_event("unusual_volatility_burst", "extreme", intensity=clamp(range_z / 5.0), confidence=0.9, continuation_prob=0.45, reversal_prob=0.55, pressure_impact=0.9)) # 9. Momentum flip if len(self._direction_buf) >= 3: dirs = self._direction_buf.as_array()[-3:] if dirs[-1] * dirs[0] < 0 and dirs[-1] * dirs[1] < 0: events.append(_make_event("momentum_flip", "moderate", intensity=0.6, confidence=0.65, continuation_prob=0.6, reversal_prob=0.4, pressure_impact=0.5)) # 10. Hidden absorption (large range, tiny body → buyers/sellers absorbing) if cr > EPSILON: body_to_range = safe_div(body, cr) if body_to_range < 0.2 and range_z > 0.5: events.append(_make_event("hidden_absorption", "moderate", intensity=1.0 - body_to_range, confidence=0.6, continuation_prob=0.45, reversal_prob=0.55, liquidity_impact=0.7)) # 11. Spoof-like behavior (OTC synthetic flag + spread anomaly) if di_result.get("synthetic_tick") and di_result.get("spread_anomaly"): events.append(_make_event("spoof_like_behavior", "moderate", intensity=0.65, confidence=0.5, continuation_prob=0.3, reversal_prob=0.7, manip_prob=0.8)) # 12. Micro reversal cluster (repeated wick rejections at same zone) rejection_zone = self._detect_rejection_cluster(candle) if rejection_zone: events.append(_make_event("micro_reversal_cluster", "moderate", intensity=0.7, confidence=0.65, continuation_prob=0.25, reversal_prob=0.75, liquidity_impact=0.6)) # 13. Liquidity exhaustion (momentum dying, range collapsing) if ms_result.get("exhaustion_prob", 0.0) > 0.7 and range_z < -0.5: events.append(_make_event("liquidity_exhaustion", "moderate", intensity=ms_result.get("exhaustion_prob", 0.5), confidence=0.6, continuation_prob=0.2, reversal_prob=0.8)) # 14. Fake continuation (gap detected + weak body) if di_result.get("gap_detected", False) and body_z < -0.5: events.append(_make_event("fake_continuation", "moderate", intensity=0.6, confidence=0.55, continuation_prob=0.2, reversal_prob=0.8, manip_prob=0.3)) # Store event names in history for e in events: self._event_history.push(e["name"]) # FIX-8: Deduplicate — suppress any event type that already fired within # _DEDUP_WINDOW ticks. Prevents liquidity_sweep (and others) from spamming # on consecutive candles when the triggering condition persists. tick = self._tick_count deduped: List[Dict] = [] for e in events: name = e["name"] if (tick - self._last_event_tick.get(name, -9999)) >= self._DEDUP_WINDOW: deduped.append(e) self._last_event_tick[name] = tick events = deduped # Aggregate result return self._aggregate(events) def _detect_rejection_cluster(self, candle: Candle) -> bool: """True if this candle's wick matches at least 2 prior rejection zones.""" if len(self._wick_zones) < 3: return False tolerance = candle.candle_range * 0.1 + EPSILON uw_zone = candle.high lw_zone = candle.low matches = 0 for (ph, pl) in list(self._wick_zones)[:-1]: if abs(ph - uw_zone) < tolerance or abs(pl - lw_zone) < tolerance: matches += 1 return matches >= 2 def _aggregate(self, events: List[Dict]) -> Dict: if not events: return self._no_event_result() # Dominant: highest intensity dominant = max(events, key=lambda e: e["intensity"]) # Overall severity severity_order = {"minor": 0, "moderate": 1, "strong": 2, "extreme": 3} max_sev = max(events, key=lambda e: severity_order.get(e["severity"], 0)) # Direction influence: weighted by intensity × reversal vs continuation dir_influence = 0.0 total_w = 0.0 for e in events: w = e["intensity"] # reversal events push direction opposite momentum, continuation forward if e["reversal_prob"] > e["continuation_prob"]: contrib = -w * (e["reversal_prob"] - 0.5) * 2.0 else: contrib = w * (e["continuation_prob"] - 0.5) * 2.0 dir_influence += contrib total_w += w dir_influence = clamp(safe_div(dir_influence, total_w + EPSILON), -1.0, 1.0) manip_flag = any(e["manip_prob"] > 0.6 for e in events) return { "events" : events, "dominant_event" : dominant["name"], "event_severity" : max_sev["severity"], "event_cluster" : len(events) >= self._CLUSTER_THRESHOLD, "direction_influence": dir_influence, "manipulation_flag" : manip_flag, } @staticmethod def _no_event_result() -> Dict: return { "events": [], "dominant_event": "none", "event_severity": "minor", "event_cluster": False, "direction_influence": 0.0, "manipulation_flag": False, } @staticmethod def _safe_default() -> Dict: return { "events": [], "dominant_event": "none", "event_severity": "minor", "event_cluster": False, "direction_influence": 0.0, "manipulation_flag": False, } # ============================================================================== # PART D3 — LIQUIDITY + PRESSURE ENGINE (Layer 5) # ============================================================================== class LiquidityPressureEngine: """ Tracks buying/selling pressure balance, liquidity reaction, micro-orderflow proxy (no order book), liquidity void/magnet detection. Input contract: update(candle, di_result, event_result) → dict Output contract: { "pressure_score": float, # [0,1] bullish pressure "liquidity_score": float, # [0,1] liquidity quality "buy_pressure": float, "sell_pressure": float, "absorption_score": float, # [0,1] absorption detected "void_score": float, # [0,1] liquidity void "magnet_zone": bool, # price near liquidity magnet "manipulation_proxy": float, # [0,1] "pressure_chain": float, # multi-candle pressure persistence "otc_fraud_score": float, # Synthetic/broker-control signal } """ _WINDOW = 20 _CHAIN_WINDOW = 7 def __init__(self) -> None: self._close_buf : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._high_buf : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._low_buf : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._range_buf : RollingBuffer = RollingBuffer(self._WINDOW) self._body_buf : RollingBuffer = RollingBuffer(self._WINDOW) self._vol_buf : RollingBuffer = RollingBuffer(self._WINDOW) self._close_pos_buf: RollingBuffer = RollingBuffer(self._WINDOW) # close position [0,1] self._direction_buf: RollingBuffer = RollingBuffer(self._CHAIN_WINDOW) self._wick_ratio_buf: RollingBuffer = RollingBuffer(self._WINDOW) # OTC fraud tracking self._smoothness_buf: RollingBuffer = RollingBuffer(self._WINDOW) # Liquidity levels: track recent swing highs/lows self._swing_highs: deque = deque(maxlen=10) self._swing_lows : deque = deque(maxlen=10) def update(self, candle: Candle, di_result: Dict, event_result: Dict) -> Dict: try: return self._update_inner(candle, di_result, event_result) except Exception as exc: ENGINE_DIAGNOSTICS.record("LiquidityPressureEngine", exc) return self._safe_default() def _update_inner(self, candle: Candle, di_result: Dict, event_result: Dict) -> Dict: if not candle.valid: return self._safe_default() # Push to buffers self._close_buf.push(candle.close) self._high_buf.push(candle.high) self._low_buf.push(candle.low) self._range_buf.push(candle.candle_range) self._body_buf.push(candle.body_size) vol = candle.volume if candle.volume > EPSILON else 0.0 self._vol_buf.push(vol) self._close_pos_buf.push(candle.close_position) self._direction_buf.push(1.0 if candle.is_bullish else -1.0) cr = candle.candle_range if cr > EPSILON: uw_ratio = safe_div(candle.upper_wick, cr) lw_ratio = safe_div(candle.lower_wick, cr) self._wick_ratio_buf.push(uw_ratio - lw_ratio) # >0 = upper wick dominant # Smoothness: high body ratio = synthetic self._smoothness_buf.push(candle.body_ratio) else: self._wick_ratio_buf.push(0.0) self._smoothness_buf.push(0.5) # Update swing high/low self._update_swings(candle) # Need warmup if not self._range_buf.is_ready(5): return self._safe_default() # --- Pressure analysis --- buy_p, sell_p, pressure = self._compute_pressure(candle) # --- Absorption --- absorption = self._compute_absorption(candle) # --- Liquidity quality --- liquidity = self._compute_liquidity(candle, di_result) # --- Void map --- void_score = self._compute_void(candle) # --- Magnet zone --- magnet = self._detect_magnet(candle) # --- Manipulation proxy --- manip_proxy = self._compute_manip_proxy(candle, di_result, event_result) # --- Pressure chain (multi-candle) --- chain = self._compute_pressure_chain() # --- OTC fraud score --- otc_fraud = self._compute_otc_fraud(di_result) return { "pressure_score" : pressure, "liquidity_score" : liquidity, "buy_pressure" : buy_p, "sell_pressure" : sell_p, "absorption_score" : absorption, "void_score" : void_score, "magnet_zone" : magnet, "manipulation_proxy": manip_proxy, "pressure_chain" : chain, "otc_fraud_score" : otc_fraud, } # --- sub-computations --- def _compute_pressure(self, candle: Candle) -> Tuple[float, float, float]: """ Aggressive candle pressure proxy (no order book). Uses body position + close position + volume proxy. """ close_pos = candle.close_position # [0,1] body_size = candle.body_size mean_body = self._body_buf.mean() + EPSILON # Volume-normalized pressure (or body-based if no volume) # AUDIT FIX (volume-proxy bug): is_ready(3) only checks that 3 values # were pushed -- but candle.volume defaults to 0.0 and 0.0 is always # pushed (see _update_inner), so for any feed that never supplies # volume, is_ready(3) becomes True after 3 ticks even though every # value in the buffer is 0.0. That made mean_vol collapse to ~EPSILON # forever, which makes vol_factor (and therefore buy_p/sell_p) lock # at exactly 0.0 permanently -- the documented "0 = absent, triggers # proxy logic" fallback never re-engaged after the 3rd candle. Also # require the recent mean to be meaningfully non-zero. if self._vol_buf.is_ready(3) and self._vol_buf.mean() > EPSILON: vol = candle.volume mean_vol = self._vol_buf.mean() + EPSILON vol_factor = clamp(safe_div(vol, mean_vol)) else: vol_factor = clamp(safe_div(body_size, mean_body)) buy_p = clamp(close_pos * vol_factor) sell_p = clamp((1.0 - close_pos) * vol_factor) # Pressure score: normalized bullish bias pressure = clamp(close_pos * 0.6 + vol_factor * 0.4) return buy_p, sell_p, pressure def _compute_absorption(self, candle: Candle) -> float: """Hidden absorption: large range + tiny body = buyers/sellers absorbing.""" cr = candle.candle_range mean_r = self._range_buf.mean() + EPSILON if cr < EPSILON: return 0.0 body_to_range = safe_div(candle.body_size, cr) range_factor = clamp(safe_div(cr, mean_r) - 1.0) return clamp((1.0 - body_to_range) * (0.5 + range_factor * 0.5)) def _compute_liquidity(self, candle: Candle, di_result: Dict) -> float: """Aggregate liquidity quality from spread, data quality, range stability.""" base = di_result.get("source_confidence", 0.7) if di_result.get("spread_anomaly"): base -= 0.2 if di_result.get("spread_expansion"): base -= 0.15 if di_result.get("noise_burst"): base -= 0.1 if di_result.get("gap_detected"): base -= di_result.get("gap_severity", 0.0) * 0.2 return clamp(base) def _compute_void(self, candle: Candle) -> float: """Liquidity void: rapid price jump through thin zone.""" if not self._range_buf.is_ready(10): return 0.0 mean_r = self._range_buf.mean() std_r = self._range_buf.std() if mean_r < EPSILON: return 0.0 z = safe_div(candle.candle_range - mean_r, std_r + EPSILON) return clamp(z * 0.3) def _detect_magnet(self, candle: Candle) -> bool: """Price approaching recent swing high or low = liquidity magnet.""" if not self._swing_highs or not self._swing_lows: return False price = candle.close mean_r = self._range_buf.mean() tolerance = max(mean_r * 1.5, EPSILON) near_high = any(abs(price - sh) < tolerance for sh in self._swing_highs) near_low = any(abs(price - sl) < tolerance for sl in self._swing_lows) return near_high or near_low def _compute_manip_proxy(self, candle: Candle, di_result: Dict, event_result: Dict) -> float: """Aggregate manipulation probability from multiple signals.""" score = 0.0 if di_result.get("synthetic_tick"): score += 0.3 if di_result.get("wick_anomaly"): score += 0.15 if di_result.get("spread_anomaly"): score += 0.15 if event_result.get("manipulation_flag"): score += 0.25 # Velocity divergence proxy: large range + tiny volume # AUDIT FIX (volume-proxy bug): same is_ready(3)-only gating issue as # _compute_pressure -- without the mean>EPSILON check, a permanently # volume-absent feed makes vol_factor=0.0 forever, so this check # degenerates into firing on any range spike regardless of volume. if self._vol_buf.is_ready(3) and self._vol_buf.mean() > EPSILON: mean_vol = self._vol_buf.mean() + EPSILON mean_r = self._range_buf.mean() + EPSILON vol_factor = safe_div(candle.volume, mean_vol) range_factor = safe_div(candle.candle_range, mean_r) if range_factor > 2.0 and vol_factor < 0.4: score += 0.2 return clamp(score) def _compute_pressure_chain(self) -> float: """Multi-candle directional pressure persistence (3-7 candles).""" if not self._direction_buf.is_ready(3): return 0.5 arr = self._direction_buf.as_array() mean_dir = float(arr.mean()) return clamp(0.5 + mean_dir * 0.5) def _compute_otc_fraud(self, di_result: Dict) -> float: """Synthetic OTC fraud score (Section 15, engine 13).""" if not self._smoothness_buf.is_ready(5): return 0.0 avg_smooth = self._smoothness_buf.mean() fraud = 0.0 fraud += clamp((avg_smooth - 0.5) * 1.5) # high body ratio → synthetic if di_result.get("synthetic_tick"): fraud += 0.3 if di_result.get("wick_anomaly"): fraud += 0.1 # Candle repetition: check if range is abnormally uniform if self._range_buf.is_ready(10): std_r = self._range_buf.std() mean_r = self._range_buf.mean() + EPSILON cv = safe_div(std_r, mean_r) # coefficient of variation if cv < 0.05: # suspiciously uniform candles fraud += 0.2 return clamp(fraud) def _update_swings(self, candle: Candle) -> None: """Track recent swing highs/lows for magnet detection.""" if not self._high_buf.is_ready(3): return highs = self._high_buf.as_array() lows = self._low_buf.as_array() # Previous candle is a swing high if its high > neighbors if len(highs) >= 3: if highs[-2] > highs[-3] and highs[-2] > highs[-1]: self._swing_highs.append(float(highs[-2])) if lows[-2] < lows[-3] and lows[-2] < lows[-1]: self._swing_lows.append(float(lows[-2])) @staticmethod def _safe_default() -> Dict: return { "pressure_score": 0.5, "liquidity_score": 0.5, "buy_pressure": 0.5, "sell_pressure": 0.5, "absorption_score": 0.0, "void_score": 0.0, "magnet_zone": False, "manipulation_proxy": 0.0, "pressure_chain": 0.5, "otc_fraud_score": 0.0, } # ============================================================================== # PART E TECHNICAL STACK # ============================================================================== import math from collections import deque from typing import Dict, List, Optional, Tuple # ============================================================================== # SHARED EMA HELPER # ============================================================================== class EMATracker: """ Incremental Exponential Moving Average. Seeded from SMA on first `period` values, then online from there. Thread-safe state in pure Python. Never raises. """ __slots__ = ("period", "_ema", "_count", "_alpha", "_seed_sum") def __init__(self, period: int) -> None: assert period > 0 self.period = period self._alpha = 2.0 / (period + 1.0) self._ema : float = 0.0 self._count : int = 0 self._seed_sum: float = 0.0 def update(self, value: float) -> Optional[float]: """Push a new value. Returns current EMA or None if not yet seeded.""" v = nan_safe(value) self._count += 1 if self._count <= self.period: self._seed_sum += v if self._count == self.period: self._ema = self._seed_sum / self.period return self._ema return None self._ema = self._alpha * v + (1.0 - self._alpha) * self._ema return self._ema @property def value(self) -> Optional[float]: return self._ema if self._count >= self.period else None def is_ready(self) -> bool: return self._count >= self.period # ============================================================================== # PART E1 — EMA 200 (Macro Trend Direction) # ============================================================================== class EMA200Engine: """ Determines macro trend direction via EMA 200. Output: { "ema200": float | None, "price_above_ema200": bool, "ema200_slope": float, "macro_bias": float } # bias [-1,1] """ def __init__(self) -> None: self._ema = EMATracker(EMA200_PERIOD) self._prev_ema: Optional[float] = None def update(self, candle: Candle) -> Dict: try: return self._update_inner(candle) except Exception as exc: ENGINE_DIAGNOSTICS.record("EMA200Engine", exc) return {"ema200": None, "price_above_ema200": False, "ema200_slope": 0.0, "macro_bias": 0.0} def _update_inner(self, candle: Candle) -> Dict: if not candle.valid: return {"ema200": None, "price_above_ema200": False, "ema200_slope": 0.0, "macro_bias": 0.0} ema_val = self._ema.update(candle.close) if ema_val is None: return {"ema200": None, "price_above_ema200": False, "ema200_slope": 0.0, "macro_bias": 0.0} slope = 0.0 if self._prev_ema is not None and self._prev_ema > EPSILON: slope = safe_div(ema_val - self._prev_ema, self._prev_ema) self._prev_ema = ema_val above = candle.close > ema_val # Macro bias: stronger when price significantly above/below EMA price_dist = safe_div(candle.close - ema_val, ema_val + EPSILON) macro_bias = clamp(price_dist * 10.0, -1.0, 1.0) # normalize return { "ema200" : ema_val, "price_above_ema200": above, "ema200_slope" : clamp(slope * 100.0, -1.0, 1.0), "macro_bias" : macro_bias, } # ============================================================================== # PART E2 — RSI ADAPTIVE ENGINE # ============================================================================== class RSIEngine: """ Adaptive RSI (period 14 by default). Thresholds sourced from asset_profile_result dynamically. RSI is a CONFIRMATION FILTER only — not standalone signal. Output: { "rsi": float, "rsi_ob_flag": bool, "rsi_os_flag": bool, "rsi_divergence": float, "rsi_momentum": float, "rsi_confirmation": float } # [-1,1] contribution to scoring """ _DIV_WINDOW: int = 10 # FIX-10: lookback window for divergence detection def __init__(self, period: int = RSI_PERIOD) -> None: self.period = period self._gains : deque = deque(maxlen=period * 2) self._losses : deque = deque(maxlen=period * 2) self._avg_gain: float = 0.0 self._avg_loss: float = 0.0 self._prev_close: Optional[float] = None self._rsi_buf : RollingBuffer = RollingBuffer(20) self._count : int = 0 # FIX-10: price close buffer for RSI divergence computation self._price_div_buf: RollingBuffer = RollingBuffer(self._DIV_WINDOW + 2) def update(self, candle: Candle, asset_result: Dict) -> Dict: try: return self._update_inner(candle, asset_result) except Exception as exc: ENGINE_DIAGNOSTICS.record("RSIEngine", exc) return self._safe_default(asset_result) def _update_inner(self, candle: Candle, asset_result: Dict) -> Dict: if not candle.valid: return self._safe_default(asset_result) close = candle.close rsi_ob = asset_result.get("rsi_overbought", RSI_OB_DEFAULT) rsi_os = asset_result.get("rsi_oversold", RSI_OS_DEFAULT) if self._prev_close is not None: delta = close - self._prev_close gain = max(0.0, delta) loss = max(0.0, -delta) else: gain = loss = 0.0 self._prev_close = close self._count += 1 self._gains.append(gain) self._losses.append(loss) if self._count < self.period: return self._safe_default(asset_result) # Smoothed RS if self._count == self.period: self._avg_gain = sum(list(self._gains)[-self.period:]) / self.period self._avg_loss = sum(list(self._losses)[-self.period:]) / self.period else: alpha = 1.0 / self.period self._avg_gain = alpha * gain + (1 - alpha) * self._avg_gain self._avg_loss = alpha * loss + (1 - alpha) * self._avg_loss if self._avg_loss < EPSILON: rsi = 100.0 else: rs = safe_div(self._avg_gain, self._avg_loss) rsi = 100.0 - safe_div(100.0, 1.0 + rs) rsi = clamp(rsi, 0.0, 100.0) self._rsi_buf.push(rsi) # FIX-10: Compute real RSI divergence (was always 0.0) self._price_div_buf.push(close) divergence = self._compute_divergence() # Momentum: slope of RSI rsi_momentum = 0.0 if self._rsi_buf.is_ready(3): arr = self._rsi_buf.as_array()[-3:] rsi_momentum = clamp(safe_div(float(arr[-1]) - float(arr[0]), 50.0), -1.0, 1.0) # Divergence: see _compute_divergence() — populated above ob_flag = rsi >= rsi_ob os_flag = rsi <= rsi_os # RSI confirmation contribution [-1, 1] # OS zone → bullish confirmation; OB zone → bearish confirmation if os_flag: rsi_confirm = clamp((rsi_os - rsi) / (rsi_os + EPSILON) * 2.0) elif ob_flag: rsi_confirm = clamp(-(rsi - rsi_ob) / (100.0 - rsi_ob + EPSILON) * 2.0, -1.0, 0.0) else: # Neutral: contribution based on distance from 50 rsi_confirm = clamp((rsi - 50.0) / 50.0, -1.0, 1.0) * 0.3 return { "rsi" : rsi, "rsi_ob_flag" : ob_flag, "rsi_os_flag" : os_flag, "rsi_divergence" : divergence, "rsi_momentum" : rsi_momentum, "rsi_confirmation" : rsi_confirm, } def _compute_divergence(self) -> float: """ FIX-10: RSI divergence — compare price direction to RSI direction over the last _DIV_WINDOW candles. Bearish divergence (returns negative): price forms a higher high while RSI forms a lower high → momentum fading despite price rising. Bullish divergence (returns positive): price forms a lower low while RSI forms a higher low → downside momentum fading. Returns [-1, 1]. 0.0 when insufficient history or no divergence. """ n = self._DIV_WINDOW if not self._price_div_buf.is_ready(n) or not self._rsi_buf.is_ready(n): return 0.0 prices = self._price_div_buf.as_array()[-n:] rsi_vals = self._rsi_buf.as_array()[-n:] if len(prices) < n or len(rsi_vals) < n: return 0.0 half = n // 2 p_early_max = float(prices[:half].max()) p_late_max = float(prices[half:].max()) p_early_min = float(prices[:half].min()) p_late_min = float(prices[half:].min()) r_early_max = float(rsi_vals[:half].max()) r_late_max = float(rsi_vals[half:].max()) r_early_min = float(rsi_vals[:half].min()) r_late_min = float(rsi_vals[half:].min()) # Bearish divergence: price higher high + RSI lower high (≥2 RSI points) if p_late_max > p_early_max and r_late_max < r_early_max - 2.0: magnitude = safe_div(r_early_max - r_late_max, 50.0) return -clamp(magnitude) # Bullish divergence: price lower low + RSI higher low (≥2 RSI points) if p_late_min < p_early_min and r_late_min > r_early_min + 2.0: magnitude = safe_div(r_late_min - r_early_min, 50.0) return clamp(magnitude) return 0.0 def _safe_default(self, asset_result: Dict) -> Dict: return { "rsi": 50.0, "rsi_ob_flag": False, "rsi_os_flag": False, "rsi_divergence": 0.0, "rsi_momentum": 0.0, "rsi_confirmation": 0.0, } # ============================================================================== # PART E3 — MACD ADAPTIVE MOMENTUM ENGINE # ============================================================================== class MACDEngine: """ MACD (12/26/9) — detects momentum acceleration/exhaustion. Suppresses small/weak crossovers (false crossover suppression). Output: { "macd": float, "signal": float, "histogram": float, "hist_slope": float, "momentum_quality": float, "macd_confirmation": float } # [-1,1] """ def __init__(self) -> None: self._ema_fast = EMATracker(MACD_FAST) self._ema_slow = EMATracker(MACD_SLOW) self._ema_signal = EMATracker(MACD_SIGNAL) self._hist_buf : RollingBuffer = RollingBuffer(10) def update(self, candle: Candle) -> Dict: try: return self._update_inner(candle) except Exception as exc: ENGINE_DIAGNOSTICS.record("MACDEngine", exc) return self._safe_default() def _update_inner(self, candle: Candle) -> Dict: if not candle.valid: return self._safe_default() fast_val = self._ema_fast.update(candle.close) slow_val = self._ema_slow.update(candle.close) if fast_val is None or slow_val is None: return self._safe_default() macd_line = fast_val - slow_val sig_val = self._ema_signal.update(macd_line) if sig_val is None: return self._safe_default() histogram = macd_line - sig_val self._hist_buf.push(histogram) # Histogram slope hist_slope = 0.0 if self._hist_buf.is_ready(3): arr = self._hist_buf.as_array()[-3:] hist_slope = clamp(float(arr[-1]) - float(arr[0]), -1.0, 1.0) # Momentum quality: slope magnitude + histogram magnitude hist_mag = abs(histogram) slope_mag = abs(hist_slope) momentum_quality = clamp(hist_mag * 5.0 + slope_mag * 2.0) # Confirmation: direction + magnitude, suppressing small crossovers mean_hist_mag = self._hist_buf.mean() if abs(histogram) < mean_hist_mag * 0.3: # Small crossover — suppress macd_confirm = clamp(histogram * 2.0, -0.2, 0.2) else: macd_confirm = clamp(histogram * 5.0 + hist_slope * 2.0, -1.0, 1.0) return { "macd" : macd_line, "signal" : sig_val, "histogram" : histogram, "hist_slope" : hist_slope, "momentum_quality" : momentum_quality, "macd_confirmation" : macd_confirm, } @staticmethod def _safe_default() -> Dict: return { "macd": 0.0, "signal": 0.0, "histogram": 0.0, "hist_slope": 0.0, "momentum_quality": 0.5, "macd_confirmation": 0.0, } # ============================================================================== # PART E4 — BOLLINGER VOLATILITY ENGINE # ============================================================================== class BollingerEngine: """ Bollinger Bands (20, 2). Identifies squeeze, band touch, expansion. NOT a standalone signal. Feeds adaptive scoring engine. Output: { "upper": float, "lower": float, "middle": float, "bandwidth": float, "bandwidth_z": float, "squeeze_flag": bool, "upper_touch": bool, "lower_touch": bool, "expansion_flag": bool, "bb_confirmation": float } """ _BW_WINDOW = 40 # bandwidth history for z-score def __init__(self) -> None: self._close_buf : RollingBuffer = RollingBuffer(BB_PERIOD * 3) self._bw_buf : RollingBuffer = RollingBuffer(self._BW_WINDOW) def update(self, candle: Candle) -> Dict: try: return self._update_inner(candle) except Exception as exc: ENGINE_DIAGNOSTICS.record("BollingerEngine", exc) return self._safe_default() def _update_inner(self, candle: Candle) -> Dict: if not candle.valid: return self._safe_default() self._close_buf.push(candle.close) if not self._close_buf.is_ready(BB_PERIOD): return self._safe_default() arr = self._close_buf.as_array()[-BB_PERIOD:] middle = float(arr.mean()) std = float(arr.std()) upper = middle + BB_STD_DEV * std lower = middle - BB_STD_DEV * std bw = safe_div(upper - lower, middle + EPSILON) self._bw_buf.push(bw) # Bandwidth z-score for squeeze/expansion detection bw_z = 0.0 squeeze = False expansion = False if self._bw_buf.is_ready(10): bw_mean = self._bw_buf.mean() bw_std = self._bw_buf.std() bw_z = safe_div(bw - bw_mean, bw_std + EPSILON) squeeze = bw_z < -1.0 expansion = bw_z > 1.5 # Touch detection (with tolerance) price = candle.close tol = std * 0.1 upper_touch = price >= upper - tol lower_touch = price <= lower + tol # Confirmation: sell zone at upper touch, buy zone at lower touch bb_confirm = 0.0 if lower_touch: bb_confirm = clamp((lower - price + tol) / (std + EPSILON)) # bullish contribution elif upper_touch: bb_confirm = clamp(-(price - upper + tol) / (std + EPSILON)) # bearish if squeeze: bb_confirm *= 0.5 # squeeze doesn't strongly confirm direction return { "upper" : upper, "lower" : lower, "middle" : middle, "bandwidth" : bw, "bandwidth_z" : bw_z, "squeeze_flag" : squeeze, "upper_touch" : upper_touch, "lower_touch" : lower_touch, "expansion_flag": expansion, "bb_confirmation": clamp(bb_confirm, -1.0, 1.0), } @staticmethod def _safe_default() -> Dict: return { "upper": 0.0, "lower": 0.0, "middle": 0.0, "bandwidth": 0.0, "bandwidth_z": 0.0, "squeeze_flag": False, "upper_touch": False, "lower_touch": False, "expansion_flag": False, "bb_confirmation": 0.0, } # ============================================================================== # PART E5 — ADVANCED SUPPORT + RESISTANCE ENGINE # ============================================================================== class SupportResistanceEngine: """ Adaptive S/R via swing highs/lows, round numbers, session levels. Level strength = touch count + reaction strength + recency. Levels decay using confidence decay logic. Output: { "nearest_support": float, "nearest_resistance": float, "at_support": bool, "at_resistance": bool, "level_strength": float, # [0,1] proximity to strong level "stop_hunt_probability": float, "sr_confirmation": float } # [-1,1] """ _MAX_LEVELS = 20 _LEVEL_DECAY = 0.02 # per tick decay def __init__(self) -> None: self._close_buf : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._high_buf : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._low_buf : RollingBuffer = RollingBuffer(DEFAULT_BUFFER_SIZE) self._range_buf : RollingBuffer = RollingBuffer(50) # Level registry: {price: {"strength": float, "type": "support"|"resistance", "age": int}} self._levels : Dict[float, Dict] = {} self._tick : int = 0 def update(self, candle: Candle, session_result: Dict) -> Dict: try: return self._update_inner(candle, session_result) except Exception as exc: ENGINE_DIAGNOSTICS.record("SupportResistanceEngine", exc) return self._safe_default() def _update_inner(self, candle: Candle, session_result: Dict) -> Dict: if not candle.valid: return self._safe_default() self._tick += 1 self._close_buf.push(candle.close) self._high_buf.push(candle.high) self._low_buf.push(candle.low) self._range_buf.push(candle.candle_range) # Update levels from swing detection self._update_levels_from_swings() # Add round number levels self._add_round_numbers(candle.close) # Decay all levels self._decay_levels() # Prune weak levels self._prune_levels() if not self._levels: return self._safe_default() price = candle.close mean_range = self._range_buf.mean() if self._range_buf.is_ready(5) else candle.candle_range tolerance = max(mean_range * 1.0, EPSILON) supports = [(p, v) for p, v in self._levels.items() if v["type"] == "support" and p < price] resistances = [(p, v) for p, v in self._levels.items() if v["type"] == "resistance" and p > price] nearest_sup = max(supports, key=lambda x: x[0])[0] if supports else price * 0.99 nearest_res = min(resistances, key=lambda x: x[0])[0] if resistances else price * 1.01 at_support = abs(price - nearest_sup) < tolerance at_resistance = abs(price - nearest_res) < tolerance # Level strength: proximity-weighted strength of nearest levels level_strength = 0.0 for p, v in self._levels.items(): dist = abs(price - p) proximity = clamp(1.0 - safe_div(dist, tolerance * 3.0)) level_strength = max(level_strength, proximity * v["strength"]) # Stop hunt: price at level + spike behavior stop_hunt_prob = clamp(level_strength * 0.5 * (1.0 if at_support or at_resistance else 0.3)) # SR confirmation: buy near support, sell near resistance sr_confirm = 0.0 if at_support: sr_confirm = clamp(level_strength) elif at_resistance: sr_confirm = clamp(-level_strength) return { "nearest_support" : nearest_sup, "nearest_resistance" : nearest_res, "at_support" : at_support, "at_resistance" : at_resistance, "level_strength" : level_strength, "stop_hunt_probability": stop_hunt_prob, "sr_confirmation" : sr_confirm, } def _update_levels_from_swings(self) -> None: if not self._high_buf.is_ready(3): return highs = self._high_buf.as_array() lows = self._low_buf.as_array() if len(highs) < 3: return # Swing high if highs[-2] > highs[-3] and highs[-2] > highs[-1]: self._register_level(float(highs[-2]), "resistance") # Swing low if lows[-2] < lows[-3] and lows[-2] < lows[-1]: self._register_level(float(lows[-2]), "support") def _add_round_numbers(self, price: float) -> None: """Add proximity-based round number levels.""" # Snap to round numbers within 2% of current price if price < EPSILON: return magnitude = 10 ** (math.floor(math.log10(price)) - 1) rounded = round(price / magnitude) * magnitude if abs(rounded - price) < price * 0.02: self._register_level(rounded, "resistance" if rounded > price else "support") def _register_level(self, price: float, level_type: str) -> None: """Register or reinforce a level.""" if len(self._levels) >= self._MAX_LEVELS and price not in self._levels: # Evict weakest weakest = min(self._levels, key=lambda p: self._levels[p]["strength"]) del self._levels[weakest] if price in self._levels: self._levels[price]["strength"] = min(1.0, self._levels[price]["strength"] + 0.1) self._levels[price]["age"] = self._tick else: self._levels[price] = {"strength": 0.3, "type": level_type, "age": self._tick} def _decay_levels(self) -> None: for p in self._levels: self._levels[p]["strength"] = max(0.0, self._levels[p]["strength"] - self._LEVEL_DECAY) def _prune_levels(self) -> None: remove = [p for p, v in self._levels.items() if v["strength"] < 0.01] for p in remove: del self._levels[p] @staticmethod def _safe_default() -> Dict: return { "nearest_support": 0.0, "nearest_resistance": 0.0, "at_support": False, "at_resistance": False, "level_strength": 0.0, "stop_hunt_probability": 0.0, "sr_confirmation": 0.0, } # ============================================================================== # TECHNICAL CONFIRMATION STACK — AGGREGATOR # ============================================================================== class TechnicalConfirmationStack: """ Coordinates EMA200, RSI, MACD, Bollinger, S&R. Applies conflict resolution (Section 14). Minimum 2-layer alignment before signal. 4-layer = max confidence. Input contract: update(candle, asset_result, session_result) → dict Output contract: { "ema200_result": dict, "rsi_result": dict, "macd_result": dict, "bb_result": dict, "sr_result": dict, "layers_aligned": int, # 0–4 layers agreeing "tech_confidence": float, # [0,1] "tech_direction": float, # [-1,1] "conflict_flag": bool, } """ def __init__(self) -> None: self._ema200 = EMA200Engine() self._rsi = RSIEngine() self._macd = MACDEngine() self._bb = BollingerEngine() self._sr = SupportResistanceEngine() def update(self, candle: Candle, asset_result: Dict, session_result: Dict) -> Dict: try: return self._update_inner(candle, asset_result, session_result) except Exception as exc: ENGINE_DIAGNOSTICS.record("TechnicalConfirmationStack", exc) return self._safe_default() def _update_inner(self, candle: Candle, asset_result: Dict, session_result: Dict) -> Dict: ema_r = self._ema200.update(candle) rsi_r = self._rsi.update(candle, asset_result) macd_r = self._macd.update(candle) bb_r = self._bb.update(candle) sr_r = self._sr.update(candle, session_result) # Gather directional signals per layer signals: List[float] = [] # EMA200 if ema_r.get("ema200") is not None: signals.append(1.0 if ema_r["price_above_ema200"] else -1.0) # RSI rsi_conf = rsi_r.get("rsi_confirmation", 0.0) if abs(rsi_conf) > 0.1: signals.append(math.copysign(1.0, rsi_conf)) # MACD macd_conf = macd_r.get("macd_confirmation", 0.0) if abs(macd_conf) > 0.1: signals.append(math.copysign(1.0, macd_conf)) # Bollinger bb_conf = bb_r.get("bb_confirmation", 0.0) if abs(bb_conf) > 0.1: signals.append(math.copysign(1.0, bb_conf)) # S&R sr_conf = sr_r.get("sr_confirmation", 0.0) if abs(sr_conf) > 0.1: signals.append(math.copysign(1.0, sr_conf)) if not signals: return self._compose(ema_r, rsi_r, macd_r, bb_r, sr_r, 0, 0.3, 0.0, False) # Count aligned layers bull_count = sum(1 for s in signals if s > 0) bear_count = sum(1 for s in signals if s < 0) aligned_count = max(bull_count, bear_count) # Conflict resolution (Section 14): # EMA bearish + RSI oversold → caution, wait for MACD ema_bear = ema_r.get("ema200") is not None and not ema_r.get("price_above_ema200", True) rsi_os = rsi_r.get("rsi_os_flag", False) if ema_bear and rsi_os: aligned_count = max(0, aligned_count - 1) conflict = bull_count > 0 and bear_count > 0 # Confidence: 2 layers min needed; scales to 4 tech_conf = clamp(safe_div(aligned_count - 1, 3)) # 1→0, 2→0.33, 3→0.67, 4→1.0 # Direction: weighted average of confirmations all_confs = [ rsi_r.get("rsi_confirmation", 0.0), macd_r.get("macd_confirmation", 0.0), bb_r.get("bb_confirmation", 0.0), sr_r.get("sr_confirmation", 0.0), ] if ema_r.get("ema200") is not None: ema_bias = ema_r.get("macro_bias", 0.0) all_confs.append(ema_bias * 0.5) tech_dir = clamp(sum(all_confs) / (len(all_confs) + EPSILON), -1.0, 1.0) # Manipulation override: if manipulation detected, suppress confirmation # (handled in scoring engine — passed through here) return self._compose(ema_r, rsi_r, macd_r, bb_r, sr_r, aligned_count, tech_conf, tech_dir, conflict) @staticmethod def _compose(ema_r, rsi_r, macd_r, bb_r, sr_r, aligned, conf, direction, conflict) -> Dict: return { "ema200_result" : ema_r, "rsi_result" : rsi_r, "macd_result" : macd_r, "bb_result" : bb_r, "sr_result" : sr_r, "layers_aligned" : aligned, "tech_confidence": conf, "tech_direction" : direction, "conflict_flag" : conflict, } @staticmethod def _safe_default() -> Dict: return { "ema200_result": {"ema200": None, "price_above_ema200": False, "ema200_slope": 0.0, "macro_bias": 0.0}, "rsi_result": {"rsi": 50.0, "rsi_ob_flag": False, "rsi_os_flag": False, "rsi_divergence": 0.0, "rsi_momentum": 0.0, "rsi_confirmation": 0.0}, "macd_result": {"macd": 0.0, "signal": 0.0, "histogram": 0.0, "hist_slope": 0.0, "momentum_quality": 0.5, "macd_confirmation": 0.0}, "bb_result": {"upper": 0.0, "lower": 0.0, "middle": 0.0, "bandwidth": 0.0, "bandwidth_z": 0.0, "squeeze_flag": False, "upper_touch": False, "lower_touch": False, "expansion_flag": False, "bb_confirmation": 0.0}, "sr_result": {"nearest_support": 0.0, "nearest_resistance": 0.0, "at_support": False, "at_resistance": False, "level_strength": 0.0, "stop_hunt_probability": 0.0, "sr_confirmation": 0.0}, "layers_aligned": 0, "tech_confidence": 0.3, "tech_direction": 0.0, "conflict_flag": False, } # ============================================================================== # PART F SCORING LIFECYCLE # ============================================================================== import math from typing import Dict, List, Optional, Tuple # ============================================================================== # PART F1 — ADAPTIVE SCORING ENGINE # ============================================================================== class AdaptiveScoringEngine: """ Computes the master confidence score and execution suitability from all upstream engine results. Weights adapt every tick based on market state, volatility, and manipulation risk — no static formula (Section 16). Input contract: score(di, ms, asset, session, tf, event, lp, tech, tick_count) → dict Output contract: { "confidence": float, "internal_trust": float, "execution_suitability": str, "momentum_confidence": float, "liquidity_confidence": float, "structure_confidence": float, "volatility_confidence": float, "manipulation_penalty": float, "noise_penalty": float, "spread_penalty": float, "latency_penalty": float, "regime_confidence": float, "continuation_probability":float, "reversal_probability": float, "tech_alignment_bonus": float, "raw_direction": float, # [-1,1] "warm_up_fraction": float, # [0,1] } """ def score(self, di_result : Dict, ms_result : Dict, asset_result : Dict, session_result: Dict, tf_result : Dict, event_result : Dict, lp_result : Dict, tech_result : Dict, tick_count : int) -> Dict: try: return self._score_inner( di_result, ms_result, asset_result, session_result, tf_result, event_result, lp_result, tech_result, tick_count) except Exception as exc: ENGINE_DIAGNOSTICS.record("AdaptiveScoringEngine", exc) return self._safe_default() def _score_inner(self, di, ms, asset, session, tf, event, lp, tech, tick_count) -> Dict: # ---------------------------------------------------------------- # 1. Warm-up fraction — confidence gated during cold start # ---------------------------------------------------------------- warm_frac = clamp(safe_div(tick_count - MIN_WARM_CANDLES, FULL_WARM_CANDLES - MIN_WARM_CANDLES)) # ---------------------------------------------------------------- # 2. Penalty signals (from data quality) # ---------------------------------------------------------------- data_quality = di.get("data_quality", 1.0) instability = di.get("instability_level", 0.0) noise_penalty = clamp(instability * 0.5 + float(di.get("noise_burst", False)) * 0.2) spread_penalty = clamp( float(di.get("spread_anomaly", False)) * 0.3 + float(di.get("spread_expansion", False)) * 0.2 ) latency_penalty = clamp( float(di.get("latency_spike", False)) * 0.15 + float(di.get("latency_drift", False)) * 0.1 ) # ---------------------------------------------------------------- # 3. Manipulation penalty # ---------------------------------------------------------------- manip_proxy = lp.get("manipulation_proxy", 0.0) event_manip = float(event.get("manipulation_flag", False)) * 0.3 # FIX-7: OTC fraud score fires falsely on low-volatility forex whose high # body-ratio candles look "synthetic" by the smoothness heuristic. # Dampen by 0.6 for every asset class that is not crypto. otc_fraud_raw = lp.get("otc_fraud_score", 0.0) if asset.get("asset_mode", "unknown") != "crypto": otc_fraud_raw *= 0.6 otc_fraud = otc_fraud_raw * 0.5 manip_penalty = clamp(manip_proxy * 0.5 + event_manip + otc_fraud) # ---------------------------------------------------------------- # 4. Component confidence scores # ---------------------------------------------------------------- momentum_conf = clamp(tech.get("macd_result", {}).get("momentum_quality", 0.5) * 0.6 + abs(tf.get("lower_bias", 0.0)) * 0.4) liquidity_conf = lp.get("liquidity_score", 0.5) structure_conf = clamp( tf.get("tf_alignment", 0.5) * 0.5 + ms.get("continuation_prob", 0.5) * 0.3 + tech.get("sr_result", {}).get("level_strength", 0.0) * 0.2 ) vol_score = asset.get("volatility_profile", "medium") volatility_conf = {"low": 0.6, "medium": 0.8, "high": 0.5, "extreme": 0.2}.get(vol_score, 0.6) regime_conf = clamp( (1.0 - ms.get("exhaustion_prob", 0.5)) * 0.5 + ms.get("continuation_prob", 0.5) * 0.5 ) continuation_p = clamp( ms.get("continuation_prob", 0.5) * 0.4 + tf.get("trend_persistence", 0.5) * 0.3 + lp.get("pressure_chain", 0.5) * 0.3 ) reversal_p = clamp( ms.get("exhaustion_prob", 0.5) * 0.4 + tf.get("reversal_probability", 0.3) * 0.4 + event.get("direction_influence", 0.0) * -0.2 ) tech_align_bonus = clamp(safe_div(tech.get("layers_aligned", 0) - 1, 3) * 0.3) # ---------------------------------------------------------------- # 5. Adaptive weights — recalibrate from market state # ---------------------------------------------------------------- market_state = ms.get("market_state", "undefined") # In trend: weight structure + momentum more # In range: weight S/R + liquidity more # In manipulation/sweep: heavily penalize if market_state in ("trend", "continuation"): w_momentum = 0.30 w_liquidity = 0.15 w_structure = 0.30 w_regime = 0.25 elif market_state in ("range", "compression"): w_momentum = 0.15 w_liquidity = 0.25 w_structure = 0.35 w_regime = 0.25 elif market_state in ("manipulation", "sweep", "unstable", "noisy"): w_momentum = 0.10 w_liquidity = 0.10 w_structure = 0.10 w_regime = 0.10 manip_penalty = clamp(manip_penalty + 0.4) elif market_state == "reversal": w_momentum = 0.25 w_liquidity = 0.20 w_structure = 0.25 w_regime = 0.30 else: w_momentum = 0.25 w_liquidity = 0.20 w_structure = 0.25 w_regime = 0.30 # Base confidence (before penalties) base_conf = weighted_mean( [momentum_conf, liquidity_conf, structure_conf, regime_conf], [w_momentum, w_liquidity, w_structure, w_regime] ) # Apply penalties total_penalty = clamp( manip_penalty * 0.4 + noise_penalty * 0.25 + spread_penalty * 0.2 + latency_penalty * 0.15 ) base_conf = clamp(base_conf * (1.0 - total_penalty)) # Technical alignment bonus base_conf = clamp(base_conf + tech_align_bonus * 0.15) # Session timing penalty timing_penalty = session.get("timing_penalty", 0.0) base_conf = clamp(base_conf * (1.0 - timing_penalty * 0.3)) # Timeframe conflict penalty tf_conflict = tf.get("tf_conflict_score", 0.0) base_conf = clamp(base_conf * (1.0 - tf_conflict * 0.25)) # Gate by warm-up base_conf = clamp(base_conf * warm_frac) # ---------------------------------------------------------------- # 6. Internal trust (meta-confidence) # ---------------------------------------------------------------- internal_trust = clamp(weighted_mean( [data_quality, 1.0 - instability, 1.0 - manip_penalty, volatility_conf, 1.0 - tf_conflict], [0.30, 0.20, 0.20, 0.15, 0.15] ) * warm_frac) # ---------------------------------------------------------------- # 7. Execution suitability # ---------------------------------------------------------------- exec_suit = self._classify_exec_suitability(base_conf, di, ms, event) # ---------------------------------------------------------------- # 8. Raw direction [-1,1] # ---------------------------------------------------------------- raw_dir = self._compute_raw_direction(tf, lp, tech, event, ms) return { "confidence" : base_conf, "internal_trust" : internal_trust, "execution_suitability" : exec_suit, "momentum_confidence" : momentum_conf, "liquidity_confidence" : liquidity_conf, "structure_confidence" : structure_conf, "volatility_confidence" : volatility_conf, "manipulation_penalty" : manip_penalty, "noise_penalty" : noise_penalty, "spread_penalty" : spread_penalty, "latency_penalty" : latency_penalty, "regime_confidence" : regime_conf, "continuation_probability": continuation_p, "reversal_probability" : reversal_p, "tech_alignment_bonus" : tech_align_bonus, "raw_direction" : raw_dir, "warm_up_fraction" : warm_frac, } def _classify_exec_suitability(self, confidence: float, di: Dict, ms: Dict, event: Dict) -> str: # Blocked conditions override confidence if di.get("suppress_signal", False): return "blocked" if ms.get("market_state") in ("manipulation", "unstable"): return "blocked" if confidence < CONF_WEAK_MAX else "weak" if event.get("manipulation_flag", False): return "blocked" if confidence < CONF_WEAK_MAX else "weak" if confidence <= CONF_BLOCKED_MAX: return "blocked" elif confidence <= CONF_WEAK_MAX: return "weak" elif confidence <= CONF_MODERATE_MAX: return "moderate" else: return "strong" def _compute_raw_direction(self, tf: Dict, lp: Dict, tech: Dict, event: Dict, ms: Dict) -> float: """Aggregate direction signal from multiple engines [-1, 1].""" signals = [ (tf.get("lower_bias", 0.0), 0.20), (tf.get("mid_bias", 0.0), 0.15), (tf.get("higher_bias", 0.0), 0.15), (lp.get("pressure_chain", 0.5) * 2.0 - 1.0, 0.15), (tech.get("tech_direction", 0.0), 0.20), (event.get("direction_influence", 0.0), 0.10), (ms.get("trend_direction", 0.0), 0.05), ] total_w = sum(w for _, w in signals) if total_w < EPSILON: return 0.0 raw = sum(v * w for v, w in signals) / total_w return clamp(raw, -1.0, 1.0) @staticmethod def _safe_default() -> Dict: return { "confidence": 0.0, "internal_trust": 0.0, "execution_suitability": "blocked", "momentum_confidence": 0.5, "liquidity_confidence": 0.5, "structure_confidence": 0.5, "volatility_confidence": 0.5, "manipulation_penalty": 0.0, "noise_penalty": 0.0, "spread_penalty": 0.0, "latency_penalty": 0.0, "regime_confidence": 0.5, "continuation_probability": 0.5, "reversal_probability": 0.5, "tech_alignment_bonus": 0.0, "raw_direction": 0.0, "warm_up_fraction": 0.0, } # ============================================================================== # PART F2 — SIGNAL LIFECYCLE ENGINE # ============================================================================== class SignalLifecycleEngine: """ Manages signal state machine: idle → forming → candidate → confirmed → cooling → expired / invalidated. TIMING-FIX-2: All hardcoded tick thresholds replaced with timeframe-adaptive properties. Stale-signal lifetime now scales with detected candle interval so 30 s / 1 m / 5 m signals persist ~15 min wall-clock instead of disappearing after a fixed 10 ticks. Implements: - Confidence hysteresis (no flip without strong evidence). - Signal cooldown / spam control. - Signal fatigue, stability, and flip-flop suppression. - Stale signal decay (timeframe-adaptive). - Expiry bucket assignment. Input contract: update(score_result, event_result, di_result, ms_result, tick_count, tf_seconds) → dict Output contract: { "direction": str, # "BUY" | "SELL" "lifecycle_state": str, "signal_freshness": float, "stale_signal_flag":bool, "expiry_bucket": str, "confidence_final": float, # after hysteresis "cooling_flag": bool, "reason_codes": list, } """ # ------------------------------------------------------------------ # Timing constants — replaced hardcoded values with defaults used # only when tf_seconds is unknown (should never happen in normal # operation because MAYTHOS always passes it). # ------------------------------------------------------------------ _DFL_STALE_TICKS = 25 # was 10 → ~25 min @ 1 m _DFL_COOLING_TICKS = 8 # was 5 → ~8 min @ 1 m _DFL_FLIP_SUPPRESS = 5 # was 3 _FORMING_THRESH = 0.10 _CANDIDATE_THRESH = 0.25 _CONFIRMED_THRESH = 0.45 _HYSTERESIS_DELTA = 0.05 # min confidence change to flip direction def __init__(self) -> None: self._state : str = "idle" self._direction : str = "BUY" self._last_dir : str = "BUY" self._confidence : float = 0.0 self._tick_in_state : int = 0 self._tick_total : int = 0 self._cooling_tick : int = 0 self._last_confirmed_tick: int = -999 self._flip_count : int = 0 self._last_flip_tick : int = -999 # TIMING-FIX-2a: hold the detected candle interval (seconds) # so _stale_ticks and _max_cooling can adapt every tick. self._last_tf_seconds : float = 60.0 # Rolling confidence for hysteresis self._conf_buf : RollingBuffer = RollingBuffer(5) # ------------------------------------------------------------------ # TIMING-FIX-2b: Timeframe-adaptive thresholds (computed properties) # ------------------------------------------------------------------ @property def _stale_ticks(self) -> int: """Target ~15 min wall-clock, clamped [12, 40] ticks.""" tf = self._last_tf_seconds if tf <= 0: return self._DFL_STALE_TICKS ticks = int(900.0 / tf) # 15 minutes / candle_interval return max(12, min(40, ticks)) @property def _max_cooling(self) -> int: """Target ~5 min wall-clock, clamped [5, 15] ticks.""" tf = self._last_tf_seconds if tf <= 0: return self._DFL_COOLING_TICKS ticks = int(300.0 / tf) # 5 minutes / candle_interval return max(5, min(15, ticks)) @property def _flip_suppress_ticks(self) -> int: """Minimum ticks between direction flips.""" return self._DFL_FLIP_SUPPRESS def update(self, score_result: Dict, event_result: Dict, di_result: Dict, ms_result: Dict, tick_count: int, tf_seconds: float = 60.0) -> Dict: try: return self._update_inner(score_result, event_result, di_result, ms_result, tick_count, tf_seconds) except Exception as exc: ENGINE_DIAGNOSTICS.record("SignalLifecycleEngine", exc) return self._safe_default() def _update_inner(self, score_result, event_result, di_result, ms_result, tick_count, tf_seconds) -> Dict: # TIMING-FIX-2c: remember the detected timeframe self._last_tf_seconds = tf_seconds if tf_seconds > 0 else 60.0 self._tick_total = tick_count self._tick_in_state += 1 raw_conf = score_result.get("confidence", 0.0) raw_dir = score_result.get("raw_direction", 0.0) exec_suit = score_result.get("execution_suitability", "blocked") suppress = di_result.get("suppress_signal", False) self._conf_buf.push(raw_conf) smoothed_conf = self._conf_buf.mean() reason_codes: List[str] = [] # ---------------------------------------------------------------- # 1. Suppression / blocked override # ---------------------------------------------------------------- if suppress or exec_suit == "blocked": if self._state not in ("idle", "forming"): self._transition("cooling") reason_codes.append("DATA_DEGRADED") return self._emit("BUY", "idle", 0.0, False, "short", False, reason_codes) # ---------------------------------------------------------------- # 2. Direction determination with hysteresis # ---------------------------------------------------------------- new_dir = "BUY" if raw_dir >= 0 else "SELL" # Flip suppression: don't flip direction rapidly without strong evidence if new_dir != self._direction: dir_flip_allowed = ( abs(raw_dir) > 0.35 and (tick_count - self._last_flip_tick) >= self._flip_suppress_ticks and smoothed_conf > self._HYSTERESIS_DELTA ) if not dir_flip_allowed: new_dir = self._direction # hold current direction else: # AUDIT FIX (flip_count bug): _flip_count previously only ever # incremented and was never reset/decayed anywhere in this # class. On a 24/7 deployment, once the engine accumulated 3 # flips at ANY point over its entire lifetime (completely # normal -- markets reverse many times over weeks/months), # _flip_count stayed >=3 permanently, so the "rapid flip-flop" # check below started misfiring on every future flip, however # isolated and well-evidenced, immediately halving confidence # and tagging TF_CONFLICT on signals that were never actually # flip-flopping. Fix: only treat this as a continuation of a # rapid-flip streak if it's close to the previous flip (reuse # the same "<5 ticks" window the check below already uses for # what counts as "rapid"); otherwise this is a fresh, isolated # flip and the streak counter restarts. if (tick_count - self._last_flip_tick) < 5: self._flip_count += 1 else: self._flip_count = 1 self._last_flip_tick = tick_count reason_codes.append("TF_ALIGNED") # Rapid flip-flop suppression if self._flip_count >= 3 and (tick_count - self._last_flip_tick) < 5: smoothed_conf = clamp(smoothed_conf * 0.5) reason_codes.append("TF_CONFLICT") self._direction = new_dir # ---------------------------------------------------------------- # 3. State machine transitions # ---------------------------------------------------------------- prev_state = self._state if self._state == "idle": if smoothed_conf >= self._FORMING_THRESH: self._transition("forming") elif self._state == "forming": if smoothed_conf >= self._CANDIDATE_THRESH: self._transition("candidate") elif smoothed_conf < self._FORMING_THRESH * 0.5: self._transition("idle") elif self._state == "candidate": if smoothed_conf >= self._CONFIRMED_THRESH: self._transition("confirmed") self._last_confirmed_tick = tick_count elif smoothed_conf < self._FORMING_THRESH: self._transition("idle") elif self._state == "confirmed": # Invalidation conditions if self._should_invalidate(event_result, ms_result, di_result): self._transition("invalidated") reason_codes.append("MANIP_HIGH") elif smoothed_conf < self._FORMING_THRESH: self._transition("expired") reason_codes.append("STALE_DECAY") elif self._tick_in_state >= self._stale_ticks: self._transition("cooling") reason_codes.append("STALE_DECAY") elif self._state == "cooling": self._cooling_tick += 1 if self._cooling_tick >= self._max_cooling: self._transition("idle") self._cooling_tick = 0 elif smoothed_conf >= self._CONFIRMED_THRESH and self._cooling_tick >= 1: self._transition("candidate") self._cooling_tick = 0 elif self._state in ("expired", "invalidated"): self._transition("idle") # ---------------------------------------------------------------- # 4. Signal freshness (TIMING-FIX-2d: gentler quadratic decay) # ---------------------------------------------------------------- stale_ticks = self._stale_ticks if self._state == "confirmed": progress = safe_div(self._tick_in_state, stale_ticks) # Quadratic decay: stays fresher longer early on, # drops faster only near the end. freshness = clamp(1.0 - progress * progress) elif self._state == "candidate": freshness = 0.5 elif self._state == "forming": freshness = 0.25 else: freshness = 0.0 stale = self._state == "confirmed" and self._tick_in_state >= stale_ticks - 2 # ---------------------------------------------------------------- # 5. Expiry bucket classification # ---------------------------------------------------------------- expiry = self._classify_expiry(smoothed_conf, tf_align=score_result.get("tech_alignment_bonus", 0.0)) # ---------------------------------------------------------------- # 6. Reason codes # ---------------------------------------------------------------- if di_result.get("data_quality", 1.0) > 0.8: reason_codes.append("DATA_OK") else: reason_codes.append("DATA_DEGRADED") if score_result.get("raw_direction", 0.0) > 0 and self._direction == "BUY": reason_codes.append("MOMENTUM_OK") # Deduplicate reason codes reason_codes = list(dict.fromkeys(reason_codes)) cooling_flag = self._state == "cooling" confidence_final = clamp(smoothed_conf if self._state == "confirmed" else smoothed_conf * 0.7) return self._emit( self._direction, self._state, freshness, stale, expiry, cooling_flag, reason_codes, confidence_final=confidence_final ) def _should_invalidate(self, event_result: Dict, ms_result: Dict, di_result: Dict) -> bool: if event_result.get("manipulation_flag"): return True if ms_result.get("market_state") in ("manipulation", "blocked"): return True if di_result.get("data_quality", 1.0) < 0.2: return True return False def _transition(self, new_state: str) -> None: if new_state != self._state: self._state = new_state self._tick_in_state = 0 def _classify_expiry(self, confidence: float, tf_align: float) -> str: if confidence > 0.65 and tf_align > 0.1: return "medium" elif confidence > 0.35: return "short" return "ultra_short" def _emit(self, direction: str, state: str, freshness: float, stale: bool, expiry: str, cooling: bool, reason_codes: List[str], confidence_final: float = 0.0) -> Dict: return { "direction" : direction, "lifecycle_state" : state, "signal_freshness" : freshness, "stale_signal_flag": stale, "expiry_bucket" : expiry, "confidence_final" : confidence_final, "cooling_flag" : cooling, "reason_codes" : reason_codes, } @staticmethod def _safe_default() -> Dict: return { "direction": "BUY", "lifecycle_state": "idle", "signal_freshness": 0.0, "stale_signal_flag": False, "expiry_bucket": "ultra_short", "confidence_final": 0.0, "cooling_flag": False, "reason_codes": ["DATA_DEGRADED"], } # ============================================================================== # PART G1 — WARM-UP CONTROLLER (Layer 8) # ============================================================================== class WarmUpController: """ Tracks candle count and enforces cold-start gating. Exposes operational_mode, degraded_mode_flag, blocked_flag. Warm-up stages: 0 → MIN_WARM_CANDLES-1 : cold_start (all signals blocked) MIN_WARM → FULL_WARM-1 : cautious (reduced confidence) FULL_WARM+ : normal (full operation) Also handles: - Data quality degradation → degraded mode - High manipulation → cautious mode - Blocked conditions → blocked mode """ _MANIP_DEGRADED_THRESHOLD = 0.6 _QUALITY_DEGRADED_THRESHOLD = 0.5 _RECOVERY_TICKS = 10 def __init__(self) -> None: self._tick_count : int = 0 self._op_mode : str = "cold_start" self._degraded_ticks: int = 0 self._blocked_ticks : int = 0 self._recovery_tick : int = 0 def tick(self, di_result: Dict, score_result: Optional[Dict] = None) -> Dict: """ Called ONCE per candle. Advances tick_count by exactly 1. FIX-A: Never call this twice per candle — use refine() for post-score update. """ try: return self._tick_inner(di_result, score_result) except Exception as exc: ENGINE_DIAGNOSTICS.record("WarmUpController", exc) return self._safe_default() def refine(self, di_result: Dict, score_result: Dict) -> Dict: """ FIX-4: Post-score mode refinement WITHOUT advancing tick_count. Previous implementation called _tick_inner() (which always increments _tick_count by 1), then manually restored the counter. That pattern risks off-by-one errors and computes mode/warm_frac with a phantom N+1 tick. This rewrite reads _tick_count directly (already correct from tick()) and re-classifies operational_mode based on the post-score signals, with zero risk of double-incrementing. """ try: tc = self._tick_count # set correctly by tick() earlier this candle manip_proxy = score_result.get("manipulation_penalty", 0.0) if score_result else 0.0 data_quality = di_result.get("data_quality", 1.0) suppress = di_result.get("suppress_signal", False) # Re-evaluate mode now that we have score feedback (only past cold-start) if tc >= MIN_WARM_CANDLES: if suppress: self._blocked_ticks += 1 # AUDIT FIX (recovery-streak bug): _recovery_tick counts # CONSECUTIVE clean ticks toward exiting degraded/blocked # mode, but was never reset when a new bad tick interrupted # an in-progress streak -- so e.g. a "bad, clean, clean" # pattern repeating would let _recovery_tick climb past # _RECOVERY_TICKS purely from non-consecutive clean ticks, # falsely declaring full recovery moments after a fresh # bad tick. Any bad tick must reset the streak. self._recovery_tick = 0 self._op_mode = "blocked" elif data_quality < self._QUALITY_DEGRADED_THRESHOLD: self._degraded_ticks += 1 self._recovery_tick = 0 # AUDIT FIX: see above self._op_mode = "degraded" elif manip_proxy > self._MANIP_DEGRADED_THRESHOLD: # High manipulation: downgrade normal/recovery to cautious self._recovery_tick = 0 # AUDIT FIX: see above if self._op_mode in ("normal", "recovery"): self._op_mode = "cautious" warm_frac = clamp(safe_div(tc - MIN_WARM_CANDLES, FULL_WARM_CANDLES - MIN_WARM_CANDLES)) blocked = tc < MIN_WARM_CANDLES or self._op_mode == "blocked" degraded = self._op_mode == "degraded" return { "operational_mode" : self._op_mode, "tick_count" : tc, "warm_up_fraction" : warm_frac, "blocked_flag" : blocked, "degraded_mode_flag" : degraded, "cold_start_flag" : tc < MIN_WARM_CANDLES, "degraded_tick_count": self._degraded_ticks, "blocked_tick_count" : self._blocked_ticks, } except Exception as exc: ENGINE_DIAGNOSTICS.record("WarmUpController", exc) return self._safe_default() def _tick_inner(self, di_result: Dict, score_result: Optional[Dict]) -> Dict: self._tick_count += 1 tc = self._tick_count data_quality = di_result.get("data_quality", 1.0) manip_proxy = 0.0 if score_result: manip_proxy = score_result.get("manipulation_penalty", 0.0) # ---- Determine operational mode ---- if tc < MIN_WARM_CANDLES: mode = "cold_start" blocked = True degraded = False elif tc < FULL_WARM_CANDLES: mode = "cautious" blocked = False degraded = False else: # Full operation but may degrade # AUDIT FIX (double-counting bug): _blocked_ticks/_degraded_ticks # are intentionally NOT incremented here anymore. refine() is # always called immediately after tick() for every non-cold-start # candle (see MAYTHOS._tick_inner) using this exact same # di_result, and re-evaluates these identical suppress/ # data_quality conditions -- so incrementing in both places # counted every blocked/degraded candle twice. refine() is also # strictly more complete (it additionally has manip_proxy, which # is never available yet here), so it is now the sole place these # persistent counters are mutated; this branch still computes the # local mode/blocked/degraded for tick()'s own immediate return # value. if di_result.get("suppress_signal", False): mode = "blocked" blocked = True degraded = False elif data_quality < self._QUALITY_DEGRADED_THRESHOLD: mode = "degraded" blocked = False degraded = True elif manip_proxy > self._MANIP_DEGRADED_THRESHOLD: mode = "cautious" blocked = False degraded = True else: # Recovery check: exiting degraded state if self._degraded_ticks > 0 or self._blocked_ticks > 0: self._recovery_tick += 1 if self._recovery_tick >= self._RECOVERY_TICKS: self._degraded_ticks = 0 self._blocked_ticks = 0 self._recovery_tick = 0 mode = "normal" blocked = False degraded = False else: mode = "recovery" blocked = False degraded = False else: mode = "normal" blocked = False degraded = False self._op_mode = mode # Warm-up fraction: 0 during cold, ramps to 1.0 at FULL_WARM warm_frac = clamp(safe_div(tc - MIN_WARM_CANDLES, FULL_WARM_CANDLES - MIN_WARM_CANDLES)) return { "operational_mode" : mode, "tick_count" : tc, "warm_up_fraction" : warm_frac, "blocked_flag" : blocked, "degraded_mode_flag" : degraded, "cold_start_flag" : tc < MIN_WARM_CANDLES, "degraded_tick_count": self._degraded_ticks, "blocked_tick_count" : self._blocked_ticks, } @property def tick_count(self) -> int: return self._tick_count @property def is_warm(self) -> bool: return self._tick_count >= FULL_WARM_CANDLES @staticmethod def _safe_default() -> Dict: return { "operational_mode": "cold_start", "tick_count": 0, "warm_up_fraction": 0.0, "blocked_flag": True, "degraded_mode_flag": False, "cold_start_flag": True, "degraded_tick_count": 0, "blocked_tick_count": 0, } # ============================================================================== # PART G2 — DEBUG TRACE ENGINE # ============================================================================== class DebugTraceEngine: """ Assembles a full debug trace dict from all engine outputs. Only populated when debug_mode=True. Otherwise returns None quickly. Never raises. Never affects signal output. """ @staticmethod def build( candle : Any, di_result : Dict, ms_result : Dict, asset_result : Dict, session_result : Dict, tf_result : Dict, event_result : Dict, lp_result : Dict, tech_result : Dict, score_result : Dict, lifecycle_result: Dict, warmup_result : Dict, debug_mode : bool = False, ) -> Optional[Dict]: if not debug_mode: return None try: return { "candle": { "ts": getattr(candle, "timestamp", 0.0), "o": getattr(candle, "open", 0.0), "h": getattr(candle, "high", 0.0), "l": getattr(candle, "low", 0.0), "c": getattr(candle, "close", 0.0), "v": getattr(candle, "volume", 0.0), }, "data_integrity" : di_result, "market_state" : ms_result, "asset_profile" : asset_result, "session" : session_result, "timeframe_fusion" : tf_result, "events" : event_result, "liquidity_pressure": lp_result, "technical_stack" : tech_result, "score" : score_result, "lifecycle" : lifecycle_result, "warmup" : warmup_result, } except Exception as exc: ENGINE_DIAGNOSTICS.record("DebugTraceEngine", exc) return {"error": "debug_trace_failed"} # ============================================================================== # PART G3 — OUTPUT FORMATTER (Layer 10) # ============================================================================== class OutputFormatter: """ Assembles the final MAYTHOS output dict from all engine results. Output contract (Section 5, expanded): Required top-level fields: direction, confidence, internal_trust, execution_suitability, market_state, regime_label, asset_mode, timeframe_alignment, manipulation_probability, liquidity_score, volatility_score, pressure_score, momentum_score, timing_score, signal_freshness, spread_health, data_quality, readability_score, stale_signal_flag, degraded_mode_flag, blocked_flag, reason_summary, reason_codes, call_bias, put_bias, expiry_suitability_ultra_short, expiry_suitability_short, expiry_suitability_medium, debug_trace, spec_version, engine_version, output_version Never raises. Returns safe default on any failure. """ @staticmethod def format( di_result : Dict, ms_result : Dict, asset_result : Dict, session_result : Dict, tf_result : Dict, event_result : Dict, lp_result : Dict, tech_result : Dict, score_result : Dict, lifecycle_result: Dict, warmup_result : Dict, debug_trace : Optional[Dict] = None, ) -> Dict: try: return OutputFormatter._format_inner( di_result, ms_result, asset_result, session_result, tf_result, event_result, lp_result, tech_result, score_result, lifecycle_result, warmup_result, debug_trace) except Exception as exc: ENGINE_DIAGNOSTICS.record("OutputFormatter", exc) return _default_output() @staticmethod def _format_inner(di, ms, asset, session, tf, event, lp, tech, score, lc, warmup, debug_trace) -> Dict: # FIX 1: blocked_flag must be consistent with execution_suitability. # During cautious warm-up, warmup.blocked_flag=False but low warm_up_fraction # can still yield exec_suit='blocked'. Unify both here for contract integrity. _warmup_blocked = warmup.get("blocked_flag", True) _exec_blocked = score.get("execution_suitability", "blocked") == "blocked" blocked = _warmup_blocked or _exec_blocked degraded = warmup.get("degraded_mode_flag", False) conf = score.get("confidence", 0.0) # If blocked, return safe output with reason if blocked: out = _default_output() out["blocked_flag"] = True out["degraded_mode_flag"] = degraded out["data_quality"] = di.get("data_quality", 1.0) out["reason_summary"] = OutputFormatter._reason_summary( warmup, di, score, lc, blocked=True) out["reason_codes"] = lc.get("reason_codes", ["DATA_OK"]) out["debug_trace"] = debug_trace out["warm_up_fraction"] = warmup.get("warm_up_fraction", 0.0) out["operational_mode"] = warmup.get("operational_mode", "cold_start") out["spec_version"] = SPEC_VERSION out["engine_version"] = ENGINE_VERSION out["output_version"] = OUTPUT_VERSION return out # ---- Core signals ---- direction = lc.get("direction", "BUY") # AUDIT FIX: was a bare `assert`, which Python silently strips # entirely when run with -O / PYTHONOPTIMIZE=1. Contract-enforcing # checks on data flowing through business logic should never depend # on an interpreter flag; this is also more graceful than a stripped # assert (which would let an invalid value leak straight through), # since it self-corrects instead of relying on an exception path. if direction not in SIGNAL_DIRECTIONS: direction = "BUY" # ---- Derived scores ---- momentum_score = clamp( tech.get("macd_result", {}).get("momentum_quality", 0.5) * 0.5 + abs(tf.get("lower_bias", 0.0)) * 0.5 ) volatility_score = clamp( {"low": 0.3, "medium": 0.6, "high": 0.8, "extreme": 1.0}.get( asset.get("volatility_profile", "medium"), 0.6) ) readability_score = clamp( ms.get("readability_hint", 0.5) * 0.4 + (1.0 - di.get("instability_level", 0.0)) * 0.3 + tf.get("tf_alignment", 0.5) * 0.3 ) timing_score = session.get("session_score", 0.5) spread_health = clamp( 1.0 - float(di.get("spread_anomaly", False)) * 0.5 - float(di.get("spread_expansion", False)) * 0.3 ) # ---- Call / Put bias ---- # BUY → higher call_bias; SELL → higher put_bias raw_dir = score.get("raw_direction", 0.0) call_bias = clamp(0.5 + raw_dir * 0.5) put_bias = clamp(1.0 - call_bias) # ---- Expiry suitability scores ---- base_conf = conf expiry_us = clamp(base_conf * 0.6) expiry_sh = clamp(base_conf * 0.8 * (1.0 + timing_score * 0.2)) expiry_me = clamp(base_conf * (0.9 + tf.get("tf_alignment", 0.5) * 0.1)) # ---- Reason summary ---- reason_summary = OutputFormatter._reason_summary(warmup, di, score, lc, blocked=False) # ---- Market recognition (from ms_result) ---- market_state = ms.get("market_state", "undefined") if market_state not in MARKET_STATES: market_state = "undefined" regime_label = ms.get("regime_label", "transition") if regime_label not in REGIME_LABELS: regime_label = "transition" asset_mode = asset.get("asset_mode", "unknown") if asset_mode not in ASSET_MODES: asset_mode = "unknown" return { # Core "direction" : direction, "confidence" : conf, "internal_trust" : score.get("internal_trust", 0.0), "execution_suitability" : score.get("execution_suitability", "blocked"), "market_state" : market_state, "regime_label" : regime_label, "asset_mode" : asset_mode, "timeframe_alignment" : tf.get("tf_alignment", 0.5), "manipulation_probability": lp.get("manipulation_proxy", 0.0), "liquidity_score" : lp.get("liquidity_score", 0.5), "volatility_score" : volatility_score, "pressure_score" : lp.get("pressure_score", 0.5), "momentum_score" : momentum_score, "timing_score" : timing_score, "signal_freshness" : lc.get("signal_freshness", 0.0), "spread_health" : spread_health, "data_quality" : di.get("data_quality", 1.0), "readability_score" : readability_score, "stale_signal_flag" : lc.get("stale_signal_flag", False), "degraded_mode_flag" : degraded, "blocked_flag" : blocked, "reason_summary" : reason_summary, "reason_codes" : lc.get("reason_codes", ["DATA_OK"]), # Optional "call_bias" : call_bias, "put_bias" : put_bias, "expiry_suitability_ultra_short" : expiry_us, "expiry_suitability_short" : expiry_sh, "expiry_suitability_medium" : expiry_me, "debug_trace" : debug_trace, # Warm-up (FIX-B: was missing from output) "warm_up_fraction" : warmup.get("warm_up_fraction", 0.0), "operational_mode" : warmup.get("operational_mode", "cold_start"), # Version "spec_version" : SPEC_VERSION, "engine_version" : ENGINE_VERSION, "output_version" : OUTPUT_VERSION, } @staticmethod def _reason_summary(warmup: Dict, di: Dict, score: Dict, lc: Dict, blocked: bool) -> str: parts: List[str] = [] if warmup.get("cold_start_flag"): parts.append("cold_start") if blocked: parts.append("blocked") if warmup.get("degraded_mode_flag"): parts.append("degraded_mode") if di.get("suppress_signal"): parts.append("data_suppressed") if di.get("spread_anomaly"): parts.append("spread_anomaly") if di.get("gap_detected"): parts.append("gap_detected") if score.get("manipulation_penalty", 0.0) > 0.4: parts.append("manipulation_risk") if lc.get("stale_signal_flag"): parts.append("stale_signal") if lc.get("cooling_flag"): parts.append("signal_cooling") exec_suit = score.get("execution_suitability", "blocked") state = lc.get("lifecycle_state", "idle") parts.append(f"exec:{exec_suit}") parts.append(f"state:{state}") return " | ".join(parts) if parts else "ok" @staticmethod def validate_output(output: Dict) -> List[str]: """ Validate that all required fields are present and within bounds. Returns list of validation errors (empty = OK). """ errors: List[str] = [] required_floats = [ "confidence", "internal_trust", "timeframe_alignment", "manipulation_probability", "liquidity_score", "volatility_score", "pressure_score", "momentum_score", "timing_score", "signal_freshness", "spread_health", "data_quality", "readability_score", "call_bias", "put_bias", "expiry_suitability_ultra_short", "expiry_suitability_short", "expiry_suitability_medium", ] required_strs = [ "direction", "execution_suitability", "market_state", "regime_label", "asset_mode", "reason_summary", ] required_bools = [ "stale_signal_flag", "degraded_mode_flag", "blocked_flag", ] required_lists = ["reason_codes"] for f in required_floats: if f not in output: errors.append(f"missing:{f}") elif not isinstance(output[f], (int, float)): errors.append(f"not_float:{f}") elif not math.isfinite(output[f]): errors.append(f"not_finite:{f}") elif not (0.0 <= output[f] <= 1.0): errors.append(f"out_of_range:{f}={output[f]}") for f in required_strs: if f not in output: errors.append(f"missing:{f}") elif not isinstance(output[f], str): errors.append(f"not_str:{f}") for f in required_bools: if f not in output: errors.append(f"missing:{f}") elif not isinstance(output[f], bool): errors.append(f"not_bool:{f}") for f in required_lists: if f not in output: errors.append(f"missing:{f}") elif not isinstance(output[f], list): errors.append(f"not_list:{f}") # Enum checks if output.get("direction") not in SIGNAL_DIRECTIONS: errors.append(f"invalid_direction:{output.get('direction')}") if output.get("execution_suitability") not in EXECUTION_SUITABILITY: errors.append(f"invalid_exec_suit:{output.get('execution_suitability')}") if output.get("market_state") not in MARKET_STATES: errors.append(f"invalid_market_state:{output.get('market_state')}") return errors # ============================================================================== # MAYTHOS ORCHESTRATOR — Top-level API # ============================================================================== class MAYTHOS: """ MAYTHOS v1.1 — Single-entry-point orchestrator. Instantiate once; call tick() on every new candle. Thread-safety: NOT thread-safe. Use one instance per symbol/feed. Parameters ---------- debug_mode : bool When True, populates debug_trace in every output dict and accumulates an internal bounded log (access via .debug_log). Max 1000 entries (FIX-9: prevents unbounded memory growth in long debug sessions). Example ------- >>> engine = MAYTHOS() >>> output = engine.tick(candle) >>> print(output["direction"], output["confidence"], output["execution_suitability"]) """ _MAX_DEBUG_LOG: int = 1000 # FIX-9: cap internal debug trace log def __init__(self, debug_mode: bool = False) -> None: self._debug_mode = debug_mode self._data_integrity = DataIntegrityEngine() self._market_state = MarketStateEngine() self._asset_profile = AssetProfileEngine() self._session = SessionIntelligenceEngine() self._timeframe_fusion = MultiTimeframeFusionEngine() self._event_detection = EventDetectionEngine() self._liquidity_pressure = LiquidityPressureEngine() self._tech_stack = TechnicalConfirmationStack() self._scorer = AdaptiveScoringEngine() self._warmup = WarmUpController() self._lifecycle = SignalLifecycleEngine() self._formatter = OutputFormatter() self._debug_engine = DebugTraceEngine() # FIX-9: bounded in-memory debug log (only populated when debug_mode=True) self._debug_log: List[Dict] = [] def tick(self, candle : "Candle", receive_time: Optional[float] = None, debug : Optional[bool] = None) -> Dict: """ Process one Candle. Returns the full MAYTHOS output dict. Parameters ---------- candle : Candle object. receive_time : Optional wall-clock receive timestamp (float, UTC epoch). debug : Override instance debug_mode for this tick only. """ _debug = self._debug_mode if debug is None else debug try: return self._tick_inner(candle, receive_time, _debug) except Exception as exc: ENGINE_DIAGNOSTICS.record("MAYTHOS", exc) out = _default_output() if _debug: out["debug_trace"] = {"exception": str(exc)} return out def _tick_inner(self, candle: "Candle", receive_time: Optional[float], debug: bool) -> Dict: # Layer 1 di = self._data_integrity.process(candle, receive_time) # Layer 8 — pre-score warm-up gate warmup = self._warmup.tick(di) if warmup["cold_start_flag"]: out = _default_output() out["data_quality"] = di.get("data_quality", 1.0) out["reason_summary"] = "cold_start: insufficient history" out["warm_up_fraction"] = warmup.get("warm_up_fraction", 0.0) out["operational_mode"] = warmup.get("operational_mode", "cold_start") if debug: out["debug_trace"] = {"warmup": warmup, "di": di} return out # Layer 2 ms = self._market_state.update(candle, di) asset = self._asset_profile.update(candle, di) sess = self._session.update(candle) # Layer 3 tf = self._timeframe_fusion.update(candle, ms) # Layer 4 ev = self._event_detection.update(candle, di, ms) # Layer 5 lp = self._liquidity_pressure.update(candle, di, ev) # Layer 6 tech = self._tech_stack.update(candle, asset, sess) # Layer 7 score = self._scorer.score( di, ms, asset, sess, tf, ev, lp, tech, tick_count=self._warmup.tick_count, ) # Layer 8 — post-score warm-up refinement (FIX-A: use refine(), not tick()) # refine() updates mode from manipulation_penalty WITHOUT advancing tick_count warmup = self._warmup.refine(di, score_result=score) # Layer 9 # TIMING-FIX-3: pass detected candle interval so lifecycle # stale-signal and cooling windows scale with chart timeframe. lc = self._lifecycle.update( score, ev, di, ms, tick_count=self._warmup.tick_count, tf_seconds=self._timeframe_fusion.detected_tf_seconds, ) # Layer 10 dtrace = self._debug_engine.build( candle, di, ms, asset, sess, tf, ev, lp, tech, score, lc, warmup, debug_mode=debug, ) # FIX-9: accumulate debug traces up to _MAX_DEBUG_LOG entries (prevents # unbounded list growth during long debug sessions) if debug and dtrace is not None: self._debug_log.append(dtrace) if len(self._debug_log) > self._MAX_DEBUG_LOG: del self._debug_log[:-self._MAX_DEBUG_LOG] output = self._formatter.format( di, ms, asset, sess, tf, ev, lp, tech, score, lc, warmup, debug_trace=dtrace, ) return output def validate(self, output: Dict) -> List[str]: """Validate output dict. Returns list of errors (empty = OK).""" return OutputFormatter.validate_output(output) def reset(self) -> None: """Hard-reset all state (for backtesting resets).""" self.__init__(debug_mode=self._debug_mode) @property def tick_count(self) -> int: return self._warmup.tick_count @property def is_warm(self) -> bool: return self._warmup.is_warm @property def debug_log(self) -> List[Dict]: """FIX-9: Read-only snapshot of the bounded debug trace log (last ≤1000 entries).""" return list(self._debug_log) @staticmethod def engine_health() -> Dict: """ AUDIT FIX: read-only snapshot of internal engine error counts/recent errors (see ENGINE_DIAGNOSTICS above). Every engine's main entry point is fault-tolerant by design and will never raise out of tick() -- this exists purely so an operator can detect when those fallback paths are actually firing (e.g. wire it into a health-check endpoint or a periodic log line in your HF Space). Process-wide across all MAYTHOS instances in this process; call ENGINE_DIAGNOSTICS.reset() to clear it (e.g. after acknowledging an alert). """ return ENGINE_DIAGNOSTICS.snapshot() # ============================================================================== # ENTRY POINT — smoke test # ============================================================================== if __name__ == "__main__": import sys print(f"MAYTHOS v{ENGINE_VERSION} — Single-file self-test") engine = MAYTHOS(debug_mode=False) import datetime base_ts = datetime.datetime(2024, 3, 12, 10, 30, 0).timestamp() errors_seen = [] for i in range(150): close = 1.0000 + i * 0.0005 candle = Candle( timestamp = base_ts + i * 60.0, open = close - 0.0002, high = close + 0.0010, low = close - 0.0008, close = close, volume = 500.0 + i * 2.0, spread = 0.0002, source_id = "test_feed", ) output = engine.tick(candle) errs = engine.validate(output) if errs: errors_seen.extend(errs) print(f" tick {i:3d}: VALIDATION ERRORS: {errs}") if errors_seen: print(f"\n[FAIL] {len(errors_seen)} validation error(s).") sys.exit(1) print(f" tick count : {engine.tick_count}") print(f" is_warm : {engine.is_warm}") print(f" direction : {output['direction']}") print(f" confidence : {output['confidence']:.4f}") print(f" exec_suit : {output['execution_suitability']}") print(f" market_state : {output['market_state']}") print(f" regime : {output['regime_label']}") print(f" asset_mode : {output['asset_mode']}") print(f" reason : {output['reason_summary']}") print(f"\n[PASS] All 150 ticks produced valid output. MAYTHOS v{ENGINE_VERSION} operational.")