yfinance Adjusted Close: Diagnose auto_adjust and Missing Columns

KeyError: 'Adj Close'

flowchart TD
    A["Raw Market Feed from Yahoo Finance"] --> B{"Download Method & Parameters"}
    B -->|"auto_adjust=False\n(yf.download legacy)"| C["Two Separate Price Columns\n• 'Close': Raw unadjusted trade price\n• 'Adj Close': Dividend & split-adjusted\n• 'Volume': Raw recorded share volume"]
    B -->|"auto_adjust=True\n(Ticker.history default)"| D["Integrated Adjusted Columns\n• 'Close': Already back-adjusted\n• 'Adj Close': OMITTED entirely\n• 'Volume': Inversely scaled for splits"]
    C --> E{"MultiIndex Column Structure?"}
    D --> E
    E -->|"yfinance >= 0.2.x default"| F["MultiIndex: (Price, Ticker)\ne.g. ('Close', 'AAPL')\nDirect df['Close'] fails or warns"]
    E -->|"Flattened / Single-level"| G["Standard 1D Index\nClean column access"]
    F --> H["Strict Normalization & Schema Validation"]
    G --> H

┌────────────────────────────────────────────────────────────────────────────────────────┐
│                   YFINANCE DOWNLOAD CONFIGURATION MATRIX                               │
├────────────────────┬──────────────────┬─────────────────────────────┬──────────────────┤
│ Method / Flag      │ Default Setting  │ 'Close' Column Content       │ 'Adj Close' Col? │
├────────────────────┼──────────────────┼─────────────────────────────┼──────────────────┤
│ yf.download()      │ auto_adjust=False│ Raw unadjusted traded price │ Present          │
│ Ticker.history()   │ auto_adjust=True │ Back-adjusted (split + div) │ OMITTED (None)   │
│ Any (auto_adj=True)│ Explicit True    │ Back-adjusted (split + div) │ OMITTED (None)   │
│ Any (auto_adj=False│ Explicit False   │ Raw unadjusted traded price │ Present          │
└────────────────────┴──────────────────┴─────────────────────────────┴──────────────────┘

┌────────────────────────────────────────────────────────────────────────────────────────┐
│                   CORPORATE ACTION ADJUSTMENT MECHANICS                                │
├─────────────────────┬───────────────────────────────┬──────────────────────────────────┤
│ Attribute           │ Stock Splits (e.g. 2-for-1)   │ Cash Dividends (e.g. $0.50/share)│
├─────────────────────┼───────────────────────────────┼──────────────────────────────────┤
│ Economic Impact     │ Neutral share redenomination  │ Real capital extraction from firm│
│ Historical Price    │ Divided by split factor (S)   │ Multiplied by (1 - D / P_close)  │
│ Historical Volume   │ Multiplied by split factor (S)│ Unchanged (shares unaffected)    │
│ Cash Balance Change │ Zero                          │ Positive cash credit to holder   │
│ Exchange Trading    │ Unadjusted price invalidates  │ Unadjusted price reflects real   │
│ Reality             │ chart continuity              │ historical order book prints     │
└─────────────────────┴───────────────────────────────┴──────────────────────────────────┘

Price_adjusted = Price_raw / S
Shares_adjusted = Shares_raw * S

Adjustment Factor (CRSP Standard):
k_t = 1 - (D_t / P_close,t-1)

P_adjusted,tau = P_raw,tau * Product_{t > tau} (k_t)

Volume_adjusted = Volume_raw * S

Pre-Split Bar:  Price = $400, Volume = 10,000,000  (Dollar Volume = $4.0B)
                                     │
                    4-for-1 Stock Split Event
                                     ▼
auto_adjust=False:  Close = $400, Adj Close = $100, Volume = 10,000,000
                    (Volume matches raw tape; mismatched to Adj Close)

auto_adjust=True:   Close = $100, Adj Close = [OMITTED], Volume = 40,000,000
                    (Volume scaled to match adjusted price; turnover preserved)

Day t-1 (Raw):  Close = $200.00,  50 SMA = $195.00  ──> Trend Bullish
Day t   (Raw):  Close = $100.00   (2-for-1 Split)
Day t   (SMA):  50 SMA = $193.10  ──> FALSE BEARISH BREAKDOWN SIGNAL!

# ANTI-PATTERN: Mixing unadjusted Close and Adj Close
df["SMA_200"] = df["Adj Close"].rolling(200).mean()
df["Signal"] = df["Close"] > df["SMA_200"]  # FATAL ERROR

Limit Buy Order submitted at $50.00.
Historical bar: Low = $48.20 (Dividend-adjusted).
Backtest engine verdict: ORDER FILLED.

# Modern yf.download('AAPL') structure:
# Price     Adj Close      Close       High        Low       Open     Volume
# Ticker         AAPL       AAPL       AAPL       AAPL       AAPL       AAPL
# Date
# 2024-01-02   185.64     185.64     188.44     183.89     187.15   82488700

"""Robust yfinance Data Ingestion and Normalization Pipeline.

Enforces deterministic corporate action handling, MultiIndex flattening,
and strict schema validation for quantitative backtesting.
"""

from typing import Literal
import numpy as np
import pandas as pd


def fetch_and_normalize_ohlcv(
    ticker: str,
    start_date: str,
    end_date: str,
    adjustment_mode: Literal["split_and_dividend", "split_only", "raw_with_adj"] = "split_and_dividend",
) -> pd.DataFrame:
    """Fetch equity data from Yahoo Finance and normalize into a strict OHLCV schema.

    Parameters
    ----------
    ticker : str
        Exchange ticker symbol (e.g. 'AAPL', 'MSFT').
    start_date : str
        ISO start date 'YYYY-MM-DD'.
    end_date : str
        ISO end date 'YYYY-MM-DD'.
    adjustment_mode : str
        - 'split_and_dividend': auto_adjust=True. Close is fully adjusted;
          Open/High/Low/Volume scaled. Ideal for total-return vectorized models.
        - 'raw_with_adj': auto_adjust=False. Preserves raw Close alongside Adj Close.
        - 'split_only': Reconstructs split-adjusted prices while keeping cash dividends
          isolated for explicit accounting.

    Returns
    -------
    pd.DataFrame
        Cleaned, single-level indexed DataFrame with validated OHLCV columns.
    """
    try:
        import yfinance as yf
    except ImportError as exc:
        raise ImportError("yfinance must be installed: pip install yfinance") from exc

    # Configure auto_adjust flag based on desired operational mode
    auto_adj = adjustment_mode == "split_and_dividend"

    # Download data with explicit parameters
    raw_df = yf.download(
        tickers=ticker,
        start=start_date,
        end=end_date,
        auto_adjust=auto_adj,
        progress=False,
    )

    if raw_df.empty:
        raise ValueError(f"No market data returned for {ticker} between {start_date} and {end_date}.")

    df = raw_df.copy()

    # 1. Flatten MultiIndex columns if present (yfinance >= 0.2.x behavior)
    if isinstance(df.columns, pd.MultiIndex):
        # Extract Level 0 (price field) if Level 1 matches our single ticker
        if df.columns.nlevels == 2:
            df.columns = df.columns.get_level_values(0)

    # Clean whitespace and standardize string types
    df.columns = [str(col).strip() for col in df.columns]

    # 2. Schema Normalization based on adjustment_mode
    if adjustment_mode == "split_and_dividend":
        required_cols = ["Open", "High", "Low", "Close", "Volume"]
        missing = [c for c in required_cols if c not in df.columns]
        if missing:
            raise KeyError(f"Missing required columns for split_and_dividend: {missing}")

        # Ensure Adj Close is not present to avoid dual-series confusion
        if "Adj Close" in df.columns:
            df = df.drop(columns=["Adj Close"])

    elif adjustment_mode == "raw_with_adj":
        required_cols = ["Open", "High", "Low", "Close", "Adj Close", "Volume"]
        missing = [c for c in required_cols if c not in df.columns]
        if missing:
            raise KeyError(
                f"Missing required columns for raw_with_adj: {missing}. "
                "Ensure auto_adjust was set to False."
            )

    elif adjustment_mode == "split_only":
        # When calculating split-only adjustments from raw + adj:
        # We fetch actions to calculate the cumulative split factor
        t = yf.Ticker(ticker)
        actions = t.actions
        splits = actions["Stock Splits"] if "Stock Splits" in actions else pd.Series(dtype=float)

        # Standardize raw bars
        if "Adj Close" in df.columns and "Close" in df.columns:
            # Reconstruct split multiplier by walking backwards
            df["Split_Factor"] = 1.0
            if not splits.empty:
                # Filter splits within target range
                sub_splits = splits[(splits.index >= df.index.min()) & (splits.index <= df.index.max())]
                for split_date, ratio in sub_splits.items():
                    if ratio > 0:
                        # Prior to split, prices were divided by ratio
                        df.loc[df.index < split_date, "Split_Factor"] *= (1.0 / ratio)

            # Apply pure split adjustment
            for col in ["Open", "High", "Low", "Close"]:
                df[col] = df[col] * df["Split_Factor"]
            df["Volume"] = df["Volume"] / df["Split_Factor"]
            df = df.drop(columns=["Split_Factor"])

    # 3. Defensive Data Validation Layer
    _validate_bar_integrity(df, ticker)

    # Standardize column naming: enforce canonical lowercase for backtesting engines
    df = df[["Open", "High", "Low", "Close", "Volume"] + ([c for c in ["Adj Close"] if c in df.columns])]
    df.index.name = "timestamp"

    return df


def _validate_bar_integrity(df: pd.DataFrame, ticker: str) -> None:
    """Validate mathematical and temporal consistency of OHLCV bars."""
    # Ensure index is monotonic increasing datetime
    if not df.index.is_monotonic_increasing:
        raise ValueError(f"Temporal anomaly: DatetimeIndex for {ticker} is not strictly monotonic.")

    # Check for NaN / null values
    null_counts = df.isnull().sum()
    if null_counts.any():
        bad_cols = null_counts[null_counts > 0].to_dict()
        raise ValueError(f"Data corruption: NaN values detected in {ticker}: {bad_cols}")

    # Check basic OHLC mathematical relationships: High must be the highest, Low the lowest
    high_violation = (df["High"] < df["Low"]) | (df["High"] < df["Open"]) | (df["High"] < df["Close"])
    if high_violation.any():
        bad_dates = df.index[high_violation].strftime("%Y-%m-%d").tolist()
        raise ValueError(f"Invalid bar geometry: High < max(Open, Close, Low) on dates: {bad_dates[:5]}")

    low_violation = (df["Low"] > df["Open"]) | (df["Low"] > df["Close"])
    if low_violation.any():
        bad_dates = df.index[low_violation].strftime("%Y-%m-%d").tolist()
        raise ValueError(f"Invalid bar geometry: Low > min(Open, Close) on dates: {bad_dates[:5]}")

    # Check for non-negative prices and volume
    if (df[["Open", "High", "Low", "Close"]] <= 0).any().any():
        raise ValueError(f"Non-positive prices detected in {ticker}.")
    if (df["Volume"] < 0).any():
        raise ValueError(f"Negative volume recorded in {ticker}.")


# =====================================================================
# Synthetic Verification Test & Output Demonstration
# =====================================================================
if __name__ == "__main__":
    print("Testing Ingestion Logic with Synthetic Data Scenario...\n")

    # Generate synthetic multi-index DataFrame mimicking yf.download output
    dates = pd.date_range("2024-01-02", periods=4, freq="B")
    tuples = [
        ("Close", "TEST"),
        ("High", "TEST"),
        ("Low", "TEST"),
        ("Open", "TEST"),
        ("Volume", "TEST"),
    ]
    columns = pd.MultiIndex.from_tuples(tuples, names=["Price", "Ticker"])
    data = [
        [185.64, 188.44, 183.89, 187.15, 82488700],
        [184.25, 185.88, 183.43, 184.22, 58414500],
        [181.91, 183.09, 180.88, 182.15, 71983600],
        [181.18, 182.76, 180.17, 182.00, 62303300],
    ]
    mock_yf_download = pd.DataFrame(data, index=dates, columns=columns)

    print("--- 1. Raw Simulated MultiIndex Input ---")
    print(mock_yf_download.head(2))

    # Demonstrate flattening and validation
    df_clean = mock_yf_download.copy()
    if isinstance(df_clean.columns, pd.MultiIndex):
        df_clean.columns = df_clean.columns.get_level_values(0)

    _validate_bar_integrity(df_clean, "TEST")
    print("\n--- 2. Normalized Flat Single-Level DataFrame ---")
    print(df_clean)

    print("\n--- 3. Validation Check Results ---")
    print("✓ Monotonic DatetimeIndex: Passed")
    print("✓ NaN Bar Audit: Zero missing values detected")
    print("✓ Candlestick Geometry: High >= max(Open, Close) and Low <= min(Open, Close)")
    print("✓ Volume non-negativity: Passed")

Testing Ingestion Logic with Synthetic Data Scenario...

--- 1. Raw Simulated MultiIndex Input ---
Price        Close      High       Low      Open    Volume
Ticker        TEST      TEST      TEST      TEST      TEST
2024-01-02  185.64    188.44    183.89    187.15  82488700
2024-01-03  184.25    185.88    183.43    184.22  58414500

--- 2. Normalized Flat Single-Level DataFrame ---
             Close    High     Low    Open    Volume
2024-01-02  185.64  188.44  183.89  187.15  82488700
2024-01-03  184.25  185.88  183.43  184.22  58414500
2024-01-04  181.91  183.09  180.88  182.15  71983600
2024-01-05  181.18  182.76  180.17  182.00  62303300

--- 3. Validation Check Results ---
✓ Monotonic DatetimeIndex: Passed
✓ NaN Bar Audit: Zero missing values detected
✓ Candlestick Geometry: High >= max(Open, Close) and Low <= min(Open, Close)
✓ Volume non-negativity: Passed

flowchart LR
    A["yfinance Normalized Feed"] --> B["Split-Adjusted OHLCV Bars\nPreserves continuous chart geometry"]
    A --> C["Corporate Actions Ledger\nCash dividends isolated as discrete events"]
    B --> D["OpenBacktesting Engine (0.2.1)\n• Bar-by-bar causal matching\n• Explicit fee & slippage modeling"]
    C --> D
    D --> E["Inspectable SQLite Trial Store"]
    E --> F["Local Studio UI Analysis\nAudit fills, equity curves, drawdown"]

┌────────────────────────────────────────────────────────────────────────────────────────┐
│                   MARKET DATA SOURCE COMPARISON MATRIX                                 │
├────────────────────┬─────────────────────────────┬─────────────────────────────────────┤
│ Dimension          │ yfinance (Scraped / Unofficial)│ Institutional Feeds (CRSP / Norgate)│
├────────────────────┼─────────────────────────────┼─────────────────────────────────────┤
│ Cost & Access      │ Free, public community tool │ Paid subscription with SLA          │
│ Licensing Terms    │ Yahoo Terms of Service:     │ Explicit commercial and live        │
│                    │ Non-commercial personal use │ trading redistribution rights       │
│ Upstream Stability │ Unannounced API HTML shifts │ Versioned, stable binary/JSON APIs  │
│ Survivorship Bias  │ Severe (delisted stocks are │ Addressed (point-in-time historical │
│                    │ purged from ticker database)│ constituent databases maintained)   │
│ Corporate Restates │ Inconsistent dividend/split │ Audited, timestamped corporate      │
│                    │ retro-adjustments           │ action distribution records         │
└────────────────────┴─────────────────────────────┴─────────────────────────────────────┤

┌────────────────────────────────────────────────────────────────────────────────────────┐
│                     PRICE ADJUSTMENT VERIFICATION CHECKLIST                            │
├────┬──────────────────────────────────────────┬────────────────────────────────────────┤
│ No │ Audit Step                               │ Verification Criteria                  │
├────┼──────────────────────────────────────────┼────────────────────────────────────────┤
│ 1  │ Parameter Audit                          │ Did you explicitly pass auto_adjust=True│
│    │                                          │ or auto_adjust=False?                  │
│ 2  │ Column Existence Check                   │ If auto_adjust=True, confirm your code │
│    │                                          │ does not query df['Adj Close'].        │
│ 3  │ MultiIndex Flattening                    │ Is df.columns a 1D Index? Have you     │
│    │                                          │ stripped (Price, Ticker) tuples?       │
│ 4  │ Indicator Consistency                    │ Are indicators computed on the same    │
│    │                                          │ price series used for entry triggers?  │
│ 5  │ Split Geometry Inspection                │ Across historic split dates, does raw  │
│    │                                          │ Close cliff while adjusted Close stays │
│    │                                          │ continuous?                            │
│ 6  │ Volume Reconciliation                    │ If using adjusted prices, is volume    │
│    │                                          │ scaled inversely by the split factor?  │
│ 7  │ Dividend Cash Flow Segregation           │ Are cash dividends credited to portfolio│
│    │                                          │ cash, rather than double-counted?      │
│ 8  │ Bar Geometry Sanity                      │ Does High >= max(Open, Close) and      │
│    │                                          │ Low <= min(Open, Close) hold true?     │
└────┴──────────────────────────────────────────┴────────────────────────────────────────┘