#!/usr/bin/env python3 import os, time, argparse, sqlite3, json from decimal import Decimal, getcontext from typing import Set, Optional from web3 import Web3 from web3.middleware import geth_poa_middleware # web3 v7+ getcontext().prec = 50 from .db import init_db, get_connection, SCHEMA_PATH, DB_PATH # and your insert/query helpers from .paths import debug_print_paths RPC_URL = os.environ.get("POLYGON_RPC", "https://polygon-rpc.com/") # six addresses because we only care about POL transfers involving these BINANCE_ADDRESSES: Set[str] = { Web3.to_checksum_address("0xF977814e90dA44bFA03b6295A0616a897441aceC"), Web3.to_checksum_address("0xe7804c37c13166fF0b37F5aE0BB07A3aEbb6e245"), Web3.to_checksum_address("0x505e71695E9bc45943c58adEC1650577BcA68fD9"), Web3.to_checksum_address("0x290275e3db66394C52272398959845170E4DCb88"), Web3.to_checksum_address("0xD5C08681719445A5Fdce2Bda98b341A49050d821"), Web3.to_checksum_address("0x082489A616aB4D46d1947eE3F912e080815b08DA"), } # We’ll store native coin rows in `transfers` with this sentinel token id: NATIVE_TOKEN = "NATIVE" # keep your schema; token_address TEXT NOT NULL def conn(): # opens sqllite db file c = sqlite3.connect(DB_PATH, timeout=30) c.row_factory = sqlite3.Row # access rows like dictionaries c.execute("PRAGMA journal_mode=WAL;") c.execute("PRAGMA synchronous=NORMAL;") return c def read_schema(): # Use SCHEMA_PATH (already resolved to /tmp/schema.sql) instead of raw "schema.sql" with open(SCHEMA_PATH, "r", encoding="utf-8") as f: return f.read() def init_db(w3: Web3, confirmations: int): # initializing db and reads the schema with conn() as c: c.executescript(read_schema()) # last_processed_block init no backfill : where to start watching the blocks rv = c.execute("SELECT value FROM meta WHERE key='last_processed_block'").fetchone() if rv is None: # If not set yet, it uses the current block number minus confirmations head = w3.eth.block_number start_at = max(0, head - confirmations) - 1 c.execute("INSERT OR REPLACE INTO meta(key,value) VALUES(?,?)", ("last_processed_block", str(start_at))) print(f"[DB] Initialized last_processed_block={start_at} (real-time only)") # cumulative init: tracks net flow at a time rv = c.execute("SELECT value FROM meta WHERE key='cumulative_netflow_dec_binance'").fetchone() if rv is None: latest = c.execute("SELECT cumulative_value FROM netflow ORDER BY created_at DESC LIMIT 1").fetchone() c.execute("INSERT OR REPLACE INTO meta(key,value) VALUES(?,?)", ("cumulative_netflow_dec_binance", latest["cumulative_value"] if latest else "0")) # tiny functions to read/write the meta table and # used to remember things like last block procressed and netflow def get_meta(key: str, default: Optional[str] = None) -> Optional[str]: with conn() as c: r = c.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone() return r["value"] if r else default def set_meta(key: str, value: str): with conn() as c: c.execute("INSERT OR REPLACE INTO meta(key,value) VALUES(?,?)", (key, str(value))) def get_last_block() -> int: # which last block processed v = get_meta("last_processed_block", "-1") return int(v) def set_last_block(n: int): # updates that block number after you finish processing set_meta("last_processed_block", str(n)) def append_transfer_native(tx_hash: str, block_number: int, log_index: int, block_time: int, frm: str, to: Optional[str], value_wei: int): if value_wei == 0: return amount_dec = (Decimal(value_wei) / Decimal(10**18)).normalize() with conn() as c: c.execute(""" INSERT OR IGNORE INTO transfers (tx_hash, block_number, log_index, block_time, from_address, to_address, amount_raw, amount_dec, token_address, received_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( tx_hash, block_number, log_index, block_time, Web3.to_checksum_address(frm), Web3.to_checksum_address(to) if to else "0x0000000000000000000000000000000000000000", str(value_wei), str(amount_dec), NATIVE_TOKEN, int(time.time()) )) # updates the running netflow number in the db def add_netflow_delta(delta_dec: Decimal): if delta_dec == 0: return cur = Decimal(get_meta("cumulative_netflow_dec_binance", "0")) new_total = cur + delta_dec ts = int(time.time()) with conn() as c: c.execute("INSERT INTO netflow(created_at, cumulative_value) VALUES (?,?)", (ts, str(new_total))) set_meta("cumulative_netflow_dec_binance", str(new_total)) return new_total # basically updated your latest step in the history and current total logbook # set up connection to the blockchain def make_web3() -> Web3: w3 = Web3(Web3.HTTPProvider(RPC_URL)) # Polygon needs PoA (proof of authority) extraData middleware in web3 v7 w3.middleware_onion.inject(geth_poa_middleware, layer=0) return w3 # reads one block from polygon and figures: # 1. did that POL move into or out of Binance # 2. if yes, move to the db and calculate netflow def process_block_native(w3: Web3, block_number: int, binance_only: bool, print_matches: bool) -> Decimal: """ Iterate full transactions in block; compute signed delta for Binance native transfers: +value if to in BINANCE, -value if from in BINANCE Persist to `transfers` table (token_address='NATIVE'). """ try: block = w3.eth.get_block(block_number, full_transactions=True) # asks web3 for block number except Exception as e: print(f"[RPC] get_block({block_number}) failed: {e}") return Decimal(0) block_time = block.timestamp delta = Decimal(0) # keeps track of the net change in Binance’s balance this block for tx in block.transactions: # sender is available as tx['from'] in web3 v7, but may require get_transaction if None frm = tx["from"] to = tx["to"] # can be None for contract creation val = int(tx["value"]) if val == 0: continue # skip value 0 # confirms if it touches Binance or not touches_binance = ((frm and Web3.to_checksum_address(frm) in BINANCE_ADDRESSES) or (to and Web3.to_checksum_address(to) in BINANCE_ADDRESSES)) # store only Binance-related if binance_only=True (default) if (not binance_only) or touches_binance: append_transfer_native( tx["hash"].hex(), block_number, tx.get("transactionIndex", 0), # or 0 if not available block_time, frm, to, val ) if touches_binance: amt_dec = Decimal(val) / Decimal(10**18) if to and Web3.to_checksum_address(to) in BINANCE_ADDRESSES: delta += amt_dec if Web3.to_checksum_address(frm) in BINANCE_ADDRESSES: delta -= amt_dec if print_matches: direction = "IN " if (to and Web3.to_checksum_address(to) in BINANCE_ADDRESSES) else "OUT" print(f"[POL(native)] blk {block_number} | {frm} -> {to} | {amt_dec} POL | {direction} Binance") return delta #Real-time loop to watch polygon def realtime_loop( binance_only: bool = True, confirmations: int = 12, poll: float = 45.0, print_matches: bool = False ): w3 = make_web3() init_db(w3, confirmations) last = get_last_block() print(f"[Indexer] start | native POL | confirmations={confirmations} | binance_only={binance_only}") while True: try: head = w3.eth.block_number except Exception as e: print("[RPC] head error:", e) time.sleep(5); continue #waits three seconds # if confirmations = 10, and head = 1000, → only process up to block 990 (because 991–1000 could still get reorged). target = head - confirmations if target > last: for blk in range(last + 1, target + 1): delta = process_block_native(w3, blk, binance_only=binance_only, print_matches=print_matches) if delta != 0: new_total = add_netflow_delta(delta) print(f"[Netflow] blk {blk} | Δ {delta} POL | cumulative {new_total} POL") set_last_block(blk) last = blk else: time.sleep(10) # this main turns raw functions into a command-line app def main(): ap = argparse.ArgumentParser(description="Polygon native POL → Binance real-time indexer") sub = ap.add_subparsers(dest="cmd", required=True) run = sub.add_parser("start", help="Run the real-time indexer (native POL)") run.add_argument("--confirmations", type=int, default=12) run.add_argument("--poll", type=float, default=1.5) run.add_argument("--all-transfers", action="store_true", help="Store ALL native transfers (default: only Binance-related)") run.add_argument("--print-matches", action="store_true", help="Print only transfers touching Binance") sub.add_parser("netflow", help="Print current cumulative netflow") args = ap.parse_args() if args.cmd == "start": realtime_loop( binance_only=(not args.all_transfers), confirmations=args.confirmations, poll=args.poll, print_matches=args.print_matches ) else: print(f"Cumulative POL netflow to Binance: {get_meta('cumulative_netflow_dec_binance','0')} POL") if __name__ == "__main__": main()