CashSense — Real-Time MSME Financial Data Pipeline

An end-to-end, real-time financial data pipeline for Micro, Small & Medium Enterprises (MSMEs) in India.

---

What Is CashSense?

CashSense is the data engineering backbone that makes AI-powered financial insights production-ready for MSMEs. It bridges raw, multi-channel transaction data — from WhatsApp payments, UPI transfers, and bank exports — into a unified, structured, and analytically queryable data store. It then surfaces those insights through an interactive BI dashboard with real-time anomaly alerts.

Stack: Python · Apache Kafka · Pandas · dbt · DuckDB · Scikit-learn · Facebook Prophet · Apache Superset · GitHub Actions · Docker

---

The Problem

MSMEs in India operate across fragmented financial channels — WhatsApp payments, UPI transactions, and bank CSV exports. This means:

• No unified cashflow view across channels
• No early warning system for unusual spending or cash flow drops
• Manual, error-prone reconciliation of transactions across platforms
• No automated alerts when financial health degrades

CashSense solves this by building a fully automated data pipeline that ingests, cleanses, transforms, and analyses financial data in real time — with zero manual intervention post-deployment.

---

How It Works

1. Ingest — Synthetic transaction data (mimicking WhatsApp, UPI, bank exports) is produced to Apache Kafka topics.
2. Cleanse — A Kafka consumer reads messages, applies Pandas-based cleaning (date normalisation, currency conversion, deduplication) and writes to DuckDB's raw schema.
3. Transform — dbt runs SQL transformation models to produce daily_cashflow, weekly_summary, and anomaly_flags tables in DuckDB's analytics schema.
4. Detect — Two ML models run against the transformed data:
   • Facebook Prophet forecasts daily cashflow per channel and flags deviations > 2 standard deviations.
   • Isolation Forest scores individual transactions for point anomalies.
5. Visualise — Apache Superset connects to DuckDB via SQLAlchemy and displays live cashflow trends, anomaly alerts, and pipeline health — refreshing every 60 seconds.
6. Automate — GitHub Actions runs dbt tests on every push, and retrains ML models every Sunday at midnight.

---

Architecture

