# Author: Simon-Pierre Boucher # Mail: contact@spboucher.ai """Enrichment worker — periodic supply snapshots per token per chain. Circulating supply is read straight from each chain (no market-data APIs): EVM eth_call totalSupply() (selector 0x18160ddd) Tron /wallet/triggerconstantcontract totalSupply() Solana getTokenSupply(mint) One snapshot at startup (so the API has data immediately), then every EXPLORER_SUPPLY_INTERVAL seconds (default hourly). Snapshots are keyed (chain, token, timestamp) so history accumulates for supply charts. Mint/burn and whale detection are query-time concerns (see api/main.py): mints are transfers FROM the zero address, burns TO it — no extra state. """ import logging import os import time from . import db from .rpc import AllEndpointsDown, RestError, RestPool, RpcError, RpcPool log = logging.getLogger("enrich") SEL_TOTAL_SUPPLY = "0x18160ddd" # keccak4("totalSupply()") # transfers from/to these are mints/burns (per family; Solana mints appear # as balance increases with no sender — see adapter docs) ZERO_ADDRESSES = { "evm": "0x" + "00" * 20, "tron": "T9yD14Nj9j7xAB4dbGeiX9h8unkKHxuWwb", # Base58Check of 0x41 + 20 zero bytes } class PriceWorker: """USD prices for non-stable assets via CoinGecko's keyless free tier — ONE bulk request per refresh. Stablecoins are seeded at 1.0. If the fetch fails, previous prices stay in place (staleness is logged, not fatal).""" def __init__(self, tokens, db_path): import pathlib import requests as _rq import yaml self.rq = _rq cfg_path = pathlib.Path(__file__).parent.parent / "config" / "prices.yaml" cfg = yaml.safe_load(cfg_path.read_text()) if cfg_path.exists() else {} self.ids = cfg.get("ids", {}) self.interval = int(cfg.get("refresh_seconds", 300)) self.stables = sorted({ t["symbol"] for toks in tokens.values() for t in toks if t.get("category", "stablecoin") == "stablecoin" }) self.db_path = db_path self.conn = None self._fail_streak = 0 def fetch(self): ids = ",".join(sorted(set(self.ids.values()))) r = self.rq.get( "https://api.coingecko.com/api/v3/simple/price", params={"ids": ids, "vs_currencies": "usd"}, timeout=25, ) if r.status_code == 429: raise RuntimeError("coingecko rate limited") r.raise_for_status() data = r.json() now = int(time.time()) rows = [] for sym, cid in self.ids.items(): usd = (data.get(cid) or {}).get("usd") if usd is not None: rows.append((sym, float(usd), now)) return rows def run(self, stop): self.conn = db.connect(self.db_path) now = int(time.time()) db.upsert_prices(self.conn, [(s, 1.0, now) for s in self.stables]) while not stop.is_set(): try: rows = self.fetch() db.upsert_prices(self.conn, rows) self._fail_streak = 0 log.info("prices: %d symbols refreshed", len(rows)) except Exception as e: self._fail_streak += 1 log.warning("price fetch failed (%s) — keeping previous prices " "(%d consecutive failures)", e, self._fail_streak) # back off harder when the free tier pushes back stop.wait(self.interval * min(4, 1 + self._fail_streak)) class StatsWorker: """Maintains the fast-read tables the API serves from: - whale_events: incremental scan of NEW transfer rows (by rowid on SQLite, timestamp overlap on PG), extracting everything above the USD floor. /whales reads this tiny table instead of rescanning millions of transfer rows. - agg_volume: rolling volume/transfers/active-address aggregates per (window, symbol, chain). Heavy COUNT(DISTINCT) work happens HERE, off the request path, every few minutes. """ WHALE_FLOOR = float(os.environ.get("WHALE_TABLE_MIN_USD", 100_000)) AGG_WINDOWS = {"1h": (3600, 180), "24h": (86400, 300), "7d": (604800, 1800)} SERIES_STEP = {"1h": 300, "24h": 3600, "7d": 21600} # matches the UI def __init__(self, db_path): self.db_path = db_path self.conn = None self._agg_last = {w: 0.0 for w in self.AGG_WINDOWS} # -- whale extraction -------------------------------------------------- def _whale_cursor(self): row = self.conn.execute( "SELECT last_block FROM cursors WHERE chain = '_whale_scan'" ).fetchone() return row[0] if row else 0 def scan_whales(self): last = self._whale_cursor() if db.is_postgres(): # no rowid on PG: rescan a 10-min overlap; the PK dedupes cond, args, head = "transfers.timestamp >= ?", [int(time.time()) - 600], last else: # snapshot the head rowid FIRST — rows inserted while we scan are # picked up next cycle instead of being skipped forever head = self.conn.execute( "SELECT COALESCE(MAX(rowid), 0) FROM transfers").fetchone()[0] cond, args = "transfers.rowid > ? AND transfers.rowid <= ?", [last, head] rows = self.conn.execute( f"SELECT transfers.*, {db.USD_PRICED} AS usd " f"FROM transfers {db.PRICE_JOIN} WHERE {cond} AND {db.USD_PRICED} >= ?", args + [self.WHALE_FLOOR], ).fetchall() if rows: self.conn.executemany( 'INSERT INTO whale_events (chain, block, tx_hash, log_index, timestamp, ' 'token, symbol, "from", "to", amount, decimals, usd) ' "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING", [(r["chain"], r["block"], r["tx_hash"], r["log_index"], r["timestamp"], r["token"], r["symbol"], r["from"], r["to"], r["amount"], r["decimals"], round(r["usd"], 2)) for r in rows], ) self.conn.execute( "INSERT INTO cursors (chain, last_block, last_hash) VALUES ('_whale_scan', ?, NULL) " "ON CONFLICT (chain) DO UPDATE SET last_block = ?", (head, head), ) self.conn.commit() if rows: log.info("whales: +%d events (floor $%d)", len(rows), self.WHALE_FLOOR) # -- rolling aggregates -------------------------------------------------- def compute_agg(self, window, seconds): since = int(time.time()) - seconds rows = self.conn.execute( "SELECT transfers.symbol AS symbol, chain, decimals, COUNT(*) AS n, " "SUM(CAST(amount AS DOUBLE PRECISION)) AS raw_sum, " 'COUNT(DISTINCT "from") AS senders, COUNT(DISTINCT "to") AS receivers, ' "COALESCE(MAX(prices.usd), 1.0) AS price " f"FROM transfers {db.PRICE_JOIN} WHERE timestamp >= ? " "GROUP BY transfers.symbol, chain, decimals", (since,), ).fetchall() now = int(time.time()) acc = {} for r in rows: key = (r["symbol"], r["chain"]) e = acc.setdefault(key, [0.0, 0, 0, 0]) e[0] += (r["raw_sum"] or 0) / 10 ** r["decimals"] * (r["price"] or 1.0) e[1] += r["n"] e[2] += r["senders"] e[3] += r["receivers"] self.conn.execute("DELETE FROM agg_volume WHERE window = ?", (window,)) self.conn.executemany( "INSERT INTO agg_volume (window, symbol, chain, volume, transfers, " "senders, receivers, updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", [(window, s, c, round(v[0], 2), v[1], v[2], v[3], now) for (s, c), v in acc.items()], ) self.conn.commit() log.info("agg %s: %d (symbol, chain) rows", window, len(acc)) def compute_series(self, window, seconds): """Bucketed volume per (symbol, chain) — ONE scan covers every token, so the chart endpoint never aggregates on the request path.""" step = self.SERIES_STEP[window] since = int(time.time()) - seconds rows = self.conn.execute( f"SELECT (timestamp / {step}) * {step} AS t, transfers.symbol AS symbol, " "chain, decimals, COUNT(*) AS n, " "SUM(CAST(amount AS DOUBLE PRECISION)) AS raw_sum, " "COALESCE(MAX(prices.usd), 1.0) AS price " f"FROM transfers {db.PRICE_JOIN} WHERE timestamp >= ? " "GROUP BY t, transfers.symbol, chain, decimals", (since,), ).fetchall() now = int(time.time()) acc = {} for r in rows: key = (r["symbol"], r["chain"], r["t"]) e = acc.setdefault(key, [0.0, 0]) e[0] += (r["raw_sum"] or 0) / 10 ** r["decimals"] * (r["price"] or 1.0) e[1] += r["n"] self.conn.execute("DELETE FROM agg_series WHERE window = ?", (window,)) self.conn.executemany( "INSERT INTO agg_series (window, symbol, chain, t, volume, transfers, updated) " "VALUES (?, ?, ?, ?, ?, ?, ?)", [(window, s, c, t, round(v[0], 2), v[1], now) for (s, c, t), v in acc.items()], ) self.conn.commit() log.info("series %s: %d bucket rows", window, len(acc)) def run(self, stop): self.conn = db.connect(self.db_path) while not stop.is_set(): try: self.scan_whales() except Exception as e: log.warning("whale scan failed: %s", e) now = time.monotonic() for window, (seconds, every) in self.AGG_WINDOWS.items(): if now - self._agg_last[window] >= every and not stop.is_set(): try: self.compute_agg(window, seconds) self.compute_series(window, seconds) self._agg_last[window] = time.monotonic() except Exception as e: log.warning("agg %s failed: %s", window, e) stop.wait(60) class SupplyWorker: def __init__(self, chains, tokens, db_path): self.chains = chains self.tokens = tokens self.db_path = db_path self.conn = None self.interval = int(os.environ.get("EXPLORER_SUPPLY_INTERVAL", 3600)) self._pools = {} def pool(self, chain): if chain not in self._pools: cfg = self.chains[chain] cls = RestPool if cfg.get("family") == "tron" else RpcPool self._pools[chain] = cls(cfg["rpcs"]) return self._pools[chain] # -- per-family supply reads ----------------------------------------- def evm_supply(self, chain, token): res = self.pool(chain).call( "eth_call", [{"to": token["address"], "data": SEL_TOTAL_SUPPLY}, "latest"] ) return int(res, 16) if res not in (None, "0x") else None def tron_supply(self, chain, token): res = self.pool(chain).post("/wallet/triggerconstantcontract", { "owner_address": ZERO_ADDRESSES["tron"], "contract_address": token["id"], "function_selector": "totalSupply()", "visible": True, }) out = (res or {}).get("constant_result") or [] return int(out[0], 16) if out else None def solana_supply(self, chain, token): res = self.pool(chain).call("getTokenSupply", [token["id"]]) val = (res or {}).get("value") or {} return int(val["amount"]) if "amount" in val else None READERS = {"evm": evm_supply, "tron": tron_supply, "solana": solana_supply} # -- worker loop ------------------------------------------------------- def snapshot_once(self): now = int(time.time()) rows = [] for chain, toks in self.tokens.items(): family = self.chains.get(chain, {}).get("family", "evm") reader = self.READERS.get(family) if reader is None: continue # family not indexed yet — no supply either for t in toks: try: supply = reader(self, chain, t) except (RpcError, RestError, AllEndpointsDown, ValueError) as e: log.warning("supply %s/%s failed: %s", chain, t["symbol"], e) continue if supply is not None: rows.append((chain, db.token_key(t), t["symbol"], str(supply), t.get("decimals"), now)) db.insert_supply(self.conn, rows) log.info("supply snapshot: %d entries @ %d", len(rows), now) return len(rows) def run(self, stop): self.conn = db.connect(self.db_path) while not stop.is_set(): try: self.snapshot_once() except Exception as e: log.error("snapshot failed: %s — retrying next interval", e) stop.wait(self.interval)