Corporate KPI Prediction Market Correlation with Share Price

2026-09-15

import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from scipy import stats

from prediction_markets.database.snowflake_writer import connector

def winsorize(df: pd.DataFrame, limits: float = 0.01) -> pd.DataFrame:
    """Clip each column to its own [limits, 1-limits] quantile range so a few
    extreme days don't dominate the correlation. NaNs are ignored by quantile
    and left untouched by clip."""
    lower = df.quantile(limits)
    upper = df.quantile(1 - limits)
    return df.clip(lower=lower, upper=upper, axis=1)

estimates_query = """
select
    price_time::DATE as date,
    price AS p,
    std_resolved_ticker as ticker,
    DATE(market_close)::VARCHAR as MARKET_CLOSE
from
    PREDICTION_MARKET_FACTORS.PUBLIC.STANDARDIZED_MARKETS
where
/*    std_resolved_ticker in ('MMM', 'T', 'ANF')
    and  */
     
       std_event_type = 'corporate KPI'
    and std_property_value = 'estimate'

    
order by date, price_time
"""

cursor = connector.cursor()
cursor.execute(estimates_query)
estimates = cursor.fetch_pandas_all()

# last quote per ticker+market_close per calendar day, carried forward
wide = (
    estimates.assign(DATE=pd.to_datetime(estimates["DATE"]))
    .pivot_table(
        index="DATE", columns=["TICKER", "MARKET_CLOSE"], values="P", aggfunc="last"
    )
    .sort_index()
)

estimate_prediction_change = winsorize(wide.diff().replace(0, np.nan))

fig = go.Figure()
for ticker, market_close in estimate_prediction_change.columns:
    fig.add_scatter(
        x=estimate_prediction_change.index,
        y=estimate_prediction_change[(ticker, market_close)],
        mode="lines",
        name=f"{ticker} ({market_close})",
    )

fig.update_layout(
    title="Daily change in market-implied estimate, by ticker and market close",
    template="plotly_white",
    height=460,
    hovermode="x unified",
    legend_title="ticker (market close)",
)
fig.update_yaxes(title="Δ implied estimate (day-over-day, %)", tickformat=".0%")
fig.update_xaxes(showgrid=False)
fig.show()

Unable to display output for mime type(s): application/vnd.plotly.v1+json

tickers = sorted(estimates["TICKER"].dropna().unique())
start_date = pd.to_datetime(estimates["DATE"]).min().date().isoformat()

placeholders = ", ".join(["%s"] * len(tickers))
PRICES_STMT = f"""
SELECT
    date AS price_date,
    ticker,
    value AS closing_price
FROM SNOWFLAKE_PUBLIC_DATA_PAID.PUBLIC_DATA.STOCK_PRICE_TIMESERIES
WHERE ticker IN ({placeholders})
  AND variable_name = 'Post-Market Close'
  AND date >= %s
ORDER BY ticker, date
"""

cursor = connector.cursor()
cursor.execute(PRICES_STMT, (*tickers, start_date))
prices = cursor.fetch_pandas_all()

price_wide = (
    prices.assign(PRICE_DATE=pd.to_datetime(prices["PRICE_DATE"]))
    .pivot_table(
        index="PRICE_DATE", columns="TICKER", values="CLOSING_PRICE", aggfunc="last"
    )
    .sort_index()
)
share_price_returns = winsorize(np.log(price_wide).diff())

rows = []
for ticker, market_close in estimate_prediction_change.columns:
    if ticker not in share_price_returns.columns:
        continue
    df = pd.concat(
        {
            "est": estimate_prediction_change[(ticker, market_close)],
            "ret": share_price_returns[ticker],
        },
        axis=1,
    ).dropna()
    df = df[df["est"] != 0]  # only days the estimate changed
    if len(df) < 3:
        continue
    rows.append(
        {
            "ticker": ticker,
            "market_close": market_close,
            "n_days": len(df),
            "pearson": df["est"].corr(df["ret"]),
            "spearman": df["est"].corr(df["ret"], method="spearman"),
        }
    )

rows_df = pd.DataFrame(rows)

corr = rows_df.set_index(["ticker", "market_close"]).sort_values(
    "pearson", ascending=False
)

pooled = pd.concat(
    pd.concat(
        {
            "est": estimate_prediction_change[(ticker, market_close)],
            "ret": share_price_returns[ticker],
        },
        axis=1,
    )
    .dropna()
    .query("est != 0")
    for ticker, market_close in estimate_prediction_change.columns
    if ticker in share_price_returns.columns
)
print(
    f"pooled  pearson={pooled['est'].corr(pooled['ret']): .3f}"
    f"  spearman={pooled['est'].corr(pooled['ret'], method='spearman'): .3f}"
    f"  n={len(pooled)}"
)

