PuGa/puga/saturation.py

"""Saturation model v1 (docs/saturation-design.md, after Opus review 2026-09-18). Pure functions.
Idea: N buildings is limited by (a) share of the market's steady traded flow and (b) the queue of
existing sell orders (supply days). Price is a haircut model, not a book walk.

is_thin() checks demand depth as an independent red flag (a genuinely dead bid book is worth
knowing about); n_out() does not multiply by it (see n_out's own docstring for why)."""
import math, statistics
from .history import trading_days

S_SHARE = 0.25     # max share of traded flow one producer takes
T_Q = 7            # days of flow the standing sell queue is compared against
T_D = 7            # days of flow standing demand should cover (used by is_thin only)
T_W = 3            # patient-selling window (days)


def tref(traded7: float, traded30: float) -> float:
    """Conservative daily flow: min(7d, 30d); with a Poisson lower bound when 30d volume is small.
    CAVEAT: the Poisson term treats each traded UNIT as an independent event, which overstates
    confidence for lumpy real trades (one 500-unit trade is one event, not 500) - see
    `tref_precise()` for the real-variance version, used once a candidate reaches stage 2 (this
    cheap version stays for stage-1 screening across every recipe, where fetching daily history
    for every ticker would be too expensive)."""
    t = min(traded7, traded30)
    if 0 < traded30 < 20:
        t = min(t, max(0.0, traded30 * (1 - 1.96 / math.sqrt(30 * traded30))))
    return max(0.0, t)


def tref_precise(tk: str, cx: str, traded7: float, traded30: float, days: int = 30) -> float:
    """Like tref(), but the lower bound comes from the REAL variance of daily traded volume
    (`puga.history.trading_days`) instead of a Poisson unit-independence assumption - one 500-unit
    trade is one data point here, not 500 independent ones. Falls back to tref() if there's too
    little real history to estimate variance from (<=2 trading days).

    `trading_days()` only returns days with a trade - zero-volume days are absent, not zero - so
    the volume list is padded with explicit zeros for the missing calendar days before computing
    mean/stdev, and `days` (not the trading-day count) is used as n. Averaging only over the rows
    that have a trade would compute the mean volume ON TRADING DAYS, overstating real
    per-CALENDAR-day flow by 1/(fraction of days traded) - worst precisely for the sparse tickers
    this function exists to protect."""
    td = trading_days(tk, cx, days)
    if len(td) <= 2:
        return tref(traded7, traded30)
    vols = [v for _, v in td] + [0.0] * (days - len(td))
    mean = statistics.mean(vols)
    sd = statistics.stdev(vols)
    bound = max(0.0, mean - 1.96 * sd / math.sqrt(days))
    return max(0.0, min(tref(traded7, traded30), bound))


def max_units_at_impact(bids: list[tuple[float, float]], ref_price: float, max_impact_pct: float = 10.0) -> float:
    """Real, ticker-specific price-impact cap, replacing the flat S_SHARE guess for THIS check:
    how many units could be sold into the LIVE bid book before the price paid degrades more than
    `max_impact_pct` below `ref_price`. Walks bids highest-first (puga.market.walk's own
    mechanics) and stops at the first tier priced below the threshold.

    A day-over-day price-vs-volume regression does not give a usable general elasticity estimate
    here: volume and price co-move mostly from DEMAND bursts (more buyers -> more volume AND
    steady/higher price), which swamps and often reverses the sign of the SUPPLY-side effect
    actually wanted (does MY added output push price down). Walking the real book sidesteps that
    confounding, but only captures the static liquidity snapshot, not how the market reprices
    over weeks of sustained extra output (nothing in available data answers that cleanly). See
    docs/decisions.md."""
    if not ref_price or ref_price <= 0:
        return 0.0
    threshold = ref_price * (1 - max_impact_pct / 100)
    cum = 0.0
    for price, qty in sorted(bids, key=lambda x: -x[0]):
        if price < threshold:
            break
        cum += qty
    return cum


def is_thin(tr: float, demand: float, q_out: float) -> bool:
    return tr < 3 * q_out or demand < 7 * q_out


def n_out(tr: float, supply: float, q_out: float) -> float:
    """Buildings the output market absorbs: flow share x queue penalty. 0 if no flow.

    Deliberately has NO resting-bid-depth term: production sells via patient ASKS (`p_patient()`)
    that get taken by buyers crossing the spread, including walk-up buyers with no resting bid at
    all - a thin resting bid book doesn't mean they won't show up, since `tr` (real traded flow,
    already in `base`) is the actual evidence they do. A market with healthy ask-side flow and
    thin standing bids is normal (most buyers don't leave a bid sitting around), not evidence of
    absent demand. Real resting-bid demand is still checked independently by `is_thin()` as a red
    flag, not folded into this score."""
    if tr <= 0 or q_out <= 0:
        return 0.0
    base = S_SHARE * tr / q_out
    queue = min(1.0, T_Q * tr / supply) if supply > 0 else 1.0
    return base * queue


def p_patient(asks: list[tuple[float, float]], tr: float, produced_per_day: float, bid: float, vwap7: float | None,
              vwap30: float | None, ask: float | None, wide_high: float | None = None) -> float:
    """Price we can hold asks at. asks = [(price, units)] ascending. Find the highest ask level whose units-ahead
    (levels strictly below) <= what buyers will absorb of the queue over T_W days once our output is counted:
    target = T_W * (tr - produced). target <= 0 => sell at the bid. Clamped to [bid, min(vwap7, vwap30, ask, wide_high)]."""
    hi = min(x for x in (vwap7, vwap30, ask, wide_high) if x)
    target = T_W * (tr - produced_per_day)
    if target <= 0:
        return bid
    p, ahead = hi, 0.0
    for price, units in asks:
        if ahead <= target:
            p = price
        ahead += units
    return max(bid, min(p, hi))


def in_band(price: float, band: tuple[float | None, float | None] | None) -> bool:
    """band = (low, high), the exchange's tradeable Price Band (FIO order_book's Narrow/WidePriceBandLow/High;
    verified live 2026-09-25 to match APEX's own displayed 'Price Band' field exactly). An order outside it is
    not just uncompetitive, it is not currently placeable at all, and existing ones there are stale artifacts
    (e.g. a leftover 334 AIC bid sitting under a 663 floor) that will never fill; not just a heuristic cutoff."""
    if not band:
        return True
    lo, hi = band
    return (lo is None or price >= lo) and (hi is None or price <= hi)


def effective_supply(asks: list[tuple[float, float]], ref_price: float | None, mult: float = 1.25,
                     band: tuple[float | None, float | None] | None = None) -> float:
    """Units of standing sell orders that actually compete: in the tradeable price band (hard filter, if given)
    AND price <= mult x reference (vwap7 or ask; soft filter for "active competition" vs merely valid-but-stale
    high asks like OCK's top tier). asks = [(price, units)]."""
    asks = [(p, u) for p, u in asks if in_band(p, band)]
    if not ref_price:
        return sum(u for _, u in asks)
    return sum(u for p, u in asks if p <= mult * ref_price)


def effective_demand(bids: list[tuple[float, float]], band: tuple[float | None, float | None] | None) -> float:
    """Units of standing buy orders that are actually tradeable: within the exchange's price band. Filters out
    stale/abandoned bids placed under a different band or long since left behind by the market (e.g. a 334 AIC
    bid when the floor is 663) rather than genuine, fillable demand. bids = [(price, units)]."""
    return sum(u for p, u in bids if in_band(p, band))