"""策略生成器：把"偏好参数"翻译成一套完整、可回测、可执行的多因子策略规格。 这是整个工具的核心卖点：用户只说人话级别的偏好（风险偏好 / 持有周期 / 风格倾向 / 持仓数量），生成器负责推导出全部具体参数并给出推导说明。 """ from __future__ import annotations import copy import datetime from typing import Dict, List, Optional # ---------------- 偏好参数定义域 ---------------- RISK_LEVELS = { "保守": dict(target_vol=0.06, max_dd_derisk=0.08, max_positions=20, max_weight=0.08, low_vol_bonus=0.20, cash_buffer=0.15, desc="低波动优先，深度分散，回撤容忍度低"), "稳健": dict(target_vol=0.10, max_dd_derisk=0.12, max_positions=15, max_weight=0.10, low_vol_bonus=0.12, cash_buffer=0.10, desc="波动可控，适度分散"), "平衡": dict(target_vol=0.15, max_dd_derisk=0.18, max_positions=12, max_weight=0.12, low_vol_bonus=0.05, cash_buffer=0.05, desc="收益风险均衡"), "进取": dict(target_vol=0.20, max_dd_derisk=0.25, max_positions=10, max_weight=0.15, low_vol_bonus=0.00, cash_buffer=0.02, desc="波动换收益，适度集中"), "激进": dict(target_vol=0.28, max_dd_derisk=0.35, max_positions=8, max_weight=0.20, low_vol_bonus=-0.05, cash_buffer=0.00, desc="高波动容忍，集中持仓"), } HORIZON_LEVELS = { "短线": dict(rebalance_days=5, momentum_lookback=20, trend_lookback=20, reversal_weight=0.20, fwd_days=5, desc="周度调仓，短周期动量+反转"), "波段": dict(rebalance_days=10, momentum_lookback=60, trend_lookback=60, reversal_weight=0.10, fwd_days=10, desc="双周调仓，中短周期"), "中线": dict(rebalance_days=20, momentum_lookback=120, trend_lookback=120, reversal_weight=0.00, fwd_days=20, desc="月度调仓，经典中期动量"), "长线": dict(rebalance_days=60, momentum_lookback=180, trend_lookback=200, reversal_weight=0.00, fwd_days=60, desc="季度调仓，长周期趋势"), "半年": dict(rebalance_days=120, momentum_lookback=250, trend_lookback=250, reversal_weight=0.00, fwd_days=120, desc="半年调仓，年度级别趋势"), } # 默认风格权重（会在风险偏好与周期推导后做调整再归一化） DEFAULT_STYLE = { "momentum": 0.30, "quality_momentum": 0.15, "low_vol": 0.20, "trend": 0.20, "reversal": 0.05, "high_52w": 0.05, "liquidity": 0.05, } FACTOR_LOOKBACKS = { "momentum": None, # 跟随周期推导 "quality_momentum": 120, "reversal": 5, "low_vol": 60, "trend": None, # 跟随周期推导 "high_52w": 250, "liquidity": 20, } ALLOWED_FACTORS = set(DEFAULT_STYLE) def normalize_style(style: Dict[str, float]) -> Dict[str, float]: """校验并归一化风格权重。""" unknown = set(style) - ALLOWED_FACTORS if unknown: raise ValueError(f"未知因子: {unknown}，可选: {sorted(ALLOWED_FACTORS)}") vals = {k: float(v) for k, v in style.items() if float(v) > 0} total = sum(vals.values()) if total <= 0: raise ValueError("风格权重总和必须为正") return {k: v / total for k, v in vals.items()} def generate_strategy( risk: str = "平衡", horizon: str = "波段", style: Optional[Dict[str, float]] = None, max_positions: Optional[int] = None, rebalance_days: Optional[int] = None, cash_buffer: Optional[float] = None, benchmark: str = "000300", universe: str = "sample", start: str = "2022-01-01", end: str = "2030-01-01", execution: Optional[Dict[str, float]] = None, extra_risk: Optional[Dict] = None, name: Optional[str] = None, ) -> Dict: """由偏好参数生成完整策略规格（dict，可 json 序列化）。 Args: risk: 保守/稳健/平衡/进取/激进 horizon: 短线/波段/中线/长线 style: 因子权重覆盖（如 {"momentum":0.4,"low_vol":0.3,"trend":0.3}） max_positions / rebalance_days / cash_buffer: 手工覆盖项 execution: 交易成本 {"commission":双边佣金率单边, "stamp_tax_sell":卖出印花税, "slippage":单边滑点} extra_risk: 风控覆盖 {"trend_filter":bool, "de_risk_ratio":0.5, "min_history":120, "top_select_from":None} """ if risk not in RISK_LEVELS: raise ValueError(f"风险偏好必须为 {list(RISK_LEVELS)}") if horizon not in HORIZON_LEVELS: raise ValueError(f"持有周期必须为 {list(HORIZON_LEVELS)}") r = copy.deepcopy(RISK_LEVELS[risk]) h = copy.deepcopy(HORIZON_LEVELS[horizon]) # 风格权重：默认 -> 风险偏好微调（越保守越偏低波，越激进越带动量）-> 用户覆盖 style_w = copy.deepcopy(DEFAULT_STYLE) style_w["low_vol"] = max(0.0, style_w["low_vol"] + r.pop("low_vol_bonus")) style_w["reversal"] = h["reversal_weight"] if style: style_w.update(style) style_w = normalize_style(style_w) lookbacks = {} for f, lb in FACTOR_LOOKBACKS.items(): if f == "momentum": lookbacks[f] = h["momentum_lookback"] elif f == "trend": lookbacks[f] = h["trend_lookback"] else: lookbacks[f] = lb strat = { "name": name or f"{risk}{horizon}多因子", "generated_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), "preferences": { "risk": risk, "risk_desc": r["desc"], "horizon": horizon, "horizon_desc": h["desc"], "style_request": style or "(默认风格)", }, "universe": {"name": universe, "benchmark": benchmark}, "backtest_window": {"start": start, "end": end}, "selection": { "factors": style_w, "factor_lookbacks": lookbacks, "max_positions": max_positions or r["max_positions"], "min_history": 120, }, "portfolio": { "rebalance_days": rebalance_days or h["rebalance_days"], "cash_buffer": cash_buffer if cash_buffer is not None else r["cash_buffer"], "max_weight": r["max_weight"], "fwd_days": h["fwd_days"], }, "risk_control": { "target_vol": r["target_vol"], "vol_lookback": 60, "max_drawdown_derisk": r["max_dd_derisk"], "de_risk_ratio": (extra_risk or {}).get("de_risk_ratio", 0.5), "trend_filter": (extra_risk or {}).get("trend_filter", True), "trend_filter_ma": 200, }, "execution": { "commission": (execution or {}).get("commission", 0.0003), "stamp_tax_sell": (execution or {}).get("stamp_tax_sell", 0.0005), "slippage": (execution or {}).get("slippage", 0.001), }, } return strat def explain(strategy: Dict) -> List[str]: """生成人类可读的推导说明。""" p, sel, port, rc = strategy["preferences"], strategy["selection"], strategy["portfolio"], strategy["risk_control"] lines = [ f"策略【{strategy['name']}】生成于 {strategy['generated_at']}", f"1. 风险偏好「{p['risk']}」：{p['risk_desc']}", f" → 目标波动率 {rc['target_vol']:.0%}，组合回撤超过 {rc['max_drawdown_derisk']:.0%} 自动降仓 " f"{rc['de_risk_ratio']:.0%}；单票上限 {port['max_weight']:.0%}；持仓 {sel['max_positions']} 只；现金缓冲 {port['cash_buffer']:.0%}", f"2. 持有周期「{p['horizon']}」：{p['horizon_desc']}", f" → 每 {port['rebalance_days']} 个交易日调仓；动量回看 {sel['factor_lookbacks']['momentum']} 日；" f"趋势回看 {sel['factor_lookbacks']['trend']} 日", "3. 因子权重（截面 Z-score 加权合成）:", " " + "，".join(f"{k} {v:.0%}" for k, v in sorted(sel["factors"].items(), key=lambda x: -x[1])), ] if rc["trend_filter"]: lines.append(f"4. 风控过滤：价格低于 {rc['trend_filter_ma']} 日均线的标的禁止入选（熊市保护）") ex = strategy["execution"] lines.append( f"5. 交易成本假设：佣金 {ex['commission']:.2%}/单边 + 卖出印花税 {ex['stamp_tax_sell']:.2%} + 滑点 {ex['slippage']:.2%}/单边" ) return lines