validator_stats.py

#!/usr/bin/env python3
"""
Ethereum Validator Stats - track validator performance via Beacon Chain APIs.

Fetches validator data from public beacon chain endpoints:
  - Consensus layer (beaconcha.in API, public beacon node APIs)
  - Attestation effectiveness, proposal history, balance tracking
  - Slashing detection, sync committee participation
  - Network-wide stats (active validators, staked ETH, participation rate)

No API key required for basic usage (beaconcha.in free tier: 10 req/min).
Optional BEACONCHA_API_KEY env var for higher limits.
"""

import json
import os
import sys
import time
import argparse
import statistics
from datetime import datetime, timezone
from collections import defaultdict
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
from urllib.parse import urlencode

# --- API Endpoints ---

BEACONCHA_BASE = "https://beaconcha.in/api/v1"

HEADERS = {
    "Content-Type": "application/json",
    "User-Agent": "validator-stats/1.0",
}


def api_get(path, params=None, base=None):
    """GET request to beacon chain API with rate limiting."""
    base_url = base or BEACONCHA_BASE
    url = f"{base_url}/{path}"
    if params:
        url += f"?{urlencode(params)}"

    api_key = os.environ.get("BEACONCHA_API_KEY")
    if api_key:
        HEADERS["Authorization"] = f"Bearer {api_key}"

    try:
        req = Request(url, headers=HEADERS)
        resp = urlopen(req, timeout=20)
        data = json.loads(resp.read())
        return data.get("data", data)
    except HTTPError as e:
        if e.code == 429:
            print("  [!] Rate limited, waiting 10s...", file=sys.stderr)
            time.sleep(10)
            return api_get(path, params, base)
        print(f"  [!] API error {e.code} for {path}", file=sys.stderr)
        return None
    except Exception as e:
        print(f"  [!] Request failed: {e}", file=sys.stderr)
        return None


def get_validator_by_index(index):
    """Get validator info by index."""
    return api_get(f"validator/{index}")


def get_validator_by_pubkey(pubkey):
    """Get validator info by public key."""
    return api_get(f"validator/{pubkey}")


def get_validators_batch(indices, batch_size=100):
    """Fetch multiple validators in batches."""
    results = []
    for i in range(0, len(indices), batch_size):
        batch = indices[i:i + batch_size]
        idx_str = ",".join(str(x) for x in batch)
        data = api_get(f"validator/{idx_str}")
        if data:
            if isinstance(data, list):
                results.extend(data)
            else:
                results.append(data)
        time.sleep(1.2)
    return results


def get_validator_attestations(index, limit=10):
    """Get recent attestation performance."""
    return api_get(f"validator/{index}/attestations", {"limit": limit})


def get_validator_proposals(index):
    """Get block proposal history."""
    return api_get(f"validator/{index}/proposals")


def get_validator_balance_history(index, days=30):
    """Get balance history over time."""
    return api_get(f"validator/{index}/balancehistory", {"days": days})


def get_network_stats():
    """Get current network-level stats."""
    return api_get("epoch/latest")


def get_epoch_info(epoch):
    """Get specific epoch data."""
    return api_get(f"epoch/{epoch}")


def get_sync_committee(index, epoch=None):
    """Check if validator is in sync committee."""
    path = f"validator/{index}/synccommittee"
    if epoch:
        return api_get(path, {"epoch": epoch})
    return api_get(path)


def get_slashings(limit=50):
    """Get recent slashing events."""
    return api_get("execution/slashedValidators", {"limit": limit})


def gwei_to_eth(gwei):
    """Convert Gwei balance to ETH."""
    return gwei / 1e9 if isinstance(gwei, (int, float)) else 0


# --- Validator Analysis ---

