import pandas as pd
 
# Each operation modifies the DataFrame in place
df['Return'] = df['Close'] / df['Close'].shift(1) - 1
df['ReturnZ'] = (df['Return'] - df['Return'].mean()) / df['Return'].std()
df = df[df['Volume'] > 1000000]  # Filter modifies DataFrame

import polars as pl
 
# All operations are expressions that compose together
df = df.with_columns([
    ((pl.col("Close") / pl.col("Close").shift(1)) - 1).alias("Return"),
    ((pl.col("Return") - pl.col("Return").mean()) / pl.col("Return").std()).alias("ReturnZ")
]).filter(pl.col("Volume") > 1_000_000)

# String operations on ticker symbols
df = df.with_columns([
    pl.col("Ticker").str.to_uppercase().alias("TickerUpper"),
    pl.col("Ticker").str.slice(0, 3).alias("TickerPrefix"),
    pl.col("Ticker").str.contains("ETF").alias("IsETF"),
    pl.col("Ticker").str.replace(" ", "-").alias("TickerNormalized"),
])
 
# Extract patterns (e.g., extract exchange from ticker format "AAPL:NASDAQ")
df = df.with_columns([
    pl.col("Ticker").str.split(":").list.get(1).alias("Exchange")
])

# Aggregate daily prices to monthly returns (respects calendar month boundaries)
monthly_returns = (
    prices.sort(["Ticker", "Date"])
    .group_by_dynamic(
        index_column="Date",
        every="1mo",           # Calendar month boundaries
        group_by="Ticker"       # Per-asset aggregation
    )
    .agg([
        pl.col("Close").last().alias("MonthClose"),
        pl.col("Return").sum().alias("MonthReturn"),
    ])
)
 
# Quarterly aggregation for earnings alignment
quarterly = prices.group_by_dynamic(
    "Date", 
    every="3mo",  # Calendar quarters
    group_by="Ticker"
).agg([...])

# Rolling statistics per ticker
features = prices.with_columns([
    # 20-day moving average
    pl.col("Close").rolling_mean(20).over("Ticker").alias("MA20"),
    
    # 20-day rolling volatility
    pl.col("Return").rolling_std(20).over("Ticker").alias("Vol20"),
    
    # 10-day rolling high (for breakout detection)
    pl.col("Close").rolling_max(10).over("Ticker").alias("High10"),
]).collect()

# Rolling statistics example
df = pl.read_csv("data/finance.csv")
df_rolling = (
    df.select(['date_id', 'V1'])
    .with_columns([
        pl.col('V1').cast(pl.Float64, strict=False).alias('V1_float')
    ])
    .filter(pl.col('V1_float').is_not_null())
    .head(100)
    .with_columns([
        pl.col('V1_float').rolling_mean(10).alias('MA10'),
        pl.col('V1_float').rolling_std(10).alias('STD10'),
    ])
)
print(df_rolling.head(20))

shape: (20, 5)
┌─────────┬──────────────────────┬──────────┬──────────┬───────┐
│ date_id ┆ V1                   ┆ V1_float ┆ MA10     ┆ STD10 │
│ ---     ┆ ---                  ┆ ---      ┆ ---     ┆ ---   │
│ i64     ┆ str                  ┆ f64      ┆ f64      ┆ f64   │
╞═════════╪══════════════════════╪══════════╪══════════╪═══════╡
│ 1006    ┆ 0.000661375661375661 ┆ 0.000661 ┆ null     ┆ null  │
│ 1007    ┆ 0.000661375661375661 ┆ 0.000661 ┆ null     ┆ null  │
│ …       ┆ …                    ┆ …        ┆ …        ┆ …     │
│ 1021    ┆ 0.000661375661375661 ┆ 0.000661 ┆ 0.000661 ┆ 0.0   │
│ 1022    ┆ 0.000661375661375661 ┆ 0.000661 ┆ 0.000661 ┆ 0.0   │
└─────────┴──────────────────────┴──────────┴──────────┴───────┘

# Load data with Date column (e.g., prices_daily.csv)
prices = pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
 
# Extract calendar components
prices = prices.with_columns([
    pl.col("Date").dt.year().alias("Year"),
    pl.col("Date").dt.quarter().alias("Quarter"),
    pl.col("Date").dt.month().alias("Month"),
    pl.col("Date").dt.weekday().alias("Weekday"),  # 1=Monday, 7=Sunday
    pl.col("Date").dt.strftime("%Y-%m").alias("YearMonth"),
])
 
# Filter by fiscal periods
q4_data = prices.filter(pl.col("Date").dt.quarter() == 4)

from polars import date
 
# Load data with Date column (e.g., prices_daily.csv)
prices = pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
 
# Filter recent data (e.g., for backtesting recent models)
recent = prices.filter(pl.col("Date") >= date(2020, 1, 1))
 
# Filter a specific historical period (e.g., a market cycle)
past_cycle = prices.filter(
    pl.col("Date").is_between(date(2018, 1, 1), date(2020, 12, 31))
)

# Load data with Date columns
prices = pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
earnings = pl.read_csv("data/earnings.csv", try_parse_dates=True)
 
# Join earnings announcements to prices (only use info available at that time)
prices_with_earnings = (
    prices.sort("Date")
    .join_asof(
        earnings.sort("AnnouncementDate"),
        left_on="Date",
        right_on="AnnouncementDate",
        by="Ticker",
        strategy="backward",  # Use most recent earnings before/on this date
        tolerance=pl.duration(days=30),  # Only match within 30 days
    )
)

# Build monthly features with rolling statistics
monthly_features = (
    prices.sort(["Ticker", "Date"])
    .with_columns([
        # Rolling features
        pl.col("Return").rolling_mean(20).over("Ticker").alias("MA20"),
        pl.col("Return").rolling_std(20).over("Ticker").alias("Vol20"),
        # Calendar features
        pl.col("Date").dt.quarter().alias("Quarter"),
    ])
    .group_by_dynamic("Date", every="1mo", group_by="Ticker")
    .agg([
        pl.col("MA20").last().alias("MonthMA20"),
        pl.col("Vol20").mean().alias("MonthVol20"),
    ])
)

pip install polars
# or
uv add polars

# Step 1: Load fact table (prices) - your core data
prices = pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
 
# Step 2: Join dimension table (company info) for cross-sectional features
company_info = pl.read_csv("data/company_info.csv")
prices_with_metadata = prices.join(company_info, on="Ticker", how="left")
 
# Step 3: Join event table (dividends) only when needed for total returns
dividends = (
    pl.read_csv("data/corporate_actions.csv", try_parse_dates=True)
    .filter(pl.col("ActionType") == "dividend")
    .select(["Date", "Ticker", "Value"])
    .rename({"Value": "Dividend"})
)
 
# Join events with time-aware join (no look-ahead bias)
features = (
    prices_with_metadata
    .join_asof(
        dividends.sort("Date"),
        left_on="Date",
        right_on="Date",
        by="Ticker",
        strategy="backward"
    )
    .with_columns(pl.col("Dividend").fill_null(0.0))
    .with_columns([
        # Total return = price return + dividend yield
        (pl.col("Return") + pl.col("Dividend") / pl.col("Close")).alias("TotalReturn")
    ])
)

import polars as pl
 
df = pl.read_csv("./data/finance.csv")
print(df.head())
print(df.describe())

