# indicators_extended.py - Extended indicators with ALL parameters # Compatible with Python 3.9+ import numpy as np import math from typing import Optional, Dict, List, Tuple from collections import deque # ================================================================ # PRICE CHANGES - MULTIPLE TIMEFRAMES # ================================================================ def calc_all_price_changes(current_price, price_history): """Calculate price changes for multiple timeframes""" timeframes = { "1m": 60, "3m": 180, "5m": 300, "10m": 600, "15m": 900, "20m": 1200, "30m": 1800, } changes = {} for name, seconds in timeframes.items(): changes[name] = calc_single_price_change(current_price, price_history, seconds) return changes def calc_single_price_change(current_price, price_history, seconds_back): """Calculate price change for a single timeframe""" if not price_history: return 0.0 import time current_time = time.time() target_time = current_time - seconds_back closest_price = None min_time_diff = float("inf") for ts, price in price_history: time_diff = abs(ts - target_time) if time_diff < min_time_diff: min_time_diff = time_diff closest_price = price if closest_price is None or closest_price == 0: return 0.0 price_change_pct = ((current_price - closest_price) / closest_price) * 100.0 return float(price_change_pct) # ================================================================ # SMOOTH (MOMENTUM SLOPE) - MULTIPLE TIMEFRAMES # ================================================================ def calc_smooth_multi_tf(price_window, price_history): """Calculate momentum slope for multiple timeframes""" # 2m smooth from price_window (120 samples) smooth_2m = compute_smooth(list(price_window)) # 5m smooth prices_5m = get_prices_for_timeframe(price_history, 300) smooth_5m = compute_smooth(prices_5m) # 10m smooth prices_10m = get_prices_for_timeframe(price_history, 600) smooth_10m = compute_smooth(prices_10m) # 15m smooth prices_15m = get_prices_for_timeframe(price_history, 900) smooth_15m = compute_smooth(prices_15m) return { "2m": smooth_2m, "5m": smooth_5m, "10m": smooth_10m, "15m": smooth_15m, } def get_prices_for_timeframe(price_history, seconds): """Extract prices for specific timeframe""" import time current_time = time.time() cutoff_time = current_time - seconds prices = [price for ts, price in price_history if ts >= cutoff_time] return prices def compute_smooth(prices, scale=0.0001): """Calculate momentum using linear regression""" if len(prices) < 5: return 0.0 x = np.arange(len(prices)) y = np.array(prices, dtype=float) try: m, _ = np.polyfit(x, y, 1) smooth = np.tanh(m / (y.mean() * scale + 1e-8)) except Exception: smooth = 0.0 return float(smooth) # ================================================================ # VOLATILITY # ================================================================ def calc_volatility(price_history): """Calculate volatility (standard deviation) for multiple timeframes""" import time current_time = time.time() # 1m volatility prices_1m = [p for ts, p in price_history if current_time - ts <= 60] vol_1m = np.std(prices_1m) / np.mean(prices_1m) * 100 if len(prices_1m) > 1 else 0.0 # 5m volatility prices_5m = [p for ts, p in price_history if current_time - ts <= 300] vol_5m = np.std(prices_5m) / np.mean(prices_5m) * 100 if len(prices_5m) > 1 else 0.0 # High-Low range for 5m if len(prices_5m) > 0: high_5m = max(prices_5m) low_5m = min(prices_5m) high_low_range = (high_5m - low_5m) / low_5m * 100 if low_5m > 0 else 0.0 else: high_low_range = 0.0 return { "1m": float(vol_1m), "5m": float(vol_5m), "high_low_range_5m": float(high_low_range), } # ================================================================ # MONEY FLOW RATIOS - MULTIPLE TIMEFRAMES # ================================================================ def calc_all_ratios(trades_current, trades_window): """Calculate buy/sell ratios for multiple timeframes""" import time current_time = time.time() # Convert deque to list for time-based filtering trades_list = list(trades_window) # 1m ratio (current batch) ratio_1m = calc_single_ratio(trades_current) # 5m ratio trades_5m = [t for t in trades_list if current_time - t["T"]/1000 <= 300] ratio_5m = calc_single_ratio(trades_5m) # 15m ratio trades_15m = [t for t in trades_list if current_time - t["T"]/1000 <= 900] ratio_15m = calc_single_ratio(trades_15m) # 20m ratio trades_20m = [t for t in trades_list if current_time - t["T"]/1000 <= 1200] ratio_20m = calc_single_ratio(trades_20m) # 30m ratio trades_30m = [t for t in trades_list if current_time - t["T"]/1000 <= 1800] ratio_30m = calc_single_ratio(trades_30m) # MF Acceleration (change in buy ratio) mf_acceleration = ratio_1m - ratio_5m return { "1m": ratio_1m, "5m": ratio_5m, "15m": ratio_15m, "20m": ratio_20m, "30m": ratio_30m, "mf_acceleration": float(mf_acceleration), } def calc_single_ratio(trades): """Calculate buy/sell ratio for a list of trades""" if not trades: return 0.5 buy = 0.0 sell = 0.0 for t in trades: qty = t["q"] if t["isBuyerMaker"]: sell += qty else: buy += qty total = buy + sell if total == 0: return 0.5 return float(buy / total) # ================================================================ # CVD EXTENDED # ================================================================ def calc_cvd_extended(cvd_current, cvd_history): """Calculate extended CVD metrics""" import time current_time = time.time() # CVD 5m average cvd_5m_values = [cvd for ts, cvd in cvd_history if current_time - ts <= 300] cvd_5m = np.mean(cvd_5m_values) if cvd_5m_values else cvd_current # CVD 15m average cvd_15m_values = [cvd for ts, cvd in cvd_history if current_time - ts <= 900] cvd_15m = np.mean(cvd_15m_values) if cvd_15m_values else cvd_current # CVD change rate (derivative) if len(cvd_5m_values) > 1: cvd_change_rate = cvd_5m_values[-1] - cvd_5m_values[0] else: cvd_change_rate = 0.0 # CVD acceleration (second derivative) if len(cvd_history) >= 3: recent_cvds = [cvd for _, cvd in list(cvd_history)[-3:]] cvd_acceleration = recent_cvds[-1] - 2*recent_cvds[-2] + recent_cvds[-3] else: cvd_acceleration = 0.0 return { "current": float(cvd_current), "5m": float(cvd_5m), "15m": float(cvd_15m), "change_rate": float(cvd_change_rate), "acceleration": float(cvd_acceleration), } # ================================================================ # WHALE EXTENDED # ================================================================ def calc_whale_extended(whale_data, whale_history): """Calculate extended whale metrics""" # Current whale data whale_total = whale_data["whale_total"] whale_buy = whale_data["whale_buy"] whale_sell = whale_data["whale_sell"] whale_count = whale_data["whale_count"] whale_largest = whale_data["whale_largest"] whale_avg = whale_data["whale_avg"] # Buy/Sell ratio if whale_sell > 0: whale_buy_sell_ratio = whale_buy / whale_sell else: whale_buy_sell_ratio = whale_buy if whale_buy > 0 else 0.0 return { "total": float(whale_total), "buy": float(whale_buy), "sell": float(whale_sell), "count": int(whale_count), "largest": float(whale_largest), "avg": float(whale_avg), "buy_sell_ratio": float(whale_buy_sell_ratio), } # ================================================================ # ORDERBOOK EXTENDED + NEW LIQUIDITY METRICS # ================================================================ def calc_orderbook_extended(orderbook, current_price): """Calculate extended orderbook metrics with liquidity analysis""" bids = orderbook.get("bids", []) asks = orderbook.get("asks", []) if not bids or not asks: return { "depth_ratio": 1.0, "best_level_ratio": 1.0, "spread": 0.0, "spread_pct": 0.0, "imbalance_5": 0.0, "imbalance_20": 0.0, "bid_wall_size": 0.0, "ask_wall_size": 0.0, "wall_distance_pct": 0.0, # New liquidity metrics "liquidity_bid_pressure": 0.0, "liquidity_ask_pressure": 0.0, "liquidity_shift": 0.0, "impact_buy_price": current_price, "impact_sell_price": current_price, "orderbook_entropy": 0.0, "liquidity_gap_bids": 0.0, "liquidity_gap_asks": 0.0, "spoof_buy_score": 0.0, "spoof_sell_score": 0.0, "iceberg_score": 0.0, } # Basic metrics bid_volume = sum(float(x[1]) for x in bids) ask_volume = sum(float(x[1]) for x in asks) depth_ratio = bid_volume / ask_volume if ask_volume > 0 else 1.0 best_bid = float(bids[0][0]) best_ask = float(asks[0][0]) best_level_ratio = best_bid / best_ask if best_ask > 0 else 1.0 spread = best_ask - best_bid spread_pct = (spread / best_bid) * 100 if best_bid > 0 else 0.0 # Imbalance bid_vol_5 = sum(float(x[1]) for x in bids[:5]) ask_vol_5 = sum(float(x[1]) for x in asks[:5]) imbalance_5 = (bid_vol_5 - ask_vol_5) / (bid_vol_5 + ask_vol_5) if (bid_vol_5 + ask_vol_5) > 0 else 0.0 bid_vol_20 = sum(float(x[1]) for x in bids[:20]) ask_vol_20 = sum(float(x[1]) for x in asks[:20]) imbalance_20 = (bid_vol_20 - ask_vol_20) / (bid_vol_20 + ask_vol_20) if (bid_vol_20 + ask_vol_20) > 0 else 0.0 # Walls bid_sizes = [float(x[1]) for x in bids] ask_sizes = [float(x[1]) for x in asks] bid_wall_size = max(bid_sizes) if bid_sizes else 0.0 ask_wall_size = max(ask_sizes) if ask_sizes else 0.0 if bid_wall_size > 0: bid_wall_idx = bid_sizes.index(bid_wall_size) bid_wall_price = float(bids[bid_wall_idx][0]) bid_wall_distance = abs(current_price - bid_wall_price) / current_price * 100 else: bid_wall_distance = 0.0 if ask_wall_size > 0: ask_wall_idx = ask_sizes.index(ask_wall_size) ask_wall_price = float(asks[ask_wall_idx][0]) ask_wall_distance = abs(ask_wall_price - current_price) / current_price * 100 else: ask_wall_distance = 0.0 wall_distance_pct = min(bid_wall_distance, ask_wall_distance) # ───────────────────────────────────────────────────────── # NEW LIQUIDITY METRICS # ───────────────────────────────────────────────────────── # Liquidity pressure (weighted by distance from price) liquidity_bid_pressure = calc_liquidity_pressure(bids, current_price, is_bid=True) liquidity_ask_pressure = calc_liquidity_pressure(asks, current_price, is_bid=False) # Liquidity shift (momentum of order book changes) liquidity_shift = (liquidity_bid_pressure - liquidity_ask_pressure) / 2.0 # Market impact prices (what price you'd get for a 10 BTC order) impact_buy_price = calc_market_impact_price(asks, 10.0) impact_sell_price = calc_market_impact_price(bids, 10.0, is_bid=True) # Orderbook entropy (measure of liquidity distribution) orderbook_entropy = calc_orderbook_entropy(bids, asks) # Liquidity gaps (detect spoofing) liquidity_gap_bids = calc_liquidity_gaps(bids) liquidity_gap_asks = calc_liquidity_gaps(asks) # Spoofing scores spoof_buy_score = calc_spoof_score(bids, bid_wall_size) spoof_sell_score = calc_spoof_score(asks, ask_wall_size) # Iceberg detection iceberg_score = calc_iceberg_score(bids, asks, bid_volume, ask_volume) return { "depth_ratio": float(depth_ratio), "best_level_ratio": float(best_level_ratio), "spread": float(spread), "spread_pct": float(spread_pct), "imbalance_5": float(imbalance_5), "imbalance_20": float(imbalance_20), "bid_wall_size": float(bid_wall_size), "ask_wall_size": float(ask_wall_size), "wall_distance_pct": float(wall_distance_pct), # New metrics "liquidity_bid_pressure": float(liquidity_bid_pressure), "liquidity_ask_pressure": float(liquidity_ask_pressure), "liquidity_shift": float(liquidity_shift), "impact_buy_price": float(impact_buy_price), "impact_sell_price": float(impact_sell_price), "orderbook_entropy": float(orderbook_entropy), "liquidity_gap_bids": float(liquidity_gap_bids), "liquidity_gap_asks": float(liquidity_gap_asks), "spoof_buy_score": float(spoof_buy_score), "spoof_sell_score": float(spoof_sell_score), "iceberg_score": float(iceberg_score), } def calc_liquidity_pressure(orders, current_price, is_bid=True): """Calculate weighted liquidity pressure""" if not orders: return 0.0 pressure = 0.0 for price_str, qty_str in orders[:20]: price = float(price_str) qty = float(qty_str) # Distance weight (closer = more weight) distance = abs(price - current_price) / current_price weight = 1.0 / (1.0 + distance * 100) # Decays with distance pressure += qty * weight return pressure def calc_market_impact_price(orders, target_qty, is_bid=False): """Calculate average price for executing target_qty""" if not orders: return float(orders[0][0]) if orders else 0.0 cumulative_qty = 0.0 total_cost = 0.0 for price_str, qty_str in orders: price = float(price_str) qty = float(qty_str) if cumulative_qty + qty >= target_qty: # This order completes our target remaining = target_qty - cumulative_qty total_cost += remaining * price cumulative_qty = target_qty break else: total_cost += qty * price cumulative_qty += qty if cumulative_qty == 0: return float(orders[0][0]) avg_price = total_cost / cumulative_qty return avg_price def calc_orderbook_entropy(bids, asks): """Calculate entropy of order book (measure of liquidity distribution)""" all_sizes = [float(x[1]) for x in bids[:20]] + [float(x[1]) for x in asks[:20]] if not all_sizes or sum(all_sizes) == 0: return 0.0 # Normalize to probabilities total = sum(all_sizes) probs = [size/total for size in all_sizes if size > 0] # Calculate entropy entropy = -sum(p * np.log2(p) for p in probs if p > 0) return entropy def calc_liquidity_gaps(orders): """Detect gaps in liquidity (potential spoofing)""" if len(orders) < 2: return 0.0 sizes = [float(x[1]) for x in orders[:20]] avg_size = np.mean(sizes) # Count significant gaps (where next level is much smaller) gaps = 0 for i in range(len(sizes)-1): if sizes[i] > avg_size * 2 and sizes[i+1] < avg_size * 0.5: gaps += 1 return float(gaps) def calc_spoof_score(orders, wall_size): """Calculate spoofing score (0-1)""" if not orders or wall_size == 0: return 0.0 sizes = [float(x[1]) for x in orders[:20]] avg_size = np.mean(sizes) # High score if there's a large wall followed by thin liquidity if wall_size > avg_size * 3: thin_count = sum(1 for s in sizes[1:] if s < avg_size * 0.5) spoof_score = min(1.0, thin_count / 10.0) else: spoof_score = 0.0 return spoof_score def calc_iceberg_score(bids, asks, bid_volume, ask_volume): """Detect potential iceberg orders""" if not bids or not asks: return 0.0 # Iceberg detection: consistent replenishment at same price level # For now, return 0 (requires historical tracking) return 0.0 # ================================================================ # VOLUME EXTENDED # ================================================================ def calc_volume_extended(trades, volume_window): """Calculate extended volume metrics""" import time current_time = time.time() # Current 1m volume volume_1m = sum(t["q"] for t in trades) # 5m volume (from window) volume_window_list = list(volume_window) volumes_5m = [vol for ts, vol in volume_window_list if current_time - ts <= 300] volume_5m = sum(volumes_5m) if volumes_5m else volume_1m # 15m volume volumes_15m = [vol for ts, vol in volume_window_list if current_time - ts <= 900] volume_15m = sum(volumes_15m) if volumes_15m else volume_1m # Average volumes avg_volume_1m = np.mean([vol for _, vol in volume_window_list]) if volume_window_list else volume_1m # Volume change rate if avg_volume_1m > 0: volume_change_rate = ((volume_1m - avg_volume_1m) / avg_volume_1m) * 100 else: volume_change_rate = 0.0 # Volume acceleration if len(volumes_5m) > 1: volume_acceleration = volumes_5m[-1] - volumes_5m[0] else: volume_acceleration = 0.0 # Average trade size avg_trade_size = volume_1m / len(trades) if trades else 0.0 # Trade frequency (trades per minute) trade_frequency = len(trades) * 60 # Large trades (>$10K) large_trade_count = sum(1 for t in trades if t["p"] * t["q"] >= 10000) large_trade_frequency = large_trade_count * 60 return { "1m": float(volume_1m), "5m": float(volume_5m), "15m": float(volume_15m), "avg_1m": float(avg_volume_1m), "change_rate": float(volume_change_rate), "acceleration": float(volume_acceleration), "avg_trade_size": float(avg_trade_size), "trade_frequency": float(trade_frequency), "large_trade_frequency": float(large_trade_frequency), } # ================================================================ # MARKET MICROSTRUCTURE + NEW METRICS # ================================================================ def calc_market_microstructure(trades, trades_window): """Calculate market microstructure metrics""" if not trades: return { "trades_per_minute": 0.0, "buy_trades_count": 0, "sell_trades_count": 0, "avg_time_between_trades": 0.0, "aggressive_buy_pct": 0.0, "aggressive_sell_pct": 0.0, "delta_strength": 0.0, "vwap_drift": 0.0, "micro_trend_3s": 0.0, "micro_trend_7s": 0.0, "micro_trend_12s": 0.0, "variance_ratio": 0.0, "breakout_energy": 0.0, } # Basic counts trades_per_minute = len(trades) * 60 buy_trades = sum(1 for t in trades if not t["isBuyerMaker"]) sell_trades = sum(1 for t in trades if t["isBuyerMaker"]) total_trades = len(trades) aggressive_buy_pct = (buy_trades / total_trades) * 100 if total_trades > 0 else 0.0 aggressive_sell_pct = (sell_trades / total_trades) * 100 if total_trades > 0 else 0.0 # Average time between trades if len(trades) > 1: timestamps = [t["T"] for t in trades] time_diffs = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps)-1)] avg_time_between_trades = np.mean(time_diffs) if time_diffs else 0.0 else: avg_time_between_trades = 0.0 # ───────────────────────────────────────────────────────── # NEW METRICS # ───────────────────────────────────────────────────────── # Delta strength (buy pressure vs sell pressure) buy_volume = sum(t["q"] for t in trades if not t["isBuyerMaker"]) sell_volume = sum(t["q"] for t in trades if t["isBuyerMaker"]) total_volume = buy_volume + sell_volume delta_strength = (buy_volume - sell_volume) / total_volume if total_volume > 0 else 0.0 # VWAP drift (deviation from VWAP) vwap = calc_vwap(trades) current_price = trades[-1]["p"] if trades else 0 vwap_drift = (current_price - vwap) / vwap if vwap > 0 else 0.0 # Micro trends (very short-term momentum) trades_list = list(trades_window) micro_trend_3s = calc_micro_trend(trades_list, 3) micro_trend_7s = calc_micro_trend(trades_list, 7) micro_trend_12s = calc_micro_trend(trades_list, 12) # Variance ratio (for mean reversion detection) variance_ratio = calc_variance_ratio(trades) # Breakout energy (sudden volume spike + price move) breakout_energy = calc_breakout_energy(trades) return { "trades_per_minute": float(trades_per_minute), "buy_trades_count": int(buy_trades), "sell_trades_count": int(sell_trades), "avg_time_between_trades": float(avg_time_between_trades), "aggressive_buy_pct": float(aggressive_buy_pct), "aggressive_sell_pct": float(aggressive_sell_pct), # New metrics "delta_strength": float(delta_strength), "vwap_drift": float(vwap_drift), "micro_trend_3s": float(micro_trend_3s), "micro_trend_7s": float(micro_trend_7s), "micro_trend_12s": float(micro_trend_12s), "variance_ratio": float(variance_ratio), "breakout_energy": float(breakout_energy), } def calc_vwap(trades): """Calculate Volume-Weighted Average Price""" if not trades: return 0.0 total_value = sum(t["p"] * t["q"] for t in trades) total_volume = sum(t["q"] for t in trades) return total_value / total_volume if total_volume > 0 else 0.0 def calc_micro_trend(trades_list, seconds): """Calculate micro trend over N seconds""" import time if not trades_list: return 0.0 current_time = time.time() cutoff = current_time - seconds recent_trades = [t for t in trades_list if t["T"]/1000 >= cutoff] if len(recent_trades) < 2: return 0.0 prices = [t["p"] for t in recent_trades] first_price = prices[0] last_price = prices[-1] trend = (last_price - first_price) / first_price if first_price > 0 else 0.0 return float(trend) def calc_variance_ratio(trades): """Calculate variance ratio (for mean reversion)""" if len(trades) < 10: return 0.0 prices = [t["p"] for t in trades] returns = np.diff(prices) / prices[:-1] if len(returns) < 2: return 0.0 var_1 = np.var(returns) var_2 = np.var(returns[::2]) # Every 2nd return variance_ratio = var_2 / (2 * var_1) if var_1 > 0 else 1.0 return float(variance_ratio) def calc_breakout_energy(trades): """Calculate breakout energy (volume * price_change)""" if len(trades) < 2: return 0.0 prices = [t["p"] for t in trades] volumes = [t["q"] for t in trades] price_change = (prices[-1] - prices[0]) / prices[0] if prices[0] > 0 else 0.0 volume_sum = sum(volumes) energy = abs(price_change) * volume_sum return float(energy) # ================================================================ # COMPOSITE INDICATORS # ================================================================ def calc_composite_indicators(smooth, buy_ratio, cvd, whale, volume_change, depth_ratio): """Calculate composite market indicators""" # Momentum Strength (0-100) momentum_strength = max(0, min(100, ( smooth * 40 + (buy_ratio - 0.5) * 100 + cvd * 5 + whale * 10 ))) # Volume-Price Trend vpt = volume_change * (1 if smooth > 0 else -1) # Buying Pressure (0-100) buying_pressure = max(0, min(100, ( (buy_ratio - 0.5) * 100 + cvd * 10 + whale * 5 + (depth_ratio - 1) * 20 ))) # Selling Pressure (inverse of buying) selling_pressure = 100 - buying_pressure # Market Sentiment (-1 to +1) market_sentiment = np.tanh(( smooth * 2 + (buy_ratio - 0.5) * 4 + cvd * 0.2 + whale * 0.1 )) return { "momentum_strength": float(momentum_strength), "volume_price_trend": float(vpt), "buying_pressure": float(buying_pressure), "selling_pressure": float(selling_pressure), "market_sentiment": float(market_sentiment), }