File size: 3,530 Bytes
b54319d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | import pandas as pd
import numpy as np
from typing import List, Dict, Tuple, Optional
class PointInTimeUniverseFilter:
"""
Enforces Point-in-Time Data Hygiene and Universe Selection Rules:
1. Min Price Threshold: Price > $5.00
2. Dollar Volume Liquidity Threshold: 20-day Average Dollar Volume (ADV20) > $50M
3. Asset Age: Minimum 252 trading days history to avoid IPO / new ETF early distortions
4. Unified Trading Calendar Alignment & Forward Fill (Zero Backfill)
"""
def __init__(
self,
min_price: float = 5.0,
min_adv20_usd: float = 50_000_000.0,
min_age_days: int = 252,
):
self.min_price = min_price
self.min_adv20_usd = min_adv20_usd
self.min_age_days = min_age_days
def align_trading_calendar(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Align all assets to a unified trading calendar.
Missing trading days are forward-filled (FFill) up to 5 days, NEVER backfilled.
"""
df = df.copy()
df["date"] = pd.to_datetime(df["date"])
df = df.sort_values(["symbol", "date"]).reset_index(drop=True)
# Get full date range
all_dates = pd.date_range(df["date"].min(), df["date"].max(), freq="B")
symbols = df["symbol"].unique()
# Multi-index reindex to guarantee alignment
full_idx = pd.MultiIndex.from_product([all_dates, symbols], names=["date", "symbol"])
df_indexed = df.set_index(["date", "symbol"])
df_aligned = df_indexed.reindex(full_idx).groupby("symbol").ffill(limit=5).reset_index()
return df_aligned
def filter_universe(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, Dict[str, List[str]]]:
"""
Filter df for each trading date t to select eligible tickers.
Returns cleaned DataFrame and a dictionary mapping `date_str -> list of eligible symbols`.
"""
df = self.align_trading_calendar(df)
df["dollar_volume"] = df["close"] * df["volume"]
# Calculate rolling 20-day ADV in USD
df["adv20"] = (
df.groupby("symbol")["dollar_volume"]
.transform(lambda s: s.rolling(20, min_periods=10).mean())
)
# Calculate history age per ticker
df["trading_age"] = df.groupby("symbol")["date"].cumcount() + 1
# Eligibility condition at timestamp t
eligible_mask = (
(df["close"] >= self.min_price) &
(df["adv20"] >= self.min_adv20_usd) &
(df["trading_age"] >= self.min_age_days)
)
df["is_eligible"] = eligible_mask
# Build universe dict per date
universe_by_date = {}
for d, group in df[df["is_eligible"]].groupby("date"):
date_str = d.strftime("%Y-%m-%d")
universe_by_date[date_str] = group["symbol"].tolist()
return df, universe_by_date
@staticmethod
def validate_no_lookahead(df: pd.DataFrame, feature_date_col: str = "date", label_start_col: str = "label_start_date") -> bool:
"""
Validation assertion to prove zero lookahead leakage:
Every feature calculated at timestamp t must strictly satisfy t < label_start_date.
"""
if label_start_col not in df.columns:
return True
violating = df[df[feature_date_col] >= df[label_start_col]]
if len(violating) > 0:
raise ValueError(f"CRITICAL ERROR: Look-ahead bias detected in {len(violating)} rows!")
return True
|