shape: (5, 98)
┌─────────┬─────┬─────┬─────┬───┬──────┬─────────────────┬────────────────┬───────────────────────────────┐
│ date_id ┆ D1  ┆ D2  ┆ D3  ┆ … ┆ V9   ┆ forward_returns ┆ risk_free_rate ┆ market_forward_excess_returns │
│ ---     ┆ --- ┆ --- ┆ --- ┆   ┆ ---  ┆ ---             ┆ ---            ┆ ---                           │
│ i64     ┆ i64 ┆ i64 ┆ i64 ┆   ┆ str  ┆ f64             ┆ f64            ┆ f64                           │
╞═════════╪═════╪═════╪═════╪═══╪══════╪═════════════════╪════════════════╪═══════════════════════════════╡
│ 0       ┆ 0   ┆ 0   ┆ 0   ┆ … ┆ null ┆ -0.002421       ┆ 0.000301       ┆ -0.003038                     │
│ 1       ┆ 0   ┆ 0   ┆ 0   ┆ … ┆ null ┆ -0.008495       ┆ 0.000303       ┆ -0.009114                     │
│ 2       ┆ 0   ┆ 0   ┆ 0   ┆ … ┆ null ┆ -0.009624       ┆ 0.000301       ┆ -0.010243                     │
│ 3       ┆ 0   ┆ 0   ┆ 0   ┆ … ┆ null ┆ 0.004662        ┆ 0.000299       ┆ 0.004046                      │
│ 4       ┆ 0   ┆ 0   ┆ 0   ┆ … ┆ null ┆ -0.011686       ┆ 0.000299       ┆ -0.012301                     │
└─────────┴─────┴─────┴─────┴───┴──────┴─────────────────┴────────────────┴───────────────────────────────┘
 
Basic statistics (describe):
┌────────────┬─────────────┬──────────┬──────────┬───┬────────────────┬────────────────┬───────────────┐
│ statistic  ┆ date_id     ┆ D1       ┆ D2       ┆ … ┆ forward_return ┆ risk_free_rat ┆ market_forwar │
│ ---        ┆ ---         ┆ ---      ┆ ---      ┆   ┆ s              ┆ e             ┆ d_excess_retu │
│ str        ┆ f64         ┆ f64      ┆ f64      ┆   ┆ ---            ┆ ---           ┆ rns           │
│            ┆             ┆          ┆          ┆   ┆ f64            ┆ f64           ┆ ---           │
│            ┆             ┆          ┆          ┆   ┆                ┆               ┆ f64           │
╞════════════╪═════════════╪══════════╪══════════╪═══╪════════════════╪═══════════════╪═══════════════╡
│ count      ┆ 9021.0      ┆ 9021.0   ┆ 9021.0   ┆ … ┆ 9021.0         ┆ 9021.0        ┆ 9021.0        │
│ null_count ┆ 0.0         ┆ 0.0      ┆ 0.0      ┆ … ┆ 0.0            ┆ 0.0           ┆ 0.0           │
│ mean       ┆ 4510.0      ┆ 0.031593 ┆ 0.031593 ┆ … ┆ 0.000471       ┆ 0.000107      ┆ 0.000053      │
│ std        ┆ 2604.28     ┆ 0.174923 ┆ 0.174923 ┆ … ┆ 0.010541       ┆ 0.000088      ┆ 0.010558      │
│ min        ┆ 0.0         ┆ 0.0      ┆ 0.0      ┆ … ┆ -0.039754      ┆ -0.000004     ┆ -0.040582     │
│ 25%        ┆ 2255.0      ┆ 0.0      ┆ 0.0      ┆ … ┆ -0.004319      ┆ 0.000008      ┆ -0.004747     │
│ 50%        ┆ 4510.0      ┆ 0.0      ┆ 0.0      ┆ … ┆ 0.000659       ┆ 0.000097      ┆ 0.000255      │
│ 75%        ┆ 6765.0      ┆ 0.0      ┆ 0.0      ┆ … ┆ 0.005896       ┆ 0.000193      ┆ 0.005479      │
│ max        ┆ 9020.0      ┆ 1.0      ┆ 1.0      ┆ … ┆ 0.040661       ┆ 0.000317      ┆ 0.040551      │
└────────────┴─────────────┴──────────┴──────────┴───┴────────────────┴────────────────┴───────────────┘

# Select specific columns by name
df.select(["Date", "Ticker", "Close", "Volume"])
 
# Select with expressions (compute new columns on the fly)
df.select([
    pl.col("Close"),
    pl.col("Volume"),
    (pl.col("Close") * pl.col("Volume")).alias("DollarVolume"),
    pl.col("Date").dt.year().alias("Year"),
])
 
# Select all columns except some
df.select([pl.all().exclude(["UnwantedCol1", "UnwantedCol2"])])

# Simple filter
df.filter(pl.col("Volume") > 1_000_000)
 
# Multiple conditions (use parentheses for clarity)
df.filter(
    (pl.col("Volume") > 1_000_000) & 
    (pl.col("Close") > 50) &
    (pl.col("Date") >= pl.date(2020, 1, 1))
)
 
# Filter with null handling
df.filter(pl.col("Return").is_not_null())
 
# Filter with string operations
df.filter(pl.col("Ticker").str.contains("ETF"))

# Filter first (reduces data size), then select (reduces columns)
recent_liquid = (
    df.filter(
        (pl.col("Date") >= pl.date(2020, 1, 1)) &
        (pl.col("Volume") > 1_000_000) &
        (pl.col("Return").is_not_null())
    )
    .select(["Date", "Ticker", "Close", "Return", "Volume"])
)

# Select specific columns
selected = df.select(["date_id", "V1", "V2", "V3", "D1", "D2"])
print(selected.head())

shape: (5, 6)
┌─────────┬──────┬──────┬──────┬─────┬─────┐
│ date_id ┆ V1   ┆ V2   ┆ V3   ┆ D1  ┆ D2  │
│ ---     ┆ ---  ┆ ---  ┆ ---  ┆ --- ┆ --- │
│ i64     ┆ str  ┆ str  ┆ str  ┆ i64 ┆ i64 │
╞═════════╪══════╪══════╪══════╪═════╪═════╡
│ 0       ┆ null ┆ null ┆ null ┆ 0   ┆ 0   │
│ 1       ┆ null ┆ null ┆ null ┆ 0   ┆ 0   │
│ 2       ┆ null ┆ null ┆ null ┆ 0   ┆ 0   │
│ 3       ┆ null ┆ null ┆ null ┆ 0   ┆ 0   │
│ 4       ┆ null ┆ null ┆ null ┆ 0   ┆ 0   │
└─────────┴──────┴──────┴──────┴─────┴─────┘

# Filter recent data
recent = df.filter(pl.col("date_id") >= 8000)
print(f"Rows after filter: {len(recent)}")
print(recent.select(["date_id", "V1", "V2"]).head())

Rows after filter: 1021
shape: (5, 3)
┌─────────┬───────────────────┬───────────────────┐
│ date_id ┆ V1                ┆ V2                │
│ ---     ┆ ---               ┆ ---               │
│ i64     ┆ str               ┆ str               │
╞═════════╪═══════════════════╪═══════════════════╡
│ 8000    ┆ 0.882275132275132 ┆ 0.853174603174603 │
│ 8001    ┆ 0.863095238095238 ┆ 0.837301587301587 │
│ 8002    ┆ 0.826719576719577 ┆ 0.823412698412698 │
│ 8003    ┆ 0.757936507936508 ┆ 0.80952380952381  │
│ 8004    ┆ 0.756613756613757 ┆ 0.792989417989418 │
└─────────┴───────────────────┴───────────────────┘

