modules/jito_tip_engine.py

"""
modules/jito_tip_engine.py — dynamic Jito tip bidding + redundant regional
bundle submission (v1.0).

Operator request (2026-07-28), backed by their own "Jito MEV Dynamic Tip
Report": replace the static SOLANA_JITO_TIP_LAMPORTS bid with the report's
auction model, and stop pointing every bundle at a single block engine.

────────────────────────────────────────────────────────────────────────────
1. WHAT THIS REPLACES
────────────────────────────────────────────────────────────────────────────
Before this module, solana_executor.py did two things with Jito:

  • tipped a FIXED _JITO_TIP_LAMPORTS (default 100,000 = 0.0001 SOL) on
    every bundle, regardless of what the auction was actually clearing at
    that second, and regardless of how much edge the trade had to spend;
  • submitted to exactly ONE block engine (_JITO_BLOCK_ENGINE, default
    mainnet.block-engine.jito.wtf).

Both are losing positions in a sealed-bid auction. A fixed tip is either
too low to ever win a contested slot (silent, invisible, looks exactly
like "no opportunities") or too high on an uncontested one (paying 0.0001
SOL when the 25th percentile cleared at 0.00001). A single engine means
one regional relay's latency and health decide whether the bundle is seen
by the leader at all.

────────────────────────────────────────────────────────────────────────────
2. THE BIDDING FUNCTION (verbatim from the operator's report)
────────────────────────────────────────────────────────────────────────────
    Tip_bid = min( Tip_max , max( Tip_floor , γ · (R_expected − C_base) ) )

subject to the strict atomic profitability condition

    ΔNet = R_expected − (Tip_bid + C_base) > 0

where
    R_expected  gross expected yield of the round trip, lamports
    C_base      signature fee + compute-unit priority budget, lamports
    Tip_floor   live floor from Jito's tip-floor feed
    Tip_max     the searcher's own risk ceiling
    γ           aggressiveness ∈ [0.90, 0.999] — the share of net revenue
                handed to the validator to outbid competing searchers.

ONE HONEST DEVIATION, and it matters. The report writes the profitability
condition as a constraint but never says what to do when the constraint
and the `max(Tip_floor, …)` clause disagree — and they can, easily: a
0.3 bps round trip on 10,000 USDC has an R−C of maybe 25,000 lamports
while the auction's 50th percentile sits at 100,000. Taken literally,
`max()` wins and the bot bids a tip larger than the entire trade's
profit. If that bundle then LANDS, the trade is a guaranteed, realised
loss — the exact opposite of what the model is for.

So the constraint is enforced LAST here, not as a footnote:

    afford = R_expected − C_base − MIN_KEEP     (MIN_KEEP > 0)
    tip    = min(Tip_max, max(Tip_floor, γ·(R−C)), afford)

and if `afford` lands below the floor by more than
JITO_FLOOR_TOLERANCE_FRACTION, the trade is reported PRICED_OUT and no
bundle is sent at all. That is not timidity: a bundle bid under the
clearing floor simply doesn't land, so skipping costs nothing but a
wasted RPC round trip, while landing an unaffordable bid costs real
money. `decide()` returns that verdict explicitly so the caller can log
"the auction was too expensive for this edge" instead of a silent miss.

────────────────────────────────────────────────────────────────────────────
3. γ IS NOT A CONSTANT (the report's own "Architectural Recommendation")
────────────────────────────────────────────────────────────────────────────
The report closes by recommending searchers "dynamically scale γ based on
real-time mempool density and validator slot leaders." This bot has no
mempool-density feed, but it has something better-grounded and free: its
own landing record. `note_outcome()` feeds every submitted bundle's fate
back in, and γ walks between JITO_GAMMA_MIN and JITO_GAMMA_MAX against
the trailing landing rate —

    landing rate low  → we are being outbid   → γ up   (bid harder)
    landing rate high → we are overpaying     → γ down (keep more edge)

with JITO_GAMMA_TARGET_LAND_RATE as the setpoint. This is a plain
integral controller on a bounded variable, not a model; it cannot run
away, and with adaptation disabled (JITO_GAMMA_ADAPTIVE=false) γ stays
pinned at JITO_GAMMA and the behaviour is exactly the static report
formula.

A word on the report's γ range while we are here. γ ∈ [0.90, 0.999]
means surrendering 90–99.9% of net revenue, which is rational in a
contested winner-takes-all race — the report's example (bidding 10.20 SOL
to keep 0.15) is a real strategy, and "guaranteed 0.15 SOL beats a
guaranteed zero" is sound game theory. It is a poor fit for the routes
THIS bot scans: public Jupiter round trips clearing 0.1–3 bps, where the
competition is usually not another searcher racing the same bundle but
simply the spread closing. Default γ here is therefore
JITO_GAMMA=0.90 — the bottom of the report's own range — and the
adaptive controller is left free to climb toward 0.999 if, and only if,
this deployment's real landing rate shows we are actually being outbid.
Set JITO_GAMMA_ADAPTIVE=false and JITO_GAMMA=0.985 for the report's
verbatim aggressive regime.

────────────────────────────────────────────────────────────────────────────
4. REDUNDANT REGIONAL SUBMISSION
────────────────────────────────────────────────────────────────────────────
The same signed bundle goes to every configured block engine at once and
the first acceptance wins. This is free — Jito dedupes identical bundles
by content, so N submissions of one bundle is one auction entry at one
tip, not N tips. The only cost is N HTTP requests.

Engine order defaults to Frankfurt first because this deployment runs on
an eu-central-1 box (i-08cb563ef1f9118a6, "garden-angel-frankfurt"), and
the relay nearest the searcher is the one whose queue the bundle reaches
soonest. Override with SOLANA_JITO_BLOCK_ENGINES (comma-separated).

Everything here is best-effort: a dead tip-floor feed falls back to the
last good reading, then to the static SOLANA_JITO_TIP_LAMPORTS the
executor used before this module existed. No path in this file can raise
into the executor's send loop.
"""