class ValidatorProfiler:
    """Profile individual validators."""

    def __init__(self, index):
        self.index = index
        self.data = None
        self.attestations = []
        self.proposals = []
        self.balance_history = []

    def load(self):
        """Fetch all validator data."""
        print(f"  Fetching validator {self.index}...", file=sys.stderr)
        self.data = get_validator_by_index(self.index)
        time.sleep(1.2)

        if not self.data:
            return False

        self.attestations = get_validator_attestations(self.index) or []
        time.sleep(1.2)

        self.proposals = get_validator_proposals(self.index) or []
        time.sleep(1.2)

        self.balance_history = get_validator_balance_history(self.index) or []
        time.sleep(1.2)

        return True

    def summary(self):
        """Generate validator profile summary."""
        if not self.data:
            return None

        d = self.data[0] if isinstance(self.data, list) else self.data

        balance_gwei = d.get("balance", 0)
        effective_balance_gwei = d.get("effectivebalance", 0)
        status = d.get("status", "unknown")
        activation_epoch = d.get("activationepoch", 0)
        exit_epoch = d.get("exitepoch", -1)
        slashed = d.get("slashed", False)

        # Attestation analysis
        att_count = len(self.attestations) if isinstance(self.attestations, list) else 0
        missed_attestations = 0
        if isinstance(self.attestations, list):
            for att in self.attestations:
                if att.get("status") == 0:
                    missed_attestations += 1

        # Proposal analysis
        prop_count = 0
        proposed_blocks = 0
        missed_proposals = 0
        if isinstance(self.proposals, list):
            prop_count = len(self.proposals)
            for p in self.proposals:
                if p.get("status") in (1, "1"):
                    proposed_blocks += 1
                else:
                    missed_proposals += 1

        # Balance trend
        balance_trend = None
        if isinstance(self.balance_history, list) and len(self.balance_history) >= 2:
            balances = [b.get("balance", 0) for b in self.balance_history]
            balance_trend = {
                "current_gwei": balances[0] if balances else 0,
                "min_gwei": min(balances),
                "max_gwei": max(balances),
                "avg_gwei": round(statistics.mean(balances)),
                "change_gwei": balances[0] - balances[-1] if len(balances) > 1 else 0,
            }

        return {
            "index": self.index,
            "pubkey": d.get("pubkey", "")[:20] + "...",
            "status": status,
            "balance_eth": round(gwei_to_eth(balance_gwei), 6),
            "effective_balance_eth": round(gwei_to_eth(effective_balance_gwei), 6),
            "activation_epoch": activation_epoch,
            "exit_epoch": exit_epoch,
            "slashed": slashed,
            "attestations_tracked": att_count,
            "missed_attestations": missed_attestations,
            "attestation_hit_rate": round(
                (1 - missed_attestations / max(att_count, 1)) * 100, 2
            ),
            "total_proposals": prop_count,
            "proposed_blocks": proposed_blocks,
            "missed_proposals": missed_proposals,
            "balance_trend": balance_trend,
        }


class NetworkStats:
    """Network-wide validator statistics."""

    def __init__(self):
        self.epoch_data = None

    def load(self):
        """Fetch network stats."""
        print("  Fetching network stats...", file=sys.stderr)
        self.epoch_data = get_network_stats()
        return self.epoch_data is not None

    def summary(self):
        """Generate network summary."""
        if not self.epoch_data:
            return None

        d = self.epoch_data
        if isinstance(d, list):
            d = d[0] if d else {}

        total_validators = d.get("validatorscount", 0)
        active_validators = d.get("activevalidators", 0)
        total_balance_gwei = d.get("totalvalidatorbalance", 0)
        participation_rate = d.get("globalparticipationrate", 0)
        epoch = d.get("epoch", 0)

        return {
            "epoch": epoch,
            "total_validators": total_validators,
            "active_validators": active_validators,
            "exited_validators": total_validators - active_validators,
            "total_staked_eth": round(gwei_to_eth(total_balance_gwei), 2),
            "participation_rate_pct": round(
                participation_rate * 100 if participation_rate < 1 else participation_rate, 2
            ),
            "average_balance_eth": round(
                gwei_to_eth(total_balance_gwei / max(active_validators, 1)), 6
            ),
        }


class BatchAnalyzer:
    """Analyze multiple validators for comparison."""

    def __init__(self):
        self.validators = []

    def add_range(self, start, count):
        """Add validator index range."""
        self.validators.extend(range(start, start + count))

    def add_list(self, indices):
        """Add specific validator indices."""
        self.validators.extend(indices)

    def analyze(self):
        """Fetch and compare all validators."""
        print(f"  Analyzing {len(self.validators)} validators...", file=sys.stderr)
        results = []

        for i, idx in enumerate(self.validators):
            profiler = ValidatorProfiler(idx)
            if profiler.load():
                summary = profiler.summary()
                if summary:
                    results.append(summary)

            if (i + 1) % 5 == 0:
                print(f"  [{i + 1}/{len(self.validators)}] done", file=sys.stderr)

        return results


