Building a Crypto Data Pipeline in Python: From Raw to Parquet

Building your own crypto data pipeline helps you evaluate dataset quality, debug problems, and extend to new assets. This article covers the core stages from raw compressed files to a consolidated Parquet master.

Stage 1: Downloading with Resume Capability

import requests
from pathlib import Path

def download_file(url: str, dest: Path) -> bool:
    if dest.exists():
        return True  # already downloaded
    dest.parent.mkdir(parents=True, exist_ok=True)
    r = requests.get(url, stream=True, timeout=30)
    if r.status_code == 404:
        return False
    r.raise_for_status()
    with open(dest, "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)
    return True

Stage 2: Auditing Before Consolidation

import pandas as pd, zipfile

def audit_file(path: Path) -> dict:
    with zipfile.ZipFile(path) as z:
        df = pd.read_csv(z.open(z.namelist()[0]), header=None)
    ts = df.iloc[0, 4]
    return {
        "rows": len(df),
        "timestamp_unit": "ms" if ts < 10**14 else "us",
        "columns": df.shape[1],
    }

Stage 3: Streaming Consolidation with PyArrow

import pyarrow as pa, pyarrow.parquet as pq, zipfile
import pandas as pd
from pathlib import Path

def consolidate(raw_dir: Path, output: Path):
    writer = None
    for f in sorted(raw_dir.glob("*.zip")):
        with zipfile.ZipFile(f) as z:
            df = pd.read_csv(z.open(z.namelist()[0]), header=None)
        if df.iloc[0, 4] > 10**14:
            df.iloc[:, 4] = df.iloc[:, 4] // 1000  # fix us->ms
        table = pa.Table.from_pandas(df, preserve_index=False)
        if writer is None:
            writer = pq.ParquetWriter(output, table.schema, compression="zstd")
        writer.write_table(table)
    if writer: writer.close()

Stage 4: Verifying Without Loading Everything

import pyarrow.parquet as pq

pf = pq.ParquetFile("master.parquet")
print(f"Rows: {pf.metadata.num_rows:,}")
print(f"Row groups: {pf.metadata.num_row_groups}")

Summary

Four stages: download with resume, audit before consolidating, stream-consolidate with PyArrow, verify with metadata. The most important and most skipped stage is the audit — it is where timestamp bugs and schema changes are caught before they corrupt your master.