copyleftdev Two Sigma Style Guide
Overview
Two Sigma is a systematic investment manager with ~$60B AUM, known for applying machine learning, distributed computing, and alternative data to financial markets. They operate like a technology company, investing heavily in data infrastructure, ML platforms, and research tooling.
Core Philosophy
"We're a technology company that happens to be in finance."
"Data is the new oil, but only if you can refine it."
"The best model is worthless without the infrastructure to deploy it."
Two Sigma believes that competitive advantage comes from data infrastructure and research velocity—the ability to test more ideas faster than competitors.
Design Principles
Data Platform First: Build the platform, then the models.
Feature Store: Features are first-class citizens, versioned and shared.
Reproducibility: Every experiment must be reproducible.
Scale Horizontally: Design for 1000x more data than you have today.
Research Velocity: Reduce time from idea to tested hypothesis.
When Building ML Trading Systems
Always
- Version everything: data, features, models, code
- Store point-in-time snapshots (no lookahead bias)
- Track feature lineage and dependencies
- Use distributed backtesting for scale
- Monitor model drift in production
- Separate feature engineering from model training
Never
- Use future data in features (even accidentally)
- Train and test on overlapping time periods
- Ignore transaction costs in backtests
- Deploy models without monitoring
- Hardcode features in model code
- Trust a single backtest run
Prefer
- Feature stores over ad-hoc feature computation
- Declarative pipelines over imperative scripts
- Ensemble methods over single models
- Online learning for adaptation
- A/B testing for model deployment
- Distributed compute over single-machine
Code Patterns
Feature Store Architecture
class FeatureStore:
"""
Two Sigma's insight: features are the real IP.
Centralize, version, and share them.
"""
def __init__(self, storage_backend, metadata_db):
self.storage = storage_backend # e.g., S3, HDFS
self.metadata = metadata_db # e.g., PostgreSQL
def register_feature(self,
name: str,
computation: Callable,
dependencies: List[str],
lookback_window: timedelta,
description: str):
"""Register a new feature definition."""
feature_def = FeatureDefinition(
name=name,
computation=computation,
dependencies=dependencies,
lookback_window=lookback_window,
description=description,
version=self.get_next_version(name),
created_at=datetime.utcnow()
)
self.metadata.save(feature_def)
return feature_def
def compute_feature(self,
feature_name: str,
as_of_date: date,
universe: List[str]) -> pd.DataFrame:
"""
Compute feature values as of a specific date.
CRITICAL: No future information leakage.
"""
feature_def = self.metadata.get_latest(feature_name)
# Get point-in-time data for dependencies
dependency_data = {}
for dep in feature_def.dependencies:
dependency_data[dep] = self.get_pit_data(
dep,
as_of_date,
lookback=feature_def.lookback_window
)
# Compute feature
result = feature_def.computation(dependency_data, universe, as_of_date)
# Cache result
self.storage.save(
feature_name=feature_name,
version=feature_def.version,
as_of_date=as_of_date,
values=result
)
return result
def get_training_data(self,
features: List[str],
target: str,
start_date: date,
end_date: date,
universe: List[str]) -> pd.DataFrame:
"""
Get feature matrix for training.
Each row is (date, symbol) with features computed as-of that date.
"""
rows = []
for current_date in date_range(start_date, end_date):
feature_values = {}
for feature_name in features:
feature_values[feature_name] = self.compute_feature(
feature_name, current_date, universe
)
# Target must be from the FUTURE (what we're predicting)
target_values = self.get_future_target(
target, current_date, universe
)
row = self.merge_features_and_target(
feature_values, target_values, current_date
)
rows.append(row)
return pd.concat(rows)
Distributed Backtesting
class DistributedBacktester:
"""
Two Sigma approach: parallelize backtesting across cluster.
Test hundreds of configurations simultaneously.
"""
def __init__(self, cluster: SparkCluster):
self.cluster = cluster
self.feature_store = FeatureStore()
def run_parameter_sweep(self,
strategy_class: Type[Strategy],
param_grid: Dict[str, List],
start_date: date,
end_date: date,
n_splits: int = 5) -> pd.DataFrame:
"""
Distributed parameter sweep with cross-validation.
"""
# Generate all parameter combinations
param_combinations = list(ParameterGrid(param_grid))
# Create time-series cross-validation splits
cv_splits = self.create_ts_splits(start_date, end_date, n_splits)
# Distribute work: (params, cv_fold) pairs
work_items = [
(params, train_dates, test_dates)
for params in param_combinations
for train_dates, test_dates in cv_splits
]
# Run in parallel on cluster
results = self.cluster.map(
self.run_single_backtest,
work_items,
strategy_class=strategy_class
)
return self.aggregate_results(results)
def run_single_backtest(self,
params: Dict,
train_dates: Tuple[date, date],
test_dates: Tuple[date, date],
strategy_class: Type[Strategy]) -> BacktestResult:
"""Single backtest run on a worker node."""
# Get training data
train_data = self.feature_store.get_training_data(
features=strategy_class.required_features(),
target='forward_returns',
start_date=train_dates[0],
end_date=train_dates[1]
)
# Train strategy
strategy = strategy_class(**params)
strategy.fit(train_data)
# Run backtest on test period
test_data = self.feature_store.get_training_data(
features=strategy_class.required_features(),
target='forward_returns',
start_date=test_dates[0],
end_date=test_dates[1]
)
returns = self.simulate_trading(strategy, test_data)
return BacktestResult(
params=params,
train_period=train_dates,
test_period=test_dates,
returns=returns,
sharpe=self.calculate_sharpe(returns),
max_drawdown=self.calculate_max_drawdown(returns)
)
Alternative Data Pipeline
class AlternativeDataPipeline:
"""
Two Sigma's edge: alternative data processed at scale.
Satellite imagery, credit card data, web scraping, etc.
"""
def __init__(self, raw_storage, processed_storage, feature_store):
self.raw = raw_storage
self.processed = processed_storage
self.feature_store = feature_store
def ingest_satellite_imagery(self,
source: str,
date: date) -> Dict[str, Any]:
"""
Example: count cars in retail parking lots.
"""
# Download raw imagery
raw_images = self.fetch_images(source, date)
# Store raw with metadata
raw_path = self.raw.save(
data=raw_images,
source=source,
date=date,
ingested_at=datetime.utcnow()
)
# Process: detect and count vehicles
processed = {}
for location_id, image in raw_images.items():
vehicle_count = self.detect_vehicles(image)
processed[location_id] = {
'vehicle_count': vehicle_count,
'image_quality': self.assess_quality(image),
'weather_conditions': self.detect_weather(image)
}
# Store processed
processed_path = self.processed.save(
data=processed,
source=source,
date=date,
processing_version='v2.3'
)
return processed
def build_retail_traffic_feature(self):
"""
Convert satellite data into trading feature.
"""
def compute(deps, universe, as_of_date):
# Get satellite data up to as_of_date
satellite = deps['satellite_parking_counts']
# Map locations to companies
location_mapping = deps['location_to_ticker']
# Aggregate by company
company_traffic = {}
for ticker in universe:
locations = location_mapping.get(ticker, [])
counts = [satellite.get(loc, {}).get('vehicle_count', np.nan)
for loc in locations]
# Year-over-year change (careful about seasonality)
current = np.nanmean(counts)
year_ago = self.get_year_ago_counts(ticker, as_of_date)
company_traffic[ticker] = {
'parking_lot_traffic': current,
'traffic_yoy_change': (current - year_ago) / year_ago if year_ago else np.nan
}
return pd.DataFrame(company_traffic).T
self.feature_store.register_feature(
name='retail_parking_traffic',
computation=compute,
dependencies=['satellite_parking_counts', 'location_to_ticker'],
lookback_window=timedelta(days=7),
description='YoY change in parking lot traffic from satellite imagery'
)
Model Monitoring and Drift Detection
class ModelMonitor:
"""
Two Sigma approach: continuous monitoring of production models.
Detect drift before it becomes a problem.
"""
def __init__(self, model_registry, metrics_store):
self.registry = model_registry
self.metrics = metrics_store
self.drift_thresholds = {}
def monitor_model(self,
model_id: str,
predictions: pd.Series,
actuals: pd.Series,
features: pd.DataFrame):
"""Real-time monitoring of model performance."""
# Performance metrics
ic = stats.spearmanr(predictions, actuals).correlation
hit_rate = (np.sign(predictions) == np.sign(actuals)).mean()
# Feature drift detection
feature_drift = self.detect_feature_drift(model_id, features)
# Prediction distribution drift
pred_drift = self.detect_prediction_drift(model_id, predictions)
# Store metrics
self.metrics.log(
model_id=model_id,
timestamp=datetime.utcnow(),
ic=ic,
hit_rate=hit_rate,
feature_drift=feature_drift,
prediction_drift=pred_drift
)
# Alert on significant drift
if feature_drift > self.drift_thresholds.get(model_id, 0.1):
self.alert(f"Feature drift detected for {model_id}: {feature_drift}")
if ic < 0:
self.alert(f"Negative IC for {model_id}: {ic}")
def detect_feature_drift(self,
model_id: str,
current_features: pd.DataFrame) -> float:
"""
Compare current feature distribution to training distribution.
Uses Population Stability Index (PSI).
"""
training_stats = self.registry.get_training_stats(model_id)
psi_scores = []
for col in current_features.columns:
expected = training_stats[col]
actual = current_features[col]
psi = self.calculate_psi(expected, actual)
psi_scores.append(psi)
return np.mean(psi_scores)
def calculate_psi(self, expected: pd.Series, actual: pd.Series) -> float:
"""Population Stability Index for drift detection."""
# Bin the distributions
bins = np.percentile(expected, np.linspace(0, 100, 11))
expected_pct = np.histogram(expected, bins=bins)[0] / len(expected)
actual_pct = np.histogram(actual, bins=bins)[0] / len(actual)
# Avoid division by zero
expected_pct = np.clip(expected_pct, 0.001, 1)
actual_pct = np.clip(actual_pct, 0.001, 1)
psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
return psi
Mental Model
Two Sigma approaches ML trading by asking:
- What data do we have? And what data could we get?
- What features can we extract? Systematic feature engineering
- How do we avoid lookahead bias? Point-in-time everything
- How do we scale? Distributed compute for backtesting
- How do we monitor? Continuous drift detection
Signature Two Sigma Moves
- Feature stores as central infrastructure
- Point-in-time data management
- Distributed backtesting at scale
- Alternative data pipelines
- Continuous model monitoring
- Experiment tracking and reproducibility
- Declarative feature definitions
- Horizontal scaling for research