from __future__ import annotations

import asyncio
import logging
import os
import time
from dataclasses import dataclass, field
from typing import Any, Optional

import httpx

logger = logging.getLogger(__name__)

# v1.71 — strong references to in-flight background floor refreshes. asyncio
# holds only a weak reference to a running task, so without this set the
# garbage collector may destroy a refresh mid-flight and the floor silently
# never updates. Entries remove themselves on completion.
_BG_REFRESHES: set = set()

LAMPORTS_PER_SOL = 1_000_000_000


def _env_float(name: str, default: float) -> float:
    try:
        raw = os.getenv(name, "").strip()
        return float(raw) if raw else default
    except (TypeError, ValueError):
        return default


def _env_int(name: str, default: int) -> int:
    try:
        raw = os.getenv(name, "").strip()
        return int(float(raw)) if raw else default
    except (TypeError, ValueError):
        return default


def _env_bool(name: str, default: bool) -> bool:
    raw = os.getenv(name, "").strip().lower()
    if not raw:
        return default
    return raw in ("1", "true", "yes", "on")


# ── The report's parameters, all env-overridable ─────────────────────────────

# γ — aggressiveness coefficient. See section 3 above for why the default
# is the BOTTOM of the report's [0.90, 0.999] range rather than the 0.985
# its worked example uses.
_GAMMA_DEFAULT = max(0.0, min(0.999, _env_float("JITO_GAMMA", 0.90)))
_GAMMA_MIN = max(0.0, min(0.999, _env_float("JITO_GAMMA_MIN", 0.50)))
_GAMMA_MAX = max(0.0, min(0.999, _env_float("JITO_GAMMA_MAX", 0.985)))
_GAMMA_ADAPTIVE = _env_bool("JITO_GAMMA_ADAPTIVE", True)
# Setpoint for the controller: the trailing landing rate we steer toward.
# 0.55 rather than something near 1.0 deliberately — a landing rate of 1.0
# is not a triumph, it means every bid was larger than it needed to be.
_GAMMA_TARGET_LAND_RATE = max(0.05, min(0.95, _env_float("JITO_GAMMA_TARGET_LAND_RATE", 0.55)))
# Per-observation step. Small: this is an integral controller with no
# damping term, and the bounded range does the rest of the work.
_GAMMA_STEP = max(0.001, min(0.2, _env_float("JITO_GAMMA_STEP", 0.02)))
# Don't move γ at all until there's a real sample to move it on.
_GAMMA_MIN_SAMPLES = max(3, _env_int("JITO_GAMMA_MIN_SAMPLES", 8))
# v1.1 — size scaling. R_ref is the edge at which full aggression is
# warranted; below it γ interpolates down toward _GAMMA_MIN. Default
# 0.5 SOL (~$150 of net edge) — comfortably above anything this bot has
# ever produced, so today every trade sits near γ_min and keeps its
# margin, and the aggressive regime only engages if edges ever get big
# enough for the winner-takes-all argument to be true.
_GAMMA_SIZE_SCALING = _env_bool("JITO_GAMMA_SIZE_SCALING", True)
_R_REF_LAMPORTS = max(0, _env_int("JITO_GAMMA_R_REF_LAMPORTS", 500_000_000))
_GAMMA_WINDOW = max(_GAMMA_MIN_SAMPLES, _env_int("JITO_GAMMA_WINDOW", 40))
# v1.2 — FIRST BLOOD. Until this many bundles have LANDED, bid the ceiling
# and skip size scaling. γ adapts toward an observed landing rate and this
# deployment has never observed one (0/0 bundles), so the controller is
# open-loop and stays wherever it started. Overpaying deliberately is the
# only way to buy the first observations. See in_first_blood(); every
# profitability guard in decide() still applies, so this can never bid away
# more than the trade actually earns.
_FIRST_BLOOD_BUNDLES = max(0, _env_int("JITO_FIRST_BLOOD_BUNDLES", 15))
_FIRST_BLOOD_ENABLED = _FIRST_BLOOD_BUNDLES > 0

# Tip_max — the searcher's hard risk ceiling, lamports. 5,000,000 (0.005
# SOL, ~$1.50) matches the clamp solana_executor.py already applied to the
# static SOLANA_JITO_TIP_LAMPORTS, so enabling this engine can never bid
# above what the previous code path could.
_TIP_MAX_LAMPORTS = max(1_000, _env_int("JITO_TIP_MAX_LAMPORTS", 5_000_000))
# Absolute lower bound on any bid we actually submit. Below Jito's own
# 1,000-lamport minimum the block engine rejects the bundle outright.
_TIP_MIN_LAMPORTS = max(1_000, _env_int("JITO_TIP_MIN_LAMPORTS", 1_000))

