25 - Functional Pipeline - Python

99 min read

Functional Data Pipeline — Python

Quote

“The object-oriented version of spaghetti code is, of course, ‘lasagna code’. Too many layers.”

— Roberto Waltman

“State is never simple. State complects value and time.”

— Rich Hickey, Simple Made Easy, Strange Loop talk (2011)

Data pipeline architecture combining five principles: functional core/imperative shell, contract-first validation, quality gates, data provenance with SHA-256 tamper detection, and semantic context propagation.

Data flow: yfinance → JSON landing → Pydantic validation → Bronze → Polars transforms → Silver → Polars aggregation → Gold → Parquet → FastAPI

Two orthogonal dimensions of data trustworthiness:

  • Structural integrity (vertical) — pure transforms, typed contracts, quality gates, immutable models
  • Semantic integrity (horizontal) — column context, business context, temporal markers, lineage tracking

This note implements an end-to-end functional data pipeline in Python using Polars, Pydantic, FastAPI, and SQL Server with medallion architecture, lineage tracking, and SHA-256 tamper detection.

Pipeline Dependencies

All imports organized by category: stdlib, data, validation, database, serving, visualization.

Reference snippet for Functional Data Pipeline — Python.

import hashlib
import importlib.util
import logging
import json
import os
import subprocess
import threading
import time
import uuid
from datetime import datetime, timezone, date as Date, timedelta
from pathlib import Path
from urllib.parse import quote_plus
 
# Suppress SQLAlchemy DBAPI2 warnings
import warnings
from sqlalchemy import exc as sa_exc
warnings.filterwarnings('ignore', category=sa_exc.SAWarning)
 
# Data & transforms
import polars as pl
import yfinance as yf
import pandas_market_calendars as mcal
 
# Validation
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from pydantic import BaseModel, Field, field_validator, model_validator, ConfigDict
 
# Database
import pyodbc
from sqlalchemy import create_engine, text
 
# Serving
from fastapi import FastAPI, HTTPException
import uvicorn
import httpx
 
# Visualization
import plotly.graph_objects as go
from IPython.display import display, HTML
 
# ── Polars HTML formatter — strip quotes, transparent background ──
_html_fmt = get_ipython().display_formatter.formatters["text/html"]  # type: ignore
_html_fmt.for_type(pl.DataFrame, lambda df: df.to_pandas().style.hide(axis="index").set_properties(**{"text-align": "left"}).to_html())
 
# ── Structured logging ──
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-5s | %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("pipeline")
No direct stdout was emitted by this cell.

1. Configuration & Constants

Central configuration: paths, SQL connection, stock universe, date range. Every downstream cell references these constants — change them here, not in individual cells.

Project configuration

Single cell that defines all pipeline-wide constants used by every downstream stage.

Python — define pipeline paths, SQL connection, and stock universe

Central Configuration Cell

All downstream cells reference these constants. Change paths, SQL connection, stock universe, and date range here only.

Reference snippet for Python — define pipeline paths, SQL connection, and stock universe.

DATA_DIR    = Path(r"C:\Users\aperi\DEV\LANG\data")
EXPORT_DIR  = DATA_DIR / "pipeline"
LINEAGE_DIR = EXPORT_DIR / "lineage"
LANDING_DIR = DATA_DIR / "pipeline" / "landing"
EXPORT_DIR.mkdir(parents=True, exist_ok=True)
LANDING_DIR.mkdir(parents=True, exist_ok=True)
LINEAGE_DIR.mkdir(parents=True, exist_ok=True)
 
# ── SQL Server (local Docker instance) ──
os.environ.setdefault("PIPELINE_SQL_USER", "sa")
os.environ.setdefault("PIPELINE_SQL_PASSWORD", "change-me")
SQL_CONN_STR = (
    "Driver={ODBC Driver 18 for SQL Server};"
    "Server=localhost,1434;Database=stoxx;"
    f"UID={os.environ["PIPELINE_SQL_USER"]};PWD={os.environ["PIPELINE_SQL_PASSWORD"]};"
    "Encrypt=yes;TrustServerCertificate=yes;"
)
sql_engine = create_engine(f"mssql+pyodbc:///?odbc_connect={quote_plus(SQL_CONN_STR)}")
 
# ── Stock universe — 5 EURO STOXX 50 components for demo ──
SYMBOLS = ["SAP.DE", "SIE.DE", "ALV.DE", "DTE.DE", "BAS.DE"]
LOOKBACK_DAYS = 365 * 2  # 2 years of history
START_DATE = (Date.today() - timedelta(days=LOOKBACK_DAYS)).isoformat()
END_DATE   = Date.today().isoformat()
 
# ── Exchange mapping: yfinance exchange code → mcal calendar name ──
EXCHANGE_MAP = {
    "GER": "XETR",    # XETRA (German stocks)
    "FRA": "XFRA",    # Frankfurt
    "PAR": "XPAR",    # Euronext Paris
    "AMS": "XAMS",    # Euronext Amsterdam
    "BRU": "XBRU",    # Euronext Brussels
    "MIL": "XMIL",    # Borsa Italiana
    "MCE": "XMAD",    # Madrid
    "NMS": "XNYS",    # NASDAQ → use NYSE calendar
    "NYQ": "XNYS",    # NYSE
    "HKG": "XHKG",    # Hong Kong
    "TKS": "XTKS",    # Tokyo
}
 
EXPORT_DIR
SYMBOLS
START_DATE, END_DATE  # date range
Notebook stdout is preserved below.
  C:\Users\aperi\DEV\LANG\data\pipeline
  ['SAP.DE', 'SIE.DE', 'ALV.DE', 'DTE.DE', 'BAS.DE']
  2024-03-29 → 2026-03-29

2. Pydantic DTOs — Schema Validation at Every Boundary

Every stage boundary has a typed contract — a Pydantic BaseModel that defines exactly what data can cross that boundary. Data that doesn’t conform is rejected BEFORE it crosses, with the rejection recorded in the quarantine table. This is Schema-on-Write (Design by Contract, Bertrand Meyer 1986): the contract is code — it runs at runtime, it’s version-controlled, it’s unit-testable. The opposite of data lake Schema-on-Read, where bad data enters freely and is discovered months later.

The models divide into three groups along two orthogonal dimensions — structural integrity (vertical: is the data correct?) and semantic integrity (horizontal: is the data meaningful?):

Structural models (boundary enforcement):

  • RawOHLCV — first line of defense: validates raw yfinance data at Bronze ingestion
  • CleanOHLCV — validates enrichment transforms at Silver boundary
  • DailySummary / SymbolProfile — validates aggregation output at Gold boundary

Operational models (provenance tracking):

  • StageLineage — forensic record: row counts, SHA-256 hash, timing per stage
  • RunContext — full pipeline execution envelope with business and temporal context

Semantic models (self-describing data):

  • ColumnContext — what each column means, its formula, unit, null semantics
  • BusinessContext — why this run was triggered (scheduled vs backfill vs correction)
  • TemporalContext — as-of date vs knowledge date (bi-temporal)
  • StageContext — propagated metadata flowing stage-to-stage with accumulated warnings

Without Typed Contracts

A renamed API field silently loads NULLs into bronze — every row, every day. A negative volume passes through to silver unchallenged. A NaN daily return poisons the gold aggregation. By the time a dashboard user notices, the damage is three layers deep and every downstream consumer has absorbed corrupt data. Contracts catch bad data at ingestion — one layer, one fix.

Fix: Pydantic Validation at Ingestion

Define a Pydantic BaseModel for each bronze schema with Field() constraints (e.g. gt=0 for price, ge=0 for volume, @model_validator for high >= low). Parse every API response through the model before writing to SQL Server. Invalid rows raise ValidationError — catch, log to quarantine, and continue. No bad row ever reaches bronze.

Medallion layer models

Structural models for each stage boundary: Bronze, Silver, Gold, and operational lineage tracking.

Pydantic — define Bronze validation model with BaseModel and Field()

Bronze Contract: RawOHLCV

Validates raw yfinance data BEFORE persistence. Enforces positive prices, non-negative volume. Model validator: high >= low (market invariant).

Reference snippet for Pydantic — define Bronze validation model with BaseModel and Field().

 
class RawOHLCV(BaseModel):
    """Schema for raw OHLCV data from yfinance — Bronze boundary."""
    model_config = ConfigDict(strict=True)
 
    symbol:       str   = Field(..., min_length=1, description="Ticker symbol")
    date:         Date  = Field(..., description="Trading date")
    open:         float = Field(..., gt=0, description="Opening price")
    high:         float = Field(..., gt=0, description="Highest price")
    low:          float = Field(..., gt=0, description="Lowest price")
    close:        float = Field(..., gt=0, description="Closing price")
    adj_close:    float = Field(..., gt=0, description="Adjusted close")
    volume:       int   = Field(..., ge=0, description="Trading volume")
    dividends:    float = Field(default=0.0, ge=0)
    stock_splits: float = Field(default=0.0, ge=0)
 
    @model_validator(mode="after")
    def high_ge_low(self):
        """Business rule: high must be >= low on any trading day."""
        if self.high < self.low:
            raise ValueError(f"high ({self.high}) < low ({self.low})")
        return self
 
# Verify the model works with sample data
sample = RawOHLCV(
    symbol="SAP.DE", date=Date(2024, 1, 2),
    open=144.5, high=146.0, low=143.8, close=145.2,
    adj_close=145.2, volume=1_200_000
)
sample.symbol, sample.date, sample.close  # RawOHLCV validated
Notebook stdout is preserved below.
SAP.DE 2024-01-02 close=145.2

Pydantic — define Silver validation model with BaseModel and Field()

Silver Contract: CleanOHLCV

Extends Bronze with computed fields: daily_return, intraday_range, sma_20. Daily return constrained to [-50%, +50%].

Reference snippet for Pydantic — define Silver validation model with BaseModel and Field().

 
class CleanOHLCV(BaseModel):
    """Schema for cleaned OHLCV data — Silver boundary."""
    model_config = ConfigDict(strict=True)
 
    symbol:         str   = Field(..., min_length=1)
    date:           Date  = Field(...)
    open:           float = Field(..., gt=0)
    high:           float = Field(..., gt=0)
    low:            float = Field(..., gt=0)
    close:          float = Field(..., gt=0)
    adj_close:      float = Field(..., gt=0)
    volume:         int   = Field(..., ge=0)
    dividends:      float = Field(default=0.0, ge=0)
    stock_splits:   float = Field(default=0.0, ge=0)
    # ── Silver enrichment fields ──
    daily_return:   float = Field(..., description="Close-to-close return pct")
    intraday_range: float = Field(..., ge=0, description="(high-low)/close pct")
    sma_20:         float | None = Field(default=None, description="20-day simple moving avg")
    batch_id:       str   = Field(..., description="Lineage batch identifier")
 
    @model_validator(mode="after")
    def high_ge_low(self):
        if self.high < self.low:
            raise ValueError(f"high ({self.high}) < low ({self.low})")
        return self
 
Notebook stdout is preserved below.
CleanOHLCV model defined — 14 fields

Pydantic — define Gold validation models with BaseModel and Field()

Gold Contracts: Two Mart Tables

DailySummary: one row per trading day. SymbolProfile: one row per symbol with full-history stats. max_drawdown always ⇐ 0.

Reference snippet for Pydantic — define Gold validation models with BaseModel and Field().

 
class DailySummary(BaseModel):
    """Daily cross-sectional summary across all symbols — Gold mart."""
    date:             Date  = Field(...)
    symbols_traded:   int   = Field(..., ge=0)
    avg_return:       float = Field(...)
    max_return:       float = Field(...)
    min_return:       float = Field(...)
    total_volume:     int   = Field(..., ge=0)
    avg_intraday_pct: float = Field(..., ge=0)
    batch_id:         str   = Field(...)
 
class SymbolProfile(BaseModel):
    """Per-symbol summary statistics — Gold mart."""
    symbol:              str   = Field(..., min_length=1)
    total_trading_days:  int   = Field(..., ge=0)
    avg_daily_return:    float = Field(...)
    volatility:          float = Field(..., ge=0, description="Std dev of daily returns")
    max_drawdown:        float = Field(..., le=0, description="Max peak-to-trough decline")
    avg_volume:          float = Field(..., ge=0)
    total_dividends:     float = Field(default=0.0, ge=0)
    first_date:          Date  = Field(...)
    last_date:           Date  = Field(...)
    batch_id:            str   = Field(...)
 
len(DailySummary.model_fields)   # DailySummary fields
len(SymbolProfile.model_fields)  # SymbolProfile fields
Notebook stdout is preserved below.
8 fields
10 fields

Pydantic — define lineage tracking models with BaseModel and Field()

Lineage Model: StageLineage

Records what each stage produced: input/output/rejected rows, SHA-256 hash for tamper detection, duration.

Reference snippet for Pydantic — define lineage tracking models with BaseModel and Field().

 
class StageLineage(BaseModel):
    """Records what a single pipeline stage produced."""
    batch_id:       str      = Field(..., description="Links all stages in one run")
    stage:          str      = Field(..., description="bronze | silver | gold | export")
    started_at:     datetime = Field(...)
    completed_at:   datetime = Field(...)
    input_rows:     int      = Field(..., ge=0)
    output_rows:    int      = Field(..., ge=0)
    rows_rejected:  int      = Field(default=0, ge=0)
    output_hash:    str      = Field(..., description="SHA-256 of output for drift detection")
 
    @property
    def duration_ms(self) -> float:
        return (self.completed_at - self.started_at).total_seconds() * 1000
No direct stdout was emitted by this cell.

Context Architecture — Semantic Metadata Layer

This is where the two dimensions of the architecture intersect. The structural models above ensure the pipeline produces correct data. The semantic models below ensure the data is meaningful to any consumer — another pipeline, a dashboard, an LLM agent, an auditor — without reading the pipeline source code.

  • ColumnContext — documents what each column means, how it was derived, what NULL signifies. Without it, volatility: 0.0187 is an opaque number. With it: unit=decimal_ratio, formula=std(daily_return), annualize with sqrt(252)
  • BusinessContext — records WHY this run happened. Without it, two batches covering the same date range are indistinguishable. With it, one is trigger=scheduled and the other is trigger=reprocess, is_correction=true
  • TemporalContext — separates “what date is this data FOR” (as_of_date) from “when did we learn about it” (knowledge_date). Without it, a backfill loading 2024 data in 2026 looks like a normal 2026 run
  • StageContext — carries all of the above THROUGH the pipeline. Each stage inherits upstream warnings and adds its own. By gold, the context contains the full warning chain from every stage

Without Semantic Context

An AI agent queries gold_symbol_profile and sees volatility: 0.0187. It doesn’t know if that’s a percentage or a decimal, daily or annual, what formula produced it, or what NULL would mean. It guesses — or hallucinates an interpretation. The data contract eliminates this: unit=decimal_ratio, formula=std(daily_return), annualize with sqrt(252). The number becomes self-describing.

Fix: Attach ColumnContext to Every Computed Field

Define a ColumnContext Pydantic model carrying unit, formula, source_columns, null_meaning, and valid_range. Attach one instance to each column in the gold schema and export it alongside the Parquet file as a context.json sidecar. Any downstream consumer — human or AI — reads the sidecar before interpreting the number.

Pydantic — define column semantic metadata model with BaseModel

Semantic Metadata: ColumnContext

Describes WHAT a column means: computation formula, source columns, null semantics, valid range, derived vs raw.

Reference snippet for Pydantic — define column semantic metadata model with BaseModel.

 
class ColumnContext(BaseModel):
    name: str = Field(..., description="Column name")
    description: str = Field(..., description="Human-readable explanation")
    unit: str = Field(..., description="Unit: decimal_ratio, EUR, count, date, identifier")
    computation: str | None = Field(default=None, description="Formula or None for source fields")
    source_columns: list[str] = Field(default_factory=list)
    valid_range: tuple[float, float] | None = Field(default=None)
    null_semantics: str = Field(default="not_applicable")
    is_business_key: bool = Field(default=False)
    is_derived: bool = Field(default=False)
No direct stdout was emitted by this cell.

Pydantic — define column registries for each medallion layer

Column Registries per Layer

Each column has a ColumnContext entry. Feeds into data contracts and StageContext for cross-stage propagation.

Reference snippet for Pydantic — define column registries for each medallion layer.

 
BRONZE_COLUMNS = [
    ColumnContext(name="symbol", description="Yahoo Finance ticker symbol", unit="identifier", is_business_key=True),
    ColumnContext(name="date", description="Trading date (exchange local)", unit="date", is_business_key=True),
    ColumnContext(name="open", description="Opening price", unit="EUR", valid_range=(0.001, 100000)),
    ColumnContext(name="high", description="Highest price", unit="EUR", valid_range=(0.001, 100000)),
    ColumnContext(name="low", description="Lowest price", unit="EUR", valid_range=(0.001, 100000)),
    ColumnContext(name="close", description="Closing price", unit="EUR", valid_range=(0.001, 100000)),
    ColumnContext(name="adj_close", description="Adjusted close", unit="EUR", valid_range=(0.001, 100000)),
    ColumnContext(name="volume", description="Shares traded", unit="count", valid_range=(0, 1e12)),
    ColumnContext(name="dividends", description="Dividend paid", unit="EUR", valid_range=(0, 1000)),
    ColumnContext(name="stock_splits", description="Split ratio", unit="ratio", valid_range=(0, 100)),
]
 
SILVER_COLUMNS = BRONZE_COLUMNS + [
    ColumnContext(name="daily_return", description="Close-to-close return", unit="decimal_ratio",
                 computation="pct_change(close).over(symbol)", source_columns=["bronze.close"],
                 valid_range=(-0.5, 0.5), null_semantics="first_row_in_series", is_derived=True),
    ColumnContext(name="intraday_range", description="(high-low)/close", unit="decimal_ratio",
                 computation="(high - low) / close", source_columns=["bronze.high", "bronze.low", "bronze.close"],
                 valid_range=(0, 0.5), is_derived=True),
    ColumnContext(name="sma_20", description="20-day moving average of close", unit="EUR",
                 computation="close.rolling_mean(20).over(symbol)", source_columns=["bronze.close"],
                 valid_range=(0.001, 100000), null_semantics="insufficient_data", is_derived=True),
]
 
GOLD_DAILY_COLUMNS = [
    ColumnContext(name="date", description="Trading date", unit="date", is_business_key=True),
    ColumnContext(name="symbols_traded", description="Distinct symbols", unit="count",
                 computation="count(distinct symbol) per date", source_columns=["silver.symbol"], is_derived=True),
    ColumnContext(name="avg_return", description="Mean daily return", unit="decimal_ratio",
                 computation="mean(daily_return) per date", source_columns=["silver.daily_return"], is_derived=True),
    ColumnContext(name="max_return", description="Best return", unit="decimal_ratio", is_derived=True),
    ColumnContext(name="min_return", description="Worst return", unit="decimal_ratio", is_derived=True),
    ColumnContext(name="total_volume", description="Sum of volume", unit="count", is_derived=True),
    ColumnContext(name="avg_intraday_pct", description="Mean intraday range", unit="decimal_ratio", is_derived=True),
]
 
GOLD_PROFILE_COLUMNS = [
    ColumnContext(name="symbol", description="Yahoo Finance ticker symbol",
                 unit="identifier", is_business_key=True, null_semantics="not_applicable"),
    ColumnContext(name="total_trading_days", description="Number of trading days with data",
                 unit="count", computation="count(*) per symbol",
                 source_columns=["silver.date"], valid_range=(1, 5000), is_derived=True),
    ColumnContext(name="avg_daily_return", description="Mean daily close-to-close return over full history",
                 unit="decimal_ratio", computation="mean(daily_return) per symbol",
                 source_columns=["silver.daily_return"], valid_range=(-0.1, 0.1), is_derived=True),
    ColumnContext(name="volatility", description="Standard deviation of daily returns \u2014 annualize by multiplying by sqrt(252)",
                 unit="decimal_ratio", computation="std(daily_return) per symbol",
                 source_columns=["silver.daily_return"], valid_range=(0, 1), is_derived=True),
    ColumnContext(name="max_drawdown", description="Largest peak-to-trough decline in cumulative return (always negative or zero)",
                 unit="decimal_ratio", computation="min(cumulative_return - running_max(cumulative_return)) per symbol",
                 source_columns=["silver.daily_return"], valid_range=(-1, 0), is_derived=True),
    ColumnContext(name="avg_volume", description="Mean daily trading volume over full history",
                 unit="count", computation="mean(volume) per symbol",
                 source_columns=["silver.volume"], valid_range=(0, 1e12), is_derived=True),
    ColumnContext(name="total_dividends", description="Sum of all dividends paid over full history",
                 unit="EUR", computation="sum(dividends) per symbol",
                 source_columns=["silver.dividends"], valid_range=(0, 10000), is_derived=True),
]
 
len(BRONZE_COLUMNS), len(SILVER_COLUMNS), len(GOLD_DAILY_COLUMNS), len(GOLD_PROFILE_COLUMNS)  # column registries
Notebook stdout is preserved below.
Bronze=10, Silver=13, Gold Daily=7, Gold Profile=7

Pydantic — define business context model with BaseModel

BusinessContext: Run Trigger

Captures WHY this pipeline ran: scheduled, manual, backfill, reprocess, or test. is_correction flags data overwrites.

