Stock Liquidity

import subprocess, syssubprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "yfinance", "pandas", "numpy"])

import yfinance as yfimport pandas as pdimport numpy as np
def liquidity_dashboard(ticker_symbol, period="3mo"):    ticker = yf.Ticker(ticker_symbol)    info = ticker.info    hist = ticker.history(period=period)
    if hist.empty:        return None
    # --- Spread metrics (from current quote) ---    bid = info.get("bid", None)    ask = info.get("ask", None)    current_price = info.get("currentPrice") or info.get("regularMarketPrice") or hist["Close"].iloc[-1]
    spread = None    spread_pct = None    if bid and ask and bid > 0 and ask > 0:        spread = round(ask - bid, 4)        midpoint = (ask + bid) / 2        spread_pct = round((spread / midpoint) * 100, 4)
    # --- Volume metrics ---    avg_volume = hist["Volume"].mean()    median_volume = hist["Volume"].median()    avg_dollar_volume = (hist["Close"] * hist["Volume"]).mean()    volume_std = hist["Volume"].std()    volume_cv = volume_std / avg_volume if avg_volume > 0 else None  # coefficient of variation
    # --- Turnover ratio ---    shares_outstanding = info.get("sharesOutstanding", None)    float_shares = info.get("floatShares", None)    base_shares = float_shares or shares_outstanding    turnover_ratio = round(avg_volume / base_shares, 6) if base_shares else None
    # --- Amihud illiquidity ratio ---    # Average of |daily return| / daily dollar volume    returns = hist["Close"].pct_change().dropna()    dollar_volume = (hist["Close"] * hist["Volume"]).iloc[1:]  # align with returns    amihud_values = returns.abs() / dollar_volume    amihud = amihud_values[amihud_values.replace([np.inf, -np.inf], np.nan).notna()].mean()
    # --- Market impact estimate (square-root model) ---    # For a hypothetical order of 1% of ADV    adv = avg_volume    order_size = adv * 0.01    daily_volatility = returns.std()    sigma = daily_volatility    participation_rate = order_size / adv if adv > 0 else 0    impact_bps = sigma * np.sqrt(participation_rate) * 10000  # in basis points
    return {        "ticker": ticker_symbol,        "current_price": round(current_price, 2),        "bid": bid,        "ask": ask,        "spread": spread,        "spread_pct": spread_pct,        "avg_daily_volume": int(avg_volume),        "median_daily_volume": int(median_volume),        "avg_dollar_volume": round(avg_dollar_volume, 0),        "volume_cv": round(volume_cv, 3) if volume_cv else None,        "shares_outstanding": shares_outstanding,        "float_shares": float_shares,        "turnover_ratio": turnover_ratio,        "amihud_illiquidity": round(amihud * 1e9, 4) if not np.isnan(amihud) else None,        "daily_volatility": round(daily_volatility * 100, 2),        "impact_1pct_adv_bps": round(impact_bps, 2),        "observations": len(hist),    }

import yfinance as yf
def spread_analysis(ticker_symbol):    ticker = yf.Ticker(ticker_symbol)    info = ticker.info
    bid = info.get("bid", 0)    ask = info.get("ask", 0)    bid_size = info.get("bidSize", None)    ask_size = info.get("askSize", None)    current_price = info.get("currentPrice") or info.get("regularMarketPrice", 0)
    result = {"bid": bid, "ask": ask, "bid_size": bid_size, "ask_size": ask_size}
    if bid > 0 and ask > 0:        midpoint = (bid + ask) / 2        result["absolute_spread"] = round(ask - bid, 4)        result["relative_spread_pct"] = round((ask - bid) / midpoint * 100, 4)        result["relative_spread_bps"] = round((ask - bid) / midpoint * 10000, 2)    return result

