DEV Community

Ray
Ray

Posted on

Self-Hosted Stripe Billing Anomaly Detection with FastAPI

Self-Hosted Stripe Billing Anomaly Detection with FastAPI

Billing anomalies are silent killers. Failed charges that go unnoticed, webhook delivery gaps, sudden MRR drops at 2 AM — by the time you see them in your dashboard, the customer is already gone.

In this post I'll walk through BillingWatch — a self-hosted FastAPI service that listens to Stripe webhooks and fires alerts the moment something looks wrong.

The Problem

Stripe's own dashboard is reactive. You find out about billing issues after they've happened, often hours later. What you want is a system that:

  • Detects failed charges in real time
  • Spots unusual refund spikes
  • Alerts when subscription churn exceeds baseline
  • Runs on your infrastructure (no SaaS subscription)

Architecture

Stripe Webhooks → FastAPI /webhook → Anomaly Engine → Alerts (WhatsApp / Slack / email)
Enter fullscreen mode Exit fullscreen mode

The service is intentionally minimal — a single FastAPI app, SQLite for event storage, and a threshold-based anomaly detector.

Webhook Handler

from fastapi import FastAPI, Request, HTTPException
import stripe, os

app = FastAPI()
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]

@app.post("/webhook")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig = request.headers.get("stripe-signature", "")

    try:
        event = stripe.Webhook.construct_event(payload, sig, WEBHOOK_SECRET)
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    if event["type"] == "invoice.payment_failed":
        await handle_payment_failure(event["data"]["object"])

    return {"status": "ok"}
Enter fullscreen mode Exit fullscreen mode

Anomaly Detection

The anomaly engine compares current event rates against a rolling 7-day baseline:

def is_anomalous(event_type: str, window_hours: int = 1) -> bool:
    recent = count_events(event_type, hours=window_hours)
    baseline = avg_events_per_hour(event_type, days=7)

    # Alert if rate is 3x above baseline
    return recent > (baseline * 3 + 1)
Enter fullscreen mode Exit fullscreen mode

This catches spikes in refunds, failed charges, and subscription cancellations without requiring ML infrastructure.

Running It

git clone https://github.com/rmbell09-lang/BillingWatch
cd BillingWatch
pip install -r requirements.txt
export STRIPE_SECRET_KEY=sk_live_...
export STRIPE_WEBHOOK_SECRET=whsec_...
uvicorn main:app --host 0.0.0.0 --port 8000
Enter fullscreen mode Exit fullscreen mode

Then point your Stripe webhook dashboard at https://yourdomain.com/webhook.

Why Self-Hosted?

No per-seat pricing, no SaaS lock-in, full control over your alert logic. For indie developers and small teams watching burn rate, that matters.

Full source, Docker setup, and deployment docs on GitHub.


Top comments (0)