scripts/verify_route.py

#!/usr/bin/env python3
"""
scripts/verify_route.py — turn a survey hit into a decision (2026-07-29).

    venv/bin/python scripts/verify_route.py ETH
    venv/bin/python scripts/verify_route.py 7vfCXTUXx5WJV5JADk17DUJ4ksgau7utNKj4b963voxs
    venv/bin/python scripts/verify_route.py ETH --rounds 5 --base USDC

────────────────────────────────────────────────────────────────────────────
WHY THIS EXISTS
────────────────────────────────────────────────────────────────────────────
scripts/survey_pairs.py answered the breadth question — of 99 pairs, four
quoted positive:

    ETH  +0.92 bps    mSOL +0.92 bps    PUMP +0.28 bps    USDT +0.00 bps

That is the first evidence this deployment has ever produced of a positive
round trip, and it is nowhere near enough to trade on. Three things stand
between that line and a route in the scan set, and this script checks all
three.

1. DECIMALS, FROM AN AUTHORITY
   Every amount sent to Jupiter is `human_units * 10**decimals`. A value
   wrong by one is a trade wrong by 10x, by two is 100x — in the direction
   of borrowing far more than intended. A wrong MINT merely fails; a wrong
   DECIMALS can succeed catastrophically. So it is read out of Jupiter's
   own token list, never typed from memory.

2. DOES IT PERSIST?
   One observation of +0.92 bps is one sample of a noisy process. Quote
   noise between two calls seconds apart is routinely larger than the edge
   being measured. Several rounds, spaced out, and the MEDIAN is what
   counts — if the median is negative, the survey caught noise.

3. DOES IT SURVIVE SIZE?
   This is the one that decides whether the edge is tradeable at all. The
   survey quoted $1,000. The bot's SOLANA_REAL_LOAN_UNITS is 50,000. Price
   impact grows superlinearly with size, so an edge measured small can be
   gone — or inverted — at the size that would make it worth having.
   Quoting the ladder is the only way to know where it breaks.

The output is either an exact pair of env lines to paste, or a clear
refusal. Read-only: quotes only, this cannot sign, send or spend anything
beyond Jupiter API calls.
"""

from __future__ import annotations

import argparse
import asyncio
import os
import statistics
import sys
import time
from pathlib import Path
from typing import Any, Optional

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

try:
    import httpx
except ImportError:
    print("httpx not installed — run from the venv: venv/bin/python scripts/verify_route.py")
    raise SystemExit(1)

from constants import JUPITER_HEADERS, JUPITER_QUOTE_API, SOLANA_TOKENS
from modules.env_file import effective_haircut_bps, load_env_file

# The bot gets .env from systemd; a hand-run script gets nothing. Without
# this the gates below are the code defaults, not the deployment's.
load_env_file()

_TOKEN_LOOKUP_URLS = [
    "https://lite-api.jup.ag/tokens/v2/search?query={q}",
    "https://tokens.jup.ag/token/{q}",
]
# Ladder in base units. Spans the survey's size through the bot's configured
# real loan size, so the answer covers the range that actually matters.
_DEFAULT_SIZES = [1_000, 5_000, 10_000, 25_000, 50_000]


def _pace_secs() -> float:
    try:
        raw = float(os.getenv("SOLANA_JUPITER_MIN_INTERVAL_SECS", "") or 2.1)
    except ValueError:
        raw = 2.1
    fast = os.getenv("SOLANA_JUPITER_ALLOW_FAST_PACING", "").strip().lower() in (
        "1", "true", "yes", "on",
    )
    return raw if fast else max(2.0, raw)


