Docs / SDK guides

Recipe: exchange flows

This recipe totals stored transfers to and from labeled exchange wallets. It prints one line per exchange name: value in, value out, and the row count. Run it from any machine with no key. A key lets it read more pages per minute.

#The method

Walk /transfers for a period.
Keep rows where from_type or to_type is exchange.
Skip rows where both sides have the same exchange group, because those are likely internal moves.
Group by the exchange name, with the trailing number removed, so Binance 14 and Binance 8 join.
Remove repeated hops: within one transaction and token, count only the largest value per direction.

#Code

# Quick look at one page: labeled exchange recipients on Ethereum
curl -s "https://farbeamlabs.com/api/v1/transfers?chain=ethereum&limit=100" \
  | jq -r '.data[] | select(.to_type == "exchange") | [.ts, .to_label, .token, (.usd | floor)] | @tsv'

// flows.mjs: run with node flows.mjs
const HOURS = 6;
const since = Date.now() - HOURS * 3600000;
const headers = process.env.FARBEAM_KEY ? { Authorization: `Bearer ${process.env.FARBEAM_KEY}` } : {};
const group = (label) => label.replace(/\s+\d+$/, "");

const best = new Map();
let before = null;
for (let page = 0; page < 20; page++) {
  const query = new URLSearchParams({ limit: "100" });
  if (before) query.set("before", String(before));
  const res = await fetch(`https://farbeamlabs.com/api/v1/transfers?${query}`, { headers });
  const { data, meta, error } = await res.json();
  if (error) throw new Error(`${error.code}: ${error.message}`);
  for (const row of data) {
    if (Date.parse(row.ts) < since || row.sampled) continue;
    const from = row.from_type === "exchange" ? group(row.from_label) : null;
    const to = row.to_type === "exchange" ? group(row.to_label) : null;
    if ((!from && !to) || from === to) continue;
    for (const [name, side] of [[from, "out"], [to, "in"]]) {
      if (!name) continue;
      const key = `${name}|${side}|${row.tx}|${row.token}`;
      best.set(key, Math.max(best.get(key) ?? 0, row.usd));
    }
  }
  const oldest = data[data.length - 1];
  if (!oldest || Date.parse(oldest.ts) < since || meta.next_cursor === null) break;
  before = meta.next_cursor;
  await new Promise((resolve) => setTimeout(resolve, 2100));
}

const totals = {};
for (const [key, usd] of best) {
  const [name, side] = key.split("|");
  totals[name] ??= { in: 0, out: 0, rows: 0 };
  totals[name][side] += usd;
  totals[name].rows += 1;
}
console.table(totals);

# flows.py: run with python3 flows.py
import os, re, time
from datetime import datetime, timedelta, timezone
import requests

since = datetime.now(timezone.utc) - timedelta(hours=6)
headers = {"Authorization": f"Bearer {os.environ['FARBEAM_KEY']}"} if os.environ.get("FARBEAM_KEY") else {}
group = lambda label: re.sub(r"\s+\d+$", "", label)
when = lambda row: datetime.fromisoformat(row["ts"].replace("Z", "+00:00"))

best, before = {}, None
for page in range(20):
    params = {"limit": 100}
    if before:
        params["before"] = before
    body = requests.get("https://farbeamlabs.com/api/v1/transfers", params=params, headers=headers, timeout=30).json()
    if "error" in body:
        raise RuntimeError(body["error"]["code"])
    for row in body["data"]:
        if when(row) < since or row["sampled"]:
            continue
        sender = group(row["from_label"]) if row["from_type"] == "exchange" else None
        recipient = group(row["to_label"]) if row["to_type"] == "exchange" else None
        if (not sender and not recipient) or sender == recipient:
            continue
        for name, side in ((sender, "out"), (recipient, "in")):
            if name:
                key = (name, side, row["tx"], row["token"])
                best[key] = max(best.get(key, 0), row["usd"])
    rows = body["data"]
    if not rows or when(rows[-1]) < since or body["meta"]["next_cursor"] is None:
        break
    before = body["meta"]["next_cursor"]
    time.sleep(2.1)

totals = {}
for (name, side, _, _), usd in best.items():
    t = totals.setdefault(name, {"in": 0, "out": 0, "rows": 0})
    t[side] += usd
    t["rows"] += 1
for name, t in sorted(totals.items(), key=lambda x: -(x[1]["in"] + x[1]["out"])):
    print(f"{name:20} in {t['in']:>14,.0f}  out {t['out']:>14,.0f}  rows {t['rows']}")

The pause of 2.1 seconds between pages stays under the anonymous limit of 30 units per minute.

#Get it as an alert

To know when such rows arrive, save a rule. A rule fires once per cooldown, so read the rows with the code above when it fires.

Beamline

(to_type == "exchange" or from_type == "exchange") and usd >= 1m and not sampled

#How to read the output

What you may see	What it supports
High "in" for one name	Labeled wallets of that exchange received value in this window
No row for an exchange	No labeled wallet moved at this size. It does not mean no activity.
Value in and out near equal	Could be treasury moves, market makers or customers. The data cannot tell.
A change from yesterday	A change in labeled, large, curated-token rows only

Do not treat "in" as selling or "out" as accumulation. See transfers vs trades. This recipe is not a trading signal, and nothing here is financial advice.

#Ways to extend it

Store each run's rows, because rows expire after 7 days.
Add a filter by token, such as USDT, to look at stablecoins alone.
Compare to the stablecoin monitor, which reads supply by chain.

PreviousStreaming NextRecipe: token screening