I Built a Stock Price Tracker with yfinance — Here’s What logical_date and Idempotent Upserts Actually Mean

A daily stock price pipeline for 5 US stocks — and the scheduling concept that every Airflow tutorial glosses over

Bilal Naseem · 9 min read · Apache Airflow · yfinance · Data Engineering · Docker

[embed]GitHub - BilalNaseem1/airflow-projects Contribute to BilalNaseem1/airflow-projects development by creating an account on GitHub.github.com

In Project 00 I built a weather pipeline and learned the basics: sensors, XComs, hooks, connections. Everything ran on @hourly with catchup=False. I never had to think about time.

Project 01 forced me to think about time. A lot. The stock price tracker runs on a @daily schedule with catchup=True — and the moment I turned it on, I realized I had no idea what Airflow actually means when it says "daily."

This article is about that confusion and how I resolved it. Along the way we build a full pipeline: 5 stocks, daily OHLCV data, idempotent upserts, weekend skipping, and automatic historical backfill.

stock_prices DAG grid view. Show green runs for weekdays, pink (skipped) for weekends.

The Pipeline

Before diving into the concept, here’s what we’re building end to end.

The complete pipeline — four tasks shown, store_prices connects after validate_prices

The Concept That Changes Everything: logical_date

Every Airflow tutorial I read mentioned logical_date in passing. Most called it execution_date (the old name) and moved on. None of them explained why it's different from when the task actually runs — and that difference is the entire mental model.

Here’s the analogy that finally made it stick for me.

💡 The Newspaper Analogy

A newspaper covers the events of Monday. But it can only be printed and delivered after Monday ends — so it arrives on Tuesday morning. The newspaper’s content date is Monday. Its delivery date is Tuesday. In Airflow: logical_date = content date. Actual run time = delivery date.

A @daily run fires AFTER the interval ends — always one day behind its logical_date

In concrete terms: with start_date = 2026-05-01 and schedule="@daily", the first run fires at May 2, 00:00 UTC. Its logical_date is May 1. It processes May 1's data.

⚠️ The Classic Mistake

Never use datetime.now() inside a DAG task. If you do, backfilling breaks completely — every historical run would fetch today's data instead of its own day's data. Always use logical_date.

In fetch_prices, this is why we write:

# ✅ Correct — uses the data interval date
def fetch_prices(trade_date: str):
    start = trade_date  # "2026-05-01" — from logical_date via XCom
    end = str((pendulum.parse(trade_date) + timedelta(days=1)).date())
    df = yf.download(tickers=STOCKS, start=start, end=end)

# ❌ Wrong — breaks every historical run
def fetch_prices_broken():
    today = datetime.now().strftime("%Y-%m-%d")
    df = yf.download(tickers=STOCKS, start=today, end=today)

catchup vs backfill — Two Ways to Load History

Both load historical data. They work differently.

catchup fills history automatically on unpause · backfill gives you manual control over any custom range

catchup=True — automatic on startup start_date May 1 today Airflow auto-triggers ALL missed runs on unpause ✓ backfill — manual, any range you choose Jan 1 — start-date — end-date today you choose this range manually airflow dags backfill stock_prices — start-date 2026–04–01 — end-date 2026–04–30

catchupbackfill Triggered byAirflow automaticallyYou manually via CLI WhenOn DAG unpauseWhenever you want Rangestart_date → nowAny range you specify Use case”Always keep up to date””Fix this specific range”

In our pipeline, catchup=True meant that the moment I unpaused the DAG, Airflow immediately started firing runs for every trading day since May 1st. 23 days of history loaded automatically — I wrote zero extra code.

Writing the DAG

Task 1: create_table — idempotency from the start

The first task creates the stock_prices table. The important detail isn't that it creates a table — it's the UNIQUE constraint:

CREATE TABLE IF NOT EXISTS stock_prices (
    id         SERIAL PRIMARY KEY,
    symbol     TEXT NOT NULL,
    open       NUMERIC,
    high       NUMERIC,
    low        NUMERIC,
    close      NUMERIC,
    volume     BIGINT,
    price_date DATE NOT NULL,
    fetched_at TIMESTAMP NOT NULL DEFAULT now(),
    CONSTRAINT uq_symbol_date UNIQUE (symbol, price_date)
);

CONSTRAINT uq_symbol_date UNIQUE (symbol, price_date) means no two rows can have the same stock on the same date. This is the foundation of idempotency — we'll use it in store_prices to make re-runs safe.

Task 2: check_market_open — skipping is not failing

US markets are closed on weekends. yfinance returns empty data for those days silently — no error, just nothing. Without a guard, we’d store empty rows and wonder why our data had gaps.

My first instinct was to raise a ValueError. That was wrong. ValueError marks the task as failed, triggers 3 retries, and fires on_failure_callback. You'd get paged at 2am because it's Saturday.

