The Problem That Breaks Every Backtest
Consider this scenario: You pull 20 years of AAPL historical prices, run a simple moving average crossover strategy, and report a Sharpe ratio of 1.42. Confident numbers. Then you deploy it live—and watch it bleed for three months before you find the root cause.
Apple executed a 7:1 stock split on June 9, 2014, and a 4:1 split on August 28, 2020. Your raw price data showed AAPL at $645 in June 2014 and $130 in September 2020. Without proper adjustment, your backtest treated a 5x price appreciation as alpha when it was purely a mechanical split effect. Every price-based indicator—moving averages, Bollinger Bands, RSI—produced garbage. Your strategy was never valid.
This is not an edge case. It is the default state of unadjusted financial data.
Raw price data suffers from two fundamental distortions: stock splits that mechanically divide share prices, and dividends that mechanically reduce cash available to shareholders. Neither represents actual investment performance. CRSP—the Center for Research in Security Prices—defined the industry standard for correcting these distortions in 1964, and it remains the backbone of virtually every academic and institutional backtesting framework.
This article dissects the complete adjustment processing pipeline: how to construct adjustment factor tables, how to apply forward and backward adjustment methods, and how to implement a production-grade pipeline that handles millions of price records without silent failures.
Understanding the Two Distortions
Stock Splits
A stock split multiplies the share count and divides the price by a split factor. A 2-for-1 split doubles every shareholder's shares and halves the price. The total market capitalization remains unchanged. For a backtest, the pre-split prices must be divided by the cumulative split factor so that all prices are expressed on a comparable basis.
| Event Date | Split Ratio | Pre-Split Price | Post-Split Price | Adjustment Factor |
|---|---|---|---|---|
| 2000-06-21 | 2:1 | $120.00 | $60.00 | 0.5 |
| 2005-02-28 | 2:1 | $90.00 | $45.00 | 0.25 (cumulative) |
| 2014-06-09 | 7:1 | $645.00 | $92.14 | 0.036 (cumulative) |
The adjustment factor accumulates multiplicatively. After three 2-for-1 splits, a pre-split price of $240 needs to be divided by 8 (2 × 2 × 2) to be comparable to post-split prices.
Dividends
A dividend distributes cash from the company to shareholders. On the ex-dividend date, the stock price typically drops by the dividend amount—shareholders are wealthier by the dividend value, but the share price mechanically decreases.
For total return calculations, this price drop must be compensated. CRSP defines total return as:
Total Return = (Price Change + Dividend) / Previous Price
If a stock closes at $100, pays a $1 dividend, and opens at $99 on the ex-dividend date, the total return is ($99 + $1 - $100) / $100 = 0%. The investment performance is zero—cash went into the investor's pocket, share price went down by the same amount.
However, for price return calculations (which exclude dividends), the -1% price return is correct. The distinction matters enormously: momentum strategies typically use price returns; buy-and-hold comparisons typically use total returns.
The Adjustment Factor Table Structure
The adjustment factor table is the foundation of the pipeline. Each row represents a corporate action event with the following schema:
from dataclasses import dataclass
from datetime import date
from decimal import Decimal, ROUND_HALF_UP
from typing import Optional
from enum import Enum
class ActionType(Enum):
SPLIT = "SPLIT"
DIVIDEND = "DIVIDEND"
SPINOFF = "SPINOFF"
STOCK_DIVIDEND = "STOCK_DIVIDEND" # e.g., 5% stock dividend
@dataclass
class AdjustmentFactor:
"""
Represents a single corporate action that requires price adjustment.
For splits: factor = 1 / cumulative_split_ratio
For dividends: factor represents the dividend yield to be added back
"""
symbol: str
event_date: date
action_type: ActionType
gross_amount: Decimal # Split ratio (e.g., 2.0 for 2:1) or dividend per share
adjustment_factor: Decimal # Cumulative multiplier for price adjustment
record_date: Optional[date] # For dividend processing
ex_date: Optional[date] # Ex-dividend date (when adjustment applies)
def __post_init__(self):
# Ensure Decimal precision for financial calculations
self.gross_amount = Decimal(str(self.gross_amount))
self.adjustment_factor = Decimal(str(self.adjustment_factor))
CRSP Adjustment Methodology
The Two Adjustment Standards
CRSP defines two adjustment methods:
Backward Adjustment (Split-adjusted prices): Historical prices are adjusted downward so that pre-event prices appear as if the split or dividend already occurred. This is the standard approach for technical analysis and price-based strategies.
Forward Adjustment (Total return prices): Historical prices are adjusted upward to reflect reinvestment of dividends. The price series reflects what the investment would be worth if dividends were reinvested. This is the standard for performance benchmarking.
The relationship between the two is:
Forward-Adjusted Price[t] = Backward-Adjusted Price[t] × Cumulative Dividend Factor[t]
Where the cumulative dividend factor chains together dividend reinvestment from the base date to time t.
CRSP Factor Calculation
CRSP calculates the adjustment factor for each event as follows:
For Splits:
adjustment_factor[new] = adjustment_factor[old] / split_ratio
For Dividends (Forward Adjustment):
price_ex_div = price_before_div - dividend_per_share
adjustment_factor[new] = adjustment_factor[old] × (price_ex_div / price_before_div)
However, the standard CRSP implementation uses a ratio-based approach:
factor = (price_before_div - dividend) / price_before_div
adjustment_factor[new] = adjustment_factor[old] / factor
This ensures that when historical prices are multiplied by the cumulative adjustment factor, dividends are "added back" to reconstruct total returns.
Pipeline Architecture
High-Level Design
┌─────────────────────────────────────────────────────────────────┐
│ ADJUSTMENT PROCESSING PIPELINE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ RAW PRICE │───▶│ CORPORATE │───▶│ ADJUSTMENT │ │
│ │ TABLE │ │ ACTION │ │ CALCULATOR │ │
│ │ │ │ TABLE │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ VALIDATED │◀───│ OUTPUT │◀───│ SPLIT │ │
│ │ OUTPUT │ │ GENERATOR │ │ + DIVIDEND │ │
│ │ TABLE │ │ │ │ APPLIER │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
Component Specifications
1. Corporate Action Ingestion
Corporate actions come from multiple sources: exchange feeds, data vendors (Bloomberg, Refinitiv), or regulatory filings (SEC EDGAR for US equities). The ingestion layer must handle:
- Duplicate detection (same event from multiple sources)
- Timing differences (record date vs. ex-date vs. payment date)
- Currency denomination (for ADRs and foreign listings)
import logging
from datetime import datetime, date
from decimal import Decimal
from typing import List, Dict, Optional, Iterator
from dataclasses import dataclass, field
import heapq
logger = logging.getLogger(__name__)
@dataclass
class CorporateActionSource:
"""Configuration for a corporate action data source."""
source_id: str
source_type: str # 'exchange', 'vendor', 'edgar'
base_url: Optional[str] = None
api_key: Optional[str] = None
poll_interval_seconds: int = 3600
class CorporateActionIngestionService:
"""
Ingests corporate actions from multiple sources with deduplication
and temporal alignment.
"""
def __init__(self):
self.sources: Dict[str, CorporateActionSource] = {}
self._action_cache: Dict[str, List[AdjustmentFactor]] = {}
def register_source(self, source: CorporateActionSource) -> None:
"""Register a new corporate action source."""
self.sources[source.source_id] = source
logger.info(f"Registered corporate action source: {source.source_id}")
def fetch_actions(
self,
symbols: List[str],
start_date: date,
end_date: date
) -> Iterator[AdjustmentFactor]:
"""
Fetch corporate actions for given symbols and date range.
Yields actions sorted by event_date to enable streaming processing.
"""
# Merge from all sources using a k-way merge
iterators = []
for source_id, source in self.sources.items():
try:
it = self._fetch_from_source(source, symbols, start_date, end_date)
iterators.append(it)
except Exception as e:
logger.error(f"Failed to fetch from {source_id}: {e}")
continue
# K-way merge by event_date
for action in heapq.merge(*iterators):
if self._is_duplicate(action):
logger.debug(f"Skipping duplicate action: {action}")
continue
yield action
def _is_duplicate(self, action: AdjustmentFactor) -> bool:
"""Check if action is a duplicate within the cache window."""
cache_key = f"{action.symbol}:{action.event_date}:{action.action_type.value}"
if cache_key in self._action_cache:
# Verify if this is a better quality record
return True
return False
2. Adjustment Factor Calculator
This component calculates cumulative adjustment factors from raw corporate actions.
from functools import reduce
class AdjustmentFactorCalculator:
"""
Calculates cumulative adjustment factors according to CRSP methodology.
CRSP adjustment factors are designed so that:
- Historical prices × cumulative_factor = split-adjusted prices (backward)
- Historical prices × cumulative_factor = total return prices (forward)
"""
SPLIT_ADJUSTMENT = Decimal("1.0") # Base adjustment
def calculate_split_factor(
self,
split_ratio: Decimal
) -> Decimal:
"""
Calculate adjustment factor for a split event.
A 2:1 split means each share becomes 2 shares.
To make pre-split prices comparable: divide by 2.
Args:
split_ratio: e.g., Decimal("2.0") for 2-for-1
Returns:
Adjustment factor to multiply historical prices by
"""
if split_ratio <= 0:
raise ValueError(f"Invalid split ratio: {split_ratio}")
return Decimal("1.0") / split_ratio
def calculate_dividend_factor(
self,
price_before_exdiv: Decimal,
dividend_per_share: Decimal,
adjustment_type: str = "forward"
) -> Decimal:
"""
Calculate adjustment factor for a dividend event.
For forward adjustment (total return):
We want: price_before × new_factor = price_exdiv + dividend
Therefore: new_factor = (price_exdiv + dividend) / price_before
= (price_before - dividend + dividend) / price_before
= 1.0
Wait, that's not quite right. Let me recalculate.
CRSP total return factor:
factor = (P[t-1] - D) / P[t-1]
To get total return price from raw price:
TR_price[t] = price[t] / factor[t]
= price[t] × P[t-1] / (P[t-1] - D)
For backward adjustment (price return, no dividends):
No factor needed; dividends simply reduce price.
"""
if price_before_exdiv <= 0:
raise ValueError(f"Invalid price: {price_before_exdiv}")
price_exdiv = price_before_exdiv - dividend_per_share
if price_exdiv <= 0:
logger.warning(
f"Dividend {dividend_per_share} exceeds price {price_before_exdiv}, "
"capping at price."
)
price_exdiv = Decimal("0.01")
# For forward adjustment (total return), the factor accounts for
# dividend reinvestment. CRSP standard:
factor = (price_exdiv / price_before_exdiv).quantize(
Decimal("0.000001"), rounding=ROUND_HALF_UP
)
return factor
def calculate_cumulative_factor(
self,
actions: List[AdjustmentFactor],
base_date: date,
target_date: date
) -> Decimal:
"""
Calculate cumulative adjustment factor from base_date to target_date.
This chains together all split and dividend factors between the dates.
Args:
actions: List of AdjustmentFactor records, sorted by event_date
base_date: The reference date (adjustment factor = 1.0)
target_date: The target date for the cumulative factor
Returns:
Cumulative adjustment factor
"""
applicable_actions = [
a for a in actions
if base_date < a.event_date <= target_date
]
cumulative = Decimal("1.0")
for action in applicable_actions:
cumulative *= action.adjustment_factor
return cumulative
3. Price Adjustment Processor
The core component that applies adjustment factors to price data.
from dataclasses import dataclass
from typing import Generator, Tuple
import pandas as pd
@dataclass
class AdjustedPriceRecord:
"""A single adjusted price record."""
symbol: str
trade_date: date
open: Decimal
high: Decimal
low: Decimal
close: Decimal
volume: int
adjustment_factor: Decimal
adjusted_close: Decimal
cumulative_dividend_factor: Optional[Decimal] = None # For total return
raw_close: Decimal = None # Original unadjusted close
def __post_init__(self):
if self.raw_close is None:
self.raw_close = self.close
class PriceAdjustmentProcessor:
"""
Applies adjustment factors to raw price data.
Supports two modes:
- 'backward': Split-adjusted prices (for technical analysis)
- 'forward': Total return prices (for performance benchmarking)
"""
def __init__(self, adjustment_type: str = "backward"):
if adjustment_type not in ("backward", "forward"):
raise ValueError(f"Invalid adjustment type: {adjustment_type}")
self.adjustment_type = adjustment_type
def process_symbol(
self,
symbol: str,
raw_prices: pd.DataFrame,
actions: List[AdjustmentFactor],
reference_date: date
) -> pd.DataFrame:
"""
Process all prices for a single symbol.
Args:
symbol: Ticker symbol
raw_prices: DataFrame with columns [trade_date, open, high, low, close, volume]
actions: Sorted list of adjustment factors
reference_date: Date at which adjustment_factor = 1.0
Returns:
DataFrame with adjusted prices and factors
"""
if raw_prices.empty:
return raw_prices
# Sort by date
df = raw_prices.sort_values("trade_date").copy()
# Build action lookup for O(1) access
action_map = self._build_action_map(actions)
# Initialize cumulative factor at reference date
# All prices BEFORE reference_date need adjustment
# All prices FROM reference_date onward are the reference
cumulative_factor = Decimal("1.0")
last_action_date = reference_date
adjusted_records = []
for idx, row in df.iterrows():
trade_date = row["trade_date"]
# Update cumulative factor if there are actions between last date and this date
actions_in_period = [
action_map[d]
for d in action_map
if last_action_date < d <= trade_date
]
for action in actions_in_period:
if self.adjustment_type == "backward":
cumulative_factor *= action.adjustment_factor
else:
# Forward adjustment: divide by factor to "add back" dividend
cumulative_factor /= action.adjustment_factor
# Apply adjustment
raw_close = Decimal(str(row["close"]))
adjusted_close = raw_close * cumulative_factor
adjusted_records.append({
"symbol": symbol,
"trade_date": trade_date,
"open": Decimal(str(row["open"])) * cumulative_factor,
"high": Decimal(str(row["high"])) * cumulative_factor,
"low": Decimal(str(row["low"])) * cumulative_factor,
"close": adjusted_close,
"volume": row["volume"],
"adjustment_factor": cumulative_factor,
"raw_close": raw_close,
"raw_volume": row["volume"] * (1 / float(cumulative_factor) if cumulative_factor > 0 else 0)
})
last_action_date = trade_date
result_df = pd.DataFrame(adjusted_records)
# Round to appropriate precision
decimal_places = 6
for col in ["open", "high", "low", "close"]:
result_df[col] = result_df[col].apply(
lambda x: float(Decimal(str(x)).quantize(
Decimal("0.000001"), rounding=ROUND_HALF_UP
))
)
return result_df
def _build_action_map(
self,
actions: List[AdjustmentFactor]
) -> Dict[date, AdjustmentFactor]:
"""Build a date-indexed lookup for actions."""
return {action.event_date: action for action in sorted(actions, key=lambda x: x.event_date)}
Handling Edge Cases
Chained Splits
When multiple splits occur in rapid succession, the cumulative factor compounds. Example: Tesla's 5:1 split on August 31, 2020, following a previous split, requires accurate chaining.
def test_chained_splits():
"""
Tesla example: 5:1 split on 2020-08-31.
Pre-split prices from 2020-08-28 would have been ~$2,213.
After 5:1 split, comparable price is ~$442.60.
"""
actions = [
AdjustmentFactor(
symbol="TSLA",
event_date=date(2020, 8, 31),
action_type=ActionType.SPLIT,
gross_amount=Decimal("5.0"),
adjustment_factor=Decimal("0.2"), # 1/5
ex_date=date(2020, 8, 31)
)
]
calculator = AdjustmentFactorCalculator()
# Price on 2020-08-28 (pre-split)
pre_split_price = Decimal("2213.40")
# Calculate what the split-adjusted price should be
cumulative = calculator.calculate_cumulative_factor(
actions,
base_date=date(2020, 8, 31), # Reference date is post-split
target_date=date(2020, 8, 28) # We want adjustment for pre-split date
)
adjusted_price = pre_split_price * cumulative
assert abs(adjusted_price - Decimal("442.68")) < Decimal("0.01"), \
f"Expected ~442.68, got {adjusted_price}"
Dividend Reinvestment Effects
True total return calculations require reinvesting dividends. For simple price adjustment, dividends are "added back" to prices. For precise CRSP total return:
def calculate_crsp_total_return(
prices: pd.DataFrame,
dividends: pd.DataFrame,
initial_investment: float = 10000.0
) -> pd.Series:
"""
Calculate CRSP-style total return index.
This reconstructs what $10,000 invested at the start would be worth
if dividends were reinvested.
"""
df = prices.merge(dividends, on=["symbol", "trade_date"], how="left")
df["dividend"] = df["dividend"].fillna(Decimal("0"))
# Daily return = (P[t] - P[t-1] + D[t]) / P[t-1]
df["daily_return"] = (
(df["close"] - df["close"].shift(1) + df["dividend"]) /
df["close"].shift(1)
)
# Cumulative return
df["cumulative_return"] = (1 + df["daily_return"]).cumprod()
df["portfolio_value"] = initial_investment * df["cumulative_return"]
return df["portfolio_value"]
Missing Price Data Around Ex-Dates
When a stock goes ex-dividend but has no trading data (holidays, delistings), the factor must still apply.
def test_missing_trading_days():
"""
If ex-date falls on a holiday with no trading,
the adjustment factor still applies on the next trading day.
"""
actions = [
AdjustmentFactor(
symbol="AAPL",
event_date=date(2024, 12, 25), # Christmas - no trading
action_type=ActionType.DIVIDEND,
gross_amount=Decimal("0.25"),
adjustment_factor=Decimal("0.9975"),
ex_date=date(2024, 12, 20) # Ex-date is Friday
)
]
# The adjustment factor must apply on the next trading day (Dec 26)
processor = PriceAdjustmentProcessor(adjustment_type="backward")
raw_prices = pd.DataFrame([
{"trade_date": date(2024, 12, 20), "open": 245.00, "high": 247.00, "low": 244.00, "close": 246.50, "volume": 50000000},
{"trade_date": date(2024, 12, 26), "open": 245.75, "high": 247.25, "low": 245.00, "close": 246.25, "volume": 45000000}, # Friday's close was 246.50, Monday opens lower due to dividend
])
result = processor.process_symbol("AAPL", raw_prices, actions, reference_date=date(2024, 12, 20))
# The adjustment factor should affect the December 26 prices
assert result.loc[1, "close"] < raw_prices.loc[1, "close"], \
"Post-ex-div prices should be adjusted downward"
Production Pipeline Implementation
Batch Processing with Checkpointing
For processing millions of records, the pipeline must be resumable.
import hashlib
import json
from pathlib import Path
from datetime import datetime
import threading
from typing import Optional
class AdjustmentPipeline:
"""
Production-grade adjustment pipeline with checkpointing and error recovery.
Key features:
- Checkpoint-based recovery for long-running jobs
- Parallel processing by symbol
- Comprehensive error handling and logging
- Idempotent operations
"""
def __init__(
self,
output_dir: Path,
checkpoint_dir: Path,
max_workers: int = 4,
batch_size: int = 1000
):
self.output_dir = Path(output_dir)
self.checkpoint_dir = Path(checkpoint_dir)
self.max_workers = max_workers
self.batch_size = batch_size
self.output_dir.mkdir(parents=True, exist_ok=True)
self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
self._checkpoint_lock = threading.Lock()
def run(
self,
symbols: List[str],
start_date: date,
end_date: date,
adjustment_type: str = "backward"
) -> Dict[str, Path]:
"""
Run the full adjustment pipeline for given symbols.
Returns:
Dict mapping symbol to output file path
"""
logger.info(f"Starting adjustment pipeline for {len(symbols)} symbols")
# Load checkpoint state
checkpoint = self._load_checkpoint()
# Filter already-processed symbols
symbols_to_process = [
s for s in symbols
if checkpoint.get(s, {}).get("status") != "completed"
]
logger.info(f"Processing {len(symbols_to_process)} symbols "
f"(skipping {len(symbols) - len(symbols_to_process)} completed)")
# Process in parallel batches
from concurrent.futures import ThreadPoolExecutor, as_completed
output_files = {}
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {
executor.submit(
self._process_single_symbol,
symbol,
start_date,
end_date,
adjustment_type,
checkpoint
): symbol
for symbol in symbols_to_process
}
for future in as_completed(futures):
symbol = futures[future]
try:
result = future.result()
if result:
output_files[symbol] = result
self._update_checkpoint(symbol, "completed", result)
logger.info(f"Completed: {symbol}")
except Exception as e:
logger.error(f"Failed processing {symbol}: {e}")
self._update_checkpoint(symbol, "failed", str(e))
return output_files
def _process_single_symbol(
self,
symbol: str,
start_date: date,
end_date: date,
adjustment_type: str,
checkpoint: Dict
) -> Optional[Path]:
"""
Process a single symbol with checkpoint recovery.
"""
# Check for partial progress
progress = checkpoint.get(symbol, {})
last_processed_date = progress.get("last_processed_date")
# Fetch raw prices (would connect to TickDB or other data source)
raw_prices = self._fetch_prices(symbol, start_date, end_date)
# Fetch corporate actions
actions = self._fetch_actions(symbol, start_date, end_date)
# Process
processor = PriceAdjustmentProcessor(adjustment_type)
result = processor.process_symbol(
symbol,
raw_prices,
actions,
reference_date=end_date # Most recent date is the reference
)
# Write output
output_path = self.output_dir / f"{symbol}_adjusted.parquet"
result.to_parquet(output_path, index=False)
return output_path
def _load_checkpoint(self) -> Dict:
"""Load checkpoint state from disk."""
checkpoint_file = self.checkpoint_dir / "pipeline_checkpoint.json"
if checkpoint_file.exists():
with open(checkpoint_file) as f:
return json.load(f)
return {}
def _update_checkpoint(
self,
symbol: str,
status: str,
result: any
) -> None:
"""Update checkpoint with thread-safe write."""
with self._checkpoint_lock:
checkpoint = self._load_checkpoint()
checkpoint[symbol] = {
"status": status,
"result": str(result) if result else None,
"timestamp": datetime.utcnow().isoformat()
}
checkpoint_file = self.checkpoint_dir / "pipeline_checkpoint.json"
with open(checkpoint_file, "w") as f:
json.dump(checkpoint, f, indent=2, default=str)
def _fetch_prices(
self,
symbol: str,
start_date: date,
end_date: date
) -> pd.DataFrame:
"""
Fetch raw price data for a symbol.
In production, this would connect to TickDB:
import os
import requests
headers = {"X-API-Key": os.environ.get("TICKDB_API_KEY")}
response = requests.get(
"https://api.tickdb.ai/v1/market/kline",
headers=headers,
params={
"symbol": f"{symbol}.US",
"interval": "1d",
"start_time": int(start_date.timestamp()),
"end_time": int(end_date.timestamp()),
"adjust": "none" # Get raw, unadjusted prices
},
timeout=(3.05, 10)
)
"""
# Placeholder implementation
raise NotImplementedError("Connect to data source")
def _fetch_actions(
self,
symbol: str,
start_date: date,
end_date: date
) -> List[AdjustmentFactor]:
"""
Fetch corporate actions for a symbol.
In production, this would query a corporate actions database
or use a vendor API.
"""
# Placeholder implementation
raise NotImplementedError("Connect to corporate actions source")
Validation and Quality Assurance
The Adjustment Validation Suite
Every adjustment pipeline must pass these validation checks before production deployment.
import unittest
from decimal import Decimal
class TestAdjustmentPipeline(unittest.TestCase):
"""
Comprehensive validation suite for adjustment calculations.
These tests verify correctness against known CRSP benchmarks.
"""
def test_split_roundtrip(self):
"""
Verify that adjusted prices survive a roundtrip.
If we apply adjustment factor f to price P:
P_adjusted = P × f
Then dividing P_adjusted by f should recover P.
"""
raw_price = Decimal("645.00")
split_ratio = Decimal("7.0")
factor = Decimal("1.0") / split_ratio
adjusted_price = raw_price * factor
recovered_price = adjusted_price / factor
self.assertAlmostEqual(
float(recovered_price),
float(raw_price),
places=6
)
def test_cumulative_split_factor(self):
"""
Verify cumulative split factor across multiple splits.
Three consecutive 2:1 splits should produce a factor of 1/8.
"""
calculator = AdjustmentFactorCalculator()
actions = [
AdjustmentFactor(
symbol="TEST", event_date=date(2020, 1, 1),
action_type=ActionType.SPLIT, gross_amount=Decimal("2.0"),
adjustment_factor=Decimal("0.5"), ex_date=date(2020, 1, 1)
),
AdjustmentFactor(
symbol="TEST", event_date=date(2020, 6, 1),
action_type=ActionType.SPLIT, gross_amount=Decimal("2.0"),
adjustment_factor=Decimal("0.25"), ex_date=date(2020, 6, 1)
),
AdjustmentFactor(
symbol="TEST", event_date=date(2021, 1, 1),
action_type=ActionType.SPLIT, gross_amount=Decimal("2.0"),
adjustment_factor=Decimal("0.125"), ex_date=date(2021, 1, 1)
),
]
cumulative = calculator.calculate_cumulative_factor(
actions,
base_date=date(2021, 1, 1),
target_date=date(2020, 1, 1)
)
# After three 2:1 splits, pre-Jan 2020 prices should be divided by 8
self.assertAlmostEqual(float(cumulative), 0.125, places=6)
def test_dividend_price_continuity(self):
"""
Verify that ex-dividend price adjustment maintains continuity.
The price drop on ex-date should equal the dividend amount
(before adjustment is applied).
"""
processor = PriceAdjustmentProcessor(adjustment_type="backward")
raw_prices = pd.DataFrame([
{"trade_date": date(2024, 3, 14), "open": 175.00, "high": 176.00,
"low": 174.50, "close": 175.50, "volume": 1000000}, # Pre-ex-div
{"trade_date": date(2024, 3, 15), "open": 174.75, "high": 175.50,
"low": 174.25, "close": 174.80, "volume": 1200000}, # Ex-div date
])
dividend = Decimal("0.24")
actions = [
AdjustmentFactor(
symbol="AAPL", event_date=date(2024, 3, 15),
action_type=ActionType.DIVIDEND, gross_amount=dividend,
adjustment_factor=Decimal("1.0") - (dividend / Decimal("175.50")),
ex_date=date(2024, 3, 15)
)
]
result = processor.process_symbol(
"AAPL", raw_prices, actions, reference_date=date(2024, 3, 15)
)
# Adjusted prices should show no discontinuity
pre_adj_close = result.loc[0, "close"]
post_adj_close = result.loc[1, "close"]
# The percentage change should be minimal (only market movement, not dividend effect)
pct_change = abs(post_adj_close - pre_adj_close) / pre_adj_close
self.assertLess(pct_change, 0.02,
f"Price discontinuity detected: {pct_change:.2%}")
def test_total_return_reconstruction(self):
"""
Verify that total return adjustment reconstructs the portfolio value.
Starting with $100, if a stock pays a 1% dividend and the price
stays flat, the total return should be 1%.
"""
prices = pd.DataFrame([
{"trade_date": date(2024, 1, 1), "open": 100.0, "high": 100.5,
"low": 99.5, "close": 100.0, "volume": 1000},
{"trade_date": date(2024, 1, 2), "open": 99.0, "high": 99.5,
"low": 98.5, "close": 99.0, "volume": 1000}, # Price drops by dividend
])
dividends = pd.DataFrame([
{"symbol": "TEST", "trade_date": date(2024, 1, 2), "dividend": Decimal("1.0")}
])
portfolio_values = calculate_crsp_total_return(prices, dividends)
# Initial: $100. After 1% dividend with flat price: $101
self.assertAlmostEqual(portfolio_values.iloc[-1], 101.0, places=2)
if __name__ == "__main__":
unittest.main()
Common Pitfalls and How to Avoid Them
| Pitfall | Symptom | Prevention |
|---|---|---|
| Using unadjusted prices | Backtest shows unrealistic Sharpe, strategy fails in live trading | Always verify adjustment status before backtesting |
| Wrong reference date | Adjustment factors are inverted | Verify that the most recent prices have factor = 1.0 |
| Missing ex-dividend dates | Small but persistent return attribution errors | Cross-reference multiple corporate action sources |
| Ignoring currency adjustments | ADRs show phantom returns from FX movements | Separate currency conversion from price adjustment |
| Integer volume overflow | Volume statistics are wrong for heavily split stocks | Use int64 for volume; adjust for splits |
| Floating-point rounding errors | Adjustment factors drift over long periods | Use Decimal type throughout; round only at output |
Deployment Recommendations by Scale
| Scale | Approach | Tools |
|---|---|---|
| Individual researcher | Single-threaded Python script with pandas | pandas, duckdb |
| Small team (1–5) | Batch processing with checkpointing | Python pipeline, PostgreSQL |
| Institutional | Distributed processing with monitoring | Spark/Dask, Kubernetes, Delta Lake |
| Real-time | Stream processing with incremental updates | Kafka, Flink, Redis |
For most quant researchers and small teams, the single-symbol processor with the validation suite provides sufficient correctness guarantees. The parallel pipeline becomes necessary when processing thousands of symbols across decades of daily data—this can generate millions of records that need to be processed within a reasonable time window.
Conclusion
Adjustment processing is not optional preprocessing. It is the foundation on which every price-based analysis rests. A backtest run on unadjusted data is not approximately wrong—it is structurally invalid. The price distortions from splits compound over time, and dividends create systematic negative bias in reported returns that grows with the holding period.
The pipeline architecture presented here follows CRSP methodology precisely: construct adjustment factors from corporate actions, chain them multiplicatively through time, and apply them to raw prices to produce either split-adjusted (backward) or total return (forward) series. The production-grade implementation includes checkpointing for resumable long-running jobs, parallel processing for throughput, and a comprehensive validation suite that verifies correctness against mathematical invariants.
The critical operational principle: validate before you trust. Every time you pull data, verify that your adjustment factors produce price continuity across known corporate action dates. When the validation suite passes, you can trust the backtest. When it fails, you will catch the error before it costs you real capital.
Next Steps
If you are building a backtesting framework, integrate the adjustment factor calculator into your data loading pipeline and verify it against CRSP-formatted benchmark data before running any strategy analysis.
If you need historical US equity data with proper adjustment, the TickDB /kline endpoint supports adjust parameters for both backward and forward adjustment modes. Configure the parameter to match your strategy's return calculation requirements.
If you are debugging an existing backtest, use the validation suite to test your adjustment factors against known split dates and dividend ex-dates. Common failure modes include using the wrong reference date, applying splits but not dividends, or processing actions in the wrong chronological order.
This article focuses on data engineering methodology. All examples use generic Python implementations; production deployments should be adapted to specific data sources and infrastructure requirements.