OVERVIEW
Unified batch + streaming market data warehouse tracking 10 major US equities with 25 years of history. Combines 4 data sources (yfinance prices, SEC EDGAR filings, FRED macro indicators, and live Kafka streams) through ~18 Airflow DAGs into an 8-table star schema on TimescaleDB. Fully containerized with 7 Docker services.
ARRIVED AS
Build a unified market data warehouse combining real-time streaming with batch historical data from multiple sources.
WHAT I BUILT
- 01Kafka producers poll 4 data sources (yfinance, SEC EDGAR, FRED, live streams) every 15 seconds.
- 02~18 Airflow DAGs orchestrate batch ETL, backfills, and aggregations across 10 major tickers.
- 038-table star schema in TimescaleDB with dimensional modeling for prices, fundamentals, earnings, SEC filings, and macro data.
WHAT CHANGED
- 25 years of historical depth for 10 major US equities (AAPL, NVDA, MSFT, GOOG, AMZN, META, TSLA, JPM, NFLX, DIS).
- Fully containerized with 7 Docker services, single command deployment.
- Interactive Dash dashboards with SMA/EMA indicators and time-window filtering.
ARCHITECTURE
Data flow
click a stage
Kafka producers poll yfinance, SEC EDGAR, and FRED; a streaming producer publishes live prices to the broker every 15s.
COMPONENT
No component mapped to this stage.
Decisions, with the cost of each.
A decision without its trade-off is marketing. Each row says what was chosen, why, and what it gave up.
TimescaleDB over plain Postgres
25 years of daily ticks across 10 tickers plus intraday macro series is time-series-heavy; TimescaleDB hypertables keep range queries fast while staying SQL-native.
One more extension to operate versus reaching for a dedicated TSDB like InfluxDB and losing relational joins to the dimension tables.
Batch-and-flush Kafka consumer, not per-message inserts
Upserting in batches of 50 (or every 10s, whichever comes first) cuts write amplification and keeps the warehouse current without hammering it.
A small bounded staleness window in exchange for far fewer transactions.
Incremental DAGs keyed on last-loaded date
Each ETL run reads the latest date already in the warehouse and only fetches the gap, so reruns are idempotent and backfills are cheap.
More bookkeeping per DAG than a naive full reload.
The part that mattered.
The numbers behind the work, and the code that produced them.
- history per ticker
- 25 yrs
- major US equities
- 10
- Airflow DAGs
- ~18
- Docker services
- 7
CREATE EXTENSION IF NOT EXISTS timescaledb;
CREATE TABLE IF NOT EXISTS dim_company (
company_key SERIAL PRIMARY KEY,
ticker TEXT NOT NULL UNIQUE,
company_name TEXT NOT NULL,
sector TEXT,
industry TEXT,
exchange TEXT,
is_current BOOLEAN NOT NULL DEFAULT TRUE,
effective_date DATE NOT NULL DEFAULT CURRENT_DATE
);
CREATE TABLE IF NOT EXISTS fact_stock_price_daily (
date DATE NOT NULL,
company_key INT NOT NULL REFERENCES dim_company(company_key),
open DOUBLE PRECISION NOT NULL,
high DOUBLE PRECISION NOT NULL,
low DOUBLE PRECISION NOT NULL,
close DOUBLE PRECISION NOT NULL,
volume BIGINT NOT NULL,
PRIMARY KEY (date, company_key)
);
Surrogate-keyed company dimension joined to a daily price fact, keyed on (date, company_key) so it becomes a clean TimescaleDB hypertable.
def _connect_kafka():
for attempt in range(1, MAX_RETRIES + 1):
try:
consumer = KafkaConsumer(
KAFKA_TOPIC,
bootstrap_servers=[os.environ.get("KAFKA_BOOTSTRAP", "...:9092")],
auto_offset_reset="earliest",
group_id="stock-data-consumer",
value_deserializer=lambda m: json.loads(m.decode("utf-8")),
)
return consumer
except Exception as e:
delay = min(BACKOFF_BASE * (2 ** (attempt - 1)), BACKOFF_CAP)
time.sleep(delay)
raise ConnectionError(f"Failed to connect after {MAX_RETRIES} attempts")
# batch by size OR time, whichever trips first
if batch and (len(batch) >= BATCH_SIZE or now - last_flush >= FLUSH_INTERVAL):
conn = _flush_batch(conn, batch)
batch, last_flush = [], now
Exponential backoff on connect, then a size-or-time batch flush that reconnects to the DB and retries the batch on failure rather than dropping data.
def _normalize_columns(df, ticker):
"""Map yfinance columns (with or without ticker suffix) to open/high/low/close/volume."""
if isinstance(df.columns, pd.MultiIndex):
df.columns = ["_".join(col) for col in df.columns]
col_map = {}
for std in ["Open", "High", "Low", "Close", "Volume"]:
suffixed = f"{std}_{ticker}"
if suffixed in df.columns:
col_map[suffixed] = std.lower()
elif std in df.columns:
col_map[std] = std.lower()
return df.rename(columns=col_map)
# only fetch the gap since the last loaded date -> reruns are idempotent
last = _get_last_loaded_date(ticker)
start = (last + timedelta(days=1)) if last else DEFAULT_BACKFILL_START
Column normalization handles yfinance's shifting multi-index, and the last-loaded-date lookup keeps each DAG run incremental and safe to rerun.
✓ LEARNED
Time-or-size batching beats both extremes
Per-message inserts melt the warehouse; pure time batching lags. Flushing on whichever of size/time trips first kept writes cheap and data fresh.
Idempotency is a schema property, not a script
Keying facts on (date, company_key) and reading the last-loaded date made reruns and backfills boring, which is exactly what you want from a pipeline.
◔ NOT DONE YET
- Add great-expectations data-quality gates between extract and load.
- Promote the Dash app to a hosted service with auth instead of a static snapshot.