5 Python Libraries That Power My Self-Hosted Billing Monitor

When I started building BillingWatch, I had one goal: catch billing anomalies before they become customer support nightmares. A refund tsunami at 2 AM. A webhook that quietly stops firing. Subscriptions flipping to "canceled" for no obvious reason.

The commercial tools — Baremetrics, ChartMogul, Datadog — can do this, but they run $50–$400/month and you're trusting a third party with your entire billing data. I wanted something self-hosted, auditable, and free to run.

Here are the 5 Python libraries that made BillingWatch work.

1. stripe — The Foundation

pip install stripe

Stripe's official Python SDK is the backbone of the whole system. BillingWatch uses it for two things: pulling historical event data for backfill and verifying webhook signatures so we know events are actually from Stripe.

The signature verification piece is critical:

import stripe



def handle_webhook(payload: bytes, sig_header: str) -> dict:

    try:

        event = stripe.Webhook.construct_event(

            payload, sig_header, settings.STRIPE_WEBHOOK_SECRET

        )

        return event

    except stripe.error.SignatureVerificationError:

        raise ValueError("Invalid webhook signature")

Without this check, any bad actor could POST fake billing events to your endpoint. The construct_event call verifies the HMAC signature Stripe includes in every webhook request.

The SDK also handles pagination cleanly when backfilling historical charges:

events = stripe.Event.list(type="charge.failed", limit=100)

for event in stripe.auto_paging_iter(events):

    process_event(event)

2. fastapi — The API Layer

pip install fastapi uvicorn

FastAPI handles the webhook receiver, the REST API for the dashboard, and the admin endpoints. I chose it over Flask for three reasons: async support (webhooks can come in fast), automatic OpenAPI docs, and Pydantic validation that makes the event parsing bulletproof.

The webhook receiver is a single async endpoint:

from fastapi import FastAPI, Request, HTTPException

import stripe



app = FastAPI()



@app.post("/webhooks/stripe")

async def stripe_webhook(request: Request):

    payload = await request.body()

    sig_header = request.headers.get("stripe-signature")



    try:

        event = stripe.Webhook.construct_event(

            payload, sig_header, WEBHOOK_SECRET

        )

    except ValueError:

        raise HTTPException(status_code=400, detail="Invalid payload")



    await process_billing_event(event)

    return {"status": "ok"}

The async design matters because Stripe has a 30-second timeout on webhook delivery. If your handler is slow, Stripe retries — and then you're processing duplicates. FastAPI's async model keeps things fast enough that this hasn't been a problem.

3. sqlalchemy + alembic — Persistent Anomaly Storage

pip install sqlalchemy alembic

Every billing event gets persisted. SQLAlchemy handles the ORM layer, Alembic handles schema migrations. The core model is simple:

from sqlalchemy import Column, String, Float, DateTime, JSON

from database import Base



class BillingEvent(Base):

    __tablename__ = "billing_events"



    id = Column(String, primary_key=True)

    event_type = Column(String, index=True)

    amount = Column(Float)

    currency = Column(String)

    customer_id = Column(String, index=True)

    created_at = Column(DateTime)

    raw_data = Column(JSON)

    anomaly_score = Column(Float, default=0.0)

    flagged = Column(Boolean, default=False)

The anomaly_score field is populated by the detection logic — things like "this customer just triggered 5 failed charges in 10 minutes" or "refund amount exceeds original charge." The flagged column drives the dashboard alerts.

Alembic migrations mean schema changes don't break production deployments:

alembic revision --autogenerate -m "add anomaly score column"

alembic upgrade head

4. apscheduler — The Heartbeat

pip install apscheduler

BillingWatch doesn't just react to webhooks — it also runs periodic digest checks. Every hour, it scans for patterns that wouldn't be obvious from individual events: rolling refund rates, sudden drops in new subscriptions, failed charge clusters.

APScheduler makes this dead simple:

from apscheduler.schedulers.asyncio import AsyncIOScheduler



scheduler = AsyncIOScheduler()



@scheduler.scheduled_job("interval", hours=1)

async def hourly_digest():

    anomalies = await check_rolling_metrics()

    if anomalies:

        await send_alert_digest(anomalies)



scheduler.start()

The async scheduler runs inside the same FastAPI process — no Celery, no Redis, no separate worker. For a self-hosted tool running on a $5 VPS, that simplicity matters a lot.

5. httpx — Outbound Alerts

pip install httpx

When something trips an anomaly threshold, BillingWatch needs to tell you. I went with httpx over requests because it's async-native, which means alert delivery doesn't block the webhook handler.

The alert dispatcher is a small utility:

import httpx



async def send_webhook_alert(webhook_url: str, payload: dict):

    async with httpx.AsyncClient(timeout=10.0) as client:

        try:

            resp = await client.post(webhook_url, json=payload)

            resp.raise_for_status()

        except httpx.HTTPError as e:

            logger.error(f"Alert delivery failed: {e}")

This powers Slack notifications, Discord alerts, or any custom webhook endpoint. BillingWatch doesn't care where the alert goes — it just needs a URL.

Putting It Together

The full stack looks like this:

Stripe → Webhook → FastAPI receiver → SQLAlchemy storage

                                    → APScheduler digest checks

                                    → httpx alert dispatch

No containers required for local dev. Just:

git clone https://github.com/rmbell09-lang/billingwatch

cd billingwatch

pip install -r requirements.txt

cp .env.example .env  # add your STRIPE_WEBHOOK_SECRET

uvicorn main:app --reload

Why Self-Host?

The honest reason: I don't want Baremetrics or Datadog seeing every charge, refund, and subscription churn event from my business. That's the full picture of revenue health — and it belongs on infrastructure I control.

BillingWatch runs on a $6/month VPS. The only external dependency is Stripe itself.

If you're running a Stripe-based product and haven't set up anomaly monitoring, it's worth 30 minutes. The next billing anomaly you miss will cost more than that.

GitHub: https://github.com/rmbell09-lang/billingwatch