# Single column
df = df.with_columns([
    (pl.col("Close") * 2).alias("CloseDoubled")
])
 
# Multiple columns in one pass (more efficient)
df = df.with_columns([
    (pl.col("Close") * 2).alias("CloseDoubled"),
    (pl.col("Close") + pl.col("Open")).alias("ClosePlusOpen"),
    pl.col("Volume").log().alias("LogVolume"),
])

# Simple conditional
df = df.with_columns([
    pl.when(pl.col("Return") > 0)
      .then(1.0)
      .otherwise(0.0)
      .alias("PositiveReturn")
])
 
# Multiple conditions
df = df.with_columns([
    pl.when(pl.col("Return") > 0.02)
      .then(1.0)  # Strong positive
    .when(pl.col("Return") > 0)
      .then(0.5)  # Weak positive
    .when(pl.col("Return") > -0.02)
      .then(-0.5)  # Weak negative
    .otherwise(-1.0)  # Strong negative
    .alias("ReturnRegime")
])

# Compute returns and volatility features
features = prices.sort(["Ticker", "Date"]).with_columns([
    # Daily return
    ((pl.col("Close") / pl.col("Close").shift(1)) - 1)
        .over("Ticker")
        .alias("Return"),
    
    # Log return (for multi-period compounding)
    (pl.col("Close").log() - pl.col("Close").log().shift(1))
        .over("Ticker")
        .alias("LogReturn"),
    
    # Volatility regime (high/low volatility indicator)
    pl.when(pl.col("Return").abs() > 0.02)
      .then(1.0)
      .otherwise(0.0)
      .alias("HighVolRegime"),
    
    # Momentum signal (5-day return)
    ((pl.col("Close") / pl.col("Close").shift(5)) - 1)
        .over("Ticker")
        .alias("Momentum5"),
]).collect()

# Create normalized features (convert strings to float first)
features = df.select(["date_id", "V1", "V2"]).with_columns([
    pl.col("V1").cast(pl.Float64, strict=False).alias("V1"),
    pl.col("V2").cast(pl.Float64, strict=False).alias("V2")
]).with_columns([
    (pl.col("V1") * 2).alias("V1_doubled"),
    (pl.col("V1") + pl.col("V2")).alias("V1_plus_V2"),
    pl.when(pl.col("V1") > 0)
      .then(1.0)
      .otherwise(0.0)
      .alias("V1_positive")
])
print(features.head())

shape: (5, 6)
┌─────────┬──────┬──────┬────────────┬────────────┬─────────────┐
│ date_id ┆ V1   ┆ V2   ┆ V1_doubled ┆ V1_plus_V2 ┆ V1_positive │
│ ---     ┆ ---  ┆ ---  ┆ ---        ┆ ---        ┆ ---         │
│ i64     ┆ f64  ┆ f64  ┆ f64        ┆ f64        ┆ f64         │
╞═════════╪══════╪══════╪════════════╪════════════╪═════════════╡
│ 0       ┆ null ┆ null ┆ null       ┆ null       ┆ 0.0         │
│ 1       ┆ null ┆ null ┆ null       ┆ null       ┆ 0.0         │
│ 2       ┆ null ┆ null ┆ null       ┆ null       ┆ 0.0         │
│ 3       ┆ null ┆ null ┆ null       ┆ null       ┆ 0.0         │
│ 4       ┆ null ┆ null ┆ null       ┆ null       ┆ 0.0         │
└─────────┴──────┴──────┴────────────┴────────────┴─────────────┘

per_ticker = prices.sort(["Ticker", "Date"]).with_columns([
    # Daily returns (percentage change)
    ((pl.col("Close") / pl.col("Close").shift(1)) - 1)
        .over("Ticker")
        .alias("Return"),
    
    # Z-score normalization (standardize returns for cross-asset comparison)
    (pl.col("Return") - pl.col("Return").mean().over("Ticker"))
        / pl.col("Return").std().over("Ticker")
        .alias("ReturnZ"),
    
    # Lagged return (for momentum and autoregressive features)
    pl.col("Return").shift(1).over("Ticker").alias("ReturnLag1"),
])
# Use: Building features for forecasting models, identifying regime changes, 
#      comparing asset performance on normalized scale

def compute_returns(prices: pl.DataFrame) -> pl.DataFrame:
    """Compute returns for price data."""
    return prices.sort(["Ticker", "Date"]).with_columns([
        ((pl.col("Close") / pl.col("Close").shift(1)) - 1)
            .over("Ticker")
            .alias("Return"),
        (pl.col("Close").log() - pl.col("Close").log().shift(1))
            .over("Ticker")
            .alias("LogReturn")
    ])
 
def compute_rolling_features(
    prices: pl.DataFrame, 
    windows: list[int] = [5, 20, 60]
) -> pl.DataFrame:
    """Compute rolling statistics for multiple windows."""
    exprs = []
    for window in windows:
        exprs.extend([
            pl.col("Close").rolling_mean(window).over("Ticker").alias(f"MA{window}"),
            pl.col("Return").rolling_std(window).over("Ticker").alias(f"Vol{window}")
        ])
    return prices.with_columns(exprs)
 
# Usage
prices_with_returns = compute_returns(prices)
features = compute_rolling_features(prices_with_returns, windows=[5, 20, 60])

class FeatureEngineer:
    """Feature engineering pipeline for financial data."""
    
    def __init__(
        self, 
        windows: list[int] = [5, 20, 60],
        volatility_threshold: float = 0.02
    ):
        self.windows = windows
        self.volatility_threshold = volatility_threshold
        self.selected_features = []
    
    def compute_returns(self, df: pl.DataFrame) -> pl.DataFrame:
        """Compute return features."""
        return df.sort(["Ticker", "Date"]).with_columns([
            ((pl.col("Close") / pl.col("Close").shift(1)) - 1)
                .over("Ticker")
                .alias("Return"),
            (pl.col("Close").log() - pl.col("Close").log().shift(1))
                .over("Ticker")
                .alias("LogReturn")
        ])
    
    def compute_rolling(self, df: pl.DataFrame) -> pl.DataFrame:
        """Compute rolling statistics."""
        exprs = []
        for window in self.windows:
            exprs.extend([
                pl.col("Close").rolling_mean(window).over("Ticker").alias(f"MA{window}"),
                pl.col("Return").rolling_std(window).over("Ticker").alias(f"Vol{window}")
            ])
        return df.with_columns(exprs)
    
    def compute_regimes(self, df: pl.DataFrame) -> pl.DataFrame:
        """Compute regime indicators."""
        return df.with_columns([
            pl.when(pl.col("Return").abs() > self.volatility_threshold)
              .then(1.0)
              .otherwise(0.0)
              .alias("HighVolRegime")
        ])
    
    def transform(self, df: pl.DataFrame) -> pl.DataFrame:
        """Apply full feature engineering pipeline."""
        df = self.compute_returns(df)
        df = self.compute_rolling(df)
        df = self.compute_regimes(df)
        return df
 
# Usage
engineer = FeatureEngineer(windows=[5, 20, 60], volatility_threshold=0.02)
features = engineer.transform(prices)