Reference snippet for Pydantic — define business context model with BaseModel.

 
class BusinessContext(BaseModel):
    trigger: str = Field(..., description="scheduled, manual, backfill, reprocess, test")
    reason: str | None = Field(default=None)
    business_date: Date = Field(default_factory=Date.today)
    is_correction: bool = Field(default=False)
    affected_symbols: list[str] | None = Field(default=None)
 
    @field_validator("trigger")
    @classmethod
    def validate_trigger(cls, v: str) -> str:
        valid = {"scheduled", "manual", "backfill", "reprocess", "test"}
        if v not in valid:
            raise ValueError(f"trigger must be one of {valid}")
        return v
No direct stdout was emitted by this cell.

Pydantic — define temporal context model with BaseModel

TemporalContext: Bi-Temporal Markers

as_of_date: business date the data represents. knowledge_date: when pipeline ingested it. Critical for backfills.

Reference snippet for Pydantic — define temporal context model with BaseModel.

 
class TemporalContext(BaseModel):
    as_of_date: Date = Field(..., description="Business date the data represents")
    knowledge_date: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    reporting_period_start: Date = Field(...)
    reporting_period_end: Date = Field(...)
    timezone: str = Field(default="UTC")
    is_backfill: bool = Field(default=False)
No direct stdout was emitted by this cell.

Pydantic — define stage context model for cross-stage propagation with BaseModel

StageContext: Cross-Stage Propagation

Created at stage start, carried forward. Each stage inherits upstream warnings and adds its own.

Reference snippet for Pydantic — define stage context model for cross-stage propagation with BaseModel.

 
class StageContext(BaseModel):
    batch_id: str = Field(...)
    stage: str = Field(...)
    upstream_stages: list[StageLineage] = Field(default_factory=list)
    data_warnings: list[str] = Field(default_factory=list)
    schema_version: str = Field(default="1.0")
    column_context: list[ColumnContext] = Field(default_factory=list)
    business_context: BusinessContext | None = Field(default=None)
    temporal_context: TemporalContext | None = Field(default=None)
 
    def add_warning(self, warning: str) -> None:
        self.data_warnings.append(warning)
        log.warning(f"  {self.stage}: {warning}")
 
    def for_next_stage(self, next_stage: str, lineage: StageLineage,
                       columns: list[ColumnContext]) -> "StageContext":
        return StageContext(
            batch_id=self.batch_id, stage=next_stage,
            upstream_stages=self.upstream_stages + [lineage],
            data_warnings=self.data_warnings.copy(),
            schema_version=self.schema_version, column_context=columns,
            business_context=self.business_context,
            temporal_context=self.temporal_context,
        )
No direct stdout was emitted by this cell.

Pydantic — define pipeline run context model with BaseModel

RunContext: Execution Envelope

Aggregates everything: stages, business context, temporal context, data warnings, contract version.

Reference snippet for Pydantic — define pipeline run context model with BaseModel.

 
class RunContext(BaseModel):
    """Full pipeline execution metadata."""
    batch_id:          str                    = Field(...)
    started_at:        datetime               = Field(...)
    completed_at:      datetime | None        = Field(default=None)
    symbols:           list[str]              = Field(...)
    date_range:        tuple[str, str]        = Field(...)
    polars_version:    str                    = Field(default=pl.__version__)
    stages:            list[StageLineage]     = Field(default_factory=list)
    status:            str                    = Field(default="running")
    business_context:  BusinessContext | None = Field(default=None)
    temporal_context:  TemporalContext | None = Field(default=None)
    data_warnings:     list[str]              = Field(default_factory=list)
    contract_version:  str                    = Field(default="1.0")
No direct stdout was emitted by this cell.

Pydantic — define data contract export function with model_json_schema()

Data Contract Export

Generates JSON Schema contracts with x-column-context: descriptions, formulas, units, null semantics.

Reference snippet for Pydantic — define data contract export function with model_json_schema().

 
def export_data_contracts(export_dir: Path) -> list[Path]:
    contracts_dir = export_dir / "contracts"
    contracts_dir.mkdir(parents=True, exist_ok=True)
    contracts = {
        "bronze_ohlcv": RawOHLCV, "silver_ohlcv": CleanOHLCV,
        "gold_daily_summary": DailySummary, "gold_symbol_profile": SymbolProfile,
    }
    registry = {
        "bronze_ohlcv": BRONZE_COLUMNS, "silver_ohlcv": SILVER_COLUMNS,
        "gold_daily_summary": GOLD_DAILY_COLUMNS, "gold_symbol_profile": GOLD_PROFILE_COLUMNS,
    }
    paths = []
    for name, model in contracts.items():
        schema = model.model_json_schema()
        schema["x-column-context"] = [col.model_dump() for col in registry.get(name, [])]
        schema["x-contract-version"] = "1.0"
        schema["x-generated-at"] = datetime.now(timezone.utc).isoformat()
        path = contracts_dir / f"{name}_contract.json"
        path.write_text(json.dumps(schema, indent=2, default=str), encoding="utf-8")
        paths.append(path)
        log.info(f"  Contract exported: {path.name}")
    return paths
 
No direct stdout was emitted by this cell.

3. Lineage & Context Infrastructure

These functions implement the ability to trace any data point from Gold back to its raw source with cryptographic proof. batch_id is the thread — every row in every table carries the UUID of the pipeline run that created it. compute_hash() produces a deterministic SHA-256: same data → same hash. If someone modifies a Silver row after the pipeline ran, the recomputed hash won’t match the recorded one. RunContext captures the full execution envelope — which symbols, what date range, which library versions, how many rejections.

Without Lineage Tracking

A stakeholder disputes a -16% drop in gold. Without lineage, you spend a day manually checking: was the source data correct? Did the transform produce the right number? Was the data modified after ingestion? With lineage, three queries answer all three questions — batch_id traces the row to its run, the hash proves no tampering, the RunContext shows zero rejections and the exact date range processed.

Fix: UUID Batch IDs and SHA-256 Row Hashes

Assign a uuid4() batch ID to every pipeline run and stamp every written row with it. Compute a SHA-256 hash of each row’s immutable fields at write time and store it in bronze_ohlcv.row_hash. On dispute, re-hash the stored row and compare against the stored hash — any mismatch proves post-write modification. The lineage_stages table provides the full audit trail by batch and stage.

Lineage helpers

Utility functions for batch identification, deterministic hashing, stage timing, and run context persistence.

uuid — generate unique batch ID with uuid4()

Batch ID: Unique Run Identifier

Every row carries this UUID. Trace any disputed value back to its pipeline run in one query.

Reference snippet for uuid — generate unique batch ID with uuid4().

 
def generate_batch_id() -> str:
    """Generate a unique batch identifier for this pipeline run."""
    return str(uuid.uuid4())
 
# Demo: generate a batch_id
demo_batch = generate_batch_id()
demo_batch  # Sample batch_id
Notebook stdout is preserved below.
22d9d5c9-3ea5-4474-b7e1-702da6e2e599

hashlib — compute deterministic DataFrame hash with sha256()

SHA-256 Hash: Tamper Detection

Same data produces the same hash. Modified rows break the hash match.

Reference snippet for hashlib — compute deterministic DataFrame hash with sha256().

 
def compute_hash(df: pl.DataFrame) -> str:
    """SHA-256 hash of DataFrame content for drift detection."""
    csv_bytes = df.sort(df.columns).write_csv().encode("utf-8")
    return hashlib.sha256(csv_bytes).hexdigest()[:16]
 
# Demo with a small frame
demo_df = pl.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
compute_hash(demo_df)  # Hash of demo frame
Notebook stdout is preserved below.
f67a232f1bb81bfa

Python — define stage start and end tracker with datetime.now()

Stage Tracking: Start/End Pattern

start_stage(): captures timestamp + input count. end_stage(): fills output metrics, computes hash.

Reference snippet for Python — define stage start and end tracker with datetime.now().

 
def start_stage(batch_id: str, stage: str, input_rows: int,
                stage_context: StageContext | None = None) -> dict:
    """Begin tracking a pipeline stage. Returns a context dict."""
    return {
        "batch_id": batch_id,
        "stage": stage,
        "started_at": datetime.now(timezone.utc),
        "input_rows": input_rows,
        "stage_context": stage_context,
    }
 
def end_stage(ctx: dict, output_df: pl.DataFrame,
              rows_rejected: int = 0) -> StageLineage:
    """Complete a pipeline stage. Returns a StageLineage record."""
    return StageLineage(
        batch_id=ctx["batch_id"],
        stage=ctx["stage"],
        started_at=ctx["started_at"],
        completed_at=datetime.now(timezone.utc),
        input_rows=ctx["input_rows"],
        output_rows=len(output_df),
        rows_rejected=rows_rejected,
        output_hash=compute_hash(output_df),
    )
 
No direct stdout was emitted by this cell.

Pydantic — save run context to JSON with model_dump_json()

Save RunContext to JSON

One JSON file per run, named by batch_id prefix. Full audit trail on disk.

Reference snippet for Pydantic — save run context to JSON with model_dump_json().

 
def save_run_context(ctx: RunContext) -> Path:
    """Serialize RunContext to JSON file in lineage directory."""
    path = LINEAGE_DIR / f"run_{ctx.batch_id[:8]}.json"
    path.write_text(ctx.model_dump_json(indent=2), encoding="utf-8")
    return path
 
Notebook stdout is preserved below.
save_run_context() defined — writes to C:\Users\aperi\DEV\LANG\data\pipeline\lineage

4. SQL Server Schema — Medallion Tables + Lineage

Nine tables implementing the full architecture — not just data storage but the complete operational infrastructure. Three groups: medallion tables (bronze_ohlcv, silver_ohlcv, gold_daily_summary, gold_symbol_profile) store data at three stages of refinement. Dimension tables (dim_symbol, dim_calendar) provide business context that enables context-driven decisions. Operational tables (lineage_stages, quarantine, context_log) store the metadata that makes the pipeline auditable, recoverable, and self-describing.

Without Operational Tables

Without lineage_stages: no record of which batch produced which rows. Without quarantine: rejected rows disappear — you never know they existed, never know what was wrong with them, can never replay them. Without context_log: the pipeline’s knowledge about holidays, expected nulls, and business triggers is lost the moment the process exits.

Fix: Create All Three Operational Tables at Schema Init

Provision lineage_stages, quarantine, and context_log in the same DDL script that creates bronze_ohlcv. Write to lineage_stages at every stage boundary (start + finish + row count). Route every ValidationError to quarantine with the raw payload and error message. Flush StageContext.warnings to context_log at the end of each run.

Table definitions

DDL for all nine schema tables: medallion data tables, dimension tables, and operational tables.

TablePurposeKey
bronze_ohlcvRaw yfinance data, untransformed(symbol, date)
silver_ohlcvEnriched with daily_return, sma_20(symbol, date)
gold_daily_summaryCross-sectional daily metrics(date)
gold_symbol_profilePer-symbol aggregate stats(symbol)
dim_symbolSCD Type 2 company metadata(symbol, valid_from)
dim_calendarPer-exchange trading day flags(date, exchange_code)
lineage_stagesStage-level execution metadata(batch_id, stage)
quarantineDead letter queue for rejected rows(batch_id, stage)
context_logSemantic context per stage per run(batch_id, stage)

SQL Server — create Bronze OHLCV table with cursor.execute()

Bronze Table: Raw Source Data

Stores raw yfinance output exactly as received. UNIQUE on (symbol, date) enables MERGE upsert.

Reference snippet for SQL Server — create Bronze OHLCV table with cursor.execute().

 
sql_conn = pyodbc.connect(SQL_CONN_STR)
cur = sql_conn.cursor()
 
cur.execute("""
IF NOT EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'bronze_ohlcv')
CREATE TABLE bronze_ohlcv (
    id           INT IDENTITY(1,1) PRIMARY KEY,
    symbol       VARCHAR(20)  NOT NULL,
    date         DATE         NOT NULL,
    [open]       FLOAT        NOT NULL,
    high         FLOAT        NOT NULL,
    low          FLOAT        NOT NULL,
    [close]      FLOAT        NOT NULL,
    adj_close    FLOAT        NOT NULL,
    volume       BIGINT       NOT NULL,
    dividends    FLOAT        NOT NULL DEFAULT 0,
    stock_splits FLOAT        NOT NULL DEFAULT 0,
    batch_id     VARCHAR(36)  NOT NULL,
    ingested_at  DATETIME2    NOT NULL DEFAULT GETUTCDATE(),
    CONSTRAINT UQ_bronze_symbol_date UNIQUE (symbol, date)
)
""")
sql_conn.commit()
Notebook stdout is preserved below.
bronze_ohlcv table ready (with UNIQUE on symbol+date)

SQL Server — create Silver OHLCV table with cursor.execute()

Silver Table DDL

Adds computed columns: daily_return, intraday_range, sma_20. UNIQUE on (symbol, date) enables MERGE upsert.

Reference snippet for SQL Server — create Silver OHLCV table with cursor.execute().

 
cur.execute("""
IF NOT EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'silver_ohlcv')
CREATE TABLE silver_ohlcv (
    id              INT IDENTITY(1,1) PRIMARY KEY NONCLUSTERED,
    symbol          VARCHAR(20)  NOT NULL,
    date            DATE         NOT NULL,
    [open]          FLOAT        NOT NULL,
    high            FLOAT        NOT NULL,
    low             FLOAT        NOT NULL,
    [close]         FLOAT        NOT NULL,
    adj_close       FLOAT        NOT NULL,
    volume          BIGINT       NOT NULL,
    dividends       FLOAT        NOT NULL DEFAULT 0,
    stock_splits    FLOAT        NOT NULL DEFAULT 0,
    daily_return    FLOAT        NOT NULL,
    intraday_range  FLOAT        NOT NULL,
    sma_20          FLOAT        NULL,
    batch_id        VARCHAR(36)  NOT NULL,
    processed_at    DATETIME2    NOT NULL DEFAULT GETUTCDATE(),
    INDEX IX_silver_symbol_date CLUSTERED (symbol, date),
    CONSTRAINT UQ_silver_symbol_date UNIQUE (symbol, date)
)
""")
sql_conn.commit()
Notebook stdout is preserved below.
silver_ohlcv table ready (with UNIQUE on symbol+date)

SQL Server — create Gold daily summary table with cursor.execute()

Gold Daily Summary DDL

One row per trading day with cross-sectional metrics. Clustered on date for efficient range scans.

Reference snippet for SQL Server — create Gold daily summary table with cursor.execute().

 
cur.execute("""
IF NOT EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'gold_daily_summary')
CREATE TABLE gold_daily_summary (
    id               INT IDENTITY(1,1) PRIMARY KEY NONCLUSTERED,
    date             DATE         NOT NULL,
    symbols_traded   INT          NOT NULL,
    avg_return       FLOAT        NOT NULL,
    max_return       FLOAT        NOT NULL,
    min_return       FLOAT        NOT NULL,
    total_volume     BIGINT       NOT NULL,
    avg_intraday_pct FLOAT        NOT NULL,
    batch_id         VARCHAR(36)  NOT NULL,
    INDEX IX_gold_daily_date CLUSTERED (date)
)
""")
sql_conn.commit()
No direct stdout was emitted by this cell.

SQL Server — create Gold symbol profile table with cursor.execute()

Gold Symbol Profile DDL

One row per symbol with aggregate statistics. Clustered on symbol for efficient lookups.

Reference snippet for SQL Server — create Gold symbol profile table with cursor.execute().

 
cur.execute("""
IF NOT EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'gold_symbol_profile')
CREATE TABLE gold_symbol_profile (
    id                 INT IDENTITY(1,1) PRIMARY KEY NONCLUSTERED,
    symbol             VARCHAR(20)  NOT NULL,
    total_trading_days INT          NOT NULL,
    avg_daily_return   FLOAT        NOT NULL,
    volatility         FLOAT        NOT NULL,
    max_drawdown       FLOAT        NOT NULL,
    avg_volume         FLOAT        NOT NULL,
    total_dividends    FLOAT        NOT NULL DEFAULT 0,
    first_date         DATE         NOT NULL,
    last_date          DATE         NOT NULL,
    batch_id           VARCHAR(36)  NOT NULL,
    INDEX IX_gold_profile_symbol CLUSTERED (symbol)
)
""")
sql_conn.commit()
No direct stdout was emitted by this cell.

SQL Server — create SCD Type 2 symbol dimension with cursor.execute()

SCD Type 2 Dimension DDL

Tracks historical changes in symbol metadata. valid_from/valid_to/is_current enable point-in-time queries.

Reference snippet for SQL Server — create SCD Type 2 symbol dimension with cursor.execute().

 
cur.execute("""
IF NOT EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'dim_symbol')
CREATE TABLE dim_symbol (
    id                      INT IDENTITY(1,1) PRIMARY KEY,
    symbol                  VARCHAR(20)   NOT NULL,
    company_name            NVARCHAR(200) NULL,
    short_name              NVARCHAR(100) NULL,
    sector                  NVARCHAR(100) NULL,
    sector_key              VARCHAR(100)  NULL,
    industry                NVARCHAR(200) NULL,
    industry_key            VARCHAR(200)  NULL,
    country                 NVARCHAR(100) NULL,
    city                    NVARCHAR(100) NULL,
    exchange                VARCHAR(20)   NULL,
    full_exchange_name      NVARCHAR(100) NULL,
    currency                VARCHAR(10)   NULL,
    market_cap              BIGINT        NULL,
    website                 VARCHAR(500)  NULL,
    -- SCD Type 2 columns
    valid_from              DATETIME2     NOT NULL DEFAULT SYSUTCDATETIME(),
    valid_to                DATETIME2     NULL,
    is_current              BIT           NOT NULL DEFAULT 1
)
""")
sql_conn.commit()
Notebook stdout is preserved below.
dim_symbol table ready (SCD Type 2)

SQL Server — create per-exchange trading calendar with cursor.execute()

Trading Calendar Dimension

Per-exchange trading calendar with holiday flags. Composite PK on (date, exchange_code).

Reference snippet for SQL Server — create per-exchange trading calendar with cursor.execute().

 
cur.execute("""
IF NOT EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'dim_calendar')
CREATE TABLE dim_calendar (
    date            DATE         NOT NULL,
    exchange_code   VARCHAR(10)  NOT NULL,
    year            SMALLINT     NOT NULL,
    quarter         TINYINT      NOT NULL,
    month           TINYINT      NOT NULL,
    week_of_year    TINYINT      NOT NULL,
    day_of_week     TINYINT      NOT NULL,
    is_trading_day  BIT          NOT NULL DEFAULT 0,
    is_month_end    BIT          NOT NULL DEFAULT 0,
    is_quarter_end  BIT          NOT NULL DEFAULT 0,
    CONSTRAINT PK_dim_calendar PRIMARY KEY (date, exchange_code)
)
""")
sql_conn.commit()
Notebook stdout is preserved below.
dim_calendar table ready (per-exchange)

SQL Server — create lineage tracking table with cursor.execute()

Lineage Table DDL

Persists StageLineage records to SQL Server. Enables querying pipeline history: which batch produced what.

Reference snippet for SQL Server — create lineage tracking table with cursor.execute().

 
cur.execute("""
IF NOT EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'lineage_stages')
CREATE TABLE lineage_stages (
    id            INT IDENTITY(1,1) PRIMARY KEY,
    batch_id      VARCHAR(36)  NOT NULL,
    stage         VARCHAR(20)  NOT NULL,
    started_at    DATETIME2    NOT NULL,
    completed_at  DATETIME2    NOT NULL,
    input_rows    INT          NOT NULL,
    output_rows   INT          NOT NULL,
    rows_rejected INT          NOT NULL DEFAULT 0,
    output_hash   VARCHAR(16)  NOT NULL
)
""")
sql_conn.commit()
No direct stdout was emitted by this cell.

SQL Server — create quarantine table for rejected rows with cursor.execute()

Quarantine: Dead Letter Queue

Stores every row that failed Pydantic validation with raw data + rejection reason.

Reference snippet for SQL Server — create quarantine table for rejected rows with cursor.execute().

 
cur.execute("""
IF NOT EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'quarantine')
CREATE TABLE quarantine (
    id              INT IDENTITY(1,1) PRIMARY KEY,
    batch_id        VARCHAR(36)   NOT NULL,
    stage           VARCHAR(20)   NOT NULL,
    symbol          VARCHAR(20)   NULL,
    date            DATE          NULL,
    raw_data        NVARCHAR(MAX) NOT NULL,
    error_message   NVARCHAR(MAX) NOT NULL,
    quarantined_at  DATETIME2     NOT NULL DEFAULT GETUTCDATE()
)
""")
sql_conn.commit()
Notebook stdout is preserved below.
quarantine table ready (dead letter queue)

Persistence helpers

Helper functions for writing to operational tables, MERGE upserts, and API retry logic.

SQL Server — create context log table with cursor.execute()

Persists StageContext records: business context, temporal context, and data warnings per stage.

Reference snippet for SQL Server — create context log table with cursor.execute().

 
cur.execute("""
IF NOT EXISTS (SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'context_log')
CREATE TABLE context_log (
    id              INT IDENTITY(1,1) PRIMARY KEY,
    batch_id        VARCHAR(36)    NOT NULL,
    stage           VARCHAR(20)    NOT NULL,
    business_date   DATE           NULL,
    trigger_type    VARCHAR(20)    NULL,
    is_correction   BIT            NOT NULL DEFAULT 0,
    schema_version  VARCHAR(10)    NOT NULL DEFAULT '1.0',
    data_warnings   NVARCHAR(MAX)  NULL,
    column_context  NVARCHAR(MAX)  NULL,
    temporal_json   NVARCHAR(MAX)  NULL,
    created_at      DATETIME2      NOT NULL DEFAULT GETUTCDATE()
)
""")
sql_conn.commit()
log.info("context_log table ready")
No direct stdout was emitted by this cell.

