""" hedging.py — 对冲决策计算器 ============================== 核心功能： 1. 给定企业月度燃油/原料消耗金额 2. 根据当前风险预测（区间+因子+regime） 3. 计算不同对冲比例下的成本-收益矩阵 4. 输出推荐对冲比例和工具建议 对冲逻辑： - 不对冲：完全暴露在油价波动中 - 部分对冲：锁定一部分成本，保留一部分上行/下行暴露 - 完全对冲：完全锁定成本，放弃上行收益但消除下行风险 - 对冲成本 = 远期升水（contango）+ 期权时间价值（简化为波动率函数） """ import numpy as np from config import INDUSTRIES, INDUSTRY_ZH # ═══════════════════════════════════════════════════════════ # INDUSTRY COST SENSITIVITY (油价弹性系数，基于公开研究) # ═══════════════════════════════════════════════════════════ # 油价变动1%对行业成本/利润的影响（弹性系数，文献值+经验校准） COST_ELASTICITY = { 'Aviation': 0.35, # 航空燃油占运营成本 25-40% 'Logistics': 0.22, # 柴油占物流成本 15-25% 'Chemicals': 0.28, # 原油是石脑油/乙烯原料 'Manufacturing': 0.12, # 能源占制造成本 8-15% 'Upstream_OG': -0.60, # 上游：油价上涨 = 收入增加 } # 典型企业月度油品相关支出（百万美元，用于示例计算） TYPICAL_EXPOSURE = { 'Aviation': 50.0, # 大型航司月均燃油 $50M 'Logistics': 15.0, # 大型物流公司 $15M 'Chemicals': 30.0, # 大型化工企业 $30M 'Manufacturing': 8.0, # 中型制造企业 $8M 'Upstream_OG': 80.0, # 油气公司产量对应营收 } # 对冲成本系数（占名义价值的百分比/月，含远期升水+交易成本） HEDGE_COST_RATES = { 'futures': 0.002, # 期货锁价：~0.2%/月（远期升水+保证金机会成本） 'put': 0.008, # 看跌期权（保护性）：~0.8%/月（时间价值衰减） 'collar': 0.003, # 零成本领：~0.3%/月（放弃部分上行） } # ═══════════════════════════════════════════════════════════ # HEDGING DECISION ENGINE # ═══════════════════════════════════════════════════════════ def compute_hedge_matrix(pred_q10, pred_q50, pred_q90, pred_vol, risk_level, risk_bias, industry, monthly_exposure=None): """ 计算不同对冲比例下的成本-收益矩阵。 Parameters ---------- pred_q10, pred_q50, pred_q90 : float 1M 预测区间（收益率，如 -0.11 表示 -11%） pred_vol : float 预测波动率 risk_level : str 'Low' / 'Medium' / 'High' risk_bias : str 'Upward' / 'Balanced' / 'Downward' industry : str 行业标识 monthly_exposure : float or None 月度油品暴露金额（百万美元），None则使用典型值 Returns ------- dict with keys: 'recommended_ratio': 推荐对冲比例 'recommended_tool': 推荐工具 'rationale': 推荐理由 'matrix': 对冲比例 × 情景 的成本矩阵 """ exposure = monthly_exposure or TYPICAL_EXPOSURE.get(industry, 20.0) elasticity = COST_ELASTICITY.get(industry, 0.20) is_upstream = industry == 'Upstream_OG' # 情景定义 scenarios = { 'downside': pred_q10, # 下行风险（10%分位） 'base': pred_q50, # 基准 'upside': pred_q90, # 上行风险（90%分位） } # 对冲比例选项 hedge_ratios = [0.0, 0.25, 0.50, 0.75, 1.0] # 计算矩阵 matrix = [] for ratio in hedge_ratios: row = {'hedge_ratio': ratio, 'hedge_ratio_pct': f'{ratio*100:.0f}%'} for scen_name, price_change in scenarios.items(): # 未对冲部分的损益 unhedged_impact = exposure * price_change * elasticity * (1 - ratio) # 对冲部分：锁定成本，不受价格影响，但有对冲成本 hedge_cost = exposure * ratio * HEDGE_COST_RATES['futures'] # 总净影响 = 未对冲损益 - 对冲成本 net_impact = unhedged_impact - hedge_cost # 上游油气反向：油价涨=收入增 if is_upstream: net_impact = -net_impact # 对冲是锁定收入 row[f'{scen_name}_impact'] = round(net_impact, 2) # VaR: 最大损失 row['worst_case'] = min(row['downside_impact'], row['upside_impact']) row['best_case'] = max(row['downside_impact'], row['upside_impact']) row['range'] = round(row['best_case'] - row['worst_case'], 2) matrix.append(row) # ── 推荐逻辑 ── recommended_ratio, recommended_tool, rationale = _recommend( risk_level, risk_bias, pred_vol, elasticity, is_upstream, pred_q10, pred_q90 ) # ── 各工具成本比较 ── tool_comparison = [] for tool, rate in HEDGE_COST_RATES.items(): monthly_cost = exposure * recommended_ratio * rate tool_comparison.append({ 'tool': tool, 'tool_zh': {'futures': '期货锁价', 'put': '看跌期权', 'collar': '零成本领'}[tool], 'monthly_cost': round(monthly_cost, 2), 'annualized_cost': round(monthly_cost * 12, 2), 'cost_pct': round(rate * 100, 2), }) return { 'industry': industry, 'industry_zh': INDUSTRY_ZH.get(industry, industry), 'exposure': exposure, 'elasticity': elasticity, 'recommended_ratio': recommended_ratio, 'recommended_ratio_pct': f'{recommended_ratio*100:.0f}%', 'recommended_tool': recommended_tool, 'rationale': rationale, 'matrix': matrix, 'tool_comparison': tool_comparison, } def _recommend(risk_level, risk_bias, pred_vol, elasticity, is_upstream, q10, q90): """推荐对冲比例和工具。""" # 基础比例由风险等级决定 base_ratio = {'Low': 0.25, 'Medium': 0.50, 'High': 0.75}.get(risk_level, 0.50) # 偏置调整 if is_upstream: # 上游：下行=收入减少=需要对冲 if risk_bias == 'Downward': base_ratio += 0.15 elif risk_bias == 'Upward': base_ratio -= 0.10 else: # 下游/成本端：上行=成本增加=需要对冲 if risk_bias == 'Upward': base_ratio += 0.15 elif risk_bias == 'Downward': base_ratio -= 0.10 # 波动率调整 if pred_vol > 0.08: base_ratio += 0.10 # 高波动 → 多对冲 # 弹性调整：暴露越大越应该对冲 if abs(elasticity) > 0.30: base_ratio += 0.05 # 尾部风险调整 tail_risk = abs(q10) if not is_upstream else abs(q90) if tail_risk > 0.15: # 尾部超过15% base_ratio += 0.10 base_ratio = max(0.0, min(1.0, round(base_ratio / 0.05) * 0.05)) # 5%步进 # 工具推荐 if risk_level == 'High' and pred_vol > 0.06: tool = 'collar' reason = f'高风险+高波动环境，零成本领策略平衡保护与成本' elif risk_bias == 'Upward' and not is_upstream: tool = 'futures' reason = f'上行偏置明显，期货锁价直接锁定成本' elif risk_bias == 'Downward' and is_upstream: tool = 'put' reason = f'下行风险突出，看跌期权保留上行收益空间' elif pred_vol < 0.04: tool = 'futures' reason = f'低波动环境，简单期货锁价成本最低' else: tool = 'collar' reason = f'均衡环境下零成本领提供灵活保护' risk_zh = {'Low': '低', 'Medium': '中等', 'High': '高'}[risk_level] bias_zh = {'Upward': '上行', 'Downward': '下行', 'Balanced': '均衡'}[risk_bias] rationale = ( f"当前风险{risk_zh}、偏置{bias_zh}、预测波动率{pred_vol*100:.1f}%。" f"建议对冲{base_ratio*100:.0f}%暴露。{reason}。" ) return base_ratio, tool, rationale def compute_all_industry_hedges(row): """为所有行业计算对冲建议。""" results = {} for ind in INDUSTRIES: results[ind] = compute_hedge_matrix( pred_q10=row.get('pred_q10_1m', -0.10), pred_q50=row.get('pred_q50_1m', 0.0), pred_q90=row.get('pred_q90_1m', 0.10), pred_vol=row.get('pred_vol', 0.05), risk_level=row.get('risk_level', 'Medium'), risk_bias=row.get('risk_bias', 'Balanced'), industry=ind, ) return results # ═══════════════════════════════════════════════════════════ # HEDGING BACKTEST # ═══════════════════════════════════════════════════════════ def backtest_hedging(results_df, lookback=60): """ 回测对冲策略：逐月计算 "按推荐比例对冲" vs "完全不对冲" 的累计成本差异。 Parameters ---------- results_df : DataFrame walk-forward 预测结果（含 risk_level, risk_bias, pred_vol, actual_ret_1m 等） lookback : int 回测月数（默认 60 个月） Returns ------- dict: 各行业的月度时间序列 + 累计节省金额 """ import pandas as pd df = results_df.tail(lookback).copy() backtest = {} for ind in INDUSTRIES: exposure = TYPICAL_EXPOSURE.get(ind, 20.0) elasticity = COST_ELASTICITY.get(ind, 0.20) is_upstream = ind == 'Upstream_OG' tool_rate = HEDGE_COST_RATES['futures'] monthly = [] cum_unhedged = 0.0 cum_hedged = 0.0 for _, row in df.iterrows(): actual_ret = row.get('actual_ret_1m', 0) if np.isnan(actual_ret): continue # Determine recommended hedge ratio for this month rl = row.get('risk_level', 'Medium') rb = row.get('risk_bias', 'Balanced') pv = row.get('pred_vol', 0.05) q10 = row.get('pred_q10_1m', -0.10) q90 = row.get('pred_q90_1m', 0.10) ratio, _, _ = _recommend(rl, rb, pv, elasticity, is_upstream, q10, q90) # Unhedged P&L: full exposure to price change price_impact = actual_ret * elasticity if is_upstream: # Upstream: revenue = price * volume. Price up = good. unhedged_pnl = exposure * actual_ret # Simplified: revenue change hedged_pnl = exposure * actual_ret * (1 - ratio) - exposure * ratio * tool_rate else: # Downstream: cost = price * consumption. Price up = bad. unhedged_pnl = -exposure * actual_ret * elasticity hedged_pnl = -exposure * actual_ret * elasticity * (1 - ratio) - exposure * ratio * tool_rate cum_unhedged += unhedged_pnl cum_hedged += hedged_pnl saving = cum_unhedged - cum_hedged # Positive = hedging saved money monthly.append({ 'date': str(row.get('test_date', '')), 'actual_ret': round(float(actual_ret) * 100, 2), 'hedge_ratio': round(ratio, 2), 'risk_level': rl, 'unhedged_pnl': round(unhedged_pnl, 2), 'hedged_pnl': round(hedged_pnl, 2), 'cum_unhedged': round(cum_unhedged, 2), 'cum_hedged': round(cum_hedged, 2), 'cum_saving': round(saving, 2), }) # Summary stats total_saving = cum_unhedged - cum_hedged # Volatility reduction unhedged_vol = np.std([m['unhedged_pnl'] for m in monthly]) if monthly else 0 hedged_vol = np.std([m['hedged_pnl'] for m in monthly]) if monthly else 0 vol_reduction = 1 - hedged_vol / unhedged_vol if unhedged_vol > 0 else 0 # Max drawdown max_dd_unhedged = min(m['cum_unhedged'] for m in monthly) if monthly else 0 max_dd_hedged = min(m['cum_hedged'] for m in monthly) if monthly else 0 backtest[ind] = { 'industry_zh': INDUSTRY_ZH.get(ind, ind), 'months': len(monthly), 'total_saving': round(total_saving, 2), 'vol_reduction': round(vol_reduction * 100, 1), 'max_dd_unhedged': round(max_dd_unhedged, 2), 'max_dd_hedged': round(max_dd_hedged, 2), 'dd_improvement': round(max_dd_hedged - max_dd_unhedged, 2), 'monthly': monthly, } return backtest