Unit Testing for Market-Data Pipeline Error Detection

Hello. In the previous lesson, you used exploratory plots to identify heavy tails, dependence, and volatility clustering in adjusted market returns. Those diagnostics are only reliable if the pipeline creating the returns is reliable. Before moving into formal time-series models, this lesson closes Module 1 with a practical safeguard: unit tests that detect market-data errors before they reach research, backtests, or risk reports.

By the end, you will be able to structure a small pytest suite around a market-data transformation, create deliberately problematic mini-datasets, and test for duplicate timestamps, invalid prices, missing-data leakage, and incorrect use of unadjusted prices.

---

Tests as executable market-data contracts

A unit test is a small, fast program that verifies one clearly stated behavior of one function. In a market-data pipeline, a useful test answers a concrete question such as:

• Does the pipeline reject duplicate timestamp–symbol observations?
• Does it preserve a missing adjusted price rather than silently inventing a zero return?
• Does it use adj_close, not raw close, when computing returns?
• Does it reject zero or negative prices before applying logarithms?
• Does it sort observations into a reproducible order?

These are not cosmetic details. A single duplicate can change a pivot operation; a forward-filled price can understate volatility; an unadjusted stock split can appear as a large loss; and a nonpositive price makes a log return mathematically undefined.

The key idea is that testing is not merely about proving that code runs. It is about encoding the data assumptions on which a quantitative result depends.

A useful separation is:

Test type	Question it answers	Example
Input-contract test	Is this input safe to process?	Required columns exist; timestamp–symbol keys are unique.
Transformation test	Did the function produce the intended output?	Adjusted prices produce the correct log returns.
Regression test	Would a previously fixed bug reappear?	A missing price must not become a fabricated zero return.
Integration test	Do several components work together?	A CSV is read, cleaned, transformed, and written correctly.

For this lesson, prioritize unit tests: in-memory DataFrames, no vendor download, no network, no dependence on today’s date. They should run in milliseconds, making them suitable for every commit.

The Arrange–Act–Assert discipline

Keep every test readable with three conceptual stages:

1. Arrange: Construct the smallest input that exposes the behavior.
2. Act: Call exactly one function under test.
3. Assert: Check the output or expected exception.

For example, a duplicate test should not load a CSV, calculate returns, produce charts, and write a file. It should construct two rows with the same timestamp and symbol, call the validation function, and confirm that a clear error is raised. Small tests fail for understandable reasons.

---

A small, testable market-data core

A pipeline becomes difficult to test when file downloading, cleaning, return calculations, plotting, and saving all happen inside one large function. Instead, isolate deterministic transformations from I/O.

Use a structure such as:

quant-research/
├── src/
│   └── market_data/
│       ├── __init__.py
│       └── transforms.py
└── tests/
    └── test_transforms.py

Ensure pytest is included as a development dependency in the reproducible environment you created earlier. It belongs in the development toolchain rather than in the runtime requirements of a research script.

Here is a deliberately compact transformation module. It assumes long-form market data with one row per instrument and timestamp.

# src/market_data/transforms.py

import numpy as np
import pandas as pd

REQUIRED_COLUMNS = {"timestamp", "symbol", "adj_close"}


def validate_and_sort_quotes(raw):
    """Validate a long-form adjusted-price table and return sorted data."""
    missing_columns = REQUIRED_COLUMNS.difference(raw.columns)
    if missing_columns:
        raise ValueError(
            f"missing required columns: {sorted(missing_columns)}"
        )

    out = raw.copy()

    out["timestamp"] = pd.to_datetime(
        out["timestamp"],
        utc=True,
        errors="raise",
    )

    if out["timestamp"].isna().any():
        raise ValueError("timestamp contains missing values")

    if out["symbol"].isna().any() or out["symbol"].astype(str).str.strip().eq("").any():
        raise ValueError("symbol contains missing or empty values")

    out["adj_close"] = pd.to_numeric(
        out["adj_close"],
        errors="raise",
    )

    duplicate_key = out.duplicated(
        subset=["timestamp", "symbol"],
        keep=False,
    )

    if duplicate_key.any():
        raise ValueError("duplicate timestamp-symbol keys found")

    observed_prices = out["adj_close"].dropna()
    if (observed_prices <= 0).any():
        raise ValueError("observed adjusted prices must be strictly positive")

    return (
        out.sort_values(["timestamp", "symbol"], kind="stable")
        .reset_index(drop=True)
    )


def compute_log_returns(raw):
    """Create a wide log-return panel from validated adjusted prices."""
    quotes = validate_and_sort_quotes(raw)

    prices = quotes.pivot(
        index="timestamp",
        columns="symbol",
        values="adj_close",
    )

    # Missing prices remain missing. No forward filling occurs here.
    return np.log(prices / prices.shift(1))

