25 - Functional Pipeline - C#

101 min read

Functional Pipeline - C#

A production-grade 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: HttpClient → JSON landing → FluentValidation → Bronze → LINQ transforms → Silver → LINQ aggregation → Gold → Parquet → HttpListener API

Two orthogonal dimensions of data trustworthiness:

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

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)

This note implements an end-to-end functional data pipeline in C#/.NET using FluentValidation, Dapper, LINQ transforms, and HttpListener with medallion architecture, lineage tracking, and SHA-256 tamper detection.

#r "nuget: Microsoft.Data.SqlClient"
#r "nuget: Dapper"
#r "nuget: FluentValidation"
#r "nuget: Polly"
#r "nuget: Plotly.NET, 5.1.0"
#r "nuget: Plotly.NET.Interactive, 5.0.0"
#r "nuget: Plotly.NET.CSharp, 0.13.0"
#r "nuget: Polars.NET"
#r "nuget: Polars.NET.Native.win-x64"
#r "nuget: ParquetSharp"
 
using System;
using System.Collections.Generic;
using System.ComponentModel.DataAnnotations;
using System.Data;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using System.Threading;
using System.Threading.Tasks;
using System.Reflection;
using Microsoft.Data.SqlClient;
using Microsoft.DotNet.Interactive;
using Microsoft.DotNet.Interactive.CSharp;
using Microsoft.DotNet.Interactive.Formatting;
using Dapper;
using FluentValidation;
using Polly;
using Polly.Retry;
using Plotly.NET;
using Plotly.NET.CSharp;
using Chart = Plotly.NET.CSharp.Chart;
using Plotly.NET.LayoutObjects;
 
// ── Warning suppression ──
var csharpKernel = (CSharpKernel)Kernel.Root.FindKernelByName("csharp");
var optionsField = typeof(CSharpKernel).GetField("_scriptOptions",
    BindingFlags.NonPublic | BindingFlags.Instance);
var scriptOptions = optionsField.GetValue(csharpKernel);
var withWarningLevel = scriptOptions.GetType().GetMethod("WithWarningLevel");
var newOptions = withWarningLevel.Invoke(scriptOptions, new object[] { 0 });
optionsField.SetValue(csharpKernel, newOptions);
 
// ── DataTable HTML formatter — transparent background ──
Formatter.Register<DataTable>((dt, writer) => {
    writer.Write("<table style='border-collapse:collapse;background:transparent;color:inherit;'>");
    writer.Write("<tr>");
    foreach (DataColumn col in dt.Columns)
        writer.Write($"<th style='text-align:left;padding:4px 12px;border-bottom:1px solid #555;'>{col.ColumnName}</th>");
    writer.Write("</tr>");
    foreach (DataRow row in dt.Rows) {
        writer.Write("<tr>");
        foreach (var val in row.ItemArray)
            writer.Write($"<td style='text-align:left;padding:4px 12px;'>{val}</td>");
        writer.Write("</tr>");
    }
    writer.Write("</table>");
}, "text/html");

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

Defines all top-level constants for the pipeline run: filesystem paths, SQL Server connection string, stock universe, and date range.

C# — define pipeline paths, SQL connection, and stock universe

Central Configuration Cell

All downstream cells reference these constants — paths, SQL connection, stock universe, and date range. Change them here, not in individual cells.

// ── Paths ──
string DATA_DIR    = @"C:\Users\aperi\DEV\LANG\data";
string EXPORT_DIR  = Path.Combine(DATA_DIR, "pipeline");
string LINEAGE_DIR = Path.Combine(EXPORT_DIR, "lineage");
string LANDING_DIR = Path.Combine(DATA_DIR, "pipeline", "landing");
Directory.CreateDirectory(EXPORT_DIR);
Directory.CreateDirectory(LANDING_DIR);
Directory.CreateDirectory(LINEAGE_DIR);
 
// ── SQL Server (local Docker instance) ──
string SQL_CONN_STR = "Server=localhost,1434;Database=stoxx;User Id=sa;Password=EsgDev2026Pass1;Encrypt=yes;TrustServerCertificate=yes;";
SqlConnection sqlConn = new SqlConnection(SQL_CONN_STR);
sqlConn.Open();
 
// ── Stock universe — 5 EURO STOXX 50 components for demo ──
string[] SYMBOLS = { "SAP.DE", "SIE.DE", "ALV.DE", "DTE.DE", "BAS.DE" };
int LOOKBACK_DAYS = 365 * 2;
string START_DATE = DateTime.Today.AddDays(-LOOKBACK_DAYS).ToString("yyyy-MM-dd");
string END_DATE   = DateTime.Today.ToString("yyyy-MM-dd");
 
// ── Exchange mapping: Yahoo Finance exchange code → calendar name ──
Dictionary<string, string> EXCHANGE_MAP = new() {
    ["GER"] = "XETR", ["FRA"] = "XFRA", ["PAR"] = "XPAR",
    ["AMS"] = "XAMS", ["BRU"] = "XBRU", ["MIL"] = "XMIL",
    ["MCE"] = "XMAD", ["NMS"] = "XNYS", ["NYQ"] = "XNYS",
    ["HKG"] = "XHKG", ["TKS"] = "XTKS",
};
 
Console.WriteLine(EXPORT_DIR);   // export dir
Console.WriteLine(string.Join(", ", SYMBOLS));  // universe
Console.WriteLine($"{START_DATE} → {END_DATE}");  // date range
C:\Users\aperi\DEV\LANG\data\pipeline
SAP.DE, SIE.DE, ALV.DE, DTE.DE, BAS.DE
2024-03-30 → 2026-03-30

2. Records + FluentValidation — Schema Validation at Every Boundary

Every stage boundary has a typed contract — a C# record 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 Yahoo Finance 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.

With Typed Contracts

C# record types with FluentValidation enforce schema and business rules at the bronze boundary. A renamed field throws a ValidationException on the first bad row — the pipeline stops, logs the rejection to quarantine, and the problem is fixed at the source before any corrupt data reaches silver or gold.

Medallion layer models

Typed record contracts for the three medallion layers (Bronze, Silver, Gold) plus the operational lineage model.

record — define Bronze validation model with record and properties

Bronze Contract: RawOhlcv

Validates raw Yahoo Finance data BEFORE persistence to SQL Server. Enforces positive prices, non-negative volume, symbol not empty. FluentValidation adds High >= Low (market invariant — any violation = bad data). C# records are immutable by default.

public record RawOhlcv(
    string Symbol,
    DateTime Date,
    double Open,
    double High,
    double Low,
    double Close,
    double AdjClose,
    long Volume,
    double Dividends = 0.0,
    double StockSplits = 0.0);
 
public class RawOhlcvValidator : AbstractValidator<RawOhlcv>
{
    public RawOhlcvValidator()
    {
        RuleFor(x => x.Symbol).NotEmpty();
        RuleFor(x => x.Open).GreaterThan(0);
        RuleFor(x => x.High).GreaterThan(0);
        RuleFor(x => x.Low).GreaterThan(0);
        RuleFor(x => x.Close).GreaterThan(0);
        RuleFor(x => x.AdjClose).GreaterThan(0);
        RuleFor(x => x.Volume).GreaterThanOrEqualTo(0);
        RuleFor(x => x.Dividends).GreaterThanOrEqualTo(0);
        RuleFor(x => x.StockSplits).GreaterThanOrEqualTo(0);
        RuleFor(x => x.High).GreaterThanOrEqualTo(x => x.Low)
            .WithMessage("high must be >= low (market invariant)");
    }
}
 
var sample = new RawOhlcv("SAP.DE", new DateTime(2024, 1, 2), 144.5, 146.0, 143.8, 145.2, 145.2, 1_200_000);
var validator = new RawOhlcvValidator();
var result = validator.Validate(sample);
SAP.DE 2024-01-02 close=145.2

record — define Silver validation model with record and properties

Silver Contract: CleanOhlcv

Extends Bronze with three computed fields: daily_return, intraday_range, sma_20. Daily return constrained to [-50%, +50%] to catch extreme calculation errors. batch_id required — every Silver row must trace back to a pipeline run.

public record CleanOhlcv(
    string Symbol,
    DateTime Date,
    double Open,
    double High,
    double Low,
    double Close,
    double AdjClose,
    long Volume,
    double Dividends,
    double StockSplits,
    double DailyReturn,
    double IntradayRange,
    double? Sma20,
    string BatchId);
 
public class CleanOhlcvValidator : AbstractValidator<CleanOhlcv>
{
    public CleanOhlcvValidator()
    {
        RuleFor(x => x.Symbol).NotEmpty();
        RuleFor(x => x.Open).GreaterThan(0);
        RuleFor(x => x.High).GreaterThan(0);
        RuleFor(x => x.Low).GreaterThan(0);
        RuleFor(x => x.Close).GreaterThan(0);
        RuleFor(x => x.AdjClose).GreaterThan(0);
        RuleFor(x => x.Volume).GreaterThanOrEqualTo(0);
        RuleFor(x => x.IntradayRange).GreaterThanOrEqualTo(0);
        RuleFor(x => x.High).GreaterThanOrEqualTo(x => x.Low)
            .WithMessage("high must be >= low");
        RuleFor(x => x.BatchId).NotEmpty();
    }
}

record — define Gold validation models with record and properties

Gold Contracts: Two Mart Tables

DailySummary: one row per trading day with cross-sectional metrics. SymbolProfile: one row per symbol with full-history aggregate stats. max_drawdown constrained to ⇐ 0 (always negative — peak-to-trough decline).

public record DailySummary(
    DateTime Date,
    int SymbolsTraded,
    double AvgReturn,
    double MaxReturn,
    double MinReturn,
    long TotalVolume,
    double AvgIntradayPct,
    string BatchId);
 
public record SymbolProfile(
    string Symbol,
    int TotalTradingDays,
    double AvgDailyReturn,
    double Volatility,
    double MaxDrawdown,
    double AvgVolume,
    double TotalDividends,
    DateTime FirstDate,
    DateTime LastDate,
    string BatchId);
 
8 fields
10 fields

record — define lineage tracking models with record and properties

Lineage Model: StageLineage

Records what a single pipeline stage produced: input_rows / output_rows / rows_rejected for data flow accounting, output_hash (SHA-256) for tamper detection, DurationMs computed from timestamps.

public record StageLineage(
    string BatchId,
    string Stage,
    DateTime StartedAt,
    DateTime CompletedAt,
    int InputRows,
    int OutputRows,
    int RowsRejected,
    string OutputHash)
{
    public double DurationMs => (CompletedAt - StartedAt).TotalMilliseconds;
}

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.

With Semantic Context

ColumnContext records attach unit, formula, null semantics, and source columns to every field in the gold contract. Any consumer — API, AI agent, analyst — reads the contract and knows exactly what each number means without consulting the pipeline author.

record — define column semantic metadata model with record

Semantic Metadata: ColumnContext

Describes WHAT a column means, not just its type. computation: formula used to derive it. source_columns: upstream dependencies. null_semantics: what NULL means (“insufficient_data” vs “source_missing”). is_derived: True = computed by pipeline, False = raw from source.

public record ColumnContext(
    string Name,
    string Description,
    string Unit,
    string Computation = null,
    string[] SourceColumns = null,
    (double Min, double Max)? ValidRange = null,
    string NullSemantics = "not_applicable",
    bool IsBusinessKey = false,
    bool IsDerived = false)
{
    public string[] SourceColumns { get; init; } = SourceColumns ?? Array.Empty<string>();
}

C# — define column registries for each medallion layer

Column Registries per Layer

Each column has a ColumnContext entry documenting what it is, how it was computed, what NULL means, and its valid range. These registries feed into data contracts (exported as JSON Schema) and attach to StageContext for cross-stage propagation.

List<ColumnContext> BRONZE_COLUMNS = new() {
    new("symbol", "Yahoo Finance ticker symbol", "identifier", IsBusinessKey: true),
    new("date", "Trading date (exchange local)", "date", IsBusinessKey: true),
    new("open", "Opening price", "EUR", ValidRange: (0.001, 100000)),
    new("high", "Highest price", "EUR", ValidRange: (0.001, 100000)),
    new("low", "Lowest price", "EUR", ValidRange: (0.001, 100000)),
    new("close", "Closing price", "EUR", ValidRange: (0.001, 100000)),
    new("adj_close", "Adjusted close", "EUR", ValidRange: (0.001, 100000)),
    new("volume", "Shares traded", "count", ValidRange: (0, 1e12)),
    new("dividends", "Dividend paid", "EUR", ValidRange: (0, 1000)),
    new("stock_splits", "Split ratio", "ratio", ValidRange: (0, 100)),
};
 
List<ColumnContext> SILVER_COLUMNS = new(BRONZE_COLUMNS) {
    new("daily_return", "Close-to-close return", "decimal_ratio",
        Computation: "pct_change(close).over(symbol)", SourceColumns: new[]{"bronze.close"},
        ValidRange: (-0.5, 0.5), NullSemantics: "first_row_in_series", IsDerived: true),
    new("intraday_range", "(high-low)/close", "decimal_ratio",
        Computation: "(high - low) / close", SourceColumns: new[]{"bronze.high","bronze.low","bronze.close"},
        ValidRange: (0, 0.5), IsDerived: true),
    new("sma_20", "20-day moving average of close", "EUR",
        Computation: "close.rolling_mean(20).over(symbol)", SourceColumns: new[]{"bronze.close"},
        ValidRange: (0.001, 100000), NullSemantics: "insufficient_data", IsDerived: true),
};
 
List<ColumnContext> GOLD_DAILY_COLUMNS = new() {
    new("date", "Trading date", "date", IsBusinessKey: true),
    new("symbols_traded", "Distinct symbols", "count",
        Computation: "count(distinct symbol) per date", SourceColumns: new[]{"silver.symbol"}, IsDerived: true),
    new("avg_return", "Mean daily return", "decimal_ratio",
        Computation: "mean(daily_return) per date", SourceColumns: new[]{"silver.daily_return"}, IsDerived: true),
    new("max_return", "Best return", "decimal_ratio", IsDerived: true),
    new("min_return", "Worst return", "decimal_ratio", IsDerived: true),
    new("total_volume", "Sum of volume", "count", IsDerived: true),
    new("avg_intraday_pct", "Mean intraday range", "decimal_ratio", IsDerived: true),
};
 
List<ColumnContext> GOLD_PROFILE_COLUMNS = new() {
    new("symbol", "Yahoo Finance ticker symbol", "identifier", IsBusinessKey: true, NullSemantics: "not_applicable"),
    new("total_trading_days", "Number of trading days with data", "count",
        Computation: "count(*) per symbol", SourceColumns: new[]{"silver.date"}, ValidRange: (1, 5000), IsDerived: true),
    new("avg_daily_return", "Mean daily close-to-close return over full history", "decimal_ratio",
        Computation: "mean(daily_return) per symbol", SourceColumns: new[]{"silver.daily_return"}, ValidRange: (-0.1, 0.1), IsDerived: true),
    new("volatility", "Standard deviation of daily returns \u2014 annualize by multiplying by sqrt(252)", "decimal_ratio",
        Computation: "std(daily_return) per symbol", SourceColumns: new[]{"silver.daily_return"}, ValidRange: (0, 1), IsDerived: true),
    new("max_drawdown", "Largest peak-to-trough decline in cumulative return (always negative or zero)", "decimal_ratio",
        Computation: "min(cumulative_return - running_max(cumulative_return)) per symbol",
        SourceColumns: new[]{"silver.daily_return"}, ValidRange: (-1, 0), IsDerived: true),
    new("avg_volume", "Mean daily trading volume over full history", "count",
        Computation: "mean(volume) per symbol", SourceColumns: new[]{"silver.volume"}, ValidRange: (0, 1e12), IsDerived: true),
    new("total_dividends", "Sum of all dividends paid over full history", "EUR",
        Computation: "sum(dividends) per symbol", SourceColumns: new[]{"silver.dividends"}, ValidRange: (0, 10000), IsDerived: true),
};
 
Console.WriteLine($"Column registries: Bronze={BRONZE_COLUMNS.Count}, Silver={SILVER_COLUMNS.Count}, " +
    $"Gold Daily={GOLD_DAILY_COLUMNS.Count}, Gold Profile={GOLD_PROFILE_COLUMNS.Count}");
Bronze=10, Silver=13, Gold Daily=7, Gold Profile=7

record — define business context model with record

BusinessContext: Run Trigger Reason

Captures WHY this pipeline execution happened. trigger: scheduled, manual, backfill, reprocess, or test. is_correction: true if overwriting previously published data. Enables downstream consumers to distinguish routine runs from corrections.

public record BusinessContext
{
    public string Trigger { get; init; }
    public string Reason { get; init; }
    public DateTime BusinessDate { get; init; } = DateTime.Today;
    public bool IsCorrection { get; init; }
    public string[] AffectedSymbols { get; init; }
 
    public static readonly HashSet<string> ValidTriggers = new() {
        "scheduled", "manual", "backfill", "reprocess", "test"
    };
}

record — define temporal context model with record

TemporalContext: Bi-Temporal Markers

as_of_date: the business date the data represents (usually T-1). knowledge_date: when the pipeline ingested it (auto-set to now). Separates “what date is this data FOR” from “when did we learn about it” — critical for backfills where knowledge_date >> as_of_date.

public record TemporalContext
{
    public DateTime AsOfDate { get; init; }
    public DateTime ReportingPeriodStart { get; init; }
    public DateTime ReportingPeriodEnd { get; init; }
    public DateTime KnowledgeDate { get; init; } = DateTime.UtcNow;
    public string Timezone { get; init; } = "UTC";
    public bool IsBackfill { get; init; }
}

record — define stage context model for cross-stage propagation with record

StageContext: Cross-Stage Propagation

Unlike StageLineage (recorded after the fact), StageContext is created at stage start and carried forward via ForNextStage(). Each stage inherits upstream warnings and adds its own. By gold, the context carries the full warning chain from all stages.

public class StageContext
{
    public string BatchId { get; set; }
    public string Stage { get; set; }
    public List<StageLineage> UpstreamStages { get; set; } = new();
    public List<string> DataWarnings { get; set; } = new();
    public string SchemaVersion { get; set; } = "1.0";
    public List<ColumnContext> ColumnCtx { get; set; } = new();
    public BusinessContext BizContext { get; set; }
    public TemporalContext TempContext { get; set; }
 
    public void AddWarning(string warning)
    {
        DataWarnings.Add(warning);
        Console.WriteLine($"  WARNING {Stage}: {warning}");
    }
 
    public StageContext ForNextStage(string nextStage, StageLineage lineage,
        List<ColumnContext> columns)
    {
        return new StageContext {
            BatchId = BatchId, Stage = nextStage,
            UpstreamStages = new List<StageLineage>(UpstreamStages) { lineage },
            DataWarnings = new List<string>(DataWarnings),
            SchemaVersion = SchemaVersion, ColumnCtx = columns,
            BizContext = BizContext, TempContext = TempContext,
        };
    }
}

record — define pipeline run context model with record

RunContext: Execution Envelope

Aggregates everything: stages, business context, temporal context, data warnings, and contract version into a single JSON artifact per pipeline execution.

// accumulated warnings, contract version. Persisted as JSON per run.
 
public class RunContext
{
    public string BatchId { get; set; }
    public DateTime StartedAt { get; set; }
    public DateTime? CompletedAt { get; set; }
    public string[] Symbols { get; set; }
    public string[] DateRange { get; set; }
    public string PolarsVersion { get; set; } = "Polars.NET";
    public List<StageLineage> Stages { get; set; } = new();
    public string Status { get; set; } = "running";
    public BusinessContext BizContext { get; set; }
    public TemporalContext TempContext { get; set; }
    public List<string> DataWarnings { get; set; } = new();
    public string ContractVersion { get; set; } = "1.0";
}

