Data Pipeline — Ingesting Real-Time Market Data | Build Your Own Trading Terminal with AI — Part 2/6

Data Sources Deep Dive

Choosing Your Data Providers

Not all data is created equal. The difference between a profitable and a broken trading system often comes down to data quality, freshness, and reliability. Let's examine each provider in detail.

Yahoo Finance — The Free Workhorse

Yahoo Finance is the most popular free data source for retail traders. It provides daily OHLCV data for US equities, ETFs, indices, crypto, and some international stocks. The catch: it's technically a scraping target, not an official API, and Yahoo has been known to break things without notice.

    Python — Yahoo Finance Provider (yfinance)
import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional

class YahooProvider:
    """Yahoo Finance data provider with error handling."""

    def __init__(self):
        self.name = "yahoo"
        self.rate_limit = 30  # calls/minute (conservative)

    def fetch_daily_bars(
        self,
        symbol: str,
        start: Optional[str] = None,
        end: Optional[str] = None,
        period: str = "1y"
    ) -> pd.DataFrame:
        """Fetch daily OHLCV bars for a symbol."""
        ticker = yf.Ticker(symbol)
        try:
            if start and end:
                df = ticker.history(start=start, end=end)
            else:
                df = ticker.history(period=period)

            if df.empty:
                raise ValueError(f"No data for {symbol}")

            # Normalize column names to our schema
            df = df.rename(columns={
                "Open": "open",
                "High": "high",
                "Low": "low",
                "Close": "close",
                "Volume": "volume",
            })
            df["symbol"] = symbol
            df["provider"] = self.name
            df.index.name = "timestamp"

            # Handle splits & dividends
            if "Stock Splits" in df.columns:
                df["split_ratio"] = df["Stock Splits"]
            if "Dividends" in df.columns:
                df["dividend"] = df["Dividends"]

            return df[["symbol","open","high","low","close","volume","provider"]]

        except Exception as e:
            print(f"[Yahoo] Error fetching {symbol}: {e}")
            return pd.DataFrame()
  

Polygon.io — The Professional Choice

If you need real-time data, options chains, or tick-level granularity, Polygon.io is the go-to. Their WebSocket streams are production-grade and their REST API supports aggregates at any timeframe. The $29/month Starter plan gets you 15-minute delayed data with unlimited API calls; $199/month gets real-time.

    Python — Polygon.io WebSocket Stream
import asyncio
import json
import websockets

class PolygonStream:
    """Real-time trade stream via Polygon WebSocket."""

    WS_URL = "wss://socket.polygon.io/stocks"

    def __init__(self, api_key: str, symbols: list[str]):
        self.api_key = api_key
        self.symbols = symbols
        self.callbacks = []

    def on_trade(self, callback):
        """Register a callback for trade events."""
        self.callbacks.append(callback)

    async def connect(self):
        async with websockets.connect(self.WS_URL) as ws:
            # Authenticate
            await ws.send(json.dumps({
                "action": "auth",
                "params": self.api_key
            }))

            # Subscribe to trade events
            subs = ",".join(f"T.{s}" for s in self.symbols)
            await ws.send(json.dumps({
                "action": "subscribe",
                "params": subs
            }))

            # Process incoming trades
            async for msg in ws:
                data = json.loads(msg)
                for event in data:
                    if event.get("ev") == "T":
                        trade = {
                            "symbol": event["sym"],
                            "price": event["p"],
                            "size": event["s"],
                            "timestamp": event["t"],
                        }
                        for cb in self.callbacks:
                            await cb(trade)

# Usage
stream = PolygonStream("YOUR_API_KEY", ["AAPL", "MSFT", "NVDA"])
stream.on_trade(lambda t: print(f"{t['symbol']}: ${t['price']}"))
asyncio.run(stream.connect())
  

Real-Time vs Delayed vs EOD

What Data Freshness Do You Actually Need?

This is the most important cost-benefit decision in your terminal. Real-time data costs 10-50x more than EOD data. The question is: does your strategy actually need it?

Data Type	Latency	Cost	Best For	Providers
Real-Time (L1)	<100ms	$100-500/mo	Day trading, scalping	Polygon, Alpaca, IB
15-min Delayed	15 min	$0-30/mo	Swing trading, monitoring	Polygon Starter, IEX
EOD (End of Day)	~4h after close	$0	Swing trading, screening	Yahoo, Tiingo, Alpha Vantage
Weekly/Monthly	Days	$0	Position trading, macro	FRED, Yahoo

The Swing Trader's Sweet Spot

If you're a swing trader (holding days to weeks), EOD data is enough for 90% of your needs. You scan after market close, place limit orders for the next day, and check positions once during the session. The $200/month you save on real-time feeds goes straight into your trading account. Add 15-min delayed for intraday monitoring — that's $30/month with Polygon Starter.

The Practical Data Stack

EOD bars (free) + 15-min delayed quotes ($29/mo) + Macro data (free) = $29/month

Building the Ingestion Layer

The Ingestion Service — Core Architecture

