Pipeline Architecture

System Overview

Click any node to view details about its role in the pipeline. Data flows left to right from the Alpha Vantage API through ingestion, transformation, and analytics layers to the interactive dashboard.

How Data Flows

# Fetch daily stock data
response = requests.get(
    "https://alphavantage.co/query",
    params={"function": "TIME_SERIES_DAILY", "symbol": symbol}
)
daily_data = response.json()["Time Series (Daily)"]

-- Upsert raw stock prices
INSERT INTO stock_prices (symbol, trade_date, open, high, low, close, volume)
VALUES (%s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (symbol, trade_date) DO UPDATE
SET open = EXCLUDED.open, close = EXCLUDED.close;

# Calculate technical indicators with pandas
df["sma_20"] = df["close"].rolling(window=20).mean()
df["ema_12"] = df["close"].ewm(span=12).mean()
df["macd"] = df["ema_12"] - df["ema_26"]
df["rsi_14"] = compute_rsi(df["close"], period=14)

-- ClickHouse MergeTree table for fast analytics
CREATE TABLE stock_analytics (
    symbol LowCardinality(String),
    trade_date Date,
    sma_20 Float64,
    rsi_14 Float64
) ENGINE = MergeTree()
ORDER BY (symbol, trade_date);

@app.get("/api/pipeline/stocks/{symbol}/analytics")
async def get_analytics(symbol: str, days: int = 90):
    query = """
        SELECT * FROM stock_analytics
        WHERE symbol = {symbol:String}
        ORDER BY trade_date DESC LIMIT {days:UInt32}
    """
    return await clickhouse.execute(query)

// Fetch and render stock analytics
const [ohlcv, analytics, volume] = await Promise.all([
  getStockOHLCV(symbol, days),
  getStockAnalytics(symbol, days),
  getStockVolume(symbol, days),
]);
return <PriceChart data={ohlcv} analytics={analytics} />;

Why This Architecture