C# — define data contract export function with JsonSerializer

Data Contract Export to JSON Schema

Generates machine-readable contracts for each pipeline boundary. Each contract includes structural schema (types, constraints) PLUS x-column-context with descriptions, formulas, units, and null semantics.

List<string> ExportDataContracts(string exportDir)
{
    var contractsDir = Path.Combine(exportDir, "contracts");
    Directory.CreateDirectory(contractsDir);
 
    var registry = new Dictionary<string, List<ColumnContext>> {
        ["bronze_ohlcv"] = BRONZE_COLUMNS,
        ["silver_ohlcv"] = SILVER_COLUMNS,
        ["gold_daily_summary"] = GOLD_DAILY_COLUMNS,
        ["gold_symbol_profile"] = GOLD_PROFILE_COLUMNS,
    };
 
    var paths = new List<string>();
    foreach (var (name, columns) in registry)
    {
        var properties = new Dictionary<string, object>();
        foreach (var col in columns)
        {
            properties[col.Name] = new Dictionary<string, object> {
                ["description"] = col.Description,
                ["x-unit"] = col.Unit,
            };
        }
 
        var schema = new Dictionary<string, object> {
            ["type"] = "object",
            ["properties"] = properties,
            ["x-column-context"] = columns.Select(c => new Dictionary<string, object> {
                ["name"] = c.Name, ["description"] = c.Description,
                ["unit"] = c.Unit, ["computation"] = c.Computation ?? "",
                ["source_columns"] = c.SourceColumns,
                ["valid_range"] = c.ValidRange.HasValue
                    ? new[] { c.ValidRange.Value.Min, c.ValidRange.Value.Max } : null,
                ["null_semantics"] = c.NullSemantics,
                ["is_business_key"] = c.IsBusinessKey,
                ["is_derived"] = c.IsDerived,
            }).ToArray(),
            ["x-contract-version"] = "1.0",
            ["x-generated-at"] = DateTime.UtcNow.ToString("o"),
        };
 
        var path = Path.Combine(contractsDir, $"{name}_contract.json");
        File.WriteAllText(path, JsonSerializer.Serialize(schema, new JsonSerializerOptions { WriteIndented = true }));
        paths.Add(path);
        Console.WriteLine($"  Contract exported: {Path.GetFileName(path)}");
    }
    return paths;
}

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.

With Lineage Tracking

batch_id links every row to its pipeline run. SHA-256 hashes prove rows were not modified after ingestion. RunContext records symbol count, date range, and rejection count at commit time. Any dispute is resolved with three SQL queries — no manual archaeology.

Lineage helpers

Core functions for generating batch IDs, computing SHA-256 tamper-detection hashes, and recording start/end stage metrics.

Guid — generate unique batch ID with Guid.NewGuid()

Batch ID: Unique Run Identifier

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

string GenerateBatchId()
{
    return Guid.NewGuid().ToString();
}
 
// Demo: generate a batch_id
var demoBatch = GenerateBatchId();
4af95c21-c810-416d-81ec-75d175e0d6d3

SHA256 — compute deterministic DataTable hash with SHA256.HashData()

SHA-256 Hash: Tamper Detection

Same data → same hash, every time. If someone modifies a row after the pipeline ran, the recomputed hash won’t match. Serializes DataTable to sorted CSV bytes before hashing.

string ComputeHash(DataTable dt)
{
    var sb = new StringBuilder();
    var colNames = dt.Columns.Cast<DataColumn>().Select(c => c.ColumnName).OrderBy(c => c).ToList();
    sb.AppendLine(string.Join(",", colNames));
    var rows = dt.AsEnumerable().OrderBy(r => string.Join(",", colNames.Select(c => r[c]?.ToString() ?? "")));
    foreach (var row in rows)
        sb.AppendLine(string.Join(",", colNames.Select(c => row[c]?.ToString() ?? "")));
    var hash = SHA256.HashData(Encoding.UTF8.GetBytes(sb.ToString()));
    return Convert.ToHexString(hash).ToLower().Substring(0, 16);
}
 
// Demo with a small table
var demoDt = new DataTable();
demoDt.Columns.Add("a", typeof(int));
demoDt.Columns.Add("b", typeof(int));
demoDt.Rows.Add(1, 4); demoDt.Rows.Add(2, 5); demoDt.Rows.Add(3, 6);
68135205be4bfc0c

C# — define stage start and end tracker with DateTime.UtcNow

Stage Tracking: Start/End Pattern

StartStage(): captures timestamp and input row count at entry. EndStage(): fills output metrics, computes SHA-256 hash, returns StageLineage. StageContext (if provided) flows alongside for semantic metadata propagation.

Dictionary<string, object> StartStage(string batchId, string stage, int inputRows,
    StageContext stageContext = null)
{
    return new Dictionary<string, object> {
        ["batch_id"] = batchId,
        ["stage"] = stage,
        ["started_at"] = DateTime.UtcNow,
        ["input_rows"] = inputRows,
        ["stage_context"] = stageContext,
    };
}
 
StageLineage EndStage(Dictionary<string, object> ctx, DataTable outputDt,
    int rowsRejected = 0)
{
    return new StageLineage(
        BatchId: (string)ctx["batch_id"],
        Stage: (string)ctx["stage"],
        StartedAt: (DateTime)ctx["started_at"],
        CompletedAt: DateTime.UtcNow,
        InputRows: (int)ctx["input_rows"],
        OutputRows: outputDt.Rows.Count,
        RowsRejected: rowsRejected,
        OutputHash: ComputeHash(outputDt)
    );
}

C# — save run context to JSON with JsonSerializer.Serialize()

Save RunContext to JSON

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

 
string SaveRunContext(RunContext ctx)
{
    var path = Path.Combine(LINEAGE_DIR, $"run_{ctx.BatchId.Substring(0, 8)}.json");
    File.WriteAllText(path, JsonSerializer.Serialize(ctx, new JsonSerializerOptions { WriteIndented = true }));
    return path;
}

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.

With Operational Tables

lineage_stages makes every run auditable and replayable. quarantine preserves rejected rows with full error context — any bad batch can be fixed and replayed without re-fetching from the API. context_log persists pipeline decisions so post-run queries can explain every warning.

TablePurposeKey
bronze_ohlcvRaw Yahoo Finance 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)

Table definitions

DDL for the nine medallion, dimension, and operational tables.

SQL Server — create Bronze OHLCV table with sqlConn.Execute()

Bronze Table: Raw Source Data

Stores raw Yahoo Finance output exactly as received. batch_id links every row to the pipeline run that ingested it. UNIQUE on (symbol, date) enables MERGE upsert for incremental loads.

sqlConn.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 Server — create Silver OHLCV table with sqlConn.Execute()

Silver Table DDL

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

 
sqlConn.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 Server — create Gold daily summary table with sqlConn.Execute()

Gold Daily Summary DDL

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

 
sqlConn.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 Server — create Gold symbol profile table with sqlConn.Execute()

Gold Symbol Profile DDL

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

 
sqlConn.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 Server — create SCD Type 2 symbol dimension with sqlConn.Execute()

SCD Type 2 Dimension DDL

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

 
sqlConn.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 Server — create per-exchange trading calendar with sqlConn.Execute()

Trading Calendar Dimension

Per-exchange trading calendar with holiday flags. Composite PK on . Aligned with the stoxx.bronze.trading_calendar schema.

 
sqlConn.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 Server — create lineage tracking table with Execute()

Lineage Table DDL

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

 
sqlConn.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 Server — create quarantine table for rejected rows with Execute()

Quarantine: Dead Letter Queue

Stores every row that failed validation. Preserves the raw data + rejection reason for investigation and replay.

 
sqlConn.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 Server — create context log table with Execute()

// Persists StageContext records — business context, temporal context, warnings
 
sqlConn.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()
)
");

Persistence helpers

Helper functions for writing stage context, lineage records, upsert data, and quarantine rows to SQL Server.

SQL Server — define context persistence helper with Execute()

Writes a StageContext record to the context_log table.

 
void PersistContext(StageContext stageCtx)
{
    if (stageCtx == null) return;
    sqlConn.Execute(
        @"INSERT INTO context_log
          (batch_id, stage, business_date, trigger_type, is_correction,
           schema_version, data_warnings, column_context, temporal_json)
          VALUES (@BatchId, @Stage, @BizDate, @Trigger, @IsCorrection,
                  @SchemaVer, @Warnings, @ColCtx, @TempJson)",
        new {
            stageCtx.BatchId, stageCtx.Stage,
            BizDate = stageCtx.BizContext?.BusinessDate,
            Trigger = stageCtx.BizContext?.Trigger,
            IsCorrection = stageCtx.BizContext?.IsCorrection ?? false,
            SchemaVer = stageCtx.SchemaVersion,
            Warnings = stageCtx.DataWarnings.Any()
                ? JsonSerializer.Serialize(stageCtx.DataWarnings) : null,
            ColCtx = stageCtx.ColumnCtx.Any()
                ? JsonSerializer.Serialize(stageCtx.ColumnCtx.Select(c => new {
                    c.Name, c.Description, c.Unit, c.Computation,
                    source_columns = c.SourceColumns, c.NullSemantics,
                    is_business_key = c.IsBusinessKey, is_derived = c.IsDerived,
                    valid_range = c.ValidRange.HasValue
                        ? new[]{ c.ValidRange.Value.Min, c.ValidRange.Value.Max } : null
                  })) : null,
            TempJson = stageCtx.TempContext != null
                ? JsonSerializer.Serialize(stageCtx.TempContext) : null,
        });
}

SQL Server — define lineage persistence helper with Execute()

Idempotent Lineage Persistence

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

 
void PersistLineage(StageLineage lineage)
{
    sqlConn.Execute(
        "DELETE FROM lineage_stages WHERE batch_id = @BatchId AND stage = @Stage",
        new { lineage.BatchId, lineage.Stage });
    sqlConn.Execute(
        @"INSERT INTO lineage_stages
          (batch_id, stage, started_at, completed_at, input_rows, output_rows, rows_rejected, output_hash)
          VALUES (@BatchId, @Stage, @StartedAt, @CompletedAt, @InputRows, @OutputRows, @RowsRejected, @OutputHash)",
        new {
            lineage.BatchId, lineage.Stage, lineage.StartedAt, lineage.CompletedAt,
            lineage.InputRows, lineage.OutputRows, lineage.RowsRejected, lineage.OutputHash
        });
}

Dapper — define DataFrame write helper with Execute()

DataFrame Write Helper

Writes a DataTable to SQL Server using Dapper row-by-row. wipes the table before insert (used by Gold tables only).

 
DataTable QueryToTable(string sql, object param = null)
{
    var reader = sqlConn.ExecuteReader(sql, param);
    var dt = new DataTable();
    dt.Load(reader);
    return dt;
}
 
void WriteToCsv(DataTable dt, string table, bool truncate = true)
{
    if (truncate)
        sqlConn.Execute($"TRUNCATE TABLE {table}");
 
    foreach (DataRow row in dt.Rows)
    {
        var cols = string.Join(", ", dt.Columns.Cast<DataColumn>()
            .Select(c => c.ColumnName == "open" || c.ColumnName == "close"
                ? $"[{c.ColumnName}]" : c.ColumnName));
        var parms = string.Join(", ", dt.Columns.Cast<DataColumn>()
            .Select(c => $"@{c.ColumnName}"));
        var parameters = new DynamicParameters();
        foreach (DataColumn col in dt.Columns)
            parameters.Add(col.ColumnName, row[col]);
        sqlConn.Execute($"INSERT INTO {table} ({cols}) VALUES ({parms})", parameters);
    }
}

SQL Server — define Bronze MERGE upsert with MERGE INTO

Bronze MERGE Upsert

Idempotent: MERGE on — safe to re-run. Existing rows get updated, new rows get inserted.

 
int MergeBronze(DataTable dt, string batchId)
{
    int rowsAffected = 0;
    foreach (DataRow row in dt.Rows)
    {
        rowsAffected += sqlConn.Execute(@"
            MERGE bronze_ohlcv AS tgt
            USING (SELECT @symbol AS symbol, @date AS date) AS src
               ON tgt.symbol = src.symbol AND tgt.date = src.date
            WHEN MATCHED THEN UPDATE SET
                [open] = @open, high = @high, low = @low, [close] = @close,
                adj_close = @adj_close, volume = @volume, dividends = @dividends,
                stock_splits = @stock_splits,
                batch_id = @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 (@symbol, @date, @open, @high, @low, @close, @adj_close,
                        @volume, @dividends, @stock_splits, @batch_id);",
            new {
                symbol = row["symbol"].ToString(), date = (DateTime)row["date"],
                open = Convert.ToDouble(row["open"]), high = Convert.ToDouble(row["high"]),
                low = Convert.ToDouble(row["low"]), close = Convert.ToDouble(row["close"]),
                adj_close = Convert.ToDouble(row["adj_close"]),
                volume = Convert.ToInt64(row["volume"]),
                dividends = Convert.ToDouble(row["dividends"]),
                stock_splits = Convert.ToDouble(row["stock_splits"]),
                batch_id = batchId,
            });
    }
    return rowsAffected;
}

SQL Server — define Silver MERGE upsert with MERGE INTO

Silver MERGE Upsert

Same idempotent pattern as Bronze, but includes enrichment columns: , , .

 
int MergeSilver(DataTable dt, string batchId)
{
    int rowsAffected = 0;
    foreach (DataRow row in dt.Rows)
    {
        rowsAffected += sqlConn.Execute(@"
            MERGE silver_ohlcv AS tgt
            USING (SELECT @symbol AS symbol, @date AS date) AS src
               ON tgt.symbol = src.symbol AND tgt.date = src.date
            WHEN MATCHED THEN UPDATE SET
                [open] = @open, high = @high, low = @low, [close] = @close,
                adj_close = @adj_close, volume = @volume, dividends = @dividends,
                stock_splits = @stock_splits,
                daily_return = @daily_return, intraday_range = @intraday_range, sma_20 = @sma_20,
                batch_id = @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 (@symbol, @date, @open, @high, @low, @close, @adj_close,
                        @volume, @dividends, @stock_splits,
                        @daily_return, @intraday_range, @sma_20, @batch_id);",
            new {
                symbol = row["symbol"].ToString(), date = (DateTime)row["date"],
                open = Convert.ToDouble(row["open"]), high = Convert.ToDouble(row["high"]),
                low = Convert.ToDouble(row["low"]), close = Convert.ToDouble(row["close"]),
                adj_close = Convert.ToDouble(row["adj_close"]),
                volume = Convert.ToInt64(row["volume"]),
                dividends = Convert.ToDouble(row["dividends"]),
                stock_splits = Convert.ToDouble(row["stock_splits"]),
                daily_return = row["daily_return"] is DBNull ? (double?)null : Convert.ToDouble(row["daily_return"]),
                intraday_range = row["intraday_range"] is DBNull ? (double?)null : Convert.ToDouble(row["intraday_range"]),
                sma_20 = row["sma_20"] is DBNull ? (double?)null : Convert.ToDouble(row["sma_20"]),
                batch_id = batchId,
            });
    }
    return rowsAffected;
}

SQL Server — define quarantine persistence helper with Execute()

// Persists a rejected row to the quarantine table with its error message
 
void QuarantineRow(string batchId, string stage, Dictionary<string, object> rowData, string error)
{
    sqlConn.Execute(
        @"INSERT INTO quarantine (batch_id, stage, symbol, date, raw_data, error_message)
          VALUES (@BatchId, @Stage, @Symbol, @Date, @RawData, @Error)",
        new {
            BatchId = batchId, Stage = stage,
            Symbol = rowData.GetValueOrDefault("symbol")?.ToString(),
            Date = rowData.ContainsKey("date") ? rowData["date"] : null,
            RawData = JsonSerializer.Serialize(rowData),
            Error = error.Length > 4000 ? error[..4000] : error,
        });
}

Polly — define API retry wrapper with WaitAndRetryAsync() exponential backoff

Polly Retry Policy

Wraps API calls with 3 attempts and exponential backoff. Catches transient failures.

 
AsyncRetryPolicy retryPolicy = Policy
    .Handle<HttpRequestException>()
    .Or<TaskCanceledException>()
    .WaitAndRetryAsync(
        retryCount: 3,
        sleepDurationProvider: attempt => TimeSpan.FromSeconds(Math.Pow(2, attempt)),
        onRetry: (ex, delay, attempt, _) =>
            Console.WriteLine($"  Retry {attempt}/3 after {ex.GetType().Name}")
    );

Quality gates

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

C# — 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.

 
public class DataQualityException : Exception
{
    public DataQualityException(string message) : base(message) { }
}

C# — assert DataTable is not empty with Rows.Count

Assert: Not Empty

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

 
(bool Passed, string Message) DqCheckNotEmpty(DataTable dt, string stage)
{
    bool ok = dt.Rows.Count > 0;
    return (ok, ok ? $"{stage}: {dt.Rows.Count} rows" : $"{stage}: EMPTY DataTable");
}

C# — assert no nulls in key columns with DBNull check

Assert: No Null Keys

Checks each key column individually, reports first failure found.

 
(bool Passed, string Message) DqCheckNoNullKeys(DataTable dt, IEnumerable<string> keys, string stage)
{
    foreach (var col in keys)
    {
        if (!dt.Columns.Contains(col))
            return (false, $"{stage}: column '{col}' missing");
        int nulls = dt.AsEnumerable().Count(r => r[col] is DBNull || r[col] == null);
        if (nulls > 0)
            return (false, $"{stage}: {nulls} nulls in '{col}'");
    }
    return (true, $"{stage}: no null keys in [{string.Join(", ", keys)}]");
}

C# — assert no duplicate rows with GroupBy()

Assert: No Duplicates

Compares total rows vs unique key combinations to detect duplicates.

 
(bool Passed, string Message) DqCheckNoDuplicates(DataTable dt, IEnumerable<string> keys, string stage)
{
    int total = dt.Rows.Count;
    int unique = dt.AsEnumerable()
        .GroupBy(r => string.Join("|", keys.Select(k => r[k]?.ToString() ?? "")))
        .Count();
    int dupes = total - unique;
    bool ok = dupes == 0;
    return (ok, ok ? $"{stage}: no duplicates" : $"{stage}: {dupes} duplicates on [{string.Join(", ", keys)}]");
}

C# — assert values within range with Where()

Assert: Value Range

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

 
(bool Passed, string Message) DqCheckRange(DataTable dt, string col, double minVal, double maxVal, string stage)
{
    int outOfRange = dt.AsEnumerable()
        .Where(r => !(r[col] is DBNull))
        .Count(r => Convert.ToDouble(r[col]) < minVal || Convert.ToDouble(r[col]) > maxVal);
    bool ok = outOfRange == 0;
    return (ok, ok
        ? $"{stage}: '{col}' within range"
        : $"{stage}: {outOfRange} values outside [{minVal}, {maxVal}] in '{col}'");
}

C# — assert data freshness against SLA with Max()

Assert: Data Freshness

Detects stale data that missed recent trading days.

 
(bool Passed, string Message) DqCheckFreshness(DataTable dt, string dateCol, int maxAgeDays, string stage)
{
    if (dt.Rows.Count == 0)
        return (false, $"{stage}: empty DataTable, can't check freshness");
    var dates = dt.AsEnumerable()
        .Where(r => !(r[dateCol] is DBNull))
        .Select(r => Convert.ToDateTime(r[dateCol]))
        .ToList();
    if (dates.Count == 0)
        return (false, $"{stage}: all dates are null");
    var latest = dates.Max();
    int age = (DateTime.Today - latest.Date).Days;
    bool ok = age <= maxAgeDays;
    return (ok, $"{stage}: latest date {latest:yyyy-MM-dd} ({age}d ago)" + (ok ? "" : $" EXCEEDS {maxAgeDays}d SLA"));
}

