spb/coinexplorer Public MIT

1# Author: Simon-Pierre Boucher2# Mail: contact@spboucher.ai3"""Bitcoin adapter — whale-only native BTC transfers via the esplora REST API4(mempool.space and blockstream.info expose the same schema, giving free5keyless failover).67UTXO model: a transaction spends inputs and creates outputs — there is no8single from/to. We record each OUTPUT >= the configured minimum as one9canonical event: `to` = the output's address, `from` = the first input's10previous-output address when the tx has exactly one distinct input address11(else null — genuinely ambiguous, e.g. exchange sweep consolidations).12Coinbase (miner reward) txs have no input address. Amounts are satoshis13(8 decimals).1415Pace: ~10-minute blocks, txs fetched 25 per page — even a full block is16~150 paged calls in a 600s budget, well inside the 2 rps bucket. Reorg17safety: 1-confirmation trail + stored block-hash check with 2-block rewind.18"""1920import logging21import time2223from .. import db24from ..rpc import AllEndpointsDown, RestError, RestPool2526log = logging.getLogger("bitcoin")2728MAX_TX_PAGES = 400  # hard cap per block (~10k txs) — logged if ever hit293031class BitcoinIndexer:32    def __init__(self, chain, cfg, tokens, db_path):33        self.chain = chain34        self.cfg = cfg35        self.db_path = db_path36        self.conn = None37        self.pool = RestPool(cfg["rpcs"], timeout=30)38        self.native = cfg["native"]  # {symbol: BTC, min, decimals: 8}3940    # -- esplora helpers ---------------------------------------------------4142    def head(self):43        return int(self.pool.get("/blocks/tip/height"))4445    def hash_at(self, height):46        h = self.pool.get(f"/block-height/{height}")  # text/plain hash47        return h.strip() if isinstance(h, str) else None4849    def block_txs(self, block_hash):50        """All txs of a block, paged 25 at a time."""51        txs, start = [], 052        for _ in range(MAX_TX_PAGES):53            page = self.pool.get(f"/block/{block_hash}/txs/{start}")54            if not page:55                break56            txs.extend(page)57            if len(page) < 25:58                break59            start += 2560        else:61            log.warning("%s: block %s hit the %d-page cap — tail not scanned",62                        self.chain, block_hash[:12], MAX_TX_PAGES)63        return txs6465    # -- pipeline ----------------------------------------------------------6667    def process_block(self, height, block_hash):68        blk = self.pool.get(f"/block/{block_hash}")69        ts = blk.get("timestamp")70        min_sats = int(self.native["min"] * 10 ** self.native["decimals"])71        rows = []72        for tx in self.block_txs(block_hash):73            in_addrs = {v.get("prevout", {}).get("scriptpubkey_address")74                        for v in tx.get("vin") or [] if v.get("prevout")}75            in_addrs.discard(None)76            frm = next(iter(in_addrs)) if len(in_addrs) == 1 else None77            for oi, out in enumerate(tx.get("vout") or []):78                val = out.get("value", 0)79                to = out.get("scriptpubkey_address")80                if val < min_sats or not to:81                    continue  # below threshold or OP_RETURN/nonstandard82                if to == frm:83                    continue  # change back to the sender — not a transfer84                rows.append({85                    "chain": self.chain, "block": height, "block_hash": block_hash,86                    "tx_hash": tx["txid"], "log_index": oi, "timestamp": ts,87                    "token": "native", "symbol": self.native["symbol"],88                    "from": frm, "to": to,89                    "amount": str(val), "decimals": self.native["decimals"],90                })91        db.insert_transfers(self.conn, rows)92        return len(rows)9394    def reorged(self, height, stored_hash):95        if not stored_hash:96            return False97        h = self.hash_at(height)98        return h is not None and h != stored_hash99100    # -- main loop -----------------------------------------------------101102    def run(self, stop):103        self.conn = db.connect(self.db_path)104        cur = db.get_cursor(self.conn, self.chain)105        confirmations = self.cfg.get("confirmations", 1)106        while not stop.is_set():107            try:108                head = self.head()109                safe = head - confirmations110                if cur is None:111                    start = max(0, safe - self.cfg.get("start_offset", 3))112                    cur = (start, None, start)113                    db.set_cursor(self.conn, self.chain, start, None)114                    db.set_backfill(self.conn, self.chain, start)115                    log.info("%s: fresh start at height %d (tip %d)", self.chain, start, head)116                height, stored_hash, _ = cur117118                if self.reorged(height, stored_hash):119                    to_block = height - 2120                    log.warning("%s: reorg at %d — rolling back to %d",121                                self.chain, height, to_block)122                    db.rollback(self.conn, self.chain, to_block)123                    height, stored_hash = to_block, None124125                while height < safe and not stop.is_set():126                    nxt = height + 1127                    bh = self.hash_at(nxt)128                    if not bh:129                        break130                    n = self.process_block(nxt, bh)131                    height, stored_hash = nxt, bh132                    db.set_cursor(self.conn, self.chain, height, bh, head=head)133                    log.info("%s: %d whale outputs in block %d (tip %d)",134                             self.chain, n, height, head)135136                cur = (height, stored_hash, None)137                db.save_rpc_health(self.conn, self.chain, self.pool.stats(), int(time.time()))138                stop.wait(60)  # ~10-min blocks — checking every minute is plenty139            except (RestError, AllEndpointsDown, Exception) as e:140                log.error("%s: %s — retrying in 30s", self.chain, e)141                stop.wait(30)142