""" Bot Shape Classifier Uses incoming market price streams and orderbook analyses to compute short-time spectrograms, extract spectral features, and classify observed bot/whale shapes (e.g., grid, spiral, oscillator, taper, accumulation/distribution). Publishes: - topic: `whale.sonar.spectrogram` payload: {symbol, detected_at, peaks, centroid, bandwidth} - topic: `whale.shape.detected` payload: {symbol, detected_at, shape: {subtype, score, reason}, spectrogram_summary} """ from __future__ import annotations from aureon.core.aureon_baton_link import link_system as _baton_link; _baton_link(__name__) import logging import math import time from collections import defaultdict, deque from dataclasses import asdict from typing import Any, Dict, List, Optional import numpy as np from aureon.core.aureon_thought_bus import get_thought_bus, Thought try: from aureon.bots.whale_metrics import whale_shape_detected_total METRICS_AVAILABLE = True except ImportError: METRICS_AVAILABLE = False whale_shape_detected_total = None # ════════════════════════════════════════════════════════════════════════════════ # 🐦 CHIRP BUS INTEGRATION - Emit strong bot patterns to Orca! # ════════════════════════════════════════════════════════════════════════════════ CHIRP_BUS_AVAILABLE = False get_chirp_bus = None try: from aureon.core.aureon_chirp_bus import get_chirp_bus CHIRP_BUS_AVAILABLE = True except ImportError: CHIRP_BUS_AVAILABLE = False # ════════════════════════════════════════════════════════════════════════════════ # 🌐 GLOBAL MARKET INTEGRATION # ════════════════════════════════════════════════════════════════════════════════ try: from aureon.exchanges.kraken_client import KrakenClient KRAKEN_AVAILABLE = True except ImportError: KRAKEN_AVAILABLE = False KrakenClient = None try: from aureon.exchanges.binance_ws_client import BinanceWebSocketClient BINANCE_WS_AVAILABLE = True except ImportError: BINANCE_WS_AVAILABLE = False BinanceWebSocketClient = None try: from aureon.exchanges.capital_client import CapitalClient CAPITAL_AVAILABLE = True except ImportError: CAPITAL_AVAILABLE = False CapitalClient = None logger = logging.getLogger(__name__) class BotShapeClassifier: def __init__(self, sample_window: int = 128, hop: int = 32): self.thought_bus = get_thought_bus() # price history per symbol: deque of (ts, price) self.price_hist: Dict[str, deque] = defaultdict(lambda: deque(maxlen=1024)) self.sample_window = int(sample_window) self.hop = int(hop) # 🌐 GLOBAL MARKET CLIENTS self.kraken_client: Optional[Any] = None self.binance_ws: Optional[Any] = None self.capital_client: Optional[Any] = None # Initialize market connections self._init_market_connections() # 🐦 CHIRP BUS - Emit strong bot patterns to Orca self.chirp_bus = None if CHIRP_BUS_AVAILABLE and get_chirp_bus: try: self.chirp_bus = get_chirp_bus() logger.info("🐦 Bot Shape Classifier → Orca CHIRP BUS connected") except Exception as e: logger.debug(f"Chirp bus init failed: {e}") # subscribe to market price updates and orderbook analyses try: self.thought_bus.subscribe('market.*', self._on_market) except Exception: logger.debug('BotShapeClassifier: failed to subscribe to market.*') try: self.thought_bus.subscribe('whale.orderbook.analyzed', self._on_orderbook) except Exception: logger.debug('BotShapeClassifier: failed to subscribe to whale.orderbook.analyzed') def _init_market_connections(self) -> None: """Initialize optional exchange client connections for live data.""" if KRAKEN_AVAILABLE and KrakenClient is not None: try: self.kraken_client = KrakenClient() logger.info("BotShapeClassifier: Kraken client connected") except Exception as e: logger.debug(f"BotShapeClassifier: Kraken init failed: {e}") if BINANCE_WS_AVAILABLE and BinanceWebSocketClient is not None: try: self.binance_ws = BinanceWebSocketClient() logger.info("BotShapeClassifier: Binance WS connected") except Exception as e: logger.debug(f"BotShapeClassifier: Binance WS init failed: {e}") if CAPITAL_AVAILABLE and CapitalClient is not None: try: self.capital_client = CapitalClient() logger.info("BotShapeClassifier: Capital.com client connected") except Exception as e: logger.debug(f"BotShapeClassifier: Capital.com init failed: {e}") def _on_market(self, thought: Thought) -> None: payload = thought.payload or {} symbol = payload.get('symbol') or payload.get('s') price = payload.get('price') or payload.get('c') or payload.get('close') ts = payload.get('ts') or thought.ts or time.time() try: if not symbol or price is None: return price = float(price) self.price_hist[symbol].append((ts, price)) except Exception: return def _on_orderbook(self, thought: Thought) -> None: payload = thought.payload or {} symbol = payload.get('symbol') exchange = payload.get('exchange') or payload.get('exch') if not symbol: return # If we have enough price history, compute spectrogram and classify ph = list(self.price_hist.get(symbol, [])) if len(ph) < self.sample_window: # not enough data -> try to fetch from external cache (best-effort) logger.debug('Not enough price history for %s (have=%d need=%d)', symbol, len(ph), self.sample_window) return timestamps, prices = zip(*ph[-self.sample_window:]) prices = np.array(prices) # Integrate harmonic analyzer (if available) for frequency/coherence/phase features harmonic_info = {} try: from aureon.bridges.aureon_probability_nexus import HarmonicAnalyzer ha = HarmonicAnalyzer() # HarmonicAnalyzer expects a price list; pass the last 100 prices if available freq, coh, ph = ha.analyze(list(prices[-100:])) harmonic_info = {'frequency': float(freq), 'coherence': float(coh), 'phase': float(ph)} except Exception: harmonic_info = {} # Work with log returns to emphasize micro-movements returns = np.diff(np.log(prices + 1e-12)) if len(returns) < 8: return # Build spectrogram via short-time FFT S = self._stft(returns, n_fft=self.sample_window//2, hop=self.hop) # magnitude spectrogram mag = np.abs(S) # Normalize per-column mag_norm = mag / (np.max(mag, axis=0, keepdims=True) + 1e-9) # Spectral features centroid = self._spectral_centroid(mag) bandwidth = self._spectral_bandwidth(mag, centroid) peaks = self._dominant_peaks(mag, top_n=3) flatness = self._spectral_flatness(mag) energy = float(np.sum(mag)) spectro_summary = { 'shape_frames': mag.shape[1], 'centroid': float(centroid), 'bandwidth': float(bandwidth), 'flatness': float(flatness), 'energy': float(energy), 'peaks': peaks, 'harmonic': harmonic_info, } # Heuristic classification layering = float(payload.get('layering_score', 0.0) or 0.0) bids_depth = float(payload.get('bids_depth', 0.0) or 0.0) asks_depth = float(payload.get('asks_depth', 0.0) or 0.0) subtype, score, reason = self._classify_shape(spectro_summary, layering, bids_depth, asks_depth) detected = { 'symbol': symbol, 'exchange': exchange, 'detected_at': time.time(), 'shape': {'subtype': subtype, 'score': float(score), 'reason': reason}, 'spectrogram': spectro_summary, } # Emit strong patterns to Orca for trading signals self.emit_shape_to_orca(symbol, detected['shape'], exchange, spectro_summary) # Publish both a spectrogram thought (compact) and a shape classification s_th = Thought(source='bot_shape_classifier', topic='whale.sonar.spectrogram', payload={'symbol': symbol, 'spectrogram': spectro_summary}) try: self.thought_bus.publish(s_th) except Exception: logger.debug('Failed to publish whale.sonar.spectrogram') # A-Z-Z-A echo signature (simple mirrored peaks) to model echolocation-like ping try: peak_bins = [p['bin'] for p in peaks] echo_sig = peak_bins + peak_bins[::-1] echo_payload = {'symbol': symbol, 'echo_signature': echo_sig, 'detected_at': time.time()} e_th = Thought(source='bot_shape_classifier', topic='whale.sonar.echo', payload=echo_payload) self.thought_bus.publish(e_th) except Exception: logger.debug('Failed to publish whale.sonar.echo') th = Thought(source='bot_shape_classifier', topic='whale.shape.detected', payload=detected) try: self.thought_bus.publish(th) # Emit metrics if METRICS_AVAILABLE and whale_shape_detected_total: whale_shape_detected_total.inc(subtype=subtype, symbol=symbol, exchange=exchange or 'unknown') except Exception: logger.debug('Failed to publish whale.shape.detected') # ----------------- signal processing helpers ----------------- def _stft(self, x: np.ndarray, n_fft: int = 32, hop: int = 8) -> np.ndarray: n = len(x) if n < n_fft: n_fft = max(8, n//2) frames = [] for start in range(0, n - n_fft + 1, hop): win = x[start:start+n_fft] * np.hanning(n_fft) fft = np.fft.rfft(win) frames.append(fft) if not frames: return np.zeros((n_fft//2+1, 0), dtype=np.complex64) return np.column_stack(frames) def _spectral_centroid(self, mag: np.ndarray) -> float: freqs = np.arange(mag.shape[0]) power = np.mean(mag, axis=1) num = np.sum(freqs * power) den = np.sum(power) + 1e-12 return float(num / den) def _spectral_bandwidth(self, mag: np.ndarray, centroid: float) -> float: freqs = np.arange(mag.shape[0]) power = np.mean(mag, axis=1) num = np.sum(((freqs - centroid) ** 2) * power) den = np.sum(power) + 1e-12 return float(math.sqrt(num / den)) def _dominant_peaks(self, mag: np.ndarray, top_n: int = 3) -> List[Dict[str, float]]: mean_spec = np.mean(mag, axis=1) idx = np.argsort(mean_spec)[::-1][:top_n] return [{'bin': int(i), 'power': float(mean_spec[i])} for i in idx] def _spectral_flatness(self, mag: np.ndarray) -> float: mean_spec = np.mean(mag, axis=1) + 1e-12 geo = float(np.exp(np.mean(np.log(mean_spec)))) arith = float(np.mean(mean_spec)) return float(geo / (arith + 1e-12)) # ----------------- classification heuristics ----------------- def _classify_shape(self, spectro: Dict[str, Any], layering: float, bids_depth: float, asks_depth: float): centroid = spectro['centroid'] flatness = spectro['flatness'] energy = spectro['energy'] peaks = spectro['peaks'] # Heuristic rules: # - Grid: high layering or low flatness with multiple evenly strong peaks # - Oscillator: dominant single peak, low bandwidth # - Spiral: shifting centroid across frames (not trivially detectable here) -> use bandwidth>threshold # - Taper: energy low and flatness increasing # - Accumulation/Distribution: inferred from wall side/imbalance (bids_depth vs asks_depth) # grid if layering > 0.6 or (len(peaks) >= 2 and flatness < 0.5 and peaks[1]['power'] > 0.4 * peaks[0]['power']): return 'grid', min(0.99, 0.4 + 0.6 * layering + (1.0 - flatness) * 0.2), 'layering_or_multi_peak' # oscillator if spectro['bandwidth'] < 2.0 and peaks and peaks[0]['power'] > 0.5 * max(0.01, sum(p['power'] for p in peaks)): return 'oscillator', min(0.99, 0.5 + (1.0 - spectro['bandwidth']/10.0)), 'dominant_peak_low_bandwidth' # spiral (bandwidth high but centroid moderate) if spectro['bandwidth'] > 4.0 and spectro['centroid'] > 2.0: return 'spiral', min(0.95, 0.4 + 0.12 * spectro['bandwidth']), 'wideband_centroid' # taper if energy < 1e-3 or spectro['flatness'] > 0.85: return 'taper', 0.6, 'low_energy_or_flat' # accumulation / distribution if bids_depth > asks_depth * 1.5: return 'accumulation', min(0.99, 0.5 + min(0.5, bids_depth / (asks_depth + 1e-9) / 10.0)), 'bid_depth_imbalance' if asks_depth > bids_depth * 1.5: return 'distribution', min(0.99, 0.5 + min(0.5, asks_depth / (bids_depth + 1e-9) / 10.0)), 'ask_depth_imbalance' # default neutral return 'neutral', 0.4, 'no_strong_signal' # Default singleton default_bot_shape_classifier: Optional[BotShapeClassifier] = None try: default_bot_shape_classifier = BotShapeClassifier() except Exception: default_bot_shape_classifier = None if __name__ == '__main__': # quick CLI test: ingest sample generated price data and a fake orderbook import argparse import math ap = argparse.ArgumentParser() ap.add_argument('symbol', nargs='?', default='TESTCOIN/USD') args = ap.parse_args() from aureon.core.aureon_thought_bus import get_thought_bus, Thought tb = get_thought_bus() # publish synthetic market ticks (sine + noise) symbol = args.symbol for t in range(256): price = 100.0 + math.sin(t/4.0) * 0.5 + (0.01 * (t % 5)) th = Thought(source='cli', topic='market.snapshot', payload={'symbol': symbol, 'price': price, 'ts': time.time()}) tb.publish(th) time.sleep(0.01) # publish a fake orderbook analyzed event analysis = {'symbol': symbol, 'detected_at': time.time(), 'walls': [{'price':100.0,'size':2000.0,'notional_usd':200000.0,'side':'bid'}], 'layering_score':0.2, 'bids_depth':200000.0, 'asks_depth':50000.0} tb.publish(Thought(source='cli', topic='whale.orderbook.analyzed', payload=analysis))