"""Microstructure indicators: orderbook imbalance, slope, microprice. Tutte le funzioni accettano bids/asks come list[list[price, qty]] (formato standard dei ticker exchange) e ritornano metriche aggregate exchange-agnostic. """ from __future__ import annotations def orderbook_imbalance( bids: list[list[float]], asks: list[list[float]], depth: int = 10, ) -> dict[str, float | None]: """Imbalance ratio = (bid_vol - ask_vol) / (bid_vol + ask_vol) sui top-`depth` livelli. Range [-1, +1]. Positivo = bid pressure, negativo = ask pressure. Microprice (Stoll-Bertsimas): mid pesato dalla size opposta → P_micro = (P_bid * Q_ask + P_ask * Q_bid) / (Q_bid + Q_ask). Best level only. Slope: variazione cumulata di volume per unità di prezzo (proxy per liquidità in profondità). """ if not bids and not asks: return { "imbalance_ratio": None, "bid_volume": 0.0, "ask_volume": 0.0, "microprice": None, "bid_slope": None, "ask_slope": None, } top_bids = bids[:depth] top_asks = asks[:depth] bid_vol = sum(q for _, q in top_bids) ask_vol = sum(q for _, q in top_asks) total = bid_vol + ask_vol ratio = None if total == 0 else (bid_vol - ask_vol) / total # Microprice: best bid, best ask. Weighted by opposite-side size. microprice = None if top_bids and top_asks: bp, bq = top_bids[0] ap, aq = top_asks[0] denom = bq + aq if denom > 0: microprice = (bp * aq + ap * bq) / denom bid_slope = _depth_slope(top_bids, ascending_price=False) ask_slope = _depth_slope(top_asks, ascending_price=True) return { "imbalance_ratio": ratio, "bid_volume": bid_vol, "ask_volume": ask_vol, "microprice": microprice, "bid_slope": bid_slope, "ask_slope": ask_slope, } def _depth_slope(levels: list[list[float]], ascending_price: bool) -> float | None: """Calcola |Δq / Δp| sul primo vs penultimo livello. Slope alto = liquidità che crolla rapidamente in profondità (book sottile). """ if len(levels) < 2: return None p_first, q_first = levels[0] p_last, q_last = levels[-1] dp = abs(p_last - p_first) if dp == 0: return None return abs(q_first - q_last) / dp