from fastapi import FastAPI, Query, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import Optional, List import os, sqlite3, datetime as dt DB_PATH = os.getenv("DB_PATH", "/tmp/pol_indexer.sqlite") DEFAULT_EXCHANGE = "binance" NATIVE_TOKEN = "NATIVE" # how we mark native POL rows in `transfers` app = FastAPI(title="Polygon POL Netflow API", version="0.1.0") # CORS (open by default; tighten in prod) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # DB helpers def _conn(): # One connection per request; SQLite is fine with that c = sqlite3.connect(DB_PATH, timeout=10, check_same_thread=False) c.row_factory = sqlite3.Row return c def _get_meta(key: str) -> Optional[str]: with _conn() as c: row = c.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone() return row["value"] if row else None # Schemas class NetflowCurrent(BaseModel): exchange: str token: str cumulative_netflow: str # decimal string class NetflowPoint(BaseModel): created_at: int created_at_iso: str cumulative_value: str class TransferRow(BaseModel): tx_hash: str block_number: int from_address: str to_address: str amount_dec: str token_address: str # Routes @app.get("/") def root(): return { "message": "Polygon POL Netflow API is running", "endpoints": ["/health", "/netflow/current", "/netflow/history", "/transfers"] } @app.get("/health") def health(): # Basic DB reachability check try: with _conn() as c: c.execute("SELECT 1") ok = True except Exception as e: ok = False return {"ok": ok} @app.get("/netflow/current", response_model=NetflowCurrent) def netflow_current(exchange: str = Query(DEFAULT_EXCHANGE)): if exchange.lower() != "binance": # ready for future exchanges; for now we only maintain 'binance' raise HTTPException(status_code=400, detail="Only 'binance' is supported right now.") val = _get_meta("cumulative_netflow_dec_binance") or "0" return NetflowCurrent(exchange="binance", token="POL(native)", cumulative_netflow=val) @app.get("/netflow/history", response_model=List[NetflowPoint]) def netflow_history( since: Optional[int] = Query(None, description="Unix epoch (inclusive)"), until: Optional[int] = Query(None, description="Unix epoch (inclusive)"), limit: int = Query(200, ge=1, le=5000, description="Max rows when no time range"), ): sql = "SELECT created_at, cumulative_value FROM netflow" params = [] conds = [] if since is not None: conds.append("created_at >= ?") params.append(since) if until is not None: conds.append("created_at <= ?") params.append(until) if conds: sql += " WHERE " + " AND ".join(conds) sql += " ORDER BY created_at ASC" if since is None and until is None: sql += " LIMIT ?" params.append(limit) with _conn() as c: rows = c.execute(sql, params).fetchall() out: List[NetflowPoint] = [] for r in rows: ts = int(r["created_at"]) out.append(NetflowPoint( created_at=ts, created_at_iso=dt.datetime.utcfromtimestamp(ts).isoformat() + "Z", cumulative_value=str(r["cumulative_value"]) )) return out @app.get("/transfers", response_model=List[TransferRow]) def transfers( binance_only: bool = Query(True, description="Only rows touching Binance"), direction: Optional[str] = Query(None, pattern="^(in|out)$", description="'in' = to Binance, 'out' = from Binance"), limit: int = Query(200, ge=1, le=2000), min_block: Optional[int] = Query(None), max_block: Optional[int] = Query(None), ): """ Return recent normalized transfer rows (native POL), newest first. """ sql = """ SELECT tx_hash, block_number, from_address, to_address, amount_dec, token_address FROM transfers WHERE 1=1 """ params = list = [] BINANCE = [ "0xF977814e90dA44bFA03b6295A0616a897441aceC", "0xe7804c37c13166fF0b37F5aE0BB07A3aEbb6e245", "0x505e71695E9bc45943c58adEC1650577BcA68fD9", "0x290275e3db66394C52272398959845170E4DCb88", "0xD5C08681719445A5Fdce2Bda98b341A49050d821", "0x082489A616aB4D46d1947eE3F912e080815b08DA", ] # Binance touch filter if binance_only: placeholders = ",".join(["?"] * len(BINANCE)) # we have 6 Binance addresses sql += f" AND (from_address IN ({placeholders}) OR to_address IN ({placeholders}))" params.extend(BINANCE) params.extend(BINANCE) # Directional filter if direction == "in": sql += " AND to_address IN (" + ",".join(["?"] * 6) + ")" params.extend(BINANCE) elif direction == "out": sql += " AND from_address IN (" + ",".join(["?"] * 6) + ")" params.extend(BINANCE) if min_block is not None: sql += " AND block_number >= ?" params.append(min_block) if max_block is not None: sql += " AND block_number <= ?" params.append(max_block) sql += " ORDER BY block_number DESC, rowid DESC LIMIT ?" params.append(limit) with _conn() as c: rows = c.execute(sql, params).fetchall() return [TransferRow(**{k: str(v) if k in ("amount_dec", "token_address") else v for k, v in dict(r).items()}) for r in rows]