import yfinance as yfimport pandas as pdimport numpy as np
def volume_analysis(ticker_symbol, period="3mo"):    ticker = yf.Ticker(ticker_symbol)    hist = ticker.history(period=period)
    if hist.empty:        return None
    vol = hist["Volume"]    close = hist["Close"]    dollar_vol = vol * close
    # Relative volume (today vs average)    rvol = vol.iloc[-1] / vol.mean() if vol.mean() > 0 else None
    # Volume trend (linear regression slope over the period)    x = np.arange(len(vol))    slope, _ = np.polyfit(x, vol.values, 1) if len(vol) > 1 else (0, 0)    trend_pct = (slope * len(vol)) / vol.mean() * 100  # % change over period
    # Volume profile by day of week    hist_copy = hist.copy()    hist_copy["DayOfWeek"] = hist_copy.index.dayofweek    day_names = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu", 4: "Fri"}    vol_by_day = hist_copy.groupby("DayOfWeek")["Volume"].mean()    vol_by_day.index = vol_by_day.index.map(day_names)
    # High/low volume days    high_vol_days = hist.nlargest(5, "Volume")[["Close", "Volume"]]    low_vol_days = hist.nsmallest(5, "Volume")[["Close", "Volume"]]
    return {        "avg_volume": int(vol.mean()),        "median_volume": int(vol.median()),        "avg_dollar_volume": round(dollar_vol.mean(), 0),        "current_volume": int(vol.iloc[-1]),        "relative_volume": round(rvol, 2) if rvol else None,        "volume_trend_pct": round(trend_pct, 1),        "volume_by_day": vol_by_day.to_dict(),        "high_vol_days": high_vol_days,        "low_vol_days": low_vol_days,        "max_volume": int(vol.max()),        "min_volume": int(vol.min()),    }

import yfinance as yfimport numpy as np
def market_impact(ticker_symbol, order_shares=None, order_dollars=None, period="3mo"):    ticker = yf.Ticker(ticker_symbol)    hist = ticker.history(period=period)    info = ticker.info
    if hist.empty:        return None
    current_price = info.get("currentPrice") or hist["Close"].iloc[-1]    avg_volume = hist["Volume"].mean()    daily_volatility = hist["Close"].pct_change().dropna().std()
    # Determine order size in shares    if order_dollars and not order_shares:        order_shares = order_dollars / current_price    elif not order_shares:        # Default: estimate for various sizes        order_shares = avg_volume * 0.01  # 1% of ADV
    participation_rate = order_shares / avg_volume if avg_volume > 0 else 0    pct_adv = (order_shares / avg_volume * 100) if avg_volume > 0 else 0
    # Square-root impact model    impact_pct = daily_volatility * np.sqrt(participation_rate) * 100    impact_bps = impact_pct * 100    impact_dollars = impact_pct / 100 * current_price * order_shares
    # Generate impact curve for multiple order sizes    sizes = [0.001, 0.005, 0.01, 0.02, 0.05, 0.10, 0.20, 0.50]  # as fraction of ADV    curve = []    for s in sizes:        q = avg_volume * s        imp = daily_volatility * np.sqrt(s) * 100        curve.append({            "pct_adv": round(s * 100, 1),            "shares": int(q),            "dollars": round(q * current_price, 0),            "impact_bps": round(imp * 100, 1),            "impact_dollars_per_share": round(imp / 100 * current_price, 4),        })
    return {        "ticker": ticker_symbol,        "current_price": round(current_price, 2),        "avg_daily_volume": int(avg_volume),        "daily_volatility_pct": round(daily_volatility * 100, 2),        "order_shares": int(order_shares),        "order_dollars": round(order_shares * current_price, 0),        "pct_of_adv": round(pct_adv, 2),        "estimated_impact_bps": round(impact_bps, 1),        "estimated_impact_pct": round(impact_pct, 4),        "estimated_impact_total_dollars": round(impact_dollars, 2),        "impact_curve": curve,    }

import yfinance as yfimport pandas as pdimport numpy as np
def turnover_analysis(ticker_symbol, period="3mo"):    ticker = yf.Ticker(ticker_symbol)    hist = ticker.history(period=period)    info = ticker.info
    if hist.empty:        return None
    avg_volume = hist["Volume"].mean()    shares_outstanding = info.get("sharesOutstanding")    float_shares = info.get("floatShares")
    result = {        "avg_daily_volume": int(avg_volume),        "shares_outstanding": shares_outstanding,        "float_shares": float_shares,    }
    if shares_outstanding:        daily_turnover = avg_volume / shares_outstanding        result["daily_turnover_ratio"] = round(daily_turnover, 6)        result["annualized_turnover"] = round(daily_turnover * 252, 2)        result["days_to_trade_float"] = round(            (float_shares or shares_outstanding) / avg_volume, 1        ) if avg_volume > 0 else None
    if float_shares:        float_turnover = avg_volume / float_shares        result["float_turnover_daily"] = round(float_turnover, 6)        result["float_turnover_annualized"] = round(float_turnover * 252, 2)
    # Turnover trend    vol = hist["Volume"]    base = float_shares or shares_outstanding    if base:        hist_copy = hist.copy()        hist_copy["turnover"] = hist_copy["Volume"] / base        recent_turnover = hist_copy["turnover"].tail(20).mean()        older_turnover = hist_copy["turnover"].head(20).mean()        if older_turnover > 0:            result["turnover_trend_pct"] = round(                (recent_turnover - older_turnover) / older_turnover * 100, 1            )
    return result