C# — assert minimum row count with Rows.Count

Assert: Minimum Row Count

Catches partial loads or missing symbols.

 
(bool Passed, string Message) DqCheckRowCount(DataTable dt, int minRows, string stage)
{
    bool ok = dt.Rows.Count >= minRows;
    return (ok, $"{stage}: {dt.Rows.Count} rows" + (ok ? "" : $" BELOW minimum {minRows}"));
}

Pipeline — run all quality gate assertions with Console.WriteLine()

Quality Gate Runner

Executes all checks for a stage, logs PASS/FAIL for each. : raises exception on first failure, blocking downstream stages.

 
DataTable RunQualityGate(IEnumerable<(bool Passed, string Message)> checks, string stage, bool failFast = true)
{
    var dt = new DataTable();
    dt.Columns.Add("check", typeof(string));
    dt.Columns.Add("status", typeof(string));
    bool allPassed = true;
 
    foreach (var (passed, msg) in checks)
    {
        var status = passed ? "PASS" : "FAIL";
        Console.WriteLine($"  DQ {status}: {msg}");
        dt.Rows.Add(msg, status);
        if (!passed) allPassed = false;
    }
 
    if (!allPassed && failFast)
        throw new DataQualityException($"Data quality gate FAILED for {stage}");
 
    return dt;
}

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.

With Trading Calendar

dim_calendar classifies each zero-volume day as a known holiday or a genuine anomaly at ingestion time. Engineers see pre-classified context warnings — holiday days are acknowledged automatically, only true anomalies trigger alerts.

Symbol metadata (SCD2)

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

HttpClient — fetch symbol metadata to JSON landing zone with GetStringAsync()

Fetch Symbol Metadata

Fetches company metadata from Yahoo Finance v8 API for each symbol. Saves raw API response to for replay.

 
string[] SCD2_COMPARE_COLS = { "company_name", "sector", "industry", "country", "exchange", "currency" };
 
async Task<string> FetchSymbolsToLanding(IEnumerable<string> symbols)
{
    var client = new HttpClient();
    client.DefaultRequestHeaders.Add("User-Agent", "Mozilla/5.0");
    var records = new List<Dictionary<string, object>>();
 
    foreach (var symbol in symbols)
    {
        try
        {
            var url = $"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}?interval=1d&range=1d";
            var json = await retryPolicy.ExecuteAsync(() => client.GetStringAsync(url));
            var doc = JsonDocument.Parse(json);
            var meta = doc.RootElement
                .GetProperty("chart").GetProperty("result")[0]
                .GetProperty("meta");
 
            var rec = new Dictionary<string, object>
            {
                ["symbol"] = symbol,
                ["longName"] = meta.TryGetProperty("longName", out var ln) ? ln.GetString() : null,
                ["shortName"] = meta.TryGetProperty("shortName", out var sn) ? sn.GetString() : null,
                ["sector"] = null,
                ["sectorKey"] = null,
                ["industry"] = null,
                ["industryKey"] = null,
                ["country"] = null,
                ["city"] = null,
                ["website"] = null,
                ["longBusinessSummary"] = null,
                ["exchange"] = meta.TryGetProperty("exchangeName", out var ex) ? ex.GetString() : null,
                ["fullExchangeName"] = meta.TryGetProperty("fullExchangeName", out var fen) ? fen.GetString() : null,
                ["exchangeTimezoneName"] = meta.TryGetProperty("exchangeTimezoneName", out var etz) ? etz.GetString() : null,
                ["exchangeTimezoneShortName"] = meta.TryGetProperty("exchangeTimezoneShortName", out var etsn) ? etsn.GetString() : null,
                ["currency"] = meta.TryGetProperty("currency", out var cur) ? cur.GetString() : null,
                ["financialCurrency"] = meta.TryGetProperty("financialCurrency", out var fc) ? fc.GetString() : null,
                ["quoteType"] = meta.TryGetProperty("instrumentType", out var qt) ? qt.GetString() : null,
                ["market"] = meta.TryGetProperty("exchangeName", out var mk) ? mk.GetString() : null,
                ["marketCap"] = null,
            };
            records.Add(rec);
            Console.WriteLine($"  {symbol}: fetched ({rec["longName"]})");
        }
        catch (Exception e)
        {
            Console.WriteLine($"  {symbol}: fetch failed ({e.Message}), using empty info");
            records.Add(new Dictionary<string, object> { ["symbol"] = symbol });
        }
    }
 
    var landingPath = Path.Combine(LANDING_DIR, "dim_symbol.json");
    File.WriteAllText(landingPath, JsonSerializer.Serialize(records, new JsonSerializerOptions { WriteIndented = true }));
    Console.WriteLine($"Landed: {landingPath} ({records.Count} symbols)");
    return landingPath;
}

JSON — load symbol metadata from landing zone with JsonSerializer.Deserialize()

Load Symbols from Landing

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

 
List<Dictionary<string, object>> LoadSymbolsFromLanding()
{
    var landingPath = Path.Combine(LANDING_DIR, "dim_symbol.json");
    var json = File.ReadAllText(landingPath);
    var docs = JsonDocument.Parse(json).RootElement;
    var results = new List<Dictionary<string, object>>();
    foreach (var el in docs.EnumerateArray())
    {
        var dict = new Dictionary<string, object>();
        foreach (var prop in el.EnumerateObject())
            dict[prop.Name] = prop.Value.ValueKind == JsonValueKind.Null
                ? null : prop.Value.ToString();
        results.Add(dict);
    }
    return results;
}

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

SCD Type 2 Dimension Upsert

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

 
string Scd2UpsertSymbol(Dictionary<string, object> rec)
{
    string Val(string key) => rec.TryGetValue(key, out var v) ? v?.ToString() : null;
    string sector = Val("sector");
    string industry = Val("industry");
 
    var dbRec = new {
        symbol = Val("symbol"),
        company_name = Val("longName") ?? Val("shortName"),
        short_name = Val("shortName"),
        sector = sector,
        sector_key = sector?.ToLower().Replace(" ", "_"),
        industry = industry,
        industry_key = industry?.ToLower().Replace(" ", "_"),
        country = Val("country"),
        city = Val("city"),
        exchange = Val("exchange"),
        full_exchange_name = Val("fullExchangeName"),
        currency = Val("currency"),
        market_cap = Val("marketCap") != null ? (long?)long.Parse(Val("marketCap")) : null,
        website = Val("website"),
    };
 
    // Check current record
    var existing = sqlConn.QueryFirstOrDefault<dynamic>(
        @"SELECT id, company_name, sector, industry, country, exchange, currency
          FROM dim_symbol WHERE symbol = @symbol AND is_current = 1",
        new { dbRec.symbol });
 
    if (existing == null)
    {
        sqlConn.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 (@symbol, @company_name, @short_name, @sector, @sector_key,
                      @industry, @industry_key, @country, @city, @exchange,
                      @full_exchange_name, @currency, @market_cap, @website)",
            dbRec);
        return "INSERT";
    }
 
    // Compare tracked columns
    var oldVals = (existing.company_name?.ToString(), existing.sector?.ToString(),
                   existing.industry?.ToString(), existing.country?.ToString(),
                   existing.exchange?.ToString(), existing.currency?.ToString());
    var newVals = (dbRec.company_name, dbRec.sector, dbRec.industry,
                   dbRec.country, dbRec.exchange, dbRec.currency);
 
    if (oldVals == newVals)
        return "UNCHANGED";
 
    // Attribute changed \u2192 close old record, insert new
    sqlConn.Execute(
        "UPDATE dim_symbol SET valid_to = SYSUTCDATETIME(), is_current = 0 WHERE id = @id",
        new { id = (int)existing.id });
    sqlConn.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 (@symbol, @company_name, @short_name, @sector, @sector_key,
                  @industry, @industry_key, @country, @city, @exchange,
                  @full_exchange_name, @currency, @market_cap, @website)",
        dbRec);
    return "SCD2_UPDATE";
}

SQL Server — orchestrate SCD Type 2 upsert for all symbols with Execute()

SCD2 Upsert Orchestration

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

 
DataTable PopulateDimSymbolFromLanding()
{
    var records = LoadSymbolsFromLanding();
    var dt = new DataTable();
    dt.Columns.Add("symbol", typeof(string));
    dt.Columns.Add("longName", typeof(string));
    dt.Columns.Add("sector", typeof(string));
    dt.Columns.Add("country", typeof(string));
    dt.Columns.Add("exchange", typeof(string));
    dt.Columns.Add("_action", typeof(string));
 
    foreach (var rec in records)
    {
        string Val(string key) => rec.TryGetValue(key, out var v) ? v?.ToString() : null;
        var action = Scd2UpsertSymbol(rec);
        dt.Rows.Add(Val("symbol"), Val("longName"), Val("sector"),
                    Val("country"), Val("exchange"), action);
        Console.WriteLine($"  {Val("symbol")}: {action}");
    }
    return dt;
}

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

// Step 1: Fetch from Yahoo Finance API \u2192 JSON landing zone
await FetchSymbolsToLanding(SYMBOLS);
 
// Step 2: Load from landing JSON \u2192 SCD2 upsert into dim_symbol
Console.WriteLine("\nStep 2: SCD2 upsert from landing zone...\n");
var dimSymbolDt = PopulateDimSymbolFromLanding();
dimSymbolDt.AsEnumerable().Take(5).CopyToDataTable()
Step 1: Fetching symbol metadata to landing zone...

  SAP.DE: fetched (SAP SE)
  SIE.DE: fetched (Siemens Aktiengesellschaft)
  ALV.DE: fetched (Allianz SE)
  DTE.DE: fetched (Deutsche Telekom AG)
  BAS.DE: fetched (BASF SE)
Landed: C:\Users\aperi\DEV\LANG\data\pipeline\landing\dim_symbol.json (5 symbols)

Step 2: SCD2 upsert from landing zone...

  SAP.DE: UNCHANGED
  SIE.DE: UNCHANGED
  ALV.DE: UNCHANGED
  DTE.DE: UNCHANGED
  BAS.DE: UNCHANGED
symbollongNamesectorcountryexchange_action
SAP.DESAP SEGERUNCHANGED
SIE.DESiemens AktiengesellschaftGERUNCHANGED
ALV.DEAllianz SEGERUNCHANGED
DTE.DEDeutsche Telekom AGGERUNCHANGED
BAS.DEBASF SEGERUNCHANGED

Trading calendar

Build and verify a per-exchange trading calendar, including holiday detection, using the dimension table populated by the Python notebook.

SQL Server — verify trading calendar exists with QueryToTable()

Read Existing Calendar