The ingestion service is the backbone of your pipeline. It handles three transport protocols: REST polling for batch data, WebSocket streams for real-time updates, and SSE (Server-Sent Events) for one-way push. Here's the unified architecture:

REST Polling

Cron-scheduled HTTP requests. Best for EOD bars, financials, earnings. Rate-limited with token bucket.

WebSocket Streams

Persistent connections for real-time trades and quotes. Auto-reconnect with exponential backoff.

SSE (Server-Sent Events)

One-way server push. Lighter than WebSocket, used by some news/sentiment feeds.

    Python — Unified Ingestion Orchestrator
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Protocol

class TransportType(Enum):
    REST = "rest"
    WEBSOCKET = "websocket"
    SSE = "sse"

class DataProvider(Protocol):
    """Interface all providers must implement."""
    name: str
    transport: TransportType

    async def fetch(self, symbols: list[str]) -> list[dict]: ...
    async def health_check(self) -> bool: ...

@dataclass
class IngestionJob:
    provider: DataProvider
    symbols: list[str]
    interval_seconds: int
    enabled: bool = True

class Orchestrator:
    """Manages all ingestion jobs with health monitoring."""

    def __init__(self, db, cache):
        self.db = db
        self.cache = cache
        self.jobs: list[IngestionJob] = []
        self.stats = {"total_fetches": 0, "errors": 0}

    def register(self, job: IngestionJob):
        self.jobs.append(job)

    async def run_job(self, job: IngestionJob):
        while job.enabled:
            try:
                rows = await job.provider.fetch(job.symbols)
                await self.db.upsert_bars(rows)
                await self.cache.invalidate(job.symbols)
                self.stats["total_fetches"] += 1
            except Exception as e:
                self.stats["errors"] += 1
                print(f"[{job.provider.name}] Error: {e}")
            await asyncio.sleep(job.interval_seconds)

    async def start(self):
        tasks = [self.run_job(j) for j in self.jobs if j.enabled]
        await asyncio.gather(*tasks)
  

Failover Strategy

Always configure a primary and fallback provider for each data type. If Yahoo Finance goes down (happens ~2x/year), your ingestion should automatically fail over to Alpha Vantage or Tiingo. Log every failover event so you can monitor provider reliability over time.

Data Normalization

The OHLCV Schema — Your Universal Language

Every data provider returns data in a slightly different format. Yahoo uses "Close" (capitalized), Polygon uses "c" (abbreviated), Alpha Vantage uses "4. close". Your normalizer translates all of them into a single canonical schema:

    SQL — TimescaleDB Schema (OHLCV)
CREATE TABLE bars_daily (
    timestamp  TIMESTAMPTZ NOT NULL,
    symbol     TEXT NOT NULL,
    open       NUMERIC(12,4),
    high       NUMERIC(12,4),
    low        NUMERIC(12,4),
    close      NUMERIC(12,4) NOT NULL,
    volume     BIGINT,
    adj_close  NUMERIC(12,4),
    provider   TEXT,
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Convert to TimescaleDB hypertable for time-series optimization
SELECT create_hypertable('bars_daily', 'timestamp');

-- Unique constraint prevents duplicate bars
CREATE UNIQUE INDEX idx_bars_daily_sym_ts
    ON bars_daily (symbol, timestamp);

-- Compression policy: compress chunks older than 30 days
ALTER TABLE bars_daily SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'symbol',
    timescaledb.compress_orderby = 'timestamp DESC'
);
SELECT add_compression_policy('bars_daily', INTERVAL '30 days');

-- Continuous aggregate: auto-generate weekly bars from daily
CREATE MATERIALIZED VIEW bars_weekly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 week', timestamp) AS week,
    symbol,
    first(open, timestamp) AS open,
    max(high) AS high,
    min(low) AS low,
    last(close, timestamp) AS close,
    sum(volume) AS volume
FROM bars_daily
GROUP BY week, symbol;
  

Handling Splits and Dividends

Stock splits and dividends are the silent killers of backtesting accuracy. A 4:1 split on NVDA means every historical price before the split date needs to be divided by 4, and every volume multiplied by 4. Here's how to handle this correctly:

    Python — Split & Dividend Adjustment
def adjust_for_splits(
    df: pd.DataFrame,
    splits: list[dict]
) -> pd.DataFrame:
    """Apply split adjustments to historical bars.

    Args:
        df: DataFrame with OHLCV data, indexed by timestamp
        splits: List of {'date': '2024-06-10', 'ratio': 10}
    """
    adjusted = df.copy()

    for split in sorted(splits, key=lambda s: s["date"]):
        split_date = pd.Timestamp(split["date"])
        ratio = split["ratio"]

        mask = adjusted.index < split_date
        for col in ["open", "high", "low", "close", "adj_close"]:
            adjusted.loc[mask, col] /= ratio
        adjusted.loc[mask, "volume"] *= ratio

    return adjusted
  

Adjusted vs. Unadjusted Prices

Always store both raw (unadjusted) and adjusted prices. Raw prices are what actually traded on that date — useful for audit trails and order reconstruction. Adjusted prices account for splits and dividends — essential for accurate backtesting and technical analysis. Yahoo Finance's adj_close column handles this, but verify against your own split data.

