03 - Functional Pipeline Architecture

Functional Pipeline Architecture

Quote

“The fundamental problem of state in data pipelines is not managing it — it is eliminating the need for it.”

— Nathan Marz (creator of Apache Storm)

Summary

This note defines a functional pipeline architecture that combines a functional core, strict boundary validation, runtime quality gates, lineage capture, immutable value objects, and context propagation so data pipelines stay testable, explainable, and safe under failure.

Architecture and core principles

• Explains the full pipeline shape, then separates pure transformation logic from the imperative shell that handles I/O, retries, and persistence.
• Treats functional structure as a practical engineering boundary for testability, reproducibility, and simpler debugging rather than as a purely stylistic preference.

Validation, gates, and trust signals

• Covers contract-first validation, runtime quality gates, quarantine handling, and the two dimensions of data trustworthiness that determine whether data may advance or must stop.
• Connects typed contracts and quality decisions to both Python and C# reference implementations so the principles stay operational rather than abstract.

Lineage, context, and consumer contracts

• Explains provenance tracking, immutable value objects, semantic context layers, and data contracts as the metadata surface that makes outputs self-describing and auditable.
• Shows how context accumulation and contract publication turn internal pipeline structure into reliable downstream interfaces.

Operations and safety

• Warnings: I/O inside transforms, missing quarantine paths, or confusing tests with runtime quality gates undermines the entire architectural separation.
• Recommendations: keep transforms pure, validate at every stage boundary, preserve rejected rows with context, and attach lineage plus semantic metadata before publishing consumer-facing contracts.

Glossary

Functional core

• The portion of a system made of pure deterministic transformations that depend only on their inputs and produce no side effects.

• It matters here because the note treats pure transforms as the foundation for testable and reproducible pipeline logic.

Debug by replay

A functional core is valuable because bugs can be reproduced with the same input dataframe rather than by reconstructing network, database, and credential state.

---

Imperative shell

• The I/O-handling layer that performs side effects such as fetching, writing, logging, retries, and orchestration around the pure core.

• It matters here because the architecture depends on keeping infrastructure concerns outside the transformation logic.

Boundary discipline required

Once database calls or file writes leak into transforms, the shell and core collapse together and the testability advantage disappears quickly.

---

Contract-first validation

• A design approach where every stage boundary is guarded by explicit typed models and validation rules before data is allowed to cross.

• It matters here because bad data is meant to fail early at the boundary instead of poisoning deeper layers silently.

Boundary, not afterthought

Validation is strongest when it is treated as part of the interface contract for a stage, not as an optional cleanup pass after loading.

---

Quality gate

• A runtime decision point that checks whether data quality signals are good enough for the next stage to proceed.

• It matters here because the note distinguishes quality enforcement in production from unit tests in development.

Not the same as tests

Tests validate code before release. Quality gates validate live data during execution, often using thresholds and operational context the test suite does not have.

---

Quarantine pattern

• A failure-handling pattern where invalid rows are diverted into a separate holding area with error context instead of being dropped or silently passed through.

• It matters here because the architecture needs a safe path for bad data that preserves evidence and supports later remediation.

Preserve the bad rows

Quarantine is useful precisely because rejected data is still operationally important. Teams need to inspect it, classify it, and sometimes fix and replay it.

---

Data provenance

• The lineage information that shows where data came from and which steps transformed it on the way to its current form.

• It matters here because the architecture aims to make every published output explainable under investigation.

Hidden lineage blocks trust

If a consumer cannot trace a number back through the pipeline, they will eventually stop trusting the number even if it is often correct.

---

Immutable value object

• A data structure whose contents do not change after creation, so later operations produce new values instead of mutating shared state.

• It matters here because immutability reduces hidden coupling and makes transformation steps easier to reason about.

State changes become explicit

Immutability forces each stage transition to be represented as a new output, which improves traceability and simplifies debugging.

---

Context propagation

• The practice of carrying metadata such as warnings, stage state, and semantic meaning along with the data as it moves through the pipeline.

• It matters here because the note treats context as part of the trust model, not as an optional annotation added later.

Meaning travels with the data

Context propagation prevents a pipeline from producing technically valid rows whose caveats or assumptions have been lost by the time they reach consumers.

