"""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))