class SlashingMonitor:
    """Monitor recent slashing events."""

    def __init__(self):
        self.slashings = []

    def load(self, limit=50):
        """Fetch recent slashings."""
        print("  Fetching slashing events...", file=sys.stderr)
        self.slashings = get_slashings(limit) or []
        return isinstance(self.slashings, list) and len(self.slashings) > 0

    def summary(self):
        """Generate slashing summary."""
        if isinstance(self.slashings, list):
            return {
                "total_slashings": len(self.slashings),
                "recent": self.slashings[:10],
            }
        return {"total_slashings": 0, "recent": []}


# --- Display Functions ---

def print_network_stats(stats):
    """Print network overview."""
    if not stats:
        print("  No network data available.")
        return

    print(f"\n{'=' * 60}")
    print("  ETHEREUM BEACON CHAIN - Network Overview")
    print(f"{'=' * 60}")
    print(f"  Epoch:                {stats['epoch']}")
    print(f"  Active Validators:    {stats['active_validators']:,}")
    print(f"  Total Validators:     {stats['total_validators']:,}")
    print(f"  Exited Validators:    {stats['exited_validators']:,}")
    print(f"  Total Staked:         {stats['total_staked_eth']:,.2f} ETH")
    print(f"  Avg Balance:          {stats['average_balance_eth']:.6f} ETH")
    print(f"  Participation Rate:   {stats['participation_rate_pct']:.2f}%")


def print_validator_profile(profile):
    """Print individual validator profile."""
    if not profile:
        print("  No data available.")
        return

    status_icon = {
        "active_online": "ON",
        "active_offline": "OFF",
        "exited": "EXIT",
        "slashed": "SLASH",
    }.get(profile["status"], "?")

    print(f"\n{'=' * 60}")
    print(f"  VALIDATOR #{profile['index']} - Profile")
    print(f"{'=' * 60}")
    print(f"  Status:             [{status_icon}] {profile['status']}")
    print(f"  Balance:            {profile['balance_eth']:.6f} ETH")
    print(f"  Effective Balance:  {profile['effective_balance_eth']:.6f} ETH")

    if profile["activation_epoch"] >= 0:
        print(f"  Activation Epoch:   {profile['activation_epoch']}")
    if profile["exit_epoch"] and profile["exit_epoch"] > 0:
        print(f"  Exit Epoch:         {profile['exit_epoch']}")
    if profile["slashed"]:
        print(f"  SLASHED:            YES")

    print(f"\n  Attestations:")
    print(f"    Tracked:          {profile['attestations_tracked']}")
    print(f"    Missed:           {profile['missed_attestations']}")
    print(f"    Hit Rate:         {profile['attestation_hit_rate']:.2f}%")

    print(f"\n  Proposals:")
    print(f"    Total:            {profile['total_proposals']}")
    print(f"    Proposed:         {profile['proposed_blocks']}")
    print(f"    Missed:           {profile['missed_proposals']}")

    bt = profile.get("balance_trend")
    if bt:
        print(f"\n  Balance Trend:")
        print(f"    Current:          {gwei_to_eth(bt['current_gwei']):.6f} ETH")
        print(f"    Min:              {gwei_to_eth(bt['min_gwei']):.6f} ETH")
        print(f"    Max:              {gwei_to_eth(bt['max_gwei']):.6f} ETH")
        print(f"    Avg:              {gwei_to_eth(bt['avg_gwei']):.6f} ETH")
        change = gwei_to_eth(bt["change_gwei"])
        direction = "+" if change >= 0 else ""
        print(f"    Change:           {direction}{change:.6f} ETH")


