CardShield is an async fraud-scoring API built with FastAPI, Redis, and PostgreSQL. It evaluates each transaction in real time using a microservice-style pipeline: record event in Redis, build features (amount, velocity, geo), load active rules, and return a risk score.
- Exposes
POST /scoreto score transactions. - Uses Redis sorted sets for low-latency sliding-window velocity features.
- Loads active fraud rules from PostgreSQL via SQLModel.
- Returns a numeric score, triggered rules, and velocity metadata.
- Includes a latency test with a sub-60 ms median target for
/score.
- FastAPI + Uvicorn (async API server)
- SQLModel + async SQLAlchemy +
asyncpg(PostgreSQL access) redis.asyncio(Redis feature store for velocity and geo state)- Docker Compose for local infrastructure (
db,redis,api)
Run everything in Docker:
docker compose up -d --buildOr run only dependencies in Docker and API locally:
docker compose up -d db redisIf running API locally, set .env to localhost endpoints:
DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/cardshield
REDIS_URL=redis://localhost:6379/0
GEOIP_DB_PATH=data/GeoLite2-City.mmdbIf running API in Docker Compose, service hostnames are used automatically (db, redis).
python -m scripts.create_tables
python -m scripts.seed_fraud_rulesOptional one-off utility if needed:
python -m scripts.add_index_fraud_rules_is_activeHealth check:
curl -s http://localhost:8000/healthScore a transaction:
curl -s -X POST http://localhost:8000/score \
-H "Content-Type: application/json" \
-d '{"user_id":"u1","amount":"15000.00"}'Example response fields:
score: total risk score from triggered rulesrules_triggered: list of rule names that firedvelocity_1m: transaction count in the last 60 seconds
Install test dependencies:
pip install -r requirements.txt pytest httpxRun all tests:
pytest tests/ -vRun latency test only:
pytest tests/test_latency.py -v -s