| """ |
| rotation.py — US sector capital-rotation monitor. |
| |
| True money-flow data for US equities (e.g. tick-rule buy/sell imbalance per |
| sector) is a paid dataset. The standard free proxy — used here — is the family |
| of 11 SPDR sector ETFs, which together cover the entire S&P 500: |
| |
| flow_proxy($) = price_change% × dollar volume (close × volume) |
| |
| A sector whose ETF rises on heavy dollar volume is absorbing capital; one that |
| falls on heavy dollar volume is shedding it. We also compute relative strength |
| vs SPY so "everything up / everything down" days don't mask rotation. |
| |
| Windows: 1-day (today's rotation), 5-day (week trend), 20-day (month trend). |
| """ |
| from __future__ import annotations |
|
|
| import pandas as pd |
|
|
| import data_us |
|
|
| SECTOR_ETFS = { |
| "XLK": "Technology", |
| "XLF": "Financials", |
| "XLE": "Energy", |
| "XLV": "Health Care", |
| "XLY": "Consumer Discretionary", |
| "XLP": "Consumer Staples", |
| "XLI": "Industrials", |
| "XLB": "Materials", |
| "XLU": "Utilities", |
| "XLRE": "Real Estate", |
| "XLC": "Communication Services", |
| } |
| BENCH = "SPY" |
|
|
|
|
| def _window_stats(df: pd.DataFrame, n: int): |
| """Return (pct_change, avg dollar volume, flow proxy $) over last n bars.""" |
| if df is None or len(df) < n + 1: |
| return None |
| closes = df["close"].iloc[-(n + 1):] |
| pct = float(closes.iloc[-1] / closes.iloc[0] - 1.0) |
| dvol = float((df["close"].iloc[-n:] * df["volume"].iloc[-n:]).mean()) |
| return pct, dvol, pct * dvol |
|
|
|
|
| def build_rotation(force: bool = False): |
| """Compute the rotation table. Returns (df_1d, df_5d, df_20d, asof_str).""" |
| frames = {} |
| for tk in list(SECTOR_ETFS) + [BENCH]: |
| frames[tk] = data_us.load_level(tk, "d", force=force) |
|
|
| spy = frames[BENCH] |
| asof = "—" |
| if spy is not None and len(spy): |
| asof = pd.Timestamp(spy["date"].iloc[-1]).strftime("%Y-%m-%d") |
|
|
| out = {} |
| for n, label in ((1, "1D"), (5, "5D"), (20, "20D")): |
| rows = [] |
| spy_stats = _window_stats(spy, n) |
| spy_pct = spy_stats[0] if spy_stats else 0.0 |
| for tk, name in SECTOR_ETFS.items(): |
| st = _window_stats(frames[tk], n) |
| if st is None: |
| continue |
| pct, dvol, flow = st |
| rows.append({ |
| "Sector": name, |
| "ETF": tk, |
| "Change": pct, |
| "Avg $ Volume": dvol, |
| "Flow proxy": flow, |
| "RS vs SPY": pct - spy_pct, |
| }) |
| if not rows: |
| out[label] = pd.DataFrame() |
| continue |
| df = pd.DataFrame(rows).sort_values("Flow proxy", ascending=False).reset_index(drop=True) |
| out[label] = df |
| return out.get("1D"), out.get("5D"), out.get("20D"), asof |
|
|
|
|
| def fmt_table(df: pd.DataFrame) -> pd.DataFrame: |
| """Human-readable formatting for the UI.""" |
| if df is None or df.empty: |
| return pd.DataFrame(columns=["Sector", "ETF", "Change", "$ Volume (avg)", |
| "Flow proxy", "RS vs SPY", "Read"]) |
| d = df.copy() |
|
|
| def _money(x): |
| ax = abs(x) |
| if ax >= 1e9: |
| return f"${x/1e9:,.2f}B" |
| if ax >= 1e6: |
| return f"${x/1e6:,.1f}M" |
| return f"${x:,.0f}" |
|
|
| def _read(row): |
| if row["Flow proxy"] > 0 and row["RS vs SPY"] > 0: |
| return "🟢 Inflow + leading" |
| if row["Flow proxy"] > 0: |
| return "🟩 Inflow" |
| if row["Flow proxy"] < 0 and row["RS vs SPY"] < 0: |
| return "🔴 Outflow + lagging" |
| if row["Flow proxy"] < 0: |
| return "🟥 Outflow" |
| return "—" |
|
|
| d["Read"] = d.apply(_read, axis=1) |
| d["$ Volume (avg)"] = d["Avg $ Volume"].map(_money) |
| d["Flow proxy"] = d["Flow proxy"].map(_money) |
| d["Change"] = d["Change"].map(lambda x: f"{x:+.2%}") |
| d["RS vs SPY"] = d["RS vs SPY"].map(lambda x: f"{x:+.2%}") |
| return d[["Sector", "ETF", "Change", "$ Volume (avg)", "Flow proxy", "RS vs SPY", "Read"]] |
|
|
|
|
| def rotation_brief(df_1d, df_5d, df_20d) -> str: |
| """Plain-text summary fed to the local LLM (and shown as fallback).""" |
| def top_bottom(df, label): |
| if df is None or df.empty: |
| return f"{label}: no data." |
| top = df.head(3) |
| bot = df.tail(3) |
| t = ", ".join(f"{r.Sector} ({r.Change:+.2%}, RS {r['RS vs SPY']:+.2%})" |
| for _, r in top.iterrows()) |
| b = ", ".join(f"{r.Sector} ({r.Change:+.2%}, RS {r['RS vs SPY']:+.2%})" |
| for _, r in bot.iterrows()) |
| return f"{label} — capital flowing INTO: {t}. Flowing OUT OF: {b}." |
| return "\n".join([top_bottom(df_1d, "1-day"), |
| top_bottom(df_5d, "5-day"), |
| top_bottom(df_20d, "20-day")]) |
|
|