class FeatureEngineer:
    # ... feature engineering methods ...
    
    def plot_feature_distributions(self, df: pl.DataFrame):
        """Visualize feature distributions (see Section 2)."""
        # Visualization code here
        pass

info = pl.read_csv("data/company_info.csv")
pt_snapshot = (
    prices.sort(["Ticker", "Date"])
          .group_by("Ticker")
          .tail(1)                      # latest row per ticker
          .select(["Ticker", "Date", "Close", "Return"])
          .join(info, on="Ticker", how="left")
)
# Use: Cross-sectional ranking (rank stocks by current metrics),
#      Portfolio initialization (select holdings based on latest data),
#      Model feature preparation (combine price and fundamental features)

actions = (
    pl.read_csv("data/corporate_actions.csv", try_parse_dates=True)
      .with_columns(pl.col("Date").cast(pl.Date))
)
 
dividends = (
    actions.filter(pl.col("ActionType") == "dividend")
           .select(["Date", "Ticker", "Value"])
           .rename({"Value": "Dividend"})
)
 
prices_with_div = (
    prices.join(dividends, on=["Ticker", "Date"], how="left")
          .with_columns(pl.col("Dividend").fill_null(0.0))
)
# Use: Computing total returns (price return + dividend yield),
#      Building dividend yield features for forecasting,
#      Adjusting prices for corporate actions

# Convert long to wide (one column per ticker)
panel_long = prices.select(["Date", "Ticker", "Close"])
panel_wide = panel_long.pivot(values="Close", index="Date", on="Ticker")
 
# Convert wide back to long (restore tidy format)
panel_long_again = panel_wide.melt(
    id_vars="Date",
    variable_name="Ticker",
    value_name="Close",
)
# Use: Feeding wide matrices into NumPy/PyTorch layers for cross-asset models,
#      Computing correlation matrices across assets,
#      Then reverting to long format for per-asset analysis

# Count missing values per column
missing = prices.select([
    pl.col("Return").is_null().sum().alias("Nulls"),
    pl.col("Return").is_not_null().sum().alias("NonNulls"),
])
 
# Check which tickers have the most gaps
missing_by_ticker = (
    prices.group_by("Ticker")
          .agg(pl.col("Return").is_null().sum().alias("MissingDays"))
          .sort("MissingDays", descending=True)
)

# Count missing values per column
missing = df.null_count()
missing_long = missing.unpivot(index=[], on=[], variable_name="column", value_name="null_count")
missing_filtered = missing_long.filter(pl.col("null_count") > 0).sort("null_count", descending=True)
print(missing_filtered.head(15))

shape: (15, 2)
┌────────┬────────────┐
│ column ┆ null_count │
│ ---    ┆ ---        │
│ str    ┆ u32        │
╞════════╪════════════╡
│ E7     ┆ 6969       │
│ V10    ┆ 6049       │
│ S3     ┆ 5733       │
│ M1     ┆ 5547       │
│ M13    ┆ 5540       │
│ …      ┆ …          │
│ M2     ┆ 3217       │
│ S8     ┆ 3009       │
│ M3     ┆ 2018       │
│ E1     ┆ 1784       │
│ P6     ┆ 1638       │
└────────┴────────────┘

# Drop rows with missing returns or volumes
clean_backtest = prices.drop_nulls(["Return", "Volume"])

# Fill missing returns with zero (assumes no price change)
forecast_fill = prices.with_columns(pl.col("Return").fill_null(0.0))

# Fill missing with zero
filled = df.select(["V1", "V2", "V3"]).with_columns([
    pl.col("V1").cast(pl.Float64, strict=False).fill_null(0.0).alias("V1_filled"),
    pl.col("V2").cast(pl.Float64, strict=False).fill_null(0.0).alias("V2_filled")
])
print(filled.head())

shape: (5, 5)
┌──────┬──────┬──────┬───────────┬───────────┐
│ V1   ┆ V2   ┆ V3   ┆ V1_filled ┆ V2_filled │
│ ---  ┆ ---  ┆ ---  ┆ ---       ┆ ---       │
│ str  ┆ str  ┆ str  ┆ f64       ┆ f64       │
╞══════╪══════╪══════╪═══════════╪═══════════╡
│ null ┆ null ┆ null ┆ 0.0       ┆ 0.0       │
│ null ┆ null ┆ null ┆ 0.0       ┆ 0.0       │
│ null ┆ null ┆ null ┆ 0.0       ┆ 0.0       │
│ null ┆ null ┆ null ┆ 0.0       ┆ 0.0       │
│ null ┆ null ┆ null ┆ 0.0       ┆ 0.0       │
└──────┴──────┴──────┴───────────┴───────────┘

# Forward-fill returns within each ticker (carry last return forward)
factor_ffill = (
    prices.sort(["Ticker", "Date"])
          .with_columns(
              pl.col("Return").fill_null(strategy="forward")
                                 .over("Ticker")
                                 .alias("Return_ffill")
          )
)

lf = prices.lazy()
 
result = (
    lf.filter(pl.col("Volume") > 0)                      # predicate pushdown
      .with_columns([
          (pl.col("Close") / pl.col("Close").shift(1) - 1)
              .over("Ticker")
              .alias("Return"),
          pl.col("Close").log().alias("LogClose"),
      ])
      .with_columns(
          pl.col("Return").rolling_std(20)
                             .over("Ticker")
                             .alias("Vol20")
      )
      .group_by_dynamic("Date", every="1mo", group_by="Ticker")
      .agg([
          pl.col("Return").mean().alias("MeanReturn"),
          pl.col("Vol20").mean().alias("MeanVol20"),
      ])
      .sort(["Ticker", "Date"])
      .collect()  # Materialize only at the end
)

# Each step materializes data, using more memory and time
prices_eager = pl.read_csv("data/prices_daily.csv")
prices_filtered = prices_eager.filter(pl.col("Volume") > 0)  # Materialized
prices_with_returns = prices_filtered.with_columns([...])     # Materialized again
monthly = prices_with_returns.group_by_dynamic(...)           # Materialized again

# Single optimized execution plan
result = (
    pl.scan_csv("data/prices_daily.csv")  # Lazy from the start
    .filter(pl.col("Volume") > 0)
    .with_columns([...])
    .group_by_dynamic(...)
    .collect()  # Materialize only once, at the end
)

# Compute returns first
returns = prices.lazy().group_by("Ticker").sort("Date").with_columns(
    pl.col("AdjClose").pct_change().alias("ret"),
    pl.col("AdjClose").log().diff().alias("log_ret"),
)
 
# Build rolling features per ticker
features = returns.with_columns(
    # 21-day rolling mean return (momentum signal)
    pl.col("log_ret").rolling_mean(window_size=21).alias("ret_1m"),
    
    # 21-day rolling volatility (risk measure)
    pl.col("log_ret").rolling_std(window_size=21).alias("vol_1m"),
    
    # First-order autocorrelation (mean-reversion signal)
    pl.corr("log_ret", pl.col("log_ret").shift(1), ddof=0)
      .over("Ticker")
      .alias("auto1"),
)

events = pl.read_csv("data/corporate_actions.csv", try_parse_dates=True)
 
joined = (
    prices.lazy()
    .join_asof(
        events.lazy(),
        left_on="Date",           # Trading dates
        right_on="ExDate",         # Event dates
        by="Ticker",               # Match within each asset
        strategy="backward",       # Use most recent event before/on this date
        tolerance=pl.duration(days=3),  # Only match events within 3 days
    )
    .with_columns(pl.col("Dividend").fill_null(0))
)

