| """Does the trend of price always align with the trend of sentiment? |
| |
| Three different questions hide inside that one, and they have different answers: |
| |
| 1. do they MOVE TOGETHER? contemporaneous correlation and sign agreement |
| 2. does one LEAD the other? cross-correlation across lags, Granger tests |
| 3. is the alignment STABLE? rolling correlation, and what follows a |
| divergence |
| |
| Sentiment series: the Crypto Fear & Greed Index, daily since 2018-02-01. It is |
| the only sentiment history available over this span, and one caveat has to be |
| stated before any number is read: the index is partly *constructed from price*. |
| Its published components include volatility and market momentum, so some |
| contemporaneous agreement with price is definitional rather than empirical. That |
| makes question 2 the one that carries information -- if sentiment merely encodes |
| recent price, it will track price with a lag and lead nothing. |
| |
| Usage: |
| python scripts/run_sentiment_alignment.py |
| python scripts/run_sentiment_alignment.py --max-lag 21 --window 90 |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import sys |
| import warnings |
| from pathlib import Path |
|
|
| import numpy as np |
| import pandas as pd |
| from scipy import stats |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT)) |
| warnings.filterwarnings("ignore") |
|
|
| from scripts.run_direction_sota_experiment import load_candles |
| from scripts.run_sentiment_replication import fear_greed |
|
|
| CACHE = ROOT / "outputs" / "volatility" / "btc_1h.csv" |
|
|
|
|
| def build() -> pd.DataFrame: |
| candles = load_candles("2018-01-01", CACHE) |
| daily = (candles.set_index("timestamp")["close"].resample("1D").last() |
| .dropna().rename("close").reset_index()) |
| daily["date"] = daily["timestamp"].dt.normalize() |
| df = daily.merge(fear_greed(), on="date", how="inner").sort_values("date").reset_index(drop=True) |
| df["price_chg"] = np.log(df["close"]).diff() |
| df["fg_chg"] = df["fear_greed"].diff() |
| |
| df["price_trend"] = np.log(df["close"]).diff(7) |
| df["fg_trend"] = df["fear_greed"].diff(7) |
| return df |
|
|
|
|
| def agreement(a: np.ndarray, b: np.ndarray) -> dict: |
| """How often do the two series move the same way, with a binomial CI?""" |
| ok = np.isfinite(a) & np.isfinite(b) & (a != 0) & (b != 0) |
| same = (np.sign(a[ok]) == np.sign(b[ok])) |
| n, k = len(same), int(same.sum()) |
| lo, hi = stats.binomtest(k, n).proportion_ci(0.95) |
| return dict(n=n, rate=k / n, lo=lo, hi=hi, |
| p_vs_coinflip=stats.binomtest(k, n, 0.5).pvalue) |
|
|
|
|
| def granger(y: np.ndarray, x: np.ndarray, lags: int) -> dict: |
| """Does past x improve a model of y beyond y's own past? (F test) |
| |
| Hand-rolled rather than taken from statsmodels, whose compiled extension is |
| blocked by this machine's application-control policy. Standard SSR F test: |
| regress y on its own lags, then on its own lags plus x's lags, and ask |
| whether the residual sum of squares falls by more than chance. |
| """ |
| def design(series, n_obs, offset): |
| return np.column_stack([series[offset - k: offset - k + n_obs] for k in range(1, lags + 1)]) |
|
|
| data = np.column_stack([y, x]) |
| data = data[np.isfinite(data).all(axis=1)] |
| yy, xx = data[:, 0], data[:, 1] |
| n_obs = len(yy) - lags |
| target = yy[lags:] |
| y_lags = design(yy, n_obs, lags) |
| x_lags = design(xx, n_obs, lags) |
| const = np.ones((n_obs, 1)) |
|
|
| def ssr(mat): |
| beta, *_ = np.linalg.lstsq(mat, target, rcond=None) |
| resid = target - mat @ beta |
| return float(resid @ resid) |
|
|
| ssr_r = ssr(np.hstack([const, y_lags])) |
| ssr_u = ssr(np.hstack([const, y_lags, x_lags])) |
| df_den = n_obs - (2 * lags + 1) |
| f = ((ssr_r - ssr_u) / lags) / (ssr_u / df_den) |
| return dict(lags=lags, F=float(f), p=float(stats.f.sf(f, lags, df_den)), |
| n=int(n_obs), r2_gain=float((ssr_r - ssr_u) / ssr_r)) |
|
|
|
|
| def main() -> None: |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--max-lag", type=int, default=21) |
| ap.add_argument("--window", type=int, default=90) |
| ap.add_argument("--granger-lags", type=int, default=7) |
| args = ap.parse_args() |
|
|
| df = build() |
| print(f"daily rows {len(df):,} {str(df['date'].min())[:10]} to {str(df['date'].max())[:10]}") |
| print("sentiment: Crypto Fear & Greed Index (partly built FROM price -- see module docstring)\n") |
|
|
| out: dict = {"rows": len(df)} |
|
|
| |
| print("=" * 92) |
| print("1. DO THEY MOVE TOGETHER?") |
| print("=" * 92) |
| pairs = [("level vs level", np.log(df["close"]), df["fear_greed"]), |
| ("daily change", df["price_chg"], df["fg_chg"]), |
| ("7-day trend", df["price_trend"], df["fg_trend"])] |
| out["correlations"] = {} |
| for name, a, b in pairs: |
| m = np.isfinite(a) & np.isfinite(b) |
| r, rp = stats.pearsonr(a[m], b[m]) |
| s, sp = stats.spearmanr(a[m], b[m]) |
| out["correlations"][name] = dict(pearson=r, pearson_p=rp, spearman=s, spearman_p=sp, n=int(m.sum())) |
| print(f" {name:<16} Pearson r = {r:+.3f} (p={rp:.2e}) Spearman = {s:+.3f} (p={sp:.2e}) n={int(m.sum()):,}") |
|
|
| print() |
| out["agreement"] = {} |
| for name, a, b in [("daily change", df["price_chg"], df["fg_chg"]), |
| ("7-day trend", df["price_trend"], df["fg_trend"])]: |
| g = agreement(a.to_numpy(), b.to_numpy()) |
| out["agreement"][name] = g |
| print(f" signs agree, {name:<14} {g['rate']:.1%} 95% CI [{g['lo']:.1%}, {g['hi']:.1%}]" |
| f" n={g['n']:,} p vs coin flip = {g['p_vs_coinflip']:.2e}") |
| print("\n -> 'always' is the wrong word if this is well below 100%.") |
|
|
| |
| print("\n" + "=" * 92) |
| print("2. DOES ONE LEAD THE OTHER?") |
| print(" correlation of price change at day t with sentiment change at day t+k") |
| print("=" * 92) |
| pc, fc = df["price_chg"].to_numpy(), df["fg_chg"].to_numpy() |
| rows = [] |
| for k in range(-args.max_lag, args.max_lag + 1): |
| shifted = pd.Series(fc).shift(-k).to_numpy() |
| m = np.isfinite(pc) & np.isfinite(shifted) |
| if m.sum() < 100: |
| continue |
| r, p = stats.pearsonr(pc[m], shifted[m]) |
| rows.append(dict(lag=k, r=r, p=p, n=int(m.sum()))) |
| out["cross_correlation"] = rows |
| best = max(rows, key=lambda r: abs(r["r"])) |
| for r in rows: |
| if abs(r["lag"]) <= 7: |
| bar = "#" * int(abs(r["r"]) * 60) |
| mark = " <= PEAK" if r["lag"] == best["lag"] else "" |
| meaning = ("sentiment leads price" if r["lag"] < 0 |
| else "price leads sentiment" if r["lag"] > 0 else "same day") |
| print(f" k={r['lag']:+3d} r={r['r']:+.3f} {bar:<34}{meaning}{mark}") |
| print(f"\n peak |r| at k = {best['lag']:+d} ({best['r']:+.3f}, p={best['p']:.2e})") |
| print(" -> k=0 or k>0 means sentiment reacts to price; only k<0 would mean it predicts price.") |
|
|
| print("\n Granger causality (daily changes, " |
| f"{args.granger_lags} lags):") |
| g1 = granger(fc, pc, args.granger_lags) |
| g2 = granger(pc, fc, args.granger_lags) |
| out["granger"] = {"price_to_sentiment": g1, "sentiment_to_price": g2} |
| print(f" price -> sentiment : F = {g1['F']:8.3f} p = {g1['p']:.3e}" |
| f" SSR reduction {g1['r2_gain'] * 100:5.2f}%" |
| f" {'SIGNIFICANT' if g1['p'] < 0.05 else 'not significant'}") |
| print(f" sentiment -> price : F = {g2['F']:8.3f} p = {g2['p']:.3e}" |
| f" SSR reduction {g2['r2_gain'] * 100:5.2f}%" |
| f" {'SIGNIFICANT' if g2['p'] < 0.05 else 'not significant'}") |
|
|
| |
| print("\n" + "=" * 92) |
| print("3. IS THE ALIGNMENT STABLE, OR DOES IT BREAK DOWN?") |
| print("=" * 92) |
| roll = (df.set_index("date")["price_trend"] |
| .rolling(args.window).corr(df.set_index("date")["fg_trend"])) |
| roll = roll.dropna() |
| out["rolling"] = dict(window=args.window, mean=float(roll.mean()), sd=float(roll.std()), |
| min=float(roll.min()), max=float(roll.max()), |
| frac_negative=float((roll < 0).mean()), |
| min_date=str(roll.idxmin())[:10], max_date=str(roll.idxmax())[:10]) |
| print(f" {args.window}-day rolling correlation of the 7-day trends:") |
| print(f" mean {roll.mean():+.3f} sd {roll.std():.3f} " |
| f"range [{roll.min():+.3f} on {str(roll.idxmin())[:10]}, " |
| f"{roll.max():+.3f} on {str(roll.idxmax())[:10]}]") |
| print(f" windows where the correlation is NEGATIVE: {(roll < 0).mean():.1%}") |
| for year, g in roll.groupby(roll.index.year): |
| print(f" {year}: mean {g.mean():+.3f} min {g.min():+.3f} max {g.max():+.3f}") |
|
|
| |
| print("\n" + "=" * 92) |
| print("4. WHEN THEY DISAGREE, WHAT HAPPENS NEXT?") |
| print(" (the only part with any trading relevance)") |
| print("=" * 92) |
| df["fwd_7d"] = np.log(df["close"]).shift(-7) - np.log(df["close"]) |
| aligned = np.sign(df["price_trend"]) == np.sign(df["fg_trend"]) |
| valid = df["price_trend"].notna() & df["fg_trend"].notna() & df["fwd_7d"].notna() |
| out["divergence"] = {} |
| print(f" {'state':<34}{'days':>7}{'mean fwd 7d':>14}{'median':>10}{'% up':>8}") |
| for label, mask in [ |
| ("aligned (both up or both down)", valid & aligned), |
| (" of which both UP", valid & aligned & (df["price_trend"] > 0)), |
| (" of which both DOWN", valid & aligned & (df["price_trend"] < 0)), |
| ("diverging", valid & ~aligned), |
| (" price up, sentiment down", valid & ~aligned & (df["price_trend"] > 0)), |
| (" price down, sentiment up", valid & ~aligned & (df["price_trend"] < 0)), |
| ]: |
| f = df.loc[mask, "fwd_7d"] |
| if len(f) < 20: |
| continue |
| out["divergence"][label.strip()] = dict(n=int(len(f)), mean=float(f.mean()), |
| median=float(f.median()), |
| pct_up=float((f > 0).mean())) |
| print(f" {label:<34}{len(f):>7,}{f.mean() * 100:>+13.2f}%{f.median() * 100:>+9.2f}%" |
| f"{(f > 0).mean() * 100:>7.1f}%") |
|
|
| a = df.loc[valid & aligned, "fwd_7d"] |
| d = df.loc[valid & ~aligned, "fwd_7d"] |
| t, p = stats.ttest_ind(a, d, equal_var=False) |
| out["divergence_test"] = dict(t=float(t), p=float(p), n_aligned=int(len(a)), n_diverging=int(len(d))) |
| print(f"\n aligned vs diverging, difference in next-7-day return: " |
| f"t = {t:+.2f}, p = {p:.3f} {'SIGNIFICANT' if p < 0.05 else 'not significant'}") |
|
|
| path = ROOT / "outputs" / "sentiment_alignment.json" |
| path.write_text(json.dumps(out, indent=2, default=str), encoding="utf-8") |
| print(f"\nwritten: {path}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
