Python client

pip install requests

"""farbeam.py: a small client for the Farbeam API. Sample code, not an official package."""
import random
import time

import requests

DEFAULT_BASE = "https://farbeamlabs.com/api/v1"


class FarbeamError(Exception):
    def __init__(self, status, code, message, request_id=None):
        super().__init__(f"{status} {code}: {message}")
        self.status = status
        self.code = code
        self.request_id = request_id


class Farbeam:
    def __init__(self, key=None, base=DEFAULT_BASE, tries=5, timeout=30):
        self.base = base
        self.tries = tries
        self.timeout = timeout
        self.rate = None
        self.session = requests.Session()
        if key:
            self.session.headers["Authorization"] = f"Bearer {key}"

    def request(self, method, path, params=None, body=None):
        params = {k: v for k, v in (params or {}).items() if v is not None}
        last_error = None
        for attempt in range(self.tries):
            wait = None
            try:
                res = self.session.request(
                    method, self.base + path, params=params, json=body, timeout=self.timeout
                )
            except requests.RequestException as err:
                last_error = err
            else:
                if "x-ratelimit-remaining" in res.headers:
                    self.rate = {
                        "limit": int(res.headers["x-ratelimit-limit"]),
                        "remaining": int(res.headers["x-ratelimit-remaining"]),
                        "reset": int(res.headers["x-ratelimit-reset"]),
                    }
                payload = res.json()
                if res.ok:
                    return payload
                e = payload["error"]
                last_error = FarbeamError(res.status_code, e["code"], e["message"], e.get("request_id"))
                retry_after = res.headers.get("retry-after")
                # Only a unit-limit 429 sends Retry-After. Other 429 caps last far longer than a retry loop.
                if res.status_code == 429 and retry_after:
                    wait = int(retry_after)
                elif res.status_code != 502:
                    raise last_error
            if attempt + 1 < self.tries:
                if wait is None:
                    wait = min(30, 2 ** attempt) * (0.5 + random.random() / 2)
                time.sleep(wait)
        raise last_error

    def get(self, path, **params):
        return self.request("GET", path, params=params)["data"]

    def transfers(self, max_pages=None, **filters):
        """Yields rows newest first. Stop early with break, or set max_pages."""
        before = filters.pop("before", None)
        page = 0
        while max_pages is None or page < max_pages:
            payload = self.request("GET", "/transfers", params={"limit": 100, **filters, "before": before})
            yield from payload["data"]
            before = payload["meta"]["next_cursor"]
            if before is None:
                return
            page += 1

    def status(self):
        return self.get("/status")

    def chains(self):
        return self.get("/chains")

    def gas(self):
        return self.get("/gas")

    def prices(self):
        return self.get("/prices")

    def transfer_stats(self, hours=24):
        return self.get("/transfers/stats", hours=hours)

    def wallet(self, address):
        return self.get(f"/wallets/{address}")

    def wallet_activity(self, address, chain=None):
        return self.get(f"/wallets/{address}/activity", chain=chain)

    def token(self, chain, address):
        return self.get(f"/tokens/{chain}/{address}")

    def tx(self, chain, tx_hash):
        return self.get(f"/tx/{chain}/{tx_hash}")

    def labels(self, **filters):
        return self.get("/labels", **filters)

    def label(self, address):
        return self.get(f"/labels/{address}")

    def validate_rule(self, expression):
        return self.request("POST", "/rules/validate", body={"expression": expression})["data"]

    def rules(self):
        return self.get("/rules")

    def create_rule(self, name, expression, url, format="json", enabled=True, cooldown_seconds=300):
        body = {"name": name, "expression": expression, "url": url, "format": format, "enabled": enabled, "cooldown_seconds": cooldown_seconds}
        return self.request("POST", "/rules", body=body)["data"]

    def update_rule(self, rule_id, **changes):
        return self.request("PATCH", f"/rules/{rule_id}", body=changes)["data"]

    def delete_rule(self, rule_id):
        return self.request("DELETE", f"/rules/{rule_id}")["data"]

    def test_rule(self, rule_id):
        return self.request("POST", f"/rules/{rule_id}/test")["data"]

    def deliveries(self, rule_id, limit=25):
        return self.get(f"/rules/{rule_id}/deliveries", limit=limit)

import itertools, os

from farbeam import Farbeam, FarbeamError

fb = Farbeam(key=os.environ.get("FARBEAM_KEY"))

# Read the state of each chain
print({c["chain"]: c["status"] for c in fb.status()["chains"]})

# Walk two pages of transfers on Ethereum, $1m or more
total = sum(row["usd"] for row in fb.transfers(max_pages=2, chain="ethereum", min_usd=1000000))
print("sum of rows", round(total), "(rows repeat value across hops)")

# Take the first 5 rows of any walk
for row in itertools.islice(fb.transfers(token="USDT"), 5):
    print(row["ts"], row["chain"], row["usd"], row["from_label"], row["to_label"])

# Handle a known error by code
try:
    fb.label("0x1111111111111111111111111111111111111111")
except FarbeamError as err:
    if err.code != "not_found":
        raise
    print("no label for this address")

# Test a rule
check = fb.validate_rule('usd >= 5m and token in ["USDC","USDT"]')
print(check["tested"], check["matched"])

# Units left in this minute
print(fb.rate)

from datetime import datetime, timezone

def parse_ts(value):
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value / 1000, tz=timezone.utc)
    return datetime.fromisoformat(value.replace("Z", "+00:00"))

import pandas as pd

from farbeam import Farbeam

fb = Farbeam()
rows = list(fb.transfers(max_pages=3, chain="base", min_usd=500000))
frame = pd.DataFrame(rows)
frame["ts"] = pd.to_datetime(frame["ts"], utc=True)
# Largest value per transaction and token, to avoid counting hops more than once
moves = frame.groupby(["tx", "token"], as_index=False)["usd"].max()
print(moves.groupby("token")["usd"].sum().sort_values(ascending=False))