---

Data contract

• A consumer-facing agreement that defines the schema, semantics, and quality expectations of a published output.

• It matters here because the architecture ends with explicit contracts rather than informal assumptions about what the pipeline emits.

Publish the guarantees you can keep

A contract that overpromises freshness, completeness, or meaning becomes another failure path. Contracts must be grounded in what the pipeline actually enforces.

---

Data trustworthiness

• The practical confidence that data is both structurally valid and contextually meaningful enough to be used safely.

• It matters here because the note treats trust as something engineered through contracts, lineage, and runtime checks rather than assumed from successful execution.

More than passing validation

Data can satisfy a schema and still be untrustworthy if its provenance, freshness, or semantic caveats are unclear.

The Complete Architecture

flowchart TB
    API[API] -->|retry| LAND[Landing]
    LAND --> VAL_B{{Validate}}
    VAL_B -->|valid| BRONZE[(Bronze)]
    VAL_B -->|invalid| DLQ[Quarantine]
    BRONZE --> TX[Transforms]
    TX --> VAL_S{{Validate}}
    VAL_S -->|valid| SILVER[(Silver)]
    VAL_S -->|invalid| DLQ
    SILVER --> QG{{Quality Gate}}
    QG -->|pass| AGG[Aggregation]
    QG -->|fail| STOP[STOP]
    AGG --> GOLD[(Gold)]
    GOLD --> PQ[Parquet]
    PQ --> SERVE[API Server]

    BRONZE -.->|hash| LIN[(Lineage)]
    SILVER -.->|hash| LIN
    GOLD -.->|hash| LIN

    BRONZE -.->|context| CTX_S[Context]
    CTX_S -.->|accumulate| CTX_G[Context]
    CTX_G -.->|export| CON[Contract]

    style API fill:#1a1a2e,stroke:#e8b84d,color:#fff
    style LAND fill:#1a1a2e,stroke:#e8b84d,color:#fff
    style VAL_B fill:#1a1a2e,stroke:#34a853,color:#fff
    style VAL_S fill:#1a1a2e,stroke:#34a853,color:#fff
    style TX fill:#1a1a2e,stroke:#4285f4,color:#fff
    style AGG fill:#1a1a2e,stroke:#4285f4,color:#fff
    style QG fill:#1a1a2e,stroke:#cc4125,color:#fff
    style BRONZE fill:#1a1a2e,stroke:#e8b84d,color:#fff
    style SILVER fill:#1a1a2e,stroke:#4285f4,color:#fff
    style GOLD fill:#1a1a2e,stroke:#4285f4,color:#fff
    style PQ fill:#1a1a2e,stroke:#4285f4,color:#fff
    style SERVE fill:#1a1a2e,stroke:#4285f4,color:#fff
    style DLQ fill:#cc4125,stroke:#a33,color:#fff
    style STOP fill:#cc4125,stroke:#a33,color:#fff
    style LIN fill:#669df6,stroke:#4285f4,color:#fff
    style CTX_S fill:#1a4d2e,stroke:#34a853,color:#fff
    style CTX_G fill:#1a4d2e,stroke:#34a853,color:#fff
    style CON fill:#1a4d2e,stroke:#34a853,color:#fff

Diagram legend

Amber border — Imperative Shell (I/O, network, database) Blue border — Functional Core (pure transforms, no side effects) Green border — Contract Enforcement (typed validation at boundaries) Red border — Quality Gates and failure paths (quarantine, stop) Blue fill — Lineage tracking (batch_id, SHA-256 hash at every stage) Green fill — Context propagation (warnings accumulate bronze → silver → gold → contract)

---

Functional Core, Imperative Shell

Origin: Gary Bernhardt, “Boundaries” talk (2012). Originally a software architecture pattern for isolating side effects from business logic.

The principle: Transforms are pure functions — given the same input DataFrame, they produce the same output DataFrame, every time. No database calls, no HTTP requests, no file I/O, no logging inside the transform. Everything with side effects (API calls, SQL writes, file saves, retry logic) lives in the “imperative shell” that calls the pure functions and handles I/O around them.

Why it matters: Pure functions are trivially testable (no mocking needed), trivially parallelizable (no shared state), and trivially debuggable (reproduce the bug by passing the same input). If a transform produces wrong output, the bug is in the function — not in a network timeout, a database lock, or a stale credential.