# Process large intraday dataset in streaming mode
large = pl.scan_parquet("s3://bucket/trades/*.parquet", streaming=True)
 
# Aggregate to 5-minute bars
binned = large.group_by_dynamic("ts", every="5m", by="Ticker").agg(
    pl.col("price").mean().alias("vwap"),      # Volume-weighted average price
    pl.col("size").sum().alias("volume"),      # Total volume
)

import polars as pl
import numpy as np
 
# Step 1: Load and inspect
prices = (
    pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
      .with_columns(pl.col("Date").cast(pl.Date))
      .sort(["Ticker", "Date"])
)
 
# Step 2: Compute returns and rolling features
features = prices.lazy().with_columns([
    # Returns
    ((pl.col("Close") / pl.col("Close").shift(1)) - 1)
        .over("Ticker")
        .alias("Return"),
    
    # Rolling statistics (21-day window)
    pl.col("Close").rolling_mean(21).over("Ticker").alias("MA21"),
    pl.col("Close").rolling_std(21).over("Ticker").alias("Vol21"),
    
    # Momentum (return over last 5 days)
    ((pl.col("Close") / pl.col("Close").shift(5)) - 1)
        .over("Ticker")
        .alias("Momentum5"),
]).with_columns([
    # Calendar features
    pl.col("Date").dt.weekday().alias("Weekday"),
    pl.col("Date").dt.month().alias("Month"),
]).drop_nulls(["Return", "MA21", "Vol21"]).collect()
 
# Step 3: Prepare for model (convert to wide format if needed, or keep long)
# For scikit-learn / XGBoost, you might want long format with Ticker as a feature
# For neural networks, you might reshape to wide format
 
# Export to NumPy
X = features.select(["MA21", "Vol21", "Momentum5", "Weekday", "Month"]).to_numpy()
y = features.select("Return").to_numpy()

# Step 1: Load prices and filter
prices = (
    pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
      .with_columns(pl.col("Date").cast(pl.Date))
      .filter(pl.col("Volume") > 1_000_000)  # Liquidity filter
      .sort(["Ticker", "Date"])
)
 
# Step 2: Compute returns and signals
signals = prices.lazy().with_columns([
    ((pl.col("Close") / pl.col("Close").shift(1)) - 1)
        .over("Ticker")
        .alias("Return"),
    
    # Simple momentum signal
    ((pl.col("Close") / pl.col("Close").shift(20)) - 1)
        .over("Ticker")
        .alias("Momentum20"),
    
    # Volatility signal
    pl.col("Close").rolling_std(20).over("Ticker").alias("Vol20"),
]).collect()
 
# Step 3: Join dividends for total returns
dividends = (
    pl.read_csv("data/corporate_actions.csv", try_parse_dates=True)
      .with_columns(pl.col("Date").cast(pl.Date))
      .filter(pl.col("ActionType") == "dividend")
      .select(["Date", "Ticker", "Value"])
      .rename({"Value": "Dividend"})
)
 
backtest_data = (
    signals.lazy()
    .join(dividends.lazy(), on=["Ticker", "Date"], how="left")
    .with_columns([
        pl.col("Dividend").fill_null(0.0),
        (pl.col("Return") + pl.col("Dividend") / pl.col("Close"))
            .alias("TotalReturn"),
    ])
    .drop_nulls(["Return", "Momentum20", "Vol20"])
    .collect()
)
 
# Step 4: Validate temporal ordering (no future data)
# This is critical - ensure signals only use past data
assert backtest_data.sort(["Ticker", "Date"]).is_sorted("Date", by="Ticker")

# Step 1: Load data
prices = (
    pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
      .with_columns(pl.col("Date").cast(pl.Date))
      .sort(["Ticker", "Date"])
)
 
fundamentals = (
    pl.read_csv("data/fundamentals.csv", try_parse_dates=True)
      .with_columns(pl.col("Date").cast(pl.Date))
)
 
# Step 2: Compute per-asset features
asset_features = prices.lazy().with_columns([
    ((pl.col("Close") / pl.col("Close").shift(1)) - 1)
        .over("Ticker")
        .alias("Return"),
    
    # 12-month momentum
    ((pl.col("Close") / pl.col("Close").shift(252)) - 1)
        .over("Ticker")
        .alias("Momentum12M"),
    
    # 20-day volatility
    pl.col("Close").rolling_std(20).over("Ticker").alias("Vol20"),
]).collect()
 
# Step 3: Join fundamentals and compute value factor (P/E, P/B, etc.)
# This is simplified - real value factors use multiple fundamental ratios
value_data = (
    asset_features.lazy()
    .join(fundamentals.lazy(), on=["Ticker", "Date"], how="left")
    .with_columns([
        (pl.col("Close") / pl.col("Earnings")).alias("PE_Ratio"),
    ])
    .collect()
)
 
# Step 4: Reshape to wide for cross-sectional ranking
# For each date, rank all assets by momentum and value
factors = (
    value_data.lazy()
    .select(["Date", "Ticker", "Momentum12M", "PE_Ratio", "Vol20"])
    .with_columns([
        # Cross-sectional z-scores (normalize across assets on each date)
        pl.col("Momentum12M").mean().over("Date").alias("MomMean"),
        pl.col("Momentum12M").std().over("Date").alias("MomStd"),
    ])
    .with_columns([
        ((pl.col("Momentum12M") - pl.col("MomMean")) / pl.col("MomStd"))
            .alias("MomentumFactor"),
        
        # Inverse PE (value factor - lower PE is better)
        (-pl.col("PE_Ratio")).rank().over("Date").alias("ValueFactorRank"),
    ])
    .select(["Date", "Ticker", "MomentumFactor", "ValueFactorRank", "Vol20"])
    .collect()
)

import polars as pl
from typing import Dict
 
# Define expected schema
EXPECTED_SCHEMA = {
    "Date": pl.Date,
    "Ticker": pl.Utf8,
    "Open": pl.Float64,
    "High": pl.Float64,
    "Low": pl.Float64,
    "Close": pl.Float64,
    "Volume": pl.Int64,
}
 
def validate_schema(df: pl.DataFrame, expected: Dict[str, pl.DataType]) -> bool:
    """Validate that DataFrame matches expected schema."""
    actual_schema = df.schema
    for col, expected_type in expected.items():
        if col not in actual_schema:
            raise ValueError(f"Missing column: {col}")
        if actual_schema[col] != expected_type:
            raise ValueError(
                f"Column {col} has type {actual_schema[col]}, expected {expected_type}"
            )
    return True
 
# Use after loading data
prices = pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
validate_schema(prices, EXPECTED_SCHEMA)