Several design choices are intentional:

• Duplicate keys fail immediately. If a provider sends two adjusted closes for the same timestamp and symbol, a separate reconciliation policy may eventually choose a preferred record. But silently selecting one inside a return function is dangerous. The policy must be explicit and testable.
• Missing adjusted prices are permitted. A missing value is not necessarily invalid data. It can represent a suspension, unavailable observation, or incomplete vendor history. The return calculation must preserve uncertainty rather than create a price.
• Observed prices must be positive. The logarithm in

$$r_t=\log \left(\frac{P_t}{P_{t-1}}\right)$$

requires positive prices.

• Returns use adj_close. This protects against mechanical discontinuities caused by splits and other corporate actions, provided the vendor’s adjustment methodology is appropriate for the research question.
• The input is copied. Validation should not silently mutate a DataFrame owned by some other stage of your pipeline.

---

Turn known market-data failures into tests

Now write a test suite that makes those assumptions executable.

# tests/test_transforms.py

import numpy as np
import pandas as pd
import pytest

from pandas.testing import assert_frame_equal, assert_series_equal

from market_data.transforms import (
    compute_log_returns,
    validate_and_sort_quotes,
)


@pytest.fixture
def valid_quotes():
    """A small, intentionally unsorted long-form price table."""
    return pd.DataFrame(
        {
            "timestamp": [
                "2024-01-03",
                "2024-01-02",
                "2024-01-03",
                "2024-01-02",
            ],
            "symbol": ["BBB", "BBB", "AAA", "AAA"],
            "adj_close": [51.0, 50.0, 101.0, 100.0],
        }
    )


def test_validation_sorts_and_normalizes_timestamps(valid_quotes):
    # Arrange
    expected = pd.DataFrame(
        {
            "timestamp": pd.to_datetime(
                [
                    "2024-01-02",
                    "2024-01-02",
                    "2024-01-03",
                    "2024-01-03",
                ],
                utc=True,
            ),
            "symbol": ["AAA", "BBB", "AAA", "BBB"],
            "adj_close": [100.0, 50.0, 101.0, 51.0],
        }
    )

    # Act
    actual = validate_and_sort_quotes(valid_quotes)

    # Assert
    assert_frame_equal(actual, expected)


def test_duplicate_timestamp_symbol_is_rejected(valid_quotes):
    # Arrange
    duplicated = pd.concat(
        [
            valid_quotes,
            pd.DataFrame(
                {
                    "timestamp": ["2024-01-03"],
                    "symbol": ["AAA"],
                    "adj_close": [102.0],
                }
            ),
        ],
        ignore_index=True,
    )

    # Act and Assert
    with pytest.raises(ValueError, match="duplicate timestamp-symbol"):
        validate_and_sort_quotes(duplicated)


@pytest.mark.parametrize("bad_price", [0.0, -5.0])
def test_nonpositive_adjusted_price_is_rejected(bad_price):
    # Arrange
    raw = pd.DataFrame(
        {
            "timestamp": ["2024-01-02"],
            "symbol": ["AAA"],
            "adj_close": [bad_price],
        }
    )

    # Act and Assert
    with pytest.raises(ValueError, match="strictly positive"):
        validate_and_sort_quotes(raw)


def test_return_uses_adjusted_close_not_raw_close():
    # Arrange
    # Raw close halves because of a two-for-one split.
    # Adjusted close correctly represents no economic price movement.
    raw = pd.DataFrame(
        {
            "timestamp": ["2024-01-02", "2024-01-03"],
            "symbol": ["AAA", "AAA"],
            "close": [100.0, 50.0],
            "adj_close": [100.0, 100.0],
        }
    )

    # Act
    returns = compute_log_returns(raw)

    # Assert
    assert np.isclose(returns["AAA"].iloc[1], 0.0)


def test_missing_price_does_not_create_a_zero_return():
    # Arrange
    raw = pd.DataFrame(
        {
            "timestamp": [
                "2024-01-02",
                "2024-01-03",
                "2024-01-04",
            ],
            "symbol": ["AAA", "AAA", "AAA"],
            "adj_close": [100.0, np.nan, 121.0],
        }
    )

    expected_index = pd.DatetimeIndex(
        pd.to_datetime(
            ["2024-01-02", "2024-01-03", "2024-01-04"],
            utc=True,
        ),
        name="timestamp",
    )

    expected = pd.Series(
        [np.nan, np.nan, np.nan],
        index=expected_index,
        name="AAA",
    )

    # Act
    actual = compute_log_returns(raw)["AAA"]

    # Assert
    assert_series_equal(actual, expected, check_freq=False)