The purity test

Can you call this function in a unit test with a hardcoded DataFrame and assert the output WITHOUT setting up a database, network, or file system? If yes — functional core. If no — imperative shell. The notebooks enforce this boundary: transforms take DataFrames and return DataFrames. Everything else is infrastructure.

Functional core (pure transforms):

What	Python	C#
Daily returns	25_py_functional_pipeline > Polars — compute daily returns with pct_change().over()	25_cs_functional_pipeline > LINQ — compute daily returns with GroupBy().SelectMany()
Intraday range	25_py_functional_pipeline > Polars — compute intraday range with with_columns()	25_cs_functional_pipeline > LINQ — compute intraday range with Select()
Moving average	25_py_functional_pipeline > Polars — compute 20-day moving average with rolling_mean().over()	25_cs_functional_pipeline > LINQ — compute 20-day SMA with Skip().Take().Average()
Gold summary	25_py_functional_pipeline > Polars — build daily cross-sectional summary with group_by().agg()	25_cs_functional_pipeline > LINQ — build Gold daily summary with GroupBy().Select()
Gold profiles	25_py_functional_pipeline > Polars — build per-symbol profile with cum_max() drawdown	25_cs_functional_pipeline > LINQ — define standard deviation extension with Sum().Sqrt()

Imperative shell (I/O boundaries):

What	Python	C#
Configuration	25_py_functional_pipeline > Python — define pipeline paths, SQL connection, and stock universe	25_cs_functional_pipeline > Constants — define pipeline paths, SQL connection, and stock universe
API fetch	25_py_functional_pipeline > yfinance — fetch OHLCV to JSON landing zone with Ticker.history()	25_cs_functional_pipeline > HttpClient — fetch OHLCV from Yahoo Finance with Polly ExecuteAsync()
DB persistence	25_py_functional_pipeline > SQLAlchemy — define DataFrame write helper with to_sql()	25_cs_functional_pipeline > Dapper — define Bronze MERGE upsert with Execute()

Anti-pattern: transforms that call the database

A transform function that reads from SQL Server mid-computation, writes intermediate results to a file, or catches network errors — this is the shell leaking into the core. The transform becomes untestable without a live database and unpredictable under network failures. Keep I/O at the boundaries.

Move all I/O into the imperative shell — transforms take and return DataFrames only

A correctly structured transform has this signature: def compute_returns(df: pl.DataFrame) -> pl.DataFrame. No database handles, no file paths, no network calls. All lookups needed by the transform are pre-fetched in the shell and passed in as parameters. The transform is then testable with a hardcoded DataFrame — no infrastructure required.

---

Contract-First Validation

Origin: Data Contracts (Andrew Jones, 2022), Schema-on-Write (traditional RDBMS philosophy), Design by Contract (Bertrand Meyer, 1986).