pooled  pearson= 0.014  spearman= 0.016  n=6756

labeled_pooled = pd.concat(
    [
        estimate_prediction_change[(ticker, market_close)]
        .rename("est")
        .to_frame()
        .assign(
            ret=share_price_returns[ticker], ticker=ticker, market_close=market_close
        )
        .dropna(subset=["est", "ret"])
        .query("est != 0")
        for ticker, market_close in estimate_prediction_change.columns
        if ticker in share_price_returns.columns
    ]
)

fig = go.Figure()
for ticker, group in labeled_pooled.groupby("ticker"):
    fig.add_scatter(
        x=group["est"],
        y=group["ret"],
        mode="markers",
        name=ticker,
        marker=dict(size=8),
    )

slope, intercept = np.polyfit(labeled_pooled["est"], labeled_pooled["ret"], 1)
x_fit = np.linspace(labeled_pooled["est"].min(), labeled_pooled["est"].max(), 50)
fig.add_scatter(
    x=x_fit,
    y=slope * x_fit + intercept,
    mode="lines",
    name="pooled OLS fit",
    line=dict(color="black", dash="dash"),
)

r = labeled_pooled["est"].corr(labeled_pooled["ret"])
fig.update_layout(
    title=f"Estimate change vs. share price return (pooled pearson r = {r:.2f}, n = {len(labeled_pooled)})",
    template="plotly_white",
    height=460,
    xaxis_title="Δ estimate (day-over-day)",
    yaxis_title="log return",
    legend_title="ticker",
)
fig.update_xaxes(tickformat=".0%")
fig.update_yaxes(tickformat=".1%")
fig.show()

Unable to display output for mime type(s): application/vnd.plotly.v1+json

def cross_correlation_pooled(
    window_pairs: list[tuple[pd.Series, pd.Series]], max_lag: int = 5
) -> pd.Series:
    """corr(est_t, ret_{t+lag}) for lag in [-max_lag, max_lag], pooling observations
    across every (est, ret) window passed in. Each window is shifted independently
    (so a shift never crosses a window boundary) and the aligned pairs are then
    pooled before computing the correlation for that lag.
    Positive lag = estimate change leads the price return by that many trading days;
    negative lag = price return leads the estimate change."""
    out = {}
    for lag in range(-max_lag, max_lag + 1):
        pooled = pd.concat(
            pd.concat({"est": est, "ret": ret.shift(-lag)}, axis=1).dropna()
            for est, ret in window_pairs
        )
        out[lag] = pooled["est"].corr(pooled["ret"]) if len(pooled) >= 3 else np.nan
    return pd.Series(out)


max_lag = 5
common_index = wide.index.intersection(price_wide.index)
window_pairs = [
    (
        wide[(ticker, market_close)].reindex(common_index).diff(),
        share_price_returns[ticker].reindex(common_index),
    )
    for ticker, market_close in estimate_prediction_change.columns
    if ticker in share_price_returns.columns
]
ccf = cross_correlation_pooled(window_pairs, max_lag=max_lag)

colors = ["#C0623B" if lag == 0 else "#3B7AB8" for lag in ccf.index]
fig = go.Figure()
fig.add_bar(x=list(ccf.index), y=ccf.values, marker_color=colors)
fig.add_hline(y=0, line=dict(color="black", width=1))

fig.update_layout(
    title="Cross-correlation: estimate change vs. share price return, by lag (all tickers and market closes pooled)",
    template="plotly_white",
    height=420,
)
fig.update_yaxes(title="correlation")
fig.update_xaxes(title="lag, trading days (+ = estimate leads price)")
fig.show()

Unable to display output for mime type(s): application/vnd.plotly.v1+json

