scripts/jito_tip_observer.py

#!/usr/bin/env python3
"""
jito_tip_observer.py

READ-ONLY. Fetches Jito's public tip floor / percentile data so you can see
the real current tip market instead of inferring it from your own bot's
occasional 99th-percentile-floor numbers.

Does NOT submit bundles, does NOT touch a wallet, does NOT place trades.

Usage:
    python3 jito_tip_observer.py
    python3 jito_tip_observer.py --watch 30    # poll every 30s, Ctrl+C to stop

Data source: Jito Labs' public bundle tip-floor API
(https://bundles.jito.wtf / mainnet.block-engine.jito.wtf), which publishes
rolling percentile tip stats across recent landed bundles. If Jito changes
or retires this endpoint, this script will fail loudly rather than silently
returning fake data -- check https://docs.jito.wtf for the current URL.
"""

import argparse
import json
import sys
import time
import urllib.request
import urllib.error

TIP_FLOOR_URL = "https://bundles.jito.wtf/api/v1/bundles/tip_floor"

LAMPORTS_PER_SOL = 1_000_000_000


def fetch_tip_floor():
    req = urllib.request.Request(TIP_FLOOR_URL, headers={"User-Agent": "tip-observer/1.0"})
    with urllib.request.urlopen(req, timeout=10) as resp:
        raw = resp.read()
    return json.loads(raw)


def fmt_sol(lamports_val: float) -> str:
    return f"{lamports_val / LAMPORTS_PER_SOL:.9f} SOL"


def print_snapshot(data):
    # The API returns a list with one summary object; handle both shapes defensively.
    if isinstance(data, list):
        if not data:
            print("Empty response from Jito tip-floor API.")
            return
        row = data[0]
    else:
        row = data

    print("-" * 70)
    print(f"Jito tip floor snapshot  (fetched {time.strftime('%Y-%m-%d %H:%M:%S UTC', time.gmtime())})")
    print("-" * 70)

    # Field names per Jito's public API as of last known schema; print raw
    # keys too so this stays useful if the schema shifts slightly.
    interesting = {
        "landed_tips_25th_percentile": "25th pct",
        "landed_tips_50th_percentile": "50th pct (median)",
        "landed_tips_75th_percentile": "75th pct",
        "landed_tips_95th_percentile": "95th pct",
        "landed_tips_99th_percentile": "99th pct",
        "ema_landed_tips_50th_percentile": "EMA 50th pct",
    }

    any_matched = False
    for key, label in interesting.items():
        if key in row:
            any_matched = True
            val_sol = row[key]  # Jito reports these in SOL already
            lamports = round(val_sol * LAMPORTS_PER_SOL)
            print(f"  {label:22s}: {lamports:>10d} lamports  ({val_sol:.9f} SOL)")

    if not any_matched:
        print("  (schema didn't match expected fields -- raw response below)")
        print(json.dumps(row, indent=2))
        return

    print()
    p99 = row.get("landed_tips_99th_percentile")
    p50 = row.get("landed_tips_50th_percentile")
    if p99 and p50:
        print(f"  Note: to reliably land, you're historically bidding near the "
              f"99th pct ({round(p99*LAMPORTS_PER_SOL)} lamports), "
              f"~{p99/p50:.1f}x the median.")
    print("  Compare this to your bot's actual profit-per-opportunity to see if")
    print("  the tip market is even clearable at your trade size right now.")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--watch", type=int, default=0,
                     help="Poll repeatedly every N seconds instead of once")
    args = ap.parse_args()

    def one_shot():
        try:
            data = fetch_tip_floor()
        except urllib.error.URLError as e:
            print(f"Failed to reach Jito tip-floor API: {e}", file=sys.stderr)
            print("Check network access / whether the endpoint has moved "
                  "(see https://docs.jito.wtf).", file=sys.stderr)
            sys.exit(1)
        except json.JSONDecodeError as e:
            print(f"Got a response but couldn't parse JSON: {e}", file=sys.stderr)
            sys.exit(1)
        print_snapshot(data)

    if args.watch > 0:
        try:
            while True:
                one_shot()
                print()
                time.sleep(args.watch)
        except KeyboardInterrupt:
            print("\nStopped.")
    else:
        one_shot()


if __name__ == "__main__":
    main()