The principle: Every stage boundary has a typed contract (Pydantic model / C# record + FluentValidation). Data that doesn’t conform is rejected BEFORE it crosses the boundary. The contract is code — version-controlled, unit-testable, and enforced at runtime.

Why it matters: Without contracts, bad data enters the pipeline silently. A renamed API field loads NULLs into bronze. A negative volume passes through to silver. A NaN daily return poisons the gold aggregation. By the time someone notices, the damage is three layers deep. Contracts catch bad data at ingestion — one layer, one fix.

Contract definitions:

Layer	Python	C#
Bronze	25_py_functional_pipeline > Pydantic — define Bronze validation model with BaseModel and Field()	25_cs_functional_pipeline > record — define Bronze OHLCV data model with Data Annotations
Silver	25_py_functional_pipeline > Pydantic — define Silver validation model with BaseModel and Field()	25_cs_functional_pipeline > record — define Silver OHLCV data model with enrichment fields
Gold	25_py_functional_pipeline > Pydantic — define Gold validation models with BaseModel and Field()	25_cs_functional_pipeline > record — define Gold daily summary data model
Validation rules	25_py_functional_pipeline > Pydantic — validate Bronze rows with BaseModel() row-level check	25_cs_functional_pipeline > FluentValidation — define Bronze validation rules with AbstractValidator<T>

Language comparison:

Aspect	Python	C#
Contract definition	Pydantic BaseModel with Field()	record with Data Annotations
Complex rules	@model_validator, @field_validator	FluentValidation AbstractValidator<T>
Strictness mode	ConfigDict(strict=True) — no coercion	Compile-time type safety + runtime validation
Error output	ValidationError with field-level messages	ValidationResult with Errors collection

See data-contracts for the broader contract theory and context-and-metadata-architecture > Schema Drift Detection for schema evolution patterns.

---

Quality Gate Pattern

Origin: Continuous Delivery (Jez Humble & David Farley, 2010). Originally a deployment concept — code must pass automated gates before reaching production. Applied here to data.

The principle: After each pipeline stage, automated assertions verify structural integrity, statistical bounds, and data freshness. If any assertion fails, the pipeline STOPS — downstream stages never see bad data. This is different from contract validation: contracts check individual rows at boundaries; quality gates check aggregate properties of the entire dataset after a stage completes.

Check	What It Catches	Example Threshold
Not empty	Failed fetch, empty API response	len(df) > 0
No null keys	Schema drift, type coercion failure	symbol, date have 0 nulls
No duplicates	Bad MERGE, double-fetch, dedup failure	0 duplicate (symbol, date) pairs
Value range	Outliers, data corruption, unit mismatch	daily_return within ±50%
Freshness	Stale data, broken source, timezone bug	Latest date within 5 days of today
Row count	Data loss, filter bug, source degradation	>= minimum expected rows

Quality gate implementations:

Check	Python	C#
Not empty	25_py_functional_pipeline > Polars — assert DataFrame is not empty with len()	25_cs_functional_pipeline > DataTable — define data quality assertion functions with AsEnumerable()
No null keys	25_py_functional_pipeline > Polars — assert no nulls in key columns with null_count()	(same file, same function)
No duplicates	25_py_functional_pipeline > Polars — assert no duplicate rows with unique()	(same file, same function)
Value range	25_py_functional_pipeline > Polars — assert values within range with filter()	(same file, same function)
Freshness	25_py_functional_pipeline > Polars — assert data freshness against SLA with max()	(same file, same function)
Orchestrator	25_py_functional_pipeline > Pipeline — run all quality gate assertions with log.info()	25_cs_functional_pipeline > DataTable — run Bronze data quality gate with RunQualityGate()
Exception	25_py_functional_pipeline > Python — define custom Exception subclass for quality gate failures	25_cs_functional_pipeline > Exception — define data quality gate failure exception

See data-quality-framework for the quality dimension taxonomy and data-pipeline-testing-strategy > Data quality assertions for where quality gates fit in the testing pyramid.

Quality gates are not tests

Tests run in CI before deployment — they catch CODE bugs. Quality gates run in production after every pipeline execution — they catch DATA bugs. You need both. A pipeline with perfect tests but no quality gates will silently ingest corrupt data from a source that changed its schema.

Implement both CI tests and runtime quality gates

CI tests (pytest, dbt test) validate your code logic against known fixtures. Quality gates (row count, null check, freshness, range bounds) run after every pipeline execution in production and stop the next stage if data integrity fails. See data-pipeline-testing-strategy for where each check belongs in the testing pyramid.

---

Data Provenance and Lineage Tracking

Origin: W3C PROV Model (2013), Data Governance literature, blockchain-inspired tamper detection.

The principle: Every row carries a batch_id linking it to the pipeline run that created it. Every stage records a StageLineage object: start/end timestamps, input/output row counts, rejection counts, and a SHA-256 hash of the output. The RunContext captures the full execution envelope (symbols processed, date range, library versions, status). All of this is persisted alongside the data.

Why it matters: When a business user disputes a gold-layer number (“why did ASML’s momentum score drop?”), you trace backward: gold row → batch_id → silver stage lineage → bronze stage lineage → landing JSON file → API call timestamp. The SHA-256 hash provides cryptographic proof that data wasn’t modified between stages.

Sample lineage query:

SELECT stage, started_at, completed_at,
       input_rows, output_rows, rows_rejected, output_hash
FROM pipeline_lineage
WHERE batch_id = '9e43b0c5-a388-4ec8-ad76-bcde6e97d37e'
ORDER BY started_at;

stage	started_at	completed_at	input_rows	output_rows	rejected	output_hash
bronze_ingest	08:00:01	08:00:04	1331	1329	2	a3f8b2…
silver_enrich	08:00:04	08:00:06	1329	1329	0	7c1d4e…
gold_aggregate	08:00:06	08:00:07	1329	50	0	e9f2a1…

Lineage implementations:

Component	Python	C#
Batch ID	25_py_functional_pipeline > uuid — generate unique batch ID with uuid4()	25_cs_functional_pipeline > Guid — generate unique batch ID with Guid.NewGuid()
SHA-256 hash	25_py_functional_pipeline > hashlib — compute deterministic DataFrame hash with sha256()	25_cs_functional_pipeline > SHA256 — compute deterministic data hash with SHA256.HashData()
Stage tracking	25_py_functional_pipeline > Python — define stage start and end tracker with datetime.now()	25_cs_functional_pipeline > DateTime — define stage start and end tracker with DateTime.UtcNow
Lineage models	25_py_functional_pipeline > Pydantic — define lineage tracking models with BaseModel and Field()	25_cs_functional_pipeline > record — define stage lineage tracking data model
Run context	25_py_functional_pipeline > Pydantic — save run context to JSON with model_dump_json()	25_cs_functional_pipeline > JsonSerializer — save run context to JSON with Serialize()
Persistence	25_py_functional_pipeline > SQL Server — define lineage persistence helper with cursor.execute()	25_cs_functional_pipeline > Dapper — define lineage persistence helper with Execute()

See context-and-metadata-architecture for the broader provenance theory including bi-temporal modeling and context propagation patterns.

---

Immutable Value Objects

Origin: Domain-Driven Design (Eric Evans, 2003), Functional Programming.

The principle: Pydantic models with strict=True and C# record types are immutable by default. Once created, they cannot be modified — you create a new instance with different values. This eliminates mutation bugs where a transform accidentally modifies its input. Combined with value equality (two records with identical fields are equal), immutability enables reliable change detection for SCD2 upserts and MERGE operations.

Implementations: The contract models from the Contract-First Validation section (above) serve double duty — they are both validation contracts AND immutable value objects. The SCD2 upsert logic relies on this:

• Python SCD2: 25_py_functional_pipeline > SQL Server — define SCD Type 2 upsert for one symbol with MERGE INTO
• C#: 25_cs_functional_pipeline > record — define Bronze OHLCV data model with Data Annotations (records provide built-in value equality)

Immutability enables change detection

SCD Type 2 upserts need to compare “current vs incoming” to detect changes. With mutable objects, comparing two instances requires field-by-field checks that may miss newly added fields. With records/immutable models, value equality is built in — old == new compares every field automatically. If they differ, close the old record and insert the new one.

---

The Quarantine Pattern

The principle: Rows that fail validation are not discarded — they’re persisted to a quarantine table with the batch_id, stage, raw data, and full error message. This enables investigation (why did these rows fail?), replay (fix the source and re-ingest), and metrics (what percentage of rows fail per source? is it getting worse?).

Implementations:

Component	Python	C#
Quarantine table DDL	25_py_functional_pipeline > SQL Server — create quarantine table for rejected rows with cursor.execute()	(same DDL)
Quarantine persistence	25_py_functional_pipeline > SQL Server — define quarantine persistence helper with cursor.execute()	25_cs_functional_pipeline > Dapper — define quarantine persistence helper with Execute()
Review quarantined rows	25_py_functional_pipeline > Polars — review quarantined rows with read_database()	25_cs_functional_pipeline > Dapper — review quarantined rows with QueryToTable()

See error-handling-and-retry-patterns for broader error handling theory and data-quality-framework > Data Quality Quarantine Pattern for the quarantine pattern in the quality framework.

Never silently drop bad rows

if not valid: continue is the most dangerous line in a data pipeline. The row disappears. Nobody knows it existed. The row count drops by one. Weeks later, someone asks why a symbol is missing from the gold report. Always quarantine — the 5 lines of code to persist rejected rows save hours of investigation.

Persist every rejected row to the quarantine table with full error context

Call quarantine_row(batch_id, stage, raw_data, error_message) for every row that fails validation. The quarantine table records what the row contained, which stage rejected it, and why. This enables investigation, replay after a fix, and quality metrics (rejection rate per source over time).

---

Two Dimensions of Data Trustworthiness

The five principles above — functional core, contract validation, quality gates, lineage, immutability — form the structural dimension. They ensure data is correct: typed, validated, auditable, and reproducible at every stage.

But correct data is not necessarily useful data. A gold table with volatility: 0.0187 is structurally perfect — it has a type, a hash, a lineage record, and it passed all quality gates. Yet no downstream consumer can interpret it without reading the pipeline source code.

The semantic dimension cuts across stages: column registries explain what each field means, business context records why the run happened, temporal context separates data date from load date, and warnings accumulate from bronze to gold. Together the two dimensions make data trustworthy.

The Intersection

Structural integrity answers: “Is this data correct?” Semantic integrity answers: “What does this data mean?” A trustworthy pipeline delivers both — verifiably correct AND self-describing.

---

Context Architecture — Semantic Metadata Layer

Context is metadata that flows THROUGH the pipeline alongside the data, growing richer at each stage. Unlike lineage (recorded after the fact), context is created at stage start and carried forward. By gold, the context contains the accumulated knowledge from every upstream stage.

ColumnContext — What Each Value Means

Each column carries structured metadata: description, unit, computation formula, source columns, null semantics, and valid range. Without it, volatility: 0.0187 is an opaque number. With it: “daily σ of close-to-close returns, decimal_ratio, annualize with √252.”

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. The data contract eliminates this: unit=decimal_ratio, formula=std(daily_return), annualize with √252. The number becomes self-describing.

Attach a ColumnContext record to every computed column

For each column in a gold table, define a ColumnContext with description, unit, formula, source_columns, and null_semantics. Export these as x-column-context in the JSON Schema contract. Any consumer — a dashboard, another pipeline, an LLM — can interpret every value correctly without reading the pipeline source code.

BusinessContext — Why This Run Happened

Records the trigger (scheduled, manual, backfill, reprocess), the is_correction flag, and the business date. Without it, two batches covering the same date range are indistinguishable — was the second a correction or a duplicate?

TemporalContext — Bi-Temporal Markers

Separates as_of_date (what date is this data FOR) from knowledge_date (when did we learn about it). Without it, a backfill loading 2024 data in 2026 looks like a normal 2026 run. See context-and-metadata-architecture > Temporal Context — As of When Is This Data True? for the broader theory.

StageContext — The Propagation Carrier

The carrier that propagates all context through the pipeline via for_next_stage(). Each stage inherits upstream warnings and adds its own. By gold, the context contains the full warning chain from every stage.

Warning Accumulation

Bronze records “116 zero-volume rows detected.” Silver inherits that warning and adds “95 SMA-20 NULLs (first 19 rows × 5 symbols).” Gold inherits both. Any consumer reading gold context sees the full chain without querying intermediate tables.

Context Persistence

Context is persisted to context_log in SQL Server — it survives the Python/C# process exit. Query it for any batch to reconstruct the full semantic state at each stage.

Implementations:

Component	Python	C#
ColumnContext model	py	cs
Column registries	py	cs
BusinessContext model	py	cs
TemporalContext model	py	cs
StageContext model	py	cs
Context persistence	py	cs
context_log DDL	py	(same DDL)

---

Data Contracts as Consumer-Facing Output

The Contract-First Validation section above covers contracts as INPUT validation — rejecting bad data at boundaries. This section covers contracts as OUTPUT — making gold values self-describing for any consumer.

The pipeline exports a JSON Schema file per gold table, enriched with x-column-context — the column registries serialized as structured metadata alongside the schema. Any consumer — a dashboard, another pipeline, an LLM agent — can interpret every value correctly without reading the pipeline source code.

{
  "x-column-context": [
    {
      "name": "volatility",
      "description": "Std dev of daily returns — annualize with √252",
      "unit": "decimal_ratio",
      "computation": "std(daily_return) per symbol",
      "source_columns": ["silver.daily_return"],
      "null_semantics": "insufficient_data"
    }
  ]
}

Contracts Turn Numbers Into Knowledge

Without the contract, volatility: 0.0187 requires reading the pipeline source. With the contract, any consumer reads: unit=decimal_ratio, formula=std(daily_return), annualize with √252 → 29.7% annual volatility. The data is self-describing.

Implementations:

Component	Python	C#
Contract export	py	cs
Contract inspection	py	cs

See data-contracts for the broader contract specification theory. See ai-augmented-data-engineering > Self-Describing Data for AI Consumers for how AI agents consume these contracts in practice.

---

Context-Driven Decisions — Real Data Proof

Context architecture produces real value through the pipeline’s own data — not hypothetical scenarios, but actual output where context answered a question that the data alone couldn’t.

Zero-Volume Classification

116 Silver rows have volume=0. Without context, each is an undifferentiated alert. The pipeline cross-referenced each zero-volume date against dim_calendar at bronze ingestion and recorded the classification (non-trading day vs genuine anomaly) as a context warning. The warning propagates through silver and gold — consumers know WHY volume is zero without investigating.

SMA-20 Null Accounting

sma_20 has 95 NULLs — exactly 19 × 5 symbols (the first 19 rows per symbol lack enough history for a 20-day average). Context recorded “95 NULL values (first 19 rows per symbol)” at silver stage. If any symbol had MORE than 19, those extras would be unexplained. Context draws the line between expected and unexpected NULLs.

Contract Interpretation

The AI agent scenario: volatility: 0.0187 is meaningless without the contract. With x-column-context, the consumer reads: unit=decimal_ratio, formula=std(daily_return), annualize with √252 → 29.7%. No source code reading required.

Implementations:

Demonstration	Python	C#
Zero-volume classification	py	cs
SMA-20 null accounting	py	cs
Contract interpretation	py	cs

Context Makes Data Self-Describing

Lineage traces data BACKWARD through the pipeline — where did this row come from? Context explains data FORWARD to any consumer — what does this value mean? Together they make data trustworthy: verifiably correct AND self-describing.

See ai-augmented-data-engineering for how AI agents consume context-enriched data.

---

Implementation Comparison

Concept	Python	C#
Validation	Pydantic v2 BaseModel	record + FluentValidation
Transforms	Polars expressions	LINQ
Retry	tenacity @retry	Polly WaitAndRetryAsync
DB access	pyodbc + SQLAlchemy	Dapper
Hashing	hashlib.sha256	SHA256.HashData
Serving	FastAPI	ASP.NET Core (HttpListener)
Immutability	ConfigDict(strict=True)	record types (default immutable)
Serialization	model_dump_json()	JsonSerializer.Serialize()
Batch ID	uuid.uuid4()	Guid.NewGuid()
Upsert	pyodbc MERGE statement	Dapper Execute() with MERGE
Quality checks	Custom dq_check_* functions	Custom Dq* assertion functions
Dead letter queue	quarantine_row() → SQL Server	QuarantineRow() → SQL Server
Column context	ColumnContext(BaseModel)	ColumnContext record
Business context	BusinessContext(BaseModel)	BusinessContext record
Context propagation	StageContext.for_next_stage()	StageContext.ForNextStage()
Data contract export	export_contracts() → JSON Schema	ExportContracts() → JSON
Context persistence	persist_context() → context_log	PersistContext() → context_log

---

Related

• 25_py_functional_pipeline — Python reference implementation (Pydantic + Polars + tenacity)
• 25_cs_functional_pipeline — C# reference implementation (FluentValidation + LINQ + Polly)
• medallion-architecture — Bronze/Silver/Gold data layering (this page builds on top of medallion)
• idempotent-pipeline-design — MERGE upsert and safe re-run patterns
• data-contracts — Contract specification, breaking vs non-breaking changes
• data-quality-framework — Quality dimensions, medallion quality gates, quarantine pattern
• data-pipeline-testing-strategy — Where quality gates fit in the testing pyramid
• error-handling-and-retry-patterns — Retry strategies, circuit breaker, dead letter queue theory
• context-and-metadata-architecture — The five types of pipeline context, bi-temporal modeling, schema evolution
• ai-augmented-data-engineering — AI agents as consumers of context-enriched data
• data-modeling-patterns — SCD Type 2 pattern used in dim_symbol