"""
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")])