Kalshi_Polymarket/arb/calc.py

"""Arb math, ported from the workbook formulas.

Authoritative source: the *Arb Positions* sheet (it holds real Excel formulas;
the *Arb Scanner* sheet only holds script-computed values).

  Direction   C  = IF(SEARCH("MINMON",ticker), "Below", "Above")
  Basis %     AC = IF("Above",(kStrike-pStrike)/kStrike,(pStrike-kStrike)/pStrike)
  Kalshi fee  AI = (rate*size*price)*(1-price)   -> per-contract: rate*p*(1-p)
  Poly fee    AJ = size*rate*price*(1-price)     -> per-contract: rate*p*(1-p)
  Per leg     win  -> (1 - price) - fee          (L-AI in the sheet)
              lose -> (- price)   - fee          (-K-AI in the sheet)
  Min Gain    AA = MIN(W+X, Y+Z)   <- the true guaranteed P&L
  In Between  AD = X+Y             <- P&L in the strike-gap region
  % Return    AG = pnl / outlay
  Annualized  AH = AG * (365 / (close - open))

A clean arb pays $1 from exactly one leg only when BOTH contracts ask the same
question (same strike + direction). When strikes differ there is a price band
between the two strikes where the position can double-WIN (free money) or
double-LOSE (basis risk). The sheet surfaces this via AA and AD; we evaluate
all resolution regions and key off the guaranteed worst case, NOT (1 - cost).
"""
from datetime import date

# Kalshi market statuses that mean the market is no longer live/tradeable.
# A settled/finalized market still returns prices, but they are stale 1.0/0.0
# sentinels (e.g. a month-end MINMON whose strike already resolved) and must
# not be treated as a real quote.
_DEAD_KALSHI = {"finalized", "settled", "closed", "determined"}


def direction(ticker):
    return "Below" if "MINMON" in (ticker or "").upper() else "Above"


def basis_pct(kalshi_strike, poly_strike, direc):
    if not kalshi_strike or not poly_strike:
        return None
    if direc == "Above":
        return (kalshi_strike - poly_strike) / kalshi_strike
    return (poly_strike - kalshi_strike) / poly_strike


def _days_to(expiry_iso, today=None):
    if not expiry_iso:
        return None
    try:
        y, m, d = (int(x) for x in expiry_iso.split("-")[:3])
        delta = (date(y, m, d) - (today or date.today())).days
        return delta if delta > 0 else None
    except (ValueError, TypeError):
        return None


def _stmt_true(price, strike, is_above):
    return price >= strike if is_above else price <= strike


def _leg(price, fee, won):
    """Per-contract P&L for one leg. Mirrors L-AI / -K-AI in Arb Positions."""
    return ((1.0 - price) - fee) if won else ((-price) - fee)


def _scenarios(ks, ps, is_above):
    """Representative prices covering every resolution region.

    Returns list of (kalshi_stmt_true, poly_stmt_true). The mid region only
    exists when strikes differ; that region is the strike-gap / basis zone.
    """
    lo = min(ks, ps) * 0.5
    hi = max(ks, ps) * 1.5 + 1.0
    pts = [lo, hi]
    if ks != ps:
        pts.append((ks + ps) / 2.0)
    return [(_stmt_true(p, ks, is_above), _stmt_true(p, ps, is_above))
            for p in pts]


