RiskQueue is a cloud-ready transaction risk and fraud-operations system. It preserves raw events, validates and enriches transactions, scores fraud risk, ranks cases by expected financial loss, records operational state, and publishes analytical history without making synchronous application requests depend on a warehouse.
The application combines AWS S3, Lambda, SQS, a containerized Python worker, PostgreSQL, Snowflake, Prefect, FastAPI, Streamlit, Terraform, scikit-learn, and XGBoost-ready modeling.
Important
The checked-in metrics are reproducible results from a deterministic synthetic demo. They are not PaySim results and do not establish performance for a real financial institution.
Fraud operations face two simultaneous constraints: transactions have unequal financial exposure, and analysts can investigate only a limited number of alerts. A useful system must do more than classify transactions. It must preserve source evidence, process events reliably, prevent duplicate financial records, expose current analyst state quickly, and retain analytical history for monitoring and policy evaluation.
RiskQueue separates these concerns:
- The model estimates fraud probability.
- The decision layer computes expected loss and review value.
- PostgreSQL serves operational application state.
- Snowflake stores analytical history outside the request path.
- SQS absorbs bursts and provides retry isolation.
- A dead-letter queue retains repeatedly failing messages.
- Prefect runs scheduled analytical workflows rather than event-level processing.
| Evaluation | Result |
|---|---|
| Held-out synthetic transactions | 1,749 |
| Fraud rate | 3.49% |
| Expected-loss value capture at 750 reviews | 97.2% |
| Probability-only value capture at 750 reviews | 92.3% |
| Expected-loss improvement | 4.9 percentage points |
| Amount drift PSI in shifted demo | 0.19 — watch |
| Local in-process batch API p95 at 1,000 | 10.90 ms |
Expected-loss ranking also captures 69.8% of fraud value in 100 reviews and 94.7% in 500 reviews. These values come from the checked-in generated artifacts and are not cloud throughput claims.
flowchart LR
A[Incoming transaction batch] --> B[(Amazon S3 raw events)]
B --> C[AWS Lambda ingestion]
C --> D[Amazon SQS processing queue]
D --> E[Containerized Python worker]
D -. repeated failure .-> DLQ[Dead-letter queue]
E --> F[Contract validation]
F --> G[History-only features]
G --> H[Fraud scoring]
H --> I[Expected-loss policy]
I --> J[(PostgreSQL OLTP)]
H --> K[(Snowflake OLAP)]
J --> L[Analyst review queue]
L --> M[FastAPI and Streamlit]
K --> N[Prefect batch flows]
J --> N
Transaction batches land under the S3 incoming/ prefix. Bucket versioning, server-side encryption, and public-access blocking are provisioned by Terraform. The storage adapter attaches ingestion timestamp and source metadata and uses conditional writes for immutable ingestion.
LocalObjectStore provides the same byte-oriented interface for local development and unit tests. Local writes use exclusive file creation and persist a metadata sidecar.
S3 object creation invokes a small Lambda handler. The function:
- validates the S3 event or supported request envelope;
- URL-decodes the object reference;
- creates a versioned processing message;
- derives a deterministic event identifier when one is not supplied;
- sends the message to SQS.
The Lambda function does not load models, create features, or perform database work.
Messages contain:
- schema version;
- event ID;
- batch ID;
- S3 bucket and key;
- optional object ETag and version ID;
- occurrence timestamp;
- event source.
The worker validates every message with Pydantic. A message is deleted only after successful processing. Failed messages remain visible for retry and move to the dead-letter queue after the Terraform-configured receive count.
The worker:
- reads and validates the SQS message;
- checks PostgreSQL for an existing event ID;
- retrieves the raw object;
- validates the inference transaction contract;
- builds the existing chronological behavioral features;
- uses a configured model artifact or deterministic development scorer;
- calculates expected loss and queue priority;
- writes operational transaction and review state;
- incrementally merges analytical scoring history into Snowflake when enabled;
- commits the processed-event marker and deletes the SQS message.
The worker uses the existing feature, scoring, and review-queue modules rather than maintaining separate cloud-only model logic.
SQS provides at-least-once delivery, so duplicate messages are expected. RiskQueue protects processing at two levels:
processed_events.event_idis a PostgreSQL primary key;processed_events.idempotency_keyis a deterministic unique SHA-256 key;operational_transactions.transaction_idis unique;review_queue.transaction_idis unique;- Snowflake uses
MERGEon(event_id, transaction_id).
A repeated event returns a duplicate result without producing another operational transaction, analytical fact, or analyst queue row. Database changes are committed only after all configured processing stages succeed.
- The SQS queue uses long polling and a three-minute visibility timeout.
- Terraform configures five receives before dead-letter routing by default.
- Invalid or failed messages are not deleted by the worker.
- Prefect warehouse and analytics tasks have explicit retry policies.
- Structured logs include event ID, batch ID, stage, status, retry count, and aggregate row counts.
- Logs deliberately omit transaction amounts, sender IDs, recipient IDs, and raw payloads.
PostgreSQL remains the application database. It stores:
- model versions;
- prediction audit events;
- processed event and idempotency records;
- current operational transaction scores;
- analyst review queue state;
- review rank and status.
FastAPI, Streamlit, and analyst workflows should use PostgreSQL for low-latency operational reads and writes.
Snowflake stores append-oriented historical and aggregate data:
- event and transaction scoring history;
- model version history attached to scores;
- expected-loss and decision history;
- daily model-monitoring aggregates;
- analytical reporting views.
The worker uses an idempotent MERGE. Snowflake can be disabled for local development, and FastAPI request latency does not depend on it. The schema is defined in sql/snowflake_schema.sql.
RiskQueue preserves the existing supervised binary-classification workflow:
- Logistic Regression baseline;
- XGBoost when available;
- deterministic Histogram Gradient Boosting fallback;
- numerical imputation and scaling;
- categorical one-hot encoding;
- class imbalance handling;
- strict chronological 70/15/15 evaluation;
- history-only sender, recipient, and relationship features;
- average precision, ROC-AUC, Brier score, log loss, precision, recall, and F1;
- validation-selected thresholds;
- PSI input-drift monitoring.
Leakage-prone post-transaction balance fields and isFlaggedFraud are excluded from model features. Equal-hour transactions see state frozen at the beginning of the hour.
The decision layer makes the operating objective explicit:
expected_loss = fraud_probability × transaction_amount × loss_fraction
expected_review_value = expected_loss − manual_review_cost
Analyst queues support probability, expected-loss, and expected-review-value ranking under a configurable capacity.
Prefect is used for scheduled analytics, not individual SQS events.
| Flow | Purpose |
|---|---|
historical_warehouse_sync |
Incrementally extracts recent PostgreSQL operational scores and merges them into Snowflake |
monitoring_dataset_flow |
Creates a parameterized PSI monitoring artifact from reference and current datasets |
daily_analytics_flow |
Refreshes daily Snowflake model-monitoring aggregates |
Run flows locally from Python:
python -c "from riskqueue.orchestration.flows import historical_warehouse_sync; historical_warehouse_sync(lookback_hours=24)"Prefect receives credentials through the process environment or a deployment secret mechanism. No credentials are stored in flow source.
FastAPI provides:
GET /healthPOST /v1/scorePOST /v1/score/batchPOST /v1/review-queueGET /v1/model/metrics
The six-view Streamlit dashboard covers executive results, model performance, analyst queues, decision policy, explainability, and monitoring.
The suite contains 73 tests and requires no live AWS or Snowflake account. Cloud behavior is verified through local adapters and injected clients.
Coverage includes:
- existing model, feature, API, database, queue-policy, and drift behavior;
- training and inference data contracts;
- SQS message validation and deterministic idempotency keys;
- S3 notification handling in Lambda;
- immutable local object storage and path traversal rejection;
- duplicate event processing;
- PostgreSQL rollback on failed processing;
- successful worker persistence and analytical writes;
- failed-message retention for SQS retry;
- Snowflake incremental
MERGEbehavior; - Prefect monitoring task behavior.
Run the quality gates:
uv sync --extra dev
uv run ruff check .
uv run pytest --cov=riskqueue --cov-report=term-missingCopy the safe configuration template and set a local database secret:
cp .env.example .envGenerate the existing deterministic demo:
uv sync --extra dev
uv run python -m scripts.run_demo
uv run streamlit run dashboard/app.pyRun the API:
uv run uvicorn riskqueue.api.main:app --reloadRun PostgreSQL, the API, and the dashboard without an AWS or Snowflake account:
docker compose up --build postgres api dashboardThe local object store defaults to data/raw-events/. Snowflake is disabled unless SNOWFLAKE_ENABLED=true.
The SQS worker is an optional Compose profile because it requires an actual queue endpoint:
docker compose --profile cloud up --build worker- AWS credentials supplied by an IAM role, AWS SSO, or the standard AWS credential chain;
- Terraform 1.6 or later;
- a PostgreSQL database initialized with
sql/schema.sql; - a built Lambda deployment package;
- optional Snowflake credentials supplied through environment variables or a secret manager.
Build the lightweight Lambda package:
bash scripts/build_lambda_package.shProvision AWS resources:
cd infra/terraform
terraform init
terraform fmt -check
terraform validate
terraform plan
terraform applyTerraform does not create, destroy, or mutate resources automatically from application startup. Review every plan before applying it.
Initialize Snowflake separately with an appropriately privileged deployment identity:
snowsql -f sql/snowflake_schema.sqlAttach Terraform's worker_iam_policy_arn output to the IAM role used by the worker container. Deploy the worker image from Dockerfile.worker in the container platform of your choice.
infra/terraform/ provisions:
- versioned S3 raw-event bucket;
- S3 server-side encryption;
- complete S3 public-access blocking;
- primary SQS processing queue;
- encrypted SQS dead-letter queue;
- retry/redrive policy;
- Lambda execution role and least-privilege inline policy;
- worker least-privilege IAM policy;
- Python 3.12 ingestion Lambda;
- S3-to-Lambda event notification;
- CloudWatch Logs write permissions.
Terraform state, crash logs, plan artifacts, credentials, and Lambda build output are ignored by Git.
Never commit .env, AWS keys, Snowflake passwords, database passwords, Terraform state, or generated deployment packages.
Important environment variables are documented in .env.example. In deployed environments:
- prefer IAM roles over static AWS access keys;
- inject PostgreSQL and Snowflake credentials from a secret manager;
- use TLS for database connections;
- restrict Snowflake roles to required schemas and operations;
- monitor the DLQ and worker failure logs;
- apply lifecycle and retention policies that match governance requirements.
| Path | Purpose |
|---|---|
riskqueue/cloud/ |
S3/local storage, SQS transport, message contracts, Lambda ingestion |
riskqueue/worker.py |
Idempotent asynchronous processing worker |
riskqueue/warehouse/ |
Optional Snowflake analytical sink |
riskqueue/orchestration/ |
Prefect batch and monitoring flows |
riskqueue/data/ |
Training and inference validation |
riskqueue/features/ |
Leakage-safe static and behavioral features |
riskqueue/modeling/ |
Training, calibration support, metrics, model artifacts |
riskqueue/decisions/ |
Expected loss, thresholds, review queues |
riskqueue/api/ |
FastAPI scoring and queue endpoints |
riskqueue/db/ |
SQLAlchemy operational models and sessions |
dashboard/ |
Six-view Streamlit application |
infra/terraform/ |
AWS S3, Lambda, SQS, DLQ, and IAM infrastructure |
sql/schema.sql |
PostgreSQL operational schema |
sql/snowflake_schema.sql |
Snowflake analytical schema and aggregates |
tests/ |
Local deterministic unit and integration tests |
The checked-in results use synthetic data. Historical features are batch-computed rather than maintained in an online feature store. The development API scorer remains deterministic when a trained model artifact is not configured. The project does not provision PostgreSQL, Snowflake, or the worker compute platform through the included AWS module. Costs, review capacity, and loss fraction are illustrative and must be validated for a real deployment.
The intended external modeling dataset is PaySim by E. A. Lopez-Rojas, A. Elmir, and S. Axelsson (2016). Setup and repository policy are documented in data/README.md; the dataset is not redistributed here.