SQL Server — define context persistence helper with cursor.execute()

Writes a StageContext record to the context_log table, including business date, trigger type, warnings, and column context JSON.

Reference snippet for SQL Server — define context persistence helper with cursor.execute().

 
def persist_context(stage_ctx: StageContext | None) -> None:
    if stage_ctx is None:
        return
    cur.execute(
        """INSERT INTO context_log
           (batch_id, stage, business_date, trigger_type, is_correction,
            schema_version, data_warnings, column_context, temporal_json)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        stage_ctx.batch_id, stage_ctx.stage,
        stage_ctx.business_context.business_date if stage_ctx.business_context else None,
        stage_ctx.business_context.trigger if stage_ctx.business_context else None,
        stage_ctx.business_context.is_correction if stage_ctx.business_context else False,
        stage_ctx.schema_version,
        json.dumps(stage_ctx.data_warnings) if stage_ctx.data_warnings else None,
        json.dumps([c.model_dump() for c in stage_ctx.column_context], default=str) if stage_ctx.column_context else None,
        stage_ctx.temporal_context.model_dump_json() if stage_ctx.temporal_context else None,
    )
    sql_conn.commit()
 
No direct stdout was emitted by this cell.

SQL Server — define lineage persistence helper with cursor.execute()

Idempotent Lineage Persistence

Inserts a StageLineage into lineage_stages. Deletes any existing record for the same batch+stage first.

Reference snippet for SQL Server — define lineage persistence helper with cursor.execute().

 
def persist_lineage(lineage: StageLineage) -> None:
    """Write a StageLineage record to SQL Server (idempotent)."""
    # Delete existing record for this batch+stage to allow re-runs
    cur.execute(
        "DELETE FROM lineage_stages WHERE batch_id = ? AND stage = ?",
        lineage.batch_id, lineage.stage
    )
    cur.execute(
        """INSERT INTO lineage_stages
           (batch_id, stage, started_at, completed_at, input_rows, output_rows, rows_rejected, output_hash)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
        lineage.batch_id, lineage.stage,
        lineage.started_at, lineage.completed_at,
        lineage.input_rows, lineage.output_rows,
        lineage.rows_rejected, lineage.output_hash
    )
    sql_conn.commit()
 
Notebook stdout is preserved below.
persist_lineage() defined — idempotent: deletes before insert

SQLAlchemy — define DataFrame write helper with to_sql()

DataFrame Write Helper

Converts Polars to pandas for SQLAlchemy. chunksize=100 avoids SQL Server 2100 parameter limit.

Reference snippet for SQLAlchemy — define DataFrame write helper with to_sql().

 
def write_to_sql(df: pl.DataFrame, table: str, truncate: bool = True) -> int:
    """Write a Polars DataFrame to SQL Server. Truncates table first by default."""
    if truncate:
        cur.execute(f"TRUNCATE TABLE {table}")
        sql_conn.commit()
    pdf = df.to_pandas()
    pdf.to_sql(table, sql_engine, if_exists="append", index=False, chunksize=100)
    return len(pdf)
 
No direct stdout was emitted by this cell.

SQL Server — define Bronze MERGE upsert with MERGE INTO

Bronze MERGE Upsert

Idempotent: MERGE on (symbol, date). Existing rows updated, new rows inserted. Safe to re-run.

Reference snippet for SQL Server — define Bronze MERGE upsert with MERGE INTO.

 
def merge_bronze(df: pl.DataFrame, batch_id: str) -> int:
    """MERGE upsert into bronze_ohlcv on (symbol, date)."""
    rows_affected = 0
    for row in df.iter_rows(named=True):
        cur.execute("""
            MERGE bronze_ohlcv AS tgt
            USING (SELECT ? AS symbol, ? AS date) AS src
               ON tgt.symbol = src.symbol AND tgt.date = src.date
            WHEN MATCHED THEN UPDATE SET
                [open] = ?, high = ?, low = ?, [close] = ?,
                adj_close = ?, volume = ?, dividends = ?, stock_splits = ?,
                batch_id = ?, ingested_at = GETUTCDATE()
            WHEN NOT MATCHED THEN INSERT
                (symbol, date, [open], high, low, [close], adj_close,
                 volume, dividends, stock_splits, batch_id)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
        """,
            row["symbol"], row["date"],
            row["open"], row["high"], row["low"], row["close"],
            row["adj_close"], row["volume"], row["dividends"], row["stock_splits"],
            batch_id,
            row["symbol"], row["date"],
            row["open"], row["high"], row["low"], row["close"],
            row["adj_close"], row["volume"], row["dividends"], row["stock_splits"],
            batch_id,
        )
        rows_affected += cur.rowcount
    sql_conn.commit()
    return rows_affected
 
No direct stdout was emitted by this cell.

SQL Server — define Silver MERGE upsert with MERGE INTO

Silver MERGE Upsert

Same pattern as Bronze, plus enrichment columns: daily_return, intraday_range, sma_20.

Reference snippet for SQL Server — define Silver MERGE upsert with MERGE INTO.

 
def merge_silver(df: pl.DataFrame, batch_id: str) -> int:
    """MERGE upsert into silver_ohlcv on (symbol, date)."""
    rows_affected = 0
    for row in df.iter_rows(named=True):
        cur.execute("""
            MERGE silver_ohlcv AS tgt
            USING (SELECT ? AS symbol, ? AS date) AS src
               ON tgt.symbol = src.symbol AND tgt.date = src.date
            WHEN MATCHED THEN UPDATE SET
                [open] = ?, high = ?, low = ?, [close] = ?,
                adj_close = ?, volume = ?, dividends = ?, stock_splits = ?,
                daily_return = ?, intraday_range = ?, sma_20 = ?,
                batch_id = ?, processed_at = GETUTCDATE()
            WHEN NOT MATCHED THEN INSERT
                (symbol, date, [open], high, low, [close], adj_close,
                 volume, dividends, stock_splits,
                 daily_return, intraday_range, sma_20, batch_id)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
        """,
            row["symbol"], row["date"],
            row["open"], row["high"], row["low"], row["close"],
            row["adj_close"], row["volume"], row["dividends"], row["stock_splits"],
            row["daily_return"], row["intraday_range"], row["sma_20"],
            batch_id,
            row["symbol"], row["date"],
            row["open"], row["high"], row["low"], row["close"],
            row["adj_close"], row["volume"], row["dividends"], row["stock_splits"],
            row["daily_return"], row["intraday_range"], row["sma_20"],
            batch_id,
        )
        rows_affected += cur.rowcount
    sql_conn.commit()
    return rows_affected
 
No direct stdout was emitted by this cell.

SQL Server — define quarantine persistence helper with cursor.execute()

Quarantine Row Persistence

Saves rejected row to quarantine table with its error message.

Reference snippet for SQL Server — define quarantine persistence helper with cursor.execute().

 
def quarantine_row(batch_id: str, stage: str, row_data: dict, error: str) -> None:
    """Save a rejected row to the quarantine table."""
    cur.execute(
        """INSERT INTO quarantine (batch_id, stage, symbol, date, raw_data, error_message)
           VALUES (?, ?, ?, ?, ?, ?)""",
        batch_id,
        stage,
        row_data.get("symbol"),
        row_data.get("date"),
        json.dumps(row_data, default=str),
        str(error)[:4000],
    )
    sql_conn.commit()
 
log.info("quarantine_row() defined \u2014 dead letter queue helper")
Notebook stdout is preserved below.
23:19:35 | INFO  | quarantine_row() defined — dead letter queue helper

tenacity — define API retry wrapper with @retry() exponential backoff

API Retry with Backoff

3 attempts, exponential backoff. Catches network errors without killing the pipeline.

Reference snippet for tenacity — define API retry wrapper with @retry() exponential backoff.

 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type((ConnectionError, TimeoutError, OSError)),
    before_sleep=lambda rs: log.warning(f"Retry {rs.attempt_number}/3, waiting..."),
)
def fetch_with_retry(ticker, start: str, end: str):
    """Fetch ticker history with automatic retry on transient failures."""
    return ticker.history(start=start, end=end, auto_adjust=False)
 
log.info("fetch_with_retry() defined \u2014 3 attempts, exponential backoff")
Notebook stdout is preserved below.
23:19:35 | INFO  | fetch_with_retry() defined — 3 attempts, exponential backoff

Quality gates

Reusable assertion functions that check DataFrame properties; run together as a gated pass/fail before each stage proceeds.

Python — define custom Exception subclass for quality gate failures

Quality Gate Exception

Raised when a data quality gate fails. Blocks downstream stages from processing bad data.

Reference snippet for Python — define custom Exception subclass for quality gate failures.

 
class DataQualityError(Exception):
    """Raised when a data quality gate fails."""
    pass
 
No direct stdout was emitted by this cell.

Polars — assert DataFrame is not empty with len()

Assert: Not Empty

Verifies the output table/frame is not empty after a stage.

Reference snippet for Polars — assert DataFrame is not empty with len().

 
def dq_check_not_empty(df: pl.DataFrame, stage: str) -> tuple[bool, str]:
    """Assert DataFrame is not empty."""
    ok = len(df) > 0
    return ok, f"{stage}: {len(df)} rows" if ok else f"{stage}: EMPTY DataFrame"
 
No direct stdout was emitted by this cell.

Polars — assert no nulls in key columns with null_count()

Assert: No Null Keys

Checks each key column individually, reports first failure found.

Reference snippet for Polars — assert no nulls in key columns with null_count().

 
def dq_check_no_null_keys(df: pl.DataFrame, keys: list[str], stage: str) -> tuple[bool, str]:
    """Assert no nulls in key columns."""
    for col in keys:
        if col not in df.columns:
            return False, f"{stage}: column '{col}' missing"
        nulls = df[col].null_count()
        if nulls > 0:
            return False, f"{stage}: {nulls} nulls in '{col}'"
    return True, f"{stage}: no null keys in {keys}"
 
No direct stdout was emitted by this cell.

Polars — assert no duplicate rows with unique()

Assert: No Duplicates

Compares total rows vs unique key combinations to detect duplicates.

Reference snippet for Polars — assert no duplicate rows with unique().

 
def dq_check_no_duplicates(df: pl.DataFrame, keys: list[str], stage: str) -> tuple[bool, str]:
    """Assert no duplicate rows on key columns."""
    total = len(df)
    unique = df.select(keys).unique().height
    dupes = total - unique
    ok = dupes == 0
    return ok, f"{stage}: {dupes} duplicates on {keys}" if not ok else f"{stage}: no duplicates"
 
No direct stdout was emitted by this cell.

Polars — assert values within range with filter()

Assert: Value Range

Reports count of out-of-range values in the specified column.

Reference snippet for Polars — assert values within range with filter().

 
def dq_check_range(df: pl.DataFrame, col: str, min_val: float, max_val: float, stage: str) -> tuple[bool, str]:
    """Assert all values in a column fall within [min_val, max_val]."""
    out_of_range = df.filter(
        (pl.col(col) < min_val) | (pl.col(col) > max_val)
    ).height
    ok = out_of_range == 0
    return ok, f"{stage}: {out_of_range} values outside [{min_val}, {max_val}] in '{col}'" if not ok else f"{stage}: '{col}' within range"
 
No direct stdout was emitted by this cell.

Polars — assert data freshness against SLA with max()

Assert: Data Freshness

Detects stale data that missed recent trading days.

Reference snippet for Polars — assert data freshness against SLA with max().

 
def dq_check_freshness(df: pl.DataFrame, date_col: str, max_age_days: int, stage: str) -> tuple[bool, str]:
    """Assert most recent date is within max_age_days of today."""
    if len(df) == 0:
        return False, f"{stage}: empty DataFrame, can't check freshness"
    latest_raw = df[date_col].max()
    if latest_raw is None:
        return False, f"{stage}: all dates are null"
    latest = Date.fromisoformat(str(latest_raw)) if not isinstance(latest_raw, Date) else latest_raw
    age = (Date.today() - latest).days
    ok = age <= max_age_days
    return ok, f"{stage}: latest date {latest} ({age}d ago)" + ("" if ok else f" EXCEEDS {max_age_days}d SLA")
 
No direct stdout was emitted by this cell.

Polars — assert minimum row count with len()

Assert: Minimum Row Count

Catches partial loads or missing symbols.

Reference snippet for Polars — assert minimum row count with len().

 
def dq_check_row_count(df: pl.DataFrame, min_rows: int, stage: str) -> tuple[bool, str]:
    """Assert DataFrame has at least min_rows."""
    ok = len(df) >= min_rows
    return ok, f"{stage}: {len(df)} rows" + ("" if ok else f" BELOW minimum {min_rows}")
 
No direct stdout was emitted by this cell.

Pipeline — run all quality gate assertions with log.info()

Quality Gate Runner

Executes all checks, logs PASS/FAIL. fail_fast=True raises DataQualityError, blocking downstream.

Reference snippet for Pipeline — run all quality gate assertions with log.info().

 
def run_quality_gate(checks: list[tuple[bool, str]], stage: str, fail_fast: bool = True) -> pl.DataFrame:
    """Run all quality checks. Logs results. Raises DataQualityError if any fail."""
    results = []
    all_passed = True
    for passed, msg in checks:
        status = "PASS" if passed else "FAIL"
        log_fn = log.info if passed else log.error
        log_fn(f"  DQ {status}: {msg}")
        results.append({"check": msg, "status": status})
        if not passed:
            all_passed = False
 
    if not all_passed and fail_fast:
        raise DataQualityError(f"Data quality gate FAILED for {stage}")
 
    return pl.DataFrame(results)
 
No direct stdout was emitted by this cell.

5. Dimension Tables — Symbol Metadata (SCD2) & Trading Calendar

Dimensions are the pipeline’s external knowledge — facts about the world that the pipeline needs but doesn’t compute. dim_symbol uses SCD Type 2 because company metadata changes over time — without historization, a join between gold scores and dim_symbol shows today’s sector for historical dates, producing misleading analysis. dim_calendar exists because zero-volume doesn’t always mean bad data — the calendar tells the pipeline whether an exchange was open, turning undifferentiated zero-volume alerts into classified holidays vs genuine anomalies.

Without Trading Calendar

Every zero-volume day triggers an investigation. Good Friday, Christmas, local exchange holidays — all flagged as anomalies. An on-call engineer wastes time cross-referencing exchange schedules. With dim_calendar, the pipeline classifies each zero-volume date at ingestion and records the classification as a context warning.

Fix: Populate dim_calendar with Exchange Holidays

Load dim_calendar from a known holiday source (e.g. pandas_market_calendars for the relevant exchange). During bronze validation, join each date against dim_calendar and automatically classify zero-volume rows as holiday, weekend, or genuine_anomaly. Record the classification as a StageContext warning — not an error — so the pipeline continues without false alerts.

Symbol metadata (SCD2)

Fetch, load, and SCD Type 2 upsert company metadata from yfinance into dim_symbol.

yfinance — fetch symbol metadata to JSON landing zone with Ticker.info

Fetch Symbol Metadata

Fetches company metadata from yfinance. Saves to landing/dim_symbol.json for replay.

Reference snippet for yfinance — fetch symbol metadata to JSON landing zone with Ticker.info.

 
SCD2_COMPARE_COLS = ["company_name", "sector", "industry", "country", "exchange", "currency"]
 
def fetch_symbols_to_landing(symbols: list[str]) -> Path:
    """Fetch metadata from yfinance and save to JSON landing zone."""
    records = []
    for symbol in symbols:
        ticker = yf.Ticker(symbol)
        try:
            info = ticker.info or {}
        except Exception as e:
            log.warning(f"Fetch failed for {symbol}: {e}, using empty info")
            info = {}
        sector = info.get("sector")
        industry = info.get("industry")
        rec = {
            "symbol": symbol,
            "longName": info.get("longName"),
            "shortName": info.get("shortName"),
            "sector": sector,
            "sectorKey": sector.lower().replace(" ", "_") if sector else None,
            "industry": industry,
            "industryKey": industry.lower().replace(" ", "_") if industry else None,
            "country": info.get("country"),
            "city": info.get("city"),
            "website": info.get("website"),
            "longBusinessSummary": info.get("longBusinessSummary"),
            "exchange": info.get("exchange"),
            "fullExchangeName": info.get("fullExchangeName"),
            "exchangeTimezoneName": info.get("exchangeTimezoneName"),
            "exchangeTimezoneShortName": info.get("exchangeTimezoneShortName"),
            "currency": info.get("currency"),
            "financialCurrency": info.get("financialCurrency"),
            "quoteType": info.get("quoteType"),
            "market": info.get("market"),
            "marketCap": info.get("marketCap"),
        }
        records.append(rec)
        log.info(f"  {symbol}: fetched ({rec['longName']})")
 
    landing_path = LANDING_DIR / "dim_symbol.json"
    landing_path.write_text(json.dumps(records, indent=2, default=str), encoding="utf-8")
    log.info(f"Landed: {landing_path} ({len(records)} symbols)")
    return landing_path
 
No direct stdout was emitted by this cell.

JSON — load symbol metadata from landing zone with json.loads()

Load Symbols from Landing

Reads the JSON landing file and returns records ready for SCD2 upsert.

Reference snippet for JSON — load symbol metadata from landing zone with json.loads().

 
def load_symbols_from_landing() -> list[dict]:
    """Read symbol metadata from JSON landing zone."""
    landing_path = LANDING_DIR / "dim_symbol.json"
    return json.loads(landing_path.read_text(encoding="utf-8"))
 
No direct stdout was emitted by this cell.

SQL Server — define SCD Type 2 upsert for one symbol with MERGE INTO

SCD Type 2 Dimension Upsert

New symbol: INSERT. Unchanged: skip. Changed: close old record, INSERT new version.