def evaluate(pair, kq, pq, settings, today=None):
    """Build one scanner row from a pair + its Kalshi/Poly quotes."""
    tkr = pair["kalshi_ticker"]
    direc = direction(tkr)
    ks = pair.get("kalshi_strike")
    ps = pair.get("poly_strike")
    bpct = basis_pct(ks, ps, direc)

    row = {
        "asset": pair.get("asset"),
        "kalshi_ticker": tkr,
        "kalshi_strike": ks,
        "poly_slug": pair.get("poly_slug"),
        "poly_strike": ps,
        "direction": direc,
        "basis_pct": bpct,
        "basis_favorable": None,
        "best_side": None,
        "kalshi_price": None, "kalshi_size": None,
        "poly_price": None, "poly_size": None,
        "combined_cost": None, "kalshi_fee": None, "poly_fee": None,
        "total_fee": None,
        "worst_pnl": None,        # guaranteed P&L / contract (sheet "Min Gain")
        "best_pnl": None,         # best-case P&L / contract (sheet "Max Gain")
        "mid_pnl": None,          # strike-gap P&L (sheet "In Between")
        "net_return": None,       # worst_pnl / combined_cost
        "annualized": None,
        "max_contracts": None, "total_gain": None,
        "poly_volume": (pq or {}).get("volume"),
        "days_to_expiry": None,
        "status": None,
        # Display metadata straight from the venues' APIs.
        "kalshi_title": (kq or {}).get("title"),
        "kalshi_rules": (kq or {}).get("rules"),
        "kalshi_yes_label": (kq or {}).get("yes_label"),
        "kalshi_no_label": (kq or {}).get("no_label"),
        "poly_question": (pq or {}).get("question"),
        "poly_description": (pq or {}).get("description"),
        "image": (pq or {}).get("image") or (pq or {}).get("icon"),
    }

    if not pair.get("poly_slug"):
        row["status"] = "NO PAIR"
        return row
    # Surface which side is missing — bare "NO DATA" hid the real cause
    # (commonly: Kalshi side liquid but the Polymarket slug doesn't resolve).
    if not ks or not ps:
        row["status"] = "NO DATA"
        return row
    if not kq and not pq:
        row["status"] = "NO DATA"
        return row
    if not pq:
        row["status"] = "NO POLY"
        return row
    if not kq or (kq.get("status") in _DEAD_KALSHI):
        # Settled/finalized Kalshi leg: prices are stale sentinels, not a quote.
        row["status"] = "NO KALSHI"
        return row

    kfee_rate = settings["kalshi_fee_rate"]
    pfee_rate = settings["poly_fee_rate"]
    is_above = (direc == "Above")
    scen = _scenarios(ks, ps, is_above)

    # Kalshi top-of-book size is only fetched for candidates; until then fall
    # back to open interest as a liquidity proxy for the size gate.
    k_oi = kq.get("open_interest")
    k_ysz = kq.get("yes_ask_size") if kq.get("yes_ask_size") is not None else k_oi
    k_nsz = kq.get("no_ask_size") if kq.get("no_ask_size") is not None else k_oi

    # Candidate hedged pairings: hold opposite sides across the two venues.
    cands = []
    if kq.get("yes_ask") and pq.get("no_ask"):
        cands.append(("YES+NO", "YES", kq["yes_ask"], k_ysz,
                      "NO", pq["no_ask"], pq.get("no_ask_size")))
    if kq.get("no_ask") and pq.get("yes_ask"):
        cands.append(("NO+YES", "NO", kq["no_ask"], k_nsz,
                      "YES", pq["yes_ask"], pq.get("yes_ask_size")))
    if not cands:
        # Both venues returned live quotes, but the opposite-side asks needed to
        # build a hedge aren't both offered (e.g. deep-OTM 'below' markets where
        # each venue only quotes the cheap YES side). There's data, just no arb.
        row["status"] = "NO ARB"
        return row

    best = None  # (worst_pnl, ...)
    for label, kside, kp, ksz, pside, pp, psz in cands:
        kfee = kfee_rate * kp * (1 - kp)
        pfee = pfee_rate * pp * (1 - pp)
        k_yes = (kside == "YES")
        p_yes = (pside == "YES")
        pnls = []
        for kt, pt in scen:
            kp_l = _leg(kp, kfee, kt == k_yes)
            pp_l = _leg(pp, pfee, pt == p_yes)
            pnls.append(kp_l + pp_l)
        worst = min(pnls)
        bestc = max(pnls)
        mid = pnls[2] if len(pnls) > 2 else None
        cand = (worst, bestc, mid, label, kside, kp, ksz, pside, pp, psz,
                kfee, pfee)
        if best is None or worst > best[0]:
            best = cand

    (worst, bestc, mid, label, kside, kp, ksz, pside, pp, psz,
     kfee, pfee) = best
    cost = kp + pp
    total_fee = kfee + pfee
    net_return = worst / cost if cost else None

    expiry = kq.get("expiry") or pq.get("end_date")
    days = _days_to(expiry, today)
    annualized = (net_return * (365.0 / days)
                  if (net_return is not None and days) else None)

    sizes = [s for s in (ksz, psz) if s is not None]
    max_contracts = min(sizes) if sizes else None
    total_gain = worst * max_contracts if max_contracts is not None else None

    # Favorable basis == strikes match, or the gap region is not a double-loss.
    fav = (ks == ps) or (mid is not None and mid >= -1e-9)

    row.update({
        "basis_favorable": fav,
        "best_side": label,
        "kalshi_price": kp, "kalshi_size": ksz,
        "poly_price": pp, "poly_size": psz,
        "combined_cost": cost, "kalshi_fee": kfee, "poly_fee": pfee,
        "total_fee": total_fee,
        "worst_pnl": worst, "best_pnl": bestc, "mid_pnl": mid,
        "net_return": net_return, "annualized": annualized,
        "max_contracts": max_contracts, "total_gain": total_gain,
        "days_to_expiry": days,
    })

    # Status precedence mirrors the workbook: DATA > BASIS > RETURN > SIZE.
    min_ret = settings["min_net_return"]
    min_vol = settings["min_poly_volume"]
    min_ct = settings["min_contracts"]
    has_edge = net_return is not None and net_return >= min_ret
    size_ok = (max_contracts is None or max_contracts >= min_ct) \
        and (row["poly_volume"] or 0) >= min_vol

    if not fav and not has_edge:
        row["status"] = "BAD BASIS"   # strike-gap double-loss kills it
    elif not has_edge:
        row["status"] = "NO ARB"      # no guaranteed edge
    elif not size_ok:
        row["status"] = "LOW SIZE"    # real edge, but untradeable size/volume
    else:
        row["status"] = "ARB"
    return row