ETF策略的四大模板：资产配置、多因子轮动，择时以及策略组合，附python代码

from dataclasses import dataclass, fieldfrom datetime import datetimefrom typing import List, Dict, Optional, Unionfrom enum import Enum

class StrategyType(Enum):    ASSET_ALLOCATION = "asset_allocation"    ROTATION = "rotation"    TIMING = "timing"    PORTFOLIO = "portfolio"

class WeightMethod(Enum):    EQUALLY = "WeightEqually"    FIXED = "WeightFixed"    RISK_PARITY = "WeightRiskParity"    VOLATILITY = "WeightVolatility"

class PeriodType(Enum):    DAILY = "RunDaily"    WEEKLY = "RunWeekly"    MONTHLY = "RunMonthly"    QUARTERLY = "RunQuarterly"    YEARLY = "RunYearly"    Once = 'RunOnce'    CUSTOM_DAYS = "custom_days"

class StopLossType(Enum):


    
    FIXED = "fixed"  # 固定止损    TRAILING = "trailing"  # 移动止损    TIME_BASED = "time_based"  # 时间止损    TECHNICAL = "technical"  # 技术指标止损    VOLATILITY = "volatility"  # 波动率止损    COMBINED = "combined"  # 组合止损

@dataclassclass FixedStopLossConfig:    """固定止损配置"""    loss_threshold: float  # 亏损阈值 X%，例如 0.1 表示亏损10%    sell_ratio: float  # 卖出仓位比例 Y%，例如 0.5 表示卖出50%仓位    enabled: bool = True
    def __post_init__(self):        if not 0 self.loss_threshold <= 1:            raise ValueError("loss_threshold 必须在 (0, 1] 范围内")        if not 0 self.sell_ratio <= 1:            raise ValueError("sell_ratio 必须在 (0, 1] 范围内")

@dataclassclass TrailingStopLossConfig:    """移动止损配置"""    profit_threshold: float  # 盈利阈值 X%，例如 0.2 表示盈利20%    pullback_threshold: float  # 回落阈值 Y%，例如 0.05 表示回落5%    sell_ratio: float  # 卖出仓位比例 Z%，例如 0.3 表示卖出30%仓位    enabled: bool = True
    def __post_init__(self):        if not 0 self.profit_threshold <= 1:            raise ValueError("profit_threshold 必须在 (0, 1] 范围内")        if not 0 self.pullback_threshold <= 1:            raise ValueError("pullback_threshold 必须在 (0, 1] 范围内")        if not 0 self.sell_ratio <= 1:            raise ValueError("sell_ratio 必须在 (0, 1] 范围内")

@dataclassclass TimeBasedStopLossConfig:    """时间止损配置"""    max_hold_days: int  # 最大持有天数    sell_ratio: float = 1.0  # 卖出比例    enabled: bool = True
    def __post_init__(self):        if self.max_hold_days <= 0:            raise ValueError("max_hold_days 必须大于0")        if not 0 self.sell_ratio <= 1:            raise ValueError("sell_ratio 必须在 (0, 1] 范围内")

@dataclassclass TechnicalStopLossConfig:    """技术指标止损配置"""    indicator: str  # 技术指标，如 'ma', 'boll', 'macd'    condition: str  # 条件，如 'close < ma20', 'close < boll_lower'    sell_ratio: float = 1.0  # 卖出比例    enabled: bool = True
    def __post_init__(self):        if not 0 self.sell_ratio <= 1:            raise ValueError("sell_ratio 必须在 (0, 1] 范围内")

@dataclassclass 


    
VolatilityStopLossConfig:    """波动率止损配置"""    atr_multiplier: float = 2.0  # ATR乘数    lookback_period: int = 14  # 回看周期    sell_ratio: float = 1.0  # 卖出比例    enabled: bool = True
    def __post_init__(self):        if self.atr_multiplier <= 0:            raise ValueError("atr_multiplier 必须大于0")        if self.lookback_period <= 0:            raise ValueError("lookback_period 必须大于0")        if not 0 self.sell_ratio <= 1:            raise ValueError("sell_ratio 必须在 (0, 1] 范围内")

@dataclassclass StopLossConfig:    """止损配置（主要用于择时策略）"""    stop_loss_type: StopLossType = StopLossType.COMBINED
    # 各种止损方式配置    fixed_stop_losses: List[FixedStopLossConfig] = field(default_factory=list)    trailing_stop_losses: List[TrailingStopLossConfig] = field(default_factory=list)    time_based_stop_losses: List[TimeBasedStopLossConfig] = field(default_factory=list)    technical_stop_losses: List[TechnicalStopLossConfig] = field(default_factory=list)    volatility_stop_losses: List[VolatilityStopLossConfig] = field(default_factory=list)
    # 止损执行优先级（数字越小优先级越高）    execution_priority: List[StopLossType] = field(default_factory=lambda: [        StopLossType.FIXED,        StopLossType.TRAILING,        StopLossType.TECHNICAL,        StopLossType.VOLATILITY,        StopLossType.TIME_BASED    ])

@dataclassclass BaseTask:    """基础策略配置"""    strategy_type: StrategyType    symbols: List[str] = field(default_factory=list)
    # 回测参数    # start_date: str = '20100101'    # end_date: str = field(default_factory=lambda: datetime.now().strftime('%Y%m%d'))    # benchmark: str = '510300.SH'    # initial_capital: float = 1000000.0
    # 执行周期    period: PeriodType = PeriodType.DAILY    period_days: Optional[int] = None
    # 风险控制    max_position_ratio: float = 1.0  # 最大持仓比例

