import uuid import time from typing import Dict, Any, List, Optional from config import settings class FootprintCluster: def __init__(self, tick_size: float = 1.0): self.tick_size = tick_size self.cluster_id = str(uuid.uuid4()) self.status = "active" self.open_time: Optional[int] = None self.close_reason: Optional[str] = None self.open_price: Optional[float] = None self.close_price: Optional[float] = None self.high: Optional[float] = None self.low: Optional[float] = None self.poc: Optional[float] = None self.total_delta: float = 0.0 self.total_volume: float = 0.0 self.total_ticks: int = 0 self.close_time: Optional[int] = None # levels: price_float -> { 'ask': float, 'bid': float } # internally we keep float keys to make sorting and arithmetic easy self.levels: Dict[float, Dict[str, float]] = {} self.stacked: Dict[str, Any] = { "buy": False, "sell": False, "price_range": [] } self.advanced_metrics: Dict[str, Any] = { "poc_position": None, "pattern": None, "top_extreme": None, "bottom_extreme": None } def add_tick(self, price: float, volume: float, is_buy: bool, timestamp_msc: int) -> None: if self.open_time is None: self.open_time = timestamp_msc self.open_price = price self.close_price = price self.close_time = timestamp_msc self.total_ticks += 1 # Round price to the nearest tick_size to avoid float precision issues rounded_price = round(price / self.tick_size) * self.tick_size if rounded_price not in self.levels: self.levels[rounded_price] = {"ask": 0.0, "bid": 0.0} if is_buy: self.levels[rounded_price]["ask"] += volume else: self.levels[rounded_price]["bid"] += volume # Update High/Low if self.high is None or rounded_price > self.high: self.high = rounded_price if self.low is None or rounded_price < self.low: self.low = rounded_price # Recalculate totals, POC, imbalances and stacked imbalances self._recalculate() def _recalculate(self) -> None: if not self.levels: return self.total_volume = 0.0 self.total_delta = 0.0 # First pass: calc totals and delta per level for price, data in self.levels.items(): ask = data["ask"] bid = data["bid"] lvl_delta = ask - bid lvl_total = ask + bid self.total_volume += lvl_total self.total_delta += lvl_delta # Find POC: level with the highest total volume. # Tie-breaker: choose the highest price level. sorted_prices = sorted(self.levels.keys()) best_price = sorted_prices[0] max_total = -1.0 for price in sorted_prices: total_vol = self.levels[price]["ask"] + self.levels[price]["bid"] if total_vol > max_total: max_total = total_vol best_price = price elif total_vol == max_total: if price > best_price: best_price = price self.poc = best_price # Second pass: calculate imbalances diagonal and stacked imbalances # To avoid division by zero: # imbalance_buy[i] is True if ask_vol[i] >= R * bid_vol[i - tick_size] # imbalance_sell[i] is True if bid_vol[i] >= R * ask_vol[i + tick_size] R = settings.IMBALANCE_RATIO imbalances_buy = {} imbalances_sell = {} for price in sorted_prices: ask_val = self.levels[price]["ask"] bid_val = self.levels[price]["bid"] # Lower level price lower_price = round((price - self.tick_size) / self.tick_size) * self.tick_size bid_below = self.levels[lower_price]["bid"] if lower_price in self.levels else 0.0 # Upper level price upper_price = round((price + self.tick_size) / self.tick_size) * self.tick_size ask_above = self.levels[upper_price]["ask"] if upper_price in self.levels else 0.0 # Buy Imbalance (diagonal): ask vs bid_below # Avoid triggering on 0 vs 0 if ask_val > 0 and ask_val >= R * bid_below: imbalances_buy[price] = True else: imbalances_buy[price] = False # Sell Imbalance (diagonal): bid vs ask_above if bid_val > 0 and bid_val >= R * ask_above: imbalances_sell[price] = True else: imbalances_sell[price] = False # Detect stacked imbalances: 3+ consecutive levels with imbalance in the same direction # Let's check contiguous price levels in steps of tick_size stacked_buy = False stacked_sell = False stacked_buy_prices = [] stacked_sell_prices = [] min_consecutive = settings.STACKED_MIN_COUNT # We need to check all possible price steps from low to high if self.low is not None and self.high is not None: current_price = self.low consec_buy = [] consec_sell = [] while current_price <= self.high: rounded_p = round(current_price / self.tick_size) * self.tick_size # Check Buy if imbalances_buy.get(rounded_p, False): consec_buy.append(rounded_p) else: if len(consec_buy) >= min_consecutive: stacked_buy = True stacked_buy_prices.extend(consec_buy) consec_buy = [] # Check Sell if imbalances_sell.get(rounded_p, False): consec_sell.append(rounded_p) else: if len(consec_sell) >= min_consecutive: stacked_sell = True stacked_sell_prices.extend(consec_sell) consec_sell = [] current_price += self.tick_size # Final check at the end of the loop if len(consec_buy) >= min_consecutive: stacked_buy = True stacked_buy_prices.extend(consec_buy) if len(consec_sell) >= min_consecutive: stacked_sell = True stacked_sell_prices.extend(consec_sell) # Build output properties for stacked self.stacked = { "buy": stacked_buy, "sell": stacked_sell, "price_range": sorted(list(set(stacked_buy_prices + stacked_sell_prices))) } # Store imbalances back in levels for JSON output for price in sorted_prices: imb = None if imbalances_buy.get(price, False) and imbalances_sell.get(price, False): imb = "both" elif imbalances_buy.get(price, False): imb = "buy" elif imbalances_sell.get(price, False): imb = "sell" self.levels[price]["delta"] = self.levels[price]["ask"] - self.levels[price]["bid"] self.levels[price]["total"] = self.levels[price]["ask"] + self.levels[price]["bid"] self.levels[price]["imbalance"] = imb # Advanced Metrics Calculation if self.high is not None and self.low is not None and self.high > self.low and self.total_volume > 0: # 1. POC Position poc_percent = (self.poc - self.low) / (self.high - self.low) if poc_percent >= 0.65: self.advanced_metrics["poc_position"] = "top" elif poc_percent <= 0.35: self.advanced_metrics["poc_position"] = "bottom" else: self.advanced_metrics["poc_position"] = "middle" # 2. P and B Patterns mid_price = (self.high + self.low) / 2.0 vol_above = sum(self.levels[p]["total"] for p in sorted_prices if p >= mid_price) vol_below = sum(self.levels[p]["total"] for p in sorted_prices if p < mid_price) if vol_above / self.total_volume > 0.65: self.advanced_metrics["pattern"] = "P" elif vol_below / self.total_volume > 0.65: self.advanced_metrics["pattern"] = "B" else: self.advanced_metrics["pattern"] = "normal" # 3. Extremes (Exhaustion / Absorption) top_vol = self.levels[self.high]["total"] next_top_price = round((self.high - self.tick_size) / self.tick_size) * self.tick_size next_top_vol = self.levels.get(next_top_price, {}).get("total", 0.0) avg_vol = self.total_volume / len(self.levels) if top_vol < avg_vol * 0.2 and next_top_vol > avg_vol * 0.5: self.advanced_metrics["top_extreme"] = "exhaustion" elif top_vol > avg_vol * 2.5: self.advanced_metrics["top_extreme"] = "absorption" else: self.advanced_metrics["top_extreme"] = "normal" bottom_vol = self.levels[self.low]["total"] next_bot_price = round((self.low + self.tick_size) / self.tick_size) * self.tick_size next_bot_vol = self.levels.get(next_bot_price, {}).get("total", 0.0) if bottom_vol < avg_vol * 0.2 and next_bot_vol > avg_vol * 0.5: self.advanced_metrics["bottom_extreme"] = "exhaustion" elif bottom_vol > avg_vol * 2.5: self.advanced_metrics["bottom_extreme"] = "absorption" else: self.advanced_metrics["bottom_extreme"] = "normal" # 4. Ratios (Exhaustion/Absorption quantification at extremes) if next_top_vol > 0: self.advanced_metrics["high_ratio"] = round(top_vol / next_top_vol, 2) else: self.advanced_metrics["high_ratio"] = round(top_vol, 2) if top_vol > 0 else 0.0 if next_bot_vol > 0: self.advanced_metrics["low_ratio"] = round(bottom_vol / next_bot_vol, 2) else: self.advanced_metrics["low_ratio"] = round(bottom_vol, 2) if bottom_vol > 0 else 0.0 # 5. Delta Divergence (Price direction vs Order Flow Delta) self.advanced_metrics["delta_divergence"] = False if self.open_price is not None and self.close_price is not None: is_bull = self.close_price > self.open_price is_bear = self.close_price < self.open_price if (is_bull and self.total_delta < 0) or (is_bear and self.total_delta > 0): self.advanced_metrics["delta_divergence"] = True def should_close(self, current_time_msc: int) -> Optional[str]: mode = settings.CLUSTER_CLOSE_MODE # Volume safety cap — always applied (prevents runaway cluster if market halts) if self.total_volume >= settings.CLUSTER_VOLUME_MAX: return "volume" # Primary closing condition based on configured mode if mode == "delta": if abs(self.total_delta) >= settings.CLUSTER_DELTA_MAX: return "delta" elif mode == "range": if self.high is not None and self.low is not None: if (self.high - self.low) >= settings.CLUSTER_RANGE_POINTS: return "range" elif mode == "time": if self.open_time is not None: elapsed = (current_time_msc - self.open_time) / 1000.0 if elapsed >= settings.CLUSTER_TIME_SECONDS: return "time" elif mode == "volume": if self.total_volume >= settings.CLUSTER_VOLUME_MAX: return "volume" return None def close(self, reason: str) -> None: self.status = "closed" self.close_reason = reason def to_json(self) -> Dict[str, Any]: # Convert keys in levels to string for JSON compatibility levels_str = {} for price, data in sorted(self.levels.items(), reverse=True): levels_str[f"{price:.5f}"] = { "ask": float(data.get("ask", 0.0)), "bid": float(data.get("bid", 0.0)), "delta": float(data.get("delta", 0.0)), "total": float(data.get("total", 0.0)), "imbalance": data.get("imbalance", None) } return { "cluster_id": self.cluster_id, "tick_size": self.tick_size, "status": self.status, "open_time": int(self.open_time) if self.open_time is not None else None, "close_reason": self.close_reason, "open_price": float(self.open_price) if self.open_price is not None else None, "close_price": float(self.close_price) if self.close_price is not None else None, "high": float(self.high) if self.high is not None else None, "low": float(self.low) if self.low is not None else None, "poc": float(self.poc) if self.poc is not None else None, "close_time": int(self.close_time) if self.close_time is not None else None, "total_delta": float(self.total_delta), "total_volume": float(self.total_volume), "total_ticks": int(self.total_ticks), "bar_volume": None, "levels": levels_str, "stacked": { "buy": bool(self.stacked.get("buy", False)), "sell": bool(self.stacked.get("sell", False)), "price_range": [float(p) for p in self.stacked.get("price_range", [])] }, "advanced_metrics": self.advanced_metrics } class Aggregator: def __init__(self, tick_size: float = 1.0): self.tick_size = tick_size self.active_cluster = FootprintCluster(tick_size=self.tick_size) self.history: List[Dict[str, Any]] = [] def process_tick(self, price: float, volume: float, is_buy: bool, timestamp_msc: int) -> tuple[Dict[str, Any], Optional[Dict[str, Any]]]: """ Process a single tick. Returns: (active_cluster_json, closed_cluster_json_if_just_closed) """ self.active_cluster.add_tick(price, volume, is_buy, timestamp_msc) # Check if the active cluster should be closed close_reason = self.active_cluster.should_close(timestamp_msc) closed_json = None if close_reason: self.active_cluster.close(close_reason) closed_json = self.active_cluster.to_json() # Save to history self.history.append(closed_json) if len(self.history) > settings.HISTORY_BUFFER_SIZE: self.history.pop(0) # Start new cluster self.active_cluster = FootprintCluster(tick_size=self.tick_size) # The next tick will set the open_time of the new cluster. return self.active_cluster.to_json(), closed_json def classify_tick(last: float, bid: float, ask: float, flags: int) -> bool: """ Classify tick aggressiveness. TICK_FLAG_BUY = 32 -> BUY (True) TICK_FLAG_SELL = 64 -> SELL (False) TICK_FLAG_ASK = 4 -> ASK (True) TICK_FLAG_BID = 2 -> BID (False) """ if flags & 32: return True elif flags & 64: return False if last > 0: if last >= ask: return True elif last <= bid: return False # Forex quote ticks fallback if flags & 4: return True elif flags & 2: return False # Standard fallback if between spread: closer to ask is BUY if ask > bid and last > 0: return (last - bid) >= (ask - last) return True