def validate_price_data(df: pl.DataFrame) -> pl.DataFrame:
    """Run finance-specific validation checks."""
    issues = []
    
    # Check 1: Prices should be positive
    negative_prices = df.filter(
        (pl.col("Close") <= 0) | 
        (pl.col("Open") <= 0) | 
        (pl.col("High") <= 0) | 
        (pl.col("Low") <= 0)
    )
    if len(negative_prices) > 0:
        issues.append(f"Found {len(negative_prices)} rows with non-positive prices")
    
    # Check 2: High >= Low, High >= Open, High >= Close, Low <= Open, Low <= Close
    invalid_ohlc = df.filter(
        (pl.col("High") < pl.col("Low")) |
        (pl.col("High") < pl.col("Open")) |
        (pl.col("High") < pl.col("Close")) |
        (pl.col("Low") > pl.col("Open")) |
        (pl.col("Low") > pl.col("Close"))
    )
    if len(invalid_ohlc) > 0:
        issues.append(f"Found {len(invalid_ohlc)} rows with invalid OHLC relationships")
    
    # Check 3: Volume should be non-negative
    negative_volume = df.filter(pl.col("Volume") < 0)
    if len(negative_volume) > 0:
        issues.append(f"Found {len(negative_volume)} rows with negative volume")
    
    # Check 4: Returns should be within reasonable bounds (e.g., -50% to +50% daily)
    returns = df.with_columns([
        ((pl.col("Close") / pl.col("Close").shift(1)) - 1).over("Ticker").alias("Return")
    ])
    extreme_returns = returns.filter(pl.col("Return").abs() > 0.5)
    if len(extreme_returns) > 0:
        issues.append(f"Found {len(extreme_returns)} rows with extreme returns (>50%)")
    
    # Check 5: Dates should be in reasonable range (e.g., 2000-2030)
    invalid_dates = df.filter(
        (pl.col("Date") < pl.date(2000, 1, 1)) | 
        (pl.col("Date") > pl.date(2030, 12, 31))
    )
    if len(invalid_dates) > 0:
        issues.append(f"Found {len(invalid_dates)} rows with dates outside expected range")
    
    if issues:
        raise ValueError("Data quality issues found:\n" + "\n".join(issues))
    
    return df
 
# Use in your pipeline
prices = pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
prices = validate_price_data(prices)

def reconcile_returns(prices: pl.DataFrame, benchmark_returns: pl.DataFrame) -> bool:
    """Reconcile computed returns against vendor-provided benchmark."""
    computed = (
        prices.sort(["Ticker", "Date"])
        .with_columns([
            ((pl.col("Close") / pl.col("Close").shift(1)) - 1)
                .over("Ticker")
                .alias("ComputedReturn")
        ])
        .select(["Ticker", "Date", "ComputedReturn"])
    )
    
    comparison = (
        computed.join(
            benchmark_returns.select(["Ticker", "Date", "Return"]),
            on=["Ticker", "Date"],
            how="inner"
        )
        .with_columns([
            (pl.col("ComputedReturn") - pl.col("Return")).abs().alias("Diff")
        ])
        .filter(pl.col("Diff") > 0.001)  # Tolerance for rounding
    )
    
    if len(comparison) > 0:
        print(f"Warning: {len(comparison)} returns don't match benchmark")
        print(comparison.head(10))
        return False
    
    return True

# In your CI/CD pipeline or data loading script
def load_and_validate_prices(filepath: str) -> pl.DataFrame:
    """Load prices with full validation."""
    df = pl.read_csv(filepath, try_parse_dates=True)
    validate_schema(df, EXPECTED_SCHEMA)
    df = validate_price_data(df)
    # Additional checks as needed
    return df

# See the optimized query plan
plan = (
    prices.lazy()
    .filter(pl.col("Volume") > 0)
    .with_columns([...])
    .group_by_dynamic("Date", every="1mo")
    .agg([...])
)
 
print(plan.explain())

# Good: Filter early
result = (
    prices.lazy()
    .filter(pl.col("Volume") > 1_000_000)  # Reduce data size early
    .filter(pl.col("Date") >= pl.date(2020, 1, 1))
    .with_columns([...])  # Expensive operations on smaller dataset
    .select(["Ticker", "Date", "Return"])  # Select only needed columns
    .collect()
)
 
# Bad: Filter late
result = (
    prices.lazy()
    .with_columns([...])  # Expensive operations on full dataset
    .filter(pl.col("Volume") > 1_000_000)  # Filter after expensive work
    .collect()
)

# Good: Lazy until the end
result = prices.lazy().with_columns([...]).group_by([...]).collect()
 
# Bad: Materializing intermediate results
intermediate = prices.with_columns([...])  # Materialized
result = intermediate.group_by([...])  # Works on materialized data

# Good: Stay in Polars
result = prices.lazy().with_columns([...]).collect()
 
# Bad: Unnecessary conversion
result = prices.to_pandas()  # Materializes entire dataset
result_polars = pl.from_pandas(result)  # Converts back

# Enable streaming for large files
large = pl.scan_parquet("s3://bucket/data/*.parquet", streaming=True)
result = large.group_by([...]).agg([...]).collect()

# Profile on sample
sample = prices.filter(pl.col("Date") >= pl.date(2023, 1, 1))  # Recent 20% of data
# Run your pipeline on sample and measure time
# Then scale up to full dataset

# Bad: Materializes intermediate result
intermediate = prices.with_columns([...])  # Materialized
result = intermediate.group_by([...])
 
# Good: Keep lazy
result = prices.lazy().with_columns([...]).group_by([...]).collect()

# Bad: apply is slow
result = df.with_columns([
    pl.col("Return").apply(lambda x: x * 2).alias("DoubleReturn")
])
 
# Good: Use built-in expressions
result = df.with_columns([
    (pl.col("Return") * 2).alias("DoubleReturn")
])

# Bad: Computes mean across all data, then filters
result = df.with_columns([
    (pl.col("Return") - pl.col("Return").mean()).alias("Deviation")
]).filter(pl.col("Ticker") == "AAPL")
 
# Good: Compute mean per group
result = df.with_columns([
    (pl.col("Return") - pl.col("Return").mean().over("Ticker")).alias("Deviation")
])

# Example: Efficient I/O with column pruning and filtering
result = (
    pl.scan_parquet("data/*.parquet")
    .select(["Ticker", "Date", "Close"])  # Only read needed columns
    .filter(pl.col("Date") >= pl.date(2020, 1, 1))  # Filter early
    .collect()
)

# Good: Operations that can be fused
result = df.with_columns([
    pl.col("Return") * 2,
    pl.col("Return") + 1,
    pl.col("Return").abs(),
])  # These can be fused into one pass
 
# Bad: Operations that can't be fused easily
result = df.with_columns([
    pl.col("Return").apply(lambda x: complex_function(x))
])  # Python UDF prevents fusion

import polars as pl
import matplotlib.pyplot as plt
import seaborn as sns
 
# Load and process data with Polars
prices = (
    pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
    .filter(pl.col("Ticker") == "AAPL")
    .sort("Date")
    .with_columns([
        ((pl.col("Close") / pl.col("Close").shift(1)) - 1).alias("Return")
    ])
)
 
# Convert to pandas for visualization (only materialize what you need)
prices_pd = prices.select(["Date", "Close", "Return"]).collect().to_pandas()
prices_pd.set_index("Date", inplace=True)
 
# Now use with matplotlib/seaborn
plt.figure(figsize=(12, 6))
plt.plot(prices_pd.index, prices_pd["Close"])
plt.title("AAPL Price Series")
plt.show()

# Load data with Polars
df = pl.read_csv("data/finance.csv")
 
# Convert to pandas (select only needed columns)
df_viz = df.select(["date_id", "forward_returns", "risk_free_rate", "market_forward_excess_returns"])
df_pd = df_viz.to_pandas()
print(df_pd.head())

   date_id  forward_returns  risk_free_rate  market_forward_excess_returns
0        0        -0.002421        0.000301                      -0.003038
1        1        -0.008495        0.000303                      -0.009114
2        2        -0.009624        0.000301                      -0.010243
3        3         0.004662        0.000299                       0.004046
4        4        -0.011686        0.000299                      -0.012301