def sign_agreement_pooled(
    window_pairs: list[tuple[pd.Series, pd.Series]], max_lag: int = 5
) -> pd.DataFrame:
    """Fraction of days where sign(est_t) == sign(ret_{t+lag}), pooled across every
    window, with a two-sided binomial test against the 50% no-relationship baseline
    (days where either series is flat are excluded, since sign() has no direction).
    Positive lag = estimate change leads the price return by that many trading days;
    negative lag = price return leads the estimate change."""
    rows = {}
    for lag in range(-max_lag, max_lag + 1):
        aligned = pd.concat(
            pd.concat({"est": est, "ret": ret.shift(-lag)}, axis=1).dropna()
            for est, ret in window_pairs
        )
        aligned = aligned[(aligned["est"] != 0) & (aligned["ret"] != 0)]
        n = len(aligned)
        agree = int((np.sign(aligned["est"]) == np.sign(aligned["ret"])).sum())
        rows[lag] = {
            "agreement_rate": agree / n if n else np.nan,
            "n": n,
            "p_value": stats.binomtest(agree, n, 0.5).pvalue if n >= 3 else np.nan,
        }
    return pd.DataFrame(rows).T


sign_agreement = sign_agreement_pooled(window_pairs, max_lag=max_lag)
sign_agreement

	agreement_rate	n	p_value
-5	0.509839	6759.0	0.108359
-4	0.507527	6776.0	0.219830
-3	0.513701	6788.0	0.024733
-2	0.517039	6808.0	0.005112
-1	0.505051	6831.0	0.410653
0	0.507452	6844.0	0.222136
1	0.496564	6839.0	0.578051
2	0.504170	6835.0	0.498181
3	0.495313	6828.0	0.445812
4	0.503741	6817.0	0.544795
5	0.508602	6801.0	0.159540

colors = ["#C0623B" if lag == 0 else "#3B7AB8" for lag in sign_agreement.index]
fig = go.Figure()
fig.add_bar(
    x=list(sign_agreement.index),
    y=sign_agreement["agreement_rate"],
    marker_color=colors,
)
fig.add_hline(y=0.5, line=dict(color="black", width=1))

fig.update_layout(
    title="Sign agreement: estimate change vs. share price return, by lag (all tickers and market closes pooled)",
    template="plotly_white",
    height=420,
)
fig.update_yaxes(title="P(sign(est) == sign(ret))", tickformat=".0%")
fig.update_xaxes(title="lag, trading days (+ = estimate leads price)")
fig.show()

Unable to display output for mime type(s): application/vnd.plotly.v1+json

for ticker, market_close in estimate_prediction_change.columns:
    if ticker not in share_price_returns.columns:
        continue
    est = estimate_prediction_change[
        (ticker, market_close)
    ].dropna()  # only days the estimate moved
    ret = share_price_returns[ticker].dropna()
    if est.empty or ret.empty:
        continue

    # every trading day the estimate window spans, not just the change days
    ret = ret[(ret.index >= est.index.min()) & (ret.index <= est.index.max())]

    fig = make_subplots(specs=[[{"secondary_y": True}]])
    fig.add_scatter(
        x=ret.index,
        y=ret,
        name="share price return",
        mode="lines",
        line=dict(color="#C0623B"),
        secondary_y=True,
    )
    fig.add_scatter(
        x=est.index,
        y=est,
        name="estimate change",
        mode="markers",
        marker=dict(color="#3B7AB8", size=7),
        secondary_y=False,
    )

    fig.update_layout(
        title=f"{ticker} ({market_close}) — estimate change vs. share price return",
        template="plotly_white",
        height=360,
        margin=dict(l=60, r=60, t=50, b=40),
        hovermode="x unified",
    )
    fig.update_yaxes(title="Δ estimate", tickformat=".1%", secondary_y=False)
    fig.update_yaxes(title="log return", tickformat=".1%", secondary_y=True)
    fig.show()

Prediction market price vs. equity returns: which better predicts the actual resolution?

Everything above compares changes in the market’s implied estimate to changes in the share price — neither is checked against what actually happened. These markets ask “will [company] beat quarterly earnings?” and eventually resolve Yes/No (STANDARDIZED_MARKETS.RESOLUTION) once the company reports. That gives a ground truth to score two candidate predictors against:

• prediction market: the market’s own last quoted probability before market_close.
• equity returns: momentum in the underlying stock (cumulative log return) over the trading days leading up to market_close, before the earnings reaction.

Scored by AUC (rank discrimination) and Brier score / log loss (calibration) against the binary resolution.

RESOLUTION_QUERY = """
select distinct
    std_resolved_ticker as ticker,
    DATE(market_close)::VARCHAR as market_close,
    resolution
from PREDICTION_MARKET_FACTORS.PUBLIC.STANDARDIZED_MARKETS
where std_event_type = 'corporate KPI'
    and std_property_value = 'estimate'
"""

cursor = connector.cursor()
cursor.execute(RESOLUTION_QUERY)
resolution_raw = cursor.fetch_pandas_all()
resolution_raw.columns = [c.upper() for c in resolution_raw.columns]

