Stock Market Data Pipeline (Airflow)

Status: planned, not started. Plan only; no DAG code exists yet.

A second Airflow pipeline, this time for market data: daily stock quotes and company profiles from the Finnhub API into Postgres, with a DuckDB layer for fast local analysis. It reuses the patterns proven in the jobs pipeline: land raw responses first, idempotent upserts, a quality gate. It's a different data shape, though: time series instead of listings.

Tools (planned)

Tool	Will be used for
Apache Airflow (existing)	Daily schedule, retries, backfill
Finnhub REST API (free key, ~60 calls/min)	Quotes, company profiles
MinIO (existing)	Raw JSON landing zone
PostgreSQL 16 (existing postgres-target)	Typed tables, upserts keyed on symbol + date
DuckDB	Analytics over exported Parquet (returns, moving averages)

Plan

1. Watchlist and raw landing

What it will do: for each symbol, call the quote endpoint and save the response untouched under stocks/<date>/<symbol>.json. It stays under the rate limit by spacing calls.

curl -s "https://finnhub.io/api/v1/quote?symbol=AAPL&token=$FINNHUB_KEY"
# {"c":…,"h":…,"l":…,"o":…,"pc":…,"t":…}

2. Typed table with an idempotent upsert

What it will do: one row per symbol per trading day. A rerun of the same day overwrites instead of duplicating.

CREATE TABLE stocks.daily_quotes (
  symbol text, trade_date date, open numeric, high numeric, low numeric, close numeric, prev_close numeric,
  PRIMARY KEY (symbol, trade_date)
);
INSERT INTO stocks.daily_quotes VALUES (…)
ON CONFLICT (symbol, trade_date) DO UPDATE SET close = EXCLUDED.close, high = EXCLUDED.high, low = EXCLUDED.low;

3. Quality gate

What it will do: fail the run if any watchlist symbol is missing, or if a price is zero or negative.

4. DuckDB analytics

What it will do: export to Parquet and compute daily returns and 20-day moving averages in DuckDB.

SELECT symbol, trade_date, close,
       close / LAG(close) OVER (PARTITION BY symbol ORDER BY trade_date) - 1 AS daily_return,
       AVG(close) OVER (PARTITION BY symbol ORDER BY trade_date ROWS 19 PRECEDING) AS ma20
FROM read_parquet('quotes/*.parquet');

Known constraints

• Delayed quotes: non-US quotes on the free tier are delayed 15+ minutes. That's fine for end-of-day data.
• Timezones: weekends and holidays return the previous close, so the date logic must use the exchange calendar, not "today".