Storage Architecture

TimescaleDB + Redis + S3 — The Three-Tier Stack

Data access patterns in a trading terminal are highly varied: you need millisecond reads for live quotes, sub-second queries for chart rendering, and bulk scans for backtesting. No single database handles all three well. That's why we use a three-tier architecture:

Hot — Redis

Latest quotes, 5-min bars, watchlist data. Sub-ms reads. 256MB memory cap with LRU eviction.

Warm — TimescaleDB

Last 2 years of daily bars, all active symbols. Hypertable compression for 10x space savings on older data.

Cold — S3/MinIO

Historical archives (10+ years). Parquet files partitioned by year/symbol. Loaded on-demand for backtesting.

    Python — Redis Cache Layer
import redis.asyncio as redis
import json

class QuoteCache:
    """Redis cache for real-time quotes and recent bars."""

    def __init__(self, url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(url)

    async def set_quote(self, symbol: str, data: dict):
        """Cache a quote with 60s TTL."""
        key = f"quote:{symbol}"
        await self.redis.setex(key, 60, json.dumps(data))

    async def get_quote(self, symbol: str) -> dict | None:
        """Get cached quote, None if expired."""
        key = f"quote:{symbol}"
        val = await self.redis.get(key)
        return json.loads(val) if val else None

    async def set_bars(self, symbol: str, tf: str, bars: list):
        """Cache bar data with 5min TTL for intraday, 1h for daily."""
        key = f"bars:{symbol}:{tf}"
        ttl = 300 if tf == "5min" else 3600
        await self.redis.setex(key, ttl, json.dumps(bars))

    async def invalidate(self, symbols: list[str]):
        """Invalidate cache for given symbols after DB update."""
        keys = []
        for s in symbols:
            keys.extend([f"quote:{s}", f"bars:{s}:*"])
        if keys:
            await self.redis.delete(*keys)
  

Cron Scheduling

Scheduling Your Ingestion Jobs

Different data types need different update frequencies. Here's a production-ready cron schedule for a swing trading terminal:

Job	Schedule	Provider	Symbols	Duration
Daily bars (US)	Mon-Fri 17:30 ET	Yahoo Finance	500 (S&P 500)	~3 min
Daily bars (EU)	Mon-Fri 18:00 CET	Yahoo Finance	100 (STOXX 600 subset)	~1 min
Crypto quotes	Every 5 min	CoinGecko	20 (top coins)	~10 sec
Macro data	Daily 08:00 ET	FRED	30 series	~30 sec
Financials	Weekly Sunday	Alpha Vantage	500	~20 min
Earnings calendar	Weekly Sunday	Yahoo Finance	Next 2 weeks	~1 min
News headlines	Every 15 min	Finnhub	Watchlist	~5 sec

    Python — APScheduler Cron Configuration
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger

scheduler = AsyncIOScheduler(timezone="US/Eastern")

# Daily bars after US market close
scheduler.add_job(
    ingest_daily_bars,
    CronTrigger(day_of_week="mon-fri", hour=17, minute=30),
    kwargs={"universe": "SP500", "provider": "yahoo"},
    id="daily_bars_us",
    name="Ingest US daily bars",
)

# Crypto quotes every 5 minutes
scheduler.add_job(
    ingest_crypto_quotes,
    "interval",
    minutes=5,
    kwargs={"coins": ["BTC","ETH","SOL","XRP"]},
    id="crypto_quotes",
)

# Macro data at 8am ET
scheduler.add_job(
    ingest_macro_data,
    CronTrigger(hour=8, minute=0),
    kwargs={"series": ["DGS10","VIXCLS","DTWEXBGS"]},
    id="macro_daily",
)

# Health check every 10 minutes
scheduler.add_job(
    check_provider_health,
    "interval",
    minutes=10,
    id="health_check",
)

scheduler.start()
  

Timing Is Everything

Schedule US daily bars at 5:30 PM ET, not 4:00 PM. Why? After-hours adjustments, split data, and dividend data take 30-90 minutes to propagate. Fetching too early gives you incomplete or incorrect data. For the same reason, schedule EU bars at 6:00 PM CET, not at market close.

Key Takeaways

Key Takeaways — Part 2

Use Yahoo Finance (free) for daily OHLCV as your primary source, with Alpha Vantage and Tiingo as failovers.
For swing trading, EOD data + 15-min delayed quotes ($29/mo) covers 95% of your needs.
Build a unified ingestion orchestrator with provider interfaces, failover logic, and health monitoring.
Normalize all data to a canonical OHLCV schema in TimescaleDB with hypertable compression.
Three-tier storage: Redis (hot, quotes) + TimescaleDB (warm, bars) + S3 (cold, archives).
Always store both raw and split-adjusted prices. Schedule ingestion 90 minutes after market close.

Part 3 of 6

Signals & Screeners — From Data to Alpha

Build Your Own Trading Terminal with AI

Complete 6-part series — From architecture to production deployment

Back to Home