Add liquidity risk module - Amihud illiquidity, bid-ask bounce, price impact modeling

liquidity_risk.py

@@ -0,0 +1,164 @@
+"""liquidity_risk.py — Liquidity Risk Measurement & Price Impact
+
+Measures liquidity risk using Amihud illiquidity, bid-ask spread estimation,
+Kyle's lambda, and volume-synchronized probability of informed trading (VPIN).
+Essential for sizing positions in illiquid assets.
+
+References:
+- Amihud 2002: "Illiquidity and Stock Returns"
+- Kyle 1985: "Continuous Auctions and Insider Trading" (Kyle's lambda)
+- Easley et al. 2012: "Flow Toxicity and Liquidity in a High-Frequency World" (VPIN)
+- Goyenko et al. 2009: "Liquidity Risk and Expected Returns"
+"""
+import numpy as np, pandas as pd
+from scipy import stats
+
+class LiquidityRiskAnalyzer:
+    """Analyzes liquidity risk and price impact for position sizing."""
+
+    def __init__(self, window=63):
+        self.window = window
+
+    def amihud_illiquidity(self, prices, volumes):
+        """Amihud (2002): |return| / dollar volume.
+        
+        Higher = more illiquid.
+        """
+        ret = prices.pct_change().abs()
+        dv = prices * volumes
+        illiq = ret / (dv + 1e-10)
+        return illiq.rolling(self.window).mean()
+
+    def effective_spread(self, prices, volumes, tick_size=0.01):
+        """Estimate effective spread from Roll (1984) serial covariance.
+        
+        Spread = 2 * sqrt(-cov(Δp_t, Δp_{t-1}))
+        """
+        dp = prices.diff().dropna()
+        cov = dp.cov(dp.shift(1))
+        if cov >= 0: return tick_size  # No bounce detected
+        return 2 * np.sqrt(-cov)
+
+    def kyle_lambda(self, prices, volumes):
+        """Kyle's lambda: price impact per unit of order flow.
+        
+        Δp = λ * V + noise
+        Higher λ = less liquid (more impact).
+        """
+        dp = prices.diff().dropna()
+        v = volumes.reindex(dp.index).fillna(0)
+        valid = (dp != 0) & (v > 0)
+        if valid.sum() < 30: return 0.0
+        X = v[valid].values.reshape(-1, 1)
+        y = dp[valid].values
+        lambda_ = np.linalg.lstsq(X, y, rcond=None)[0][0]
+        return lambda_
+
+    def vpin(self, prices, volumes, bucket_size=50):
+        """Volume-Synchronized Probability of Informed Trading.
+        
+        VPIN ≈ 1 means toxic flow (informed traders dominating).
+        VPIN ≈ 0 means balanced flow.
+        """
+        ret = prices.pct_change().abs().dropna()
+        v = volumes.reindex(ret.index).fillna(0)
+        # Classify as buy/sell-initiated based on price direction
+        buy_vol = v.where(prices.diff() > 0, 0)
+        sell_vol = v.where(prices.diff() <= 0, 0)
+        # Volume bucket approach
+        total_vol = v.cumsum()
+        buckets = []
+        current = 0; buy_sum = 0; sell_sum = 0
+        for i in range(len(v)):
+            current += v.iloc[i]
+            buy_sum += buy_vol.iloc[i]
+            sell_sum += sell_vol.iloc[i]
+            if current >= bucket_size:
+                buckets.append(abs(buy_sum - sell_sum) / current)
+                current = buy_sum = sell_sum = 0
+        if buckets:
+            return np.mean(buckets)
+        return 0.0
+
+    def liquidity_adjusted_var(self, returns, illiquidity, confidence=0.05, holding_period=1):
+        """Liquidity-Adjusted VaR (LaVaR).
+        
+        Accounts for the probability of not being able to exit at VaR price.
+        """
+        r = returns.dropna()
+        var = np.percentile(r, confidence * 100)
+        # Liquidity cost component
+        liq_cost = illiquidity.dropna().mean() * holding_period
+        return float(var - liq_cost)
+
+    def position_capacity(self, prices, volumes, max_participation=0.05, max_impact_bps=50):
+        """Maximum position size given liquidity constraints.
+        
+        Args:
+            max_participation: max fraction of ADV
+            max_impact_bps: max acceptable market impact in basis points
+        """
+        adv = volumes.tail(20).mean()
+        price = prices.iloc[-1]
+        # Participation constraint
+        max_shares_part = adv * max_participation
+        # Impact constraint (square root law)
+        impact_frac = max_impact_bps / 10000.0
+        # Impact ≈ gamma * sigma * sqrt(participation)
+        sigma = prices.pct_change().tail(20).std()
+        # Solve for participation: impact_frac = sigma * sqrt(x) => x = (impact_frac/sigma)^2
+        if sigma > 0:
+            max_part_impact = (impact_frac / sigma) ** 2
+            max_shares_impact = adv * max_part_impact
+        else:
+            max_shares_impact = float('inf')
+        max_shares = min(max_shares_part, max_shares_impact)
+        max_notional = max_shares * price
+        return {
+            'max_shares': int(max_shares),
+            'max_notional': float(max_notional),
+            'adv': float(adv),
+            'participation_limit': float(max_participation),
+            'impact_limit': float(max_part_impact if sigma > 0 else 0),
+            'binding_constraint': 'participation' if max_shares_part < max_shares_impact else 'impact'
+        }
+
+    def report(self, prices, volumes):
+        """Full liquidity risk report."""
+        illiq = self.amihud_illiquidity(prices, volumes)
+        spread = self.effective_spread(prices, volumes)
+        kyle = self.kyle_lambda(prices, volumes)
+        vpin_val = self.vpin(prices, volumes)
+        capacity = self.position_capacity(prices, volumes)
+        ret = prices.pct_change().dropna()
+        lavar = self.liquidity_adjusted_var(ret, illiq)
+        return f"""## Liquidity Risk Report
+
+| Metric | Value | Interpretation |
+|--------|-------|--------------|
+| Amihud Illiquidity | {illiq.iloc[-1]*1e6:.2f} | {'Low' if illiq.iloc[-1]*1e6 < 0.1 else 'Medium' if illiq.iloc[-1]*1e6 < 1 else 'High'} |
+| Effective Spread (est.) | {spread*100:.3f}% | {'Tight' if spread < 0.001 else 'Wide'} |
+| Kyle's Lambda | {kyle:.6f} | {'Low impact' if abs(kyle) < 1e-6 else 'Medium' if abs(kyle) < 1e-5 else 'High impact'} |
+| VPIN | {vpin_val:.3f} | {'Balanced' if vpin_val < 0.3 else 'Elevated' if vpin_val < 0.6 else 'Toxic'} |
+| Liquidity-Adj VaR (5%) | {lavar*100:.2f}% | — |
+
+**Position Capacity:**
+
+| Constraint | Limit |
+|------------|-------|
+| Max Shares | {capacity['max_shares']:,} |
+| Max Notional | ${capacity['max_notional']:,.0f} |
+| ADV | {capacity['adv']:,.0f} |
+| Binding Constraint | {capacity['binding_constraint'].upper()} |
+
+**Recommendation:** {'✅ Liquid — standard sizing OK' if illiq.iloc[-1]*1e6 < 0.5 else '⚠️ Reduce position by 50%' if illiq.iloc[-1]*1e6 < 2 else '🔴 Highly illiquid — avoid or use limit orders'}
+"""
+
+if __name__ == '__main__':
+    np.random.seed(42)
+    prices = pd.Series(np.cumprod(1 + np.random.normal(0.0005, 0.015, 500)),
+                       index=pd.date_range('2022-01-01', periods=500, freq='B'))
+    volumes = pd.Series(np.random.lognormal(15, 0.5, 500),
+                        index=pd.date_range('2022-01-01', periods=500, freq='B'))
+    lra = LiquidityRiskAnalyzer()
+    print(lra.report(prices, volumes))