Run the suite from the project root:

python -m pytest -q

The tests are designed to detect real errors, not merely to increase a coverage number:

Test	Bug it would catch
test_validation_sorts_and_normalizes_timestamps	Non-reproducible ordering or inconsistent timestamp handling
test_duplicate_timestamp_symbol_is_rejected	Ambiguous observations entering a pivot or return calculation
test_nonpositive_adjusted_price_is_rejected	Invalid logarithms and corrupted returns
test_return_uses_adjusted_close_not_raw_close	Using unadjusted prices across a corporate action
test_missing_price_does_not_create_a_zero_return	Forward filling that fabricates an observed return

The last test is especially valuable. A seemingly convenient implementation such as filling prices before returns would convert the missing price into an apparent period of zero movement or produce a return across a gap. Either outcome can distort realized volatility, correlations, factor estimates, and a later backtest.

Notice also the use of @pytest.mark.parametrize. Rather than maintaining separate tests for zero and negative prices, one behavioral statement is tested against two invalid cases. This is appropriate when the inputs differ but the expected policy is identical.

---

Compare DataFrames deliberately

A DataFrame is more than a collection of visible values. It includes index labels, column names, row order, dtypes, and sometimes time-index metadata. Using a generic equality check can produce confusing results or overlook a meaningful structural mismatch.

For the sorting test, assert_frame_equal(actual, expected) is intentionally strict. A wrong timestamp dtype, unexpected index, reordered row, or changed column name should be treated as a failure because downstream time-series code may depend on exactly those details.

For numerical outputs, strict equality is not always appropriate. Floating-point operations can differ slightly because of operation order or platform-level numerical details. When testing an algorithm with nontrivial floating-point calculations, state a justified tolerance:

assert_frame_equal(
    actual,
    expected,
    rtol=1e-10,
    atol=1e-12,
)

The tolerance is part of the scientific claim. A tolerance of $10^{-2}$ could hide a material pricing or risk error; a tolerance of $10^{-12}$ may be unnecessarily strict for a simulation. Choose it according to the scale and purpose of the calculation.

Do not use check_dtype=False as a default escape hatch. It is reasonable only when the behavior under test concerns values and storage type is truly irrelevant. For example, a downstream optimizer may require float64; in that case, dtype is part of the contract and should remain checked.

---

Test failures are useful research feedback

A passing test suite does not prove that your data is economically correct. It proves that the code conforms to the assumptions you chose to encode. That distinction matters in quantitative work.

For instance:

• A test can prove that every observed adjusted price is positive.
• It cannot prove that the vendor adjusted every dividend correctly.
• A test can prove that timestamps are unique after your reconciliation rule.
• It cannot prove that the source did not omit an entire trading session.
• A test can prove that missing prices are not silently forward-filled.
• It cannot decide whether a missing value should lead to exclusion, delayed ingestion, or a separate imputation model.

Therefore, maintain two related layers:

1. Unit tests use tiny, hand-built fixtures to protect transformation logic.
2. Runtime data validations inspect every actual incoming dataset and stop or flag the pipeline when a contract is violated.

For a portfolio repository, a strong practice is to add a test whenever you discover a real data bug. Suppose a NIFTY constituent changes ticker, a vendor unexpectedly returns duplicate records, or a holiday alignment bug appears. First reduce the incident to the smallest possible DataFrame. Then write a failing test that reproduces it. Finally, fix the function until the test passes. The test becomes permanent evidence that the same failure should not quietly return.

A quick quality check for any proposed test is this: Can you state, in one sentence, the real-world error it would catch? If not, the test may be coupled to an implementation detail rather than protecting a research-relevant behavior.

---

Key takeaways

Unit tests make market-data assumptions visible, repeatable, and enforceable.

• Keep market transformations as small, deterministic functions operating on in-memory DataFrames.
• Use a timestamp–symbol pair as an explicit uniqueness contract for long-form market data.
• Reject invalid observed prices before calculating log returns.
• Test that adjusted prices, rather than raw closes, drive return calculations.
• Treat missing observations carefully: do not forward-fill merely to obtain a rectangular panel.
• Use pytest fixtures for reusable sample DataFrames and parametrization for families of edge cases.
• Use assert_frame_equal and assert_series_equal when structural equality matters, with explicit numerical tolerances only when justified.
• Convert every meaningful past data incident into a small regression test.

Next, Module 2 begins with statistical inference for financial data. You will model correlated asset returns with multivariate distributions and Cholesky-based simulation, using the clean, validated return panels that this workflow is designed to protect.