near-toll-audit / pipeline/fetch_public.py · raw
#!/usr/bin/env python3"""Keyless public pulls. Sequential, cached in raw/latest, stops a source at the first 402/403/429.A failed source keeps its previous snapshot and is listed in raw/latest/_fetch_status.json, which thedashboard shows, so a stale number can never pass for a fresh one."""import html, json, os, pathlib, re, signal, sys, time, urllib.parsesys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))from common import StopSource, http_get, http_json, load, now_utc, saveLLAMA = "https://api.llama.fi"STABLE = "https://stablecoins.llama.fi"COINS = "https://coins.llama.fi"CG = "https://api.coingecko.com/api/v3"NEARBLOCKS = "https://api.nearblocks.io/v1"PEER_COINS = ["near", "solana", "sui", "aptos", "avalanche-2", "the-open-network", "arbitrum", "thorchain", "chainflip", "internet-computer", "cosmos", "injective-protocol", "polkadot", "ethereum"]RAILS = {"near-intents": "NEAR Intents", "thorchain-dex": "THORChain", "chainflip-amm": "Chainflip", "maya-protocol": "Maya"}FEE_SLUGS = ["near-intents", "thorchain", "chainflip", "relay", "across", "debridge", "mayan-finance", "li.fi", "cow-swap"]CHAIN_FEE_NAMES = {"Solana", "Sui", "Aptos", "Avalanche", "Near", "TON", "Arbitrum", "Base", "Tron", "ICP", "Injective", "Polygon", "Ethereum", "BSC", "Sei", "Monad"}def trim_chart(d, keep=("name", "total24h", "total7d", "total30d", "total1y", "totalAllTime", "methodology", "methodologyURL")): out = {k: d.get(k) for k in keep if k in d} out["totalDataChart"] = d.get("totalDataChart") or [] return outdef intents_fees(): out = {} for kind in ("dailyFees", "dailyRevenue", "dailyHoldersRevenue", "dailySupplySideRevenue"): out[kind] = trim_chart(http_json(f"{LLAMA}/summary/fees/near-intents?dataType={kind}")) time.sleep(0.6) return outdef rail_volumes(): out = {} for slug, label in RAILS.items(): d = http_json(f"{LLAMA}/summary/dexs/{slug}") out[slug] = {"label": label, **trim_chart(d)} time.sleep(0.6) return outdef rail_fees(): out = {} for slug in FEE_SLUGS: try: d = http_json(f"{LLAMA}/summary/fees/{urllib.parse.quote(slug)}?dataType=dailyFees") except StopSource: continue out[slug] = {k: d.get(k) for k in ("name", "total30d", "total1y", "totalAllTime")} time.sleep(0.6) return outdef peer_chain_fees(): d = http_json(f"{LLAMA}/overview/fees?excludeTotalDataChart=true&excludeTotalDataChartBreakdown=true&dataType=dailyFees") rows = [{k: p.get(k) for k in ("name", "category", "total7d", "total30d", "total1y")} for p in d.get("protocols", []) if p.get("category") in ("Chain", "Rollup") and p.get("name") in CHAIN_FEE_NAMES] return {"chains": rows}def dex_market(): d = http_json(f"{LLAMA}/overview/dexs?excludeTotalDataChart=true&excludeTotalDataChartBreakdown=true") cats = {} for p in d.get("protocols", []): cats[p.get("category") or "?"] = cats.get(p.get("category") or "?", 0) + (p.get("total30d") or 0) return {"total30d": d.get("total30d"), "by_category_30d": cats}def bridge_aggregators(): d = http_json(f"{LLAMA}/overview/bridge-aggregators?excludeTotalDataChart=true&excludeTotalDataChartBreakdown=true") return {"total30d": d.get("total30d"), "protocols": [{"name": p.get("name"), "total30d": p.get("total30d")} for p in d.get("protocols", [])]}def chain_tvl(): rows = http_json(f"{LLAMA}/v2/chains") return [{k: r.get(k) for k in ("name", "tvl", "gecko_id", "tokenSymbol")} for r in rows]def stable_chains(): rows = http_json(f"{STABLE}/stablecoinchains") return [{"name": r.get("name"), "usd": (r.get("totalCirculatingUSD") or {}).get("peggedUSD")} for r in rows]def stable_near(): rows = http_json(f"{STABLE}/stablecoincharts/Near") return [{"t": int(r["date"]), "usd": (r.get("totalCirculatingUSD") or {}).get("peggedUSD")} for r in rows]def price_history(): start = 1602720000 # NEAR mainnet transfers enabled, Oct 2020 out = {} while start < time.time(): d = http_json(f"{COINS}/chart/coingecko:near?start={start}&span=400&period=1d&searchWidth=600") for p in d["coins"]["coingecko:near"]["prices"]: out[p["timestamp"] // 86400] = {"t": p["timestamp"], "p": p["price"]} start += 400 * 86400 time.sleep(0.7) return [out[k] for k in sorted(out)]def coinbase_candles(): rows = http_json("https://api.exchange.coinbase.com/products/NEAR-USD/candles?granularity=86400") return [{"t": r[0], "low": r[1], "high": r[2], "open": r[3], "close": r[4], "vol": r[5]} for r in rows]def coingecko_markets(): ids = ",".join(PEER_COINS) return http_json(f"{CG}/coins/markets?vs_currency=usd&ids={ids}&per_page=30&sparkline=false&price_change_percentage=30d,1y")def coingecko_near(): d = http_json(f"{CG}/coins/near?localization=false&tickers=false&community_data=false&developer_data=false") m = d["market_data"] return {"price": m["current_price"]["usd"], "mcap": m["market_cap"]["usd"], "fdv": m["fully_diluted_valuation"]["usd"], "circulating": m["circulating_supply"], "total": m["total_supply"], "rank": m["market_cap_rank"], "ath": m["ath"]["usd"], "ath_date": m["ath_date"]["usd"], "volume_24h": m["total_volume"]["usd"]}def nearblocks_charts(): return http_json(f"{NEARBLOCKS}/charts", headers={"Accept": "application/json"})["charts"]def nearblocks_stats(): return http_json(f"{NEARBLOCKS}/stats", headers={"Accept": "application/json"})["stats"][0]def oneclick_tokens(): return http_json("https://1click.chaindefuser.com/v0/tokens")def revenue_page(): """The official revenue dashboard ships its series inside the page. We read them as a claim to check, not as truth.""" body = http_get("https://revenue.near.org", headers={"Accept": "text/html"}).decode("utf8", "replace") chunks = re.findall(r'self\.__next_f\.push\(\[1,"([^"\\]*(?:\\.[^"\\]*)*)"\]\)', body) if not chunks: raise StopSource("page layout changed: no data payload") text = max(chunks, key=len) text = json.loads('"' + text + '"') out = {} for key in ("intentVolumeSeries", "revenueSeries", "totalFeesSeries", "tvlChartSeries", "emissionsMonthly", "absoluteRevEmissions", "stats"): m = re.search(r'"' + key + r'":\[', text) if not m: continue arr, _ = json.JSONDecoder().raw_decode(text, m.end() - 1) out[key] = arr split = re.search(r'\{"pp_fe_all_time"[^{}]{0,2000}\}', text) out["capture_split"] = json.loads(split.group(0)) if split else None wallets = re.findall(r'"([a-z0-9_.-]+\.near)"[^{}]{0,200}?"nearAmount":"?([0-9.,]+)', text) out["wallets"] = wallets claims = [c for c in re.findall(r'"children":"([^"]{20,200})"', text) if "buyback" in c.lower() or "circulation" in c.lower()] answers = re.findall(r'"question":"([^"]+)","answer":"([^"]+)"', text) out["claims"] = claims[:6] out["faq"] = [{"q": html.unescape(q), "a": html.unescape(a)} for q, a in answers] return outdef github_activity(): token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") head = {"Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"} if token: head["Authorization"] = f"Bearer {token}" repos = {"NEAR": ["near/nearcore", "near/near-sdk-rs", "near/intents", "near/mpc", "near/near-api-js", "nearai/cloud-api"], "Solana": ["anza-xyz/agave", "solana-foundation/solana-web3.js"], "Sui": ["MystenLabs/sui"], "Aptos": ["aptos-labs/aptos-core"], "Avalanche": ["ava-labs/avalanchego"], "Arbitrum": ["OffchainLabs/nitro"]} out = {} since = time.strftime("%Y-%m-%dT00:00:00Z", time.gmtime(time.time() - 90 * 86400)) for chain, names in repos.items(): for repo in names: row = {"chain": chain, "repo": repo} try: d = http_json(f"https://api.github.com/repos/{repo}/stats/participation", headers=head) weeks = d.get("all") or [] row["commits_52w"] = sum(weeks) row["commits_13w"] = sum(weeks[-13:]) row["weekly"] = weeks authors = set() for page in range(1, 6): commits = http_json(f"https://api.github.com/repos/{repo}/commits?since={since}&per_page=100&page={page}", headers=head) for c in commits: login = (c.get("author") or {}).get("login") or (c.get("commit", {}).get("author") or {}).get("name") if login and "[bot]" not in login: authors.add(login) if len(commits) < 100: break time.sleep(0.5) row["active_authors_90d"] = len(authors) row["authors_capped"] = len(commits) == 100 except StopSource as exc: row["error"] = str(exc) out[repo] = row time.sleep(0.7) return outCOMP_SLUGS = {"HYPE": ["hyperliquid-perps", "hyperliquid-spot-orderbook", "hyperliquid-l1"], "PUMP": ["pump.fun"], "JUP": ["jupiter-perpetual-exchange", "jupiter-aggregator"], "RAY": ["raydium-amm"], "AERO": ["aerodrome-slipstream", "aerodrome-v1"], "SKY": ["sky-lending"], "UNI": ["uniswap-v3"], "CAKE": ["pancakeswap-amm-v3"], "FLIP": ["chainflip-amm"], "RUNE": ["thorchain-dex"], "NEAR": ["near-intents"]}COMP_COINS = {"HYPE": "hyperliquid", "PUMP": "pump-fun", "JUP": "jupiter-exchange-solana", "RAY": "raydium", "AERO": "aerodrome-finance", "SKY": "sky", "UNI": "uniswap", "CAKE": "pancakeswap-token", "FLIP": "chainflip", "RUNE": "thorchain", "NEAR": "near"}def holder_revenue_comps(): d = http_json(f"{LLAMA}/overview/fees?excludeTotalDataChart=true&excludeTotalDataChartBreakdown=true&dataType=dailyHoldersRevenue") by_slug = {p.get("slug"): p for p in d.get("protocols", [])} time.sleep(0.7) ids = ",".join(COMP_COINS.values()) caps = {m["id"]: m for m in http_json(f"{CG}/coins/markets?vs_currency=usd&ids={ids}&per_page=30&sparkline=false")} rows = [] for sym, slugs in COMP_SLUGS.items(): cap = caps.get(COMP_COINS[sym]) if not cap: continue rows.append({"symbol": sym, "slugs": slugs, "mcap": cap.get("market_cap"), "fdv": cap.get("fully_diluted_valuation"), "holders_rev_30d": sum((by_slug.get(s) or {}).get("total30d") or 0 for s in slugs), "holders_rev_1y": sum((by_slug.get(s) or {}).get("total1y") or 0 for s in slugs)}) return rowsdef etf_holdings(): text = http_get("https://nrretf.com", headers={"Accept": "text/html"}).decode("utf8", "replace") flat = re.sub(r"<[^<>]{0,2000}>", " ", text) flat = re.sub(r"\s+", " ", flat) held = re.search(r"NEAR\s+in\s+Trust\s+([0-9][0-9,]{0,24}\.[0-9]{1,18})", flat) or re.search(r"([0-9][0-9,]{0,24}\.[0-9]{1,18})\s{0,4}NEAR\s+in\s+Trust", flat) aum = re.search(r"\$([0-9][0-9,]{6,24})", flat) if not held: raise StopSource("page layout changed: NEAR held not found") return {"near_held": float(held.group(1).replace(",", "")), "aum_usd": float(aum.group(1).replace(",", "")) if aum else None}def token_price_history(): """Weekly USD prices for every coingecko id that Intents tokens map to (public 1Click token list).""" import csv here = pathlib.Path(__file__).resolve().parent import gzip rows = list(csv.DictReader((here / "sql" / "token_prices.csv").open())) flows = here.parent / "raw" / "bigquery" / "09_intents_flows_weekly.csv.gz" weight = {} with gzip.open(flows, "rt") as fh: for r in csv.DictReader(fh): weight[r["token"]] = weight.get(r["token"], 0) + int(r["events"]) ranked = sorted(rows, key=lambda r: -weight.get(r["token"], 0)) ids = [] for r in ranked: if r["coingecko_id"] and r["coingecko_id"] not in ids and weight.get(r["token"], 0) > 0: ids.append(r["coingecko_id"]) start = 1751241600 # 2025-06-30 span = min(100, int((time.time() - start) / 604800) + 1) out = {} for cid in ids[:45]: try: d = http_json(f"{COINS}/chart/coingecko:{urllib.parse.quote(cid)}?start={start}&span={span}&period=1w&searchWidth=86400") except StopSource: continue # coverage of what stayed unpriced is reported downstream for key, val in (d.get("coins") or {}).items(): out[key.split(":", 1)[1]] = [[p["timestamp"], p["price"]] for p in val.get("prices", [])] time.sleep(0.5) if not out: raise StopSource("no token prices returned") return outJOBS = { "token_prices_weekly": (token_price_history, "coins.llama.fi/chart weekly"), "comps_holder_revenue": (holder_revenue_comps, "api.llama.fi/overview/fees + coingecko markets"), "etf_nrr": (etf_holdings, "nrretf.com"), "llama_intents_fees": (intents_fees, "api.llama.fi/summary/fees/near-intents"), "llama_rail_volumes": (rail_volumes, "api.llama.fi/summary/dexs/<rail>"), "llama_rail_fees": (rail_fees, "api.llama.fi/summary/fees/<rail>"), "llama_peer_chain_fees": (peer_chain_fees, "api.llama.fi/overview/fees"), "llama_dex_market": (dex_market, "api.llama.fi/overview/dexs"), "llama_bridge_aggregators": (bridge_aggregators, "api.llama.fi/overview/bridge-aggregators"), "llama_near_tvl": (lambda: http_json(f"{LLAMA}/v2/historicalChainTvl/Near"), "api.llama.fi/v2/historicalChainTvl/Near"), "llama_chain_tvl": (chain_tvl, "api.llama.fi/v2/chains"), "llama_near_chain_fees": (lambda: trim_chart(http_json(f"{LLAMA}/summary/fees/near?dataType=dailyFees")), "api.llama.fi/summary/fees/near"), "llama_stable_chains": (stable_chains, "stablecoins.llama.fi/stablecoinchains"), "llama_stable_near": (stable_near, "stablecoins.llama.fi/stablecoincharts/Near"), "price_near_daily": (price_history, "coins.llama.fi/chart/coingecko:near"), "coinbase_near": (coinbase_candles, "api.exchange.coinbase.com NEAR-USD candles"), "cg_markets": (coingecko_markets, "api.coingecko.com/api/v3/coins/markets"), "cg_near": (coingecko_near, "api.coingecko.com/api/v3/coins/near"), "nearblocks_charts": (nearblocks_charts, "api.nearblocks.io/v1/charts"), "nearblocks_stats": (nearblocks_stats, "api.nearblocks.io/v1/stats"), "oneclick_tokens": (oneclick_tokens, "1click.chaindefuser.com/v0/tokens"), "official_revenue_page": (revenue_page, "revenue.near.org"), "github_activity": (github_activity, "api.github.com repo stats"),}def on_alarm(*_): raise StopSource("source took too long")def main(argv): only = set(argv[1:]) status = load("_fetch_status", {}) or {} bad = 0 signal.signal(signal.SIGALRM, on_alarm) for name, (fn, source) in JOBS.items(): if only and name not in only: continue signal.alarm(240) # one hostile or slow source must not hold up the rest try: rec = save(name, fn(), source) status[name] = {"ok": True, "pulled_at": rec["pulled_at"], "sha256": rec["sha256"][:16], "source": source} print(f"ok {name}: {rec['bytes']/1024:.0f} KB") except Exception as exc: # any failure leaves the last good snapshot in place and names the source on the page bad += 1 prev = status.get(name, {}) status[name] = {"ok": False, "failed_at": now_utc(), "error": str(exc)[:160], "source": source, "pulled_at": prev.get("pulled_at")} print(f"FAIL {name}: {str(exc)[:160]}") finally: signal.alarm(0) save("_fetch_status", status, "pipeline", manifest=False) # a stall later in the run still leaves an honest status time.sleep(0.5) save("_fetch_status", status, "pipeline") print(f"{len(JOBS) - bad if not only else len(only) - bad} sources ok, {bad} failed")if __name__ == "__main__": main(sys.argv)