Add market microstructure features: Kyle's lambda, VPIN, Roll measure, OFI, Amihud

market_microstructure.py

@@ -0,0 +1,382 @@
+"""Market Microstructure Features
+
+Based on Marcos Lopez de Prado and the mlfinlab library.
+
+This is what separates retail technical analysis from institutional quant.
+Order flow, liquidity, and market impact contain genuine alpha.
+"""
+import numpy as np
+import pandas as pd
+from typing import Dict, List, Optional, Tuple
+import warnings
+warnings.filterwarnings('ignore')
+
+
+class MicrostructureFeatures:
+    """
+    Extract market microstructure features from tick-level data.
+    
+    Key insight: The market is not a continuous price stream.
+    It is a series of discrete transactions driven by informed vs.
+    uninformed traders. Microstructure features detect this.
+    """
+    
+    @staticmethod
+    def bid_ask_spread(bid: pd.Series, ask: pd.Series) -> pd.Series:
+        """
+        Raw bid-ask spread.
+        
+        Wider spreads = lower liquidity, higher execution cost.
+        """
+        return ask - bid
+    
+    @staticmethod
+    def relative_spread(bid: pd.Series, ask: pd.Series, 
+                        mid: Optional[pd.Series] = None) -> pd.Series:
+        """
+        Spread as percentage of mid price.
+        """
+        if mid is None:
+            mid = (bid + ask) / 2
+        return (ask - bid) / mid
+    
+    @staticmethod
+    def effective_spread(price: pd.Series, bid: pd.Series, 
+                         ask: pd.Series) -> pd.Series:
+        """
+        Effective spread = 2 * |trade_price - mid_price|.
+        
+        Measures actual execution cost vs. quoted spread.
+        """
+        mid = (bid + ask) / 2
+        return 2 * np.abs(price - mid) / mid
+    
+    @staticmethod
+    def realized_spread(price: pd.Series, bid: pd.Series, ask: pd.Series,
+                        future_mid: pd.Series) -> pd.Series:
+        """
+        Realized spread = 2 * |trade_price - future_mid|.
+        
+        Measures adverse selection. If realized spread > effective spread,
+        your trade moved the market against you.
+        """
+        mid = (bid + ask) / 2
+        return 2 * np.abs(price - future_mid) / mid
+    
+    @staticmethod
+    def price_impact(price: pd.Series, volume: pd.Series,
+                     bid: pd.Series, ask: pd.Series) -> pd.Series:
+        """
+        Kyle's Lambda — price impact coefficient.
+        
+        delta_price = lambda * signed_volume + noise
+        
+        Higher lambda = less liquid market, your orders move prices more.
+        """
+        mid = (bid + ask) / 2
+        mid_change = mid.diff()
+        
+        # Signed volume: Lee-Ready tick test
+        signed_vol = np.where(
+            price > mid.shift(1), volume,
+            np.where(price < mid.shift(1), -volume, 0)
+        )
+        
+        # Rolling regression via covariance/variance ratio
+        return pd.Series(signed_vol).rolling(100).cov(
+            pd.Series(mid_change).rolling(100)
+        ) / pd.Series(signed_vol).rolling(100).var().replace(0, np.nan)
+    
+    @staticmethod
+    def order_flow_imbalance(bid_size: pd.Series, ask_size: pd.Series) -> pd.Series:
+        """
+        OFI = (bid_size - ask_size) / (bid_size + ask_size).
+        
+        Positive = more buying pressure = bullish.
+        
+        This is genuine short-term alpha in liquid markets.
+        """
+        return (bid_size - ask_size) / (bid_size + ask_size + 1e-10)
+    
+    @staticmethod
+    def volume_imbalance(buy_volume: pd.Series, sell_volume: pd.Series) -> pd.Series:
+        """
+        Volume imbalance = (buy_vol - sell_vol) / (buy_vol + sell_vol).
+        
+        Classification via tick test or quote test.
+        """
+        return (buy_volume - sell_volume) / (buy_volume + sell_volume + 1e-10)
+    
+    @staticmethod
+    def trade_sign_classification(price: pd.Series, 
+                                   bid: pd.Series,
+                                   ask: pd.Series) -> pd.Series:
+        """
+        Lee-Ready tick test for trade direction classification.
+        
+        If trade price > mid → buy (aggressor is buyer)
+        If trade price < mid → sell (aggressor is seller)
+        If trade price = mid → use tick test (compare to previous trade)
+        """
+        mid = (bid + ask) / 2
+        
+        # Quote test
+        sign = np.where(price > mid, 1, np.where(price < mid, -1, 0))
+        
+        # Tick test for mid-trades
+        price_change = price.diff()
+        tick_sign = np.where(
+            price_change > 0, 1,
+            np.where(price_change < 0, -1, 0)
+        )
+        
+        # Use tick test where quote test is inconclusive
+        sign = np.where(sign == 0, tick_sign, sign)
+        
+        # If still 0, carry forward
+        sign = pd.Series(sign).fillna(method='ffill').fillna(0).values
+        
+        return pd.Series(sign, index=price.index)
+    
+    @staticmethod
+    def amihud_illiquidity(price: pd.Series, volume: pd.Series,
+                           window: int = 21) -> pd.Series:
+        """
+        Amihud illiquidity = |return| / (price * volume).
+        
+        Higher = less liquid. 
+        
+        Used in academic literature to measure market quality.
+        Predicts returns (illiquid stocks earn premium).
+        """
+        returns = price.pct_change().abs()
+        dollar_volume = price * volume
+        
+        return (returns / dollar_volume).rolling(window).mean() * 1e6
+    
+    @staticmethod
+    def kyles_lambda(price: pd.Series, volume: pd.Series,
+                     trade_sign: pd.Series, window: int = 100) -> pd.Series:
+        """
+        Kyle's Lambda — price impact per unit of order flow.
+        
+        Lambda = Cov(delta_price, signed_volume) / Var(signed_volume)
+        
+        Proxy for adverse selection and market depth.
+        """
+        delta_price = price.diff()
+        signed_volume = trade_sign * volume
+        
+        cov = delta_price.rolling(window).cov(signed_volume)
+        var = signed_volume.rolling(window).var()
+        
+        return cov / var.replace(0, np.nan)
+    
+    @staticmethod
+    def vpin_approximation(price: pd.Series, volume: pd.Series,
+                           bucket_vol: float = 10000) -> float:
+        """
+        VPIN — Volume-Synchronized Probability of Informed Trading.
+        
+        Simplified approximation using equal-volume buckets.
+        
+        High VPIN = high probability of informed trading = adverse selection risk.
+        """
+        # Classify trades
+        mid = price.rolling(2).mean()
+        trade_sign = np.where(price > mid.shift(1), 1, -1)
+        
+        signed_volume = trade_sign * volume
+        buy_volume = np.where(signed_volume > 0, volume, 0)
+        sell_volume = np.where(signed_volume < 0, volume, 0)
+        
+        # Create volume buckets
+        cumulative = np.cumsum(volume)
+        n_buckets = int(cumulative[-1] / bucket_vol)
+        
+        if n_buckets < 10:
+            return np.nan
+        
+        bucket_boundaries = np.linspace(0, cumulative[-1], n_buckets + 1)
+        
+        bucket_buy = []
+        bucket_sell = []
+        
+        for i in range(n_buckets):
+            mask = (cumulative >= bucket_boundaries[i]) & (cumulative < bucket_boundaries[i+1])
+            bucket_buy.append(np.sum(buy_volume[mask]))
+            bucket_sell.append(np.sum(sell_volume[mask]))
+        
+        bucket_buy = np.array(bucket_buy)
+        bucket_sell = np.array(bucket_sell)
+        bucket_volume = bucket_buy + bucket_sell
+        
+        # VPIN = average |buy - sell| / volume
+        vpin_values = np.abs(bucket_buy - bucket_sell) / (bucket_volume + 1e-10)
+        
+        return np.mean(vpin_values)
+    
+    @staticmethod
+    def roll_measure(price: pd.Series, window: int = 20) -> pd.Series:
+        """
+        Roll's measure — estimate bid-ask spread from serial covariance.
+        
+        Spread = 2 * sqrt(-Cov(delta_price_t, delta_price_{t-1}))
+        
+        Only valid when covariance is negative.
+        """
+        delta = price.diff()
+        cov = delta.rolling(window).cov(delta.shift(1))
+        
+        # Roll's measure
+        roll = 2 * np.sqrt(np.maximum(-cov, 0))
+        
+        return roll
+    
+    @staticmethod
+    def hasbrouck_lambda(price: pd.Series, volume: pd.Series,
+                         window: int = 100) -> pd.Series:
+        """
+        Hasbrouck's Lambda — information-based price impact.
+        
+        Measures how much of the price change is due to information
+        vs. liquidity demand.
+        """
+        # Simplified: correlation of returns with lagged signed volume
+        returns = price.pct_change()
+        trade_sign = np.sign(price.diff().fillna(0))
+        signed_volume = trade_sign * volume
+        
+        return returns.rolling(window).corr(signed_volume.shift(1))
+
+
+def compute_all_microstructure_features(df: pd.DataFrame) -> pd.DataFrame:
+    """
+    Compute all microstructure features from a tick DataFrame.
+    
+    Required columns: price, volume, bid, ask, bid_size, ask_size
+    """
+    required = ['price', 'volume', 'bid', 'ask', 'bid_size', 'ask_size']
+    for col in required:
+        if col not in df.columns:
+            raise ValueError(f"Missing required column: {col}")
+    
+    features = pd.DataFrame(index=df.index)
+    
+    # Basic spread
+    features['spread'] = MicrostructureFeatures.bid_ask_spread(df['bid'], df['ask'])
+    features['relative_spread'] = MicrostructureFeatures.relative_spread(
+        df['bid'], df['ask']
+    )
+    
+    # Effective spread
+    features['effective_spread'] = MicrostructureFeatures.effective_spread(
+        df['price'], df['bid'], df['ask']
+    )
+    
+    # Order flow imbalance
+    features['ofi'] = MicrostructureFeatures.order_flow_imbalance(
+        df['bid_size'], df['ask_size']
+    )
+    
+    # Trade sign classification
+    features['trade_sign'] = MicrostructureFeatures.trade_sign_classification(
+        df['price'], df['bid'], df['ask']
+    )
+    
+    # Signed volume
+    features['signed_volume'] = features['trade_sign'] * df['volume']
+    features['volume_imbalance'] = MicrostructureFeatures.volume_imbalance(
+        np.where(features['trade_sign'] > 0, df['volume'], 0),
+        np.where(features['trade_sign'] < 0, df['volume'], 0)
+    )
+    
+    # Amihud illiquidity (using daily approximation from intraday)
+    features['amihud_illiquidity'] = MicrostructureFeatures.amihud_illiquidity(
+        df['price'], df['volume']
+    )
+    
+    # Kyle's lambda
+    features['kyle_lambda'] = MicrostructureFeatures.kyles_lambda(
+        df['price'], df['volume'], features['trade_sign']
+    )
+    
+    # Roll's measure
+    features['roll_measure'] = MicrostructureFeatures.roll_measure(df['price'])
+    
+    # Hasbrouck lambda
+    features['hasbrouck_lambda'] = MicrostructureFeatures.hasbrouck_lambda(
+        df['price'], df['volume']
+    )
+    
+    # VPIN (computed once, broadcast)
+    vpin = MicrostructureFeatures.vpin_approximation(df['price'], df['volume'])
+    features['vpin'] = vpin
+    
+    return features.replace([np.inf, -np.inf], np.nan).fillna(method='ffill').fillna(0)
+
+
+def generate_synthetic_tick_data(n_ticks: int = 10000,
+                                  base_price: float = 100.0,
+                                  volatility: float = 0.001,
+                                  spread_bps: float = 1.0) -> pd.DataFrame:
+    """
+    Generate synthetic tick-level data for testing microstructure features.
+    """
+    np.random.seed(42)
+    
+    # Price process: random walk with slight mean reversion
+    prices = [base_price]
+    for _ in range(n_ticks - 1):
+        # Small random walk
+        change = np.random.randn() * volatility * base_price
+        # Mean reversion
+        change -= 0.01 * (prices[-1] - base_price)
+        prices.append(max(prices[-1] + change, 0.01))
+    
+    prices = np.array(prices)
+    
+    # Bid-ask spread
+    half_spread = prices * spread_bps / 20000  # bps to dollars
+    bid = prices - half_spread
+    ask = prices + half_spread
+    
+    # Sizes (power law: few large orders, many small)
+    bid_size = np.random.lognormal(8, 1.5, n_ticks).astype(int)
+    ask_size = np.random.lognormal(8, 1.5, n_ticks).astype(int)
+    
+    # Volume (trades happen at mid mostly)
+    volume = np.random.lognormal(6, 1.2, n_ticks).astype(int)
+    
+    # Timestamp
+    times = pd.date_range('2024-01-01 09:30', periods=n_ticks, freq='1s')
+    
+    return pd.DataFrame({
+        'timestamp': times,
+        'price': prices,
+        'bid': bid,
+        'ask': ask,
+        'bid_size': bid_size,
+        'ask_size': ask_size,
+        'volume': volume
+    }).set_index('timestamp')
+
+
+if __name__ == '__main__':
+    # Test microstructure features
+    tick_data = generate_synthetic_tick_data(n_ticks=5000)
+    features = compute_all_microstructure_features(tick_data)
+    
+    print("Market Microstructure Features")
+    print("=" * 60)
+    print(f"\nDataset: {len(tick_data)} ticks")
+    print(f"Features computed: {len(features.columns)}")
+    print(f"\nFeature Summary:")
+    print(features.describe().round(6))
+    
+    print(f"\nVPIN (Volume-Synchronized Probability of Informed Trading):")
+    print(f"  {features['vpin'].iloc[0]:.4f}")
+    
+    print(f"\nSample Features (last 5 ticks):")
+    print(features[['spread', 'relative_spread', 'ofi', 'kyle_lambda', 
+                     'amihud_illiquidity']].tail().round(6))