┌─────────────────────────────────────────────────────────────┐
│                        DATA SOURCES                          │
│  [WhatsApp Business API] [UPI CSV Exports] [OpenExchangeRates│
└─────────────────┬───────────────┬───────────────┬───────────┘
                  │               │               │
                  ▼               ▼               ▼
┌─────────────────────────────────────────────────────────────┐
│                     INGESTION LAYER                          │
│              Apache Kafka (3 Topics)                         │
│  [whatsapp-transactions] [upi-transactions] [currency-rates] │
└─────────────────────────────┬───────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                   STREAM PROCESSING                          │
│              Python Kafka Consumers + Pandas                 │
│     Cleanse → Normalise → Deduplicate → Validate            │
└─────────────────────────────┬───────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                     TRANSFORMATION                           │
│                    dbt (Data Build Tool)                     │
│        daily_cashflow | weekly_summary | anomaly_flags       │
└─────────────────────────────┬───────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                       STORAGE                                │
│                       DuckDB                                 │
│         raw schema (ingested) | analytics schema (modelled)  │
└──────────────────┬──────────────────────┬───────────────────┘
                   │                      │
                   ▼                      ▼
┌──────────────────────────┐  ┌──────────────────────────────┐
│      ML LAYER            │  │      VISUALISATION            │
│  Prophet (forecasting)   │  │  Apache Superset Dashboard    │
│  Isolation Forest        │  │  BI Charts + Anomaly Alerts   │
│  (anomaly detection)     │  │  Refreshes every 60 seconds   │
└──────────────────────────┘  └──────────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────────────────────┐
│                    CI/CD (GitHub Actions)                    │
│        dbt tests on every push | ML retraining weekly        │
└─────────────────────────────────────────────────────────────┘

Data Flow (Example)

Stage	Data
Raw (Kafka message)	{ "source": "upi", "amount": 5000, "direction": "inflow", "timestamp": "2026-04-15T14:32:00Z" }
Cleansed (DuckDB raw)	transaction_id, source, amount_inr, direction, counterparty, transaction_date
Transformed (dbt)	2026-04-15 | upi | inflow: 12000 | outflow: 8500 | net: 3500
ML Output	2026-04-15 | upi | anomaly_score: 0.12 | is_anomaly: true

---

Project Structure

cashsense/
├── .github/
│   └── workflows/
│       ├── dbt_test.yml          # Runs dbt tests on every push to main
│       └── ml_retrain.yml        # Retrains ML models every Sunday at midnight
├── producers/
│   ├── whatsapp_producer.py      # Streams WhatsApp-style transactions to Kafka
│   └── upi_producer.py           # Streams UPI CSV data to Kafka
├── consumers/
│   └── transaction_consumer.py   # Reads Kafka, cleanses, writes to DuckDB
├── cleansing/
│   └── cleaner.py                # Pandas-based cleansing & normalisation logic
├── storage/
│   └── duckdb_loader.py          # DuckDB schema init & dataframe loader
├── ml/
│   ├── prophet_model.py          # Facebook Prophet cashflow forecasting
│   ├── isolation_forest_model.py # Isolation Forest transaction anomaly scoring
│   └── run_models.py             # Orchestrates both ML models end-to-end
├── data/
│   ├── synthetic_generator.py    # Generates synthetic MSME transaction data
│   └── sample_upi_transactions.csv
├── dbt/
│   ├── models/
│   │   ├── daily_cashflow.sql    # Aggregates inflow/outflow per channel per day
│   │   ├── weekly_summary.sql    # 7-day rolling cashflow trends
│   │   └── schema.yml            # dbt schema tests (not_null, unique, accepted_values)
│   ├── dbt_project.yml
│   └── profiles.yml
├── cashsense.db                  # DuckDB database file (raw + analytics schemas)
├── docker-compose.yml            # Kafka + Zookeeper local setup
├── superset_config.py            # Apache Superset configuration
├── requirements.txt
└── README.md

---

Local Setup

Prerequisites

• Python 3.11+
• Docker Desktop
• pip

Step 1 — Clone & Install

git clone https://github.com/LevelVoid/CashSense.git
cd CashSense

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

pip install -r requirements.txt

Step 2 — Start Kafka (via Docker)

docker-compose up -d

This starts Zookeeper and a single-broker Kafka instance on localhost:9092.

Step 3 — Generate Synthetic Data & Stream to Kafka

# Generate a sample UPI CSV with injected anomalies
python data/synthetic_generator.py

# Stream UPI and WhatsApp transactions to Kafka
python producers/upi_producer.py
python producers/whatsapp_producer.py

Step 4 — Run the Consumer (Cleanse & Load to DuckDB)

python consumers/transaction_consumer.py

This reads messages from Kafka, applies cleansing (date normalisation, deduplication, amount extraction), and writes to cashsense.db in the raw schema.

Step 5 — Run dbt Transformations

cd dbt
dbt run     # Build daily_cashflow and weekly_summary models
dbt test    # Validate schema tests (not_null, unique, accepted_values)

Step 6 — Run ML Anomaly Detection

python ml/run_models.py

This runs Prophet (cashflow forecasting) and Isolation Forest (transaction scoring), writing results to the analytics.anomaly_results table in DuckDB.

Step 7 — Launch Apache Superset Dashboard

superset db upgrade
superset fab create-admin   # Set username/password when prompted
superset init
superset run -p 8088 --with-threads --reload --debugger

Open http://localhost:8088 and add a DuckDB database connection:

SQLAlchemy URI: duckdb:////<absolute-path-to-repo>/cashsense.db

Step 8 — Verify End-to-End in DuckDB CLI

python -c "
import duckdb
conn = duckdb.connect('cashsense.db')
print(conn.execute('SELECT * FROM analytics.anomaly_results WHERE is_anomaly = TRUE LIMIT 10').df())
"

---

CI/CD

Trigger	Workflow	Action
Every push to main	dbt_test.yml	Runs dbt run + dbt test; fails the check if any schema test breaks
Every Sunday midnight (UTC)	ml_retrain.yml	Retrains Prophet & Isolation Forest on latest DuckDB data
Manual workflow_dispatch	ml_retrain.yml	Same as above, triggerable on demand

---

Progress Status

✅ Done

Component	Files	Notes
Kafka Producers	producers/upi_producer.py, producers/whatsapp_producer.py	Both streaming to Kafka topics
Synthetic Data Generator	data/synthetic_generator.py, data/sample_upi_transactions.csv	1000-record generation with anomaly injection (5%)
Kafka Consumer + Cleansing	consumers/transaction_consumer.py, cleansing/cleaner.py	Full cleanse, validate, dedup, and DuckDB write
DuckDB Storage Layer	storage/duckdb_loader.py, cashsense.db	Schema init, raw + analytics schemas live
dbt Transformation Models	dbt/models/daily_cashflow.sql, weekly_summary.sql, schema.yml	Tests passing for not_null, accepted_values
ML — Isolation Forest	ml/isolation_forest_model.py	Feature engineering, scoring, DuckDB write-back
ML — Prophet Forecasting	ml/prophet_model.py	Per-channel forecast, anomaly flag, DuckDB write-back
ML Orchestration	ml/run_models.py	Runs both models end-to-end
GitHub Actions — dbt Tests	.github/workflows/dbt_test.yml	Triggers on push + PR to main
GitHub Actions — ML Retrain	.github/workflows/ml_retrain.yml	Weekly schedule + manual dispatch
Docker Compose (Kafka)	docker-compose.yml	Zookeeper + Kafka broker

⏳ Remaining

Component	Status	Notes
currency_producer.py	❌ Not built	SRS requires a 3rd Kafka topic (currency-rates) for OpenExchangeRates; producer missing
anomaly_flags.sql (dbt)	❌ Not built	SRS specifies a dedicated dbt model for pre-ML anomaly flagging; only daily_cashflow and weekly_summary exist
Superset Dashboard Panels	⚠️ Not configured	superset_config.py exists but the 5 dashboard panels (cashflow trend, channel bar, anomaly timeline, weekly trend, big-number alert count) are not built
.env.example	❌ Not created	SRS lists this as required for documenting API keys (OpenExchangeRates) and Kafka broker config
pytest Test Suite	❌ Not created	SRS specifies tests/ directory with unit tests for cleaner.py and integration tests for producer→consumer flow
Superset Alerts	❌ Not configured	No-data alert and anomaly spike (>5/day) alerts not set up
Structured JSON Logging	⚠️ Partial	SRS requires log_event() JSON logging consistently across all pipeline stages

---

Tech Stack Decisions

Technology	Role	Why
Apache Kafka	Message streaming	Decouples producers & consumers; handles backpressure; horizontally scalable via partitions
DuckDB	Data warehouse	Embedded OLAP, zero licensing cost, first-class dbt adapter, in-process Python querying
dbt	Transformation	Declarative SQL models, built-in schema tests, version-controlled data lineage
Facebook Prophet	Cashflow forecasting	Handles weekly seasonality & missing data; interpretable for business stakeholders
Isolation Forest	Transaction anomaly detection	No labelled data required; O(n log n) scale; robust to irrelevant features
Apache Superset	BI Dashboard	SQL-driven, DuckDB-native, financial chart types — correct tool vs. Grafana (infra monitoring)
GitHub Actions	CI/CD	Free for public repos; native cron scheduling for weekly ML retraining

---

Built by Pradeep Biswas — CashSense v2.0, May 2026
Stack: Python · Kafka · Pandas · dbt · DuckDB · Scikit-learn · Prophet · Superset · GitHub Actions · Docker