# -*- coding: utf-8 -*- """验证 开盘成交占比 → 次日反转 是真信号还是 小盘/流动性/开盘跳空 的伪装。 上一步: 开盘占比(10:00 bar amount / 全天) decile D10-D1 次日 -33.5% t=-8.6 (强负)。 本步三关: ① size 中性(5市值组内各自分档) ② 与开盘跳空(overnight gap)解耦(gap分层内) ③ 展示 各组平均市值/换手 看混淆程度。 survives 三关 → 可成文; 否则记负结论。 """ import sys, glob, os, json, sqlite3 sys.path.insert(0, "/root/cb-allotment/scripts") import numpy as np, pandas as pd from lib.factor_lab import DB, CACHE OSP = f"{CACHE}/open_share_panel.parquet" if os.path.exists(OSP): OS = pd.read_parquet(OSP); OS.index = pd.to_datetime(OS.index) print(f"OS 缓存载入 {OS.shape}", file=sys.stderr) else: frames = {} 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) day = d.groupby("date").amount.sum() a10 = d[d.t == "10:00:00"].set_index("date").amount frames[code] = (a10 / day.reindex(a10.index)).replace([np.inf, -np.inf], np.nan) OS = pd.DataFrame(frames); OS.index = pd.to_datetime(OS.index); OS = OS.sort_index() OS.to_parquet(OSP); print(f"OS 构建并缓存 {OS.shape}", file=sys.stderr) codes = list(OS.columns) conn = sqlite3.connect(DB) qs = ",".join(f"'{c}'" for c in codes) dd = pd.read_sql(f"SELECT trade_date,ts_code,open,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(OS.index) / 100 O = dd.pivot(index="dt", columns="ts_code", values="open").reindex(OS.index) PCL = dd.pivot(index="dt", columns="ts_code", values="pre_close").reindex(OS.index) CL = dd.pivot(index="dt", columns="ts_code", values="close").reindex(OS.index) GAP = (O / PCL - 1) # 开盘跳空 limit = ((CL / PCL - 1).abs() > 0.095) fwd = RET.shift(-1).where(~limit.shift(-1).astype(bool)) # 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(OS.index.union(MVm.index)).sort_index().ffill().reindex(OS.index) MV = MV[[c for c in codes if c in MV.columns]] def ann(v): return float(np.nanmean(v) * 244 * 100) # ① 原始 + size中性 def size_neutral(sig, n_size=5, n_dec=5): by = {q: [] for q in range(n_size)}; overall = []; szmean = {q: [] for q in range(n_size)} for t in OS.index[:-1]: u = sig.loc[t].notna() & fwd.loc[t].notna() & MV.loc[t].notna() s = sig.loc[t][u]; mv = MV.loc[t][u]; f = fwd.loc[t] if len(s) < 300: 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) < 30: continue dd_ = pd.qcut(sub.rank(method="first"), n_dec, labels=False, duplicates="drop") sp = f[sub.index[dd_ == dd_.max()]].mean() - f[sub.index[dd_ == 0]].mean() by[q].append(sp); ms.append(sp) if ms: overall.append(np.mean(ms)) ov = pd.Series(overall) return {"sn_spread": 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)}} # ② gap 解耦: gap 分5层内 decile OS def gap_neutral(n_gap=5, n_dec=5): overall = [] for t in OS.index[:-1]: u = OS.loc[t].notna() & fwd.loc[t].notna() & GAP.loc[t].notna() s = OS.loc[t][u]; g = GAP.loc[t][u]; f = fwd.loc[t] if len(s) < 300: continue gq = pd.qcut(g.rank(method="first"), n_gap, labels=False); ms = [] for q in range(n_gap): sub = s[gq == q] if len(sub) < 30: continue dd_ = pd.qcut(sub.rank(method="first"), n_dec, labels=False, duplicates="drop") ms.append(f[sub.index[dd_ == dd_.max()]].mean() - f[sub.index[dd_ == 0]].mean()) if ms: overall.append(np.mean(ms)) ov = pd.Series(overall) return {"gap_neutral_spread": round(ann(ov), 1), "gap_neutral_t": round(ov.mean()/ov.std()*np.sqrt(len(ov)), 1)} # 各 OS decile 的平均 log市值 / 平均换手代理(amount) —— 看混淆 def profile(groups=10): lm = {g: [] for g in range(groups)} for t in OS.index[:-1]: u = OS.loc[t].notna() & MV.loc[t].notna() s = OS.loc[t][u] if len(s) < 300: continue q = pd.qcut(s.rank(method="first"), groups, labels=False) for g in range(groups): idx = s.index[q == g] lm[g].append(np.log(MV.loc[t][idx]).mean()) return {"log_mv_by_decile": [round(float(np.nanmean(lm[g])), 2) for g in range(groups)]} def raw_decile(sig, groups=10): grp = {g: [] for g in range(groups)}; spread = [] for t in OS.index[:-1]: u = sig.loc[t].notna() & fwd.loc[t].notna() s = sig.loc[t][u]; f = fwd.loc[t] if len(s) < 300: continue q = pd.qcut(s.rank(method="first"), groups, labels=False) gm = {} for g in range(groups): r = f[s.index[q == g]].mean(); grp[g].append(r); gm[g] = r spread.append((t, gm[groups-1] - gm[0])) sp = pd.Series(dict(spread)) tr = sp[sp.index < "2022-01-01"]; te = sp[sp.index >= "2022-01-01"] 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(tr), 1), "test": round(ann(te), 1)} res = {"panel": [OS.shape[0], OS.shape[1]]} res["raw_decile"] = raw_decile(OS) res["size_neutral"] = size_neutral(OS) res["gap_neutral"] = gap_neutral() res["profile"] = profile() print(f"size中性: {res['size_neutral']['sn_spread']}% t={res['size_neutral']['sn_t']} by_size={res['size_neutral']['by_size']}", file=sys.stderr) print(f"gap中性: {res['gap_neutral']['gap_neutral_spread']}% t={res['gap_neutral']['gap_neutral_t']}", file=sys.stderr) print(f"log市值 D1..D10: {res['profile']['log_mv_by_decile']}", file=sys.stderr) json.dump(res, open("/root/cb-allotment/my-app/public/reports/opening-volume-reversal/backtest_result.json", "w"), ensure_ascii=False, indent=1) print("saved", file=sys.stderr)