#!/usr/bin/env python3 """ scripts/lighthouse.py — where do I stand, and what do I do next? venv/bin/python scripts/lighthouse.py venv/bin/python scripts/lighthouse.py --quick # skip the live probes ──────────────────────────────────────────────────────────────────────────── WHY THIS EXISTS ──────────────────────────────────────────────────────────────────────────── Operator: "we are at darkness we need go to light." There are now eleven screens — /why, /tune, /doctor, /pipeline, /bench, /gates, /capture, /decay, /routes, /decisions, /nearmiss — and each one is correct. Together they are a reading assignment. Every one answers a different question and none of them answers THE question, which is: is there money here, and if not, what is the single next thing to do about it. So this reads all of them and returns one verdict. It is deliberately opinionated. A screen that lists ten facts and lets you choose is what got us here; this ranks the facts, states which one dominates, and names one action. Where it is uncertain it says so rather than hedging into uselessness — "not measured yet" is a finding, and it usually has a command attached. ──────────────────────────────────────────────────────────────────────────── THE FIVE QUESTIONS, IN THE ORDER THEY GATE EACH OTHER ──────────────────────────────────────────────────────────────────────────── 1 CAN IT LOOK is the scanner quoting anything at all? 2 IS THERE EDGE does any round trip come back above the fees? 3 CAN IT DECIDE do quotes clear the bar the config sets? 4 CAN IT REACH does a signal survive to a signed transaction? 5 CAN IT LAND does that transaction get included? Answering 5 while 2 is unknown is the mistake this whole deployment has been making, in both bots. A landing problem is invisible while there is nothing to land. ──────────────────────────────────────────────────────────────────────────── v1.2 — THE VERDICT COMES FROM ONE PLACE (2026-08-02) ──────────────────────────────────────────────────────────────────────────── The five questions are now answered by modules.lighthouse_report.lighthouse() — the same call that serves /lighthouse in Telegram and /api/lighthouse on bot.elghaly.dev. This script used to evaluate them a second time, which meant three copies of one judgement free to disagree with each other. Two screens giving the operator different one-line answers is precisely the failure that /doctor and /why were fixed for; shipping a third copy of it inside the screen whose entire job is to be the summary would have been absurd. What stays local is the PRINTING, and it is deliberately richer than the other two surfaces: budget counts, the flash fee and dollar floor in both units, the slowest pipeline stage and how much of it is stages that should cost nothing, whether a Jito bundle has ever been submitted at all. Those belong on a terminal, where there is room, and not in a Telegram message. So: one evaluation, three renderings. The states and the verdict cannot drift. The detail can be as long as the medium allows. Read-only. Quotes only. Cannot sign, send, or spend anything but API calls. """ from __future__ import annotations import argparse import os import sys from pathlib import Path from typing import Any sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from modules.env_file import load_env_file from modules.lighthouse_report import lighthouse load_env_file() _G, _Y, _R, _B, _0 = "\033[32m", "\033[33m", "\033[31m", "\033[1m", "\033[0m" # The mark for each state, so a stage's ✓/✗ can never contradict the colour # the same stage gets in Telegram or on the web — they read one field. _MARK = { "ok": f" {_G}✓{_0} ", "warn": f" {_Y}!{_0} ", "bad": f" {_R}✗{_0} ", "unknown": " ? ", } def _safe(fn, default=None): try: return fn() except Exception: # noqa: BLE001 — a diagnostic must never raise return default def _env_f(name: str, default: float) -> float: try: raw = os.getenv(name, "").strip() return float(raw) if raw else default except (TypeError, ValueError): return default def _hdr(text: str) -> None: print(f"\n{_B}{text}{_0}") print(" " + "─" * 66) def _state(stage: dict[str, Any]) -> None: """Print the shared verdict line for a stage.""" print(_MARK.get(stage["state"], " ? ") + stage["detail"]) # ── the five questions ────────────────────────────────────────────────────── # # Each prints the shared state line, then whatever extra measurement is # worth a terminal's width. None of them decides anything. def q1_can_it_look(stage: dict[str, Any]) -> None: _hdr("1 · CAN IT LOOK — is the scanner quoting?") st = _safe(lambda: __import__( "modules.route_scorer", fromlist=["get_scorer"]).get_scorer().status(), {}) or {} used, budget = st.get("budget_used", 0), st.get("budget_total", 0) if budget: print(f" jupiter budget {used:,} / {budget:,}") if st.get("blackout") and st.get("blackout_reason"): print(f" reason {st['blackout_reason']}") _state(stage) if stage["state"] == "warn": print(" Routes are being skipped for lack of quota, not lack of merit.") def q2_is_there_edge(stage: dict[str, Any]) -> None: _hdr("2 · IS THERE EDGE — does any round trip beat the fees?") haircut = _safe(lambda: __import__( "modules.env_file", fromlist=["effective_haircut_bps"]).effective_haircut_bps()[0], 0.3) or 0.3 fee = _env_f("SOLANA_FLASH_FEE_BPS", 0.0) print(f" leg-2 haircut {haircut:.2f} bps") print(f" flash fee {fee:.2f} bps") print(f" total to beat {haircut + fee:.2f} bps") _state(stage) if stage["state"] == "bad": print(" Jupiter's router exists to erase exactly the difference this") print(" is asking it to find. Price each venue separately before") print(" concluding the edge is not there.") def q3_can_it_decide(stage: dict[str, Any]) -> None: _hdr("3 · CAN IT DECIDE — is the configured bar the real bar?") floor_usd = _env_f("MIN_PROFIT_FLOOR_USD", 0.20) floor_bps = _env_f("MIN_PROFIT_FLOOR_BPS", 0.0) probe = _env_f("SOLANA_PROBE_BASE_UNITS", 10_000) print(f" dollar floor ${floor_usd:.2f} ({floor_usd / probe * 10_000:.2f} bps " f"at a {probe:,.0f} probe)") if floor_bps: print(f" percentage floor {floor_bps:.2f} bps (${probe * floor_bps / 10_000:.2f})") _state(stage) if stage["state"] == "bad": print(" That is measured from a slow pipeline. Fix the pipeline and it") print(" falls by itself; raising the floor by hand would be treating a") print(" symptom of latency as if it were a preference.") def q4_can_it_reach(stage: dict[str, Any]) -> None: _hdr("4 · CAN IT REACH — does a signal survive to a signed transaction?") pipe = _safe(lambda: __import__( "modules.pipeline_trace", fromlist=["get_log"]).get_log().status(), {}) or {} stages = {s["stage"]: s for s in pipe.get("stages", [])} slowest = pipe.get("slowest") if slowest: print(f" slowest stage {slowest} at " f"{stages.get(slowest, {}).get('p50_ms', 0):.0f} ms") print(" a solana slot is 400 ms") _state(stage) if stage["state"] == "bad": # Stages that a warm cache should make free. When they dominate, the # answer is a restart, not a faster network. wasteful = {"freeze", "fee", "reserve", "vault", "alt", "blockhash", "tip", "pace"} waste = sum(s["p50_ms"] for n, s in stages.items() if n in wasteful) if waste > 100: print(f" {waste:.0f} ms of that is stages that should cost nothing —") print(" they load warm at startup. A restart may be the whole fix.") def q5_can_it_land(stage: dict[str, Any]) -> None: _hdr("5 · CAN IT LAND — does the transaction get included?") cap = _safe(lambda: __import__( "modules.capture_report", fromlist=["capture"]).capture(), {}) or {} print(f" signalled {cap.get('signalled', 0)} · " f"attempted {cap.get('attempted') or 0} · landed {cap.get('landed') or 0}") _state(stage) if stage["state"] == "bad": jito = _safe(lambda: __import__( "modules.jito_tip_engine", fromlist=["get_tip_engine"]).get_tip_engine().status(), {}) or {} if not jito.get("bundles_submitted"): print(" No Jito bundle has ever been submitted, so the tip auction has") print(" no landing rate to learn from. First blood mode bids the ceiling") print(" until one lands — check /jito that it is on.") def _verdict(v: dict[str, Any]) -> None: _hdr("VERDICT") state = v.get("state") if state == "ok": print(f" {_G}{v['headline']}.{_0}") print(f" {v['note']}") return if state == "unknown": # Unknown is not healthy. A green verdict derived from an absence of # evidence is the exact failure this script exists to replace. print(f" {_Y}{v['headline']}.{_0} {v['note'].split('.')[0]}.") print(" That is a finding, not a clean bill of health — start the bot,") print(" let it run a few cycles, and run this again.") return print(f" {_R}{v['headline']}.{_0}") print() print(" This is the ONE thing to fix. Everything else is either a") print(" symptom of it or cannot be judged until it is resolved.") if v.get("action"): print() print(f" {_B}next:{_0} {v['action']}") if v.get("also"): print() print(" also seen, in order — do NOT act on these yet:") for t in v["also"]: print(f" · {t}") def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--quick", action="store_true", help="skip anything that makes a network call") ap.parse_args() print() print(f"{_B} 🔦 lighthouse — where do I stand?{_0}") print(" Eleven screens, one answer. Read-only; nothing is sent.") data = lighthouse() by_key = {s["key"]: s for s in data.get("stages", [])} # A missing stage means the shared evaluator changed shape under us. # Say that rather than crashing or, worse, silently skipping a question. blank = {"state": "unknown", "detail": "stage not reported by the evaluator"} q1_can_it_look(by_key.get("looking", blank)) q2_is_there_edge(by_key.get("edge", blank)) q3_can_it_decide(by_key.get("deciding", blank)) q4_can_it_reach(by_key.get("reaching", blank)) q5_can_it_land(by_key.get("landing", blank)) _verdict(data.get("verdict") or {"state": "unknown", "headline": "Verdict unavailable", "note": "The evaluator returned nothing."}) print() return 0 if __name__ == "__main__": raise SystemExit(main())