DataFlow

A modular Python data pipeline framework with automated profiling, quality monitoring, a Streamlit dashboard, and LLM-powered natural language querying.

Architecture

dataflow/
├── dataflow/
│   ├── cli.py                 # Unified CLI entry point
│   ├── pipeline.py            # Core ETL engine (YAML → ingest → transform → quality → output)
│   ├── ingestors/
│   │   ├── base.py            # Abstract ingestor interface
│   │   ├── csv_ingestor.py    # CSV file ingestion
│   │   ├── api_ingestor.py    # REST API ingestion
│   │   ├── sql_ingestor.py    # SQL database ingestion (SQLAlchemy)
│   │   └── yahoo_finance.py   # Yahoo Finance market data
│   ├── transformers/
│   │   └── engine.py          # Declarative transform engine (10+ transform types)
│   ├── quality/
│   │   └── checker.py         # 8 quality check types with weighted scoring
│   ├── llm/
│   │   └── query_engine.py    # Natural language → pandas code via local LLM
│   └── scheduler/
│       └── runner.py          # APScheduler-based pipeline orchestration
├── configs/
│   └── market_data.yaml       # Example pipeline config
├── dashboard.py               # Streamlit dashboard
├── requirements.txt
└── setup.py

Quick Start

# Install
cd dataflow
pip install -r requirements.txt
pip install -e .

# Run the pipeline
dataflow run configs/market_data.yaml

# Launch dashboard
streamlit run dashboard.py

# Profile a dataset
dataflow profile data/processed/market_data_latest.parquet

# Query with natural language (requires LM Studio at localhost:1234)
dataflow query data/processed/market_data_latest.parquet
dataflow query data/processed/market_data_latest.parquet -q "What ticker has the highest Sharpe ratio?"

# Check quality
dataflow quality data/processed/market_data_latest.parquet configs/market_data.yaml

# View run history
dataflow history

# Schedule recurring runs
dataflow schedule configs/market_data.yaml --cron "0 9 * * 1-5"   # weekdays 9am
dataflow schedule configs/market_data.yaml --interval 60           # every hour

Pipeline Configuration (YAML)

Pipelines are fully declarative. Each YAML config defines four stages:

1. Ingestion

ingestion:
  type: yahoo_finance    # csv | api | sql | yahoo_finance
  tickers: [SPY, QQQ, GLD, BTC-USD]
  period: "2y"

2. Transforms

transforms:
  - type: sort
    by: [Ticker, Date]
  - type: rolling
    column: Close
    window: 20
    func: mean
    output: SMA_20
    group_by: Ticker
  - type: fill_na
    mapping: { Volume: 0 }
  - type: add_column
    name: Spread
    expr: "High - Low"

Supported transforms: rename_columns, drop_columns, filter_rows, fill_na, cast_types, add_column, sort, deduplicate, rolling, groupby_agg

3. Quality Checks

quality_checks:
  - type: null_check
    columns: [Close, Volume]
    threshold: 0.02
    weight: 2.0
  - type: freshness_check
    date_column: Date
    max_age_days: 5
  - type: anomaly_check
    columns: [Daily_Return]
    z_threshold: 4.0

Supported checks: null_check, type_check, range_check, uniqueness_check, freshness_check, anomaly_check, completeness_check, custom_check

4. Output

output:
  format: parquet   # parquet | csv | sqlite
  filename: market_data_latest

LLM Query Engine

Connects to LM Studio (or any OpenAI-compatible API) and translates natural language into pandas code:

📊 Ask: What is the average daily return for each ticker?

💡 Code:
result = df.groupby('Ticker')['Daily_Return'].mean().sort_values(ascending=False)

📈 Result:
Ticker
BTC-USD    0.001245
SPY        0.000523
QQQ        0.000412
...

The engine sends the DataFrame schema, sample data, and statistics to the LLM, then safely executes the generated code in a sandboxed namespace.

Dashboard

The Streamlit dashboard provides:

• Overview: Key metrics, normalized price charts, volatility heatmap, quality trend
• Data Explorer: Interactive filtering by ticker, date, and metric with schema profiling
• Quality Report: Detailed check results with score breakdown visualization
• Pipeline History: Run logs, duration tracking, success/failure analysis
• LLM Query: Browser-based natural language queries against your data

Tech Stack

Component	Technology
Pipeline Engine	Python, pandas, PyYAML
Data Sources	Yahoo Finance, CSV, REST API, SQL
Quality	Custom framework (8 check types, weighted scoring)
LLM	LM Studio (OpenAI-compatible API)
Dashboard	Streamlit, Plotly
Scheduling	APScheduler
Storage	Parquet, CSV, SQLite
Version Control	Git