企业官网建设流程全解析

因果对冲是一种实测有效的执行策略，这里示例一个理想化的zpos因果对冲的代码实现。

包括对冲book来源，因果beta如何估计，对冲如何执行，如何通过逐日执行获得收益。

1 因果对冲

1.1 因果对冲计算

经典的因果对冲执行逻辑流程如下

1）每个交易日 t，用截至 t-1 的 60 日滚动窗口估计zpos book对benchmark的beta，

2）令当日空头比率等于该滞后 beta，然后计算

3）最后以零基准评估其绝对收益、波动、Sharpe 和最大回撤；

1.2 因果对冲代码

这里实现了一个实时、无前视的简化的动态beta 因果对冲(causal hedge)。

每天用截至前一日滚动beta决定当日做空基准的比例，再从策略book收益中扣掉这部分基准暴露。

"""Beta-neutralization experiment — separate factor alpha from market exposure. Motivation ---------- Every long-only monetization (ew_top50 / rank_all / zpos) of the surviving skewness signal carries full market beta: max drawdowns run -48%..-68% because the books ride the underlying index through 2015/2018/2024. The market cycle dwarfs the few %/yr of factor alpha, so "excess vs EW" is dominated by beta mismatch rather than factor P&L (see docs/research_findings.md). Two separable questions, both labelled explicitly: Q1 feature-level (cross-sectional) neutralization Before ranking, strip the systematic beta tilt out of the composite: z_tilde_i = z_i - (a + b_t * beta_i) (OLS residual per date) with beta_i = 60d trailing CAPM beta of stock i vs the equal-weight universe. Nothing about the ranking mechanics changes; only the score fed to it is orthogonalized. Reported as the zpos book's ex-ante beta tilt before/after (should go ~0 by construction) and the resulting zpos/rank_all/zls metrics. Q2 portfolio-level hedge (market-neutrality) Hold the long-only book and short beta_book x benchmark: r_hedged_t = r_book_t - beta_hedge_t * r_bm_t Two beta_hedge estimators are reported: * ex-post : full-sample regression of the book on the benchmark (a hindsight diagnostic = the frictionless-hedge ceiling) * causal : 60d trailing beta, shifted so day t uses data <= t-1 (estimable in real time) A hedged book has beta ~ 0, so IR-vs-benchmark is a category error: we report annualized return, ann vol, absolute Sharpe and max drawdown. Conventions (entry-only cost, halt renormalization by absolute weight mass, last-day gross = 0) are mirrored from BaseEngine / ic_weighted_test.evaluate and NOT re-derived. ic_weighted_test.py stays frozen: its JSONs are the reference set of docs/research_findings.md §3/§5. """ import argparse import json import math import os import sys import numpy as np import pandas as pd sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import ic_weighted_test as base from mfm.combine import combine_factors_icir_weight from mfm.pipelines.csi300_screen import ( CSI300ScreenConfig, build_rolling_weights, load_universe, prepare_context, validate_and_screen, ) BETA_WINDOW = 60 # trailing window for per-stock CAPM beta BETA_MINP = 30 HEDGE_WINDOW = 60 # trailing window for the book's realized beta HEDGE_MINP = 30 COMPOSITES = { "skew": ["skewness"], "core3": ["skewness", "vol_20d", "vol_60d"], } def compute_market_beta(returns, window=BETA_WINDOW, minp=BETA_MINP): """60d trailing CAPM beta of each stock vs the equal-weight universe.""" r_mkt = returns.mean(axis=1) var = r_mkt.rolling(window, min_periods=minp).var() cov = returns.rolling(window, min_periods=minp).cov(r_mkt) return cov.div(var, axis=0) def neutralize(z, beta, min_obs=12): """Per-date OLS: keep the residual of z on beta (feature-level neutrality).""" resid = pd.DataFrame(np.nan, index=z.index, columns=z.columns) cols = z.columns for t in z.index: x = beta.loc[t].reindex(cols) y = z.loc[t].reindex(cols) m = x.notna() & y.notna() nm = int(m.sum()) if nm < min_obs: continue xm = x[m].to_numpy(dtype=float) ym = y[m].to_numpy(dtype=float) A = np.column_stack([np.ones(nm), xm]) coef, *_ = np.linalg.lstsq(A, ym, rcond=None) resid.loc[t, cols[m]] = ym - A @ coef return resid def build_scheme_returns(comp, ret_next, rebal_dates, cfg, scheme): """Net daily return series for ONE scheme. Mirrors ic_weighted_test.evaluate's inner loop verbatim for `scheme` (zpos / rank_all), returning the (gross - cost) net Series so it can be hedged at the portfolio level. """ idx = comp.index all_cols = ret_next.columns w_rows, costs = {}, {} prev_support, prev_w = None, None for T in rebal_dates: if T not in comp.index: continue w = base.scheme_weights(comp.loc[T], scheme, cfg.n_hold).reindex(all_cols).fillna(0.0) w_rows[T] = w support = w.index[w != 0.0] if prev_support is None: notional = float(w.abs().sum()) # full cost on initial entry else: entered = support.difference(prev_support) notional = float(w.loc[entered].abs().sum()) if len(entered) else 0.0 costs[T] = cfg.cost * notional prev_support, prev_w = support, w W = pd.DataFrame(w_rows).T.reindex(idx).ffill().fillna(0.0) valid = ret_next.notna() & (W != 0.0) w_valid = W.where(valid, 0.0) mass = w_valid.abs().sum(axis=1) numer = (w_valid * ret_next.fillna(0.0)).sum(axis=1) gross = (numer / mass.where(mass != 0.0)).fillna(0.0) gross.iloc[-1] = 0.0 # engine last-day convention cost_s = pd.Series(0.0, index=idx) for T, c in costs.items(): cost_s.loc[T] = c return gross - cost_s def exante_beta_tilt(comp, beta, rebal_dates, cfg, scheme): """Weighted-avg 60d beta of the book's target weights (rebalance-time).""" tilts = [] for T in rebal_dates: if T not in comp.index: continue w = base.scheme_weights(comp.loc[T], scheme, cfg.n_hold) b = beta.loc[T] common = w.index.intersection(b.dropna().index) if len(common) == 0: continue num = float((w.loc[common] * b.loc[common]).sum()) den = float(w.loc[common].abs().sum()) if den > 0: tilts.append(num / den) return float(np.mean(tilts)) if tilts else float("nan") def realized_beta(ret, bm): """Full-sample regression beta of a book on the benchmark (ex-post).""" a = ret - ret.mean() b = bm - bm.mean() denom = float((b * b).sum()) return float((a * b).sum() / denom) if denom > 0 else 0.0 def rolling_hedge_beta(book, bm, window=HEDGE_WINDOW, minp=HEDGE_MINP): """Causal 60d book beta; day t uses data <= t-1 (shifted).""" var = bm.rolling(window, min_periods=minp).var() cov = book.rolling(window, min_periods=minp).cov(bm) return (cov / var).shift(1) def hedge(book, bm, beta_hedge): """r_book - beta_hedge * r_bm (beta_hedge: float or aligned Series).""" bm_a = bm.reindex(book.index).fillna(0.0) if isinstance(beta_hedge, pd.Series): bh = beta_hedge.reindex(book.index).fillna(0.0) else: bh = beta_hedge return book - bh * bm_a def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--tag", required=True) ap.add_argument("--start", required=True) ap.add_argument("--end", required=True) ap.add_argument("--instruments", default="data/cn_data/instruments/csi300.txt") ap.add_argument("--outdir", default="./output_btn") args = ap.parse_args() os.makedirs(args.outdir, exist_ok=True) cfg = CSI300ScreenConfig( start=args.start, end=args.end, instruments_path=args.instruments, out_dir=args.outdir, ) print(f"[1/4] Loading & validating ({args.tag})...") prices, returns, mask, uf = load_universe(cfg) screen = validate_and_screen(uf, prices, cfg) print("[2/4] Rolling weights (shared OOS window)...") roll_icir, roll_blend, oos_idx = build_rolling_weights(screen, prices.index, cfg) ctx = prepare_context(prices, returns, mask, uf, roll_icir, roll_blend, oos_idx, cfg) print("[3/4] Market beta (60d trailing vs EW universe)...") beta_all = compute_market_beta(returns).loc[oos_idx] ret_next = returns.shift(-1).loc[oos_idx] bm = ctx.benchmark zero = pd.Series(0.0, index=bm.index) out = { "tag": args.tag, "oos_window": [str(oos_idx[0].date()), str(oos_idx[-1].date())], "oos_days": int(len(oos_idx)), "beta_window": BETA_WINDOW, "hedge_window": HEDGE_WINDOW, "composites": {}, } print("[4/4] Neutralization & hedge...") for cname, factors in COMPOSITES.items(): w = { f: base.FIXED_SIGNS.get(f, 1.0 if screen.icir[f] >= 0 else -1.0) for f in factors } comp_raw = combine_factors_icir_weight(ctx.norm_all, w).where(ctx.mask_oos) comp_neu = neutralize(comp_raw, beta_all).where(ctx.mask_oos) cell = {"weights": w, "schemes_raw": None, "schemes_neutral": None, "hedge": {}} cell["schemes_raw"] = base.evaluate(comp_raw, ret_next, ctx.rebal_dates, bm, cfg) cell["schemes_neutral"] = base.evaluate(comp_neu, ret_next, ctx.rebal_dates, bm, cfg) tilt_raw = exante_beta_tilt(comp_raw, beta_all, ctx.rebal_dates, cfg, "zpos") tilt_neu = exante_beta_tilt(comp_neu, beta_all, ctx.rebal_dates, cfg, "zpos") cell["zpos_exante_beta_tilt"] = {"raw": tilt_raw, "neutral": tilt_neu} book_raw = build_scheme_returns(comp_raw, ret_next, ctx.rebal_dates, cfg, "zpos") book_neu = build_scheme_returns(comp_neu, ret_next, ctx.rebal_dates, cfg, "zpos") for bk_name, book in (("raw", book_raw), ("neutral", book_neu)): bep = realized_beta(book, bm) broll = rolling_hedge_beta(book, bm) cell["hedge"][bk_name] = { "book_beta_expost": bep, "unhedged": base.calc_metrics(book, zero), "hedged_expost": base.calc_metrics(hedge(book, bm, bep), zero), "hedged_causal": base.calc_metrics(hedge(book, bm, broll), zero), } out["composites"][cname] = cell wtxt = ", ".join(f"{k2}:{v2:+.0f}" for k2, v2 in w.items()) print(f"\n=== [{cname}] w=({wtxt}) zpos ex-ante beta tilt: " f"raw={tilt_raw:+.2f} neutral={tilt_neu:+.2f}") print(" -- feature-level (Q1): raw vs neutralized composite --") for name in ("ew_top50", "rank_all", "zpos", "zls"): mr = cell["schemes_raw"][name] mn = cell["schemes_neutral"][name] print(f" {name:9s} raw IR={mr['information_ratio']:+.2f} " f"ann={mr['annual_return']:+.2%} maxDD={mr['max_drawdown']:.1%}" f" | neu IR={mn['information_ratio']:+.2f} " f"ann={mn['annual_return']:+.2%} maxDD={mn['max_drawdown']:.1%}") print(" -- portfolio hedge (Q2): zpos unhedged vs ex-post vs causal --") for bk_name in ("raw", "neutral"): h = cell["hedge"][bk_name] u, ex, cx = h["unhedged"], h["hedged_expost"], h["hedged_causal"] vol_u = book_raw.std() * math.sqrt(252) if bk_name == "raw" \ else book_neu.std() * math.sqrt(252) print(f" [{bk_name}] book beta(ex-post)={h['book_beta_expost']:+.2f} " f"unhedged vol={vol_u:.1%}") print(f" unhedged ann={u['annual_return']:+.2%} " f"sharpe={u['sharpe_ratio']:+.2f} maxDD={u['max_drawdown']:.1%}") print(f" hedged_ex ann={ex['annual_return']:+.2%} " f"sharpe={ex['sharpe_ratio']:+.2f} maxDD={ex['max_drawdown']:.1%}") print(f" hedged_caus ann={cx['annual_return']:+.2%} " f"sharpe={cx['sharpe_ratio']:+.2f} maxDD={cx['max_drawdown']:.1%}") path = os.path.join(args.outdir, f"btn_{args.tag}.json") with open(path, "w") as fh: json.dump(out, fh, indent=1, ensure_ascii=False) print(f"\nSaved -> {path}") if __name__ == "__main__": main()

