# -*- coding: utf-8 -*-
"""一天里哪半小时的涨跌最能预测明天? — 全时段 → 次日 系统扫描 (1736全池 30min)。

对每个 30 分钟 bar(10:00,10:30,...,15:00 共8段)算每股该段收益(close/open-1),
日度截面分 10 组, 看次日收益价差(D10-D1)。哪段最有信息? 尾盘反转是其中一段的特例。
"""
import sys, glob, json, sqlite3
sys.path.insert(0, "/root/cb-allotment/scripts")
import numpy as np, pandas as pd
from lib.factor_lab import DB, CACHE

BARS = ["10:00:00", "10:30:00", "11:00:00", "11:30:00", "13:30:00", "14:00:00", "14:30:00", "15:00:00"]
LABEL = {"10:00:00": "09:30-10:00 开盘", "10:30:00": "10:00-10:30", "11:00:00": "10:30-11:00",
         "11:30:00": "11:00-11:30", "13:30:00": "13:00-13:30", "14:00:00": "13:30-14:00",
         "14:30:00": "14:00-14:30", "15:00:00": "14:30-15:00 尾盘"}

# 每段收益面板
ret_by_bar = {b: {} for b in BARS}
for fp in glob.glob(f"{CACHE}/eod30/*.parquet"):
    code = fp.split("/")[-1].replace(".parquet", "").replace("_", ".", 1)
    d = pd.read_parquet(fp); d["dt"] = pd.to_datetime(d.trade_time); d["date"] = d.dt.dt.date
    d["t"] = d.dt.dt.time.astype(str)
    for b in BARS:
        bb = d[d.t == b].set_index("date")
        ret_by_bar[b][code] = (bb.close / bb.open - 1)
panels = {}
for b in BARS:
    P = pd.DataFrame(ret_by_bar[b]); P.index = pd.to_datetime(P.index); panels[b] = P.sort_index()
idx = panels["15:00:00"].index
codes = list(panels["15:00:00"].columns)
print(f"面板 {panels['15:00:00'].shape}", file=sys.stderr)

# 次日收益 + 涨跌停剔除
conn = sqlite3.connect(DB); qs = ",".join(f"'{c}'" for c in codes)
dd = pd.read_sql(f"SELECT trade_date,ts_code,close,pre_close,pct_chg FROM daily WHERE ts_code IN ({qs}) AND trade_date>='20160101'", conn)
conn.close()
dd["dt"] = pd.to_datetime(dd.trade_date, format="%Y%m%d")
RET = dd.pivot(index="dt", columns="ts_code", values="pct_chg").reindex(idx) / 100
C = dd.pivot(index="dt", columns="ts_code", values="close").reindex(idx)
PC = dd.pivot(index="dt", columns="ts_code", values="pre_close").reindex(idx)
limit = ((C / PC - 1).abs() > 0.095)
fwd = RET.shift(-1).where(~limit.shift(-1).astype(bool))

def ann(v): return float(np.nanmean(v) * 244 * 100)
def decile(sig, groups=10):
    grp = {g: [] for g in range(groups)}; spread = []
    common = sig.columns.intersection(fwd.columns)
    for t in idx[:-1]:
        s = sig.loc[t, common].dropna(); f = fwd.loc[t]
        s = s[f[s.index].notna()]
        if len(s) < 300: continue
        q = pd.qcut(s.rank(method="first"), groups, labels=False, duplicates="drop")
        gm = {}
        for g in range(groups):
            r = f[s.index[q == g]].mean(); grp[g].append(r); gm[g] = r
        spread.append(gm.get(groups-1, np.nan) - gm.get(0, np.nan))
    sp = pd.Series(spread).dropna()
    return {"spread_ann": round(ann(sp), 1), "spread_t": round(float(sp.mean()/sp.std()*np.sqrt(len(sp))), 1),
            "d10": round(ann(grp[groups-1]), 1), "d1": round(ann(grp[0]), 1)}

# size 中性 (mv 月度 ffill 到日), 只对 开盘/尾盘 两端做
mo = pd.read_parquet(f"{CACHE}/monthly.parquet")
MVm = mo["mv"].unstack("ts_code"); MVm.index = pd.to_datetime(MVm.index)
MV = MVm.reindex(idx.union(MVm.index)).sort_index().ffill().reindex(idx)
def size_neutral(sig, n_size=5, n_dec=5):
    overall = []
    common = sig.columns.intersection(fwd.columns).intersection(MV.columns)
    for t in idx[:-1]:
        s = sig.loc[t, common].dropna(); mv = MV.loc[t, common]; f = fwd.loc[t]
        u = s.index[f[s.index].notna() & mv[s.index].notna()]
        s = s[u]
        if len(s) < 400: continue
        sz = pd.qcut(mv[s.index].rank(method="first"), n_size, labels=False); ms = []
        for q in range(n_size):
            sub = s[sz == q]
            if len(sub) < 30: 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": [panels["15:00:00"].shape[0], panels["15:00:00"].shape[1]], "by_bar": {}}
for b in BARS:
    r = decile(panels[b]); res["by_bar"][b] = {"label": LABEL[b], **r}
    print(f"{LABEL[b]:16s}: D10-D1 价差 {r['spread_ann']:+6.1f}% t={r['spread_t']:+5.1f} (D1 {r['d1']:+.0f} / D10 {r['d10']:+.0f})", file=sys.stderr)
for b in ["10:00:00", "15:00:00"]:
    sn, snt = size_neutral(panels[b]); res["by_bar"][b]["sn_spread"] = sn; res["by_bar"][b]["sn_t"] = snt
    print(f"size中性 {LABEL[b]}: {sn}% t={snt}", file=sys.stderr)
json.dump(res, open("/root/cb-allotment/research/timeofday_scan_result.json", "w"), ensure_ascii=False, indent=1)
print("saved", file=sys.stderr)