# Which percentile of the live tip-floor feed counts as "the floor".
# Jito publishes 25th/50th/75th/95th/99th plus an EMA of the 50th.
# 25th is the honest reading of "the price of entry" — the 50th is the
# price of a comfortable win, and bidding it on every uncontested slot is
# exactly the overpayment this module exists to stop.
_TIP_FLOOR_PERCENTILE = os.getenv("JITO_TIP_FLOOR_PERCENTILE", "25").strip() or "25"
# v1.72 — how far ABOVE the clearing price to bid. The bid is anchored to
# the floor (see decide()), and this is the headroom that turns "matches the
# clearing price" into "beats it". 1.5 is deliberately modest: on an
# uncontested slot anything over the floor wins, and the money saved by not
# overbidding is the entire edge on trades this size. Raise it only once
# bundles are being SUBMITTED and LOSING.
_TIP_FLOOR_MULTIPLIER = max(1.0, _env_float("JITO_TIP_FLOOR_MULTIPLIER", 1.5))
# How far below the floor an affordable bid may sit before we call the
# trade priced out. 0.5 == "bid anyway if we can afford at least half the
# floor" — a losing bundle costs nothing, so a little optimism here is
# free, but a bid at 2% of the floor is pure wasted latency.
_FLOOR_TOLERANCE_FRACTION = max(0.0, min(1.0, _env_float("JITO_FLOOR_TOLERANCE_FRACTION", 0.5)))
# Profit that must survive the tip no matter what the formula says, in
# lamports. Enforces the report's strict ΔNet > 0 with a real, non-zero
# margin instead of a knife-edge equality.
_MIN_KEEP_LAMPORTS = max(0, _env_int("JITO_MIN_KEEP_LAMPORTS", 5_000))

_TIP_FLOOR_URL = os.getenv(
    "JITO_TIP_FLOOR_URL", "https://bundles.jito.wtf/api/v1/bundles/tip_floor",
).strip()
_TIP_FLOOR_TTL_SECS = max(1.0, _env_float("JITO_TIP_FLOOR_TTL_SECS", 20.0))

# Regional block engines, Frankfurt first — see section 4. Jito's canonical
# mainnet relays; the bare `mainnet.block-engine.jito.wtf` global endpoint
# stays in the list as a last resort.
_DEFAULT_BLOCK_ENGINES = (
    "https://frankfurt.mainnet.block-engine.jito.wtf",
    "https://amsterdam.mainnet.block-engine.jito.wtf",
    "https://london.mainnet.block-engine.jito.wtf",
    "https://ny.mainnet.block-engine.jito.wtf",
)


def block_engines() -> list[str]:
    """Configured block-engine base URLs, in submission-preference order.

    SOLANA_JITO_BLOCK_ENGINES (comma-separated) wins. Falls back to the
    legacy single-value SOLANA_JITO_BLOCK_ENGINE — so an existing .env
    that only ever set the old variable keeps working unchanged, just
    with the regional relays appended behind it rather than replacing it.
    """
    raw = os.getenv("SOLANA_JITO_BLOCK_ENGINES", "").strip()
    if raw:
        urls = [u.strip().rstrip("/") for u in raw.split(",") if u.strip()]
        if urls:
            return urls
    legacy = os.getenv("SOLANA_JITO_BLOCK_ENGINE", "").strip().rstrip("/")
    urls = list(_DEFAULT_BLOCK_ENGINES)
    if legacy:
        # Operator's explicit choice leads; the regionals become redundancy
        # behind it instead of overriding a deliberate setting.
        urls = [legacy] + [u for u in urls if u != legacy]
    return urls


@dataclass
class TipDecision:
    """Everything the caller needs to log an auditable bid — or to explain,
    in one line, why no bundle was sent."""

    tip_lamports: int
    should_bundle: bool
    reason: str
    gamma: float
    floor_lamports: int
    affordable_lamports: int
    expected_revenue_lamports: int
    base_cost_lamports: int
    floor_source: str = "none"

    @property
    def tip_sol(self) -> float:
        return self.tip_lamports / LAMPORTS_PER_SOL

    def describe(self) -> str:
        if not self.should_bundle:
            return f"no bundle — {self.reason}"
        return (
            f"tip {self.tip_lamports:,} lamports ({self.tip_sol:.6f} SOL) "
            f"— γ={self.gamma:.3f}, floor {self.floor_lamports:,} "
            f"({self.floor_source}), affordable {self.affordable_lamports:,}"
        )


@dataclass
class _FloorCache:
    lamports: int = 0
    fetched_at: float = 0.0
    source: str = "none"
    percentiles: dict[str, int] = field(default_factory=dict)


