# -*- coding: utf-8 -*-
"""「情绪过热」综合信号: 合并 换手率 + 异常放量 + 特质波动 三个反向信号, 是否强于单个?

本站已分别证明(全A月频, 反向): 换手率水平高→跑输(炒作温度计)、异常放量→跑输(GKM反转)、
特质波动 ivol 高→跑输(A股ivol之谜)。三者都是"关注度/情绪过热"的不同侧面。
问: 把它们 z 合成一个"过热分", decile 是否比任一单信号 更强/更单调/更稳健(OOS+size中性)?
数据: monthly.parquet (turn/ivol/mo/mv). 全A月频. train<2022/test.
"""
import sys, json, numpy as np, pandas as pd
sys.path.insert(0, "/root/cb-allotment/scripts")

mo = pd.read_parquet("/root/cb-allotment/data/factor_lab/monthly.parquet")
RET = mo["mo"].unstack("ts_code").sort_index()
TURN = mo["turn"].unstack("ts_code").reindex_like(RET)
IVOL = mo["ivol"].unstack("ts_code").reindex_like(RET)
MV = mo["mv"].unstack("ts_code").reindex_like(RET)
PX = mo["close"].unstack("ts_code").reindex_like(RET)
months = RET.index; fwd = RET.shift(-1)
uni = (PX > 2) & TURN.notna() & RET.notna()
TRAIN = pd.Timestamp("2022-01-01")

SHOCK = np.log(TURN / TURN.rolling(6, min_periods=3).mean())   # 异常放量

def z(df):
    r = df.rank(axis=1)
    return r.sub(r.mean(axis=1), axis=0).div(r.count(axis=1), axis=0)

# 三信号(均 高=过热=预期反向) + 合成
zt, zi, zs = z(TURN), z(IVOL), z(SHOCK)
COMPOSITE = (zt + zi + zs) / 3

def ann(v): return float(np.nanmean(v) * 12 * 100)

def decile(sig, groups=10):
    grp = {g: [] for g in range(groups)}; spread = []
    for m in months[:-1]:
        s = sig.loc[m][uni.loc[m]].dropna()
        if len(s) < 300: continue
        q = pd.qcut(s.rank(method="first"), groups, labels=False, duplicates="drop")
        f = fwd.loc[m]; gm = {}
        for g in range(groups):
            r = f[s.index[q == g]].mean(); grp[g].append(r); gm[g] = r
        spread.append((m, gm.get(groups-1, np.nan) - gm.get(0, np.nan)))
    sp = pd.Series(dict(spread)).dropna()
    rho = pd.Series([ann(grp[g]) for g in range(groups)]).corr(pd.Series(range(groups)), method="spearman")
    return {"groups_ann": [round(ann(grp[g]), 1) for g in range(groups)],
            "spread_ann": round(ann(sp), 1), "spread_t": round(float(sp.mean()/sp.std()*np.sqrt(len(sp))), 1),
            "monotonic_rho": round(float(rho), 2),
            "train": round(ann(sp[sp.index < TRAIN]), 1), "test": round(ann(sp[sp.index >= TRAIN]), 1),
            "hitrate_pos": round(float((sp < 0).mean()) * 100, 0)}   # 反向: 价差<0 的月占比

def size_neutral(sig, n_size=5, n_dec=5):
    by = {q: [] for q in range(n_size)}; overall = []
    for m in months[:-1]:
        u = uni.loc[m] & MV.loc[m].notna() & sig.loc[m].notna()
        s = sig.loc[m][u]; mv = MV.loc[m][u]; f = fwd.loc[m]
        if len(s) < 500: continue
        sz = pd.qcut(mv.rank(method="first"), n_size, labels=False); ms = []
        for q in range(n_size):
            sub = s[sz == q]
            if len(sub) < 40: continue
            dq = pd.qcut(sub.rank(method="first"), n_dec, labels=False, duplicates="drop")
            ms.append(f[sub.index[dq == dq.max()]].mean() - f[sub.index[dq == 0]].mean())
        if ms: overall.append(np.mean(ms))
    ov = pd.Series(overall)
    return round(ann(ov), 1), round(float(ov.mean()/ov.std()*np.sqrt(len(ov))), 1)

res = {"panel": [RET.shape[0], RET.shape[1], str(months.min().date()), str(months.max().date())]}
for name, sig in [("turnover", TURN), ("ivol", IVOL), ("volshock", SHOCK), ("composite", COMPOSITE)]:
    d = decile(sig); sn, snt = size_neutral(sig)
    d["sn_spread"] = sn; d["sn_t"] = snt
    res[name] = d
    print(f"{name:10s}: 价差{d['spread_ann']:+6.1f}% t={d['spread_t']:+5.1f} ρ={d['monotonic_rho']:+.2f} "
          f"tr{d['train']:+.0f}/te{d['test']:+.0f} 反向胜率{d['hitrate_pos']:.0f}% | size中性{sn:+.1f}% t={snt:+.1f}", file=sys.stderr)
json.dump(res, open("/root/cb-allotment/my-app/public/reports/sentiment-overheating/backtest_result.json", "w"), ensure_ascii=False, indent=1)
print("saved", file=sys.stderr)