Reference snippet for SQL Server — define SCD Type 2 upsert for one symbol with MERGE INTO.

 
def scd2_upsert_symbol(rec: dict) -> str:
    """SCD Type 2 upsert for one symbol from landed JSON. Returns action taken."""
    db_rec = {
        "symbol": rec["symbol"],
        "company_name": rec.get("longName") or rec.get("shortName"),
        "short_name": rec.get("shortName"),
        "sector": rec.get("sector"),
        "sector_key": rec.get("sectorKey"),
        "industry": rec.get("industry"),
        "industry_key": rec.get("industryKey"),
        "country": rec.get("country"),
        "city": rec.get("city"),
        "exchange": rec.get("exchange"),
        "full_exchange_name": rec.get("fullExchangeName"),
        "currency": rec.get("currency"),
        "market_cap": rec.get("marketCap"),
        "website": rec.get("website"),
    }
 
    # Check current record
    cur.execute(
        "SELECT id, company_name, sector, industry, country, exchange, currency "
        "FROM dim_symbol WHERE symbol = ? AND is_current = 1",
        db_rec["symbol"]
    )
    existing = cur.fetchone()
 
    if existing is None:
        cur.execute("""
            INSERT INTO dim_symbol
                (symbol, company_name, short_name, sector, sector_key,
                 industry, industry_key, country, city, exchange,
                 full_exchange_name, currency, market_cap, website)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
            db_rec["symbol"], db_rec["company_name"], db_rec["short_name"],
            db_rec["sector"], db_rec["sector_key"], db_rec["industry"], db_rec["industry_key"],
            db_rec["country"], db_rec["city"], db_rec["exchange"],
            db_rec["full_exchange_name"], db_rec["currency"], db_rec["market_cap"],
            db_rec["website"],
        )
        sql_conn.commit()
        return "INSERT"
 
    # Compare tracked columns
    old_vals = (existing[1], existing[2], existing[3], existing[4], existing[5], existing[6])
    new_vals = (db_rec["company_name"], db_rec["sector"], db_rec["industry"],
                db_rec["country"], db_rec["exchange"], db_rec["currency"])
 
    if old_vals == new_vals:
        return "UNCHANGED"
 
    # Attribute changed → close old record, insert new
    cur.execute(
        "UPDATE dim_symbol SET valid_to = SYSUTCDATETIME(), is_current = 0 WHERE id = ?",
        existing[0]
    )
    cur.execute("""
        INSERT INTO dim_symbol
            (symbol, company_name, short_name, sector, sector_key,
             industry, industry_key, country, city, exchange,
             full_exchange_name, currency, market_cap, website)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """,
        db_rec["symbol"], db_rec["company_name"], db_rec["short_name"],
        db_rec["sector"], db_rec["sector_key"], db_rec["industry"], db_rec["industry_key"],
        db_rec["country"], db_rec["city"], db_rec["exchange"],
        db_rec["full_exchange_name"], db_rec["currency"], db_rec["market_cap"],
        db_rec["website"],
    )
    sql_conn.commit()
    return "SCD2_UPDATE"
 
No direct stdout was emitted by this cell.

SQL Server — orchestrate SCD Type 2 upsert for all symbols with cursor.execute()

SCD2 Upsert Orchestration

Read landing JSON, SCD2 upsert each symbol, log action taken per symbol.

Reference snippet for SQL Server — orchestrate SCD Type 2 upsert for all symbols with cursor.execute().

 
def populate_dim_symbol_from_landing() -> pl.DataFrame:
    """Load from landing JSON and apply SCD Type 2 upsert."""
    records = load_symbols_from_landing()
    results = []
    for rec in records:
        action = scd2_upsert_symbol(rec)
        rec["_action"] = action
        results.append(rec)
        log.info(f"  {rec['symbol']}: {action}")
    return pl.DataFrame(results)
 
No direct stdout was emitted by this cell.

SQL Server — load symbols from landing and SCD2 upsert with MERGE INTO

This orchestration step makes the two-stage boundary explicit: fetch external metadata into the landing zone first, then apply the SCD2 MERGE against dim_symbol. Keeping the fetch and merge in one visible sequence makes it easier to audit whether a bad attribute came from the source payload or from warehouse logic.

Reference snippet for SQL Server — load symbols from landing and SCD2 upsert with MERGE INTO.

# Step 1: Fetch from yfinance API → JSON landing zone
fetch_symbols_to_landing(SYMBOLS)
 
# Step 2: Load from landing JSON → SCD2 upsert into dim_symbol
dim_symbol_df = populate_dim_symbol_from_landing()
dim_symbol_df.select("symbol", "longName", "sector", "country", "exchange", "_action")
Notebook stdout is preserved below.
23:19:35 | INFO  |   SAP.DE: fetched (SAP SE)
23:19:36 | INFO  |   SIE.DE: fetched (Siemens Aktiengesellschaft)
23:19:36 | INFO  |   ALV.DE: fetched (Allianz SE)
23:19:36 | INFO  |   DTE.DE: fetched (Deutsche Telekom AG)
23:19:36 | INFO  |   BAS.DE: fetched (BASF SE)
23:19:36 | INFO  | Landed: C:\Users\aperi\DEV\LANG\data\pipeline\landing\dim_symbol.json (5 symbols)
23:19:36 | INFO  |   SAP.DE: UNCHANGED
23:19:36 | INFO  |   SIE.DE: UNCHANGED
23:19:36 | INFO  |   ALV.DE: UNCHANGED
23:19:36 | INFO  |   DTE.DE: UNCHANGED
23:19:36 | INFO  |   BAS.DE: UNCHANGED
symbollongNamesectorcountryexchange_action
SAP.DESAP SETechnologyGermanyGERUNCHANGED
SIE.DESiemens AktiengesellschaftIndustrialsGermanyGERUNCHANGED
ALV.DEAllianz SEFinancial ServicesGermanyGERUNCHANGED
DTE.DEDeutsche Telekom AGCommunication ServicesGermanyGERUNCHANGED
BAS.DEBASF SEBasic MaterialsGermanyGERUNCHANGED

Trading calendar

Build and persist a per-exchange trading calendar using pandas-market-calendars, including holiday detection.

pandas-market-calendars — generate trading calendar with get_calendar().schedule()

Exchange Calendar Builder

Uses pandas-market-calendars for accurate per-exchange trading days with holiday detection.

Reference snippet for pandas-market-calendars — generate trading calendar with get_calendar().schedule().

 
def generate_dim_calendar(start: str, end: str, exchange_codes: list[str]) -> pl.DataFrame:
    """Generate a per-exchange calendar using pandas-market-calendars."""
    frames = []
 
    for yf_code in exchange_codes:
        # Map yfinance exchange code to mcal calendar name
        mcal_name = EXCHANGE_MAP.get(yf_code, yf_code)
 
        try:
            cal = mcal.get_calendar(mcal_name)
        except RuntimeError:
            print(f"  Warning: no calendar for {yf_code} ({mcal_name}), using XETR fallback")
            cal = mcal.get_calendar("XETR")
 
        # Get actual trading days from market calendar
        schedule = cal.schedule(start_date=start, end_date=end)
        trading_dates = {d.date() for d in schedule.index}
 
        # Build full date range
        all_dates = pl.date_range(
            Date.fromisoformat(start),
            Date.fromisoformat(end),
            eager=True
        ).alias("date")
 
        df = pl.DataFrame({"date": all_dates}).with_columns(
            pl.lit(yf_code).alias("exchange_code"),
            pl.col("date").dt.year().cast(pl.Int16).alias("year"),
            pl.col("date").dt.quarter().cast(pl.UInt8).alias("quarter"),
            pl.col("date").dt.month().cast(pl.UInt8).alias("month"),
            pl.col("date").dt.week().cast(pl.UInt8).alias("week_of_year"),
            pl.col("date").dt.weekday().cast(pl.UInt8).alias("day_of_week"),
        )
 
        # Mark trading days from mcal schedule
        trading_list = sorted(trading_dates)
        df = df.with_columns(
            pl.col("date").is_in(trading_list).cast(pl.Int8).alias("is_trading_day")
        )
 
        # Mark month-end and quarter-end
        df = df.with_columns(
            (pl.col("date") == pl.col("date").dt.month_end()).cast(pl.Int8).alias("is_month_end"),
            ((pl.col("month").is_in([3, 6, 9, 12])) &
             (pl.col("date") == pl.col("date").dt.month_end())).cast(pl.Int8).alias("is_quarter_end"),
        )
 
        frames.append(df)
        print(f"  {yf_code} ({mcal_name}): {len(trading_dates)} trading days, "
              f"{len(all_dates) - len(trading_dates)} non-trading")
 
    return pl.concat(frames).sort(["exchange_code", "date"])
 
# Get exchanges from dim_symbol (just populated)
exchanges = dim_symbol_df.select("exchange").unique().to_series().to_list()
exchanges = [e for e in exchanges if e is not None]
cal_df = generate_dim_calendar(START_DATE, END_DATE, exchanges)
cal_df.head()
Notebook-rendered table output is preserved below.
dateexchange_codeyearquartermonthweek_of_yearday_of_weekis_trading_dayis_month_endis_quarter_end
2024-03-29 00:00:00GER202413135000
2024-03-30 00:00:00GER202413136000
2024-03-31 00:00:00GER202413137011
2024-04-01 00:00:00GER202424141000
2024-04-02 00:00:00GER202424142100

SQL Server — persist calendar dimension with MERGE INTO

Calendar MERGE Upsert

MERGE upsert into dim_calendar. Key: (date, exchange_code).

Reference snippet for SQL Server — persist calendar dimension with MERGE INTO.

 
def persist_dim_calendar(cal_df: pl.DataFrame) -> int:
    """MERGE upsert calendar dimension into SQL Server."""
    count = 0
    for row in cal_df.iter_rows(named=True):
        cur.execute("""
            MERGE dim_calendar AS tgt
            USING (SELECT ? AS date, ? AS exchange_code) AS src
               ON tgt.date = src.date AND tgt.exchange_code = src.exchange_code
            WHEN MATCHED THEN UPDATE SET
                year = ?, quarter = ?, month = ?, week_of_year = ?,
                day_of_week = ?, is_trading_day = ?,
                is_month_end = ?, is_quarter_end = ?
            WHEN NOT MATCHED THEN INSERT
                (date, exchange_code, year, quarter, month, week_of_year,
                 day_of_week, is_trading_day, is_month_end, is_quarter_end)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
        """,
            row["date"], row["exchange_code"],
            row["year"], row["quarter"], row["month"], row["week_of_year"],
            row["day_of_week"], row["is_trading_day"],
            row["is_month_end"], row["is_quarter_end"],
            row["date"], row["exchange_code"],
            row["year"], row["quarter"], row["month"], row["week_of_year"],
            row["day_of_week"], row["is_trading_day"],
            row["is_month_end"], row["is_quarter_end"],
        )
        count += 1
    sql_conn.commit()
    print(f"dim_calendar: {count} rows upserted")
    return count
 
persist_dim_calendar(cal_df)
No direct stdout was emitted by this cell.
731

Polars — display detected exchange holidays with filter()

Detected Exchange Holidays

Display holidays detected by pandas-market-calendars (weekdays with no trading).

Reference snippet for Polars — display detected exchange holidays with filter().

 
holidays = cal_df.filter(
    (pl.col("day_of_week").is_between(1, 5)) &  # weekday
    (pl.col("is_trading_day") == 0)
).select("date", "exchange_code", "day_of_week").sort("exchange_code", "date")
 
len(holidays)  # holidays detected (weekdays with no trading)
holidays.head()
Notebook-rendered table output is preserved below.
dateexchange_codeday_of_week
2024-03-29 00:00:00GER5
2024-04-01 00:00:00GER1
2024-05-01 00:00:00GER3
2024-12-24 00:00:00GER2
2024-12-25 00:00:00GER3

6. Bronze Layer — Landing Zone + Incremental Ingestion

Bronze implements two principles. The landing zone decouples API fetching from database loading — API calls are unreliable and unrepeatable, so raw responses are saved as JSON files first. If the MERGE fails, data is still on disk. If the pipeline is replayed, it reads from files without re-calling the API. Contract enforcement at the bronze boundary is the first line of defense — business rule violations (high < low, negative prices, empty symbols) are caught here, not three stages later. Rejected rows go to the quarantine table with full error context — preserved for investigation and replay, never silently dropped.

Without Landing Zone

The pipeline calls the API and writes directly to SQL Server. The API changes its response format. The MERGE fails mid-batch. 3 of 5 symbols are loaded, 2 are missing, and there’s no way to replay because the API response is gone. With the landing zone, the raw JSON is on disk — fix the parser, re-run the load, no re-fetch needed.

Fix: Write Raw API Response to Landing Zone First

Persist every API response as a timestamped JSON file in the landing zone (landing/<symbol>/<date>.json) before parsing or loading. The fetch stage and the load stage are now independent — a parser bug can be fixed and the load re-run against the saved JSON without a second API call. Retain files for at least 30 days to cover any delayed re-processing need.

Data fetching

Functions to download OHLCV data from yfinance to the JSON landing zone and reload it into Polars.

yfinance — fetch OHLCV to JSON landing zone with Ticker.history()

Landing Zone: OHLCV Fetch

Downloads OHLCV data to landing/ohlcv_{symbol}.json. Each symbol gets its own file.

Reference snippet for yfinance — fetch OHLCV to JSON landing zone with Ticker.history().

 
def fetch_ohlcv_to_landing(symbol: str, start: str, end: str) -> Path | None:
    """Download OHLCV data from yfinance and save to JSON landing zone."""
    ticker = yf.Ticker(symbol)
    pdf = fetch_with_retry(ticker, start, end)
 
    if pdf.empty:
        return None
 
    # Convert to records for JSON serialization
    pdf = pdf.reset_index()
    records = []
    for _, row in pdf.iterrows():
        records.append({
            "symbol": symbol,
            "date": row["Date"].strftime("%Y-%m-%d"),
            "open": float(row["Open"]),
            "high": float(row["High"]),
            "low": float(row["Low"]),
            "close": float(row["Close"]),
            "adj_close": float(row["Adj Close"]),
            "volume": int(row["Volume"]),
            "dividends": float(row["Dividends"]),
            "stock_splits": float(row["Stock Splits"]),
        })
 
    safe_symbol = symbol.replace(".", "_")
    landing_path = LANDING_DIR / f"ohlcv_{safe_symbol}.json"
    landing_path.write_text(json.dumps(records, indent=2), encoding="utf-8")
    return landing_path
 
No direct stdout was emitted by this cell.

Polars — load OHLCV from JSON landing zone with pl.DataFrame()

Load OHLCV from Landing

Reads a symbol JSON landing file into a Polars DataFrame. Casts dates.

Reference snippet for Polars — load OHLCV from JSON landing zone with pl.DataFrame().

 
def load_ohlcv_from_landing(symbol: str) -> pl.DataFrame:
    """Read OHLCV data from JSON landing zone into Polars DataFrame."""
    safe_symbol = symbol.replace(".", "_")
    landing_path = LANDING_DIR / f"ohlcv_{safe_symbol}.json"
    if not landing_path.exists():
        return pl.DataFrame()
 
    records = json.loads(landing_path.read_text(encoding="utf-8"))
    if not records:
        return pl.DataFrame()
 
    df = pl.DataFrame(records)
    df = df.with_columns(pl.col("date").str.to_date("%Y-%m-%d"))
    return df
 
No direct stdout was emitted by this cell.

yfinance — test single symbol landing zone fetch with fetch_ohlcv_to_landing()

Verifies the landing zone pattern: fetch to JSON, then load back into a Polars DataFrame.

Reference snippet for yfinance — test single symbol landing zone fetch with fetch_ohlcv_to_landing().

 
test_path = fetch_ohlcv_to_landing("SAP.DE", "2024-06-01", "2024-06-30")
if test_path:
    print(f"Landed: {test_path} ({test_path.stat().st_size / 1024:.1f} KB)")
else:
    print("No data returned")
 
test_df = load_ohlcv_from_landing("SAP.DE")
len(test_df), test_df.columns  # rows, columns
test_df.head()
Notebook-rendered table output is preserved below.
symboldateopenhighlowcloseadj_closevolumedividendsstock_splits
SAP.DE2024-06-03 00:00:00169.740005169.820007166.960007168.259995166.75280815317280.0000000.000000
SAP.DE2024-06-04 00:00:00168.520004170.440002167.660004168.600006167.08976715920710.0000000.000000
SAP.DE2024-06-05 00:00:00170.000000171.820007169.080002171.520004169.98361213529160.0000000.000000
SAP.DE2024-06-06 00:00:00176.020004180.240005176.000000177.720001176.12806720895490.0000000.000000
SAP.DE2024-06-07 00:00:00177.500000178.259995175.699997177.360001175.77130112248630.0000000.000000

Validation

Row-level Pydantic validation at the Bronze boundary; rejected rows are routed to the quarantine table.

Pydantic — validate Bronze rows with BaseModel() row-level check

Bronze Row-Level Validation

Valid rows collected; rejected rows go to quarantine with error details.

Reference snippet for Pydantic — validate Bronze rows with BaseModel() row-level check.

 
def validate_bronze(df: pl.DataFrame, batch_id: str = "") -> tuple[pl.DataFrame, int]:
    """Validate each row through RawOHLCV. Quarantines rejected rows."""
    valid_rows = []
    rejected = 0
 
    for row in df.iter_rows(named=True):
        try:
            record = RawOHLCV(
                symbol=str(row["symbol"]),
                date=row["date"],
                open=float(row["open"]),
                high=float(row["high"]),
                low=float(row["low"]),
                close=float(row["close"]),
                adj_close=float(row["adj_close"]),
                volume=int(row["volume"]),
                dividends=float(row["dividends"]),
                stock_splits=float(row["stock_splits"]),
            )
            valid_rows.append(record.model_dump())
        except Exception as e:
            rejected += 1
            if batch_id:
                quarantine_row(batch_id, "bronze", dict(row), str(e))
 
    if not valid_rows:
        return pl.DataFrame(), rejected
 
    return pl.DataFrame(valid_rows), rejected
 
log.info("validate_bronze() defined \u2014 rejects go to quarantine")
Notebook stdout is preserved below.
23:19:37 | INFO  | validate_bronze() defined — rejects go to quarantine

Pydantic — test Bronze validation on sample data

All rows should pass since yfinance data is generally clean.

Reference snippet for Pydantic — test Bronze validation on sample data.

 
valid_df, rejected = validate_bronze(test_df, batch_id="test")
len(valid_df), rejected  # valid rows, rejected rows
valid_df.head(5)
Notebook-rendered table output is preserved below.
symboldateopenhighlowcloseadj_closevolumedividendsstock_splits
SAP.DE2024-06-03 00:00:00169.740005169.820007166.960007168.259995166.75280815317280.0000000.000000
SAP.DE2024-06-04 00:00:00168.520004170.440002167.660004168.600006167.08976715920710.0000000.000000
SAP.DE2024-06-05 00:00:00170.000000171.820007169.080002171.520004169.98361213529160.0000000.000000
SAP.DE2024-06-06 00:00:00176.020004180.240005176.000000177.720001176.12806720895490.0000000.000000
SAP.DE2024-06-07 00:00:00177.500000178.259995175.699997177.360001175.77130112248630.0000000.000000

Ingestion pipeline

End-to-end Bronze orchestration: incremental fetch, Pydantic validation, MERGE upsert, and quality gate.

Bronze — define incremental ingestion pipeline with landing zone + MERGE INTO

Bronze Ingestion Pipeline

Three-step: land, validate, MERGE upsert. Fetches only new data since last known date per symbol.

Reference snippet for Bronze — define incremental ingestion pipeline with landing zone + MERGE INTO.

 
def ingest_bronze(symbols: list[str], start: str, end: str, batch_id: str) -> tuple[pl.DataFrame, StageLineage]:
    """Incrementally fetch to landing zone, validate, and MERGE upsert."""
    stage_ctx = start_stage(batch_id, "bronze", input_rows=0)
    all_frames = []
    total_rejected = 0
    total_fetched = 0
 
    for symbol in symbols:
        # Check last known date in SQL Server
        cur.execute("SELECT MAX(date) FROM bronze_ohlcv WHERE symbol = ?", symbol)
        row = cur.fetchone()
        last_date = row[0] if row and row[0] else None
 
        # Determine fetch range
        if last_date:
            fetch_start = (last_date + timedelta(days=1)).isoformat()
            if fetch_start >= end:
                log.info(f"  {symbol}: up to date (last: {last_date})")
                continue
        else:
            fetch_start = start
 
        # Step 1: Fetch from yfinance → JSON landing zone
        landing_path = fetch_ohlcv_to_landing(symbol, fetch_start, end)
        if landing_path is None:
            log.info(f"  {symbol}: no new data from {fetch_start}")
            continue
 
        # Step 2: Load from landing zone
        raw_df = load_ohlcv_from_landing(symbol)
        total_fetched += len(raw_df)
 
        # Step 3: Validate through Pydantic
        valid_df, rejected = validate_bronze(raw_df, batch_id)
        total_rejected += rejected
 
        if len(valid_df) > 0:
            all_frames.append(valid_df)
            # MERGE upsert into SQL Server
            merged = merge_bronze(valid_df, batch_id)
            size_kb = landing_path.stat().st_size / 1024
            log.info(f"  {symbol}: {len(raw_df)} landed ({size_kb:.1f} KB), "
                  f"{len(valid_df)} valid, {rejected} rejected, {merged} merged")
 
    # Combine new data for lineage hash
    new_df = pl.concat(all_frames) if all_frames else pl.DataFrame()
    if len(new_df) > 0:
        new_df = new_df.with_columns(pl.lit(batch_id).alias("batch_id"))
 
    stage_ctx["input_rows"] = total_fetched
    lineage = end_stage(stage_ctx, new_df if len(new_df) > 0 else pl.DataFrame({"_": []}), total_rejected)
    persist_lineage(lineage)
 
    # Return FULL bronze dataset for downstream stages
    bronze_full = pl.read_database(
        "SELECT symbol, date, [open] as [open], high, low, [close] as [close], "
        "adj_close, volume, dividends, stock_splits, batch_id "
        "FROM bronze_ohlcv ORDER BY symbol, date",
        connection=sql_engine
    )
 
    return bronze_full, lineage
 
Notebook stdout is preserved below.
ingest_bronze() defined — landing zone + incremental MERGE

Bronze — execute incremental ingestion for all symbols

Bronze Execution with Context

Creates BusinessContext, TemporalContext, StageContext, then runs Bronze ingestion.

Reference snippet for Bronze — execute incremental ingestion for all symbols.

 
batch_id = generate_batch_id()
 
biz_ctx = BusinessContext(trigger="scheduled", business_date=Date.today() - timedelta(days=1))
temp_ctx = TemporalContext(
    as_of_date=Date.today() - timedelta(days=1),
    reporting_period_start=Date.fromisoformat(START_DATE),
    reporting_period_end=Date.fromisoformat(END_DATE), timezone="CET",
)
bronze_stage_ctx = StageContext(
    batch_id=batch_id, stage="bronze",
    column_context=BRONZE_COLUMNS, business_context=biz_ctx, temporal_context=temp_ctx,
)
 
log.info(f"Pipeline batch_id: {batch_id[:8]}...")
log.info(f"Range: {START_DATE} \u2192 {END_DATE}")
 
t0 = time.time()
bronze_df, bronze_lineage = ingest_bronze(SYMBOLS, START_DATE, END_DATE, batch_id)
elapsed = (time.time() - t0) * 1000
 
# Detect zero-volume rows and classify using trading calendar
zero_vol = bronze_df.filter(pl.col("volume") == 0)
if len(zero_vol) > 0:
    for row in zero_vol.select("symbol", "date").unique().iter_rows(named=True):
        cal = pl.read_database(
            f"SELECT is_trading_day FROM dim_calendar "
            f"WHERE date = '{row['date']}' AND exchange_code = 'XETR'",
            connection=sql_engine
        )
        if len(cal) > 0 and not bool(cal["is_trading_day"][0]):
            bronze_stage_ctx.add_warning(
                f"{row['symbol']}: zero volume on {row['date']} \u2014 non-trading day (calendar)"
            )
        elif len(cal) > 0 and bool(cal["is_trading_day"][0]):
            bronze_stage_ctx.add_warning(
                f"{row['symbol']}: zero volume on {row['date']} \u2014 TRADING DAY (anomaly)"
            )
 
persist_context(bronze_stage_ctx)
silver_stage_ctx = bronze_stage_ctx.for_next_stage("silver", bronze_lineage, SILVER_COLUMNS)
log.info(f"Bronze complete: {len(bronze_df)} rows in {elapsed:.0f}ms")
Notebook stdout is preserved below.
23:19:37 | INFO  | Pipeline batch_id: 05a35d97...
23:19:37 | INFO  | Range: 2024-03-29 → 2026-03-29
23:19:37 | INFO  |   SAP.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
23:19:37 | INFO  |   SIE.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
23:19:37 | INFO  |   ALV.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
23:19:37 | INFO  |   DTE.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
23:19:37 | INFO  |   BAS.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
23:19:37 | INFO  | Bronze complete: 2530 rows in 85ms

Polars — display Bronze sample data with head()

Shows the first rows of ingested data to verify schema and values.

Reference snippet for Polars — display Bronze sample data with head().

 
bronze_df.head(5)
Notebook-rendered table output is preserved below.
symboldateopenhighlowcloseadj_closevolumedividendsstock_splitsbatch_id
ALV.DE2024-03-28 00:00:00277.000000278.100006276.450012277.799988252.8253949191730.0000000.0000009c135c08-937d-4413-8bb4-1b66407ed9a5
ALV.DE2024-04-02 00:00:00278.200012280.000000272.200012273.899994249.27600110131760.0000000.0000009c135c08-937d-4413-8bb4-1b66407ed9a5
ALV.DE2024-04-03 00:00:00274.500000276.600006273.899994274.399994249.7310647821020.0000000.0000009c135c08-937d-4413-8bb4-1b66407ed9a5
ALV.DE2024-04-04 00:00:00274.100006275.200012272.200012272.399994247.9108736905510.0000000.0000009c135c08-937d-4413-8bb4-1b66407ed9a5
ALV.DE2024-04-05 00:00:00270.000000270.200012267.100006268.799988244.6344919308740.0000000.0000009c135c08-937d-4413-8bb4-1b66407ed9a5

Polars — display Bronze row counts per symbol with group_by().agg()

Verifies all symbols were ingested with reasonable row counts and correct date ranges.

Reference snippet for Polars — display Bronze row counts per symbol with group_by().agg().

 
bronze_df.group_by("symbol").agg(
    pl.col("date").count().alias("rows"),
    pl.col("date").min().alias("first_date"),
    pl.col("date").max().alias("last_date"),
).sort("symbol")
Notebook-rendered table output is preserved below.
symbolrowsfirst_datelast_date
ALV.DE5062024-03-28 00:00:002026-03-27 00:00:00
BAS.DE5062024-03-28 00:00:002026-03-27 00:00:00
DTE.DE5062024-03-28 00:00:002026-03-27 00:00:00
SAP.DE5062024-03-28 00:00:002026-03-27 00:00:00
SIE.DE5062024-03-28 00:00:002026-03-27 00:00:00

Pipeline — run Bronze data quality gate with run_quality_gate()

Bronze Quality Gate

All checks must pass before Silver processing begins.

Reference snippet for Pipeline — run Bronze data quality gate with run_quality_gate().

 
bronze_dq = run_quality_gate([
    dq_check_not_empty(bronze_df, "bronze"),
    dq_check_no_null_keys(bronze_df, ["symbol", "date"], "bronze"),
    dq_check_no_duplicates(bronze_df, ["symbol", "date"], "bronze"),
    dq_check_range(bronze_df, "close", 0.01, 100_000, "bronze"),
    dq_check_range(bronze_df, "volume", 0, 10_000_000_000, "bronze"),
    dq_check_freshness(bronze_df, "date", 5, "bronze"),
    dq_check_row_count(bronze_df, len(SYMBOLS) * 200, "bronze"),
], stage="bronze")
 
bronze_dq
Notebook stdout is preserved below.
23:19:38 | INFO  |   DQ PASS: bronze: 2530 rows
23:19:38 | INFO  |   DQ PASS: bronze: no null keys in ['symbol', 'date']
23:19:38 | INFO  |   DQ PASS: bronze: no duplicates
23:19:38 | INFO  |   DQ PASS: bronze: 'close' within range
23:19:38 | INFO  |   DQ PASS: bronze: 'volume' within range
23:19:38 | INFO  |   DQ PASS: bronze: latest date 2026-03-27 (2d ago)
23:19:38 | INFO  |   DQ PASS: bronze: 2530 rows
checkstatus
bronze: 2530 rowsPASS
bronze: no null keys in ['symbol', 'date']PASS
bronze: no duplicatesPASS
bronze: 'close' within rangePASS
bronze: 'volume' within rangePASS
bronze: latest date 2026-03-27 (2d ago)PASS
bronze: 2530 rowsPASS

7. Silver Layer — Cleaning & Enrichment

Silver is where the Functional Core principle (Gary Bernhardt, ‘Boundaries’ 2012) is most visible. The three transforms (daily_return, intraday_range, sma_20) are pure functions — DataFrame in, DataFrame out, no database calls, no file I/O, no side effects. Pure functions are trivially testable (pass a 10-row hardcoded DataFrame, assert the output), trivially debuggable (the bug is in the formula, not in a network timeout), and trivially parallelizable (no shared state). The imperative shell (MERGE upsert, lineage persistence, context propagation) wraps AROUND the pure transforms, never inside them.

Without Pure Transforms

A transform function that reads from SQL Server mid-computation becomes untestable without a live database. A transform that writes intermediate results to a file fails unpredictably under disk pressure. Keeping transforms pure means the only thing that can go wrong is the formula — and formulas can be verified with a unit test in milliseconds.

Fix: Pure Functions — Input DataFrame, Output DataFrame

Write every silver transform as a function that accepts a Polars DataFrame and returns a new DataFrame — no database calls, no file I/O, no global state. The caller (imperative shell) handles reading from SQL Server and writing back. This pattern makes every transform unit-testable with pl.DataFrame(...) literals in under a second, with no mocking required.

Transform functions

Pure Polars functions computing daily return, intraday range, and 20-day SMA — no side effects.

Polars — compute daily returns with pct_change().over()

Transform: Daily Returns

Close-to-close percentage change per symbol. Pure function: DataFrame in, DataFrame out.

Reference snippet for Polars — compute daily returns with pct_change().over().

 
def compute_daily_returns(df: pl.DataFrame) -> pl.DataFrame:
    """Add daily_return column: close-to-close percentage change per symbol."""
    return df.sort(["symbol", "date"]).with_columns(
        pl.col("close")
          .pct_change()
          .over("symbol")
          .fill_null(0.0)
          .round(6)
          .alias("daily_return")
    )
 
# Test on Bronze data
test_returns = compute_daily_returns(bronze_df.drop("batch_id"))
test_returns.filter(pl.col("symbol") == "SAP.DE").select("symbol", "date", "close", "daily_return").head(5)
Notebook-rendered table output is preserved below.
symboldateclosedaily_return
SAP.DE2024-03-28 00:00:00180.4600070.000000
SAP.DE2024-04-02 00:00:00177.059998-0.018841
SAP.DE2024-04-03 00:00:00178.2200010.006551
SAP.DE2024-04-04 00:00:00178.020004-0.001122
SAP.DE2024-04-05 00:00:00177.419998-0.003370

Polars — compute intraday range with with_columns()

Transform: Intraday Range

(high - low) / close: normalized daily price spread. Higher = more volatile.

Reference snippet for Polars — compute intraday range with with_columns().

 
def compute_intraday_range(df: pl.DataFrame) -> pl.DataFrame:
    """Add intraday_range column: (high - low) / close as percentage."""
    return df.with_columns(
        ((pl.col("high") - pl.col("low")) / pl.col("close"))
        .round(6)
        .alias("intraday_range")
    )
 
# Test on returns data
test_range = compute_intraday_range(test_returns)
test_range.filter(pl.col("symbol") == "SAP.DE").select("symbol", "date", "high", "low", "close", "intraday_range").head(5)
Notebook-rendered table output is preserved below.
symboldatehighlowcloseintraday_range
SAP.DE2024-03-28 00:00:00181.860001179.100006180.4600070.015294
SAP.DE2024-04-02 00:00:00181.919998177.059998177.0599980.027448
SAP.DE2024-04-03 00:00:00179.520004176.559998178.2200010.016609
SAP.DE2024-04-04 00:00:00178.460007176.339996178.0200040.011909
SAP.DE2024-04-05 00:00:00177.960007174.779999177.4199980.017924

Polars — compute 20-day moving average with rolling_mean().over()

Transform: 20-Day SMA

Rolling mean of close over 20-day window per symbol. First 19 rows are NULL.

Reference snippet for Polars — compute 20-day moving average with rolling_mean().over().

 
def compute_sma(df: pl.DataFrame, window: int = 20) -> pl.DataFrame:
    """Add sma_20 column: rolling mean of close price per symbol."""
    return df.sort(["symbol", "date"]).with_columns(
        pl.col("close")
          .rolling_mean(window_size=window)
          .over("symbol")
          .round(4)
          .alias(f"sma_{window}")
    )
 
# Test on range data
test_sma = compute_sma(test_range)
test_sma.filter(pl.col("symbol") == "SAP.DE").select("symbol", "date", "close", "sma_20").tail(5)
Notebook-rendered table output is preserved below.
symboldateclosesma_20
SAP.DE2026-03-23 00:00:00153.860001166.034000
SAP.DE2026-03-24 00:00:00147.619995165.123000
SAP.DE2026-03-25 00:00:00146.899994164.129000
SAP.DE2026-03-26 00:00:00144.639999162.750000
SAP.DE2026-03-27 00:00:00142.559998161.330000

Polars — compose all Silver transforms with function chaining

Transform Composition Pipeline

Chains three pure functions. Each independent and unit-testable. Validates before MERGE.

Reference snippet for Polars — compose all Silver transforms with function chaining.

 
def transform_silver(bronze_df: pl.DataFrame) -> pl.DataFrame:
    """Apply all Silver enrichment transforms in sequence."""
    # Drop batch_id from Bronze — Silver adds its own
    df = bronze_df.drop("batch_id") if "batch_id" in bronze_df.columns else bronze_df
    df = compute_daily_returns(df)
    df = compute_intraday_range(df)
    df = compute_sma(df, window=20)
    return df
 
Notebook stdout is preserved below.
transform_silver() defined — composes all Silver transforms

Pydantic — validate Silver rows with BaseModel() row-level check

Silver Row-Level Validation

Valid rows collected; rejected rows quarantined with error details.

Reference snippet for Pydantic — validate Silver rows with BaseModel() row-level check.

 
def validate_silver(df: pl.DataFrame, batch_id: str) -> tuple[pl.DataFrame, int]:
    """Validate each row through CleanOHLCV. Quarantines rejected rows."""
    valid_rows = []
    rejected = 0
 
    for row in df.iter_rows(named=True):
        try:
            record = CleanOHLCV(
                symbol=str(row["symbol"]),
                date=row["date"],
                open=float(row["open"]),
                high=float(row["high"]),
                low=float(row["low"]),
                close=float(row["close"]),
                adj_close=float(row["adj_close"]),
                volume=int(row["volume"]),
                dividends=float(row["dividends"]),
                stock_splits=float(row["stock_splits"]),
                daily_return=float(row["daily_return"]),
                intraday_range=float(row["intraday_range"]),
                sma_20=float(row["sma_20"]) if row["sma_20"] is not None else None,
                batch_id=batch_id,
            )
            valid_rows.append(record.model_dump())
        except Exception as e:
            rejected += 1
            quarantine_row(batch_id, "silver", dict(row), str(e))
 
    if not valid_rows:
        return pl.DataFrame(), rejected
 
    return pl.DataFrame(valid_rows), rejected
 
log.info("validate_silver() defined \u2014 rejects go to quarantine")
Notebook stdout is preserved below.
23:19:38 | INFO  | validate_silver() defined — rejects go to quarantine

Enrichment pipeline

Imperative shell: applies transforms, validates through CleanOHLCV, MERGE upserts to SQL Server, and runs the quality gate.

Silver — define enrichment pipeline with transform + MERGE INTO

Silver Enrichment Pipeline

Transform, validate (Pydantic), MERGE (SQL), lineage. Transforms FULL bronze for correct SMA.

Reference snippet for Silver — define enrichment pipeline with transform + MERGE INTO.

 
def process_silver(bronze_df: pl.DataFrame, batch_id: str) -> tuple[pl.DataFrame, StageLineage]:
    """Transform, validate, and MERGE upsert Silver data."""
    stage_ctx = start_stage(batch_id, "silver", input_rows=len(bronze_df))
 
    # Apply all transforms on full bronze (SMA/returns need full history)
    enriched_df = transform_silver(bronze_df)
 
    # Validate through Pydantic
    valid_df, rejected = validate_silver(enriched_df, batch_id)
 
    # MERGE upsert into SQL Server
    if len(valid_df) > 0:
        merged = merge_silver(valid_df, batch_id)
        log.info(f"Silver: {merged} rows merged ({len(valid_df)} valid, {rejected} rejected)")
 
    # Complete lineage
    lineage = end_stage(stage_ctx, valid_df, rejected)
    persist_lineage(lineage)
 
    # Return FULL silver dataset for Gold layer
    silver_full = pl.read_database(
        "SELECT symbol, date, [open] as [open], high, low, [close] as [close], "
        "adj_close, volume, dividends, stock_splits, "
        "daily_return, intraday_range, sma_20, batch_id "
        "FROM silver_ohlcv ORDER BY symbol, date",
        connection=sql_engine
    )
 
    return silver_full, lineage
 
Notebook stdout is preserved below.
process_silver() defined — MERGE upsert, returns full dataset

Silver — execute enrichment on full Bronze data

Silver Execution with Context

Runs Silver enrichment, records SMA-20 null warnings in StageContext.

Reference snippet for Silver — execute enrichment on full Bronze data.

 
t0 = time.time()
silver_df, silver_lineage = process_silver(bronze_df, batch_id)
elapsed = (time.time() - t0) * 1000
 
sma_null_count = silver_df.filter(pl.col("sma_20").is_null()).height
if sma_null_count > 0:
    silver_stage_ctx.add_warning(f"sma_20: {sma_null_count} NULL values (first 19 rows per symbol)")
 
persist_context(silver_stage_ctx)
gold_stage_ctx = silver_stage_ctx.for_next_stage("gold", silver_lineage, GOLD_DAILY_COLUMNS)
log.info(f"Silver complete: {len(silver_df)} rows in {elapsed:.0f}ms")
Notebook stdout is preserved below.
23:19:41 | INFO  | Silver: 2530 rows merged (2530 valid, 0 rejected)
23:19:41 | WARNING |   silver: sma_20: 95 NULL values (first 19 rows per symbol)
23:19:41 | INFO  | Silver complete: 2530 rows in 3098ms

Polars — display Silver enriched columns with filter().select()

Verifies that daily_return, intraday_range, and sma_20 are populated after Silver enrichment.

Reference snippet for Polars — display Silver enriched columns with filter().select().

 
silver_df.filter(pl.col("symbol") == "SAP.DE").select(
    "symbol", "date", "close", "daily_return", "intraday_range", "sma_20"
).tail(5)
Notebook-rendered table output is preserved below.
symboldateclosedaily_returnintraday_rangesma_20
SAP.DE2026-03-23 00:00:00153.8600010.0002600.072274166.034000
SAP.DE2026-03-24 00:00:00147.619995-0.0405560.034142165.123000
SAP.DE2026-03-25 00:00:00146.899994-0.0048770.035262164.129000
SAP.DE2026-03-26 00:00:00144.639999-0.0153850.031527162.750000
SAP.DE2026-03-27 00:00:00142.559998-0.0143810.036616161.330000

Polars — display Silver statistics per symbol with group_by().agg()

Summary statistics to verify enrichment quality across all symbols.

Reference snippet for Polars — display Silver statistics per symbol with group_by().agg().

 
silver_df.group_by("symbol").agg(
    pl.col("daily_return").mean().round(6).alias("avg_return"),
    pl.col("daily_return").std().round(6).alias("volatility"),
    pl.col("intraday_range").mean().round(6).alias("avg_intraday"),
    pl.col("sma_20").null_count().alias("sma_nulls"),
    pl.col("date").count().alias("rows"),
).sort("symbol")
Notebook-rendered table output is preserved below.
symbolavg_returnvolatilityavg_intradaysma_nullsrows
ALV.DE0.0005320.0118510.01410619506
BAS.DE0.0001210.0175080.02139819506
DTE.DE0.0007650.0132630.01572019506
SAP.DE-0.0002850.0189190.02067219506
SIE.DE0.0004750.0192230.02150019506

Pipeline — run Silver data quality gate with run_quality_gate()

Silver Quality Gate

Hard gate: blocks on structural issues. Soft gate: logs warnings on statistical anomalies.

Reference snippet for Pipeline — run Silver data quality gate with run_quality_gate().

silver_dq = run_quality_gate([
    dq_check_not_empty(silver_df, "silver"),
    dq_check_no_null_keys(silver_df, ["symbol", "date", "daily_return"], "silver"),
    dq_check_no_duplicates(silver_df, ["symbol", "date"], "silver"),
    dq_check_range(silver_df, "daily_return", -0.5, 0.5, "silver"),
    dq_check_range(silver_df, "intraday_range", 0, 0.5, "silver"),
    dq_check_freshness(silver_df, "date", 5, "silver"),
    dq_check_row_count(silver_df, len(SYMBOLS) * 200, "silver"),
], stage="silver")
 
# Soft checks — warn but don't block (daily return > 10% is unusual for blue chips)
log.info("Outlier checks (warnings only):")
outliers = silver_df.filter(pl.col("daily_return").abs() > 0.10)
if len(outliers) > 0:
    log.warning(f"  {len(outliers)} rows with |daily_return| > 10%:")
    display(outliers.select("symbol", "date", "close", "daily_return", "volume").sort("daily_return"))
else:
    log.info("  No outliers detected")
 
silver_dq
Notebook stdout is preserved below.
23:19:41 | INFO  |   DQ PASS: silver: 2530 rows
23:19:41 | INFO  |   DQ PASS: silver: no null keys in ['symbol', 'date', 'daily_return']
23:19:41 | INFO  |   DQ PASS: silver: no duplicates
23:19:41 | INFO  |   DQ PASS: silver: 'daily_return' within range
23:19:41 | INFO  |   DQ PASS: silver: 'intraday_range' within range
23:19:41 | INFO  |   DQ PASS: silver: latest date 2026-03-27 (2d ago)
23:19:41 | INFO  |   DQ PASS: silver: 2530 rows
23:19:41 | INFO  | Outlier checks (warnings only):
23:19:41 | WARNING |   3 rows with |daily_return| > 10%:
symboldateclosedaily_returnvolume
SAP.DE2026-01-29 00:00:00164.619995-0.16070215846791
SAP.DE2025-04-23 00:00:00241.6999970.1061783410054
BAS.DE2025-03-05 00:00:0053.6600000.1070779216640
checkstatus
silver: 2530 rowsPASS
silver: no null keys in ['symbol', 'date', 'daily_return']PASS
silver: no duplicatesPASS
silver: 'daily_return' within rangePASS
silver: 'intraday_range' within rangePASS
silver: latest date 2026-03-27 (2d ago)PASS
silver: 2530 rowsPASS

8. Gold Layer — Aggregations & Mart Tables

Gold produces consumption-ready data products from Silver. Two aggregations, both pure functions: DailySummary (cross-sectional: all symbols for each date) and SymbolProfile (longitudinal: full history for each symbol). Gold is always a full rebuild — truncate and recompute from Silver on every run. This is simpler than incremental and guarantees consistency. Acceptable because Gold tables are small (~50 symbols x 1 row + ~500 daily rows). Both are validated through the typed contracts before persistence.

Without Gold Validation

An aggregation bug produces max_drawdown = 0.15 (positive). This is mathematically impossible — drawdown is always negative. Without the le=0 constraint, the bad value reaches the dashboard. A portfolio manager sees “positive drawdown” and makes decisions on nonsensical data.

Fix: Gold Quality Gate with Invariant Assertions

Define a Pydantic GoldRecord model with mathematically-constrained fields: max_drawdown: float = Field(le=0), volatility: float = Field(ge=0), etc. Run every aggregated row through the model before writing to the gold table. Rows that violate an invariant are rejected to quarantine with the constraint name and the actual value — never silently written.

Aggregation functions

Pure Polars functions building DailySummary (cross-sectional) and SymbolProfile (longitudinal) from Silver data.

Polars — build daily cross-sectional summary with group_by().agg()

Aggregation: Daily Summary

Groups by date: mean/max/min return, total volume, avg intraday range.

Reference snippet for Polars — build daily cross-sectional summary with group_by().agg().

 
def build_daily_summary(silver_df: pl.DataFrame, batch_id: str) -> pl.DataFrame:
    """Aggregate Silver data into daily cross-sectional summary."""
    summary = silver_df.group_by("date").agg(
        pl.col("symbol").n_unique().alias("symbols_traded"),
        pl.col("daily_return").mean().round(6).alias("avg_return"),
        pl.col("daily_return").max().alias("max_return"),
        pl.col("daily_return").min().alias("min_return"),
        pl.col("volume").sum().alias("total_volume"),
        pl.col("intraday_range").mean().round(6).alias("avg_intraday_pct"),
    ).sort("date").with_columns(
        pl.lit(batch_id).alias("batch_id")
    )
    return summary
 
daily_summary_df = build_daily_summary(silver_df, batch_id)
len(daily_summary_df)  # trading days
daily_summary_df.head(5)
Notebook-rendered table output is preserved below.
datesymbols_tradedavg_returnmax_returnmin_returntotal_volumeavg_intraday_pctbatch_id
2024-03-28 00:00:0050.0000000.0000000.000000141152410.01124605a35d97-97ef-43f8-a752-d45524767f62
2024-04-02 00:00:005-0.0062610.016815-0.018841154600600.02085005a35d97-97ef-43f8-a752-d45524767f62
2024-04-03 00:00:0050.0048620.012820-0.002239123444750.01461705a35d97-97ef-43f8-a752-d45524767f62
2024-04-04 00:00:005-0.0006310.007522-0.007289102387480.01083005a35d97-97ef-43f8-a752-d45524767f62
2024-04-05 00:00:005-0.014092-0.003370-0.021460163640360.01779105a35d97-97ef-43f8-a752-d45524767f62

Polars — build per-symbol profile with cum_max() drawdown

Aggregation: Symbol Risk Profile

Per-symbol: avg return, volatility, max drawdown, total dividends.

Reference snippet for Polars — build per-symbol profile with cum_max() drawdown.

 
def build_symbol_profile(silver_df: pl.DataFrame, batch_id: str) -> pl.DataFrame:
    """Aggregate Silver data into per-symbol summary statistics."""
    profiles = []
 
    for symbol in silver_df.select("symbol").unique().sort("symbol").to_series():
        sym_df = silver_df.filter(pl.col("symbol") == symbol).sort("date")
 
        # Max drawdown: peak-to-trough decline using cumulative max of close
        cum_max = sym_df.select(pl.col("close").cum_max().alias("peak"))
        drawdowns = (sym_df["close"] - cum_max["peak"]) / cum_max["peak"]
        max_dd = round(drawdowns.to_list()[-1] if len(drawdowns) == 0 else min(drawdowns.to_list()), 6)
 
        # Extract values as plain Python types to satisfy type checker
        returns = sym_df["daily_return"].to_list()
        volumes = sym_df["volume"].to_list()
        dividends = sym_df["dividends"].to_list()
        dates = sym_df["date"].to_list()
 
        profiles.append({
            "symbol": symbol,
            "total_trading_days": len(sym_df),
            "avg_daily_return": round(sum(returns) / len(returns), 6),
            "volatility": round((sum((r - sum(returns)/len(returns))**2 for r in returns) / len(returns)) ** 0.5, 6),
            "max_drawdown": round(max_dd, 6),
            "avg_volume": round(sum(volumes) / len(volumes), 2),
            "total_dividends": round(sum(dividends), 4),
            "first_date": dates[0],
            "last_date": dates[-1],
            "batch_id": batch_id,
        })
 
    return pl.DataFrame(profiles)
 
symbol_profile_df = build_symbol_profile(silver_df, batch_id)
len(symbol_profile_df)  # symbol profiles
symbol_profile_df
Notebook-rendered table output is preserved below.
symboltotal_trading_daysavg_daily_returnvolatilitymax_drawdownavg_volumetotal_dividendsfirst_datelast_datebatch_id
ALV.DE5060.0005320.011839-0.123504631486.14000029.2000002024-03-28 00:00:002026-03-27 00:00:0005a35d97-97ef-43f8-a752-d45524767f62
BAS.DE5060.0001210.017491-0.2767662518782.1400005.6500002024-03-28 00:00:002026-03-27 00:00:0005a35d97-97ef-43f8-a752-d45524767f62
DTE.DE5060.0007650.013250-0.2661096531019.0600001.6700002024-03-28 00:00:002026-03-27 00:00:0005a35d97-97ef-43f8-a752-d45524767f62
SAP.DE506-0.0002850.018900-0.4914021676455.1900004.5500002024-03-28 00:00:002026-03-27 00:00:0005a35d97-97ef-43f8-a752-d45524767f62
SIE.DE5060.0004750.019204-0.2732511155097.07000010.5500002024-03-28 00:00:002026-03-27 00:00:0005a35d97-97ef-43f8-a752-d45524767f62

Pydantic — validate Gold daily summary with BaseModel() row-level check

Validates each daily summary row through the DailySummary model to catch aggregation errors before persistence.

Reference snippet for Pydantic — validate Gold daily summary with BaseModel() row-level check.

 
def validate_gold_daily(df: pl.DataFrame) -> tuple[pl.DataFrame, int]:
    """Validate daily summary rows through DailySummary model."""
    valid_rows = []
    rejected = 0
    for row in df.iter_rows(named=True):
        try:
            record = DailySummary(**row)
            valid_rows.append(record.model_dump())
        except Exception:
            rejected += 1
    return (pl.DataFrame(valid_rows) if valid_rows else pl.DataFrame()), rejected
 
valid_daily, rej_daily = validate_gold_daily(daily_summary_df)
len(valid_daily), rej_daily  # daily summary validation: valid, rejected
No direct stdout was emitted by this cell.

Pydantic — validate Gold symbol profiles with BaseModel() row-level check

Validates each symbol profile through the SymbolProfile model to catch calculation errors before persistence.

Reference snippet for Pydantic — validate Gold symbol profiles with BaseModel() row-level check.

 
def validate_gold_profiles(df: pl.DataFrame) -> tuple[pl.DataFrame, int]:
    """Validate symbol profile rows through SymbolProfile model."""
    valid_rows = []
    rejected = 0
    for row in df.iter_rows(named=True):
        try:
            record = SymbolProfile(**row)
            valid_rows.append(record.model_dump())
        except Exception as e:
            rejected += 1
            print(f"  Rejected {row.get('symbol', '?')}: {e}")
    return (pl.DataFrame(valid_rows) if valid_rows else pl.DataFrame()), rejected
 
valid_profiles, rej_profiles = validate_gold_profiles(symbol_profile_df)
len(valid_profiles), rej_profiles  # symbol profile validation: valid, rejected
No direct stdout was emitted by this cell.

Persistence

Validate Gold rows through typed contracts, then truncate and reload both mart tables.

SQL Server — define Gold persistence function with TRUNCATE + to_sql()

Gold: Truncate and Rebuild

Full rebuild from Silver. TRUNCATE both Gold tables, INSERT new aggregations.

Reference snippet for SQL Server — define Gold persistence function with TRUNCATE + to_sql().

 
def persist_gold(daily_df: pl.DataFrame, profile_df: pl.DataFrame, batch_id: str) -> StageLineage:
    """Persist Gold mart tables to SQL Server with lineage tracking."""
    total_input = len(daily_df) + len(profile_df)
    stage_ctx = start_stage(batch_id, "gold", input_rows=total_input)
 
    # Truncate and insert
    if len(daily_df) > 0:
        write_to_sql(daily_df, "gold_daily_summary", truncate=True)
    if len(profile_df) > 0:
        write_to_sql(profile_df, "gold_symbol_profile", truncate=True)
 
    # Combined hash
    combined = pl.concat([
        daily_df.select(pl.all().cast(pl.Utf8)),
        profile_df.select(pl.all().cast(pl.Utf8)),
    ], how="diagonal")
 
    lineage = end_stage(stage_ctx, combined, 0)
    persist_lineage(lineage)
    return lineage
 
No direct stdout was emitted by this cell.

SQL Server — persist Gold marts with TRUNCATE + to_sql()

Gold Execution with Context

Persists Gold marts, checks for missing symbols, records warnings.

Reference snippet for SQL Server — persist Gold marts with TRUNCATE + to_sql().

 
gold_lineage = persist_gold(valid_daily, valid_profiles, batch_id)
 
missing_symbols = valid_daily.filter(pl.col("symbols_traded") < len(SYMBOLS))
if len(missing_symbols) > 0:
    for row in missing_symbols.head(5).iter_rows(named=True):
        gold_stage_ctx.add_warning(f"gold: {row['date']} only {row['symbols_traded']}/{len(SYMBOLS)} symbols")
 
persist_context(gold_stage_ctx)
log.info(f"Gold persisted: hash={gold_lineage.output_hash}")
Notebook stdout is preserved below.
23:19:41 | INFO  | Gold persisted: hash=57256313be661629

Polars — display Gold daily summary with sort().tail()

Shows the most recent trading days with cross-sectional metrics.

Reference snippet for Polars — display Gold daily summary with sort().tail().

 
valid_daily.sort("date").tail()
Notebook-rendered table output is preserved below.
datesymbols_tradedavg_returnmax_returnmin_returntotal_volumeavg_intraday_pctbatch_id
2026-03-23 00:00:0050.0120350.037055-0.002530219007460.07127605a35d97-97ef-43f8-a752-d45524767f62
2026-03-24 00:00:0050.0038540.041800-0.040556149909440.02866305a35d97-97ef-43f8-a752-d45524767f62
2026-03-25 00:00:0050.0080210.023951-0.004877135363250.01951205a35d97-97ef-43f8-a752-d45524767f62
2026-03-26 00:00:005-0.0060630.014394-0.015385157222940.01961005a35d97-97ef-43f8-a752-d45524767f62
2026-03-27 00:00:005-0.0037680.026803-0.023123162296340.02484805a35d97-97ef-43f8-a752-d45524767f62

Polars — display Gold symbol profiles with select()

Final per-symbol summary statistics across the full two-year history.

Reference snippet for Polars — display Gold symbol profiles with select().

 
valid_profiles.select(
    "symbol", "total_trading_days", "avg_daily_return",
    "volatility", "max_drawdown", "avg_volume", "total_dividends"
)
Notebook-rendered table output is preserved below.
symboltotal_trading_daysavg_daily_returnvolatilitymax_drawdownavg_volumetotal_dividends
ALV.DE5060.0005320.011839-0.123504631486.14000029.200000
BAS.DE5060.0001210.017491-0.2767662518782.1400005.650000
DTE.DE5060.0007650.013250-0.2661096531019.0600001.670000
SAP.DE506-0.0002850.018900-0.4914021676455.1900004.550000
SIE.DE5060.0004750.019204-0.2732511155097.07000010.550000

9. Parquet Export — Pre-Materialized Data Products

The serving layer reads Parquet files, not SQL Server. This is the pre-materialized views pattern — the pipeline produces finished data products as files, the API is a thin reader with zero database dependency at serving time. Deployment is a file copy, not a migration. Cache invalidation = re-run the pipeline. Data contracts (JSON Schema with column semantics) are exported alongside the Parquet files, making each data product self-describing.

Without Pre-Materialization

The API queries SQL Server on every request. A slow query blocks the response. A database restart takes the API down. With Parquet files, the API has no database dependency — it reads a file that the pipeline pre-computed. The API can serve data even if SQL Server is down.

Fix: Pipeline Writes Parquet, API Reads File

At the end of each pipeline run, export the gold aggregation to a versioned Parquet file (gold/daily_summary_<date>.parquet). The FastAPI endpoint reads the latest file at startup and caches it in memory — zero database queries at serve time. Cache invalidation is a file replacement: re-run the pipeline, the next request loads the new file.

Export operations

Write Gold and Silver data products to Parquet, record the export lineage stage, and verify file integrity.

Polars — export daily summary to Parquet with write_parquet()

Parquet: Pre-Materialized View

API reads this file directly. Parquet preserves types without CSV parsing overhead.

Reference snippet for Polars — export daily summary to Parquet with write_parquet().

 
daily_path = EXPORT_DIR / "gold_daily_summary.parquet"
valid_daily.write_parquet(daily_path)
size_kb = daily_path.stat().st_size / 1024
daily_path.name, size_kb, len(valid_daily)  # exported file, KB, rows
Notebook stdout is preserved below.
gold_daily_summary.parquet (21.2 KB, 506 rows)

Polars — export symbol profiles to Parquet with write_parquet()

Pre-materialized per-symbol summary for the comparison dashboard.

Reference snippet for Polars — export symbol profiles to Parquet with write_parquet().

 
profile_path = EXPORT_DIR / "gold_symbol_profile.parquet"
valid_profiles.write_parquet(profile_path)
size_kb = profile_path.stat().st_size / 1024
profile_path.name, size_kb, len(valid_profiles)  # exported file, KB, rows
Notebook stdout is preserved below.
gold_symbol_profile.parquet (3.9 KB, 5 rows)

Polars — export Silver data to Parquet with write_parquet()

Silver Parquet Export

Full Silver dataset exported for time-series and per-symbol drill-down endpoints.

Reference snippet for Polars — export Silver data to Parquet with write_parquet().

 
silver_path = EXPORT_DIR / "silver_ohlcv.parquet"
silver_df.write_parquet(silver_path)
size_kb = silver_path.stat().st_size / 1024
silver_path.name, size_kb, len(silver_df)  # exported file, KB, rows
Notebook stdout is preserved below.
silver_ohlcv.parquet (96.9 KB, 2530 rows)

Lineage — record export stage with end_stage()

Tracks which files were exported and their sizes as a stage lineage record.

Reference snippet for Lineage — record export stage with end_stage().

 
export_ctx = start_stage(batch_id, "export", input_rows=len(valid_daily) + len(valid_profiles) + len(silver_df))
 
# Combined export DataFrame for hash (all exported data)
export_combined = pl.concat([
    valid_daily.select(pl.all().cast(pl.Utf8)),
    valid_profiles.select(pl.all().cast(pl.Utf8)),
    silver_df.select(pl.all().cast(pl.Utf8)),
], how="diagonal")
 
export_lineage = end_stage(export_ctx, export_combined, 0)
persist_lineage(export_lineage)
 
export_lineage.output_rows, export_lineage.output_hash  # export lineage: total rows, hash
Notebook stdout is preserved below.
3041 total rows, hash=5daff78bb463b750

Polars — verify exported Parquet files with read_parquet()

Round-trip test: write, then read back to verify row counts match.

Reference snippet for Polars — verify exported Parquet files with read_parquet().

 
for name in ["gold_daily_summary", "gold_symbol_profile", "silver_ohlcv"]:
    path = EXPORT_DIR / f"{name}.parquet"
    df = pl.read_parquet(path)
    print(f"  {name}: {len(df)} rows, {len(df.columns)} cols")
Notebook stdout is preserved below.
506 rows (gold_daily_summary), 8 cols
  5 rows (gold_symbol_profile), 10 cols
  2530 rows (silver_ohlcv), 14 cols

Pydantic \u2014 export data contracts as JSON Schema with model_json_schema()

Exports machine-readable JSON Schema contracts for every pipeline boundary.

Reference snippet for Pydantic \u2014 export data contracts as JSON Schema with model_json_schema().

 
contract_paths = export_data_contracts(EXPORT_DIR)
for p in contract_paths:
    print(f"  {p.name}: {p.stat().st_size:,} bytes")
Notebook stdout is preserved below.
23:19:41 | INFO  |   Contract exported: bronze_ohlcv_contract.json
23:19:41 | INFO  |   Contract exported: silver_ohlcv_contract.json
23:19:41 | INFO  |   Contract exported: gold_daily_summary_contract.json
23:19:41 | INFO  |   Contract exported: gold_symbol_profile_contract.json

5,031 bytes
  6,711 bytes
  3,453 bytes
  4,606 bytes

10. Lineage Review — Pipeline Execution Audit

After all stages complete, the full execution trail is available for review across five artifacts: stage lineage (timing, row counts, hashes), RunContext JSON (execution envelope with business and temporal context), context log (warnings per stage in SQL Server), quarantine (every rejected row with its error), and data contracts (column-level semantics as JSON Schema). Together these answer any question about what the pipeline did, why it did it, what it knew, and what it produced.

Pipeline audit

Finalize and persist the RunContext, then query lineage, quarantine, and context log tables for the current batch.

Pydantic — build and save run context with RunContext()

Finalize RunContext

Combines stage lineage, business context, temporal context, warnings into final record.

Reference snippet for Pydantic — build and save run context with RunContext().

 
run_context = RunContext(
    batch_id=batch_id,
    started_at=bronze_lineage.started_at,
    completed_at=export_lineage.completed_at,
    symbols=SYMBOLS,
    date_range=(START_DATE, END_DATE),
    stages=[bronze_lineage, silver_lineage, gold_lineage, export_lineage],
    status="completed",
    business_context=biz_ctx,
    temporal_context=temp_ctx,
    data_warnings=gold_stage_ctx.data_warnings,
    contract_version="1.0",
)
 
ctx_path = save_run_context(run_context)
total_ms = sum(s.duration_ms for s in run_context.stages)
ctx_path.name  # RunContext saved
batch_id[:8], len(run_context.data_warnings)  # batch prefix, warnings
total_ms  # processing time (ms)
Notebook stdout is preserved below.
run_05a35d97.json
05a35d97... | Warnings: 1
3187ms

Polars — display lineage summary as DataFrame

Shows all stages with timing, row counts, and output hashes.

Reference snippet for Polars — display lineage summary as DataFrame.

 
lineage_records = [
    {
        "stage": s.stage,
        "input_rows": s.input_rows,
        "output_rows": s.output_rows,
        "rejected": s.rows_rejected,
        "duration_ms": round(s.duration_ms, 1),
        "output_hash": s.output_hash,
    }
    for s in run_context.stages
]
pl.DataFrame(lineage_records)
Notebook-rendered table output is preserved below.
stageinput_rowsoutput_rowsrejectedduration_msoutput_hash
bronze55052.300000426cfbf011fd33f5
silver2530253003080.90000012eeb022d46cec3e
gold511511052.70000057256313be661629
export3041304101.0000005daff78bb463b750

JSON — read back persisted run context with json.loads()

Verify RunContext JSON

Check the JSON file is complete and parseable. Shows business_context and temporal_context.

Reference snippet for JSON — read back persisted run context with json.loads().

 
ctx_json = json.loads(ctx_path.read_text(encoding="utf-8"))
 
# Display key fields without the bulky stages array
display_ctx = {k: v for k, v in ctx_json.items() if k != "stages"}
display_ctx["stages"] = f"[{len(ctx_json.get('stages', []))} stage records]"
json.dumps(display_ctx, indent=2, default=str)
Notebook stdout is preserved below.
{
  "batch_id": "05a35d97-97ef-43f8-a752-d45524767f62",
  "started_at": "2026-03-29T21:19:37.708034Z",
  "completed_at": "2026-03-29T21:19:41.357077Z",
  "symbols": [
    "SAP.DE",
    "SIE.DE",
    "ALV.DE",
    "DTE.DE",
    "BAS.DE"
  ],
  "date_range": [
    "2024-03-29",
    "2026-03-29"
  ],
  "polars_version": "1.39.3",
  "status": "completed",
  "business_context": {
    "trigger": "scheduled",
    "reason": null,
    "business_date": "2026-03-28",
    "is_correction": false,
    "affected_symbols": null
  },
  "temporal_context": {
    "as_of_date": "2026-03-28",
    "knowledge_date": "2026-03-29T21:19:37.706538Z",
    "reporting_period_start": "2024-03-29",
    "reporting_period_end": "2026-03-29",
    "timezone": "CET",
    "is_backfill": false
  },
  "data_warnings": [
    "sma_20: 95 NULL values (first 19 rows per symbol)"
  ],
  "contract_version": "1.0",
  "stages": "[4 stage records]"
}

Polars — query lineage table with read_database()

Verifies lineage records were persisted to SQL Server for the current batch.

Reference snippet for Polars — query lineage table with read_database().

 
lineage_query = pl.read_database(
    f"SELECT stage, input_rows, output_rows, rows_rejected, output_hash FROM lineage_stages WHERE batch_id = '{batch_id}'",
    connection=sql_engine
)
lineage_query
Notebook-rendered table output is preserved below.
stageinput_rowsoutput_rowsrows_rejectedoutput_hash
bronze550426cfbf011fd33f5
silver25302530012eeb022d46cec3e
gold511511057256313be661629
export3041304105daff78bb463b750

Polars — review quarantined rows with read_database()

Review Quarantined Rows

Shows rejected rows with error messages for investigation.

Reference snippet for Polars — review quarantined rows with read_database().

 
quarantine_df = pl.read_database(
    f"SELECT stage, symbol, date, error_message, quarantined_at "
    f"FROM quarantine WHERE batch_id = '{batch_id}' ORDER BY quarantined_at",
    connection=sql_engine
)
 
if len(quarantine_df) > 0:
    log.warning(f"Quarantined rows: {len(quarantine_df)}")
    display(quarantine_df)
else:
    log.info("No quarantined rows \u2014 all data passed validation")
    print("No quarantined rows")
Notebook stdout is preserved below.
23:19:41 | INFO  | No quarantined rows — all data passed validation

No quarantined rows

Polars — query context log for this batch with read_database()

Context Audit per Stage

Lineage = what happened. Context = what the pipeline knew at each stage.

Reference snippet for Polars — query context log for this batch with read_database().

 
context_df = pl.read_database(
    f"SELECT stage, business_date, trigger_type, schema_version, data_warnings "
    f"FROM context_log WHERE batch_id = '{batch_id}' ORDER BY created_at",
    connection=sql_engine
)
context_df
Notebook-rendered table output is preserved below.
stagebusiness_datetrigger_typeschema_versiondata_warnings
bronze2026-03-28 00:00:00scheduled1.0None
silver2026-03-28 00:00:00scheduled1.0["sma_20: 95 NULL values (first 19 rows per symbol)"]
gold2026-03-28 00:00:00scheduled1.0["sma_20: 95 NULL values (first 19 rows per symbol)"]

Python — display accumulated data warnings with log.warning()

Shows all data warnings accumulated across stages from the gold StageContext.

Reference snippet for Python — display accumulated data warnings with log.warning().

 
if gold_stage_ctx and gold_stage_ctx.data_warnings:
    print(f"Data Warnings ({len(gold_stage_ctx.data_warnings)} total):")
    for w in gold_stage_ctx.data_warnings:
        print(f"  {w}")
else:
    print("No data warnings — clean run")
Notebook stdout is preserved below.
Data Warnings (1 total):
  sma_20: 95 NULL values (first 19 rows per symbol)

JSON — inspect exported data contract with json.loads()

Inspects the Silver contract to show structural schema and column semantics side by side.

Reference snippet for JSON — inspect exported data contract with json.loads().

 
contract_path = EXPORT_DIR / "contracts" / "silver_ohlcv_contract.json"
if contract_path.exists():
    contract = json.loads(contract_path.read_text())
    print(f"Contract: {contract_path.name}")
    print(f"  Version: {contract.get('x-contract-version')}")
    print(f"  Generated: {contract.get('x-generated-at')}")
    print(f"  Fields: {len(contract.get('properties', {}))}")
    print(f"  Column contexts: {len(contract.get('x-column-context', []))}")
    print()
    for col in contract.get("x-column-context", []):
        if col.get("is_derived"):
            print(f"  {col['name']}:")
            print(f"    {col['description']}")
            print(f"    Computation: {col['computation']}")
            print(f"    Sources: {col['source_columns']}")
            print(f"    Null means: {col['null_semantics']}")
            print()
Notebook stdout is preserved below.
  daily_return:
    Close-to-close return

  intraday_range:
    (high-low)/close

  sma_20:
    20-day moving average of close

11. FastAPI Serving Layer — Pre-Materialized Parquet API

FastAPI serves the Gold data products by reading pre-materialized Parquet files. No database connection at runtime — the API reads files that the pipeline produced. Five endpoints serve different consumer needs: health (operational monitoring), daily-summary (market overview), symbol-profile (stock comparison), timeseries (per-symbol drill-down), and lineage (pipeline execution audit).

API models

Pydantic response models that define the JSON shape returned by each endpoint.

Pydantic — define daily summary API response model with BaseModel

API Response Schema

Response model for /daily-summary endpoint. No strict mode for FastAPI compatibility.

Reference snippet for Pydantic — define daily summary API response model with BaseModel.

 
class DailySummaryResponse(BaseModel):
    date:             Date
    symbols_traded:   int
    avg_return:       float
    max_return:       float
    min_return:       float
    total_volume:     int
    avg_intraday_pct: float
 
No direct stdout was emitted by this cell.

Pydantic — define symbol profile API response model with BaseModel

Response schema for the /symbol-profile endpoint.

Reference snippet for Pydantic — define symbol profile API response model with BaseModel.

 
class SymbolProfileResponse(BaseModel):
    symbol:              str
    total_trading_days:  int
    avg_daily_return:    float
    volatility:          float
    max_drawdown:        float
    avg_volume:          float
    total_dividends:     float
    first_date:          Date
    last_date:           Date
 
No direct stdout was emitted by this cell.

Pydantic — define timeseries row API response model with BaseModel

Response schema for the /symbol/{symbol}/timeseries endpoint.

Reference snippet for Pydantic — define timeseries row API response model with BaseModel.

 
class TimeSeriesRow(BaseModel):
    date:           Date
    open:           float
    high:           float
    low:            float
    close:          float
    volume:         int
    daily_return:   float
    intraday_range: float
    sma_20:         float | None
 
No direct stdout was emitted by this cell.

Endpoints

FastAPI application instance and five route handlers: health, daily-summary, symbol-profile, timeseries, and lineage.

FastAPI — create application instance with FastAPI()

Initializes the FastAPI application for serving pre-materialized Parquet data.

Reference snippet for FastAPI — create application instance with FastAPI().

 
app = FastAPI(title="Gold Data Pipeline API", version="1.0.0")
 
Notebook stdout is preserved below.
FastAPI app created

FastAPI — define health endpoint with @app.get()

Healthcheck endpoint that verifies Parquet files exist and reports their sizes.

Reference snippet for FastAPI — define health endpoint with @app.get().

 
@app.get("/health")
def health():
    """Healthcheck \u2014 verify Parquet files exist."""
    files = {f.stem: f.stat().st_size for f in EXPORT_DIR.glob("*.parquet")}
    return {"status": "healthy", "files": files}
 
No direct stdout was emitted by this cell.

FastAPI — define daily summary endpoint with @app.get()

Returns the daily cross-sectional summary from Parquet, with optional date range filter.

Reference snippet for FastAPI — define daily summary endpoint with @app.get().

 
@app.get("/daily-summary", response_model=list[DailySummaryResponse])
def get_daily_summary(start_date: Date | None = None, end_date: Date | None = None):
    """Return daily cross-sectional summary, optionally filtered by date range."""
    df = pl.read_parquet(EXPORT_DIR / "gold_daily_summary.parquet")
    if start_date:
        df = df.filter(pl.col("date") >= start_date)
    if end_date:
        df = df.filter(pl.col("date") <= end_date)
    return df.drop("batch_id").sort("date").to_dicts()
 
No direct stdout was emitted by this cell.

FastAPI — define symbol profile endpoint with @app.get()

Returns per-symbol summary statistics from the pre-materialized Parquet file.

Reference snippet for FastAPI — define symbol profile endpoint with @app.get().

 
@app.get("/symbol-profile", response_model=list[SymbolProfileResponse])
def get_symbol_profiles():
    """Return per-symbol summary statistics."""
    df = pl.read_parquet(EXPORT_DIR / "gold_symbol_profile.parquet")
    return df.drop("batch_id").sort("symbol").to_dicts()
 
No direct stdout was emitted by this cell.

FastAPI — define symbol timeseries endpoint with @app.get()

Returns daily OHLCV plus enrichment columns for one symbol from the Silver Parquet file.

Reference snippet for FastAPI — define symbol timeseries endpoint with @app.get().

 
@app.get("/symbol/{symbol}/timeseries", response_model=list[TimeSeriesRow])
def get_timeseries(symbol: str, limit: int = 100):
    """Return daily OHLCV + enrichment for one symbol."""
    df = pl.read_parquet(EXPORT_DIR / "silver_ohlcv.parquet")
    sym_df = df.filter(pl.col("symbol") == symbol.upper())
    if len(sym_df) == 0:
        raise HTTPException(status_code=404, detail=f"Symbol {symbol} not found")
    return sym_df.select(
        "date", "open", "high", "low", "close", "volume",
        "daily_return", "intraday_range", "sma_20"
    ).sort("date", descending=True).head(limit).to_dicts()
 
No direct stdout was emitted by this cell.

FastAPI — define lineage endpoint with @app.get()

Returns the RunContext JSON for a batch ID prefix, enabling audit queries from the API.

Reference snippet for FastAPI — define lineage endpoint with @app.get().

 
@app.get("/lineage/{batch_id_prefix}")
def get_lineage(batch_id_prefix: str):
    """Return RunContext JSON for a batch (matches by prefix)."""
    matches = list(LINEAGE_DIR.glob(f"run_{batch_id_prefix}*.json"))
    if not matches:
        raise HTTPException(status_code=404, detail="Batch not found")
    return json.loads(matches[0].read_text(encoding="utf-8"))
 
Notebook stdout is preserved below.
GET /lineage/{batch_id} registered — 9 total routes

Testing

Start the server in a background thread and smoke-test all five endpoints with httpx.

uvicorn — start API server in background with threading.Thread()

Background API Server

Port 8099 to avoid conflicts. Background thread allows notebook to continue.

Reference snippet for uvicorn — start API server in background with threading.Thread().

 
API_PORT = 8099
 
def run_server():
    """Run uvicorn in a background thread."""
    config = uvicorn.Config(app, host="127.0.0.1", port=API_PORT, log_level="warning")
    server = uvicorn.Server(config)
    server.run()
 
# Start server in background
server_thread = threading.Thread(target=run_server, daemon=True)
server_thread.start()
 
# Wait briefly for server to start
time.sleep(2)
API_PORT  # FastAPI server running at http://127.0.0.1
Notebook stdout is preserved below.
ERROR:    [Errno 10048] error while attempting to bind on address ('127.0.0.1', 8099): only one usage of each socket address (protocol/network address/port) is normally permitted

FastAPI server running at http://127.0.0.1:8099
http://127.0.0.1:8099/docs

httpx — test health endpoint with httpx.get()

Verifies the API server is running and that all Parquet files are accessible.

Reference snippet for httpx — test health endpoint with httpx.get().

 
resp = httpx.get(f"http://127.0.0.1:{API_PORT}/health")
resp.status_code  # Status
json.dumps(resp.json(), indent=2)
Notebook stdout is preserved below.
23:19:43 | INFO  | HTTP Request: GET http://127.0.0.1:8099/health "HTTP/1.1 200 OK"

200
{
  "status": "healthy",
  "files": {
    "gold_daily_summary": 21685,
    "gold_symbol_profile": 4036,
    "silver_ohlcv": 99214
  }
}

httpx — test daily summary endpoint with httpx.get()

Fetches the last 5 trading days of cross-sectional summary from the API.

Reference snippet for httpx — test daily summary endpoint with httpx.get().

 
five_days_ago = (Date.today() - timedelta(days=10)).isoformat()
resp = httpx.get(f"http://127.0.0.1:{API_PORT}/daily-summary", params={"start_date": five_days_ago})
resp.status_code, len(resp.json())  # status, rows
 
# Display as Polars DataFrame
pl.DataFrame(resp.json())
Notebook stdout is preserved below.
23:19:43 | INFO  | HTTP Request: GET http://127.0.0.1:8099/daily-summary?start_date=2026-03-19 "HTTP/1.1 200 OK"

200, rows: 7
datesymbols_tradedavg_returnmax_returnmin_returntotal_volumeavg_intraday_pct
2026-03-195-0.023291-0.008920-0.044730195135180.026323
2026-03-205-0.020986-0.002818-0.038625424862700.040470
2026-03-2350.0120350.037055-0.002530219007460.071276
2026-03-2450.0038540.041800-0.040556149909440.028663
2026-03-2550.0080210.023951-0.004877135363250.019512
2026-03-265-0.0060630.014394-0.015385157222940.019610
2026-03-275-0.0037680.026803-0.023123162296340.024848

httpx — test symbol profile endpoint with httpx.get()

Fetches all symbol profiles from the API and renders them as a DataFrame.

Reference snippet for httpx — test symbol profile endpoint with httpx.get().

 
resp = httpx.get(f"http://127.0.0.1:{API_PORT}/symbol-profile")
resp.status_code, len(resp.json())  # status, profiles
 
pl.DataFrame(resp.json())
Notebook stdout is preserved below.
23:19:44 | INFO  | HTTP Request: GET http://127.0.0.1:8099/symbol-profile "HTTP/1.1 200 OK"

200, profiles: 5
symboltotal_trading_daysavg_daily_returnvolatilitymax_drawdownavg_volumetotal_dividendsfirst_datelast_date
ALV.DE5060.0005320.011839-0.123504631486.14000029.2000002024-03-282026-03-27
BAS.DE5060.0001210.017491-0.2767662518782.1400005.6500002024-03-282026-03-27
DTE.DE5060.0007650.013250-0.2661096531019.0600001.6700002024-03-282026-03-27
SAP.DE506-0.0002850.018900-0.4914021676455.1900004.5500002024-03-282026-03-27
SIE.DE5060.0004750.019204-0.2732511155097.07000010.5500002024-03-282026-03-27

httpx — test symbol timeseries endpoint with httpx.get()

Fetches the last 10 days of SAP.DE time series data from the API.

Reference snippet for httpx — test symbol timeseries endpoint with httpx.get().

 
resp = httpx.get(f"http://127.0.0.1:{API_PORT}/symbol/SAP.DE/timeseries", params={"limit": 10})
resp.status_code, len(resp.json())  # status, rows
 
pl.DataFrame(resp.json()).head()
Notebook stdout is preserved below.
23:19:44 | INFO  | HTTP Request: GET http://127.0.0.1:8099/symbol/SAP.DE/timeseries?limit=10 "HTTP/1.1 200 OK"

200, rows: 10
dateopenhighlowclosevolumedaily_returnintraday_rangesma_20
2026-03-27145.740005147.320007142.100006142.5599983568581-0.0143810.036616161.330000
2026-03-26145.399994148.080002143.520004144.6399993752998-0.0153850.031527162.750000
2026-03-25148.779999150.539993145.360001146.8999943757697-0.0048770.035262164.129000
2026-03-24149.759995151.039993146.000000147.6199954380715-0.0405560.034142165.123000
2026-03-23150.460007161.520004150.399994153.86000141653680.0002600.072274166.034000

httpx — test lineage endpoint with httpx.get()

Fetches the pipeline execution metadata for the current run from the lineage endpoint.

Reference snippet for httpx — test lineage endpoint with httpx.get().

 
resp = httpx.get(f"http://127.0.0.1:{API_PORT}/lineage/{batch_id[:8]}")
resp.status_code  # Status
data = resp.json()
data['batch_id'][:8], data['status']  # batch, status
 
pl.DataFrame(data["stages"]).select("stage", "input_rows", "output_rows", "rows_rejected", "output_hash")
Notebook stdout is preserved below.
23:19:44 | INFO  | HTTP Request: GET http://127.0.0.1:8099/lineage/05a35d97 "HTTP/1.1 200 OK"

200
05a35d97... | completed
stageinput_rowsoutput_rowsrows_rejectedoutput_hash
bronze550426cfbf011fd33f5
silver25302530012eeb022d46cec3e
gold511511057256313be661629
export3041304105daff78bb463b750

12. Pipeline Visualization — Charts & Metrics

Visual validation of the pipeline output. Each chart answers a specific question about the data: daily return volatility (how noisy is the market?), cumulative investment performance (how would a 1 EUR investment have grown?), risk-return positioning (which stocks offer the best return per unit of risk?), and pipeline execution timing (which stage is the bottleneck?). Charts use dark-theme compatible transparent backgrounds.

Charts

Four Plotly charts covering return time series, cumulative performance, risk-return scatter, and stage timing.

Plotly — plot daily return time series with go.Scatter()

Overlaid line chart showing daily returns across all 5 symbols for the last 3 months.

Reference snippet for Plotly — plot daily return time series with go.Scatter().

 
three_months_ago = Date.today() - timedelta(days=90)
 
fig = go.Figure()
for symbol in SYMBOLS:
    sym_df = silver_df.filter(
        (pl.col("symbol") == symbol) & (pl.col("date") >= three_months_ago)
    ).sort("date")
    fig.add_trace(go.Scatter(
        x=sym_df["date"].to_list(),
        y=sym_df["daily_return"].to_list(),
        mode="lines", name=symbol, opacity=0.7
    ))
 
fig.update_layout(
    title="Daily Returns \u2014 Last 3 Months",
    template="plotly_dark",
    paper_bgcolor="rgba(0,0,0,0)",
    plot_bgcolor="rgba(0,0,0,0)",
    yaxis_title="Daily Return",
    xaxis_title="Date",
    legend=dict(orientation="h", y=-0.25),
)
fig.show()
Notebook-rendered visual output is preserved below.

Plotly — plot cumulative returns comparison with cum_prod()

Shows how a €1 investment in each symbol would have grown over the full history.

Reference snippet for Plotly — plot cumulative returns comparison with cum_prod().

 
fig = go.Figure()
for symbol in SYMBOLS:
    sym_df = silver_df.filter(pl.col("symbol") == symbol).sort("date")
    cum_ret = (1 + sym_df["daily_return"]).cum_prod()
    fig.add_trace(go.Scatter(
        x=sym_df["date"].to_list(),
        y=cum_ret.to_list(),
        mode="lines", name=symbol
    ))
 
fig.update_layout(
    title="Cumulative Returns — €1 Investment",
    template="plotly_dark",
    paper_bgcolor="rgba(0,0,0,0)",
    plot_bgcolor="rgba(0,0,0,0)",
    yaxis_title="Growth of €1",
    xaxis_title="Date",
    legend=dict(orientation="h", y=-0.15),
)
fig.show()
Notebook-rendered visual output is preserved below.

Plotly — plot risk-return scatter with go.Scatter()

Risk-return visualization using Gold symbol profile data: volatility vs average daily return.

Reference snippet for Plotly — plot risk-return scatter with go.Scatter().

 
fig = go.Figure()
fig.add_trace(go.Scatter(
    x=[v * 100 for v in valid_profiles["volatility"].to_list()],
    y=[v * 100 for v in valid_profiles["avg_daily_return"].to_list()],
    mode="markers+text",
    text=valid_profiles["symbol"].to_list(),
    textposition="top center",
    marker=dict(size=12, color="#4285F4"),
))
 
fig.update_layout(
    title="Risk-Return Profile — Volatility vs Avg Daily Return",
    template="plotly_dark",
    paper_bgcolor="rgba(0,0,0,0)",
    plot_bgcolor="rgba(0,0,0,0)",
    xaxis_title="Daily Volatility (%)",
    yaxis_title="Avg Daily Return (%)",
)
fig.show()
Notebook-rendered visual output is preserved below.

Plotly — plot pipeline stage timing with go.Bar()

Shows how long each pipeline stage took in milliseconds, identifying the bottleneck.

Reference snippet for Plotly — plot pipeline stage timing with go.Bar().

 
stages = [s.stage for s in run_context.stages]
durations = [s.duration_ms for s in run_context.stages]
 
fig = go.Figure()
fig.add_trace(go.Bar(
    x=stages, y=durations,
    marker_color=["#4285F4", "#34A853", "#FBBC04", "#EA4335"],
    text=[f"{d:.0f}ms" for d in durations],
    textposition="outside",
))
 
fig.update_layout(
    title="Pipeline Stage Duration",
    template="plotly_dark",
    paper_bgcolor="rgba(0,0,0,0)",
    plot_bgcolor="rgba(0,0,0,0)",
    yaxis_title="Duration (ms)",
    xaxis_title="Stage",
)
fig.show()
Notebook-rendered visual output is preserved below.

13. Audit — Investigating a Disputed Data Point

The audit section demonstrates lineage in action. A stakeholder disputes a specific data point — the pipeline traces it through five persisted checkpoints: Bronze (raw values as ingested), Silver (computed return verified mathematically), Gold (propagation to aggregation), Lineage (batch metadata with SHA-256 hash), and RunContext (execution fingerprint). The landing-zone JSON file is checked as supplemental corroboration only, because incremental fetches can overwrite the current raw file.

Disputed data point investigation

Five-step forensic trace from Gold back to persisted evidence: Bronze → Silver → Gold → Lineage → RunContext. Landing-zone JSON is a supplemental corroboration check.

SQL Server — query Bronze table for raw ingested values with read_database()

Start the forensic trace at Bronze because batch_id, ingested_at, and the raw OHLCV fields are the persisted source of truth for the disputed day. If the Bronze row is wrong, every downstream layer is only reproducing that original defect.

Reference snippet for SQL Server — query Bronze table for raw ingested values with read_database().

# Step 2: Check Bronze — exact values as persisted, with batch_id and ingestion timestamp
 
bronze_audit = pl.read_database(
    "SELECT symbol, date, [open] as [open], high, low, [close] as [close], "
    "adj_close, volume, dividends, stock_splits, batch_id, ingested_at "
    "FROM bronze_ohlcv "
    "WHERE symbol = 'SAP.DE' AND date = '2026-01-29'",
    connection=sql_engine
)
print("Bronze table (raw ingested):")
bronze_audit
Notebook stdout is preserved below.
Bronze table (raw ingested):
symboldateopenhighlowcloseadj_closevolumedividendsstock_splitsbatch_idingested_at
SAP.DE2026-01-29 00:00:00179.000000180.160004162.119995164.619995164.619995158467910.0000000.0000009c135c08-937d-4413-8bb4-1b66407ed9a52026-03-28 23:21:33.570000

SQL Server — query Silver table for enriched values with read_database()

Silver Return Verification

Verify daily_return = (close - prev_close) / prev_close mathematically.

Reference snippet for SQL Server — query Silver table for enriched values with read_database().

 
silver_audit = pl.read_database(
    "SELECT symbol, date, [close] as [close], daily_return, intraday_range, sma_20, "
    "batch_id, processed_at "
    "FROM silver_ohlcv "
    "WHERE symbol = 'SAP.DE' AND date BETWEEN '2026-01-28' AND '2026-01-30' "
    "ORDER BY date",
    connection=sql_engine
)
print("Silver table (enriched, 3-day window):")
 
# Manually verify the return calculation
rows = silver_audit.to_dicts()
if len(rows) >= 2:
    prev_close = rows[0]["close"]
    curr_close = rows[1]["close"]
    expected_return = (curr_close - prev_close) / prev_close
    actual_return = rows[1]["daily_return"]
    print(f"  Previous close (Jan 28): {prev_close:.2f}")
    print(f"  Current close  (Jan 29): {curr_close:.2f}")
    print(f"  Expected return: ({curr_close:.2f} - {prev_close:.2f}) / {prev_close:.2f} = {expected_return:.6f}")
    print(f"  Actual return:   {actual_return:.6f}")
    print(f"  Match: {abs(expected_return - actual_return) < 0.000001}")
    print()
 
silver_audit
Notebook stdout is preserved below.
Silver table (enriched, 3-day window):

  Previous close (Jan 28): 196.14
  Current close  (Jan 29): 164.62
  Expected return: (164.62 - 196.14) / 196.14 = -0.160702
  Actual return:   -0.160702
  Match: True
symboldateclosedaily_returnintraday_rangesma_20batch_idprocessed_at
SAP.DE2026-01-28 00:00:00196.1399990.0030680.020598202.37950005a35d97-97ef-43f8-a752-d45524767f622026-03-29 21:19:39.550000
SAP.DE2026-01-29 00:00:00164.619995-0.1607020.109586200.19300005a35d97-97ef-43f8-a752-d45524767f622026-03-29 21:19:39.550000
SAP.DE2026-01-30 00:00:00170.5599980.0360830.038227198.62350005a35d97-97ef-43f8-a752-d45524767f622026-03-29 21:19:39.553333

SQL Server — query Gold tables for aggregated impact with read_database()

Gold is where a single row-level anomaly becomes a mart-level effect, so this query verifies that the disputed Bronze/Silver event materially changed the daily aggregate. The goal is not to restate the row value, but to prove the downstream business surface reflects the same batch_id lineage.

Reference snippet for SQL Server — query Gold tables for aggregated impact with read_database().

# Step 4: Check Gold — verify the data point propagated to aggregations
 
gold_daily_audit = pl.read_database(
    "SELECT date, symbols_traded, avg_return, min_return, max_return, total_volume, batch_id "
    "FROM gold_daily_summary WHERE date = '2026-01-29'",
    connection=sql_engine
)
print("Gold daily summary (Jan 29):")
print(f"  The min_return on this day should reflect SAP's -16% drop")
gold_daily_audit
Notebook stdout is preserved below.
Gold daily summary (Jan 29):
  The min_return on this day should reflect SAP's -16% drop
datesymbols_tradedavg_returnmin_returnmax_returntotal_volumebatch_id
2026-01-29 00:00:005-0.025652-0.1607020.0201282535724705a35d97-97ef-43f8-a752-d45524767f62

SQL Server — query lineage table for pipeline run metadata with read_database()

Lineage Chain Trace

batch_id traces disputed row to pipeline run. Output hash proves no post-ingestion tampering.

Reference snippet for SQL Server — query lineage table for pipeline run metadata with read_database().

batch_from_bronze = pl.read_database(
    "SELECT batch_id FROM bronze_ohlcv WHERE symbol = 'SAP.DE' AND date = '2026-01-29'",
    connection=sql_engine
)["batch_id"][0]
 
print(f"Disputed row: SAP.DE / 2026-01-29")
batch_from_bronze  # Batch ID (from row)
 
# Trace that batch through every pipeline stage
lineage_audit = pl.read_database(
    f"SELECT stage, started_at, completed_at, input_rows, output_rows, "
    f"rows_rejected, output_hash "
    f"FROM lineage_stages WHERE batch_id = '{batch_from_bronze}' "
    f"ORDER BY started_at",
    connection=sql_engine
)
print("Full pipeline execution for this batch:")
lineage_audit
Notebook stdout is preserved below.
SAP.DE / 2026-01-29
9c135c08-937d-4413-8bb4-1b66407ed9a5

Full pipeline execution for this batch:
stagestarted_atcompleted_atinput_rowsoutput_rowsrows_rejectedoutput_hash
bronze2026-03-28 23:21:34.3505442026-03-28 23:21:37.061475253025300914eccd231d933a2
silver2026-03-28 23:25:09.9800662026-03-28 23:25:12.568855253025300d3c2286d9c3a8b8c
gold2026-03-28 23:29:18.3832732026-03-28 23:29:18.4330445115110fabdf6a3896591c1
export2026-03-28 23:29:41.3456502026-03-28 23:29:41.346711304130410ad1a03cae4b23504

JSON — verify RunContext execution metadata with json.loads()

RunContext Execution Proof

Proves symbols processed, date range, Polars version, status, zero rejections, output hash.

Reference snippet for JSON — verify RunContext execution metadata with json.loads().

 
ctx_files = list(LINEAGE_DIR.glob(f"run_{batch_from_bronze[:8]}*.json"))
if ctx_files:
    ctx = json.loads(ctx_files[0].read_text(encoding="utf-8"))
    print(f"RunContext: {ctx_files[0].name}")
    print(f"  Status:       {ctx['status']}")
    print(f"  Symbols:      {ctx['symbols']}")
    print(f"  Date range:   {ctx['date_range']}")
    print(f"  Polars:       {ctx['polars_version']}")
    total_rejected = sum(s['rows_rejected'] for s in ctx['stages'])
    print(f"  Rejected:     {total_rejected} rows (all data passed validation)")
    print(f"  Output hash:  {ctx['stages'][0]['output_hash']} (tamper-proof)")
else:
    print(f"No RunContext found for batch {batch_from_bronze[:8]}")
Notebook stdout is preserved below.
    ['2024-03-29', '2026-03-29']

JSON — verify landing-zone file continuity with json.loads()

Landing Zone File Proof

Check whether the current landing-zone file still carries the disputed date. Archived landing files are corroborative, not authoritative.

Reference snippet for JSON — verify landing-zone file continuity with json.loads().

 
landing_file = LANDING_DIR / "ohlcv_SAP_DE.json"
raw_records = json.loads(landing_file.read_text(encoding="utf-8"))
 
jan29_in_file = [r for r in raw_records if r["date"] == "2026-01-29"]
if jan29_in_file:
    live_close_29 = jan29_in_file[0]["close"]
    print(f"Landing file: {landing_file.name}")
    print(f"  Total records: {len(raw_records)}")
    print(f"  Jan 29 record found:")
    for k, v in jan29_in_file[0].items():
        print(f"    {k:15s}: {v}")
    print(f"\n  Bronze close: {float(bronze_audit['close'][0]):.2f}")
    print(f"  Landing close: {live_close_29}")
    print(f"  Match: {abs(float(live_close_29) - float(bronze_audit['close'][0])) < 0.01}")
else:
    # File has been overwritten by incremental fetch — Jan 29 no longer in file
    live_close_29 = float(bronze_audit["close"][0])
    dates = sorted(set(r["date"] for r in raw_records))
    print(f"Landing file covers: {dates[0]} to {dates[-1]}")
    print(f"Jan 29 not in current file (overwritten by incremental fetch)")
    print(f"In production, landing files are archived and would contain this record")
Notebook stdout is preserved below.
2026-03-27 to 2026-03-27
Jan 29 not in current file (overwritten by incremental fetch)
In production, landing files are archived and would contain this record

Polars — display full audit trail summary as DataFrame

Audit: Full Chain of Evidence

Assembles all audit steps into a summary. Each row is one verification step with evidence.

Reference snippet for Polars — display full audit trail summary as DataFrame.

 
bronze_close = float(bronze_audit["close"][0])
silver_row = silver_audit.filter(pl.col("date") == Date.fromisoformat("2026-01-29"))
silver_close = float(silver_row["close"][0])
 
audit_summary = pl.DataFrame([
    {"step": "1. Bronze table", "source": "bronze_ohlcv",              "close": bronze_close,  "evidence": f"Batch {batch_from_bronze[:8]}"},
    {"step": "2. Silver table", "source": "silver_ohlcv",              "close": silver_close,  "evidence": "Return = -16.07% verified"},
    {"step": "3. Gold table",   "source": "gold_daily_summary",        "close": None,          "evidence": "min_return reflects the drop"},
    {"step": "4. Lineage",      "source": "lineage_stages",            "close": None,          "evidence": f"Hash: {lineage_audit['output_hash'][0]}"},
    {"step": "5. RunContext",   "source": f"run_{batch_from_bronze[:8]}.json", "close": None,   "evidence": "0 rejected, status=completed"},
])
 
print("AUDIT CONCLUSION: The SAP.DE -16% drop on 2026-01-29 is AUTHENTIC.")
print("  - Same close price in Bronze and Silver")
print("  - Daily return verified mathematically from consecutive closes")
print("  - Zero rows rejected by Pydantic validation")
print("  - Output hash proves no post-ingestion tampering")
audit_summary
Notebook stdout is preserved below.
AUDIT CONCLUSION: The SAP.DE -16% drop on 2026-01-29 is AUTHENTIC.
  - Same close price in Bronze and Silver
  - Daily return verified mathematically from consecutive closes
  - Zero rows rejected by Pydantic validation
  - Output hash proves no post-ingestion tampering
stepsourcecloseevidence
1. Bronze tablebronze_ohlcv164.619995Batch 9c135c08
2. Silver tablesilver_ohlcv164.619995Return = -16.07% verified
3. Gold tablegold_daily_summarynanmin_return reflects the drop
4. Lineagelineage_stagesnanHash: 914eccd231d933a2
5. RunContextrun_9c135c08.jsonnan0 rejected, status=completed

Context-Driven Analysis — Metadata in Action

The three demonstrations below use real data from this pipeline run to show what context adds beyond lineage:

  • Zero-volume classification — context + trading calendar turns 116 undifferentiated alerts into classified holidays vs genuine anomalies
  • SMA-20 null accounting — context explains exactly how many nulls are expected per symbol and flags any that exceed the baseline
  • Data contract interpretation — column-level metadata makes Gold values self-describing without reading the pipeline source code

Zero-Volume Classification — Holiday or Anomaly?

Silver contains rows with volume=0. Without context, each is an undifferentiated alert. With the trading calendar cross-reference recorded at bronze ingestion, each is classified as 📅 non-trading day or 🔴 genuine anomaly.

Context: Zero-Volume Classification

Silver rows with volume=0 classified using ColumnContext. Distinguishes real data from quality issues.

Reference snippet for Zero-Volume Classification — Holiday or Anomaly?.

 
zero_vol = silver_df.filter(pl.col("volume") == 0)
 
if len(zero_vol) > 0:
    zero_dates = zero_vol.select("symbol", "date").unique()
 
    classified = []
    for row in zero_dates.iter_rows(named=True):
        cal = pl.read_database(
            f"SELECT is_trading_day FROM dim_calendar "
            f"WHERE date = '{row['date']}' AND exchange_code = 'XETR'",
            connection=sql_engine
        )
        is_trading = bool(cal["is_trading_day"][0]) if len(cal) > 0 else None
        classified.append({
            "symbol": row["symbol"],
            "date": row["date"],
            "is_trading_day": is_trading,
            "verdict": "anomaly" if is_trading else "non-trading day",
        })
 
    classification = pl.DataFrame(classified)
    anomalies = classification.filter(pl.col("verdict") == "anomaly")
    holidays = classification.filter(pl.col("verdict") == "non-trading day")
 
    display(classification.sort("date").head())
    print(f"\nTotal zero-volume: {len(classification)}  |  "
          f"Non-trading days: {len(holidays)}  |  "
          f"Anomalies to investigate: {len(anomalies)}")
else:
    print(f"No zero-volume rows in Silver \u2014 all {len(silver_df)} rows have volume > 0")
Notebook-rendered table output is preserved below.
symboldateis_trading_dayverdict
ALV.DE2024-09-20 00:00:00Nonenon-trading day
SAP.DE2024-10-21 00:00:00Nonenon-trading day
BAS.DE2024-10-21 00:00:00Nonenon-trading day
ALV.DE2024-11-01 00:00:00Nonenon-trading day
BAS.DE2024-11-01 00:00:00Nonenon-trading day
Total zero-volume: 116  |  Non-trading days: 116  |  Anomalies to investigate: 0

SMA-20 Null Accounting — Expected vs Unexpected

sma_20 requires 20 data points — the first 19 rows per symbol are NULL by mathematical necessity. Context recorded this at silver stage. If any symbol has MORE than 19 nulls, those extras are unexplained and need investigation.

Context: SMA-20 Null Accounting

sma_20 needs 20 data points. First 19 per symbol are NULL by mathematical necessity.

Reference snippet for SMA-20 Null Accounting — Expected vs Unexpected.

 
sma_nulls = silver_df.filter(pl.col("sma_20").is_null())
actual_null_count = len(sma_nulls)
 
# Expected: first 19 rows per symbol have no 20-day history
symbols_count = silver_df["symbol"].n_unique()
expected_null_count = 19 * symbols_count
 
# Show the null distribution per symbol
null_per_symbol = (
    sma_nulls.group_by("symbol").agg(
        pl.col("date").count().alias("null_count"),
        pl.col("date").min().alias("first_null"),
        pl.col("date").max().alias("last_null"),
    )
    .with_columns(
        pl.lit(19).alias("expected"),
        (pl.col("null_count") - 19).alias("unexplained"),
    )
    .sort("symbol")
)
 
display(null_per_symbol)
print(f"\nExpected nulls: {expected_null_count} (19 x {symbols_count} symbols)  |  "
      f"Actual: {actual_null_count}  |  "
      f"Unexplained: {actual_null_count - expected_null_count}")
 
# Context warning recorded this at silver stage
silver_warnings = [w for w in gold_stage_ctx.data_warnings if "sma_20" in w]
if silver_warnings:
    print(f"Context recorded: {silver_warnings[0]}")
Notebook-rendered table output is preserved below.
symbolnull_countfirst_nulllast_nullexpectedunexplained
ALV.DE192024-03-28 00:00:002024-04-25 00:00:00190
BAS.DE192024-03-28 00:00:002024-04-25 00:00:00190
DTE.DE192024-03-28 00:00:002024-04-25 00:00:00190
SAP.DE192024-03-28 00:00:002024-04-25 00:00:00190
SIE.DE192024-03-28 00:00:002024-04-25 00:00:00190
Expected nulls: 95 (19 x 5 symbols)  |  Actual: 95  |  Unexplained: 0
sma_20: 95 NULL values (first 19 rows per symbol)

Data Contract — Column Semantics as Structured Data

Each exported JSON Schema contract includes x-column-context with the computation formula, source columns, unit, and null semantics for every derived column. This is what turns volatility: 0.0187 into “daily σ of close-to-close returns, annualize with √252 → 29.7%“.

Context: Data Contract Metadata

Reads exported JSON Schema and displays x-column-context entries for gold_symbol_profile.

Reference snippet for Data Contract — Column Semantics as Structured Data.

 
contract = json.loads(
    (EXPORT_DIR / "contracts" / "gold_symbol_profile_contract.json").read_text()
)
 
derived = [
    {
        "column": c["name"],
        "description": c["description"],
        "unit": c["unit"],
        "computation": c.get("computation", "\u2014"),
        "source_columns": ", ".join(c.get("source_columns", [])) or "\u2014",
        "null_means": c.get("null_semantics", "\u2014"),
    }
    for c in contract["x-column-context"]
    if c.get("is_derived")
]
 
pl.DataFrame(derived)
Notebook-rendered table output is preserved below.
columndescriptionunitcomputationsource_columnsnull_means
total_trading_daysNumber of trading days with datacountcount(*) per symbolsilver.datenot_applicable
avg_daily_returnMean daily close-to-close return over full historydecimal_ratiomean(daily_return) per symbolsilver.daily_returnnot_applicable
volatilityStandard deviation of daily returns — annualize by multiplying by sqrt(252)decimal_ratiostd(daily_return) per symbolsilver.daily_returnnot_applicable
max_drawdownLargest peak-to-trough decline in cumulative return (always negative or zero)decimal_ratiomin(cumulative_return - running_max(cumulative_return)) per symbolsilver.daily_returnnot_applicable
avg_volumeMean daily trading volume over full historycountmean(volume) per symbolsilver.volumenot_applicable
total_dividendsSum of all dividends paid over full historyEURsum(dividends) per symbolsilver.dividendsnot_applicable

Context: Interpreting Gold Values

volatility=0.0187 means nothing without context. Contract says: std(daily_return), annualize by sqrt(252).

Reference snippet for Data Contract — Column Semantics as Structured Data.

 
import math
 
vol_meta = next(c for c in contract["x-column-context"] if c["name"] == "volatility")
german = valid_profiles.filter(pl.col("symbol").str.ends_with(".DE"))
 
interpretation = german.select(
    "symbol",
    pl.col("volatility").round(4).alias("daily_vol"),
    (pl.col("volatility") * math.sqrt(252) * 100).round(1).alias("annual_vol_%"),
).with_columns(
    pl.lit(vol_meta["unit"]).alias("unit"),
    pl.lit(vol_meta["computation"]).alias("formula"),
)
 
display(interpretation)
print(f"\nContract says: '{vol_meta['description']}'")
Notebook-rendered table output is preserved below.
symboldaily_volannual_vol_%unitformula
ALV.DE0.01180018.800000decimal_ratiostd(daily_return) per symbol
BAS.DE0.01750027.800000decimal_ratiostd(daily_return) per symbol
DTE.DE0.01320021.000000decimal_ratiostd(daily_return) per symbol
SAP.DE0.01890030.000000decimal_ratiostd(daily_return) per symbol
SIE.DE0.01920030.500000decimal_ratiostd(daily_return) per symbol
'Standard deviation of daily returns — annualize by multiplying by sqrt(252)'

Risk Notes

Boundary copies and async blocking

Crossing the boundary from pl.DataFrame to df.to_pandas() creates another mutable representation, and calling blocking work inside async def routes removes the concurrency benefits of FastAPI unless the operation is offloaded with run_in_executor() or asyncio.to_thread().

Measure event-loop blocking against thread offload for two equal-duration tasks.

import asyncio
import time
 
async def blocked():
    start = time.perf_counter()
 
    async def bad():
        time.sleep(0.05)
 
    await asyncio.gather(bad(), bad())
    return time.perf_counter() - start
 
async def offloaded():
    start = time.perf_counter()
 
    async def good():
        await asyncio.to_thread(time.sleep, 0.05)
 
    await asyncio.gather(good(), good())
    return time.perf_counter() - start
 
print(f"blocking={asyncio.run(blocked()):.3f}s")
print(f"offloaded={asyncio.run(offloaded()):.3f}s")
blocking=0.100s
offloaded=0.051s

Replay integrity and source hygiene

Bronze must remain append-only on (symbol, date) so prior batch_id evidence survives reruns, and raw vendor rows with None or inf must fail the RawOHLCV gate before they contaminate Silver.

Compare append-only merge semantics with truncate-and-reload behavior.

bronze = {("SAP.DE", "2026-01-29"): {"close": 164.62, "batch_id": "old"}}
incoming = {("SAP.DE", "2026-01-30"): {"close": 170.56, "batch_id": "new"}}
append_only = bronze | incoming
truncate_reload = incoming
invalid_close = None
 
print(f"append_only_keys={sorted(append_only)}")
print(f"truncate_keys={sorted(truncate_reload)}")
print(f"close_valid={invalid_close is not None}")
append_only_keys=[('SAP.DE', '2026-01-29'), ('SAP.DE', '2026-01-30')]
truncate_keys=[('SAP.DE', '2026-01-30')]
close_valid=False

Implementation Notes

Frozen contracts and targeted retries

Keep Gold DTOs immutable with ConfigDict(frozen=True) and scope retries to transient faults with retry_if_exception_type(ConnectionError | TimeoutError | OSError) instead of retrying business-rule failures.

Show that a frozen contract rejects post-construction mutation.

from pydantic import BaseModel, ConfigDict
 
class GoldRow(BaseModel):
    model_config = ConfigDict(frozen=True)
    symbol: str
 
row = GoldRow(symbol="SAP.DE")
try:
    row.symbol = "BAS.DE"
except Exception as exc:
    print(type(exc).__name__)
    print("mutation_blocked")
ValidationError
mutation_blocked

Configuration and lineage keys

Build SQL_CONN_STR from PIPELINE_SQL_USER and PIPELINE_SQL_PASSWORD at runtime, pre-materialize write_parquet() outputs before API startup, and stamp every persisted row with batch_id so exported marts and audit queries stay joinable.

Assemble the connection string from environment variables and carry the lineage key forward.

import os
 
os.environ["PIPELINE_SQL_USER"] = "etl_reader"
os.environ["PIPELINE_SQL_PASSWORD"] = "example-secret"
sql_conn_str = (
    "Driver={ODBC Driver 18 for SQL Server};"
    "Server=localhost,1434;Database=stoxx;"
    f"UID={os.environ['PIPELINE_SQL_USER']};PWD={os.environ['PIPELINE_SQL_PASSWORD']};"
)
row = {"symbol": "SAP.DE", "batch_id": "05a35d97", "parquet": "gold_daily_summary.parquet"}
 
print(sql_conn_str)
print(f"batch_id={row['batch_id']} parquet={row['parquet']}")
Driver={ODBC Driver 18 for SQL Server};Server=localhost,1434;Database=stoxx;UID=etl_reader;PWD=example-secret;
batch_id=05a35d97 parquet=gold_daily_summary.parquet

Failure Scenarios

Bronze contract rejects null numerics

A pydantic.ValidationError on close or another required field should go straight to quarantine; coercing None through Bronze only defers the failure into downstream aggregates.

Trigger the same class of Bronze validation failure raised by a missing numeric field.

from pydantic import BaseModel, ValidationError
 
class BronzeRow(BaseModel):
    close: float
 
try:
    BronzeRow(close=None)
except ValidationError as exc:
    first = exc.errors()[0]
    print(first["type"])
    print(first["loc"][0])
float_type
close

Silver transform receives string numerics

If a Bronze extract lands close as String, cast with pl.col("close").cast(pl.Float64) before pct_change() or other arithmetic; otherwise the transform boundary is operating on the wrong schema.

Cast a string price column before computing numeric aggregates.

import polars as pl
 
bronze_df = pl.DataFrame({"close": ["164.62", "170.56"]})
casted = bronze_df.with_columns(pl.col("close").cast(pl.Float64))
 
print(bronze_df.schema["close"])
print(casted.schema["close"])
print(f"mean_close={casted['close'].mean():.2f}")
String
Float64
mean_close=167.59

Export paths or hashes drift from expectations

If the API reads a different PARQUET_PATH than the export stage writes, users see stale marts; if the recomputed sha256 no longer matches output_hash, treat the row as tampered and investigate normalization or post-write mutation.

Contrast an endpoint path mismatch and a payload hash mismatch.

from pathlib import Path
import hashlib
 
export_path = Path("pipeline/gold_daily_summary.parquet")
api_path = Path("pipeline/gold_symbol_profile.parquet")
payload = "SAP.DE|2026-01-29|164.62"
mutated = "SAP.DE|2026-01-29|164.62 "
 
print(f"path_match={export_path == api_path}")
print(hashlib.sha256(payload.encode()).hexdigest()[:8])
print(hashlib.sha256(mutated.encode()).hexdigest()[:8])
path_match=False
b5931d6e
b2a77fb9

Graph View

Backlinks