scripts/arb_log_analysis.py

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

Read-only analysis tool for arb-bot / GardenAngelArbAlerts-style log text.
Does NOT execute trades, does NOT touch the bot, does NOT submit anything.
It just parses text you paste/export and tells you what actually happened.

Usage:
    python3 arb_log_analysis.py path/to/logfile.txt
    # or pipe text in:
    cat logs.txt | python3 arb_log_analysis.py -

Expected input: raw text containing lines like the Telegram alerts, e.g.

    SOL-USDC opportunity via aggregate: net ~13308 lamports (bidding 39921 tip ...)
    SOL-USDC sent ... net ~13308 · tip 39921 vs 99th floor 1000 ...
    SOL-USDC — bundle dropped — engine lost track of it (expected net ~9322 lamports)

The parser is deliberately tolerant: it looks for "sent" events (attempts,
with tip + expected net) and "dropped" events (failures), and matches them
up in order per pair. It does NOT assume any attempt that isn't explicitly
marked dropped actually landed and was profitable -- see the "unconfirmed"
bucket in the output. Only mark something a real win if you have independent
on-chain confirmation (e.g. a signature you looked up), because these alerts
alone cannot prove settlement.
"""

import re
import sys
from dataclasses import dataclass, field
from collections import defaultdict

LAMPORTS_PER_SOL = 1_000_000_000

SENT_RE = re.compile(
    r"(?P<pair>[A-Z]{2,10}-[A-Z]{2,10})\s+sent.*?"
    r"net\s*~?(?P<net>-?\d+)\s*.*?tip\s*(?P<tip>\d+)",
    re.IGNORECASE | re.DOTALL,
)

DROPPED_RE = re.compile(
    r"(?P<pair>[A-Z]{2,10}-[A-Z]{2,10}).*?bundle dropped.*?"
    r"expected net\s*~?(?P<net>-?\d+)\s*lamports",
    re.IGNORECASE | re.DOTALL,
)

TIMESTAMP_RE = re.compile(r"\d{4}-\d{2}-\d{2}T[\d:.]+Z")


@dataclass
class Attempt:
    pair: str
    tip_lamports: int
    expected_net_lamports: int
    dropped: bool = False


@dataclass
class PairStats:
    attempts: int = 0
    dropped: int = 0
    unconfirmed: int = 0  # sent, not explicitly marked dropped -- status unknown
    total_tip_lamports: int = 0
    total_expected_net_lamports: int = 0  # only counted for dropped/unconfirmed as "at risk", not "won"


def parse_log(text: str):
    """Split into blocks separated by blank-ish boundaries and classify each."""
    # Break on lines that start a new alert (rocket/magnifier/red-circle emoji lines
    # collapse in plain text, so instead we chunk on pair-name + keyword occurrences).
    attempts = []

    for m in SENT_RE.finditer(text):
        pair = m.group("pair").upper()
        net = int(m.group("net"))
        tip = int(m.group("tip"))
        attempts.append(Attempt(pair=pair, tip_lamports=tip, expected_net_lamports=net))

    dropped_events = []
    for m in DROPPED_RE.finditer(text):
        pair = m.group("pair").upper()
        net = int(m.group("net"))
        dropped_events.append((pair, net))

    # Match dropped events to the most recent unmatched "sent" attempt for that pair
    # with the same expected net (best-effort; logs don't carry a shared ID).
    used = [False] * len(attempts)
    for pair, net in dropped_events:
        matched = False
        for i in range(len(attempts) - 1, -1, -1):
            if (
                not used[i]
                and attempts[i].pair == pair
                and attempts[i].expected_net_lamports == net
            ):
                attempts[i].dropped = True
                used[i] = True
                matched = True
                break
        if not matched:
            # dropped event with no matching sent line -- record as a standalone dropped attempt
            attempts.append(Attempt(pair=pair, tip_lamports=0, expected_net_lamports=net, dropped=True))

    return attempts


def summarize(attempts):
    by_pair = defaultdict(PairStats)
    overall = PairStats()

    for a in attempts:
        s = by_pair[a.pair]
        s.attempts += 1
        overall.attempts += 1

        s.total_tip_lamports += a.tip_lamports
        overall.total_tip_lamports += a.tip_lamports

        if a.dropped:
            s.dropped += 1
            overall.dropped += 1
        else:
            # NOTE: "not explicitly dropped" != "confirmed landed profitably".
            # These alerts only show us attempts and drops; a true landed,
            # profitable fill would need on-chain confirmation to count as a win.
            s.unconfirmed += 1
            overall.unconfirmed += 1

        s.total_expected_net_lamports += a.expected_net_lamports
        overall.total_expected_net_lamports += a.expected_net_lamports

    return by_pair, overall


def fmt_sol(lamports: int) -> str:
    return f"{lamports / LAMPORTS_PER_SOL:.6f} SOL"


def print_report(by_pair, overall):
    print("=" * 70)
    print("ARB-BOT LOG ANALYSIS (read-only, no trades executed)")
    print("=" * 70)
    print()
    print(f"Total attempts parsed: {overall.attempts}")
    print(f"  Explicitly dropped:   {overall.dropped}")
    print(f"  Unconfirmed status:   {overall.unconfirmed}  "
          f"(sent, no drop message seen -- NOT proof of a profitable landed fill)")
    print()
    print(f"Total tips paid (bid, whether or not it landed): "
          f"{overall.total_tip_lamports} lamports ({fmt_sol(overall.total_tip_lamports)})")
    print(f"Total *expected* net profit across all attempts if every single one had "
          f"landed at quoted size:")
    print(f"  {overall.total_expected_net_lamports} lamports "
          f"({fmt_sol(overall.total_expected_net_lamports)})")
    print()

    if overall.dropped > 0 and overall.attempts > 0:
        drop_rate = 100 * overall.dropped / overall.attempts
        print(f"Drop rate: {drop_rate:.1f}% of attempts confirmed failed")
    print()

    print("-" * 70)
    print("By pair:")
    print("-" * 70)
    for pair, s in sorted(by_pair.items()):
        print(f"\n  {pair}")
        print(f"    attempts: {s.attempts}  dropped: {s.dropped}  unconfirmed: {s.unconfirmed}")
        print(f"    tips paid: {s.total_tip_lamports} lamports ({fmt_sol(s.total_tip_lamports)})")
        print(f"    expected net (if landed): {s.total_expected_net_lamports} lamports "
              f"({fmt_sol(s.total_expected_net_lamports)})")
        if s.total_expected_net_lamports > 0:
            ratio = s.total_tip_lamports / s.total_expected_net_lamports
            print(f"    tip / expected-net ratio: {ratio:.2f}x "
                  f"{'  <-- paying more than the edge is worth' if ratio > 1 else ''}")

    print()
    print("=" * 70)
    print("READ THIS PART")
    print("=" * 70)
    print("""