# Extract arrays directly
dates = prices.select("Date").collect().to_series().to_numpy()
returns = prices.select("Return").collect().to_series().to_numpy()
 
# Plot with matplotlib
plt.figure(figsize=(12, 6))
plt.plot(dates, returns)
plt.title("AAPL Returns")
plt.show()

# Convert to NumPy arrays
returns_array = df_viz["forward_returns"].to_numpy()
print(f"Returns array shape: {returns_array.shape}")
print(f"First 10 values: {returns_array[:10]}")

Returns array shape: (9021,)
First 10 values: [-0.00242117 -0.00849468 -0.00962448  0.0046624  -0.01168577 -0.00644942
  0.00364423 -0.02456598 -0.00849183  0.01126298]

# Good: Materialize only selected columns
plot_data = (
    prices.lazy()
    .filter(pl.col("Ticker") == "AAPL")
    .select(["Date", "Close", "Return"])
    .collect()  # Materialize here
    .to_pandas()
)
 
# Bad: Materializing entire dataset
prices_full = prices.collect()  # Loads everything
plot_data = prices_full.select(["Date", "Close"]).to_pandas()  # Wastes memory

import polars as pl
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
 
# Load and compute returns
prices = pl.read_csv("data/prices_daily.csv", try_parse_dates=True)
returns = (
    prices.sort(["Ticker", "Date"])
    .with_columns([
        ((pl.col("Close") / pl.col("Close").shift(1)) - 1)
            .over("Ticker")
            .alias("Return")
    ])
    .filter(pl.col("Return").is_not_null())
    .select("Return")
    .collect()
    .to_series()
    .to_numpy()
)
 
# Histogram with normal overlay
fig, ax = plt.subplots(figsize=(10, 6))
ax.hist(returns, bins=50, density=True, alpha=0.7, label="Returns")
mu, sigma = np.mean(returns), np.std(returns)
x = np.linspace(returns.min(), returns.max(), 100)
ax.plot(x, np.exp(-0.5 * ((x - mu) / sigma) ** 2) / (sigma * np.sqrt(2 * np.pi)), 
        'r-', linewidth=2, label="Normal fit")
ax.set_xlabel("Daily Return")
ax.set_ylabel("Density")
ax.set_title("Returns Distribution vs Normal")
ax.legend()
ax.grid(True, alpha=0.3)
plt.show()

# Distribution analysis
returns_array = df_viz["forward_returns"].to_numpy()
 
# Statistics
print(f"Mean: {returns_array.mean():.6f}")
print(f"Std: {returns_array.std():.6f}")
print(f"Min: {returns_array.min():.6f}")
print(f"Max: {returns_array.max():.6f}")
print(f"25th percentile: {np.percentile(returns_array, 25):.6f}")
print(f"75th percentile: {np.percentile(returns_array, 75):.6f}")

Mean: 0.000471
Std: 0.010540
Min: -0.039754
Max: 0.040661
25th percentile: -0.004319
75th percentile: 0.005896

# Compare return distributions across assets
tickers = ["AAPL", "MSFT", "GOOGL"]
returns_by_ticker = {}
 
for ticker in tickers:
    ticker_returns = (
        prices.filter(pl.col("Ticker") == ticker)
        .with_columns([
            ((pl.col("Close") / pl.col("Close").shift(1)) - 1).alias("Return")
        ])
        .select("Return")
        .collect()
        .to_series()
        .to_numpy()
    )
    returns_by_ticker[ticker] = ticker_returns[ticker_returns != np.nan]
 
# Box plot comparison
fig, ax = plt.subplots(figsize=(10, 6))
ax.boxplot([returns_by_ticker[t] for t in tickers], labels=tickers)
ax.set_ylabel("Daily Return")
ax.set_title("Return Distributions by Asset")
ax.grid(True, alpha=0.3)
plt.show()

# Plot price series for multiple assets
fig, axes = plt.subplots(2, 1, figsize=(14, 10), sharex=True)
 
for ticker in ["AAPL", "MSFT"]:
    ticker_data = (
        prices.filter(pl.col("Ticker") == ticker)
        .sort("Date")
        .select(["Date", "Close"])
        .collect()
        .to_pandas()
        .set_index("Date")
    )
    
    axes[0].plot(ticker_data.index, ticker_data["Close"], label=ticker, linewidth=1.5)
 
axes[0].set_ylabel("Price ($)")
axes[0].set_title("Price Series")
axes[0].legend()
axes[0].grid(True, alpha=0.3)
 
# Plot returns below
for ticker in ["AAPL", "MSFT"]:
    ticker_data = (
        prices.filter(pl.col("Ticker") == ticker)
        .sort("Date")
        .with_columns([
            ((pl.col("Close") / pl.col("Close").shift(1)) - 1).alias("Return")
        ])
        .select(["Date", "Return"])
        .collect()
        .to_pandas()
        .set_index("Date")
    )
    
    axes[1].plot(ticker_data.index, ticker_data["Return"], label=ticker, alpha=0.7, linewidth=1)
 
axes[1].axhline(y=0, color='k', linestyle='--', alpha=0.3)
axes[1].set_ylabel("Daily Return")
axes[1].set_xlabel("Date")
axes[1].legend()
axes[1].grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

# Prepare time series data for visualization
df = pl.read_csv("data/finance.csv")
df_ts = df.select(['date_id', 'forward_returns']).head(200)
 
print('Time series data summary:')
print(f'Date range: {df_ts["date_id"].min()} to {df_ts["date_id"].max()}')
print(f'Returns range: {df_ts["forward_returns"].min():.6f} to {df_ts["forward_returns"].max():.6f}')
print(f'Mean return: {df_ts["forward_returns"].mean():.6f}')
print(f'Std return: {df_ts["forward_returns"].std():.6f}')
print(df_ts.head(10))

Time series data summary:
Date range: 0 to 199
Returns range: -0.030141 to 0.031990
Mean return: -0.000734
Std return: 0.010048
 
shape: (10, 2)
┌─────────┬─────────────────┐
│ date_id ┆ forward_returns │
│ ---     ┆ ---             │
│ i64     ┆ f64             │
╞═════════╪═════════════════╡
│ 0       ┆ -0.002421       │
│ 1       ┆ -0.008495       │
│ 2       ┆ -0.009624       │
│ 3       ┆ 0.004662        │
│ 4       ┆ -0.011686       │
│ 5       ┆ -0.006449       │
│ 6       ┆ 0.003644        │
│ 7       ┆ -0.024566       │
│ 8       ┆ -0.008492       │
│ 9       ┆ 0.011263        │
└─────────┴─────────────────┘

# Identify volatility regimes
aapl_data = (
    prices.filter(pl.col("Ticker") == "AAPL")
    .sort("Date")
    .with_columns([
        ((pl.col("Close") / pl.col("Close").shift(1)) - 1).alias("Return"),
        pl.col("Close").rolling_std(20).alias("Vol20")
    ])
    .collect()
    .to_pandas()
    .set_index("Date")
)
 
# Plot returns with volatility regime shading
fig, ax = plt.subplots(figsize=(14, 6))
ax.plot(aapl_data.index, aapl_data["Return"], alpha=0.6, label="Returns")
 