resolution = (
    resolution_raw.drop_duplicates(["TICKER", "MARKET_CLOSE"])
    .set_index(["TICKER", "MARKET_CLOSE"])["RESOLUTION"]
    .eq("Yes")  # 1 = beat the estimate, 0 = missed
    .astype(int)
    .rename("resolution")
)

print(
    f"{len(resolution)} resolved markets, {resolution.mean():.1%} beat their estimate"
)

1247 resolved markets, 74.8% beat their estimate

# predictor 1: the market's own last quoted probability strictly before market_close
market_prob = {}
for ticker, market_close in wide.columns:
    close_date = pd.Timestamp(market_close)
    quotes = wide[(ticker, market_close)].dropna()
    quotes = quotes[quotes.index < close_date]
    if not quotes.empty:
        market_prob[(ticker, market_close)] = quotes.iloc[-1]
market_prob = pd.Series(market_prob, name="market_prob")
market_prob.index.names = ["TICKER", "MARKET_CLOSE"]

# predictor 2: cumulative log return in the underlying stock over the trailing
# window ending the trading day before market_close (momentum leading into the
# earnings date, not the earnings-day reaction itself)
MOMENTUM_WINDOW_DAYS = 10

equity_momentum = {}
for ticker, market_close in wide.columns:
    if ticker not in price_wide.columns:
        continue
    close_date = pd.Timestamp(market_close)
    quotes = price_wide[ticker].dropna()
    quotes = quotes[quotes.index < close_date]
    if len(quotes) < MOMENTUM_WINDOW_DAYS + 1:
        continue
    window = quotes.iloc[-(MOMENTUM_WINDOW_DAYS + 1) :]
    equity_momentum[(ticker, market_close)] = np.log(window.iloc[-1] / window.iloc[0])
equity_momentum = pd.Series(equity_momentum, name="momentum")
equity_momentum.index.names = ["TICKER", "MARKET_CLOSE"]

predictors = pd.concat([resolution, market_prob, equity_momentum], axis=1).dropna()
predictors.columns = ["resolution", "market_prob", "momentum"]
print(f"{len(predictors)} markets with both predictors and a known resolution")

1215 markets with both predictors and a known resolution

from sklearn.linear_model import LogisticRegression
from sklearn.metrics import brier_score_loss, log_loss, roc_auc_score

y = predictors["resolution"].to_numpy()


def evaluate(x: pd.Series, y: np.ndarray, fit_logistic: bool = True) -> dict:
    """AUC / Brier / log loss of x as a predictor of y.
    fit_logistic=True recalibrates x into a probability via pooled logistic
    regression first (needed for momentum, which isn't itself a probability);
    the raw market price is already a probability so it's also scored as-is."""
    x = x.to_numpy(dtype=float).reshape(-1, 1)
    prob = (
        LogisticRegression().fit(x, y).predict_proba(x)[:, 1]
        if fit_logistic
        else x.ravel()
    )
    return {
        "n": len(y),
        "auc": roc_auc_score(y, prob),
        "brier": brier_score_loss(y, prob),
        "log_loss": log_loss(y, prob, labels=[0, 1]),
    }


base_rate = np.full_like(y, y.mean(), dtype=float)
comparison = pd.DataFrame(
    {
        "prediction market (raw price)": evaluate(
            predictors["market_prob"], y, fit_logistic=False
        ),
        "prediction market (recalibrated)": evaluate(predictors["market_prob"], y),
        f"equity momentum ({MOMENTUM_WINDOW_DAYS}d)": evaluate(
            predictors["momentum"], y
        ),
        "base rate only": {
            "n": len(y),
            "auc": np.nan,
            "brier": brier_score_loss(y, base_rate),
            "log_loss": log_loss(y, base_rate, labels=[0, 1]),
        },
    }
).T

comparison

	n	auc	brier	log_loss
prediction market (raw price)	1215.0	0.793499	0.145201	0.451012
prediction market (recalibrated)	1215.0	0.793499	0.146253	0.456586
equity momentum (10d)	1215.0	0.574901	0.186285	0.559019
base rate only	1215.0	NaN	0.188830	0.565255

