"""策略生成器：把"偏好参数"翻译成一套完整、可回测、可执行的多因子策略规格。
这是整个工具的核心卖点：用户只说人话级别的偏好（风险偏好 / 持有周期 /
风格倾向 / 持仓数量），生成器负责推导出全部具体参数并给出推导说明。
"""
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