因果对冲是一种实测有效的执行策略这里示例一个理想化的zpos因果对冲的代码实现。包括对冲book来源因果beta如何估计对冲如何执行如何通过逐日执行获得收益。1 因果对冲1.1 因果对冲计算经典的因果对冲执行逻辑流程如下1每个交易日 t用截至 t-1 的 60 日滚动窗口估计zpos book对benchmark的beta2令当日空头比率等于该滞后 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 PL (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 books 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 books realized beta HEDGE_MINP 30 COMPOSITES { skew: [skewness], core3: [skewness, vol_20d, vol_60d], } def compute_market_beta(returns, windowBETA_WINDOW, minpBETA_MINP): 60d trailing CAPM beta of each stock vs the equal-weight universe. r_mkt returns.mean(axis1) var r_mkt.rolling(window, min_periodsminp).var() cov returns.rolling(window, min_periodsminp).cov(r_mkt) return cov.div(var, axis0) def neutralize(z, beta, min_obs12): Per-date OLS: keep the residual of z on beta (feature-level neutrality). resid pd.DataFrame(np.nan, indexz.index, columnsz.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(dtypefloat) ym y[m].to_numpy(dtypefloat) A np.column_stack([np.ones(nm), xm]) coef, *_ np.linalg.lstsq(A, ym, rcondNone) 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.evaluates 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(axis1) numer (w_valid * ret_next.fillna(0.0)).sum(axis1) 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, indexidx) 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 books 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, windowHEDGE_WINDOW, minpHEDGE_MINP): Causal 60d book beta; day t uses data t-1 (shifted). var bm.rolling(window, min_periodsminp).var() cov book.rolling(window, min_periodsminp).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, requiredTrue) ap.add_argument(--start, requiredTrue) ap.add_argument(--end, requiredTrue) ap.add_argument(--instruments, defaultdata/cn_data/instruments/csi300.txt) ap.add_argument(--outdir, default./output_btn) args ap.parse_args() os.makedirs(args.outdir, exist_okTrue) cfg CSI300ScreenConfig( startargs.start, endargs.end, instruments_pathargs.instruments, out_dirargs.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, indexbm.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: fraw{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} fann{mr[annual_return]:.2%} maxDD{mr[max_drawdown]:.1%} f | neu IR{mn[information_ratio]:.2f} fann{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} funhedged vol{vol_u:.1%}) print(f unhedged ann{u[annual_return]:.2%} fsharpe{u[sharpe_ratio]:.2f} maxDD{u[max_drawdown]:.1%}) print(f hedged_ex ann{ex[annual_return]:.2%} fsharpe{ex[sharpe_ratio]:.2f} maxDD{ex[max_drawdown]:.1%}) print(f hedged_caus ann{cx[annual_return]:.2%} fsharpe{cx[sharpe_ratio]:.2f} maxDD{cx[max_drawdown]:.1%}) path os.path.join(args.outdir, fbtn_{args.tag}.json) with open(path, w) as fh: json.dump(out, fh, indent1, ensure_asciiFalse) print(f\nSaved - {path}) if __name__ __main__: main()2 对冲链路梳理这里基于以上示例代码按执行链路进行深入详细的梳理。2.1 因果对冲位置这里因果对冲实验分两条线1Q1特征层中性化neutralize(comp_raw, beta_all)在排序前把composite 对股票beta做横截面OLS取残差。改变的是选股分数。2Q2 组合层对冲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_betadef rolling_hedge_beta(book, bm, window60, minp30):var bm.rolling(window, min_periodsminp).var()cov book.rolling(window, min_periodsminp).cov(bm)return (cov / var).shift(1)rolling_hedge_beta实现逻辑如下1. 对基准 bm计算 60 日滚动方差。2. 对 book和bm计算 60 日滚动协方差。3. 得到滚动 beta4. .shift(1)即t日使用的对冲比率只基于t-1及之前的数据避免用到t日收益防止前视偏差。rolling_hedge_beta其他参数说明如下1BETA_WINDOW 60滚动窗口 60 天。2BETA_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_hedgereturn book - bh * bm_ahedge逐日执行如下计算实现细节如下- 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或core3book_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 betabroll 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, indexbm.index)。所以calc_metrics(..., zero)是以 0 为基准报告的是绝对表现比如年化收益、年化波动、绝对 Sharpe、最大回撤等。对冲后 book beta 接近 0因此不能再按相对基准 IR解释代码注释里也明确说IR-vs-benchmark是类别错误。2.6 逐日执行时序因果对冲的核心是今日对冲比率只来自昨日及更早信息。以 t 日为例执行时序说明如下1t-1 收盘后用截至 t-1的过去60日book和bm收益计算滚动 beta2t日交易前/持有期开始设置对冲比率3t 日持有期间持有zpos股票组合同时做空倍benchmark。4t日收益实现5t 日收盘后更新滚动窗口计算供 t1 日使用。3 与ex-post的区别联系3.1 区别这里进一步对比hedged_expost示例其作用。1hedged_exposthedged_expost是全样本回归beta不可以实盘因为有前视用于理想上限/诊断。2hedged_causalhedged_causalbela来源于60 日滚动 beta并shift(1)可实盘实时可估是实际可执行版。3unhedgedunhedged其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缺失被填0rolling_hedge_beta初期 NaNhedge中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