Financial Data Pipeline 📈

Table of Contents

Overview

Requirements

5-Minute Quick Start

Option 1: Terminal Quick Start (No Code Required)

# 1. Clone and setup
git clone https://github.com/josetraderx/Financial-Data-Pipeline.git
cd Financial-Data-Pipeline
pip install -r requirements.txt

# 2. Download Bitcoin data for the last 7 days (1-hour intervals)
python -m Data_ETL.cli download --symbol BTCUSDT --timeframe 1h --days 7

# 3. View your data
ls data/processed/  # Your CSV files are here!

Option 2: Python Quick Start (3 Lines of Code)

from Data_ETL.pipelines.crypto_pipeline import CryptoPipeline

# Download and process Bitcoin data
pipeline = CryptoPipeline()
data = pipeline.quick_download('BTCUSDT', timeframe='1h', days=7)
print(f"✅ Downloaded {len(data)} Bitcoin price records!")

✅ Downloaded 168 Bitcoin price records!
   Timestamp         Open     High      Low    Close      Volume
0  2024-08-04 00:00  65420.5  65580.0  65200.0  65350.2  1250.43
1  2024-08-04 01:00  65350.2  65480.1  65180.5  65420.8  980.52
...

Option 3: One-Liner with File Export

# Download data and save to CSV in one line
CryptoPipeline().quick_download('ETHUSDT', '4h', 14).to_csv('ethereum_2weeks.csv')

Architecture

System Flow Diagram

graph TD
    A[Data Sources] --> B[Data Providers]
    B --> C[Data Validation]
    C --> D[Data Processing]
    D --> E[Data Splitting]
    E --> F[Storage Layer]
    F --> G[Export Formats]
    
    B1[Bybit API] --> B
    B2[Binance API] --> B
    B3[Yahoo Finance] --> B
    
    C --> C1[OHLCV Validation]
    C --> C2[Outlier Detection]
    C --> C3[Missing Data Handling]
    
    F --> F1[TimescaleDB]
    F --> F2[PostgreSQL]
    F --> F3[File System]
    
    G --> G1[Parquet]
    G --> G2[CSV]
    G --> G3[JSON]
    
    subgraph "Monitoring & Quality"
        H[Data Quality Metrics]
        I[Error Logging]
        J[Performance Monitoring]
    end
    
    D --> H
    C --> I
    F --> J

Project Structure

Data_ETL/
├── providers/          # Data provider implementations
│   ├── crypto/         # Cryptocurrency data providers
│   ├── equities/       # Stock market data providers
│   ├── options/        # Options data providers
│   └── other_derivatives/  # Other derivative instruments
├── processing/         # Data processing and validation
│   ├── enhanced_metadata_manager.py  # Centralized metadata management
│   ├── data_cleaner.py              # Data validation and cleaning
│   └── data_splitter.py             # Data splitting utilities
├── storage/            # Database storage solutions
│   ├── timeseries_db.py             # TimescaleDB for time series data
│   └── metadata_db.py               # PostgreSQL for metadata
├── pipelines/          # Complete data pipelines
│   ├── crypto_pipeline.py           # Crypto data pipeline
│   ├── config_manager.py            # Configuration management
│   └── example_usage.py             # Usage examples
├── feeds/              # Data feeds
├── raw/                # Raw data and archiving
├── validation/         # Data validation modules
├── tests/              # Test scripts and setup
├── notebooks/          # Jupyter notebooks and demos
├── cli/                # Command line interface
└── README.md           # Project documentation

Key Features

1. Data Providers

2. Data Processing

3. Data Splitting

4. Storage Solutions

5. Pipeline Orchestration

Installation & Setup

1. Basic Installation

# Clone the repository
git clone https://github.com/josetraderx/Financial-Data-Pipeline.git
cd Financial-Data-Pipeline

# Using Poetry (Recommended)
poetry install
poetry shell

# Or using pip
pip install -r requirements.txt

2. Database Setup (Optional)

# Ubuntu/Debian
sudo apt-get install postgresql postgresql-contrib
sudo apt-get install timescaledb-postgresql

# macOS with Homebrew
brew install postgresql timescaledb

# Create database
sudo -u postgres createdb exodus_data
sudo -u postgres psql -d exodus_data -c "CREATE EXTENSION IF NOT EXISTS timescaledb;"

-- Connect to PostgreSQL as superuser
sudo -u postgres psql

-- Create user and database
CREATE USER exodus_user WITH PASSWORD 'your_password';
CREATE DATABASE exodus_data OWNER exodus_user;
GRANT ALL PRIVILEGES ON DATABASE exodus_data TO exodus_user;

-- Enable TimescaleDB extension
\c exodus_data
CREATE EXTENSION IF NOT EXISTS timescaledb;

Configuration

Environment Variables

# Database Configuration
export DB_HOST=localhost
export DB_PORT=5432
export DB_NAME=exodus_data
export DB_USER=exodus_user
export DB_PASSWORD=your_password

# Bybit API Configuration
export BYBIT_API_KEY=your_api_key
export BYBIT_API_SECRET=your_api_secret
export BYBIT_TESTNET=true

Configuration File Example