def print_batch_comparison(results):
    """Print comparison table for multiple validators."""
    if not results:
        print("  No validators to compare.")
        return

    print(f"\n{'=' * 80}")
    print("  VALIDATOR COMPARISON")
    print(f"{'=' * 80}")
    print(f"  {'Index':<10} {'Status':<16} {'Balance':>14} {'Att %':>8} {'Props':>6} {'Missed':>6} {'Slashed':>7}")
    print(f"  {'-' * 75}")

    for v in sorted(results, key=lambda x: x["index"]):
        status_short = v["status"][:14]
        print(
            f"  {v['index']:<10} {status_short:<16} {v['balance_eth']:>12.4f}  "
            f"{v['attestation_hit_rate']:>6.1f}% {v['total_proposals']:>5} "
            f"{v['missed_proposals']:>5} {'YES' if v['slashed'] else '-':>7}"
        )

    # Stats summary
    balances = [v["balance_eth"] for v in results]
    att_rates = [v["attestation_hit_rate"] for v in results]
    total_missed_att = sum(v["missed_attestations"] for v in results)
    total_missed_prop = sum(v["missed_proposals"] for v in results)
    slashed_count = sum(1 for v in results if v["slashed"])

    print(f"\n  {'-' * 75}")
    print(f"  Summary:")
    print(f"    Validators:       {len(results)}")
    print(f"    Slashed:          {slashed_count}")
    print(f"    Avg Balance:      {statistics.mean(balances):.4f} ETH")
    print(f"    Avg Att. Rate:    {statistics.mean(att_rates):.2f}%")
    print(f"    Total Missed Att: {total_missed_att}")
    print(f"    Total Missed Prop:{total_missed_prop}")


def print_slashings(slashings):
    """Print recent slashing events."""
    data = slashings.get("recent", [])
    total = slashings.get("total_slashings", 0)

    print(f"\n{'=' * 60}")
    print(f"  RECENT SLASHINGS ({total} events)")
    print(f"{'=' * 60}")

    if not data:
        print("  No recent slashing events found.")
        return

    if isinstance(data, list):
        for s in data[:10]:
            if isinstance(s, dict):
                idx = s.get("validatorindex", s.get("index", "?"))
                reason = s.get("reason", "unknown")
                epoch = s.get("epoch", "?")
                slot = s.get("slot", "?")
                print(f"  Validator #{idx} - {reason} (epoch {epoch}, slot {slot})")
            else:
                print(f"  {s}")
    else:
        print(f"  Data: {str(data)[:200]}")


# --- Main CLI ---

def main():
    parser = argparse.ArgumentParser(
        description="Ethereum Validator Stats - track validator performance via Beacon Chain APIs"
    )
    sub = parser.add_subparsers(dest="command")

    # Network overview
    sub.add_parser("network", help="Network-wide validator statistics")

    # Single validator
    p_validator = sub.add_parser("validator", help="Single validator profile")
    p_validator.add_argument("index", type=int, help="Validator index number")

    # Batch comparison
    p_batch = sub.add_parser("batch", help="Compare multiple validators")
    p_batch.add_argument("--range", type=str, help="Validator range (e.g., 100000:100020)")
    p_batch.add_argument(
        "--indices", type=str, help="Comma-separated indices (e.g., 100000,100001,100002)"
    )

    # Slashing monitor
    p_slash = sub.add_parser("slashings", help="Recent slashing events")
    p_slash.add_argument("--limit", type=int, default=50, help="Events to fetch")

    # JSON output
    parser.add_argument("--json", action="store_true", help="JSON output")

    args = parser.parse_args()
    if not args.command:
        parser.print_help()
        sys.exit(1)

    if args.command == "network":
        ns = NetworkStats()
        ns.load()
        stats = ns.summary()
        if args.json:
            print(json.dumps(stats, indent=2))
        else:
            print_network_stats(stats)

    elif args.command == "validator":
        profiler = ValidatorProfiler(args.index)
        if profiler.load():
            summary = profiler.summary()
            if args.json:
                print(json.dumps(summary, indent=2))
            else:
                print_validator_profile(summary)
        else:
            print("Failed to fetch validator data.", file=sys.stderr)
            sys.exit(1)

    elif args.command == "batch":
        analyzer = BatchAnalyzer()
        if args.range:
            parts = args.range.split(":")
            if len(parts) == 2:
                analyzer.add_range(int(parts[0]), int(parts[1]) - int(parts[0]))
        elif args.indices:
            indices = [int(x.strip()) for x in args.indices.split(",")]
            analyzer.add_list(indices)
        else:
            print("Must specify --range or --indices", file=sys.stderr)
            sys.exit(1)

        results = analyzer.analyze()
        if args.json:
            print(json.dumps(results, indent=2))
        else:
            print_batch_comparison(results)

    elif args.command == "slashings":
        sm = SlashingMonitor()
        sm.load(args.limit)
        summary = sm.summary()
        if args.json:
            print(json.dumps(summary, indent=2))
        else:
            print_slashings(summary)


if __name__ == "__main__":
    main()