Spaces:
Running
Running
#!/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; Yahoo Finance via yfinance + 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)
Yahoo Finance live mode:
feed = YahooFinanceFeed("AAPL", interval="1m", period="1d")
output = engine.tick_yahoo(feed)
"""
# ==============================================================================
# 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
# Yahoo Finance is the only live-market data source used by this file. Keep
# the import optional so the analysis engine can still be imported for offline
# backtests; YahooFinanceFeed raises a clear install error when live mode is
# requested without yfinance.
try:
import yfinance as yf
except ImportError: # pragma: no cover - depends on the host environment
yf = None # type: ignore[assignment]
# ==============================================================================
# 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 = 100 # full confidence unlock
MAX_CACHE_ENTRIES = 64 # bounded memory-behavior caches
AGE_DECAY_TICKS = 50 # ticks before cache entry ages out
SIGNAL_COOLDOWN_TICKS = 3 # minimum ticks between same-direction signals
# RSI periods and defaults
RSI_PERIOD = 14
MACD_FAST = 12
MACD_SLOW = 26
MACD_SIGNAL = 9
BB_PERIOD = 20
BB_STD_DEV = 2.0
EMA200_PERIOD = 200
RSI_OB_CRYPTO = 80.0
RSI_OS_CRYPTO = 20.0
RSI_OB_DEFAULT = 70.0
RSI_OS_DEFAULT = 30.0
# Confidence bands (Section 17)
CONF_BLOCKED_MAX = 0.24
CONF_WEAK_MAX = 0.44
CONF_MODERATE_MAX = 0.69
# above 0.69 → strong
# Sudden candle expansion multiplier
SUDDEN_EXPANSION_MULT = 3.0
# ==============================================================================
# NUMERIC UTILITY FUNCTIONS
# ==============================================================================
def clamp(value: float, lo: float = 0.0, hi: float = 1.0) -> float:
"""Clamp value to [lo, hi]. Handles NaN → lo."""
if not math.isfinite(value):
return lo
return max(lo, min(hi, value))
def safe_div(numerator: float, denominator: float, fallback: float = 0.0) -> float:
"""Division protected against zero/NaN/inf. Returns fallback on error."""
if not math.isfinite(numerator) or not math.isfinite(denominator):
return fallback
if abs(denominator) < EPSILON:
return fallback
result = numerator / denominator
return result if math.isfinite(result) else fallback
def safe_sqrt(value: float, fallback: float = 0.0) -> float:
"""Square root protected against negative and non-finite inputs."""
if not math.isfinite(value) or value < 0.0:
return fallback
return math.sqrt(value)
def is_finite_positive(value: float) -> bool:
return math.isfinite(value) and value > 0.0
def normalize_01(value: float, min_val: float, max_val: float,
fallback: float = 0.5) -> float:
"""Normalize value to [0, 1] given range. Clamped. Returns fallback if range is zero."""
rng = max_val - min_val
if abs(rng) < EPSILON:
return fallback
return clamp(safe_div(value - min_val, rng))
def weighted_mean(values: List[float], weights: List[float]) -> float:
"""Weighted average. Returns 0.5 if all weights are zero."""
total_w = sum(weights)
if total_w < EPSILON:
return 0.5
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
# ==============================================================================
# ROLLING FIXED-SIZE BUFFERS
# ==============================================================================
class RollingBuffer:
"""
Fixed-size FIFO ring buffer backed by a deque.
Provides O(1) append/access and numpy array views.
Memory bounded: maxlen enforced at construction.
"""
__slots__ = ("_buf", "maxlen")
def __init__(self, maxlen: int) -> None:
assert maxlen > 0, "maxlen must be positive"
self.maxlen: int = maxlen
self._buf: Deque[float] = deque(maxlen=maxlen)
def push(self, value: float) -> None:
v = nan_safe(value)
self._buf.append(v)
def as_array(self) -> np.ndarray:
return np.array(self._buf, dtype=np.float64)
def last(self, fallback: float = 0.0) -> float:
return self._buf[-1] if self._buf else fallback
def __len__(self) -> int:
return len(self._buf)
@property
def full(self) -> bool:
return len(self._buf) == self.maxlen
def is_ready(self, min_len: int) -> bool:
return len(self._buf) >= min_len
def mean(self) -> float:
if not self._buf:
return 0.0
return float(np.mean(self.as_array()))
def std(self) -> float:
if len(self._buf) < 2:
return 0.0
return float(np.std(self.as_array()))
def min(self) -> float:
if not self._buf:
return 0.0
return float(np.min(self.as_array()))
def max(self) -> float:
if not self._buf:
return 0.0
return float(np.max(self.as_array()))
class RollingObjectBuffer:
"""Fixed-size FIFO ring buffer for arbitrary Python objects (candles, events)."""
__slots__ = ("_buf", "maxlen")
def __init__(self, maxlen: int) -> None:
assert maxlen > 0
self.maxlen = maxlen
self._buf: Deque[Any] = deque(maxlen=maxlen)
def push(self, obj: Any) -> None:
self._buf.append(obj)
def last(self, fallback: Any = None) -> Any:
return self._buf[-1] if self._buf else fallback
def to_list(self) -> List[Any]:
return list(self._buf)
def __len__(self) -> int:
return len(self._buf)
def is_ready(self, min_len: int) -> bool:
return len(self._buf) >= min_len
class BoundedCache:
"""
Bounded key-value cache with age-based eviction.
Maximum MAX_CACHE_ENTRIES entries.
"""
def __init__(self, maxlen: int = MAX_CACHE_ENTRIES) -> None:
self.maxlen = maxlen
self._data: Dict[str, Tuple[Any, int]] = {} # key → (value, insert_tick)
self._tick: int = 0
def tick(self) -> None:
self._tick += 1
def set(self, key: str, value: Any) -> None:
if len(self._data) >= self.maxlen and key not in self._data:
self._evict_oldest()
self._data[key] = (value, self._tick)
def get(self, key: str, fallback: Any = None) -> Any:
entry = self._data.get(key)
return entry[0] if entry is not None else fallback
def evict_stale(self, max_age: int = AGE_DECAY_TICKS) -> None:
stale_keys = [k for k, (_, t) in self._data.items()
if self._tick - t > max_age]
for k in stale_keys:
del self._data[k]
def _evict_oldest(self) -> None:
if not self._data:
return
oldest_key = min(self._data, key=lambda k: self._data[k][1])
del self._data[oldest_key]
def __contains__(self, key: str) -> bool:
return key in self._data
def __len__(self) -> int:
return len(self._data)
# ==============================================================================
# CANDLE DATACLASS — minimal validated representation
# ==============================================================================
@dataclass
class Candle:
"""
Validated OHLCV candle. Optional fields default to safe sentinels.
Immutable once constructed.
"""
timestamp : float
open : float
high : float
low : float
close : float
volume : float = 0.0 # 0 = absent, triggers proxy logic
spread : float = 0.0
bid : float = 0.0
ask : float = 0.0
source_id : str = "default"
session_label: str = "unknown"
is_closed : bool = True # False = intrabar/live tick
def __post_init__(self) -> None:
# Clamp float fields against non-finite
for attr in ("open", "high", "low", "close", "volume", "spread", "bid", "ask"):
v = getattr(self, attr)
if not math.isfinite(v):
object.__setattr__(self, attr, 0.0)
# Ensure OHLC consistency after sanitization; if broken, mark invalid
object.__setattr__(self, "_valid", _validate_candle_fields(
self.open, self.high, self.low, self.close))
@property
def valid(self) -> bool:
return self._valid # type: ignore[attr-defined]
@property
def body_size(self) -> float:
return abs(self.close - self.open)
@property
def candle_range(self) -> float:
return self.high - self.low
@property
def upper_wick(self) -> float:
return self.high - max(self.open, self.close)
@property
def lower_wick(self) -> float:
return min(self.open, self.close) - self.low
@property
def is_bullish(self) -> bool:
return self.close >= self.open
@property
def body_ratio(self) -> float:
r = self.candle_range
return safe_div(self.body_size, r) if r > EPSILON else 0.5
@property
def close_position(self) -> float:
"""Where the close sits within the full range [0=bottom, 1=top]."""
r = self.candle_range
return safe_div(self.close - self.low, r) if r > EPSILON else 0.5
# ==============================================================================
# YAHOO FINANCE LIVE DATA SOURCE
# ==============================================================================
class YahooFinanceFeed:
"""Fetch the latest OHLCV candle and quote from Yahoo Finance via yfinance.
This is the only live-price adapter in MAYTHOS. No broker websocket or
alternate market-data provider is used.
Parameters
----------
symbol : str
Yahoo Finance ticker, for example ``AAPL``, ``BTC-USD``, or
``EURUSD=X``.
interval : str
Yahoo Finance interval. ``1m`` is the default for a live candle.
period : str
History window used to obtain the latest candle. Yahoo Finance
limits very small intervals; ``1d`` is suitable for ``1m`` data.
prepost : bool
Include pre-market and after-hours data where Yahoo provides it.
"""
def __init__(self, symbol: str, interval: str = "1m",
period: str = "1d", prepost: bool = False) -> None:
symbol = str(symbol).strip()
if not symbol:
raise ValueError("Yahoo Finance symbol must not be empty")
if not interval:
raise ValueError("Yahoo Finance interval must not be empty")
if not period:
raise ValueError("Yahoo Finance period must not be empty")
self.symbol = symbol
self.interval = interval
self.period = period
self.prepost = bool(prepost)
self._ticker: Any = None
@staticmethod
def _require_yfinance() -> Any:
if yf is None:
raise RuntimeError(
"Live Yahoo Finance mode requires yfinance. "
"Install it with: pip install yfinance numpy"
)
return yf
def _get_ticker(self) -> Any:
yahoo = self._require_yfinance()
if self._ticker is None:
self._ticker = yahoo.Ticker(self.symbol)
return self._ticker
@staticmethod
def _finite_float(value: Any, fallback: float = 0.0) -> float:
"""Convert a Yahoo/pandas scalar to a finite float."""
try:
# A MultiIndex row can occasionally return a one-item Series.
if hasattr(value, "iloc") and not isinstance(value, (str, bytes)):
value = value.iloc[0]
result = float(value)
return result if math.isfinite(result) else fallback
except (TypeError, ValueError, IndexError):
return fallback
@classmethod
def _row_value(cls, row: Any, name: str) -> float:
try:
value = row[name]
except (KeyError, TypeError, IndexError):
return 0.0
return cls._finite_float(value)
@staticmethod
def _latest_quote(ticker: Any, fallback: float) -> float:
"""Read Yahoo's latest quote; fall back only to Yahoo history close."""
try:
fast_info = ticker.fast_info
if hasattr(fast_info, "get"):
quote = fast_info.get("last_price", fallback)
else:
quote = fast_info["last_price"]
return YahooFinanceFeed._finite_float(quote, fallback)
except (KeyError, TypeError, AttributeError, ValueError, IndexError):
return fallback
@staticmethod
def _timestamp(index_value: Any) -> float:
try:
timestamp = float(index_value.timestamp())
if math.isfinite(timestamp) and timestamp > 0.0:
return timestamp
except (AttributeError, TypeError, ValueError, OSError, OverflowError):
pass
return time.time()
def fetch_candle(self) -> Candle:
"""Return the newest Yahoo Finance candle with Yahoo's latest quote."""
ticker = self._get_ticker()
try:
history = ticker.history(
period=self.period,
interval=self.interval,
prepost=self.prepost,
auto_adjust=False,
actions=False,
)
except Exception as exc:
raise RuntimeError(
f"Yahoo Finance history request failed for {self.symbol}: {exc}"
) from exc
if history is None or len(history) == 0:
raise RuntimeError(
f"Yahoo Finance returned no OHLCV data for {self.symbol}"
)
# yfinance normally returns one-level columns for Ticker.history. A
# defensive flatten keeps this source compatible with versions that
# return a one-symbol MultiIndex.
columns = getattr(history, "columns", None)
if columns is not None and getattr(columns, "nlevels", 1) > 1:
history = history.copy()
history.columns = [
column[0] if isinstance(column, tuple) else column
for column in columns
]
row = history.iloc[-1]
open_price = self._row_value(row, "Open")
high_price = self._row_value(row, "High")
low_price = self._row_value(row, "Low")
close_price = self._row_value(row, "Close")
volume = self._row_value(row, "Volume")
if not is_finite_positive(close_price):
raise RuntimeError(
f"Yahoo Finance returned an invalid latest price for {self.symbol}"
)
# fast_info is still yfinance/Yahoo Finance and gives the newest quote
# available. Keep history OHLC values while extending the current
# candle if the quote has moved outside that interval's range.
latest_quote = self._latest_quote(ticker, close_price)
if is_finite_positive(latest_quote):
close_price = latest_quote
high_price = max(high_price, close_price)
low_price = min(low_price, close_price)
# A malformed Yahoo row must never enter the signal engine.
if not is_finite_positive(open_price):
open_price = close_price
if not is_finite_positive(high_price):
high_price = max(open_price, close_price)
if not is_finite_positive(low_price):
low_price = min(open_price, close_price)
high_price = max(high_price, open_price, close_price)
low_price = min(low_price, open_price, close_price)
return Candle(
timestamp=self._timestamp(history.index[-1]),
open=open_price,
high=high_price,
low=low_price,
close=close_price,
volume=max(0.0, volume),
source_id=f"yahoo_finance:{self.symbol}",
is_closed=False,
)
def fetch_candles(self, limit: Optional[int] = None) -> List[Candle]:
"""Return closed historical Yahoo candles for engine warm-up."""
ticker = self._get_ticker()
try:
history = ticker.history(
period=self.period,
interval=self.interval,
prepost=self.prepost,
auto_adjust=False,
actions=False,
)
except Exception as exc:
raise RuntimeError(
f"Yahoo Finance history request failed for {self.symbol}: {exc}"
) from exc
if history is None or len(history) == 0:
raise RuntimeError(
f"Yahoo Finance returned no historical data for {self.symbol}"
)
columns = getattr(history, "columns", None)
if columns is not None and getattr(columns, "nlevels", 1) > 1:
history = history.copy()
history.columns = [
column[0] if isinstance(column, tuple) else column
for column in columns
]
if limit is not None:
if limit < 1:
raise ValueError("limit must be positive or None")
history = history.iloc[-limit:]
candles: List[Candle] = []
for index_value, row in history.iterrows():
open_price = self._row_value(row, "Open")
high_price = self._row_value(row, "High")
low_price = self._row_value(row, "Low")
close_price = self._row_value(row, "Close")
volume = max(0.0, self._row_value(row, "Volume"))
if not is_finite_positive(close_price):
continue
if not is_finite_positive(open_price):
open_price = close_price
if not is_finite_positive(high_price):
high_price = max(open_price, close_price)
if not is_finite_positive(low_price):
low_price = min(open_price, close_price)
high_price = max(high_price, open_price, close_price)
low_price = min(low_price, open_price, close_price)
candles.append(Candle(
timestamp=self._timestamp(index_value),
open=open_price,
high=high_price,
low=low_price,
close=close_price,
volume=volume,
source_id=f"yahoo_finance:{self.symbol}",
is_closed=True,
))
if not candles:
raise RuntimeError(
f"Yahoo Finance returned no valid historical candles for {self.symbol}"
)
return candles
def fetch_latest_price(self) -> float:
"""Return only the latest Yahoo Finance price."""
return self.fetch_candle().close
# ==============================================================================
# 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
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 ---
self._source_scores : Dict[str, RollingBuffer] = {}
# --- 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:
# 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
# Yahoo's newest interval is an open candle, so repeated polls can
# legitimately carry the same timestamp while its price changes.
# Closed candles still use the normal duplicate guard.
if candle.is_closed and 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:
self._source_scores[source_id] = RollingBuffer(30)
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:
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
# 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:
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 log return
if self._return_buf.is_ready(1):
prev_close = self._return_buf.last()
if prev_close > EPSILON:
ret = abs(safe_div(candle.close - prev_close, prev_close))
self._return_buf.push(candle.close)
else:
ret = 0.0
self._return_buf.push(candle.close)
else:
ret = 0.0
self._return_buf.push(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:
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):
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:
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"
@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
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)
def update(self, candle: Candle, di_result: Dict, ms_result: Dict) -> Dict:
try:
return self._update_inner(candle, di_result, ms_result)
except Exception:
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._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()
# 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_range + 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"])
# 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:
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)
if self._vol_buf.is_ready(3):
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
if self._vol_buf.is_ready(3):
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:
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
"""
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
def update(self, candle: Candle, asset_result: Dict) -> Dict:
try:
return self._update_inner(candle, asset_result)
except Exception:
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)
# 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 proxy: compare RSI direction to price direction
divergence = 0.0 # populated further in scoring engine
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 _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:
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:
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:
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:
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:
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
otc_fraud = lp.get("otc_fraud_score", 0.0) * 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.
Implements:
- Confidence hysteresis (no flip without strong evidence).
- Signal cooldown / spam control.
- Signal fatigue, stability, and flip-flop suppression.
- Stale signal decay.
- Expiry bucket assignment.
Input contract:
update(score_result, event_result, di_result, ms_result, tick_count) → 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,
}
"""
_COOLDOWN_TICKS = SIGNAL_COOLDOWN_TICKS
_FORMING_THRESH = 0.10
_CANDIDATE_THRESH = 0.25
_CONFIRMED_THRESH = 0.45
_HYSTERESIS_DELTA = 0.05 # min confidence change to flip direction
_STALE_TICKS = 10
_MAX_COOLING = 5
_FLIP_SUPPRESS_TICKS = 3 # minimum ticks before same signal can flip
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
# Rolling confidence for hysteresis
self._conf_buf : RollingBuffer = RollingBuffer(5)
def update(self, score_result: Dict, event_result: Dict,
di_result: Dict, ms_result: Dict, tick_count: int) -> Dict:
try:
return self._update_inner(score_result, event_result,
di_result, ms_result, tick_count)
except Exception:
return self._safe_default()
def _update_inner(self, score_result, event_result, di_result, ms_result, tick_count) -> Dict:
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:
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
# ----------------------------------------------------------------
if self._state == "confirmed":
freshness = clamp(1.0 - safe_div(self._tick_in_state, self._STALE_TICKS))
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 >= self._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 G OUTPUT WARMUP
# ==============================================================================
import math
import time
from typing import Any, Dict, List, Optional
# ==============================================================================
# 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:
return self._safe_default()
def refine(self, di_result: Dict, score_result: Dict) -> Dict:
"""
FIX-A: Post-score mode refinement WITHOUT advancing tick_count.
Incorporates manipulation_penalty into operational_mode.
Does NOT increment _tick_count. Corrects warm_frac to use
the already-counted tick (not a temporary N+1).
"""
try:
saved = self._tick_count
result = self._tick_inner(di_result, score_result)
self._tick_count = saved # restore — must NOT increment again
# Recompute warm_frac with correct (non-incremented) tick count
correct_wf = clamp(safe_div(saved - MIN_WARM_CANDLES,
FULL_WARM_CANDLES - MIN_WARM_CANDLES))
result["warm_up_fraction"] = correct_wf
return result
except Exception:
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
if di_result.get("suppress_signal", False):
self._blocked_ticks += 1
mode = "blocked"
blocked = True
degraded = False
elif data_quality < self._QUALITY_DEGRADED_THRESHOLD:
self._degraded_ticks += 1
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:
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:
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")
assert direction in SIGNAL_DIRECTIONS
# ---- Computed 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.
Example
-------
>>> engine = MAYTHOS()
>>> output = engine.tick(candle)
>>> print(output["direction"], output["confidence"], output["execution_suitability"])
"""
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()
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:
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
lc = self._lifecycle.update(score, ev, di, ms,
tick_count=self._warmup.tick_count)
# Layer 10
dtrace = self._debug_engine.build(
candle, di, ms, asset, sess, tf, ev, lp, tech,
score, lc, warmup, debug_mode=debug,
)
output = self._formatter.format(
di, ms, asset, sess, tf, ev, lp, tech,
score, lc, warmup, debug_trace=dtrace,
)
return output
def tick_yahoo(self, feed: "YahooFinanceFeed",
receive_time: Optional[float] = None,
debug: Optional[bool] = None) -> Dict:
"""Fetch one Yahoo Finance candle and process it through MAYTHOS."""
if not isinstance(feed, YahooFinanceFeed):
raise TypeError("feed must be a YahooFinanceFeed instance")
candle = feed.fetch_candle()
if receive_time is None:
receive_time = time.time()
return self.tick(candle, receive_time=receive_time, debug=debug)
def stream_yahoo(self, feed: "YahooFinanceFeed",
poll_seconds: float = 60.0,
max_ticks: Optional[int] = None,
debug: Optional[bool] = None):
"""Yield MAYTHOS outputs from Yahoo Finance at a fixed poll interval.
Set ``max_ticks`` for a finite run (useful in scripts and tests). The
default is continuous polling until the caller stops iteration.
"""
if not isinstance(feed, YahooFinanceFeed):
raise TypeError("feed must be a YahooFinanceFeed instance")
if not math.isfinite(poll_seconds) or poll_seconds < 0.0:
raise ValueError("poll_seconds must be a finite non-negative number")
if max_ticks is not None and max_ticks < 0:
raise ValueError("max_ticks must be non-negative or None")
emitted = 0
while max_ticks is None or emitted < max_ticks:
yield self.tick_yahoo(feed, debug=debug)
emitted += 1
if max_ticks is None or emitted < max_ticks:
time.sleep(poll_seconds)
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
# ==============================================================================
# ENTRY POINT — smoke test
# ==============================================================================
if __name__ == "__main__":
import argparse
import json
import sys
parser = argparse.ArgumentParser(description="MAYTHOS Yahoo Finance engine")
parser.add_argument(
"--symbol",
help="Yahoo Finance symbol, e.g. AAPL, BTC-USD, or EURUSD=X",
)
parser.add_argument("--interval", default="1m")
parser.add_argument("--period", default="1d")
parser.add_argument("--prepost", action="store_true")
parser.add_argument(
"--ticks",
type=int,
default=1,
help="Number of Yahoo polls to perform when --symbol is supplied",
)
parser.add_argument("--poll-seconds", type=float, default=60.0)
args = parser.parse_args()
if args.symbol:
if args.ticks < 1:
parser.error("--ticks must be at least 1")
feed = YahooFinanceFeed(
args.symbol,
interval=args.interval,
period=args.period,
prepost=args.prepost,
)
engine = MAYTHOS(debug_mode=False)
try:
for result in engine.stream_yahoo(
feed,
poll_seconds=args.poll_seconds,
max_ticks=args.ticks,
):
print(json.dumps(result, sort_keys=True, default=str))
except (RuntimeError, TypeError, ValueError) as exc:
print(f"Yahoo Finance error: {exc}", file=sys.stderr)
sys.exit(1)
sys.exit(0)
print(f"MAYTHOS v{ENGINE_VERSION} — Single-file self-test")
engine = MAYTHOS(debug_mode=False)
import datetime
base_ts = datetime.datetime(2024, 3, 12, 10, 30, 0).timestamp()
errors_seen = []
for i in range(150):
close = 1.0000 + i * 0.0005
candle = Candle(
timestamp = base_ts + i * 60.0,
open = close - 0.0002,
high = close + 0.0010,
low = close - 0.0008,
close = close,
volume = 500.0 + i * 2.0,
spread = 0.0002,
source_id = "test_feed",
)
output = engine.tick(candle)
errs = engine.validate(output)
if errs:
errors_seen.extend(errs)
print(f" tick {i:3d}: VALIDATION ERRORS: {errs}")
if errors_seen:
print(f"\n[FAIL] {len(errors_seen)} validation error(s).")
sys.exit(1)
print(f" tick count : {engine.tick_count}")
print(f" is_warm : {engine.is_warm}")
print(f" direction : {output['direction']}")
print(f" confidence : {output['confidence']:.4f}")
print(f" exec_suit : {output['execution_suitability']}")
print(f" market_state : {output['market_state']}")
print(f" regime : {output['regime_label']}")
print(f" asset_mode : {output['asset_mode']}")
print(f" reason : {output['reason_summary']}")
print(f"\n[PASS] All 150 ticks produced valid output. MAYTHOS v{ENGINE_VERSION} operational.")
#!/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; Yahoo Finance via yfinance + 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)
Yahoo Finance live mode:
feed = YahooFinanceFeed("AAPL", interval="1m", period="1d")
output = engine.tick_yahoo(feed)
"""
# ==============================================================================
# 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
# Yahoo Finance is the only live-market data source used by this file. Keep
# the import optional so the analysis engine can still be imported for offline
# backtests; YahooFinanceFeed raises a clear install error when live mode is
# requested without yfinance.
try:
import yfinance as yf
except ImportError: # pragma: no cover - depends on the host environment
yf = None # type: ignore[assignment]
# ==============================================================================
# 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"})