src/api.py

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]