This script cannot tell you a trade was profitable just because it wasn't
marked "dropped" -- these Telegram alerts don't confirm on-chain settlement.
To get a REAL win rate:

  1. Pull the tx signatures your bot submitted (it should log these, or you
     can find them via the Jito bundle IDs).
  2. Look each one up on a Solana explorer (or via RPC getSignatureStatuses /
     getTransaction) to confirm it actually landed and check the real token
     balance delta.
  3. Only count a lamport net-positive, confirmed, landed transaction as a
     genuine win. Everything else (dropped, unconfirmed, simulation-failed)
     is a cost with zero revenue.

Based purely on what's in this log text: every attempt either explicitly
dropped, or has no confirmation it landed at all. If your total tips-paid
figure above is nonzero and you have zero confirmed on-chain profitable
fills, your realized P&L right now is AT BEST zero and likely negative
(tips are typically deducted/attempted regardless of bundle inclusion
depending on your tip payment mechanism -- check your own fee logs for
actual SOL balance changes, don't trust the bot's self-reported "net").
""")


def main():
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(1)

    src = sys.argv[1]
    if src == "-":
        text = sys.stdin.read()
    else:
        with open(src, "r", encoding="utf-8", errors="replace") as f:
            text = f.read()

    attempts = parse_log(text)
    if not attempts:
        print("No matching 'sent' or 'bundle dropped' lines found in the input.")
        print("Check that you pasted the raw alert text (including 'net ~N' and 'tip N').")
        sys.exit(0)

    by_pair, overall = summarize(attempts)
    print_report(by_pair, overall)


if __name__ == "__main__":
    main()