scripts/lighthouse.py

#!/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())