# Shade high volatility periods
vol_threshold = aapl_data["Vol20"].quantile(0.75)
high_vol = aapl_data["Vol20"] > vol_threshold
ax.fill_between(aapl_data.index, aapl_data["Return"].min(), aapl_data["Return"].max(),
                where=high_vol, alpha=0.2, color='red', label="High Volatility Regime")
 
ax.axhline(y=0, color='k', linestyle='--', alpha=0.3)
ax.set_ylabel("Daily Return")
ax.set_xlabel("Date")
ax.set_title("Returns with Volatility Regimes")
ax.legend()
ax.grid(True, alpha=0.3)
plt.show()

# Compute correlation matrix for features
features = (
    prices.filter(pl.col("Ticker") == "AAPL")
    .sort("Date")
    .with_columns([
        ((pl.col("Close") / pl.col("Close").shift(1)) - 1).alias("Return"),
        pl.col("Close").rolling_mean(20).alias("MA20"),
        pl.col("Close").rolling_std(20).alias("Vol20"),
        pl.col("Volume").log().alias("LogVolume"),
    ])
    .select(["Return", "MA20", "Vol20", "LogVolume"])
    .drop_nulls()
    .collect()
    .to_pandas()
)
 
corr_matrix = features.corr()
 
# Heatmap
fig, ax = plt.subplots(figsize=(10, 8))
sns.heatmap(corr_matrix, annot=True, fmt=".2f", cmap="coolwarm", center=0,
            square=True, linewidths=1, cbar_kws={"shrink": 0.8}, ax=ax)
ax.set_title("Feature Correlation Matrix")
plt.tight_layout()
plt.show()

# Compute correlation matrix
numeric_cols = ["forward_returns", "risk_free_rate", "market_forward_excess_returns"]
corr_data = df_viz.select(numeric_cols).to_pandas()
corr_matrix = corr_data.corr()
print(corr_matrix)

                               forward_returns  risk_free_rate  market_forward_excess_returns
forward_returns                       1.000000       -0.001019                       0.999943
risk_free_rate                       -0.001019        1.000000                      -0.009386
market_forward_excess_returns         0.999943       -0.009386                       1.000000

# Compute correlation matrix across assets
tickers = ["AAPL", "MSFT", "GOOGL", "AMZN"]
returns_matrix = []
 
for ticker in tickers:
    ticker_returns = (
        prices.filter(pl.col("Ticker") == ticker)
        .sort("Date")
        .with_columns([
            ((pl.col("Close") / pl.col("Close").shift(1)) - 1).alias("Return")
        ])
        .select(["Date", "Return"])
        .collect()
        .to_pandas()
        .set_index("Date")
    )
    returns_matrix.append(ticker_returns["Return"])
 
returns_df = pd.DataFrame(returns_matrix).T
returns_df.columns = tickers
corr_matrix = returns_df.corr()
 
# Heatmap
fig, ax = plt.subplots(figsize=(8, 6))
sns.heatmap(corr_matrix, annot=True, fmt=".2f", cmap="coolwarm", center=0,
            square=True, linewidths=1, cbar_kws={"shrink": 0.8}, ax=ax)
ax.set_title("Cross-Asset Return Correlations")
plt.tight_layout()
plt.show()

# Visualize missing data patterns
prices_sample = prices.select(["Date", "Ticker", "Close", "Volume"]).collect()
 
# Convert to wide format for visualization
prices_wide = (
    prices_sample
    .pivot(values="Close", index="Date", on="Ticker")
    .to_pandas()
    .set_index("Date")
)
 
# Plot missing data
fig, ax = plt.subplots(figsize=(14, 8))
ax.imshow(prices_wide.isnull().T, aspect='auto', cmap='viridis', interpolation='nearest')
ax.set_xlabel("Date")
ax.set_ylabel("Ticker")
ax.set_title("Missing Data Pattern (Yellow = Missing)")
plt.tight_layout()
plt.show()

# Count missing values per column
missing = df.null_count()
missing_long = missing.unpivot(index=[], on=[], variable_name="column", value_name="null_count")
missing_filtered = missing_long.filter(pl.col("null_count") > 0).sort("null_count", descending=True)
print(missing_filtered.head(20))

shape: (20, 2)
┌────────┬────────────┐
│ column ┆ null_count │
│ ---    ┆ ---        │
│ str    ┆ u32        │
╞════════╪════════════╡
│ E7     ┆ 6969       │
│ V10    ┆ 6049       │
│ S3     ┆ 5733       │
│ M1     ┆ 5547       │
│ M13    ┆ 5540       │
│ …      ┆ …          │
│ E20    ┆ 1616       │
│ P7     ┆ 1616       │
│ P5     ┆ 1574       │
│ V5     ┆ 1512       │
│ S5     ┆ 1511       │
└────────┴────────────┘

import xgboost as xgb
 
# Train model (example - see Chapter 4 for details)
# model = xgb.XGBRegressor().fit(X_train, y_train)
 
# Get feature importance
feature_importance = pd.DataFrame({
    'feature': feature_names,
    'importance': model.feature_importances_
}).sort_values('importance', ascending=False)
 
# Plot
fig, ax = plt.subplots(figsize=(10, 8))
ax.barh(feature_importance['feature'][:20], feature_importance['importance'][:20])
ax.set_xlabel("Importance")
ax.set_title("Top 20 Feature Importances")
ax.invert_yaxis()
plt.tight_layout()
plt.show()

# Simulated feature importance
feature_names = ["D1", "D2", "D3", "V1", "V2", "V3", "M1", "M2", "M3", "S1", "S2", "S3"]
importance_scores = np.random.exponential(scale=0.1, size=len(feature_names))
importance_scores = importance_scores / importance_scores.sum()  # Normalize
 
importance_df = pl.DataFrame({
    "feature": feature_names,
    "importance": importance_scores
}).sort("importance", descending=True)
print(importance_df)

shape: (12, 2)
┌─────────┬────────────┐
│ feature ┆ importance │
│ ---     ┆ ---        │
│ str     ┆ f64        │
╞═════════╪════════════╡
│ S3      ┆ 0.25399    │
│ D2      ┆ 0.218219   │
│ M2      ┆ 0.145804   │
│ D3      ┆ 0.095457   │
│ S1      ┆ 0.089259   │
│ …       ┆ …          │
│ D1      ┆ 0.03402    │
│ V2      ┆ 0.012297   │
│ V3      ┆ 0.012295   │
│ M1      ┆ 0.004338   │
│ S2      ┆ 0.001508   │
└─────────┴────────────┘

# Actual vs Predicted
fig, axes = plt.subplots(2, 1, figsize=(14, 10))
 
# Time series plot
axes[0].plot(y_test.index, y_test.values, label="Actual", alpha=0.7, linewidth=1.5)
axes[0].plot(y_test.index, y_pred, label="Predicted", alpha=0.7, linewidth=1.5)
axes[0].set_ylabel("Return")
axes[0].set_title("Actual vs Predicted Returns")
axes[0].legend()
axes[0].grid(True, alpha=0.3)
 
# Scatter plot
axes[1].scatter(y_test.values, y_pred, alpha=0.5)
axes[1].plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()], 
             'r--', linewidth=2, label="Perfect Prediction")
axes[1].set_xlabel("Actual Return")
axes[1].set_ylabel("Predicted Return")
axes[1].set_title("Prediction Accuracy")
axes[1].legend()
axes[1].grid(True, alpha=0.3)
 
plt.tight_layout()
plt.show()