# robustness: is 10 trading days a favorable window, or does equity momentum stay
# weak at other horizons? and does momentum add anything on top of the market
# price rather than just being a noisier version of the same information?
momentum_windows = {}
for window_days in (5, 10, 20, 40):
    momentum = {}
    for ticker, market_close in wide.columns:
        if ticker not in price_wide.columns:
            continue
        close_date = pd.Timestamp(market_close)
        quotes = price_wide[ticker].dropna()
        quotes = quotes[quotes.index < close_date]
        if len(quotes) < window_days + 1:
            continue
        window = quotes.iloc[-(window_days + 1) :]
        momentum[(ticker, market_close)] = np.log(window.iloc[-1] / window.iloc[0])
    momentum = pd.Series(momentum)
    momentum.index.names = ["TICKER", "MARKET_CLOSE"]

    aligned = pd.concat([resolution, momentum], axis=1).dropna()
    aligned.columns = ["resolution", "momentum"]
    momentum_windows[f"{window_days}d"] = evaluate(
        aligned["momentum"], aligned["resolution"].to_numpy()
    )

momentum_window_sensitivity = pd.DataFrame(momentum_windows).T
momentum_window_sensitivity.index.name = "trailing window"
print(
    "equity momentum AUC/Brier by trailing window (market price for comparison: "
    f"auc={comparison.loc['prediction market (raw price)', 'auc']:.3f}, "
    f"brier={comparison.loc['prediction market (raw price)', 'brier']:.3f}):"
)
display(momentum_window_sensitivity)

joint_x = predictors[["market_prob", "momentum"]].to_numpy()
joint_prob = LogisticRegression().fit(joint_x, y).predict_proba(joint_x)[:, 1]
print(
    f"\njoint model (market price + momentum): "
    f"auc={roc_auc_score(y, joint_prob):.3f}, brier={brier_score_loss(y, joint_prob):.3f}"
    " -- essentially no improvement over the market price alone, i.e. momentum adds"
    " ~nothing once the market price is known."
)

equity momentum AUC/Brier by trailing window (market price for comparison: auc=0.793, brier=0.145):

	n	auc	brier	log_loss
trailing window
5d	1238.0	0.581445	0.187238	0.561339
10d	1218.0	0.573712	0.186410	0.559294
20d	1175.0	0.583929	0.185783	0.558313
40d	989.0	0.589310	0.182882	0.550752

joint model (market price + momentum): auc=0.796, brier=0.145 -- essentially no improvement over the market price alone, i.e. momentum adds ~nothing once the market price is known.

bar_labels = [
    "prediction market",
    f"equity momentum ({MOMENTUM_WINDOW_DAYS}d)",
    "base rate only",
]
bar_auc = [
    comparison.loc["prediction market (raw price)", "auc"],
    comparison.loc[f"equity momentum ({MOMENTUM_WINDOW_DAYS}d)", "auc"],
    0.5,
]
bar_brier = [
    comparison.loc["prediction market (raw price)", "brier"],
    comparison.loc[f"equity momentum ({MOMENTUM_WINDOW_DAYS}d)", "brier"],
    comparison.loc["base rate only", "brier"],
]

fig = make_subplots(
    rows=1,
    cols=2,
    subplot_titles=("AUC (higher = better)", "Brier score (lower = better)"),
)
fig.add_bar(
    x=bar_labels,
    y=bar_auc,
    marker_color=["#3B7AB8", "#C0623B", "#999999"],
    row=1,
    col=1,
)
fig.add_hline(y=0.5, line=dict(color="black", width=1, dash="dot"), row=1, col=1)
fig.add_bar(
    x=bar_labels,
    y=bar_brier,
    marker_color=["#3B7AB8", "#C0623B", "#999999"],
    row=1,
    col=2,
)

fig.update_layout(
    title="Predicting the actual resolution: prediction market price vs. equity momentum",
    template="plotly_white",
    height=420,
    showlegend=False,
)
fig.show()

Unable to display output for mime type(s): application/vnd.plotly.v1+json

Result

The prediction market’s own price is a much better predictor of whether a company actually beats its earnings estimate than pre-earnings share-price momentum: AUC ≈0.79 for the market price vs. ≈0.57-0.59 for momentum (barely above the 0.50 no-skill line, and stable across 5-40 day windows), and Brier score 0.145 vs. 0.183-0.187 (a naive “always predict the base rate” guess already scores 0.189, so momentum is only marginally better than not looking at the stock at all). Feeding both into a joint model barely moves AUC (0.793 → 0.796), so equity returns carry almost no information about the outcome beyond what the market has already priced in.

This isn’t surprising: the market price is a real-time forecast of this exact beat/miss event, aggregated from participants who can trade directly on it, while share-price momentum is only an indirect, noisy proxy for the same underlying expectation.