diff --git "a/models/maythos/engine.py" "b/models/maythos/engine.py" deleted file mode 100644--- "a/models/maythos/engine.py" +++ /dev/null @@ -1,4450 +0,0 @@ -#!/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) -SIGNAL_COOLDOWN_TICKS = 5 # TIMING-FIX-5: was 3 → 5 ticks minimum between same-direction signals - -# Indicator periods are imported from the project-wide config so the batch -# indicator pipeline (indicators/) and these streaming technical engines share -# a single source of truth. See MERGE_NOTES.md (Decision 4). -from config import ( - RSI_PERIOD, - MACD_FAST, - MACD_SLOW, - MACD_SIGNAL, - BB_PERIOD, - BB_STD_DEV, -) -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 - - -# ============================================================================== -# 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. - 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() -