Alternative Data for Trading: Satellite, Social, and Web Data
Alternative data encompasses any dataset that provides investment insight beyond traditional financial statements, price data, and analyst reports. The alternative data industry has exploded from $200 million in 2016 to over $7 billion in 2025, driven by the recognition that non-traditional datasets can provide information advantages that persist for years before being arbitraged away. This guide covers the most actionable categories of alternative data, how to process them into trading signals, and the practical challenges of working with messy, unstructured, and incomplete datasets.Key Takeaways
- Alternative data provides an information edge by capturing real-world activity (consumer spending, foot traffic, shipping) before it appears in financial statements.
- The signal is typically weak but persistent. Alternative data rarely provides 10% alpha; it provides 0.5-2% that compounds reliably.
- Data quality is the primary challenge. Coverage gaps, reporting lags, survivorship bias, and vendor changes require extensive validation.
- Point-in-time accuracy is critical. Any alternative dataset must be time-stamped to prevent look-ahead bias.
Web Scraping for Financial Data
Web data provides real-time insight into product pricing, inventory levels, job postings, and corporate activity. ``python
import requests
from bs4 import BeautifulSoup
import pandas as pd
import numpy as np
from datetime import datetime
import time
import hashlib
import json
class WebDataCollector:
"""
Collect alternative data from web sources.
Includes rate limiting, caching, and change detection.
"""
def __init__(
self,
cache_dir: str = "data/web_cache",
rate_limit_seconds: float = 2.0,
):
self.cache_dir = cache_dir
self.rate_limit = rate_limit_seconds
self.last_request_time = 0
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Research Bot)"
})
def _rate_limited_get(self, url: str) -> requests.Response:
"""Rate-limited HTTP GET."""
elapsed = time.time() - self.last_request_time
if elapsed < self.rate_limit:
time.sleep(self.rate_limit - elapsed)
self.last_request_time = time.time()
return self.session.get(url, timeout=30)
def scrape_job_postings(
self, company: str, source: str = "careers_page"
) -> dict:
"""
Track job postings as a leading indicator.
Rising postings = growth; falling = contraction.
In production, use a proper job posting API
(Revelio Labs, LinkUp, Thinknum).
"""
# Simulated scraping result structure
return {
"company": company,
"timestamp": datetime.now().isoformat(),
"total_postings": 0, # Would come from actual scraping
"engineering_postings": 0,
"sales_postings": 0,
"categories": {},
"change_30d": 0.0,
}
def track_pricing_changes(
self, product_urls: list[str]
) -> pd.DataFrame:
"""
Monitor product pricing for revenue insights.
Price increases + stable demand = margin expansion.
"""
results = []
for url in product_urls:
url_hash = hashlib.md5(url.encode()).hexdigest()[:8]
try:
response = self._rate_limited_get(url)
soup = BeautifulSoup(response.text, "html.parser")
# Generic price extraction
price_elements = soup.find_all(
attrs={"class": lambda c: c and "price" in c.lower()}
)
prices = []
for elem in price_elements:
text = elem.get_text(strip=True)
# Extract numeric price
import re
numbers = re.findall(r"\$?([\d,]+\.?\d*)", text)
for n in numbers:
try:
prices.append(float(n.replace(",", "")))
except ValueError:
pass
results.append({
"url_hash": url_hash,
"timestamp": datetime.now().isoformat(),
"prices_found": len(prices),
"min_price": min(prices) if prices else None,
"max_price": max(prices) if prices else None,
"avg_price": np.mean(prices) if prices else None,
})
except Exception as e:
results.append({
"url_hash": url_hash,
"error": str(e),
})
return pd.DataFrame(results)
`
Social Media Sentiment
Social media provides real-time sentiment signals with particular value for consumer-facing companies and meme stocks.
`python
class SocialSentimentProcessor:
"""
Process social media data into trading signals.
Sources: Reddit (r/wallstreetbets, r/stocks), StockTwits, Twitter/X.
In production, use data vendors (SocialSentiment.io,
Alternative.me) for reliable feeds.
"""
def __init__(
self,
baseline_window: int = 30,
alert_threshold: float = 3.0,
):
self.baseline_window = baseline_window
self.alert_threshold = alert_threshold
def process_mentions(
self, mentions: pd.DataFrame
) -> pd.DataFrame:
"""
Process raw mention data into signals.
Input columns: ticker, timestamp, text, source, sentiment_score
Output: ticker-level signals with volume and sentiment anomalies
"""
# Aggregate by ticker and day
mentions["date"] = pd.to_datetime(mentions["timestamp"]).dt.date
daily = mentions.groupby(["ticker", "date"]).agg(
mention_count=("text", "count"),
avg_sentiment=("sentiment_score", "mean"),
median_sentiment=("sentiment_score", "median"),
sentiment_std=("sentiment_score", "std"),
positive_pct=("sentiment_score", lambda x: (x > 0.1).mean()),
negative_pct=("sentiment_score", lambda x: (x < -0.1).mean()),
).reset_index()
signals = []
for ticker in daily["ticker"].unique():
ticker_data = daily[daily["ticker"] == ticker].sort_values("date")
if len(ticker_data) < self.baseline_window:
continue
# Volume anomaly (z-score of mention count)
rolling_mean = ticker_data["mention_count"].rolling(
self.baseline_window
).mean()
rolling_std = ticker_data["mention_count"].rolling(
self.baseline_window
).std()
volume_z = (ticker_data["mention_count"] - rolling_mean) / rolling_std.replace(0, 1)
# Sentiment shift
sentiment_ma = ticker_data["avg_sentiment"].rolling(
self.baseline_window
).mean()
sentiment_shift = ticker_data["avg_sentiment"] - sentiment_ma
# Composite signal
for i in range(self.baseline_window, len(ticker_data)):
row = ticker_data.iloc[i]
vz = volume_z.iloc[i]
ss = sentiment_shift.iloc[i]
signals.append({
"ticker": ticker,
"date": row["date"],
"mention_volume_z": vz,
"sentiment_shift": ss,
"is_volume_spike": abs(vz) > self.alert_threshold,
"signal": ss (1 + min(abs(vz), 3) 0.3),
"confidence": min(1.0, row["mention_count"] / 50),
})
return pd.DataFrame(signals)
`
Satellite and Geolocation Data
Satellite imagery and geolocation data provide ground-truth observations of economic activity.
`python
class SatelliteDataProcessor:
"""
Process satellite and geolocation data for trading signals.
Data types:
- Parking lot counts (retail foot traffic)
- Oil storage tank levels (commodity supply)
- Shipping/port activity (trade flow)
- Construction activity (real estate/capex)
- Nighttime light intensity (economic activity)
"""
def __init__(self):
pass
def process_parking_lot_counts(
self, counts: pd.DataFrame
) -> pd.DataFrame:
"""
Convert parking lot car counts into retail sales signals.
Input: location_id, date, car_count, store_chain
Output: chain-level same-store-sales proxy
"""
# Aggregate to chain level
daily_chain = counts.groupby(["store_chain", "date"]).agg(
total_cars=("car_count", "sum"),
location_count=("location_id", "nunique"),
avg_cars_per_location=("car_count", "mean"),
).reset_index()
signals = []
for chain in daily_chain["store_chain"].unique():
chain_data = daily_chain[
daily_chain["store_chain"] == chain
].sort_values("date")
if len(chain_data) < 30:
continue
# Year-over-year comparison (same-store-sales proxy)
chain_data = chain_data.set_index("date")
chain_data.index = pd.to_datetime(chain_data.index)
# 7-day rolling average to smooth weekly patterns
smoothed = chain_data["avg_cars_per_location"].rolling(7).mean()
# YoY change
yoy = smoothed.pct_change(periods=365)
# Trend (30-day momentum)
trend = smoothed.pct_change(periods=30)
for date in chain_data.index[365:]:
signals.append({
"store_chain": chain,
"date": date,
"avg_cars": smoothed.loc[date],
"yoy_change": yoy.loc[date] if date in yoy.index else None,
"trend_30d": trend.loc[date] if date in trend.index else None,
"signal": (
0.6 * (yoy.loc[date] if date in yoy.index and not pd.isna(yoy.loc[date]) else 0)
+ 0.4 * (trend.loc[date] if date in trend.index and not pd.isna(trend.loc[date]) else 0)
),
})
return pd.DataFrame(signals)
def process_oil_storage(
self, tank_levels: pd.DataFrame
) -> pd.DataFrame:
"""
Process oil tank fill levels from satellite imagery.
Tank shadows in satellite images reveal fill levels.
Rising inventories = bearish for crude prices.
"""
# Aggregate to regional level
regional = tank_levels.groupby(["region", "date"]).agg(
total_capacity_pct=("fill_level_pct", "mean"),
n_tanks=("tank_id", "nunique"),
).reset_index()
regional["date"] = pd.to_datetime(regional["date"])
regional = regional.sort_values(["region", "date"])
# Change in storage levels
for region in regional["region"].unique():
mask = regional["region"] == region
regional.loc[mask, "storage_change_7d"] = (
regional.loc[mask, "total_capacity_pct"].diff(7)
)
regional.loc[mask, "storage_change_30d"] = (
regional.loc[mask, "total_capacity_pct"].diff(30)
)
# Signal: falling storage = bullish for oil
regional["signal"] = -regional["storage_change_7d"].fillna(0)
return regional
`
Building an Alternative Data Pipeline
`python
class AltDataPipeline:
"""
End-to-end pipeline for ingesting, processing,
and scoring alternative data signals.
"""
def __init__(self):
self.signals = {}
def add_signal(
self,
name: str,
data: pd.DataFrame,
ticker_col: str,
date_col: str,
signal_col: str,
weight: float = 1.0,
decay_days: int = 5,
):
"""Register an alternative data signal."""
self.signals[name] = {
"data": data,
"ticker_col": ticker_col,
"date_col": date_col,
"signal_col": signal_col,
"weight": weight,
"decay_days": decay_days,
}
def compute_composite(
self,
tickers: list[str],
as_of_date: pd.Timestamp,
) -> pd.DataFrame:
"""
Compute composite alternative data score for each ticker.
"""
results = []
for ticker in tickers:
component_scores = {}
for name, config in self.signals.items():
df = config["data"]
ticker_mask = df[config["ticker_col"]] == ticker
date_mask = pd.to_datetime(df[config["date_col"]]) <= as_of_date
recency_mask = (
as_of_date - pd.to_datetime(df[config["date_col"]])
).dt.days <= config["decay_days"]
relevant = df[ticker_mask & date_mask & recency_mask]
if len(relevant) > 0:
# Most recent signal value
latest = relevant.sort_values(config["date_col"]).iloc[-1]
score = latest[config["signal_col"]]
# Apply time decay
days_old = (
as_of_date - pd.to_datetime(latest[config["date_col"]])
).days
decay = np.exp(-0.5 * days_old / config["decay_days"])
component_scores[name] = score decay config["weight"]
else:
component_scores[name] = 0.0
# Weighted composite
total_weight = sum(
config["weight"] for config in self.signals.values()
)
composite = sum(component_scores.values()) / total_weight
results.append({
"ticker": ticker,
"date": as_of_date,
"composite_score": composite,
**{f"signal_{k}": v for k, v in component_scores.items()},
})
return pd.DataFrame(results)
``