A production-grade financial data engineering pipeline that collects live stock market data via APIs, streams it through Apache Kafka, transforms it with Python, stores structured results, and delivers insights through Power BI and Python dashboards — built to demonstrate enterprise-level ETL, real-time processing, and BI integration.
Financial institutions — banks, investment firms, SACCOs, and fintech companies — require real-time visibility into market movements to support trading decisions, risk management, and portfolio analytics. Batch-processed reports are insufficient for fast-moving financial environments.
This project solves that by building a complete, automated, real-time financial data pipeline that:
- Fetches live stock data from external market APIs
- Streams it through Apache Kafka for reliable, decoupled ingestion
- Transforms and cleans raw market data into structured analytical formats
- Stores processed data for querying and historical trend analysis
- Visualizes insights via Power BI dashboards and Python-generated graphs
- Automates the entire workflow using Apache Airflow DAGs
┌──────────────┐ Live API Feed ┌──────────────────┐
│ Market API │ ───────────────────► │ fetch_data.py │
│ (Alpha/IEX) │ │ (Data Ingestion) │
└──────────────┘ └────────┬─────────┘
│
JSON Stream (per tick)
│
┌────────▼─────────┐
│ Apache Kafka │
│ Producer/Consumer│
└────────┬─────────┘
│
Raw Market Events
│
┌────────▼─────────┐
│ process_data.py │
│ (Transform & │
│ Clean) │
└────────┬─────────┘
│
Structured CSV / DB
│
┌─────────────────┴──────────────────┐
│ │
┌────────▼─────────┐ ┌─────────▼────────┐
│ Power BI / Viz │ │ load_data.py │
│ (Dashboards) │ │ (Storage Layer) │
└──────────────────┘ └──────────────────┘
▲
┌────────┴─────────┐
│ Apache Airflow │
│ (Scheduling & │
│ Monitoring) │
└──────────────────┘
| Layer | Tool / Framework | Purpose |
|---|---|---|
| Data Ingestion | Python + Market APIs | Live stock price collection |
| Message Broker | Apache Kafka | Reliable event streaming |
| Transformation | Python (Pandas) | Data cleaning & enrichment |
| Storage | CSV / Relational DB | Structured persistent storage |
| BI & Visualization | Power BI, Python (Matplotlib/Plotly) | Dashboards & analytical graphs |
| Orchestration | Apache Airflow | ETL scheduling & pipeline monitoring |
| Language | Python 3.9+ | End-to-end scripting |
real-time-stock-analysis-pipeline/
│
├── README.md # Project documentation
├── requirements.txt # Python dependencies
│
├── src/ # Core pipeline scripts
│ ├── fetch_data.py # Fetches live stock data from APIs
│ ├── process_data.py # Cleans, transforms & enriches raw data
│ ├── load_data.py # Loads structured data to storage
│ ├── visualize_data.py # Generates Python analytical graphs
│ └── airflow_dag.py # Automates ETL pipeline scheduling
│
├── scripts/ # Kafka streaming scripts
│ ├── kafka_producer.py # Publishes stock events to Kafka topic
│ └── kafka_consumer.py # Consumes & routes Kafka messages
│
├── dashboards/ # BI Dashboard files
│ ├── power_bi_dashboard.pbix # Power BI real-time dashboard
│ └── tableau_dashboard.twb # Tableau visualization
│
├── data/ # Data files
│ ├── sample_data.csv # Sample raw stock data
│ └── processed_data.csv # Transformed & structured output
│
├── config/ # Configuration
│ ├── api_keys.json # API credentials (gitignored in production)
│ └── db_config.yaml # Database connection config
│
└── docs/ # Documentation
├── architecture_diagram.png # Pipeline architecture visual
├── dataset_description.md # Data dictionary & schema
└── presentation.pdf # Stakeholder presentation
git clone https://github.com/evans25575/real-time-stock-analysis-pipeline.git
cd real-time-stock-analysis-pipeline
pip install -r requirements.txt// config/api_keys.json
{
"alpha_vantage_key": "YOUR_API_KEY",
"iex_cloud_token": "YOUR_TOKEN"
}# Step 1: Fetch live data
python src/fetch_data.py
# Step 2: Start Kafka producer
python scripts/kafka_producer.py
# Step 3: Start Kafka consumer
python scripts/kafka_consumer.py
# Step 4: Transform & clean
python src/process_data.py
# Step 5: Load to storage
python src/load_data.py
# Step 6: Visualize
python src/visualize_data.pycp src/airflow_dag.py ~/airflow/dags/
airflow scheduler &
airflow webserver --port 8080| Step | Operation | Detail |
|---|---|---|
| Extract | API Fetch | Pulls OHLCV data (Open, High, Low, Close, Volume) per ticker |
| Validate | Schema Check | Enforces data types, rejects malformed records |
| Clean | Null Handling | Fills missing values using forward-fill for price continuity |
| Enrich | Feature Engineering | Adds moving averages (MA7, MA30), daily % change, volatility |
| Load | Storage | Saves structured records to CSV and/or relational DB |
Raw API Response:
{"symbol": "AAPL", "price": 189.45, "volume": 52341200, "timestamp": "2024-07-26T12:00:00Z"}After Transformation:
| symbol | date | open | close | volume | ma7 | daily_chg |
|--------|------------|--------|--------|-----------|--------|-----------|
| AAPL | 2024-07-26 | 188.10 | 189.45 | 52,341,200| 187.32 | +0.71% |
Power BI Dashboard Metrics:
- Live price ticker with rolling average overlay
- Volume spike detection chart
- Daily % change heatmap across tracked symbols
- Volatility index trend line
- Data Freshness: Sub-minute latency from API fetch to stored record
- Throughput: Handles multiple ticker symbols simultaneously
- Reliability: Kafka consumer offset tracking prevents data loss on restart
- Automation: Airflow DAG triggers every scheduled interval with failure alerts
| Use Case | Application |
|---|---|
| Market Risk Monitoring | Real-time price feeds for portfolio risk exposure |
| Treasury Operations | Live FX and equity data for asset-liability management |
| Customer Analytics | Transaction pattern analysis using similar ETL architecture |
| Regulatory Reporting | Structured, auditable data storage for compliance needs |
| BI & Executive Reporting | Power BI dashboards for management decision support |
Evans Kiplangat — Data Engineer & Data Analyst
MIT License — see LICENSE for details.