2 对冲链路梳理

这里基于以上示例代码，按执行链路进行深入详细的梳理。

2.1 因果对冲位置

这里因果对冲实验分两条线：

1）Q1特征层中性化

neutralize(comp_raw, beta_all)

在排序前把composite 对股票beta做横截面OLS，取残差。

改变的是选股分数。

2）Q2 组合层对冲

hedge(book, bm, beta_hedge)`

不改变选股，只改变收益序列：

其中causal版本用滚动、滞后一期的 beta。

Q2只对zpos方案生成的book做对冲，不是对ew_top50 / rank_all / zls全部做。

2.2 被对冲的book怎么来

这里示例被对冲的book的计算过程，具体为build_scheme_returns。

build_scheme_returns(comp, ret_next, rebal_dates, cfg, "zpos")

负责生成zpos组合的日净收益，细节如下：

1. 每个调仓日T，用base.scheme_weights(comp.loc[T], "zpos", n_hold)生成目标权重。

2. 记录权重，计算 entry-only 成本：

- 初始建仓：全部绝对权重；

- 后续调仓：只对新增股票entered的绝对权重收费。
3. 将调仓权重ffill到日频，得到W。
4. 用 ret_next计算每日 gross：

并对无效收益、零权重做质量归一化。
5. 扣成本，最后一日 gross.iloc[-1] = 0.0，得到 book。

所以book_raw和book_neu分别是：

- 原始 composite 的zpos净收益；

- 特征中性化后 composite 的zpos净收益。

2.3 因果beta的估计

因果beta的估计的实现rolling_hedge_beta

def rolling_hedge_beta(book, bm, window=60, minp=30):
var = bm.rolling(window, min_periods=minp).var()
cov = book.rolling(window, min_periods=minp).cov(bm)
return (cov / var).shift(1)

rolling_hedge_beta实现逻辑如下

1. 对基准 bm计算 60 日滚动方差。

2. 对 book和bm计算 60 日滚动协方差。

3. 得到滚动 beta：

4. .shift(1)：

即t日使用的对冲比率只基于t-1及之前的数据，避免用到t日收益，防止前视偏差。

rolling_hedge_beta其他参数说明如下

1）BETA_WINDOW = 60，滚动窗口 60 天。

2）BETA_MINP = 30，至少 30 个观测才计算。

因此前约30个交易日beta为NaN，后续fillna(0.0)后会变成 0，等价于早期不对冲。

2.4 对冲执行

对冲执行函数是hedge，具体如下：

def hedge(book, bm, beta_hedge):
bm_a = bm.reindex(book.index).fillna(0.0)
if isinstance(beta_hedge, pd.Series):
bh = beta_hedge.reindex(book.index).fillna(0.0)
else:
bh = beta_hedge
return book - bh * bm_a

hedge逐日执行如下计算

实现细节如下

- bm对齐到book.index，缺失基准收益按 0 处理。

- 如果beta_hedge是 Series，则按日对齐；缺失 beta 按 0 处理。

- 如果beta_hedge是 float，则是常数对冲，例如全样本 ex-post beta。

2.5 main实际调用链

main示例了因果对冲的实际对以上函数实现的调用链条。

1）调用build_scheme_returns

对每个 composite，例如skew或core3：

book_raw = build_scheme_returns(comp_raw, ret_next, ctx.rebal_dates, cfg, "zpos")
book_neu = build_scheme_returns(comp_neu, ret_next, ctx.rebal_dates, cfg, "zpos")

2）计算beta

然后对raw和neutral两个book 分别做：

bep = realized_beta(book, bm) # 全样本 ex-post beta
broll = rolling_hedge_beta(book, bm) # 因果滚动 beta

其中：

realized_beta(book, bm)

用全样本回归得到book对 benchmark 的 beta：

这是上帝视角，用了全样本信息，不能实盘，只作为摩擦无成本对冲上限诊断。

rolling_hedge_beta(book, bm)

得到逐日因果 beta，用于实盘可执行版本。

3）最后计算三组指标

这里为计算三种不同对冲实现的年化指标。

"unhedged": base.calc_metrics(book, zero),
"hedged_expost": base.calc_metrics(hedge(book, bm, bep), zero),
"hedged_causal": base.calc_metrics(hedge(book, bm, broll), zero),

注意 zero = pd.Series(0.0, index=bm.index)。

所以calc_metrics(..., zero)是以 0 为基准，报告的是绝对表现：

比如年化收益、年化波动、绝对 Sharpe、最大回撤等。

对冲后 book beta 接近 0，因此不能再按相对基准 IR解释，

代码注释里也明确说IR-vs-benchmark是类别错误。

2.6 逐日执行时序

因果对冲的核心是，今日对冲比率只来自昨日及更早信息。

以 t 日为例，执行时序说明如下

1）t-1 收盘后：

用截至 t-1的过去60日book和bm收益，计算滚动 beta：

2）t日交易前/持有期开始：

设置对冲比率：

3）t 日持有期间：
持有zpos股票组合，同时做空倍benchmark。

4）t日收益实现：

5）t 日收盘后：

更新滚动窗口，计算，供 t+1 日使用。

3 与ex-post的区别联系

3.1 区别

这里进一步对比hedged_expost，示例其作用。

1）hedged_expost

hedged_expost是全样本回归beta，不可以实盘，因为有前视，用于理想上限/诊断。

2）hedged_causal

hedged_causal，bela来源于60 日滚动 beta，并shift(1)，可实盘，实时可估，是实际可执行版。

3）unhedged

unhedged，其beta不对冲，可实盘，是原始zpos 表现。

3.2 联系

全样本 ex-post beta只作为前视诊断上限。

比较hedged_expost和hedged_causal，可以判断出如下重要信息。

1）滚动 beta 是否稳定；

2）实时对冲相对理想对冲损失多少；

3）市场暴露是否被有效抵消。

4 注意事项

4.1 只对zpos做Q2对冲

build_scheme_returns(..., zpos)写死了。

ew_top50 / rank_all / zls只出现在Q1的 raw vs neutral 对比里。

4.2 基准口径可能不一致

Q1 的股票 beta 是compute_market_beta(returns)，即相对等权 universe 的 beta。

Q2 对冲用的是bm = ctx.benchmark。

如果ctx.benchmark不是同一个等权 universe，那么特征中性化 beta和组合对冲beta口径不同。

4.3早期beta缺失被填0

rolling_hedge_beta初期 NaN，hedge中fillna(0.0)，意味着前约 30 天不 hedge。

若严格评估，可以考虑从有 beta 的日期开始统计。

4.4 无对冲交易成本

代码只扣股票book的entry-only 成本，没有扣：

- 做空 benchmark 的借券/期货/融券成本；

- 保证金占用；

- 基差、展期、冲击成本；

- 对冲比率调整带来的换手成本。

所以hedged_causal是理想化实时对冲。

4.5 calc_metrics(..., zero)口径

传入零基准后，得到的不是相对 benchmark 的 IR，而是绝对收益序列的 Sharpe 类指标。

所以，打印阶段进一步补充计算vol_u = book.std()*sqrt(252)

reference

---

多因子Beta对冲工具的管理分析

https://blog.csdn.net/liliang199/article/details/165348356

如何生成日度策略收益序列

https://blog.csdn.net/liliang199/article/details/165328624

滚动CAPM贝塔的计算示例和分析

https://blog.csdn.net/liliang199/article/details/165293799