class JitoTipEngine:
    """Stateful across the process: caches the tip floor, tracks the landing
    rate, and walks γ. One instance per bot (see `get_tip_engine()`)."""

    def __init__(self) -> None:
        self._gamma = _GAMMA_DEFAULT
        self._floor = _FloorCache()
        self._lock = asyncio.Lock()
        # Bounded outcome ring — True == the bundle landed. Only ever read
        # as a rate, never as a history, so a plain list sliced to
        # _GAMMA_WINDOW is the whole data structure needed.
        self._outcomes: list[bool] = []
        self._bundles_submitted = 0
        self._bundles_landed = 0
        self._tip_lamports_paid = 0
        self._priced_out = 0
        # v1.71 — set while a background floor refresh is in flight, so
        # decide_fast() starts at most one. Without it, a burst of attempts
        # against a stale floor would each schedule their own refresh of the
        # same rate-limited feed. Plain bool rather than a lock: it is only
        # ever touched from the event loop thread, and the cost of a rare
        # duplicate refresh is far below the cost of awaiting a lock on the
        # send path.
        self._refresh_inflight = False

    # ── γ ────────────────────────────────────────────────────────────────
    @property
    def gamma(self) -> float:
        return self._gamma

    def in_first_blood(self) -> bool:
        """True while this bot has never landed a bundle (v1.2).

        Operator: "i dont have problem to pay for jito and make profit at
        the first nothing i pay so give them high profit."

        Right, and there is a control-theory reason as well as a
        willingness one. γ adapts toward a target LANDING RATE, and
        _landing_rate() returns None until _GAMMA_MIN_SAMPLES bundles have
        resolved. With zero bundles ever submitted — which is this
        deployment's actual state, `Bundles: 0/0 landed` on /jito — the
        controller has no feedback at all and γ sits wherever it was
        initialised, forever. It cannot learn its way to a winning bid
        because it has never made one.

        The only way out of that is to overpay deliberately until there
        are observations to adapt from. A landed bundle at a bad price is
        worth far more than an unlanded one at a good price: it converts
        the landing rate from unknown to measured, and everything after it
        is a real control loop instead of an open one.

        Ends by itself. Once JITO_FIRST_BLOOD_BUNDLES have landed, γ goes
        back to being whatever the controller has learned.
        """
        if not _FIRST_BLOOD_ENABLED:
            return False
        return self._bundles_landed < _FIRST_BLOOD_BUNDLES

    def _scaled_gamma(self, net_lamports: int) -> float:
        """γ(R) — see decide()'s own comment for why size matters here.

        Interpolates from _GAMMA_MIN at a negligible edge up to the
        controller's current γ at _R_REF and above. With
        JITO_GAMMA_SIZE_SCALING=false this returns the controller's γ
        unchanged, which is the pre-v1.1 behaviour exactly.
        """
        # v1.2 — first blood: bid the ceiling, and do NOT scale it down for
        # a small edge. Size scaling is the right policy once landing is a
        # known quantity; while it is unknown, shading a small trade's bid
        # down is how a bot never lands anything and therefore never finds
        # out what landing costs. The profitability constraint in decide()
        # is untouched — this can still never bid away more than the trade
        # earns. See in_first_blood().
        if self.in_first_blood():
            return _GAMMA_MAX
        if not _GAMMA_SIZE_SCALING or _R_REF_LAMPORTS <= 0:
            return self._gamma
        ratio = min(1.0, max(0.0, net_lamports / _R_REF_LAMPORTS))
        low = min(_GAMMA_MIN, self._gamma)
        return low + (self._gamma - low) * ratio

    def _landing_rate(self) -> Optional[float]:
        if len(self._outcomes) < _GAMMA_MIN_SAMPLES:
            return None
        return sum(1 for o in self._outcomes if o) / len(self._outcomes)

    def note_outcome(self, landed: bool, tip_lamports: int = 0) -> None:
        """Feed one submitted bundle's fate back into the controller.

        Called from the executor after _confirm() resolves. Never raises;
        an unknown/ambiguous outcome should simply not be reported rather
        than guessed at, since a wrong sample moves γ the wrong way.
        """
        # v1.3 — attach the fate to the remembered bid, so the landing rate
        # survives a restart. These counters are RAM: the operator restarts
        # several times an hour, and "0/0 landed" on /jito was a statement
        # about the last few minutes of uptime that read identically to
        # "we have never won".
        try:
            from modules.tip_memory import record_outcome

            record_outcome(bool(landed))
        except Exception as exc:  # noqa: BLE001
            logger.debug("[JitoTip] outcome not remembered: %s", exc)

        self._outcomes.append(bool(landed))
        if len(self._outcomes) > _GAMMA_WINDOW:
            del self._outcomes[: len(self._outcomes) - _GAMMA_WINDOW]
        self._bundles_submitted += 1
        if landed:
            self._bundles_landed += 1
            self._tip_lamports_paid += max(0, int(tip_lamports))

        if not _GAMMA_ADAPTIVE:
            return
        rate = self._landing_rate()
        if rate is None:
            return
        before = self._gamma
        if rate < _GAMMA_TARGET_LAND_RATE:
            # Losing more auctions than we want — the report's core claim is
            # that a conservative γ guarantees a loss against aggressive
            # searchers. Bid harder.
            self._gamma = min(_GAMMA_MAX, self._gamma + _GAMMA_STEP)
        elif rate > _GAMMA_TARGET_LAND_RATE:
            # Winning nearly everything means the bids were larger than they
            # needed to be; every lamport above the clearing price was pure
            # donation. Keep more of the edge.
            self._gamma = max(_GAMMA_MIN, self._gamma - _GAMMA_STEP)
        if abs(self._gamma - before) > 1e-9:
            logger.info(
                "[JitoTip] γ %.3f -> %.3f (landing rate %.0f%% over last %d bundles, "
                "target %.0f%%)",
                before, self._gamma, rate * 100, len(self._outcomes),
                _GAMMA_TARGET_LAND_RATE * 100,
            )

    # ── Tip floor ────────────────────────────────────────────────────────
    async def refresh_tip_floor(self, client: httpx.AsyncClient) -> int:
        """Live tip floor in lamports, cached for _TIP_FLOOR_TTL_SECS.

        Jito's feed returns a single-element array of SOL-denominated
        percentiles. Any failure serves the last good reading (however
        stale) and, failing that, 0 — which `decide()` reads as "no floor
        known", not as "the floor is free".
        """
        now = time.monotonic()
        if self._floor.lamports and (now - self._floor.fetched_at) < _TIP_FLOOR_TTL_SECS:
            return self._floor.lamports
        async with self._lock:
            # Re-check after acquiring: a concurrent caller may have just
            # refreshed it, and this feed is rate-limited like any other.
            now = time.monotonic()
            if self._floor.lamports and (now - self._floor.fetched_at) < _TIP_FLOOR_TTL_SECS:
                return self._floor.lamports
            try:
                r = await client.get(_TIP_FLOOR_URL, timeout=5.0)
                r.raise_for_status()
                body = r.json()
                row = body[0] if isinstance(body, list) and body else body
                if not isinstance(row, dict):
                    raise ValueError(f"unexpected tip_floor payload: {type(row).__name__}")
                percentiles: dict[str, int] = {}
                for key, value in row.items():
                    if not isinstance(value, (int, float)):
                        continue
                    # Feed is denominated in SOL, e.g. 0.000012 — convert once.
                    percentiles[key] = int(float(value) * LAMPORTS_PER_SOL)
                wanted = (
                    f"landed_tips_{_TIP_FLOOR_PERCENTILE}th_percentile",
                    f"ema_landed_tips_{_TIP_FLOOR_PERCENTILE}th_percentile",
                    "landed_tips_25th_percentile",
                    "landed_tips_50th_percentile",
                )
                lamports, source = 0, "none"
                for key in wanted:
                    if percentiles.get(key):
                        lamports, source = percentiles[key], key
                        break
                if lamports <= 0:
                    raise ValueError("tip_floor payload carried no usable percentile")
                self._floor = _FloorCache(
                    lamports=lamports, fetched_at=time.monotonic(),
                    source=source, percentiles=percentiles,
                )
                logger.debug(
                    "[JitoTip] tip floor %s = %d lamports (%.6f SOL)",
                    source, lamports, lamports / LAMPORTS_PER_SOL,
                )
                # v1.3 — persist it. The percentiles were fetched, used for
                # one decision and dropped, so /jito could only ever
                # describe this process's uptime. See modules/tip_memory.py.
                try:
                    from modules.tip_memory import record_floor

                    record_floor(percentiles)
                except Exception as exc:  # noqa: BLE001
                    logger.debug("[JitoTip] floor not remembered: %s", exc)
            except Exception as exc:  # noqa: BLE001 — best-effort by design
                # v1.3 — a REMEMBERED floor beats no floor.
                #
                # When the feed failed and this process had no cached floor
                # — every restart, and any outage — the engine bid into a
                # live auction with floor=0. `max(Tip_floor, γ·(R−C))` then
                # has nothing to hold the bid up, which is how a bundle
                # goes out at the minimum without anyone deciding that.
                #
                # An hour-old 25th percentile is a far better prior than
                # zero, and it is labelled `remembered` so no reader
                # mistakes it for live.
                if not self._floor.lamports:
                    try:
                        from modules.tip_memory import remembered_floor

                        lam, age_secs = remembered_floor()
                        if lam:
                            self._floor = _FloorCache(
                                lamports=lam, fetched_at=time.monotonic(),
                                source=f"remembered ({age_secs / 60:.0f}m old)",
                            )
                            logger.info(
                                "[JitoTip] feed down — using remembered floor "
                                "%d lamports (%.0fm old) rather than bidding "
                                "with none", lam, age_secs / 60,
                            )
                    except Exception:  # noqa: BLE001
                        pass
                age = time.monotonic() - self._floor.fetched_at if self._floor.fetched_at else None
                logger.debug(
                    "[JitoTip] tip-floor fetch failed (%s) — %s",
                    str(exc)[:120],
                    f"serving cached floor {self._floor.lamports} lamports "
                    f"({age:.0f}s old)" if self._floor.lamports else
                    "no cached floor, bidding without one",
                )
        return self._floor.lamports

    # ── The bid ──────────────────────────────────────────────────────────
    def decide(
        self,
        expected_revenue_lamports: int,
        base_cost_lamports: int,
        floor_lamports: Optional[int] = None,
        min_keep_lamports: Optional[int] = None,
    ) -> TipDecision:
        """Evaluate the report's bidding function for one trade.

        `expected_revenue_lamports` is R_expected — the round trip's GROSS
        expected yield, already converted to lamports by the caller (the
        executor knows the live SOL/USD figure; this module deliberately
        does not, so it can never disagree with the margin gate about what
        a dollar is worth).

        `base_cost_lamports` is C_base — signature fee plus the
        compute-unit priority budget this attempt will actually bid. It
        must NOT include the tip; that is what this function returns.
        """
        R = max(0, int(expected_revenue_lamports))
        C = max(0, int(base_cost_lamports))
        floor = int(self._floor.lamports if floor_lamports is None else floor_lamports)
        floor = max(0, floor)

        # v1.4 — THE TIP MUST LEAVE ENOUGH FOR THE MARGIN GATE.
        #
        # _MIN_KEEP_LAMPORTS defaults to 5,000 (~$0.00036 at $73/SOL). The
        # executor's margin gate requires SOLANA_MIN_REAL_NET_MARGIN, which
        # on this deployment is $0.002 — about 27,400 lamports, five times
        # larger. So this engine was computing a bid that left less than the
        # gate demands, and the gate was then obliged to refuse the trade the
        # engine had just priced.
        #
        # The operator's own decline message is the arithmetic:
        #
        #   net $+0.0654 - $0.0645 network fee (… Jito tip, 888,254 lamports)
        #   < margin $0.002 — aborted before broadcasting anything
        #
        #   edge          $0.0654
        #   tip           $0.0586   90% of it, exactly gamma
        #   base+priority $0.0062
        #   left          $0.0006   against a $0.0020 gate  -> REJECT
        #
        # At gamma 0.90 that is not bad luck, it is arithmetic: bidding 90%
        # of the edge cannot leave the gate its 3%. Every such trade was
        # rejected by construction, which is why 20 attempts produced 0
        # sends and /tune then recommended raising gamma to 0.985 — advice
        # that would have made the guarantee stronger.
        #
        # The caller passes the gate's own requirement so the two cannot
        # drift apart. Without it the constant is used, exactly as before.
        keep = _MIN_KEEP_LAMPORTS if min_keep_lamports is None else max(
            _MIN_KEEP_LAMPORTS, int(min_keep_lamports))
        net_before_tip = R - C
        # v1.1 (2026-07-29, operator: "make γ scale with size instead of a
        # flat constant … giving up 98% only makes sense when what's left is
        # still worth having") — correct, and it fixes a real defect in the
        # flat version.
        #
        #     γ(R) = γ_min + (γ_max − γ_min) × min(1, R / R_ref)
        #
        # The flat γ treated a $1.15 edge and a $3,000 edge identically,
        # which is precisely the mistake in transplanting the report's
        # worked example onto this bot's actual trade sizes. At γ = 0.985 a
        # $1.15 edge keeps $0.017 — 1,465 landed trades to clear $25 — while
        # the same γ on a $3,000 edge keeps $46, which is worth having and
        # is what that number was calibrated for.
        #
        # Size-scaling makes the aggression EARNED rather than assumed:
        # small edges keep the margin that exists, and only edges above
        # JITO_GAMMA_R_REF climb toward the aggressive end where the
        # winner-takes-all argument actually applies.
        #
        # The landing-rate controller still moves the BAND (self._gamma);
        # this scales within it by trade size. The two compose: "how
        # contested is the auction lately" times "is this edge big enough to
        # be worth fighting for".
        gamma = self._scaled_gamma(net_before_tip)

        # The strict condition, applied before anything else: a trade whose
        # gross yield doesn't already cover its own base cost has nothing to
        # bid with, and no tip can rescue it.
        if net_before_tip <= keep:
            self._priced_out += 1
            return TipDecision(
                tip_lamports=0, should_bundle=False,
                reason=(
                    f"expected revenue {R:,} lamports does not clear base cost "
                    f"{C:,} + minimum keep {keep:,} — nothing to bid with"
                ),
                gamma=gamma, floor_lamports=floor, affordable_lamports=0,
                expected_revenue_lamports=R, base_cost_lamports=C,
                floor_source=self._floor.source,
            )

        affordable = net_before_tip - keep

        # v1.72 — γ IS A CEILING, NOT A SECOND FLOOR.
        #
        # The old line was `tip = min(TIP_MAX, max(floor, γ·(R−C)))`, taken
        # literally from the tip report. Read what `max` does there: the bid
        # is the LARGER of the clearing price and a fixed share of the edge.
        # So whenever γ·(R−C) exceeds the floor — which is essentially
        # always — the floor stops being an input at all and the bot bids
        # half its edge into an auction that is clearing for pennies.
        #
        # THIS IS WHY 0 BUNDLES WERE EVER SUBMITTED. From the operator's own
        # /jito and the attempt that motivated this, at $73/SOL:
        #
        #     edge R                880,821 lamports   $0.0643
        #     live floor              1,122 lamports   $0.0001
        #     γ·(R−C) at γ=0.5      397,910 lamports   $0.0290   ← bid
        #
        # A bid 355x the clearing price, eating 45% of the edge — after
        # which execute_flash_arb's own margin gate correctly refused to
        # send, every single time. The bot was not losing auctions. It was
        # pricing itself out before reaching one, and /jito has been
        # reporting `Bundles: 0/0` as the proof.
        #
        # An auction is won by beating the clearing price, not by donating a
        # share of your profit. So: anchor the bid to the FLOOR, multiply it
        # by enough to be comfortably above it, and let γ·(R−C) act as the
        # ceiling it always should have been — the most this trade is
        # willing to pay, not the amount it must pay.
        #
        # With floor 1,122 the same trade now bids ~1,683 lamports ($0.0001)
        # and keeps $0.0597 of its $0.0643. That is the difference between a
        # bot that trades and one that watches.
        #
        # JITO_TIP_FLOOR_MULTIPLIER tunes the headroom over the clearing
        # price. Raise it when bundles are submitted and LOSING — that is
        # the real signal to bid harder, and it is a signal this deployment
        # has never once produced.
        aggressive = int(gamma * net_before_tip)      # the ceiling
        anchored = int(max(floor, 0) * _TIP_FLOOR_MULTIPLIER)
        tip = max(_TIP_MIN_LAMPORTS, anchored)
        if aggressive > 0:
            tip = min(tip, aggressive)
        tip = min(_TIP_MAX_LAMPORTS, tip)
        # No floor known (fresh box, feed down and nothing remembered) is
        # the one case where the anchor cannot be computed. Fall back to the
        # old share-of-edge behaviour rather than bidding the bare minimum
        # into an auction whose price is genuinely unknown.
        if floor <= 0 and aggressive > 0:
            tip = min(_TIP_MAX_LAMPORTS, max(_TIP_MIN_LAMPORTS, aggressive))

        # …subject to ΔNet > 0. This is the clause the report leaves
        # implicit; see the module docstring, section 2.
        if tip > affordable:
            if floor > 0 and affordable < floor * _FLOOR_TOLERANCE_FRACTION:
                self._priced_out += 1
                return TipDecision(
                    tip_lamports=0, should_bundle=False,
                    reason=(
                        f"priced out of the auction — this edge can afford at most "
                        f"{affordable:,} lamports of tip but the "
                        f"{self._floor.source if self._floor.source != 'none' else 'live'} "
                        f"floor is {floor:,} ({affordable / floor:.0%} of it, below the "
                        f"{_FLOOR_TOLERANCE_FRACTION:.0%} tolerance). Bidding under the "
                        f"clearing price would not land; bidding over it would land at a loss"
                    ),
                    gamma=gamma, floor_lamports=floor, affordable_lamports=affordable,
                    expected_revenue_lamports=R, base_cost_lamports=C,
                    floor_source=self._floor.source,
                )
            tip = affordable

        if tip < _TIP_MIN_LAMPORTS:
            self._priced_out += 1
            return TipDecision(
                tip_lamports=0, should_bundle=False,
                reason=(
                    f"affordable tip {tip:,} lamports is below Jito's "
                    f"{_TIP_MIN_LAMPORTS:,}-lamport minimum — the block engine would "
                    f"reject the bundle"
                ),
                gamma=gamma, floor_lamports=floor, affordable_lamports=affordable,
                expected_revenue_lamports=R, base_cost_lamports=C,
                floor_source=self._floor.source,
            )

        # v1.3 — write the bid down before returning it. Only this path,
        # deliberately: the earlier returns are decisions NOT to bundle, and
        # recording them as bids would put rows in the auction history for
        # auctions this bot never entered.
        try:
            from modules.tip_memory import record_bid

            record_bid(tip_lamports=int(tip), floor_lamports=floor,
                       gamma=gamma, net_lamports=net_before_tip,
                       floor_source=self._floor.source)
        except Exception as exc:  # noqa: BLE001 — memory never blocks a bid
            logger.debug("[JitoTip] bid not remembered: %s", exc)

        return TipDecision(
            tip_lamports=int(tip), should_bundle=True,
            reason=(
                f"γ·(R−C) = {aggressive:,}, floor {floor:,}, ceiling "
                f"{_TIP_MAX_LAMPORTS:,}, affordable {affordable:,}"
            ),
            gamma=gamma, floor_lamports=floor, affordable_lamports=affordable,
            expected_revenue_lamports=R, base_cost_lamports=C,
            floor_source=self._floor.source,
        )

    async def decide_live(
        self, client: httpx.AsyncClient,
        expected_revenue_lamports: int, base_cost_lamports: int,
        min_keep_lamports: Optional[int] = None,
    ) -> TipDecision:
        """`decide()` with a freshly-refreshed floor. One HTTP call at most
        per _TIP_FLOOR_TTL_SECS across the whole process.

        `min_keep_lamports` is the caller's own margin gate, passed through
        so the bid can never be sized to leave less than the gate will
        demand back — see decide() for the arithmetic that made every one
        of the operator's 20 attempts reject by construction."""
        floor = await self.refresh_tip_floor(client)
        return self.decide(expected_revenue_lamports, base_cost_lamports, floor,
                           min_keep_lamports)

    def decide_fast(
        self, client: httpx.AsyncClient,
        expected_revenue_lamports: int, base_cost_lamports: int,
        min_keep_lamports: Optional[int] = None,
    ) -> TipDecision:
        """`decide_live()` without the network wait. Never awaits, never raises.

        v1.71 — WHY THIS EXISTS.

        decide_live() awaits refresh_tip_floor(), which is an HTTPS GET to
        bundles.jito.wtf with `timeout=5.0`. The executor called it from the
        middle of the send path, while holding the Jupiter lock, and — because
        `tip` was declared in pipeline_trace.STAGES but `trace.stage("tip")`
        was never opened around it — the time it took was invisible, landing
        in `unaccounted_ms` where no report could name it. A five-second
        worst case sat on the critical path of a trade whose edge decays at
        $0.92/s, and /pipeline could not have told anyone.

        The floor does not need to be fetched inside the trade. It is a 20s-
        TTL reading of a market-wide auction statistic — it describes what
        OTHER people paid recently, not anything about this trade — and
        refresh_tip_floor() already serves a cached value, already falls back
        to tip_memory's remembered floor, and already treats a failure as
        "serve what we have". Everything needed to answer instantly is
        present; the only reason it blocked is that the refresh and the read
        were the same call.

        So this splits them: read what we know NOW, and if that reading is
        stale, start the refresh in the background for the NEXT trade. The
        bid is made against a floor that is at most a few tens of seconds
        old instead of against one fetched at a cost that can exceed the
        edge being bid for.
        """
        # If the cached reading is stale, refresh behind us — never in front.
        try:
            age = (time.monotonic() - self._floor.fetched_at
                   if self._floor.fetched_at else float("inf"))
            if age >= _TIP_FLOOR_TTL_SECS and not self._refresh_inflight:
                self._refresh_inflight = True

                async def _bg() -> None:
                    try:
                        await self.refresh_tip_floor(client)
                    except Exception as exc:  # noqa: BLE001
                        logger.debug("[JitoTip] background floor refresh failed: %s", exc)
                    finally:
                        self._refresh_inflight = False

                task = asyncio.ensure_future(_bg())
                # Strong reference: asyncio only weakly references running
                # tasks, so without this the refresh can be collected before
                # it completes and the floor never updates.
                _BG_REFRESHES.add(task)
                task.add_done_callback(_BG_REFRESHES.discard)
        except Exception as exc:  # noqa: BLE001 — a bid must never fail on its own telemetry
            self._refresh_inflight = False
            logger.debug("[JitoTip] could not schedule floor refresh: %s", exc)

        return self.decide(expected_revenue_lamports, base_cost_lamports,
                           self._floor.lamports, min_keep_lamports)

    # ── Redundant submission ─────────────────────────────────────────────
    async def send_bundle(
        self, client: httpx.AsyncClient, tx_b64: str,
        engines: Optional[list[str]] = None,
    ) -> tuple[bool, str]:
        """Submit one already-signed transaction as a single-tx bundle to
        every configured block engine concurrently. Returns (accepted,
        detail).

        Identical bundles are deduped by the auction, so this buys landing
        probability at zero extra tip — the same bid simply reaches more
        relays, and whichever one is closest to the current leader wins.
        The first acceptance is enough; remaining requests are cancelled.

        Never raises. A total failure returns (False, why) so the caller
        can fall back to plain RPC sendTransaction exactly as before.
        """
        urls = engines if engines is not None else block_engines()
        if not urls:
            return False, "no Jito block engines configured"

        payload = {
            "jsonrpc": "2.0", "id": 1, "method": "sendBundle",
            "params": [[tx_b64], {"encoding": "base64"}],
        }

        async def _submit(base_url: str) -> str:
            r = await client.post(
                f"{base_url.rstrip('/')}/api/v1/bundles", json=payload, timeout=8.0,
            )
            r.raise_for_status()
            body = r.json()
            if isinstance(body, dict) and body.get("error"):
                raise RuntimeError(f"{base_url}: {str(body['error'])[:150]}")
            bundle_id = body.get("result") if isinstance(body, dict) else None
            return f"{base_url} accepted (bundle {str(bundle_id)[:16]})"

        tasks = [asyncio.create_task(_submit(u)) for u in urls]
        errors: list[str] = []
        try:
            for coro in asyncio.as_completed(tasks):
                try:
                    detail = await coro
                except Exception as exc:  # noqa: BLE001 — try the next relay
                    errors.append(str(exc)[:120])
                    continue
                accepted_by = len(urls) - len(errors)
                logger.info(
                    "[JitoTip] bundle accepted — %s (submitted to %d engine(s), "
                    "%d rejected so far)", detail, len(urls), len(errors),
                )
                return True, f"{detail}; submitted to {len(urls)} engine(s)"
        finally:
            for t in tasks:
                if not t.done():
                    t.cancel()
        return False, (
            f"all {len(urls)} block engine(s) rejected the bundle: "
            + " | ".join(errors[:3])
        )

    # ── Introspection (for /doctor, /jito, the dashboard) ────────────────
    def status(self) -> dict[str, Any]:
        rate = self._landing_rate()
        floor_age = (
            None if not self._floor.fetched_at
            else round(time.monotonic() - self._floor.fetched_at, 1)
        )
        return {
            "gamma": round(self._gamma, 4),
            "gamma_adaptive": _GAMMA_ADAPTIVE,
            "gamma_size_scaling": _GAMMA_SIZE_SCALING,
            "gamma_r_ref_sol": round(_R_REF_LAMPORTS / LAMPORTS_PER_SOL, 4),
            "gamma_bounds": [_GAMMA_MIN, _GAMMA_MAX],
            "target_land_rate": _GAMMA_TARGET_LAND_RATE,
            "landing_rate": None if rate is None else round(rate, 3),
            # v1.2 — say plainly when the ceiling is being bid on purpose,
            # so a large tip reads as a decision rather than a bug.
            "first_blood": self.in_first_blood(),
            "first_blood_bundles": _FIRST_BLOOD_BUNDLES,
            "samples": len(self._outcomes),
            "bundles_submitted": self._bundles_submitted,
            "bundles_landed": self._bundles_landed,
            "priced_out": self._priced_out,
            "tip_sol_paid": round(self._tip_lamports_paid / LAMPORTS_PER_SOL, 6),
            "tip_floor_lamports": self._floor.lamports,
            "tip_floor_source": self._floor.source,
            "tip_floor_age_secs": floor_age,
            "tip_max_lamports": _TIP_MAX_LAMPORTS,
            "min_keep_lamports": _MIN_KEEP_LAMPORTS,
            "block_engines": block_engines(),
        }


_engine: Optional[JitoTipEngine] = None


def get_tip_engine() -> JitoTipEngine:
    """Process-wide singleton — γ and the landing record are only meaningful
    if every bundle this bot sends contributes to the same sample."""
    global _engine
    if _engine is None:
        _engine = JitoTipEngine()
    return _engine