# -*- coding: utf-8 -*-
"""高成交量收益溢价 (Gervais-Kaniel-Mingelgrin 2001 JF "The High-Volume Return Premium") A股版。

GKM(美股): 相对自身历史 异常放量 的股票, 未来一段跑赢; 缩量的跑输。是"关注度/可见性"溢价。
本站已知: 换手率 LEVEL 是反向信号(高换手=炒作温度计→未来跑输)。
本篇分清 LEVEL vs SHOCK: 换手率水平 与 相对自身历史的异常放量, 预测方向是否不同?

数据: monthly.parquet (turn=月均换手, mo=月收益, mv=市值)。全A月频. size中性. 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)
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")
print(f"面板 {RET.shape} {months.min().date()}..{months.max().date()}", file=sys.stderr)

# 信号
LEVEL = TURN
SHOCK6 = np.log(TURN / TURN.rolling(6, min_periods=3).mean())   # 相对6月均值 异常放量
SHOCK3 = np.log(TURN / TURN.rolling(3, min_periods=2).mean())

def decile(signal, groups=10):
    grp = {g: [] for g in range(groups)}
    spread = []
    for i, m in enumerate(months[:-1]):
        s = signal.loc[m][uni.loc[m]].dropna()
        if len(s) < 300: continue
        q = pd.qcut(s.rank(method="first"), groups, labels=False)
        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[groups - 1] - gm[0]))
    def ann(v): return float(np.nanmean(v) * 12 * 100)
    sp = pd.Series(dict(spread))
    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(sp.mean()/sp.std()*np.sqrt(len(sp)), 1),
            "monotonic_rho": round(rho, 2),
            "train": round(ann(sp[sp.index < TRAIN]), 1), "test": round(ann(sp[sp.index >= TRAIN]), 1)}

def size_neutral(signal, 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() & signal.loc[m].notna()
        s = signal.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
            d = pd.qcut(sub.rank(method="first"), n_dec, labels=False)
            sp = f[sub.index[d == n_dec-1]].mean() - f[sub.index[d == 0]].mean()
            by[q].append(sp); ms.append(sp)
        if ms: overall.append(np.mean(ms))
    def ann(v): return float(np.nanmean(v) * 12 * 100)
    ov = pd.Series(overall)
    return {"sn_spread_ann": round(ann(ov), 1), "sn_t": round(ov.mean()/ov.std()*np.sqrt(len(ov)), 1),
            "by_size": {f"Q{q+1}": round(ann(by[q]), 1) for q in range(n_size)}}

# 异常放量 × 当月涨跌方向 (GKM 是纯量, 但直觉: 放量上涨 vs 放量下跌)
def shock_by_dir(shock, groups=5):
    """放量组内, 按当月收益正负分, 看 forward。"""
    up_hi, dn_hi = [], []
    for m in months[:-1]:
        u = uni.loc[m] & shock.loc[m].notna() & RET.loc[m].notna()
        s = shock.loc[m][u]; r = RET.loc[m][u]; f = fwd.loc[m]
        if len(s) < 300: continue
        thr = s.quantile(0.8)   # 放量前20%
        hi = s.index[s >= thr]
        up = [c for c in hi if r[c] > 0]; dn = [c for c in hi if r[c] <= 0]
        if up: up_hi.append(f[up].mean())
        if dn: dn_hi.append(f[dn].mean())
    def ann(v): return float(np.nanmean(v) * 12 * 100)
    return {"放量上涨_fwd": round(ann(up_hi), 1), "放量下跌_fwd": round(ann(dn_hi), 1)}

result = {"panel": [RET.shape[0], RET.shape[1], str(months.min().date()), str(months.max().date())]}
for name, sig in [("turnover_level", LEVEL), ("volume_shock_6m", SHOCK6), ("volume_shock_3m", SHOCK3)]:
    result[name] = decile(sig)
    r = result[name]
    print(f"{name}: 价差{r['spread_ann']:+}% t={r['spread_t']} ρ={r['monotonic_rho']} tr{r['train']:+}/te{r['test']:+} "
          f"D1..D10={r['groups_ann']}", file=sys.stderr)
def grid2x2(shock, groups=5):
    """2x2: {放量top20%/缩量bot20%} × {当月涨/跌} -> forward 年化 + 基准(全样本涨/跌)。"""
    cells = {k: [] for k in ["放量涨", "放量跌", "缩量涨", "缩量跌", "全涨", "全跌"]}
    for m in months[:-1]:
        u = uni.loc[m] & shock.loc[m].notna() & RET.loc[m].notna()
        s = shock.loc[m][u]; r = RET.loc[m][u]; f = fwd.loc[m]
        if len(s) < 300: continue
        hi = s >= s.quantile(0.8); lo = s <= s.quantile(0.2); up = r > 0
        def mn(mask):
            idx = s.index[mask]; return f[idx].mean() if len(idx) else np.nan
        cells["放量涨"].append(mn(hi & up)); cells["放量跌"].append(mn(hi & ~up))
        cells["缩量涨"].append(mn(lo & up)); cells["缩量跌"].append(mn(lo & ~up))
        cells["全涨"].append(mn(up)); cells["全跌"].append(mn(~up))
    return {k: round(float(np.nanmean(v)) * 12 * 100, 1) for k, v in cells.items()}
result["grid_2x2"] = grid2x2(SHOCK6)
print(f"2x2: {result['grid_2x2']}", file=sys.stderr)

result["sn_turnover_level"] = size_neutral(LEVEL)
result["sn_volume_shock_6m"] = size_neutral(SHOCK6)
result["shock_by_direction"] = shock_by_dir(SHOCK6)
print(f"size中性: level {result['sn_turnover_level']['sn_spread_ann']}% | shock {result['sn_volume_shock_6m']['sn_spread_ann']}%", file=sys.stderr)
print(f"放量方向: {result['shock_by_direction']}", file=sys.stderr)
json.dump(result, open("/root/cb-allotment/my-app/public/reports/high-volume-premium/backtest_result.json", "w"), ensure_ascii=False, indent=1)
print("saved", file=sys.stderr)