spb/coinexplorer Public MIT
Self-hosted, zero-API-key explorer for stablecoins and major crypto.
Python 60.3%
HTML 23.6%
JavaScript 8.1%
CSS 6.8%
SQL 1%
1# Author: Simon-Pierre Boucher2# Mail: contact@spboucher.ai3"""Enrichment worker — periodic supply snapshots per token per chain.45Circulating supply is read straight from each chain (no market-data APIs):6 EVM eth_call totalSupply() (selector 0x18160ddd)7 Tron /wallet/triggerconstantcontract totalSupply()8 Solana getTokenSupply(mint)910One snapshot at startup (so the API has data immediately), then every11EXPLORER_SUPPLY_INTERVAL seconds (default hourly). Snapshots are keyed12(chain, token, timestamp) so history accumulates for supply charts.1314Mint/burn and whale detection are query-time concerns (see api/main.py):15mints are transfers FROM the zero address, burns TO it — no extra state.16"""1718import logging19import os20import time2122from . import db23from .rpc import AllEndpointsDown, RestError, RestPool, RpcError, RpcPool2425log = logging.getLogger("enrich")2627SEL_TOTAL_SUPPLY = "0x18160ddd" # keccak4("totalSupply()")2829# transfers from/to these are mints/burns (per family; Solana mints appear30# as balance increases with no sender — see adapter docs)31ZERO_ADDRESSES = {32 "evm": "0x" + "00" * 20,33 "tron": "T9yD14Nj9j7xAB4dbGeiX9h8unkKHxuWwb", # Base58Check of 0x41 + 20 zero bytes34}353637class PriceWorker:38 """USD prices for non-stable assets via CoinGecko's keyless free tier —39 ONE bulk request per refresh. Stablecoins are seeded at 1.0. If the fetch40 fails, previous prices stay in place (staleness is logged, not fatal)."""4142 def __init__(self, tokens, db_path):43 import pathlib4445 import requests as _rq46 import yaml47 self.rq = _rq48 cfg_path = pathlib.Path(__file__).parent.parent / "config" / "prices.yaml"49 cfg = yaml.safe_load(cfg_path.read_text()) if cfg_path.exists() else {}50 self.ids = cfg.get("ids", {})51 self.interval = int(cfg.get("refresh_seconds", 300))52 self.stables = sorted({53 t["symbol"] for toks in tokens.values() for t in toks54 if t.get("category", "stablecoin") == "stablecoin"55 })56 self.db_path = db_path57 self.conn = None58 self._fail_streak = 05960 def fetch(self):61 ids = ",".join(sorted(set(self.ids.values())))62 r = self.rq.get(63 "https://api.coingecko.com/api/v3/simple/price",64 params={"ids": ids, "vs_currencies": "usd"}, timeout=25,65 )66 if r.status_code == 429:67 raise RuntimeError("coingecko rate limited")68 r.raise_for_status()69 data = r.json()70 now = int(time.time())71 rows = []72 for sym, cid in self.ids.items():73 usd = (data.get(cid) or {}).get("usd")74 if usd is not None:75 rows.append((sym, float(usd), now))76 return rows7778 def run(self, stop):79 self.conn = db.connect(self.db_path)80 now = int(time.time())81 db.upsert_prices(self.conn, [(s, 1.0, now) for s in self.stables])82 while not stop.is_set():83 try:84 rows = self.fetch()85 db.upsert_prices(self.conn, rows)86 self._fail_streak = 087 log.info("prices: %d symbols refreshed", len(rows))88 except Exception as e:89 self._fail_streak += 190 log.warning("price fetch failed (%s) — keeping previous prices "91 "(%d consecutive failures)", e, self._fail_streak)92 # back off harder when the free tier pushes back93 stop.wait(self.interval * min(4, 1 + self._fail_streak))949596class StatsWorker:97 """Maintains the fast-read tables the API serves from:9899 - whale_events: incremental scan of NEW transfer rows (by rowid on100 SQLite, timestamp overlap on PG), extracting everything above the101 USD floor. /whales reads this tiny table instead of rescanning102 millions of transfer rows.103 - agg_volume: rolling volume/transfers/active-address aggregates per104 (window, symbol, chain). Heavy COUNT(DISTINCT) work happens HERE,105 off the request path, every few minutes.106 """107108 WHALE_FLOOR = float(os.environ.get("WHALE_TABLE_MIN_USD", 100_000))109 AGG_WINDOWS = {"1h": (3600, 180), "24h": (86400, 300), "7d": (604800, 1800)}110 SERIES_STEP = {"1h": 300, "24h": 3600, "7d": 21600} # matches the UI111112 def __init__(self, db_path):113 self.db_path = db_path114 self.conn = None115 self._agg_last = {w: 0.0 for w in self.AGG_WINDOWS}116117 # -- whale extraction --------------------------------------------------118119 def _whale_cursor(self):120 row = self.conn.execute(121 "SELECT last_block FROM cursors WHERE chain = '_whale_scan'"122 ).fetchone()123 return row[0] if row else 0124125 def scan_whales(self):126 last = self._whale_cursor()127 if db.is_postgres():128 # no rowid on PG: rescan a 10-min overlap; the PK dedupes129 cond, args, head = "transfers.timestamp >= ?", [int(time.time()) - 600], last130 else:131 # snapshot the head rowid FIRST — rows inserted while we scan are132 # picked up next cycle instead of being skipped forever133 head = self.conn.execute(134 "SELECT COALESCE(MAX(rowid), 0) FROM transfers").fetchone()[0]135 cond, args = "transfers.rowid > ? AND transfers.rowid <= ?", [last, head]136 rows = self.conn.execute(137 f"SELECT transfers.*, {db.USD_PRICED} AS usd "138 f"FROM transfers {db.PRICE_JOIN} WHERE {cond} AND {db.USD_PRICED} >= ?",139 args + [self.WHALE_FLOOR],140 ).fetchall()141 if rows:142 self.conn.executemany(143 'INSERT INTO whale_events (chain, block, tx_hash, log_index, timestamp, '144 'token, symbol, "from", "to", amount, decimals, usd) '145 "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING",146 [(r["chain"], r["block"], r["tx_hash"], r["log_index"], r["timestamp"],147 r["token"], r["symbol"], r["from"], r["to"], r["amount"],148 r["decimals"], round(r["usd"], 2)) for r in rows],149 )150 self.conn.execute(151 "INSERT INTO cursors (chain, last_block, last_hash) VALUES ('_whale_scan', ?, NULL) "152 "ON CONFLICT (chain) DO UPDATE SET last_block = ?",153 (head, head),154 )155 self.conn.commit()156 if rows:157 log.info("whales: +%d events (floor $%d)", len(rows), self.WHALE_FLOOR)158159 # -- rolling aggregates --------------------------------------------------160161 def compute_agg(self, window, seconds):162 since = int(time.time()) - seconds163 rows = self.conn.execute(164 "SELECT transfers.symbol AS symbol, chain, decimals, COUNT(*) AS n, "165 "SUM(CAST(amount AS DOUBLE PRECISION)) AS raw_sum, "166 'COUNT(DISTINCT "from") AS senders, COUNT(DISTINCT "to") AS receivers, '167 "COALESCE(MAX(prices.usd), 1.0) AS price "168 f"FROM transfers {db.PRICE_JOIN} WHERE timestamp >= ? "169 "GROUP BY transfers.symbol, chain, decimals",170 (since,),171 ).fetchall()172 now = int(time.time())173 acc = {}174 for r in rows:175 key = (r["symbol"], r["chain"])176 e = acc.setdefault(key, [0.0, 0, 0, 0])177 e[0] += (r["raw_sum"] or 0) / 10 ** r["decimals"] * (r["price"] or 1.0)178 e[1] += r["n"]179 e[2] += r["senders"]180 e[3] += r["receivers"]181 self.conn.execute("DELETE FROM agg_volume WHERE window = ?", (window,))182 self.conn.executemany(183 "INSERT INTO agg_volume (window, symbol, chain, volume, transfers, "184 "senders, receivers, updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",185 [(window, s, c, round(v[0], 2), v[1], v[2], v[3], now)186 for (s, c), v in acc.items()],187 )188 self.conn.commit()189 log.info("agg %s: %d (symbol, chain) rows", window, len(acc))190191 def compute_series(self, window, seconds):192 """Bucketed volume per (symbol, chain) — ONE scan covers every token,193 so the chart endpoint never aggregates on the request path."""194 step = self.SERIES_STEP[window]195 since = int(time.time()) - seconds196 rows = self.conn.execute(197 f"SELECT (timestamp / {step}) * {step} AS t, transfers.symbol AS symbol, "198 "chain, decimals, COUNT(*) AS n, "199 "SUM(CAST(amount AS DOUBLE PRECISION)) AS raw_sum, "200 "COALESCE(MAX(prices.usd), 1.0) AS price "201 f"FROM transfers {db.PRICE_JOIN} WHERE timestamp >= ? "202 "GROUP BY t, transfers.symbol, chain, decimals",203 (since,),204 ).fetchall()205 now = int(time.time())206 acc = {}207 for r in rows:208 key = (r["symbol"], r["chain"], r["t"])209 e = acc.setdefault(key, [0.0, 0])210 e[0] += (r["raw_sum"] or 0) / 10 ** r["decimals"] * (r["price"] or 1.0)211 e[1] += r["n"]212 self.conn.execute("DELETE FROM agg_series WHERE window = ?", (window,))213 self.conn.executemany(214 "INSERT INTO agg_series (window, symbol, chain, t, volume, transfers, updated) "215 "VALUES (?, ?, ?, ?, ?, ?, ?)",216 [(window, s, c, t, round(v[0], 2), v[1], now) for (s, c, t), v in acc.items()],217 )218 self.conn.commit()219 log.info("series %s: %d bucket rows", window, len(acc))220221 def run(self, stop):222 self.conn = db.connect(self.db_path)223 while not stop.is_set():224 try:225 self.scan_whales()226 except Exception as e:227 log.warning("whale scan failed: %s", e)228 now = time.monotonic()229 for window, (seconds, every) in self.AGG_WINDOWS.items():230 if now - self._agg_last[window] >= every and not stop.is_set():231 try:232 self.compute_agg(window, seconds)233 self.compute_series(window, seconds)234 self._agg_last[window] = time.monotonic()235 except Exception as e:236 log.warning("agg %s failed: %s", window, e)237 stop.wait(60)238239240class SupplyWorker:241 def __init__(self, chains, tokens, db_path):242 self.chains = chains243 self.tokens = tokens244 self.db_path = db_path245 self.conn = None246 self.interval = int(os.environ.get("EXPLORER_SUPPLY_INTERVAL", 3600))247 self._pools = {}248249 def pool(self, chain):250 if chain not in self._pools:251 cfg = self.chains[chain]252 cls = RestPool if cfg.get("family") == "tron" else RpcPool253 self._pools[chain] = cls(cfg["rpcs"])254 return self._pools[chain]255256 # -- per-family supply reads -----------------------------------------257258 def evm_supply(self, chain, token):259 res = self.pool(chain).call(260 "eth_call", [{"to": token["address"], "data": SEL_TOTAL_SUPPLY}, "latest"]261 )262 return int(res, 16) if res not in (None, "0x") else None263264 def tron_supply(self, chain, token):265 res = self.pool(chain).post("/wallet/triggerconstantcontract", {266 "owner_address": ZERO_ADDRESSES["tron"],267 "contract_address": token["id"],268 "function_selector": "totalSupply()",269 "visible": True,270 })271 out = (res or {}).get("constant_result") or []272 return int(out[0], 16) if out else None273274 def solana_supply(self, chain, token):275 res = self.pool(chain).call("getTokenSupply", [token["id"]])276 val = (res or {}).get("value") or {}277 return int(val["amount"]) if "amount" in val else None278279 READERS = {"evm": evm_supply, "tron": tron_supply, "solana": solana_supply}280281 # -- worker loop -------------------------------------------------------282283 def snapshot_once(self):284 now = int(time.time())285 rows = []286 for chain, toks in self.tokens.items():287 family = self.chains.get(chain, {}).get("family", "evm")288 reader = self.READERS.get(family)289 if reader is None:290 continue # family not indexed yet — no supply either291 for t in toks:292 try:293 supply = reader(self, chain, t)294 except (RpcError, RestError, AllEndpointsDown, ValueError) as e:295 log.warning("supply %s/%s failed: %s", chain, t["symbol"], e)296 continue297 if supply is not None:298 rows.append((chain, db.token_key(t), t["symbol"], str(supply),299 t.get("decimals"), now))300 db.insert_supply(self.conn, rows)301 log.info("supply snapshot: %d entries @ %d", len(rows), now)302 return len(rows)303304 def run(self, stop):305 self.conn = db.connect(self.db_path)306 while not stop.is_set():307 try:308 self.snapshot_once()309 except Exception as e:310 log.error("snapshot failed: %s — retrying next interval", e)311 stop.wait(self.interval)312