from airflow.exceptions import AirflowSkipException

@task
def check_market_open(logical_date=None):
    weekday = logical_date.day_of_week  # 0=Monday, 6=Sunday
    if weekday >= 5:
        raise AirflowSkipException(
            f"Market closed on {logical_date.date()} — skipping."
        )
    return str(logical_date.date())  # passes date forward via XCom

Task 3: fetch_prices — XCom in practice

check_market_open returns a date string. fetch_prices receives it as a parameter. Airflow handles the transfer automatically via XCom — the metadata database acts as the bridge between tasks running in complete isolation.

XComs make data flow look like regular Python function calls — Airflow handles the transfer

⚠️ XCom Size Limit

XComs are stored in the metadata database. They are designed for small values only — a string, a number, a file path. We write prices to a CSV file and pass only the path through XCom. Never push a DataFrame, a large list, or binary data through XCom.

Task 5: store_prices — idempotent upsert

This is where idempotency is enforced at the database level. The SQL does the heavy lifting:

INSERT INTO stock_prices (symbol, open, high, low, close, volume, price_date)
VALUES (%s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (symbol, price_date)
DO UPDATE SET
    open       = EXCLUDED.open,
    high       = EXCLUDED.high,
    low        = EXCLUDED.low,
    close      = EXCLUDED.close,
    volume     = EXCLUDED.volume,
    fetched_at = now();

ON CONFLICT DO UPDATE — What Happens on Re-run

Re-running the same DAG run never duplicates data — ON CONFLICT DO UPDATE handles it at the SQL level

Wiring: Implicit vs Explicit Dependencies

The pipeline has two kinds of task dependencies. Understanding the difference matters for reading any Airflow DAG.

# implicit — inferred from XCom data flow
date      = check_market_open()  # must run first
file      = fetch_prices(date)   # must run after check_market_open
validated = validate_prices(file)
store_prices(validated)

# explicit — no data flows, order still matters
create_table >> date  # table must exist before we try to write

The implicit dependencies stay in sync with the code automatically — if you change what fetch_prices receives, the dependency updates too. Prefer implicit wherever data flows between tasks.

The Mistakes I Made

1. ValueError for weekends

My first implementation used raise ValueError("Market closed"). Every weekend run showed red, triggered 3 retries each, and fired on_failure_callback. The fix was AirflowSkipException — one import, one word change, all weekend runs went from red to pink.

2. Wiring inside a task function

I accidentally indented the wiring code inside store_prices. Python treated it as part of the function body — it never executed during DAG parsing. Only create_table ran, nothing else. The fix was pure indentation — move the wiring to the DAG body level.

# ❌ Wrong — wiring inside store_prices function body
@task
def store_prices(file_path: str):
    # ... store logic ...
        date = check_market_open()  # ← never runs during parsing
        create_table >> date

# ✅ Correct — wiring at DAG body level
@task
def store_prices(file_path: str):
    # ... store logic ...

date = check_market_open()  # ← at DAG level, runs during parsing
create_table >> date

3. Wrong import path for Airflow 3.x

airflow.providers.postgres.operators.postgres no longer exists in newer provider versions. The operator moved to airflow.providers.common.sql.operators.sql. The DAG showed a red import error in the UI until I fixed this.

4. Wrong credentials in the Airflow Connection

I set the connection login as postgres but my infra .env had POSTGRES_USER=admin. Every create_table run failed with "password authentication failed." The task log was the only place the actual error appeared — always check the task log first, not the container logs.

What the Data Looks Like

After unpausing with catchup=True, Airflow loaded every trading day from May 1st automatically. After a few minutes, 12 trading days × 5 stocks = 60 rows.

SELECT symbol, close, price_date
FROM stock_prices
ORDER BY price_date DESC, symbol
LIMIT 10;

What This Pipeline Taught Me

• 01 logical_date is not execution time. Airflow processes the data interval after it ends. Always pass logical_date to external APIs — never datetime.now().
• 02 catchup=True is backfilling for free. Unpause the DAG and history loads automatically. Zero extra code. The value of getting the mental model right.
• 03 Skipping is not failing. AirflowSkipException exists for expected conditions. ValueError is for unexpected errors. The distinction matters for on-call noise.
• 04 Idempotency is a database concern. ON CONFLICT DO UPDATE makes re-runs safe at the SQL level — no application logic needed.
• 05 The task log is your best friend. Container logs show infrastructure noise. Task logs show exactly what your code did. Always check task logs first.
• 06 Indentation is logic. In Python, where you put code determines when it runs. Wiring inside a function body is dead code during DAG parsing.

Next up: Project 02 — Reddit posts to AWS S3 with Terraform. First time touching cloud infrastructure.