C# reads dim_calendar populated by the Python notebook rather than regenerating.

 
DataTable calDt = QueryToTable(
    @"SELECT exchange_code, COUNT(*) as total_days,
             SUM(CAST(is_trading_day AS INT)) as trading_days,
             MIN(date) as first_date, MAX(date) as last_date
      FROM dim_calendar GROUP BY exchange_code ORDER BY exchange_code");
 
calDt.AsEnumerable().Take(5).CopyToDataTable()
Calendar dimension (from Python notebook):
exchange_codetotal_daystrading_daysfirst_datelast_date
GER73250628-Mar-24 0:00:0029-Mar-26 0:00:00

SQL Server — persist calendar dimension with MERGE INTO

// Calendar already populated by Python notebook \u2014 verify row count
 
int calRows = sqlConn.ExecuteScalar<int>("SELECT COUNT(*) FROM dim_calendar");
Console.WriteLine($"dim_calendar: {calRows} rows verified");
dim_calendar: 732 rows verified

SQL Server — display detected exchange holidays with QueryToTable()

// Display holidays detected (weekdays marked as non-trading)
 
DataTable holidays = QueryToTable(
    @"SELECT TOP 20 date, exchange_code, day_of_week
      FROM dim_calendar
      WHERE day_of_week BETWEEN 1 AND 5 AND is_trading_day = 0
      ORDER BY exchange_code, date");
 
int holidayCount = sqlConn.ExecuteScalar<int>(
    @"SELECT COUNT(*) FROM dim_calendar
      WHERE day_of_week BETWEEN 1 AND 5 AND is_trading_day = 0");
Console.WriteLine($"Holidays detected: {holidayCount} (weekdays with no trading)");
holidays
Holidays detected: 16 (weekdays with no trading)
dateexchange_codeday_of_week
29-Mar-24 0:00:00GER5
01-Apr-24 0:00:00GER1
01-May-24 0:00:00GER3
24-Dec-24 0:00:00GER2
25-Dec-24 0:00:00GER3
26-Dec-24 0:00:00GER4
31-Dec-24 0:00:00GER2
01-Jan-25 0:00:00GER3
18-Apr-25 0:00:00GER5
21-Apr-25 0:00:00GER1
01-May-25 0:00:00GER4
24-Dec-25 0:00:00GER3
25-Dec-25 0:00:00GER4
26-Dec-25 0:00:00GER5
31-Dec-25 0:00:00GER3
01-Jan-26 0:00:00GER4

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.

With Landing Zone

Raw API responses are written to disk before any parsing or database operations. If the MERGE fails, re-run the load step against the saved files — no API call, no rate-limit risk, no missing symbols. The landing zone decouples the unreliable (API) from the recoverable (database).

Data fetching

Functions to download OHLCV data from the Yahoo Finance API to the JSON landing zone and reload it into a DataTable.

HttpClient — fetch OHLCV to JSON landing zone with GetStringAsync()

Landing Zone: OHLCV Fetch

Downloads OHLCV data from Yahoo Finance API and saves to . Each symbol gets its own file.

 
HttpClient httpClient = new HttpClient();
httpClient.DefaultRequestHeaders.Add("User-Agent", "Mozilla/5.0");
 
async Task<string> FetchOhlcvToLanding(string symbol, string start, string end)
{
    long period1 = new DateTimeOffset(DateTime.Parse(start)).ToUnixTimeSeconds();
    long period2 = new DateTimeOffset(DateTime.Parse(end)).ToUnixTimeSeconds();
    string url = $"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}?period1={period1}&period2={period2}&interval=1d";
 
    string json = await retryPolicy.ExecuteAsync(() => httpClient.GetStringAsync(url));
 
    var safeSymbol = symbol.Replace(".", "_");
    var landingPath = Path.Combine(LANDING_DIR, $"ohlcv_{safeSymbol}.json");
 
    // Parse v8 response and extract OHLCV records
    using var doc = JsonDocument.Parse(json);
    var result = doc.RootElement.GetProperty("chart").GetProperty("result")[0];
    var timestamps = result.GetProperty("timestamp");
    var quote = result.GetProperty("indicators").GetProperty("quote")[0];
    var adjClose = result.GetProperty("indicators").GetProperty("adjclose")[0].GetProperty("adjclose");
 
    var records = new List<Dictionary<string, object>>();
    for (int i = 0; i < timestamps.GetArrayLength(); i++)
    {
        var ts = DateTimeOffset.FromUnixTimeSeconds(timestamps[i].GetInt64()).DateTime;
        records.Add(new Dictionary<string, object> {
            ["symbol"] = symbol,
            ["date"] = ts.ToString("yyyy-MM-dd"),
            ["open"] = quote.GetProperty("open")[i].GetDouble(),
            ["high"] = quote.GetProperty("high")[i].GetDouble(),
            ["low"] = quote.GetProperty("low")[i].GetDouble(),
            ["close"] = quote.GetProperty("close")[i].GetDouble(),
            ["adj_close"] = adjClose[i].GetDouble(),
            ["volume"] = quote.GetProperty("volume")[i].GetInt64(),
            ["dividends"] = 0.0,
            ["stock_splits"] = 0.0,
        });
    }
 
    File.WriteAllText(landingPath, JsonSerializer.Serialize(records, new JsonSerializerOptions { WriteIndented = true }));
    return landingPath;
}

C# — load OHLCV from JSON landing zone with JsonSerializer.Deserialize()

Load OHLCV from Landing

Reads a symbol JSON landing file into a DataTable. Casts date strings to DateTime.

 
DataTable LoadOhlcvFromLanding(string symbol)
{
    var safeSymbol = symbol.Replace(".", "_");
    var landingPath = Path.Combine(LANDING_DIR, $"ohlcv_{safeSymbol}.json");
    if (!File.Exists(landingPath)) return new DataTable();
 
    var json = File.ReadAllText(landingPath);
    var records = JsonSerializer.Deserialize<List<Dictionary<string, JsonElement>>>(json);
    if (records == null || records.Count == 0) return new DataTable();
 
    var dt = new DataTable();
    dt.Columns.Add("symbol", typeof(string));
    dt.Columns.Add("date", typeof(DateTime));
    dt.Columns.Add("open", typeof(double));
    dt.Columns.Add("high", typeof(double));
    dt.Columns.Add("low", typeof(double));
    dt.Columns.Add("close", typeof(double));
    dt.Columns.Add("adj_close", typeof(double));
    dt.Columns.Add("volume", typeof(long));
    dt.Columns.Add("dividends", typeof(double));
    dt.Columns.Add("stock_splits", typeof(double));
 
    foreach (var rec in records)
    {
        dt.Rows.Add(
            rec["symbol"].GetString(),
            DateTime.Parse(rec["date"].GetString()),
            rec["open"].GetDouble(),
            rec["high"].GetDouble(),
            rec["low"].GetDouble(),
            rec["close"].GetDouble(),
            rec["adj_close"].GetDouble(),
            rec["volume"].GetInt64(),
            rec.ContainsKey("dividends") ? rec["dividends"].GetDouble() : 0.0,
            rec.ContainsKey("stock_splits") ? rec["stock_splits"].GetDouble() : 0.0
        );
    }
    return dt;
}

HttpClient — test single symbol landing zone fetch with FetchOhlcvToLanding()

Verifies the landing zone pattern: fetch → JSON → load → DataTable.

 
var testPath = await FetchOhlcvToLanding("SAP.DE", "2024-06-01", "2024-06-30");
if (testPath != null)
    Console.WriteLine($"Landed: {Path.GetFileName(testPath)} ({new FileInfo(testPath).Length / 1024.0:F1} KB)");
 
var testDt = LoadOhlcvFromLanding("SAP.DE");
Console.WriteLine($"Loaded: {testDt.Rows.Count} rows, columns: {string.Join(", ", testDt.Columns.Cast<DataColumn>().Select(c => c.ColumnName))}");
testDt.AsEnumerable().Take(5).CopyToDataTable()
Landed: ohlcv_SAP_DE.json (5.7 KB)
Loaded: 20 rows, columns: symbol, date, open, high, low, close, adj_close, volume, dividends, stock_splits
symboldateopenhighlowcloseadj_closevolumedividendsstock_splits
SAP.DE03-Jun-24 0:00:00169.74000549316406169.82000732421875166.9600067138672168.25999450683594166.7528076171875153172800
SAP.DE04-Jun-24 0:00:00168.52000427246094170.44000244140625167.66000366210938168.60000610351562167.0897674560547159207100
SAP.DE05-Jun-24 0:00:00170171.82000732421875169.0800018310547171.52000427246094169.98361206054688135291600
SAP.DE06-Jun-24 0:00:00176.02000427246094180.24000549316406176177.72000122070312176.12806701660156208954900
SAP.DE07-Jun-24 0:00:00177.5178.25999450683594175.6999969482422177.36000061035156175.77130126953125122486300

Validation

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

FluentValidation — validate Bronze rows with Validate() row-level check

Bronze Row-Level Validation

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

 
(DataTable Valid, int Rejected) ValidateBronze(DataTable dt, string batchId = "")
{
    var validator = new RawOhlcvValidator();
    var valid = dt.Clone();
    int rejected = 0;
 
    foreach (DataRow row in dt.Rows)
    {
        try
        {
            var record = new RawOhlcv(
                row["symbol"].ToString(), (DateTime)row["date"],
                Convert.ToDouble(row["open"]), Convert.ToDouble(row["high"]),
                Convert.ToDouble(row["low"]), Convert.ToDouble(row["close"]),
                Convert.ToDouble(row["adj_close"]), Convert.ToInt64(row["volume"]),
                Convert.ToDouble(row["dividends"]), Convert.ToDouble(row["stock_splits"]));
            var result = validator.Validate(record);
            if (!result.IsValid) throw new Exception(string.Join("; ", result.Errors));
            valid.ImportRow(row);
        }
        catch (Exception ex)
        {
            rejected++;
            if (!string.IsNullOrEmpty(batchId))
                QuarantineRow(batchId, "bronze",
                    dt.Columns.Cast<DataColumn>().ToDictionary(c => c.ColumnName, c => row[c] as object),
                    ex.Message);
        }
    }
    return (valid, rejected);
}

FluentValidation — test Bronze validation on sample data

Should pass all rows since Yahoo Finance data is generally clean.

 
var (validDt, rejectedCount) = ValidateBronze(testDt, "test");
Console.WriteLine($"Valid: {validDt.Rows.Count} rows | Rejected: {rejectedCount} rows");
var preview = validDt.AsEnumerable().Take(5).CopyToDataTable();
preview
Valid: 20 rows | Rejected: 0 rows
symboldateopenhighlowcloseadj_closevolumedividendsstock_splits
SAP.DE03-Jun-24 0:00:00169.74000549316406169.82000732421875166.9600067138672168.25999450683594166.7528076171875153172800
SAP.DE04-Jun-24 0:00:00168.52000427246094170.44000244140625167.66000366210938168.60000610351562167.0897674560547159207100
SAP.DE05-Jun-24 0:00:00170171.82000732421875169.0800018310547171.52000427246094169.98361206054688135291600
SAP.DE06-Jun-24 0:00:00176.02000427246094180.24000549316406176177.72000122070312176.12806701660156208954900
SAP.DE07-Jun-24 0:00:00177.5178.25999450683594175.6999969482422177.36000061035156175.77130126953125122486300

Ingestion pipeline

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

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

Bronze Ingestion Pipeline

Three-step process: land → validate → MERGE upsert. Checks last known date per symbol, fetches only new data, validates with FluentValidation, MERGE upserts to SQL Server.

 
async Task<(DataTable, StageLineage)> IngestBronze(string[] symbols, string start, string end, string batchId)
{
    var stageCtx = StartStage(batchId, "bronze", 0);
    int totalFetched = 0;
    int totalRejected = 0;
    var allRows = new DataTable();
 
    foreach (var symbol in symbols)
    {
        // Check last known date in SQL Server
        var lastDate = sqlConn.ExecuteScalar<DateTime?>(
            "SELECT MAX(date) FROM bronze_ohlcv WHERE symbol = @symbol",
            new { symbol });
 
        // Determine fetch range
        string fetchStart;
        if (lastDate.HasValue)
        {
            fetchStart = lastDate.Value.AddDays(1).ToString("yyyy-MM-dd");
            if (string.Compare(fetchStart, end) >= 0)
            {
                Console.WriteLine($"  {symbol}: up to date (last: {lastDate.Value:yyyy-MM-dd})");
                continue;
            }
        }
        else
        {
            fetchStart = start;
        }
 
        // Step 1: Fetch from Yahoo Finance → JSON landing zone
        string landingPath;
        try
        {
            landingPath = await FetchOhlcvToLanding(symbol, fetchStart, end);
        }
        catch
        {
            Console.WriteLine($"  {symbol}: no new data from {fetchStart}");
            continue;
        }
 
        // Step 2: Load from landing zone
        var rawDt = LoadOhlcvFromLanding(symbol);
        totalFetched += rawDt.Rows.Count;
 
        // Step 3: Validate through FluentValidation
        var (validDt, rejected) = ValidateBronze(rawDt, batchId);
        totalRejected += rejected;
 
        if (validDt.Rows.Count > 0)
        {
            if (allRows.Columns.Count == 0) allRows = validDt.Clone();
            foreach (DataRow r in validDt.Rows) allRows.ImportRow(r);
 
            // MERGE upsert into SQL Server
            int merged = MergeBronze(validDt, batchId);
            double sizeKb = new FileInfo(landingPath).Length / 1024.0;
            Console.WriteLine($"  {symbol}: {rawDt.Rows.Count} landed ({sizeKb:F1} KB), "
                + $"{validDt.Rows.Count} valid, {rejected} rejected, {merged} merged");
        }
    }
 
    // Build lineage via StartStage/EndStage
    stageCtx["input_rows"] = totalFetched;
    var emptyDt = allRows.Rows.Count > 0 ? allRows : new DataTable();
    if (emptyDt.Columns.Count == 0) { emptyDt.Columns.Add("_"); emptyDt.Rows.Add("_"); }
    var lineage = EndStage(stageCtx, emptyDt, totalRejected);
    PersistLineage(lineage);
 
    // Return FULL bronze dataset for downstream stages
    var bronzeFull = QueryToTable(
        "SELECT symbol, date, [open], high, low, [close], "
        + "adj_close, volume, dividends, stock_splits, batch_id "
        + "FROM bronze_ohlcv ORDER BY symbol, date");
 
    return (bronzeFull, lineage);
}

Bronze — execute incremental ingestion for all symbols

Bronze Execution with Context

Creates (scheduled, T-1) and , initializes , then runs the Bronze ingestion pipeline.

 
var batchId = Guid.NewGuid().ToString();
 
var bizCtx = new BusinessContext {
    Trigger = "scheduled",
    BusinessDate = DateTime.Today.AddDays(-1),
};
var tempCtx = new TemporalContext {
    AsOfDate = DateTime.Today.AddDays(-1),
    ReportingPeriodStart = DateTime.Parse(START_DATE),
    ReportingPeriodEnd = DateTime.Parse(END_DATE),
    Timezone = "CET",
};
var bronzeStageCtx = new StageContext {
    BatchId = batchId, Stage = "bronze",
    ColumnCtx = BRONZE_COLUMNS,
    BizContext = bizCtx, TempContext = tempCtx,
};
 
Console.WriteLine($"Pipeline batch_id: {batchId.Substring(0, 8)}...");
Console.WriteLine($"Range: {START_DATE} → {END_DATE}");
 
var sw = System.Diagnostics.Stopwatch.StartNew();
var (bronzeDt, bronzeLineage) = await IngestBronze(SYMBOLS, START_DATE, END_DATE, batchId);
sw.Stop();
 
// Detect zero-volume rows and classify using trading calendar
var zeroVol = bronzeDt.AsEnumerable()
    .Where(r => Convert.ToInt64(r["volume"]) == 0)
    .Select(r => new { Symbol = r["symbol"].ToString(), Date = (DateTime)r["date"] })
    .Distinct().ToList();
 
if (zeroVol.Count > 0)
{
    foreach (var row in zeroVol)
    {
        var cal = QueryToTable(
            Console.WriteLine($"SELECT is_trading_day FROM dim_calendar ");
            + $"WHERE date = '{row.Date:yyyy-MM-dd}' AND exchange_code = 'XETR'");
        if (cal.Rows.Count > 0 && !Convert.ToBoolean(cal.Rows[0]["is_trading_day"]))
            bronzeStageCtx.AddWarning(
                $"{row.Symbol}: zero volume on {row.Date:yyyy-MM-dd} — non-trading day (calendar)");
        else if (cal.Rows.Count > 0 && Convert.ToBoolean(cal.Rows[0]["is_trading_day"]))
            bronzeStageCtx.AddWarning(
                $"{row.Symbol}: zero volume on {row.Date:yyyy-MM-dd} — TRADING DAY (anomaly)");
    }
}
 
PersistContext(bronzeStageCtx);
var silverStageCtx = bronzeStageCtx.ForNextStage("silver", bronzeLineage, SILVER_COLUMNS);
Console.WriteLine($"Bronze complete: {bronzeDt.Rows.Count} rows in {sw.ElapsedMilliseconds}ms");
Pipeline batch_id: d99b77ff...
Range: 2024-03-30 → 2026-03-30
  SAP.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
  SIE.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
  ALV.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
  DTE.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
  BAS.DE: 1 landed (0.3 KB), 1 valid, 0 rejected, 1 merged
Bronze complete: 2530 rows in 583ms

SQL Server — display Bronze sample data with QueryToTable()

// Show first rows of ingested data to verify schema and values
 
QueryToTable("SELECT TOP 5 * FROM bronze_ohlcv ORDER BY symbol, date")
idsymboldateopenhighlowcloseadj_closevolumedividendsstock_splitsbatch_idingested_at
1013ALV.DE28-Mar-24 0:00:00277278.1000061035156276.45001220703125277.79998779296875252.8253936767578919173009c135c08-937d-4413-8bb4-1b66407ed9a528-Mar-26 23:21:34
1014ALV.DE02-Apr-24 0:00:00278.20001220703125280272.20001220703125273.8999938964844249.27600097656251013176009c135c08-937d-4413-8bb4-1b66407ed9a528-Mar-26 23:21:34
1015ALV.DE03-Apr-24 0:00:00274.5276.6000061035156273.8999938964844274.3999938964844249.73106384277344782102009c135c08-937d-4413-8bb4-1b66407ed9a528-Mar-26 23:21:34
1016ALV.DE04-Apr-24 0:00:00274.1000061035156275.20001220703125272.20001220703125272.3999938964844247.91087341308594690551009c135c08-937d-4413-8bb4-1b66407ed9a528-Mar-26 23:21:34
1017ALV.DE05-Apr-24 0:00:00270270.20001220703125267.1000061035156268.79998779296875244.63449096679688930874009c135c08-937d-4413-8bb4-1b66407ed9a528-Mar-26 23:21:34

SQL Server — display Bronze row counts per symbol with GROUP BY

Verifies all symbols were ingested with reasonable row counts.

 
QueryToTable(@"SELECT symbol, COUNT(*) as rows, MIN(date) as first_date, MAX(date) as last_date
    FROM bronze_ohlcv GROUP BY symbol ORDER BY symbol")
symbolrowsfirst_datelast_date
ALV.DE50628-Mar-24 0:00:0027-Mar-26 0:00:00
BAS.DE50628-Mar-24 0:00:0027-Mar-26 0:00:00
DTE.DE50628-Mar-24 0:00:0027-Mar-26 0:00:00
SAP.DE50628-Mar-24 0:00:0027-Mar-26 0:00:00
SIE.DE50628-Mar-24 0:00:0027-Mar-26 0:00:00

Pipeline — run Bronze data quality gate with RunQualityGate()

Bronze Quality Gate

All checks must pass before Silver processing begins.

 
var bronzeDq = RunQualityGate(new[] {
    DqCheckNotEmpty(bronzeDt, "bronze"),
    DqCheckNoNullKeys(bronzeDt, new[] { "symbol", "date" }, "bronze"),
    DqCheckNoDuplicates(bronzeDt, new[] { "symbol", "date" }, "bronze"),
    DqCheckRange(bronzeDt, "close", 0.01, 100_000, "bronze"),
    DqCheckRange(bronzeDt, "volume", 0, 10_000_000_000, "bronze"),
    DqCheckFreshness(bronzeDt, "date", 5, "bronze"),
    DqCheckRowCount(bronzeDt, SYMBOLS.Length * 200, "bronze"),
}, "bronze");
 
bronzeDq
DQ PASS: bronze: 2530 rows
  DQ PASS: bronze: no null keys in [symbol, date]
  DQ PASS: bronze: no duplicates
  DQ PASS: bronze: 'close' within range
  DQ PASS: bronze: 'volume' within range
  DQ PASS: bronze: latest date 2026-03-27 (3d ago)
  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 (3d 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.

With Pure Transforms

Each silver transform is a function: DataTable in → DataTable out, no I/O, no side effects. Unit tests pass a 10-row hardcoded DataTable and assert the output in milliseconds — no database, no network, no flakiness. The imperative shell (MERGE, lineage, context) wraps around pure transforms, never inside them.

Transform functions

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

LINQ — compute daily returns with grouped percentage change

Transform: Daily Returns

Close-to-close percentage change, partitioned by symbol. Pure function: DataTable in → DataTable out.

 
DataTable ComputeDailyReturns(DataTable dt)
{
    if (!dt.Columns.Contains("daily_return"))
        dt.Columns.Add("daily_return", typeof(double));
 
    var groups = dt.AsEnumerable().GroupBy(r => r.Field<string>("symbol"));
    foreach (var group in groups)
    {
        var rows = group.OrderBy(r => r.Field<DateTime>("date")).ToList();
        for (int i = 0; i < rows.Count; i++)
        {
            if (i == 0)
                rows[i]["daily_return"] = 0.0;
            else
            {
                double prevClose = Convert.ToDouble(rows[i - 1]["close"]);
                double currClose = Convert.ToDouble(rows[i]["close"]);
                rows[i]["daily_return"] = Math.Round((currClose - prevClose) / prevClose, 6);
            }
        }
    }
    return dt;
}
 
// Test on Bronze data
var testReturns = ComputeDailyReturns(bronzeDt.Copy());
var sapRows = testReturns.AsEnumerable()
    .Where(r => r.Field<string>("symbol") == "SAP.DE")
    .Take(5);
var preview = testReturns.Clone();
foreach (var r in sapRows) preview.ImportRow(r);
preview
symboldateopenhighlowcloseadj_closevolumedividendsstock_splitsbatch_iddaily_return
SAP.DE28-Mar-24 0:00:00181.82000732421875181.86000061035156179.10000610351562180.4600067138672176.60925292968751700825009c135c08-937d-4413-8bb4-1b66407ed9a50
SAP.DE02-Apr-24 0:00:00181181.9199981689453177.05999755859375177.05999755859375173.281799316406251838833009c135c08-937d-4413-8bb4-1b66407ed9a5-0.018841
SAP.DE03-Apr-24 0:00:00178.32000732421875179.52000427246094176.55999755859375178.22000122070312174.417053222656251501774009c135c08-937d-4413-8bb4-1b66407ed9a50.006551
SAP.DE04-Apr-24 0:00:00177.9199981689453178.4600067138672176.33999633789062178.02000427246094174.221328735351561126985009c135c08-937d-4413-8bb4-1b66407ed9a5-0.001122
SAP.DE05-Apr-24 0:00:00175.39999389648438177.9600067138672174.77999877929688177.4199981689453173.634124755859382099406009c135c08-937d-4413-8bb4-1b66407ed9a5-0.00337

LINQ — compute intraday range with (high - low) / close

Transform: Intraday Range

— normalized daily price spread. Higher values = more volatile day.

 
DataTable ComputeIntradayRange(DataTable dt)
{
    if (!dt.Columns.Contains("intraday_range"))
        dt.Columns.Add("intraday_range", typeof(double));
 
    foreach (DataRow row in dt.Rows)
    {
        double high = Convert.ToDouble(row["high"]);
        double low = Convert.ToDouble(row["low"]);
        double close = Convert.ToDouble(row["close"]);
        row["intraday_range"] = Math.Round((high - low) / close, 6);
    }
    return dt;
}
 
// Test on returns data
var testRange = ComputeIntradayRange(testReturns);
var sapRange = testRange.AsEnumerable()
    .Where(r => r.Field<string>("symbol") == "SAP.DE")
    .Take(5);
var previewRange = testRange.Clone();
foreach (var r in sapRange) previewRange.ImportRow(r);
previewRange
symboldateopenhighlowcloseadj_closevolumedividendsstock_splitsbatch_iddaily_returnintraday_range
SAP.DE28-Mar-24 0:00:00181.82000732421875181.86000061035156179.10000610351562180.4600067138672176.60925292968751700825009c135c08-937d-4413-8bb4-1b66407ed9a500.015294
SAP.DE02-Apr-24 0:00:00181181.9199981689453177.05999755859375177.05999755859375173.281799316406251838833009c135c08-937d-4413-8bb4-1b66407ed9a5-0.0188410.027448
SAP.DE03-Apr-24 0:00:00178.32000732421875179.52000427246094176.55999755859375178.22000122070312174.417053222656251501774009c135c08-937d-4413-8bb4-1b66407ed9a50.0065510.016609
SAP.DE04-Apr-24 0:00:00177.9199981689453178.4600067138672176.33999633789062178.02000427246094174.221328735351561126985009c135c08-937d-4413-8bb4-1b66407ed9a5-0.0011220.011909
SAP.DE05-Apr-24 0:00:00175.39999389648438177.9600067138672174.77999877929688177.4199981689453173.634124755859382099406009c135c08-937d-4413-8bb4-1b66407ed9a5-0.003370.017924

LINQ — compute 20-day moving average with rolling window

Transform: 20-Day SMA

Rolling mean of close price over 20-day window, per symbol. First 19 rows per symbol are NULL (insufficient data).

 
DataTable ComputeSma(DataTable dt, int window = 20)
{
    string col = $"sma_{window}";
    if (!dt.Columns.Contains(col))
        dt.Columns.Add(col, typeof(object));
 
    var groups = dt.AsEnumerable().GroupBy(r => r.Field<string>("symbol"));
    foreach (var group in groups)
    {
        var rows = group.OrderBy(r => r.Field<DateTime>("date")).ToList();
        for (int i = 0; i < rows.Count; i++)
        {
            if (i < window - 1)
                rows[i][col] = DBNull.Value;
            else
            {
                double sum = 0;
                for (int j = i - window + 1; j <= i; j++)
                    sum += Convert.ToDouble(rows[j]["close"]);
                rows[i][col] = Math.Round(sum / window, 4);
            }
        }
    }
    return dt;
}
 
// Test on range data
var testSma = ComputeSma(testRange);
var sapSma = testSma.AsEnumerable()
    .Where(r => r.Field<string>("symbol") == "SAP.DE")
    .Reverse().Take(5).Reverse();
var previewSma = testSma.Clone();
foreach (var r in sapSma) previewSma.ImportRow(r);
previewSma
symboldateopenhighlowcloseadj_closevolumedividendsstock_splitsbatch_iddaily_returnintraday_rangesma_20
SAP.DE23-Mar-26 0:00:00150.4600067138672161.52000427246094150.39999389648438153.86000061035156153.860000610351564165368009c135c08-937d-4413-8bb4-1b66407ed9a50.000260.072274166.034
SAP.DE24-Mar-26 0:00:00149.75999450683594151.0399932861328146147.6199951171875147.61999511718754380715009c135c08-937d-4413-8bb4-1b66407ed9a5-0.0405560.034142165.123
SAP.DE25-Mar-26 0:00:00148.77999877929688150.5399932861328145.36000061035156146.89999389648438146.899993896484383757697009c135c08-937d-4413-8bb4-1b66407ed9a5-0.0048770.035262164.129
SAP.DE26-Mar-26 0:00:00145.39999389648438148.0800018310547143.52000427246094144.63999938964844144.639999389648443752998009c135c08-937d-4413-8bb4-1b66407ed9a5-0.0153850.031527162.75
SAP.DE27-Mar-26 0:00:00145.74000549316406147.32000732421875142.10000610351562142.55999755859375142.55999755859375356858100d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f-0.0143810.036616161.33

C# — compose all Silver transforms with function chaining

Transform Composition Pipeline

Chains three pure functions — each is independent and unit-testable. The composed pipeline validates through FluentValidation before MERGE.

 
DataTable TransformSilver(DataTable bronzeDt)
{
    var dt = bronzeDt.Copy();
    if (dt.Columns.Contains("batch_id")) dt.Columns.Remove("batch_id");
    dt = ComputeDailyReturns(dt);
    dt = ComputeIntradayRange(dt);
    dt = ComputeSma(dt, 20);
    return dt;
}

FluentValidation — validate Silver rows with Validate() row-level check

Silver Row-Level Validation

Valid rows collected; rejected rows quarantined with error details.

 
(DataTable Valid, int Rejected) ValidateSilver(DataTable dt, string batchId)
{
    var validator = new CleanOhlcvValidator();
    var valid = dt.Clone();
    int rejected = 0;
 
    foreach (DataRow row in dt.Rows)
    {
        try
        {
            var sma20 = row["sma_20"] == DBNull.Value ? (double?)null : Convert.ToDouble(row["sma_20"]);
            var record = new CleanOhlcv(
                row["symbol"].ToString(), (DateTime)row["date"],
                Convert.ToDouble(row["open"]), Convert.ToDouble(row["high"]),
                Convert.ToDouble(row["low"]), Convert.ToDouble(row["close"]),
                Convert.ToDouble(row["adj_close"]), Convert.ToInt64(row["volume"]),
                Convert.ToDouble(row["dividends"]), Convert.ToDouble(row["stock_splits"]),
                Convert.ToDouble(row["daily_return"]),
                Convert.ToDouble(row["intraday_range"]),
                sma20, batchId);
            var result = validator.Validate(record);
            if (!result.IsValid) throw new Exception(string.Join("; ", result.Errors));
            valid.ImportRow(row);
        }
        catch (Exception ex)
        {
            rejected++;
            QuarantineRow(batchId, "silver",
                dt.Columns.Cast<DataColumn>().ToDictionary(c => c.ColumnName, c => row[c] as object),
                ex.Message);
        }
    }
    return (valid, rejected);
}

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

Orchestrates: transform (pure) → validate (FluentValidation) → MERGE (SQL) → lineage. Transforms the FULL bronze dataset (needed for correct SMA/returns).

 
(DataTable, StageLineage) ProcessSilver(DataTable bronzeDt, string batchId)
{
    var stageCtx = StartStage(batchId, "silver", bronzeDt.Rows.Count);
 
    // Apply all transforms on full bronze (SMA/returns need full history)
    var enrichedDt = TransformSilver(bronzeDt);
 
    // Validate through FluentValidation
    var (validDt, rejected) = ValidateSilver(enrichedDt, batchId);
 
    // MERGE upsert into SQL Server
    if (validDt.Rows.Count > 0)
    {
        int merged = MergeSilver(validDt, batchId);
        Console.WriteLine($"Silver: {merged} rows merged ({validDt.Rows.Count} valid, {rejected} rejected)");
    }
 
    // Complete lineage
    var lineage = EndStage(stageCtx, validDt, rejected);
    PersistLineage(lineage);
 
    // Return FULL silver dataset for Gold layer
    var silverFull = QueryToTable(
        "SELECT symbol, date, [open], high, low, [close], "
        + "adj_close, volume, dividends, stock_splits, "
        + "daily_return, intraday_range, sma_20, batch_id "
        + "FROM silver_ohlcv ORDER BY symbol, date");
 
    return (silverFull, lineage);
}

Silver — execute enrichment on full Bronze data

Silver Execution with Context

Runs the Silver enrichment, then records SMA-20 null warnings in for downstream visibility.

 
var swSilver = System.Diagnostics.Stopwatch.StartNew();
var (silverDt, silverLineage) = ProcessSilver(bronzeDt, batchId);
swSilver.Stop();
 
int smaNullCount = silverDt.AsEnumerable()
    .Count(r => r["sma_20"] == DBNull.Value || r["sma_20"] == null);
if (smaNullCount > 0)
    silverStageCtx.AddWarning($"sma_20: {smaNullCount} NULL values (first 19 rows per symbol)");
 
PersistContext(silverStageCtx);
var goldStageCtx = silverStageCtx.ForNextStage("gold", silverLineage, GOLD_DAILY_COLUMNS);
Console.WriteLine($"Silver complete: {silverDt.Rows.Count} rows in {swSilver.ElapsedMilliseconds}ms");
Silver: 2530 rows merged (2530 valid, 0 rejected)
  WARNING silver: sma_20: 95 NULL values (first 19 rows per symbol)
Silver complete: 2530 rows in 5837ms

SQL Server — display Silver enriched columns with QueryToTable()

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

 
QueryToTable(@"SELECT TOP 5 symbol, date, [close], daily_return, intraday_range, sma_20
    FROM silver_ohlcv WHERE symbol = 'SAP.DE' ORDER BY date DESC")
symboldateclosedaily_returnintraday_rangesma_20
SAP.DE27-Mar-26 0:00:00142.55999755859375-0.0143810.036616161.33
SAP.DE26-Mar-26 0:00:00144.63999938964844-0.0153850.031527162.75
SAP.DE25-Mar-26 0:00:00146.89999389648438-0.0048770.035262164.129
SAP.DE24-Mar-26 0:00:00147.6199951171875-0.0405560.034142165.123
SAP.DE23-Mar-26 0:00:00153.860000610351560.000260.072274166.034

SQL Server — display Silver statistics per symbol with GROUP BY

// Summary stats to verify enrichment quality across all symbols
 
QueryToTable(@"SELECT symbol,
    ROUND(AVG(daily_return), 6) as avg_return,
    ROUND(STDEV(daily_return), 6) as volatility,
    ROUND(AVG(intraday_range), 6) as avg_intraday,
    SUM(CASE WHEN sma_20 IS NULL THEN 1 ELSE 0 END) as sma_nulls,
    COUNT(*) as rows
    FROM silver_ohlcv GROUP BY symbol ORDER BY symbol")
symbolavg_returnvolatilityavg_intradaysma_nullsrows
ALV.DE0.0005320.0118510.01410619506
BAS.DE0.0001210.0175080.02139819506
DTE.DE0.0007650.0132630.0157219506
SAP.DE-0.0002850.0189190.02067219506
SIE.DE0.0004750.0192230.021519506

Pipeline — run Silver data quality gate with RunQualityGate()

Silver Quality Gate

Hard gate: blocks pipeline on structural issues. Soft gate: logs warnings on statistical anomalies (e.g., return outliers).

var silverDq = RunQualityGate(new[] {
    DqCheckNotEmpty(silverDt, "silver"),
    DqCheckNoNullKeys(silverDt, new[] { "symbol", "date", "daily_return" }, "silver"),
    DqCheckNoDuplicates(silverDt, new[] { "symbol", "date" }, "silver"),
    DqCheckRange(silverDt, "daily_return", -0.5, 0.5, "silver"),
    DqCheckRange(silverDt, "intraday_range", 0, 0.5, "silver"),
    DqCheckFreshness(silverDt, "date", 5, "silver"),
    DqCheckRowCount(silverDt, SYMBOLS.Length * 200, "silver"),
}, "silver");
 
// Soft checks — warn but don't block (daily return > 10% is unusual for blue chips)
var outliers = silverDt.AsEnumerable()
    .Where(r => Math.Abs(Convert.ToDouble(r["daily_return"])) > 0.10)
    .ToList();
if (outliers.Count > 0)
{
    Console.WriteLine($"  {outliers.Count} rows with |daily_return| > 10%:");
    var outlierDt = silverDt.Clone();
    foreach (var r in outliers) outlierDt.ImportRow(r);
    display(outlierDt);
}
else
{
    Console.WriteLine("  No outliers detected");
}
 
silverDq
DQ PASS: silver: 2530 rows
  DQ PASS: silver: no null keys in [symbol, date, daily_return]
  DQ PASS: silver: no duplicates
  DQ PASS: silver: 'daily_return' within range
  DQ PASS: silver: 'intraday_range' within range
  DQ PASS: silver: latest date 2026-03-27 (3d ago)
  DQ PASS: silver: 2530 rows
Outlier checks (warnings only):
  3 rows with |daily_return| > 10%:
symboldateopenhighlowcloseadj_closevolumedividendsstock_splitsdaily_returnintraday_rangesma_20batch_id
BAS.DE05-Mar-25 0:00:0050.09999847412109453.6800003051757850.0499992370605553.6599998474121150.940132141113289216640000.1070770.06764849.203d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
SAP.DE23-Apr-25 0:00:00242.39999389648438244.1999969482422236.1999969482422241.6999969482422239.534973144531253410054000.1061780.033099235.6525d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
SAP.DE29-Jan-26 0:00:00179180.16000366210938162.1199951171875164.6199951171875164.61999511718751584679100-0.1607020.109586200.193d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
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 (3d 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.

With Gold Validation

The gold contract enforces mathematical invariants on every aggregated row before persistence — max_drawdown <= 0, avg_volume >= 0, sharpe_ratio finite. A bad aggregation is rejected at the gold boundary, not discovered by a portfolio manager three days later.

Aggregation functions

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

LINQ — build daily cross-sectional summary with GroupBy()

Aggregation: Daily Summary

Groups all symbols by date: mean/max/min return, total volume, avg intraday range. One row per trading day.

 
DataTable BuildDailySummary(DataTable silverDt, string batchId)
{
    var dt = new DataTable();
    dt.Columns.Add("date", typeof(DateTime));
    dt.Columns.Add("symbols_traded", typeof(int));
    dt.Columns.Add("avg_return", typeof(double));
    dt.Columns.Add("max_return", typeof(double));
    dt.Columns.Add("min_return", typeof(double));
    dt.Columns.Add("total_volume", typeof(long));
    dt.Columns.Add("avg_intraday_pct", typeof(double));
    dt.Columns.Add("batch_id", typeof(string));
 
    var groups = silverDt.AsEnumerable()
        .GroupBy(r => r.Field<DateTime>("date"))
        .OrderBy(g => g.Key);
 
    foreach (var group in groups)
    {
        var returns = group.Select(r => Convert.ToDouble(r["daily_return"])).ToList();
        var volumes = group.Select(r => Convert.ToInt64(r["volume"])).ToList();
        var intraday = group.Select(r => Convert.ToDouble(r["intraday_range"])).ToList();
 
        dt.Rows.Add(
            group.Key,
            group.Select(r => r.Field<string>("symbol")).Distinct().Count(),
            Math.Round(returns.Average(), 6),
            returns.Max(),
            returns.Min(),
            volumes.Sum(),
            Math.Round(intraday.Average(), 6),
            batchId
        );
    }
    return dt;
}
 
var dailySummaryDt = BuildDailySummary(silverDt, batchId);
Console.WriteLine($"Daily summary: {dailySummaryDt.Rows.Count} trading days");
var previewDaily = dailySummaryDt.Clone();
foreach (DataRow r in dailySummaryDt.AsEnumerable().Take(5)) previewDaily.ImportRow(r);
previewDaily
Daily summary: 506 trading days
datesymbols_tradedavg_returnmax_returnmin_returntotal_volumeavg_intraday_pctbatch_id
28-Mar-24 0:00:005000141152410.011246d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
02-Apr-24 0:00:005-0.0062610.016815-0.018841154600600.02085d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
03-Apr-24 0:00:0050.0048620.01282-0.002239123444750.014617d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
04-Apr-24 0:00:005-0.0006310.007522-0.007289102387480.01083d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
05-Apr-24 0:00:005-0.014092-0.00337-0.02146163640360.017791d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f

LINQ — build per-symbol profile with cumulative max drawdown

Aggregation: Symbol Risk Profile

Per-symbol over full history: avg return, volatility (daily sigma), max drawdown (peak-to-trough), total dividends.

 
DataTable BuildSymbolProfile(DataTable silverDt, string batchId)
{
    var dt = new DataTable();
    dt.Columns.Add("symbol", typeof(string));
    dt.Columns.Add("total_trading_days", typeof(int));
    dt.Columns.Add("avg_daily_return", typeof(double));
    dt.Columns.Add("volatility", typeof(double));
    dt.Columns.Add("max_drawdown", typeof(double));
    dt.Columns.Add("avg_volume", typeof(double));
    dt.Columns.Add("total_dividends", typeof(double));
    dt.Columns.Add("first_date", typeof(DateTime));
    dt.Columns.Add("last_date", typeof(DateTime));
    dt.Columns.Add("batch_id", typeof(string));
 
    var groups = silverDt.AsEnumerable()
        .GroupBy(r => r.Field<string>("symbol"))
        .OrderBy(g => g.Key);
 
    foreach (var group in groups)
    {
        var rows = group.OrderBy(r => r.Field<DateTime>("date")).ToList();
        var returns = rows.Select(r => Convert.ToDouble(r["daily_return"])).ToList();
        var volumes = rows.Select(r => Convert.ToInt64(r["volume"])).ToList();
        var dividends = rows.Select(r => Convert.ToDouble(r["dividends"])).ToList();
        var closes = rows.Select(r => Convert.ToDouble(r["close"])).ToList();
        var dates = rows.Select(r => r.Field<DateTime>("date")).ToList();
 
        // Max drawdown: peak-to-trough decline using cumulative max of close
        double peak = closes[0];
        double maxDd = 0;
        foreach (var c in closes)
        {
            if (c > peak) peak = c;
            double dd = (c - peak) / peak;
            if (dd < maxDd) maxDd = dd;
        }
 
        double avgRet = returns.Average();
        double vol = Math.Sqrt(returns.Select(r => Math.Pow(r - avgRet, 2)).Average());
 
        dt.Rows.Add(
            group.Key,
            rows.Count,
            Math.Round(avgRet, 6),
            Math.Round(vol, 6),
            Math.Round(maxDd, 6),
            Math.Round(volumes.Average(), 2),
            Math.Round(dividends.Sum(), 4),
            dates.First(),
            dates.Last(),
            batchId
        );
    }
    return dt;
}
 
var symbolProfileDt = BuildSymbolProfile(silverDt, batchId);
Console.WriteLine($"Symbol profiles: {symbolProfileDt.Rows.Count} symbols");
symbolProfileDt
Symbol profiles: 5 symbols
symboltotal_trading_daysavg_daily_returnvolatilitymax_drawdownavg_volumetotal_dividendsfirst_datelast_datebatch_id
ALV.DE5060.0005320.011839-0.123504631486.1429.228-Mar-24 0:00:0027-Mar-26 0:00:00d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
BAS.DE5060.0001210.017491-0.2767662518782.145.6528-Mar-24 0:00:0027-Mar-26 0:00:00d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
DTE.DE5060.0007650.01325-0.2661096531019.061.6728-Mar-24 0:00:0027-Mar-26 0:00:00d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
SAP.DE506-0.0002850.0189-0.4914021676455.194.5528-Mar-24 0:00:0027-Mar-26 0:00:00d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
SIE.DE5060.0004750.019204-0.2732511155097.0710.5528-Mar-24 0:00:0027-Mar-26 0:00:00d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f

FluentValidation — validate Gold daily summary with row-level check

// Validates each row to catch aggregation errors before persistence
 
(DataTable Valid, int Rejected) ValidateGoldDaily(DataTable dt)
{
    var valid = dt.Clone();
    int rejected = 0;
    foreach (DataRow row in dt.Rows)
    {
        try
        {
            var record = new DailySummary(
                (DateTime)row["date"],
                Convert.ToInt32(row["symbols_traded"]),
                Convert.ToDouble(row["avg_return"]),
                Convert.ToDouble(row["max_return"]),
                Convert.ToDouble(row["min_return"]),
                Convert.ToInt64(row["total_volume"]),
                Convert.ToDouble(row["avg_intraday_pct"]),
                row["batch_id"].ToString());
            valid.ImportRow(row);
        }
        catch
        {
            rejected++;
        }
    }
    return (valid, rejected);
}
 
var (validDaily, rejDaily) = ValidateGoldDaily(dailySummaryDt);
Console.WriteLine($"Daily summary validation: {validDaily.Rows.Count} valid, {rejDaily} rejected");
Daily summary validation: 506 valid, 0 rejected

FluentValidation — validate Gold symbol profiles with row-level check

// Validates each profile to catch calculation errors
 
(DataTable Valid, int Rejected) ValidateGoldProfiles(DataTable dt)
{
    var valid = dt.Clone();
    int rejected = 0;
    foreach (DataRow row in dt.Rows)
    {
        try
        {
            var record = new SymbolProfile(
                row["symbol"].ToString(),
                Convert.ToInt32(row["total_trading_days"]),
                Convert.ToDouble(row["avg_daily_return"]),
                Convert.ToDouble(row["volatility"]),
                Convert.ToDouble(row["max_drawdown"]),
                Convert.ToDouble(row["avg_volume"]),
                Convert.ToDouble(row["total_dividends"]),
                (DateTime)row["first_date"],
                (DateTime)row["last_date"],
                row["batch_id"].ToString());
            valid.ImportRow(row);
        }
        catch (Exception ex)
        {
            rejected++;
            Console.WriteLine($"  Rejected {row["symbol"]}: {ex.Message}");
        }
    }
    return (valid, rejected);
}
 
var (validProfiles, rejProfiles) = ValidateGoldProfiles(symbolProfileDt);
Console.WriteLine($"Symbol profile validation: {validProfiles.Rows.Count} valid, {rejProfiles} rejected");
Symbol profile validation: 5 valid, 0 rejected

Persistence

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

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

Gold: Truncate and Rebuild

Gold is always a full rebuild from Silver — not incremental. TRUNCATE both Gold tables, then INSERT new aggregations.

 
StageLineage PersistGold(DataTable dailyDt, DataTable profileDt, string batchId)
{
    int totalInput = dailyDt.Rows.Count + profileDt.Rows.Count;
    var stageCtx = StartStage(batchId, "gold", totalInput);
 
    // Truncate and insert
    if (dailyDt.Rows.Count > 0)
    {
        sqlConn.Execute("TRUNCATE TABLE gold_daily_summary");
        WriteToCsv(dailyDt, "gold_daily_summary", false);
    }
    if (profileDt.Rows.Count > 0)
    {
        sqlConn.Execute("TRUNCATE TABLE gold_symbol_profile");
        WriteToCsv(profileDt, "gold_symbol_profile", false);
    }
 
    // Combined DataTable for hash
    var combined = dailyDt.Copy();
    foreach (DataRow r in profileDt.Rows)
    {
        var newRow = combined.NewRow();
        combined.Rows.Add(newRow);
    }
 
    var lineage = EndStage(stageCtx, combined, 0);
    PersistLineage(lineage);
    return lineage;
}

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

Gold Execution with Context

Persists Gold marts, checks for days with missing symbols, records warnings in .

 
var goldLineage = PersistGold(validDaily, validProfiles, batchId);
 
var missingSymbols = validDaily.AsEnumerable()
    .Where(r => Convert.ToInt32(r["symbols_traded"]) < SYMBOLS.Length)
    .ToList();
if (missingSymbols.Count > 0)
{
    foreach (var row in missingSymbols.Take(5))
    {
        goldStageCtx.AddWarning(
            $"gold: {((DateTime)row["date"]):yyyy-MM-dd} only {row["symbols_traded"]}/{SYMBOLS.Length} symbols");
    }
}
 
PersistContext(goldStageCtx);
Console.WriteLine($"Gold persisted: hash={goldLineage.OutputHash}");
Gold persisted: hash=a87ddaf98043eb7c

SQL Server — display Gold daily summary with QueryToTable()

// Show the most recent trading days with cross-sectional metrics
 
QueryToTable("SELECT TOP 5 * FROM gold_daily_summary ORDER BY date DESC")
iddatesymbols_tradedavg_returnmax_returnmin_returntotal_volumeavg_intraday_pctbatch_id
50627-Mar-26 0:00:005-0.0037680.026803-0.023123162296340.024848d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
50526-Mar-26 0:00:005-0.0060630.014394-0.015385157222940.01961d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
50425-Mar-26 0:00:0050.0080210.023951-0.004877135363250.019512d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
50324-Mar-26 0:00:0050.0038540.0418-0.040556149909440.028663d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f
50223-Mar-26 0:00:0050.0120350.037055-0.00253219007460.071276d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f

SQL Server — display Gold symbol profiles with QueryToTable()

// Final per-symbol summary statistics
 
QueryToTable(@"SELECT symbol, total_trading_days, avg_daily_return,
    volatility, max_drawdown, avg_volume, total_dividends
    FROM gold_symbol_profile ORDER BY symbol")
symboltotal_trading_daysavg_daily_returnvolatilitymax_drawdownavg_volumetotal_dividends
ALV.DE5060.0005320.011839-0.123504631486.1429.2
BAS.DE5060.0001210.017491-0.2767662518782.145.65
DTE.DE5060.0007650.01325-0.2661096531019.061.67
SAP.DE506-0.0002850.0189-0.4914021676455.194.55
SIE.DE5060.0004750.019204-0.2732511155097.0710.55

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.

With Pre-Materialization

The pipeline exports finished Parquet files as part of each run. The ASP.NET API reads those files directly — zero SQL dependency at serving time. A SQL Server restart has no impact on API availability. Deployment is a file copy, not a migration.

Export operations

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

ParquetSharp — export daily summary to Parquet with WriteDataTableToParquet()

Parquet: Pre-Materialized View

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

 
using ParquetSharp;
 
void WriteDataTableToParquet(DataTable dt, string path)
{
    // Detect which columns have DBNull values (nullable)
    var hasNulls = new HashSet<string>();
    foreach (DataColumn col in dt.Columns)
        if (dt.AsEnumerable().Any(r => r[col] is DBNull))
            hasNulls.Add(col.ColumnName);
 
    var columns = new List<Column>();
    foreach (DataColumn col in dt.Columns)
    {
        bool nullable = hasNulls.Contains(col.ColumnName);
        if (col.DataType == typeof(DateTime))
            columns.Add(nullable ? new Column<DateTime?>(col.ColumnName) : new Column<DateTime>(col.ColumnName));
        else if (col.DataType == typeof(double) || col.DataType == typeof(object))
            columns.Add(nullable ? new Column<double?>(col.ColumnName) : new Column<double>(col.ColumnName));
        else if (col.DataType == typeof(long) || col.DataType == typeof(int))
            columns.Add(nullable ? new Column<long?>(col.ColumnName) : new Column<long>(col.ColumnName));
        else
            columns.Add(new Column<string>(col.ColumnName));
    }
 
    using var file = new ParquetFileWriter(path, columns.ToArray());
    using var rowGroup = file.AppendRowGroup();
    foreach (DataColumn col in dt.Columns)
    {
        bool nullable = hasNulls.Contains(col.ColumnName);
        if (col.DataType == typeof(DateTime) && nullable)
        {
            using var w = rowGroup.NextColumn().LogicalWriter<DateTime?>();
            w.WriteBatch(dt.AsEnumerable().Select(r => r[col] is DBNull ? (DateTime?)null : (DateTime)r[col]).ToArray());
        }
        else if (col.DataType == typeof(DateTime))
        {
            using var w = rowGroup.NextColumn().LogicalWriter<DateTime>();
            w.WriteBatch(dt.AsEnumerable().Select(r => (DateTime)r[col]).ToArray());
        }
        else if ((col.DataType == typeof(double) || col.DataType == typeof(object)) && nullable)
        {
            using var w = rowGroup.NextColumn().LogicalWriter<double?>();
            w.WriteBatch(dt.AsEnumerable().Select(r => r[col] is DBNull ? (double?)null : Convert.ToDouble(r[col])).ToArray());
        }
        else if (col.DataType == typeof(double) || col.DataType == typeof(object))
        {
            using var w = rowGroup.NextColumn().LogicalWriter<double>();
            w.WriteBatch(dt.AsEnumerable().Select(r => Convert.ToDouble(r[col])).ToArray());
        }
        else if ((col.DataType == typeof(long) || col.DataType == typeof(int)) && nullable)
        {
            using var w = rowGroup.NextColumn().LogicalWriter<long?>();
            w.WriteBatch(dt.AsEnumerable().Select(r => r[col] is DBNull ? (long?)null : Convert.ToInt64(r[col])).ToArray());
        }
        else if (col.DataType == typeof(long) || col.DataType == typeof(int))
        {
            using var w = rowGroup.NextColumn().LogicalWriter<long>();
            w.WriteBatch(dt.AsEnumerable().Select(r => Convert.ToInt64(r[col])).ToArray());
        }
        else
        {
            using var w = rowGroup.NextColumn().LogicalWriter<string>();
            w.WriteBatch(dt.AsEnumerable().Select(r => r[col] is DBNull ? "" : r[col].ToString()).ToArray());
        }
    }
}
 
var dailyPath = Path.Combine(EXPORT_DIR, "gold_daily_summary.parquet");
WriteDataTableToParquet(validDaily, dailyPath);
var sizeKb = new FileInfo(dailyPath).Length / 1024.0;
Console.WriteLine($"Exported: {Path.GetFileName(dailyPath)} ({sizeKb:F1} KB, {validDaily.Rows.Count} rows)");
Exported: gold_daily_summary.parquet (26.8 KB, 506 rows)

ParquetSharp — export symbol profiles to Parquet with WriteDataTableToParquet()

// ── Pre-materialized view: per-symbol summary for the comparison dashboard ──
 
var profilePath = Path.Combine(EXPORT_DIR, "gold_symbol_profile.parquet");
WriteDataTableToParquet(validProfiles, profilePath);
var sizeKb = new FileInfo(profilePath).Length / 1024.0;
Console.WriteLine($"Exported: {Path.GetFileName(profilePath)} ({sizeKb:F1} KB, {validProfiles.Rows.Count} rows)");
Exported: gold_symbol_profile.parquet (2.6 KB, 5 rows)

ParquetSharp — export Silver data to Parquet with WriteDataTableToParquet()

Silver Parquet Export

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

 
var silverPath = Path.Combine(EXPORT_DIR, "silver_ohlcv.parquet");
WriteDataTableToParquet(silverDt, silverPath);
var sizeKb = new FileInfo(silverPath).Length / 1024.0;
Console.WriteLine($"Exported: {Path.GetFileName(silverPath)} ({sizeKb:F1} KB, {silverDt.Rows.Count} rows)");
Exported: silver_ohlcv.parquet (160.7 KB, 2530 rows)

Lineage — record export stage with EndStage()

// ── Track which files were exported and their sizes ──
 
var exportCtx = StartStage(batchId, "export", inputRows: validDaily.Rows.Count + validProfiles.Rows.Count + silverDt.Rows.Count);
 
// Combined export DataTable for hash (all exported data)
var exportCombined = new DataTable();
exportCombined.Columns.Add("data", typeof(string));
foreach (DataTable dt in new[] { validDaily, validProfiles, silverDt })
    foreach (DataRow row in dt.Rows)
        exportCombined.Rows.Add(string.Join("|", row.ItemArray.Select(v => v?.ToString() ?? "")));
 
var exportLineage = EndStage(exportCtx, exportCombined, 0);
PersistLineage(exportLineage);
 
Console.WriteLine($"Export lineage recorded: {exportLineage.OutputRows} total rows, hash={exportLineage.OutputHash}");
Export lineage recorded: 3041 total rows, hash=4915a3cad5280dd8

ParquetSharp — verify exported Parquet files with ParquetFileReader

// ── Round-trip test: verify Parquet files exist and report sizes ──
 
foreach (var name in new[] { "gold_daily_summary", "gold_symbol_profile", "silver_ohlcv" })
{
    var path = Path.Combine(EXPORT_DIR, $"{name}.parquet");
    using var reader = new ParquetFileReader(path);
    var meta = reader.FileMetaData;
    Console.WriteLine($"  {name}: {meta.NumRows} rows, {meta.NumColumns} cols");
}
gold_daily_summary: 506 rows, 8 cols
  gold_symbol_profile: 5 rows, 10 cols
  silver_ohlcv: 2530 rows, 14 cols

C# — export data contracts as JSON Schema with JsonSerializer

// ── Export machine-readable contracts for every pipeline boundary ──
 
var contractPaths = ExportDataContracts(EXPORT_DIR);
foreach (var p in contractPaths)
    Console.WriteLine($"  {Path.GetFileName(p)}: {new FileInfo(p).Length:N0} bytes");
Contract exported: bronze_ohlcv_contract.json
  Contract exported: silver_ohlcv_contract.json
  Contract exported: gold_daily_summary_contract.json
  Contract exported: gold_symbol_profile_contract.json
  bronze_ohlcv_contract.json: 4'279 bytes
  silver_ohlcv_contract.json: 5'861 bytes
  gold_daily_summary_contract.json: 3'080 bytes
  gold_symbol_profile_contract.json: 4'074 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.

C# — build and save run context with RunContext

Finalize RunContext

Combines stage lineage, business context, temporal context, and data warnings into the final execution record. Persisted as JSON.

 
var runContext = new RunContext {
    BatchId = batchId,
    StartedAt = bronzeLineage.StartedAt,
    CompletedAt = exportLineage.CompletedAt,
    Symbols = SYMBOLS,
    DateRange = new[] { START_DATE, END_DATE },
    Stages = new List<StageLineage> { bronzeLineage, silverLineage, goldLineage, exportLineage },
    Status = "completed",
    BizContext = bizCtx,
    TempContext = tempCtx,
    DataWarnings = goldStageCtx.DataWarnings,
    ContractVersion = "1.0",
};
 
string SaveRunContext(RunContext ctx)
{
    var path = Path.Combine(LINEAGE_DIR, $"run_{ctx.BatchId[..8]}.json");
    File.WriteAllText(path, JsonSerializer.Serialize(ctx, new JsonSerializerOptions { WriteIndented = true }));
    return path;
}
 
var ctxPath = SaveRunContext(runContext);
var totalMs = runContext.Stages.Sum(s => s.DurationMs);
Console.WriteLine($"RunContext saved: {Path.GetFileName(ctxPath)}");
Console.WriteLine($"Batch: {batchId[..8]}... | Warnings: {runContext.DataWarnings.Count}");
Console.WriteLine($"Processing time: {totalMs:F0}ms");
RunContext saved: run_d99b77ff.json
Batch: d99b77ff... | Warnings: 1
Processing time: 7219ms

C# — display lineage summary as DataTable

// ── Shows all stages with timing, row counts, and hashes ──
 
var lineageDt = new DataTable();
lineageDt.Columns.Add("stage", typeof(string));
lineageDt.Columns.Add("input_rows", typeof(int));
lineageDt.Columns.Add("output_rows", typeof(int));
lineageDt.Columns.Add("rejected", typeof(int));
lineageDt.Columns.Add("duration_ms", typeof(double));
lineageDt.Columns.Add("output_hash", typeof(string));
 
foreach (var s in runContext.Stages)
    lineageDt.Rows.Add(s.Stage, s.InputRows, s.OutputRows, s.RowsRejected, Math.Round(s.DurationMs, 1), s.OutputHash);
 
lineageDt
stageinput_rowsoutput_rowsrejectedduration_msoutput_hash
bronze550561959fbdf2bc0d14a5
silver2530253005776.976d1db80852e66dc
gold5115110877.2a87ddaf98043eb7c
export3041304104.34915a3cad5280dd8

JSON — read back persisted run context with JsonSerializer.Deserialize()

Verify RunContext JSON

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

 
var ctxJson = JsonSerializer.Deserialize<JsonElement>(File.ReadAllText(ctxPath));
 
// Display key fields without the bulky stages array
var displayCtx = new Dictionary<string, object>();
foreach (var prop in ctxJson.EnumerateObject())
{
    if (prop.Name == "Stages")
        displayCtx[prop.Name] = $"[{prop.Value.GetArrayLength()} stage records]";
    else if (prop.Value.ValueKind == JsonValueKind.Object || prop.Value.ValueKind == JsonValueKind.Array)
        displayCtx[prop.Name] = JsonSerializer.Deserialize<object>(prop.Value.GetRawText());
    else
        displayCtx[prop.Name] = prop.Value.ToString();
}
Console.WriteLine(JsonSerializer.Serialize(displayCtx, new JsonSerializerOptions { WriteIndented = true }));
{
  "BatchId": "d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f",
  "StartedAt": "2026-03-29T23:24:03.1918047Z",
  "CompletedAt": "2026-03-29T23:24:12.045971Z",
  "Symbols": [
    "SAP.DE",
    "SIE.DE",
    "ALV.DE",
    "DTE.DE",
    "BAS.DE"
  ],
  "DateRange": [
    "2024-03-30",
    "2026-03-30"
  ],
  "PolarsVersion": "Polars.NET",
  "Stages": "[4 stage records]",
  "Status": "completed",
  "BizContext": {
    "Trigger": "scheduled",
    "Reason": null,
    "BusinessDate": "2026-03-29T00:00:00\u002B01:00",
    "IsCorrection": false,
    "AffectedSymbols": null
  },
  "TempContext": {
    "AsOfDate": "2026-03-29T00:00:00\u002B01:00",
    "ReportingPeriodStart": "2024-03-30T00:00:00",
    "ReportingPeriodEnd": "2026-03-30T00:00:00",
    "KnowledgeDate": "2026-03-29T23:24:03.1903432Z",
    "Timezone": "CET",
    "IsBackfill": false
  },
  "DataWarnings": [
    "sma_20: 95 NULL values (first 19 rows per symbol)"
  ],
  "ContractVersion": "1.0"
}

SQL Server — query lineage table with QueryToTable()

// ── Verify lineage records were persisted to SQL Server ──
 
var lineageQuery = QueryToTable(
    Console.WriteLine($"SELECT stage, input_rows, output_rows, rows_rejected, output_hash FROM lineage_stages WHERE batch_id = '{batchId}'");
);
lineageQuery
stageinput_rowsoutput_rowsrows_rejectedoutput_hash
bronze550959fbdf2bc0d14a5
silver25302530076d1db80852e66dc
gold5115110a87ddaf98043eb7c
export3041304104915a3cad5280dd8

SQL Server — review quarantined rows with QueryToTable()

Review Quarantined Rows

Shows rejected rows with error messages for investigation.

 
var quarantineDt = QueryToTable(
    Console.WriteLine($"SELECT stage, symbol, date, error_message, quarantined_at FROM quarantine WHERE batch_id = '{batchId}' ORDER BY quarantined_at");
);
 
if (quarantineDt.Rows.Count > 0)
{
    Console.WriteLine($"WARNING: Quarantined rows: {quarantineDt.Rows.Count}");
    display(quarantineDt);
}
else
{
}
No quarantined rows — all data passed validation

SQL Server — query context log for this batch with QueryToTable()

Context Audit per Stage

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

 
var contextDt = QueryToTable(
    Console.WriteLine($"SELECT stage, business_date, trigger_type, schema_version, data_warnings FROM context_log WHERE batch_id = '{batchId}' ORDER BY created_at");
);
contextDt
stagebusiness_datetrigger_typeschema_versiondata_warnings
bronze29-Mar-26 0:00:00scheduled1.0
silver29-Mar-26 0:00:00scheduled1.0["sma_20: 95 NULL values (first 19 rows per symbol)"]
gold29-Mar-26 0:00:00scheduled1.0["sma_20: 95 NULL values (first 19 rows per symbol)"]

C# — display accumulated data warnings with Console.WriteLine()

// ── Show all data warnings accumulated across stages ──
 
if (goldStageCtx != null && goldStageCtx.DataWarnings.Count > 0)
{
    Console.WriteLine($"Data Warnings ({goldStageCtx.DataWarnings.Count} total):");
    foreach (var w in goldStageCtx.DataWarnings)
        Console.WriteLine($"  {w}");
}
else
{
}
Data Warnings (1 total):
  sma_20: 95 NULL values (first 19 rows per symbol)

JSON — inspect exported data contract with JsonSerializer.Deserialize()

// ── Inspect the Silver contract — shows structural schema + column semantics ──
 
var contractPath = Path.Combine(EXPORT_DIR, "contracts", "silver_ohlcv_contract.json");
if (File.Exists(contractPath))
{
    var contract = JsonSerializer.Deserialize<JsonElement>(File.ReadAllText(contractPath));
    Console.WriteLine($"Contract: {Path.GetFileName(contractPath)}");
    Console.WriteLine($"  Version: {(contract.TryGetProperty("x-contract-version", out var v) ? v.ToString() : "N/A")}");
    Console.WriteLine($"  Generated: {(contract.TryGetProperty("x-generated-at", out var g) ? g.ToString() : "N/A")}");
    Console.WriteLine($"  Fields: {(contract.TryGetProperty("properties", out var props) ? props.EnumerateObject().Count() : 0)}");
 
    if (contract.TryGetProperty("x-column-context", out var colCtx))
    {
        Console.WriteLine($"  Column contexts: {colCtx.GetArrayLength()}");
        foreach (var col in colCtx.EnumerateArray())
        {
            if (col.TryGetProperty("is_derived", out var isDerived) && isDerived.GetBoolean())
            {
                Console.WriteLine($"  {col.GetProperty("name")}:");
                Console.WriteLine($"    {col.GetProperty("description")}");
                Console.WriteLine($"    Computation: {col.GetProperty("computation")}");
                Console.WriteLine($"    Sources: {col.GetProperty("source_columns")}");
                Console.WriteLine($"    Null means: {col.GetProperty("null_semantics")}");
            }
        }
    }
}
Contract: silver_ohlcv_contract.json
  Version: 1.0
  Generated: 2026-03-29T23:24:12.1161340Z
  Fields: 13
  Column contexts: 13

  daily_return:
    Close-to-close return
    Computation: pct_change(close).over(symbol)
    Sources: [
        "bronze.close"
      ]
    Null means: first_row_in_series

  intraday_range:
    (high-low)/close
    Computation: (high - low) / close
    Sources: [
        "bronze.high",
        "bronze.low",
        "bronze.close"
      ]
    Null means: not_applicable

  sma_20:
    20-day moving average of close
    Computation: close.rolling_mean(20).over(symbol)
    Sources: [
        "bronze.close"
      ]
    Null means: insufficient_data

11. HttpListener Serving Layer — Pre-Materialized JSON API

HttpListener serves the Gold data products by reading pre-materialized JSON files. No database connection at runtime — the API reads files that the pipeline produced. Four endpoints: health (operational monitoring), daily-summary (market overview), symbol-profile (stock comparison), and lineage (pipeline execution audit).

API models

Typed record contracts that define the JSON shape returned by each HttpListener endpoint.

C# — define daily summary API response record

// ── Response schema for the /daily-summary endpoint ──
 
public record DailySummaryResponse(
    DateTime Date,
    int SymbolsTraded,
    double AvgReturn,
    double MaxReturn,
    double MinReturn,
    long TotalVolume,
    double AvgIntradayPct
);

C# — define symbol profile API response record

// ── Response schema for the /symbol-profile endpoint ──
 
public record SymbolProfileResponse(
    string Symbol,
    int TotalTradingDays,
    double AvgDailyReturn,
    double Volatility,
    double MaxDrawdown,
    double AvgVolume,
    double TotalDividends,
    DateTime FirstDate,
    DateTime LastDate
);

C# — define timeseries row API response record

// ── Response schema for the /symbol/{symbol}/timeseries endpoint ──
 
public record TimeSeriesRow(
    DateTime Date,
    double Open,
    double High,
    double Low,
    double Close,
    long Volume,
    double DailyReturn,
    double IntradayRange,
    double? Sma20
);

Endpoints

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

HttpListener — create listener instance with HttpListener()

// ── Serialize pipeline outputs to JSON for HttpListener endpoints ──
 
var dailyJson = JsonSerializer.Serialize(validDaily.AsEnumerable().Select(r => new {
    date = r.Field<DateTime>("date").ToString("yyyy-MM-dd"),
    symbols_traded = Convert.ToInt32(r["symbols_traded"]),
    avg_return = Convert.ToDouble(r["avg_return"]),
    max_return = Convert.ToDouble(r["max_return"]),
    min_return = Convert.ToDouble(r["min_return"]),
    total_volume = Convert.ToInt64(r["total_volume"]),
    avg_intraday_pct = Convert.ToDouble(r["avg_intraday_pct"]),
}).ToArray());
 
var profilesJson = JsonSerializer.Serialize(validProfiles.AsEnumerable().Select(r => new {
    symbol = r["symbol"]?.ToString() ?? "",
    total_trading_days = Convert.ToInt32(r["total_trading_days"]),
    avg_daily_return = Convert.ToDouble(r["avg_daily_return"]),
    volatility = Convert.ToDouble(r["volatility"]),
    max_drawdown = Convert.ToDouble(r["max_drawdown"]),
    avg_volume = Convert.ToDouble(r["avg_volume"]),
    total_dividends = Convert.ToDouble(r["total_dividends"]),
    first_date = r.Field<DateTime>("first_date").ToString("yyyy-MM-dd"),
    last_date = r.Field<DateTime>("last_date").ToString("yyyy-MM-dd"),
}).ToArray());
 
var silverSymbolJson = new Dictionary<string, string>();
foreach (var grp in silverDt.AsEnumerable().GroupBy(r => r["symbol"]?.ToString() ?? ""))
{
    silverSymbolJson[grp.Key.ToUpper()] = JsonSerializer.Serialize(grp.Select(r => new {
        date = r.Field<DateTime>("date").ToString("yyyy-MM-dd"),
        open = Convert.ToDouble(r["open"]),
        high = Convert.ToDouble(r["high"]),
        low = Convert.ToDouble(r["low"]),
        close = Convert.ToDouble(r["close"]),
        volume = Convert.ToInt64(r["volume"]),
        daily_return = Convert.ToDouble(r["daily_return"]),
        intraday_range = Convert.ToDouble(r["intraday_range"]),
        sma_20 = r["sma_20"] == DBNull.Value ? (double?)null : Convert.ToDouble(r["sma_20"]),
    }).ToArray());
}
 
JSON data prepared for HttpListener: daily=80'209 chars, profiles=1'089 chars, symbols=5

HttpListener — define health endpoint handler

// ── Healthcheck: verifies Parquet files exist and reports their sizes ──
 
string HandleHealth()
{
    var files = Directory.GetFiles(EXPORT_DIR, "*.parquet")
        .ToDictionary(f => Path.GetFileNameWithoutExtension(f), f => new FileInfo(f).Length);
    return JsonSerializer.Serialize(new { status = "healthy", files });
}

HttpListener — define daily summary endpoint handler

// ── Returns daily cross-sectional summary from pre-serialized JSON ──
 
string HandleDailySummary()
{
    return dailyJson;
}

HttpListener — define symbol profile endpoint handler

// ── Returns per-symbol summary statistics from pre-serialized JSON ──
 
string HandleSymbolProfile()
{
    return profilesJson;
}

HttpListener — define symbol timeseries endpoint handler

// ── Returns daily OHLCV + enrichment for one symbol from pre-serialized JSON ──
 
string HandleTimeseries(string symbol)
{
    return silverSymbolJson.GetValueOrDefault(symbol.ToUpper(), "[]");
}

HttpListener — define lineage endpoint handler

// ── Returns RunContext JSON for a batch, matched by prefix ──
 
string HandleLineage(string batchIdPrefix)
{
    var files = Directory.GetFiles(LINEAGE_DIR, $"run_{batchIdPrefix}*.json");
    return files.Length > 0 ? File.ReadAllText(files[0]) : "{}";
}

HttpListener — start server in background with Thread()

Background HTTP Server

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

 
int API_PORT = 8098;
var listener = new HttpListener();
listener.Prefixes.Add($"http://localhost:{API_PORT}/");
listener.Start();
 
async Task HandleRequests()
{
    while (listener.IsListening)
    {
        try
        {
            var ctx = await listener.GetContextAsync();
            var path = ctx.Request.Url.AbsolutePath;
            string responseJson = "";
            int statusCode = 200;
 
            if (path == "/health")
                responseJson = HandleHealth();
            else if (path == "/daily-summary")
                responseJson = HandleDailySummary();
            else if (path == "/symbol-profile")
                responseJson = HandleSymbolProfile();
            else if (path.StartsWith("/symbol/") && path.EndsWith("/timeseries"))
            {
                var symbol = path.Split('/')[2];
                responseJson = HandleTimeseries(symbol);
            }
            else if (path.StartsWith("/lineage/"))
            {
                var prefix = path.Split('/')[2];
                responseJson = HandleLineage(prefix);
            }
            else
                statusCode = 404;
 
            var bytes = System.Text.Encoding.UTF8.GetBytes(responseJson);
            ctx.Response.StatusCode = statusCode;
            ctx.Response.ContentType = "application/json";
            ctx.Response.ContentLength64 = bytes.Length;
            await ctx.Response.OutputStream.WriteAsync(bytes);
            ctx.Response.Close();
        }
        catch (HttpListenerException) { break; }
    }
}
 
var serverThread = new Thread(() => HandleRequests().Wait()) { IsBackground = true };
serverThread.Start();
Thread.Sleep(1000);
HttpListener running at http://localhost:8098

Testing

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

HttpClient — test health endpoint with GetStringAsync()

// ── Verify the API server is running and Parquet files are accessible ──
 
var client = new HttpClient();
var resp = await client.GetStringAsync($"http://localhost:{API_PORT}/health");
Console.WriteLine($"Status: 200");
Console.WriteLine(JsonSerializer.Serialize(JsonSerializer.Deserialize<JsonElement>(resp), new JsonSerializerOptions { WriteIndented = true }));
Status: 200
{
  "status": "healthy",
  "files": {
    "gold_daily_summary": 27461,
    "gold_symbol_profile": 2680,
    "silver_ohlcv": 164515
  }
}

HttpClient — test daily summary endpoint with GetStringAsync()

// ── Fetch daily cross-sectional summary ──
 
var resp = await client.GetStringAsync($"http://localhost:{API_PORT}/daily-summary");
var dailyArr = JsonSerializer.Deserialize<JsonElement>(resp);
Console.WriteLine($"Status: 200, rows: {dailyArr.GetArrayLength()}");
 
// Display as DataTable
var dailyRespDt = new DataTable();
dailyRespDt.Columns.Add("date", typeof(string));
dailyRespDt.Columns.Add("symbols_traded", typeof(int));
dailyRespDt.Columns.Add("avg_return", typeof(double));
dailyRespDt.Columns.Add("total_volume", typeof(long));
foreach (var item in dailyArr.EnumerateArray().Take(5))
    dailyRespDt.Rows.Add(
        item.GetProperty("date").GetString(),
        item.GetProperty("symbols_traded").GetInt32(),
        item.GetProperty("avg_return").GetDouble(),
        item.GetProperty("total_volume").GetInt64()
    );
dailyRespDt
Status: 200, rows: 506
datesymbols_tradedavg_returntotal_volume
2024-03-285014115241
2024-04-025-0.00626115460060
2024-04-0350.00486212344475
2024-04-045-0.00063110238748
2024-04-055-0.01409216364036

HttpClient — test symbol profile endpoint with GetStringAsync()

// ── Fetch all symbol profiles from the API ──
 
var resp = await client.GetStringAsync($"http://localhost:{API_PORT}/symbol-profile");
var profileArr = JsonSerializer.Deserialize<JsonElement>(resp);
Console.WriteLine($"Status: 200, profiles: {profileArr.GetArrayLength()}");
 
var profileRespDt = new DataTable();
profileRespDt.Columns.Add("symbol", typeof(string));
profileRespDt.Columns.Add("total_trading_days", typeof(int));
profileRespDt.Columns.Add("avg_daily_return", typeof(double));
profileRespDt.Columns.Add("volatility", typeof(double));
foreach (var item in profileArr.EnumerateArray())
    profileRespDt.Rows.Add(
        item.GetProperty("symbol").GetString(),
        item.GetProperty("total_trading_days").GetInt32(),
        item.GetProperty("avg_daily_return").GetDouble(),
        item.GetProperty("volatility").GetDouble()
    );
profileRespDt
Status: 200, profiles: 5
symboltotal_trading_daysavg_daily_returnvolatility
ALV.DE5060.0005320.011839
BAS.DE5060.0001210.017491
DTE.DE5060.0007650.01325
SAP.DE506-0.0002850.0189
SIE.DE5060.0004750.019204

HttpClient — test symbol timeseries endpoint with GetStringAsync()

// ── Fetch last 10 days of SAP.DE time series data ──
 
var resp = await client.GetStringAsync($"http://localhost:{API_PORT}/symbol/SAP.DE/timeseries");
var tsArr = JsonSerializer.Deserialize<JsonElement>(resp);
Console.WriteLine($"Status: 200, rows: {tsArr.GetArrayLength()}");
 
var tsRespDt = new DataTable();
tsRespDt.Columns.Add("date", typeof(string));
tsRespDt.Columns.Add("open", typeof(double));
tsRespDt.Columns.Add("high", typeof(double));
tsRespDt.Columns.Add("low", typeof(double));
tsRespDt.Columns.Add("close", typeof(double));
tsRespDt.Columns.Add("volume", typeof(long));
foreach (var item in tsArr.EnumerateArray().Take(5))
    tsRespDt.Rows.Add(
        item.GetProperty("date").GetString(),
        item.GetProperty("open").GetDouble(),
        item.GetProperty("high").GetDouble(),
        item.GetProperty("low").GetDouble(),
        item.GetProperty("close").GetDouble(),
        item.GetProperty("volume").GetInt64()
    );
tsRespDt
Status: 200, rows: 506
dateopenhighlowclosevolume
2024-03-28181.82000732421875181.86000061035156179.10000610351562180.46000671386721700825
2024-04-02181181.9199981689453177.05999755859375177.059997558593751838833
2024-04-03178.32000732421875179.52000427246094176.55999755859375178.220001220703121501774
2024-04-04177.9199981689453178.4600067138672176.33999633789062178.020004272460941126985
2024-04-05175.39999389648438177.9600067138672174.77999877929688177.41999816894532099406

HttpClient — test lineage endpoint with GetStringAsync()

// ── Fetch pipeline execution metadata for this run ──
 
var resp = await client.GetStringAsync($"http://localhost:{API_PORT}/lineage/{batchId[..8]}");
Console.WriteLine($"Status: 200");
var data = JsonSerializer.Deserialize<JsonElement>(resp);
Console.WriteLine($"Batch: {data.GetProperty("BatchId").GetString()[..8]}... | Status: {data.GetProperty("Status")}");
 
var stagesDt = new DataTable();
stagesDt.Columns.Add("stage", typeof(string));
stagesDt.Columns.Add("input_rows", typeof(int));
stagesDt.Columns.Add("output_rows", typeof(int));
stagesDt.Columns.Add("rows_rejected", typeof(int));
stagesDt.Columns.Add("output_hash", typeof(string));
foreach (var s in data.GetProperty("Stages").EnumerateArray())
    stagesDt.Rows.Add(
        s.GetProperty("Stage").GetString(),
        s.GetProperty("InputRows").GetInt32(),
        s.GetProperty("OutputRows").GetInt32(),
        s.GetProperty("RowsRejected").GetInt32(),
        s.GetProperty("OutputHash").GetString()
    );
stagesDt
Status: 200
Batch: d99b77ff... | Status: completed
stageinput_rowsoutput_rowsrows_rejectedoutput_hash
bronze550959fbdf2bc0d14a5
silver25302530076d1db80852e66dc
gold5115110a87ddaf98043eb7c
export3041304104915a3cad5280dd8

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.NET charts covering return time series, cumulative performance, risk-return scatter, and stage timing.

Plotly.NET — plot daily return time series with Chart.Line()

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

 
var threeMonthsAgo = DateTime.Today.AddDays(-90);
 
var returnTraces = SYMBOLS.Select(symbol =>
{
    var symDt = QueryToTable(
        Console.WriteLine($"SELECT date, daily_return FROM silver_ohlcv ");
        + $"WHERE symbol = '{symbol}' AND date >= '{threeMonthsAgo:yyyy-MM-dd}' "
        + "ORDER BY date");
    var dates = symDt.AsEnumerable().Select(r => r.Field<DateTime>("date")).ToArray();
    var returns = symDt.AsEnumerable().Select(r => Convert.ToDouble(r["daily_return"])).ToArray();
    return Plotly.NET.CSharp.Chart.Line<DateTime, double, string>(x: dates, y: returns, Name: symbol);
}).ToArray();
 
Plotly.NET.CSharp.Chart.Combine(returnTraces)
    .WithTitle("Daily Returns — Last 3 Months")
    .WithXAxisStyle(Title.init("Date"))
    .WithYAxisStyle(Title.init("Daily Return"))
    .WithSize(750, 450)
    .WithLayout(Layout.init<string>(
        PaperBGColor: Color.fromString("transparent"),
        PlotBGColor: Color.fromString("transparent"),
        Font: Font.init(Color: Color.fromHex("#cccccc"))))

Plotly.NET — plot cumulative returns comparison with cumulative product

// Shows how a €1 investment in each symbol would have grown
 
var cumTraces = SYMBOLS.Select(symbol =>
{
    var symDt = QueryToTable(
        Console.WriteLine($"SELECT date, daily_return FROM silver_ohlcv ");
        + $"WHERE symbol = '{symbol}' ORDER BY date");
    var dates = symDt.AsEnumerable().Select(r => r.Field<DateTime>("date")).ToArray();
    var returns = symDt.AsEnumerable().Select(r => Convert.ToDouble(r["daily_return"])).ToArray();
 
    var cumRet = new double[returns.Length];
    cumRet[0] = 1.0 + returns[0];
    for (int i = 1; i < returns.Length; i++)
        cumRet[i] = cumRet[i - 1] * (1.0 + returns[i]);
 
    return Plotly.NET.CSharp.Chart.Line<DateTime, double, string>(x: dates, y: cumRet, Name: symbol);
}).ToArray();
 
Plotly.NET.CSharp.Chart.Combine(cumTraces)
    .WithTitle("Cumulative Returns — €1 Investment")
    .WithXAxisStyle(Title.init("Date"))
    .WithYAxisStyle(Title.init("Growth of €1"))
    .WithSize(750, 450)
    .WithLayout(Layout.init<string>(
        PaperBGColor: Color.fromString("transparent"),
        PlotBGColor: Color.fromString("transparent"),
        Font: Font.init(Color: Color.fromHex("#cccccc"))))

Plotly.NET — plot risk-return scatter with Chart.Point()

// Risk-return visualization using Gold symbol profile data
 
var profDt = QueryToTable("SELECT symbol, volatility, avg_daily_return FROM gold_symbol_profile");
 
var scatterSymbols = profDt.AsEnumerable().Select(r => r.Field<string>("symbol")).ToArray();
var vols = profDt.AsEnumerable().Select(r => Convert.ToDouble(r["volatility"]) * 100).ToArray();
var avgRets = profDt.AsEnumerable().Select(r => Convert.ToDouble(r["avg_daily_return"]) * 100).ToArray();
 
Plotly.NET.CSharp.Chart.Point<double, double, string>(x: vols, y: avgRets, Name: "Symbols",
        MultiText: scatterSymbols, TextPosition: StyleParam.TextPosition.TopCenter,
        MarkerColor: Color.fromHex("#4285F4"))
    .WithMarkerStyle(Size: 12)
    .WithTitle("Risk-Return Profile — Volatility vs Avg Daily Return")
    .WithXAxisStyle(Title.init("Daily Volatility (%)"))
    .WithYAxisStyle(Title.init("Avg Daily Return (%)"))
    .WithSize(750, 450)
    .WithLayout(Layout.init<string>(
        PaperBGColor: Color.fromString("transparent"),
        PlotBGColor: Color.fromString("transparent"),
        Font: Font.init(Color: Color.fromHex("#cccccc"))))

Plotly.NET — plot pipeline stage timing with Chart.Column()

// Shows how long each pipeline stage took in milliseconds
 
var stageNames = runContext.Stages.Select(s => s.Stage).ToArray();
var durations = runContext.Stages.Select(s => s.DurationMs).ToArray();
 
Plotly.NET.CSharp.Chart.Column<double, string, string>(values: durations, Keys: stageNames, Name: "Duration")
    .WithMarkerStyle(Color: Color.fromColors(new[] {
        Color.fromHex("#4285F4"), Color.fromHex("#34A853"),
        Color.fromHex("#FBBC04"), Color.fromHex("#EA4335") }))
    .WithTitle("Pipeline Stage Duration")
    .WithXAxisStyle(Title.init("Stage"))
    .WithYAxisStyle(Title.init("Duration (ms)"))
    .WithSize(750, 450)
    .WithLayout(Layout.init<string>(
        PaperBGColor: Color.fromString("transparent"),
        PlotBGColor: Color.fromString("transparent"),
        Font: Font.init(Color: Color.fromHex("#cccccc"))))

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 from Gold back to the raw source in seven steps, each independently verifiable: Bronze (raw values as ingested), Silver (computed return verified mathematically), Gold (propagation to aggregation), Lineage (batch metadata with SHA-256 hash), RunContext (execution fingerprint), Landing Zone (raw JSON file on disk), and Live API (corroboration with current source). This is the proof that the architecture’s lineage tracking delivers real forensic capability.

Disputed data point investigation

Seven-step forensic trace from Gold back to raw source: Bronze → Silver → Gold → Lineage → RunContext → Landing Zone → Live API.

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
 
var bronzeAudit = QueryToTable(
    @"SELECT symbol, date, [open], high, low, [close],
             adj_close, volume, dividends, stock_splits, batch_id, ingested_at
      FROM bronze_ohlcv
      WHERE symbol = 'SAP.DE' AND date = '2026-01-29'");
bronzeAudit
Bronze table (raw ingested):
symboldateopenhighlowcloseadj_closevolumedividendsstock_splitsbatch_idingested_at
SAP.DE29-Jan-26 0:00:00179180.16000366210938162.1199951171875164.6199951171875164.619995117187515846791009c135c08-937d-4413-8bb4-1b66407ed9a528-Mar-26 23:21:33

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

// Step 3: Check Silver — verify the daily_return calculation is correct
 
var silverAudit = QueryToTable(
    @"SELECT symbol, date, [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");
 
if (silverAudit.Rows.Count >= 2)
{
    var prevClose = Convert.ToDouble(silverAudit.Rows[0]["close"]);
    var currClose = Convert.ToDouble(silverAudit.Rows[1]["close"]);
    var expectedReturn = (currClose - prevClose) / prevClose;
    var actualReturn = Convert.ToDouble(silverAudit.Rows[1]["daily_return"]);
    Console.WriteLine($"  Previous close (Jan 28): {prevClose:F2}");
    Console.WriteLine($"  Current close  (Jan 29): {currClose:F2}");
    Console.WriteLine($"  Expected return: ({currClose:F2} - {prevClose:F2}) / {prevClose:F2} = {expectedReturn:F6}");
    Console.WriteLine($"  Actual return:   {actualReturn:F6}");
    Console.WriteLine($"  Match: {Math.Abs(expectedReturn - actualReturn) < 0.000001}");
}
silverAudit
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.DE28-Jan-26 0:00:00196.139999389648440.0030680.020598202.3795d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f29-Mar-26 23:24:06
SAP.DE29-Jan-26 0:00:00164.6199951171875-0.1607020.109586200.193d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f29-Mar-26 23:24:06
SAP.DE30-Jan-26 0:00:00170.559997558593750.0360830.038227198.6235d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f29-Mar-26 23:24:06

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

// Step 4: Check Gold — verify the data point propagated to aggregations
 
var goldDailyAudit = QueryToTable(
    @"SELECT date, symbols_traded, avg_return, min_return, max_return, total_volume, batch_id
      FROM gold_daily_summary WHERE date = '2026-01-29'");
goldDailyAudit
Gold daily summary (Jan 29):
  The min_return on this day should reflect SAP's -16% drop
datesymbols_tradedavg_returnmin_returnmax_returntotal_volumebatch_id
29-Jan-26 0:00:005-0.025652-0.1607020.02012825357247d99b77ff-42f2-4cde-a7b2-fbfc54ad6e6f

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

// Step 5: Trace the disputed data point back to its pipeline run
var batchFromBronze = sqlConn.ExecuteScalar<string>(
    "SELECT batch_id FROM bronze_ohlcv WHERE symbol = 'SAP.DE' AND date = '2026-01-29'");
 
Console.WriteLine($"Disputed row: SAP.DE / 2026-01-29");
Console.WriteLine($"Batch ID (from row): {batchFromBronze}");
 
var lineageAudit = QueryToTable(
    $@"SELECT stage, started_at, completed_at, input_rows, output_rows,
              rows_rejected, output_hash
       FROM lineage_stages WHERE batch_id = '{batchFromBronze}'
       ORDER BY started_at");
lineageAudit
Disputed row: SAP.DE / 2026-01-29
Batch ID (from row): 9c135c08-937d-4413-8bb4-1b66407ed9a5

Full pipeline execution for this batch:
stagestarted_atcompleted_atinput_rowsoutput_rowsrows_rejectedoutput_hash
bronze28-Mar-26 23:21:3428-Mar-26 23:21:37253025300914eccd231d933a2
silver28-Mar-26 23:25:0928-Mar-26 23:25:12253025300d3c2286d9c3a8b8c
gold28-Mar-26 23:29:1828-Mar-26 23:29:185115110fabdf6a3896591c1
export28-Mar-26 23:29:4128-Mar-26 23:29:41304130410ad1a03cae4b23504

JSON — verify RunContext execution metadata with JsonSerializer

// Step 6: Load RunContext — the pipeline's execution fingerprint
 
var ctxFiles = Directory.GetFiles(LINEAGE_DIR, $"run_{batchFromBronze[..8]}*.json");
if (ctxFiles.Length > 0)
{
    var ctxText = File.ReadAllText(ctxFiles[0]);
    // Handle BOM if present
    if (ctxText[0] == '') ctxText = ctxText[1..];
    var ctx = JsonSerializer.Deserialize<JsonElement>(ctxText);
    Console.WriteLine($"RunContext: {Path.GetFileName(ctxFiles[0])}");
 
    // Helper: try PascalCase (C#) then snake_case (Python) keys
    JsonElement Prop(JsonElement el, string pascal, string snake) =>
        el.TryGetProperty(pascal, out var v) ? v :
        el.TryGetProperty(snake, out var v2) ? v2 :
        throw new Exception($"Key {pascal}/{snake} not found");
 
    Console.WriteLine($"  Status:       {Prop(ctx, "Status", "status")}");
    Console.WriteLine($"  Symbols:      [{string.Join(", ", Prop(ctx, "Symbols", "symbols").EnumerateArray())}]");
    Console.WriteLine($"  Date range:   [{string.Join(", ", Prop(ctx, "DateRange", "date_range").EnumerateArray())}]");
    var stages = Prop(ctx, "Stages", "stages");
    var totalRejected = stages.EnumerateArray()
        .Sum(s => Prop(s, "RowsRejected", "rows_rejected").GetInt32());
    Console.WriteLine($"  Rejected:     {totalRejected} rows");
    Console.WriteLine($"  Output hash:  {Prop(stages.EnumerateArray().First(), "OutputHash", "output_hash")}");
}
else
{
    Console.WriteLine($"No RunContext found for batch {batchFromBronze[..8]}");
}
RunContext: run_9c135c08.json
  Status:       completed
  Symbols:      [SAP.DE, SIE.DE, ALV.DE, DTE.DE, BAS.DE]
  Date range:   [2024-03-29, 2026-03-29]
  Rejected:     0 rows
  Output hash:  914eccd231d933a2

C# — corroborate with landing zone file

// Step 7: Prove the landing zone file exists and contains the record
 
var landingFile = Path.Combine(LANDING_DIR, "ohlcv_SAP_DE.json");
var rawRecords = JsonSerializer.Deserialize<List<Dictionary<string, JsonElement>>>(
    File.ReadAllText(landingFile));
 
var jan29InFile = rawRecords.Where(r => r["date"].GetString() == "2026-01-29").ToList();
if (jan29InFile.Any())
{
    var liveClose29 = jan29InFile[0]["close"].GetDouble();
    Console.WriteLine($"Landing file: {Path.GetFileName(landingFile)}");
    Console.WriteLine($"  Total records: {rawRecords.Count}");
    Console.WriteLine($"  Jan 29 record found:");
    foreach (var kv in jan29InFile[0])
        Console.WriteLine($"    {kv.Key,-15}: {kv.Value}");
    Console.WriteLine($"\n  Bronze close: {Convert.ToDouble(bronzeAudit.Rows[0]["close"]):F2}");
    Console.WriteLine($"  Landing close: {liveClose29}");
    Console.WriteLine($"  Match: {Math.Abs(liveClose29 - Convert.ToDouble(bronzeAudit.Rows[0]["close"])) < 0.01}");
}
else
{
    var dates = rawRecords.Select(r => r["date"].GetString()).OrderBy(d => d).ToList();
    Console.WriteLine($"Landing file covers: {dates.First()} to {dates.Last()}");
    Console.WriteLine("Jan 29 not in current file (overwritten by incremental fetch)");
    Console.WriteLine("In production, landing files are archived and would contain this record");
}
Landing file covers: 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

C# — display full audit trail summary as DataTable

// ── Audit conclusion: full chain of evidence ──
 
var bronzeClose = Convert.ToDouble(bronzeAudit.Rows[0]["close"]);
var silverRow = silverAudit.AsEnumerable()
    .First(r => r.Field<DateTime>("date").ToString("yyyy-MM-dd") == "2026-01-29");
var silverClose = Convert.ToDouble(silverRow["close"]);
 
var auditSummary = new DataTable();
auditSummary.Columns.Add("step", typeof(string));
auditSummary.Columns.Add("source", typeof(string));
auditSummary.Columns.Add("close", typeof(double));
auditSummary.Columns.Add("evidence", typeof(string));
 
auditSummary.Rows.Add("1. Bronze table", "bronze_ohlcv", bronzeClose,
    $"Batch {batchFromBronze[..8]}");
auditSummary.Rows.Add("2. Silver table", "silver_ohlcv", silverClose,
    "Return = -16.07% verified");
auditSummary.Rows.Add("3. Gold table", "gold_daily_summary", DBNull.Value,
    "min_return reflects the drop");
auditSummary.Rows.Add("4. Lineage", "lineage_stages", DBNull.Value,
    $"Hash: {lineageAudit.Rows[0]["output_hash"]}");
auditSummary.Rows.Add("5. RunContext", $"run_{batchFromBronze[..8]}.json", DBNull.Value,
    "0 rejected, status=completed");
 
Console.WriteLine("AUDIT CONCLUSION: The SAP.DE -16% drop on 2026-01-29 is AUTHENTIC.");
Console.WriteLine("  - Same close price in Bronze and Silver");
Console.WriteLine("  - Daily return verified mathematically from consecutive closes");
Console.WriteLine("  - Zero rows rejected by Pydantic validation");
Console.WriteLine("  - Output hash proves no post-ingestion tampering");
auditSummary
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.6199951171875Batch 9c135c08
2. Silver tablesilver_ohlcv164.6199951171875Return = -16.07% verified
3. Gold tablegold_daily_summarymin_return reflects the drop
4. Lineagelineage_stagesHash: 914eccd231d933a2
5. RunContextrun_9c135c08.json0 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 demonstration: zero-volume classification ──
 
var zeroVol = QueryToTable(
    "SELECT DISTINCT symbol, date FROM silver_ohlcv WHERE volume = 0");
 
if (zeroVol.Rows.Count > 0)
{
    var classified = new DataTable();
    classified.Columns.Add("symbol", typeof(string));
    classified.Columns.Add("date", typeof(DateTime));
    classified.Columns.Add("is_trading_day", typeof(bool));
    classified.Columns.Add("verdict", typeof(string));
 
    foreach (DataRow row in zeroVol.Rows)
    {
        var isTradingDay = sqlConn.ExecuteScalar<bool?>(
            $"SELECT is_trading_day FROM dim_calendar WHERE date = @Date AND exchange_code = 'XETR'",
            new { Date = row["date"] });
        classified.Rows.Add(row["symbol"], row["date"],
            isTradingDay,
            isTradingDay == true ? "anomaly" : "non-trading day");
    }
 
    var anomalies = classified.AsEnumerable().Count(r => r.Field<string>("verdict") == "anomaly");
    var holidays = classified.AsEnumerable().Count(r => r.Field<string>("verdict") == "non-trading day");
 
    classified.DefaultView.Sort = "date ASC";
    var sorted = classified.DefaultView.ToTable();
    // show first 5
    var preview = sorted.Clone();
    for (int i = 0; i < Math.Min(5, sorted.Rows.Count); i++)
        preview.ImportRow(sorted.Rows[i]);
    display(preview);
 
    Console.WriteLine($"\nTotal zero-volume: {classified.Rows.Count}  |  " +
        $"Non-trading days: {holidays}  |  Anomalies to investigate: {anomalies}");
}
else
{
    Console.WriteLine($"No zero-volume rows in Silver");
}
symboldateis_trading_dayverdict
ALV.DE20-Sep-24 0:00:00non-trading day
BAS.DE21-Oct-24 0:00:00non-trading day
SAP.DE21-Oct-24 0:00:00non-trading day
ALV.DE01-Nov-24 0:00:00non-trading day
BAS.DE01-Nov-24 0:00:00non-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 demonstration: SMA-20 null accounting ──
 
var smaNulls = QueryToTable(
    @"SELECT symbol, COUNT(*) as null_count, MIN(date) as first_null, MAX(date) as last_null
      FROM silver_ohlcv WHERE sma_20 IS NULL GROUP BY symbol ORDER BY symbol");
 
int actualNullCount = sqlConn.ExecuteScalar<int>("SELECT COUNT(*) FROM silver_ohlcv WHERE sma_20 IS NULL");
int symbolsCount = sqlConn.ExecuteScalar<int>("SELECT COUNT(DISTINCT symbol) FROM silver_ohlcv");
int expectedNullCount = 19 * symbolsCount;
 
// Add expected + unexplained columns
smaNulls.Columns.Add("expected", typeof(int));
smaNulls.Columns.Add("unexplained", typeof(int));
foreach (DataRow row in smaNulls.Rows)
{
    row["expected"] = 19;
    row["unexplained"] = Convert.ToInt32(row["null_count"]) - 19;
}
 
display(smaNulls);
Console.WriteLine($"\nExpected nulls: {expectedNullCount} (19 x {symbolsCount} symbols)  |  " +
    $"Actual: {actualNullCount}  |  Unexplained: {actualNullCount - expectedNullCount}");
 
var smaWarnings = goldStageCtx.DataWarnings.Where(w => w.Contains("sma_20")).ToList();
if (smaWarnings.Any())
    Console.WriteLine($"Context recorded: {smaWarnings[0]}");
symbolnull_countfirst_nulllast_nullexpectedunexplained
ALV.DE1928-Mar-24 0:00:0025-Apr-24 0:00:00190
BAS.DE1928-Mar-24 0:00:0025-Apr-24 0:00:00190
DTE.DE1928-Mar-24 0:00:0025-Apr-24 0:00:00190
SAP.DE1928-Mar-24 0:00:0025-Apr-24 0:00:00190
SIE.DE1928-Mar-24 0:00:0025-Apr-24 0:00:00190
Expected nulls: 95 (19 x 5 symbols)  |  Actual: 95  |  Unexplained: 0
Context recorded: 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 demonstration: data contract as structured metadata ──
 
var contractJson = File.ReadAllText(Path.Combine(EXPORT_DIR, "contracts", "gold_symbol_profile_contract.json"));
var contract = JsonSerializer.Deserialize<JsonElement>(contractJson);
 
var derived = new DataTable();
derived.Columns.Add("column", typeof(string));
derived.Columns.Add("description", typeof(string));
derived.Columns.Add("unit", typeof(string));
derived.Columns.Add("computation", typeof(string));
derived.Columns.Add("source_columns", typeof(string));
derived.Columns.Add("null_means", typeof(string));
 
foreach (var col in contract.GetProperty("x-column-context").EnumerateArray())
{
    if (col.TryGetProperty("is_derived", out var isDerived) && isDerived.GetBoolean())
    {
        derived.Rows.Add(
            col.GetProperty("name").GetString(),
            col.GetProperty("description").GetString(),
            col.GetProperty("unit").GetString(),
            col.TryGetProperty("computation", out var comp) ? comp.GetString() : "—",
            col.TryGetProperty("source_columns", out var src)
                ? string.Join(", ", src.EnumerateArray().Select(s => s.GetString())) : "—",
            col.TryGetProperty("null_semantics", out var ns) ? ns.GetString() : "—"
        );
    }
}
derived
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 demonstration: interpreting a Gold value ──
 
var volMeta = contract.GetProperty("x-column-context").EnumerateArray()
    .First(c => c.GetProperty("name").GetString() == "volatility");
 
var germanProfiles = QueryToTable(
    "SELECT symbol, volatility FROM gold_symbol_profile WHERE symbol LIKE '%.DE'");
 
var interpretation = new DataTable();
interpretation.Columns.Add("symbol", typeof(string));
interpretation.Columns.Add("daily_vol", typeof(double));
interpretation.Columns.Add("annual_vol_%", typeof(double));
interpretation.Columns.Add("unit", typeof(string));
interpretation.Columns.Add("formula", typeof(string));
 
foreach (DataRow row in germanProfiles.Rows)
{
    var vol = Convert.ToDouble(row["volatility"]);
    interpretation.Rows.Add(
        row["symbol"],
        Math.Round(vol, 4),
        Math.Round(vol * Math.Sqrt(252) * 100, 1),
        volMeta.GetProperty("unit").GetString(),
        volMeta.GetProperty("computation").GetString()
    );
}
 
display(interpretation);
Console.WriteLine($"\nContract says: '{volMeta.GetProperty("description").GetString()}'");
symboldaily_volannual_vol_%unitformula
ALV.DE0.011818.8decimal_ratiostd(daily_return) per symbol
BAS.DE0.017527.8decimal_ratiostd(daily_return) per symbol
DTE.DE0.013221decimal_ratiostd(daily_return) per symbol
SAP.DE0.018930decimal_ratiostd(daily_return) per symbol
SIE.DE0.019230.5decimal_ratiostd(daily_return) per symbol

C# Functional Pipeline Warnings

Mutable class DTOs passed through pipeline stages allow silent state mutation

If a Bronze DTO is a class with public setters, any method in the transform chain can modify it. A Silver enrichment step that accidentally writes back a corrected value to the Bronze object corrupts the immutable source record.

Correct pattern

Declare all pipeline DTOs as record types. Use with expressions to produce a new instance with the changed field rather than mutating the existing one: var silverRow = bronzeRow with { CloseAdj = adjusted };.

Using string concatenation in Dapper queries creates SQL injection risk

Dapper executes whatever SQL string you provide. Building queries with $"SELECT * FROM {tableName} WHERE symbol = '{symbol}'" is exploitable if any input comes from user-controlled or untrusted sources.

Correct pattern

Always use parameterized queries: conn.Query<BronzeRow>("SELECT * FROM bronze WHERE symbol = @symbol", new { symbol }). Dapper maps named parameters safely.

HttpListener is not safe to expose on a production internet-facing endpoint

HttpListener has no TLS termination, no authentication middleware, no rate limiting, and no routing beyond basic prefix matching. Binding it to http://+:8080/ on a production server exposes an unprotected endpoint.

Correct pattern

Use HttpListener only for local development, integration testing, or pipeline-internal communication. For production, replace it with ASP.NET Core Minimal API, which adds TLS, middleware, dependency injection, and OpenAPI generation.

Polly v7 and v8 builder APIs are incompatible and cannot coexist in the same call chain

Polly v8 introduced a breaking redesign — Policy.Handle<>()...Build() is replaced by new ResiliencePipelineBuilder()...Build(). Mixing the two in one project compiles but produces runtime exceptions.

Correct pattern

Standardize on one Polly version per project. For new code, use Polly v8 with ResiliencePipelineBuilder<T>. Migrate old v7 policies incrementally by wrapping them as ResiliencePipeline delegates.

C# Functional Pipeline Recommendations

  • Enforce immutability with record types for all DTOs. The compiler generates Equals, GetHashCode, and with for free, and the init-only setters prevent post-construction mutation without explicit copying.
  • Run FluentValidation rules before any I/O. Validate in memory first; only write to SQL Server after the validator returns no failures. A failed database write with a partial row is harder to clean up than a rejected in-memory object.
  • Configure Polly retry with ShouldHandle to distinguish transient from permanent errors. Retry on HttpRequestException and SqlException with error code 1205 (deadlock); do not retry on JsonException or ValidationException — those are permanent failures.
  • Use SqlBulkCopy for Bronze inserts above ~1,000 rows. Dapper’s Execute with INSERT statements sends one round-trip per row. SqlBulkCopy uses the TDS bulk load protocol and is typically 10–50× faster for batch inserts.
  • Store the pipeline run ID in every Bronze and Silver row. This enables forensic Dapper queries that filter WHERE run_id = @runId to reconstruct exactly what a specific execution produced.
  • Pre-materialize all JSON before the HttpListener loop starts. Serializing JsonSerializer.Serialize(goldRows) once at startup is far cheaper than re-serializing on every request, especially under concurrent load.
  • Version FluentValidation rules alongside schema migrations. If a Silver validator’s rules change, Bronze rows that were valid under the previous rule set may fail under the new one — keep a changelog in the configuration constants section.
  • Monitor the lineage table for missing run IDs. A Dapper query like SELECT expected_date FROM schedule WHERE run_id IS NULL is the authoritative check that a pipeline executed — do not rely on log file presence alone.

C# Functional Pipeline Troubleshooting

ProblemCauseFix
FluentValidation.ValidationException on Silver ingestionBronze row has a null or out-of-range value that passed initial landingAdd a field_validator equivalent (RuleFor(x => x.Close).GreaterThan(0)) and log rejected rows to a bronze_quarantine table
SqlException: Cannot insert duplicate key in object 'bronze'Incremental ingestion re-downloaded a date range already in the tableAdd a WHERE NOT EXISTS (SELECT 1 FROM bronze WHERE symbol = @symbol AND date = @date) guard to the insert statement
HttpListenerException: Access is denied when starting the serving layerWindows requires explicit netsh URL reservation for non-localhost prefixesRun netsh http add urlacl url=http://+:8080/ user=Everyone once, or change the prefix to http://localhost:8080/ for local use
Polly retry fires on every request even for non-transient errorsShouldHandle predicate is too broad (matching all exceptions)Narrow the predicate: handle only HttpRequestException and specific SqlException error codes; re-throw all others immediately
Parquet file not found when the API startsExport step ran before the Gold writes were committed, or path constants divergedConfirm PARQUET_PATH in the export step and the API startup code reference the same Path.Combine(BASE_DIR, ...) expression
SHA-256 mismatch for rows that were never manually modifiedTrailing whitespace or encoding difference in a string field produces a different hashNormalize all string fields to .Trim().ToLowerInvariant() before hashing; document the normalization contract next to the hash utility method
Dapper InvalidOperationException: Sequence contains no elementsQuerySingle used on a query that returned zero rowsReplace QuerySingle with QuerySingleOrDefault and check for null before using the result
JsonException: The JSON value could not be converted to System.DoubleGold row has a NaN or Infinity value (from a zero-price division) that System.Text.Json cannot serializeGuard against division by zero in the Gold aggregation step; replace invalid doubles with null before serialization
Contract says: 'Standard deviation of daily returns — annualize by multiplying by sqrt(252)'

Graph View

Backlinks