async def resolve_token(client: httpx.AsyncClient, query: str) -> Optional[dict[str, Any]]:
    """Symbol or mint -> {symbol, mint, decimals}, from Jupiter.

    Built-ins short-circuit. Everything else must come back from Jupiter
    with an integer `decimals` — an absent or non-integer value is a
    refusal, not a default, for the reason in the module docstring.
    """
    for symbol, meta in SOLANA_TOKENS.items():
        if query.upper() == symbol.upper() or query == meta["mint"]:
            return {"symbol": symbol, "mint": meta["mint"],
                    "decimals": meta["decimals"], "source": "built-in"}

    for template in _TOKEN_LOOKUP_URLS:
        try:
            r = await client.get(template.format(q=query), headers=JUPITER_HEADERS, timeout=20.0)
            r.raise_for_status()
            body = r.json()
        except Exception as exc:  # noqa: BLE001 — try the next source
            print(f"    {template.split('?')[0]} — {type(exc).__name__}")
            continue
        rows = body if isinstance(body, list) else [body]
        for row in rows:
            if not isinstance(row, dict):
                continue
            mint = row.get("id") or row.get("address") or row.get("mint")
            decimals = row.get("decimals")
            symbol = row.get("symbol") or ""
            if not mint or not isinstance(decimals, int):
                continue
            if query == mint or query.upper() == str(symbol).upper():
                return {"symbol": symbol or query, "mint": mint,
                        "decimals": decimals, "source": template.split("?")[0]}
    return None


async def quote(
    client: httpx.AsyncClient, in_mint: str, out_mint: str,
    amount_raw: int, slippage_bps: int,
) -> Optional[int]:
    params = {
        "inputMint": in_mint, "outputMint": out_mint, "amount": str(amount_raw),
        "slippageBps": str(slippage_bps), "onlyDirectRoutes": "false",
        "swapMode": "ExactIn",
    }
    try:
        r = await client.get(f"{JUPITER_QUOTE_API}/quote", params=params,
                             headers=JUPITER_HEADERS, timeout=15.0)
        if r.status_code != 200:
            return None
        out = r.json().get("outAmount")
        return int(out) if out else None
    except Exception:  # noqa: BLE001
        return None


async def round_trip_bps(
    client: httpx.AsyncClient, base: dict, mid: dict,
    size: float, slippage_bps: int, pace: float,
) -> Optional[float]:
    amount_raw = int(size * (10 ** base["decimals"]))
    out_mid = await quote(client, base["mint"], mid["mint"], amount_raw, slippage_bps)
    await asyncio.sleep(pace)
    if not out_mid:
        return None
    back = await quote(client, mid["mint"], base["mint"], out_mid, slippage_bps)
    await asyncio.sleep(pace)
    if not back:
        return None
    out_units = back / (10 ** base["decimals"])
    return ((out_units - size) / size) * 10_000.0