@dataclassclass AssetAllocationTask(BaseTask):    """资产配置策略"""    strategy_type: StrategyType = StrategyType.ASSET_ALLOCATION
    # 权重配置    weight_method: WeightMethod = WeightMethod.EQUALLY    weight_fixed: Dict[str, float] = field(default_factory=dict)    risk_parity_target: Optional[float] = None  # 风险平价目标波动率
    # 再平衡配置    rebalance_threshold: float = 0.05  # 再平衡阈值    dynamic_rebalance: bool = True  # 是否动态再平衡

@dataclassclass RotationTask(BaseTask):    """轮动策略"""    strategy_type: StrategyType = StrategyType.ROTATION
    # 筛选规则    select_buy_rules: List[str] = field(default_factory=list)    select_sell_rules: List[str


    
] = field(default_factory=list)
    # 数量控制    min_hold_count: int = 1    max_hold_count: int = 10    buy_at_least_count: int = 0    sell_at_least_count: int = 1
    # 排序配置    order_by_signal: str = ''    order_by_topK: int = 5    order_by_dropN: int = 0    order_by_desc: bool = True
    # 轮动参数    rotation_threshold: float = 0.02  # 轮动阈值    hold_period: Optional[int] = None  # 最小持有期

@dataclassclass TimingTask(BaseTask):    """择时策略"""    strategy_type: StrategyType = StrategyType.TIMING
    # 信号规则    buy_rules: List[str] = field(default_factory=list)    sell_rules: List[str] = field(default_factory=list)
    # 仓位管理    position_size: float = 1.0  # 每次开仓比例    pyramiding: int = 1  # 加仓次数
    # 止盈止损配置（择时策略特有）    stop_loss_config: StopLossConfig = field(default_factory=StopLossConfig)
    # 技术指标参数    use_technical: bool = False    technical_params: Dict[str, float] = field(default_factory=dict)

@dataclassclass PortfolioTask(BaseTask):    """策略组合"""    strategy_type: StrategyType = StrategyType.PORTFOLIO
    # 子策略配置    sub_strategies: List[Union[AssetAllocationTask, RotationTask, TimingTask]] = field(default_factory=list)
    # 资金分配    weight_method: WeightMethod = WeightMethod.EQUALLY    strategy_weights: Dict[str, float] = field(default_factory=dict)  # 策略名称:权重
    # 组合风控    correlation_threshold: float = 0.8  # 策略相关性阈值    max_drawdown_control: bool = True

# 使用示例if __name__ == "__main__":    # 资产配置策略示例（不需要止盈止损）    asset_task = AssetAllocationTask(        symbols=['510300.SH', '510500.SH', '159915.SZ'],        weight_method=WeightMethod.FIXED,        weight_fixed={'510300.SH': 0.5, '510500.SH': 0.3, '159915.SZ': 0.2},        period=PeriodType.MONTHLY,        rebalance_threshold=0.1    )
    # 轮动策略示例（通常也不需要复杂的止盈止损）    rotation_task = RotationTask(        symbols=['512100.SH', '512000.SH', '512700.SH', '515000.SH'],        select_buy_rules=['pe_ratio < 20', 'pb_ratio < 2'],        order_by_signal='momentum_score',        order_by_topK=2,        period=PeriodType.WEEKLY    )
    # 择时策略示例（需要丰富的止盈止损配置）    timing_task = TimingTask(        symbols=['510300.SH'],        buy_rules=['close > ma20', 'rsi < 30'],        sell_rules=['close < ma20', 'rsi > 70'],


    
        position_size=0.8,        stop_loss_config=StopLossConfig(            fixed_stop_losses=[                FixedStopLossConfig(loss_threshold=0.05, sell_ratio=0.3),  # 亏损5%卖出30%                FixedStopLossConfig(loss_threshold=0.08, sell_ratio=0.5),  # 亏损8%卖出50%                FixedStopLossConfig(loss_threshold=0.12, sell_ratio=1.0)  # 亏损12%全部卖出            ],            trailing_stop_losses=[                TrailingStopLossConfig(profit_threshold=0.15, pullback_threshold=0.03, sell_ratio=0.2),                # 盈利15%后回落3%卖出20%                TrailingStopLossConfig(profit_threshold=0.25, pullback_threshold=0.05, sell_ratio=0.5),                # 盈利25%后回落5%卖出50%                TrailingStopLossConfig(profit_threshold=0.4, pullback_threshold=0.08, sell_ratio=1.0)  # 盈利40%后回落8%全部卖出            ],            time_based_stop_losses=[                TimeBasedStopLossConfig(max_hold_days=30, sell_ratio=0.5),  # 持有30天卖出50%                TimeBasedStopLossConfig(max_hold_days=60, sell_ratio=1.0)  # 持有60天全部卖出            ],            technical_stop_losses=[                TechnicalStopLossConfig(                    indicator='ma',                    condition='close < ma20',                    sell_ratio=0.5                ),                TechnicalStopLossConfig(                    indicator='boll',                    condition='close < boll_lower',                    sell_ratio=0.3                )            ],            volatility_stop_losses=[                VolatilityStopLossConfig(                    atr_multiplier=2.0,                    lookback_period=14,                    sell_ratio=0.5                )            ]        )    )
    # 策略组合示例    portfolio_task = PortfolioTask(        sub_strategies=[asset_task, rotation_task, timing_task],        weight_method=WeightMethod.FIXED,        strategy_weights={'asset_allocation': 0.4, 'rotation': 0.4, 'timing': 0.2}    )