{
  "data_dir": "data/processed",
  "db_config": {
    "host": "localhost",
    "port": 5432,
    "database": "exodus_data",
    "user": "exodus_user",
    "password": "your_password"
  },
  "providers": {
    "bybit": {
      "api_key": "your_api_key",
      "api_secret": "your_api_secret",
      "testnet": true
    }
  },
  "validation_config": {
    "handle_missing": "interpolate",
    "outlier_method": "iqr",
    "outlier_threshold": 1.5
  }
}

Data Validation

Validation Checks

Validation Report

{
    "is_valid": true,
    "total_records": 720,
    "valid_records": 718,
    "invalid_records": 2,
    "missing_values": 0,
    "outliers_detected": 5,
    "duplicates_found": 0,
    "quality_score": 0.97,
    "errors": [],
    "warnings": ["Some outliers detected"],
    "validation_details": {
        "column_stats": {...},
        "missing_analysis": {...},
        "outlier_analysis": {...}
    }
}

Data Storage

TimescaleDB Integration

from Data_ETL.storage.timeseries_db import TimeSeriesDB

# Connect to TimescaleDB
db = TimeSeriesDB(db_config)
db.connect()
db.create_hypertable()

# Insert data
db.insert_data(dataframe)

# Query data
data = db.query_data(
    symbol='BTCUSDT',
    start_time=start_date,
    end_time=end_date
)

Metadata Management

from Data_ETL.storage.metadata_db import MetadataDB

# Connect to metadata database
metadata_db = MetadataDB(db_config)
metadata_db.connect()
metadata_db.create_tables()

# Store metadata
dataset_id = metadata_db.insert_dataset_metadata(metadata)

# Store validation report
metadata_db.insert_validation_report(dataset_id, report)

Data Splitting

Chronological Split

from Data_ETL.processing.data_splitter import DataSplitter

splitter = DataSplitter()
train_data, test_data = splitter.train_test_split(
    data, 
    test_size=0.2, 
    method='chronological'
)

Date-based Split

splits = splitter.split_by_date(
    data, 
    split_date=datetime(2024, 1, 1)
)

Sliding Windows

windows = splitter.create_sliding_windows(
    data, 
    window_size=100, 
    step_size=10
)

Monitoring and Quality Assurance

Data Quality Metrics

Monitoring Dashboard

# Get quality metrics
metrics = validator.get_quality_metrics(data)

# Store metrics in database
metadata_db.insert_quality_metrics(dataset_id, metrics)

Error Handling

Performance Optimization

Database Optimization

Memory Management

Testing

Unit Tests

# Run unit tests
pytest Data_ETL/processing/tests/

# Run specific test module
pytest Data_ETL/processing/tests/test_data_cleaner.py

Integration Tests

# Run integration tests
pytest tests/setup_test_db.py

Examples

Complete Examples

Basic Pipeline Example

from Data_ETL.pipelines.crypto_pipeline import CryptoPipeline
from Data_ETL.pipelines.config_manager import PipelineConfig

# Load configuration
config = PipelineConfig('config/pipeline_config.json')

# Create pipeline
pipeline = CryptoPipeline(config.get())

# Configure pipeline run
pipeline_config = config.create_pipeline_config(
    provider='bybit',
    symbol='BTCUSDT',
    timeframe='1h',
    days_back=30,
    save_files=True,
    store_db=False
)

# Run pipeline
results = pipeline.run_pipeline(pipeline_config)

Advanced Pipeline with Database Storage

# Configure advanced pipeline
pipeline_config = config.create_pipeline_config(
    provider='bybit',
    symbol='ETHUSDT',
    timeframe='4h',
    days_back=30,
    splits={
        'train_test_split': {
            'test_size': 0.2,
            'method': 'chronological'
        }
    },
    store_db=True,
    save_files=True
)

# Run pipeline
results = pipeline.run_pipeline(pipeline_config)

Running Examples

# Run all examples
python Data_ETL/pipelines/example_usage.py

# Set up configuration
python Data_ETL/pipelines/config_manager.py

Extension Points

Adding New Providers

Custom Validation Rules

Storage Backends

Frequently Asked Questions (FAQ)

General Questions

Installation & Setup

pip install -r requirements.txt
# or
poetry install

Configuration

export BYBIT_API_KEY=your_key_here
export BYBIT_API_SECRET=your_secret_here

symbols = ['BTCUSDT', 'ETHUSDT', 'ADAUSDT']
for symbol in symbols:
    pipeline.quick_download(symbol, '1h', 30)

Data Quality & Validation

Performance & Scaling

Troubleshooting Data Issues

results = pipeline.run_pipeline(config, debug=True)
print(results['validation_report'])

API & Rate Limiting

Development & Contributing

Troubleshooting

Common Issues

Debug Mode

# Enable debug logging
import logging
logging.basicConfig(level=logging.DEBUG)

# Run pipeline with debug info
results = pipeline.run_pipeline(config, debug=True)

Getting Help

Contributing

Development Setup

# Clone and setup development environment
git clone https://github.com/josetraderx/Financial-Data-Pipeline.git
cd Financial-Data-Pipeline

# Install development dependencies
poetry install --dev
poetry shell

# Run tests
pytest

# Run linting
flake8 Data_ETL/
black Data_ETL/

Pull Request Process

License