async def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("token", help="symbol or mint address to verify")
    ap.add_argument("--base", default="USDC", help="base asset (default USDC)")
    ap.add_argument("--rounds", type=int, default=3, help="repeats per size (default 3)")
    ap.add_argument("--slippage", type=int, default=50)
    ap.add_argument("--sizes", default="", help="comma-separated base units, overrides the default ladder")
    args = ap.parse_args()

    base_meta = SOLANA_TOKENS.get(args.base)
    if not base_meta:
        print(f"unknown base '{args.base}' — pick one of: {', '.join(SOLANA_TOKENS)}")
        return 2
    base = {"symbol": args.base, **base_meta}
    sizes = ([float(s) for s in args.sizes.split(",") if s.strip()]
             if args.sizes else list(_DEFAULT_SIZES))
    pace = _pace_secs()

    async with httpx.AsyncClient() as client:
        print(f"\n1/3  resolving '{args.token}'")
        mid = await resolve_token(client, args.token)
        if not mid:
            print(f"\n  ✗ could not resolve '{args.token}' to a mint with a known "
                  f"decimals.\n    Nothing is added on a guess — a wrong decimals "
                  f"scales every trade\n    by a power of ten. Pass the exact mint "
                  f"address if the symbol is ambiguous.\n")
            return 1
        print(f"     {mid['symbol']}  {mid['mint']}")
        print(f"     decimals {mid['decimals']}  (source: {mid['source']})")
        if mid["mint"] == base["mint"]:
            print("\n  ✗ that is the base asset itself.\n")
            return 2

        total = len(sizes) * args.rounds
        print(f"\n2/3  quoting {base['symbol']}->{mid['symbol']}->{base['symbol']} — "
              f"{len(sizes)} size(s) x {args.rounds} round(s), "
              f"~{total * 2 * pace / 60:.1f} min\n")

        results: dict[float, list[float]] = {}
        for size in sizes:
            samples: list[float] = []
            for r in range(args.rounds):
                bps = await round_trip_bps(client, base, mid, size, args.slippage, pace)
                if bps is None:
                    print(f"     {size:>9,.0f} {base['symbol']}  round {r+1}: no quote")
                    continue
                samples.append(bps)
                mark = "🟢" if bps > 0 else "  "
                print(f"  {mark} {size:>9,.0f} {base['symbol']}  round {r+1}: {bps:>9.3f} bps")
            if samples:
                results[size] = samples

    if not results:
        print("\n  ✗ nothing could be priced. Check the Jupiter key and network.\n")
        return 1

    print("\n3/3  verdict\n")
    print(f"     {'size':>10}  {'median':>9}  {'min':>9}  {'max':>9}  {'n':>3}")
    print("     " + "-" * 48)
    medians: dict[float, float] = {}
    for size, samples in results.items():
        med = statistics.median(samples)
        medians[size] = med
        mark = "🟢" if med > 0 else "  "
        print(f"  {mark} {size:>10,.0f}  {med:>9.3f}  {min(samples):>9.3f}  "
              f"{max(samples):>9.3f}  {len(samples):>3}")

    # The impact slope, so the table answers "what about 100k / 200k?"
    # without another run per size.
    #
    # Operator: "make this round 100k and 200k let us see what happens with
    # big numbers." Naming two more sizes gets two more rows; fitting the
    # curve the rows already describe gets every size at once, including the
    # ones too big to be worth quoting. The refusals in size_optimizer.py
    # matter here as much as the fit: extrapolating an impact curve past
    # what was measured is precisely where a linear model fails, and it
    # fails in the direction that costs money — so a peak beyond the ladder
    # is reported as a lower bound and never as a size to trade.
    try:
        from modules.size_optimizer import fit_curve

        fit = fit_curve([(s, s * m / 10_000.0) for s, m in medians.items()])
        if fit and fit.r2 >= 0.5:
            # net(s) = alpha*s - beta*s^2, so gross bps at size s is
            # 10_000*net/s = 10_000*(alpha - beta*s). Zero crossing at
            # alpha/beta, and the slope says what each extra 10k costs.
            per_10k = fit.beta * 10_000.0 * 10_000.0
            print(f"\n     impact: every extra 10,000 {base['symbol']} costs "
                  f"{per_10k:.3f} bps  (fit R²={fit.r2:.2f})")
            if fit.beta > 0:
                zero_at = fit.alpha / fit.beta
                if zero_at > 0:
                    print(f"     the round trip crosses zero around "
                          f"{zero_at:,.0f} {base['symbol']}"
                          + ("" if zero_at <= max(medians)
                             else " — extrapolated past what was measured, "
                                  "treat as a direction, not a number"))
    except Exception:  # noqa: BLE001 — a diagnostic must not break the verdict
        pass

    positive = {s: m for s, m in medians.items() if m > 0}
    print()
    if not positive:
        print(
            "  ✗ NOT VERIFIED — the median is negative at every size tested.\n\n"
            "    The survey caught noise, not an edge. Quote-to-quote variation\n"
            "    here is larger than the number being measured, which is exactly\n"
            "    why one observation was never enough to act on. Do not add it.\n"
        )
        return 1

    best_size = max(positive, key=lambda s: positive[s] * s)
    best_bps = positive[best_size]
    largest_ok = max(positive)
    gross_usd = best_size * best_bps / 10_000.0

    # Is the winning median bigger than the scatter it came out of?
    #
    # Added after a 2026-07-30 run on USDC->SOL that reported "HOLDS at 1 of
    # 5" on this table:
    #
    #     10,000   median +0.159   min -0.627   max +0.401
    #
    # A spread of 1.03 bps around a median of 0.16, from three quotes
    # seconds apart. The median is real arithmetic and it is also six times
    # smaller than the noise it was computed from — with n=3 the sign of
    # that median is close to a coin flip. Printing "HOLDS" without saying
    # so invites adding a route on the strength of a rounding error, which
    # is the exact mistake this script was written to prevent.
    #
    # Not a refusal: the median is the best estimate available and it may
    # well be right. It is a statement about how much weight it can carry.
    spread = max(results[best_size]) - min(results[best_size])
    n_best = len(results[best_size])
    noisy = spread > abs(best_bps)

    print(f"  ✓ HOLDS at {len(positive)} of {len(medians)} size(s).")
    if noisy:
        print(f"    ⚠ but the {n_best} quotes at {best_size:,.0f} spanned "
              f"{spread:.3f} bps around a\n"
              f"      median of {best_bps:+.3f} — the noise is "
              f"{spread / abs(best_bps):.1f}x the edge. With n={n_best} the\n"
              f"      SIGN of that median is barely established. Re-run with "
              f"--rounds 9\n      before trusting it; if the median flips, "
              f"there was never an edge.")
    print(f"    Best expected value: {best_bps:+.3f} bps at {best_size:,.0f} "
          f"{base['symbol']} = ${gross_usd:+.4f} gross per round trip.")
    if largest_ok < max(medians):
        print(f"    Breaks down above {largest_ok:,.0f} {base['symbol']} — price "
              f"impact eats it.\n    Cap the size; do not let "
              f"SOLANA_REAL_LOAN_UNITS exceed that.")

    floor = float(os.getenv("MIN_PROFIT_FLOOR_USD", "0.20") or 0.20)
    haircut, source = effective_haircut_bps()
    print()
    if haircut >= best_bps:
        print(f"  ⚠ the leg-2 haircut is {haircut:.2f} bps ({source}) and this "
              f"edge is {best_bps:.2f} bps.\n    The haircut is subtracted from "
              f"every quote, so this route would score\n    "
              f"{best_bps - haircut:+.2f} bps and never once signal."
              + ("\n    That value is LEARNED — it rises only on a real on-chain "
                 "6024 and\n    decays back down on its own. Setting "
                 "SOLANA_LEG2_HAIRCUT_BPS will not\n    change it while "
                 "HAIRCUT_ADAPTIVE_ENABLED is on." if source == "learned" else
                 "\n    Lower the haircut below the edge, or adding the route "
                 "changes nothing."))
    if gross_usd < floor:
        print(f"  ⚠ MIN_PROFIT_FLOOR_USD is ${floor:.2f} and the best gross here is "
              f"${gross_usd:.4f}.\n    Nothing clears the floor at any size that "
              f"still quotes positive.")

    if mid["source"] != "built-in":
        print("\n  Add it with:\n")
        print(f"    ./scripts/setenv.sh \\\n"
              f"      SOLANA_EXTRA_TOKENS={mid['symbol']}:{mid['mint']}:{mid['decimals']} \\\n"
              f"      SOLANA_EXTRA_ROUTES={base['symbol']}>{mid['symbol']}")
    else:
        print(f"\n  {mid['symbol']} is already a configured token. Add the route with:\n")
        print(f"    ./scripts/setenv.sh SOLANA_EXTRA_ROUTES={base['symbol']}>{mid['symbol']}")
    print("\n  Then restart, and read /nearmiss after a few hours — a route that\n"
          "  signals and never lands is losing a race, which is a different\n"
          "  problem from one that never signals.\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))