etf/docs/量化策略通用框架抽象设计.md

量化策略通用框架抽象设计

一、三种策略核心能力对比

能力维度	轮动策略	趋势跟踪策略	反转策略
核心逻辑	动量排序，选出Top N持有	识别趋势方向，顺势交易	识别超买超卖，逆势交易
因子类型	动量因子、加权动量	趋势指标（均线、MACD、通道）	反转指标（RSI、KDJ、布林带）
信号频率	定期调仓（每日/每周）	趋势变化时调仓	反转点入场
持仓周期	中短期（1-30天）	中长期（数周/数月）	短期（数天）
持仓数量	多标的（Top 3-5）	单一或少数标的	单一标的
风险控制	分散化、溢价过滤	止损、趋势跟踪止损	止损、仓位控制
执行方式	T+1执行	信号当日或次日执行	反转点当日执行

---

二、通用能力抽象

2.1 共性能力矩阵

能力层	具体能力	轮动策略	趋势策略	反转策略	通用化程度
数据层	多源数据获取	✓	✓	✓	100%
	交易日历对齐	✓	✓	✓	100%
	缓存管理	✓	✓	✓	100%
因子层	因子计算接口	✓	✓	✓	100%
	因子组合/加权	✓	✓	✓	100%
	因子参数化	✓	✓	✓	100%
信号层	信号生成接口	✓	✓	✓	100%
	信号过滤机制	✓	✓	✓	100%
	信号强度评估	部分	✓	✓	80%
执行层	回测执行器	✓	✓	✓	100%
	模拟盘执行器	✓	✓	✓	100%
	实盘执行器	✓	✓	✓	100%
风控层	止损控制	部分	✓	✓	100%
	仓位控制	✓	✓	✓	100%
	敞口控制	✓	✓	✓	100%
报告层	KPI计算	✓	✓	✓	100%
	可视化报告	✓	✓	✓	100%

2.2 策略差异抽象

差异维度	轮动策略	趋势策略	反转策略	抽象方式
选股逻辑	Top N排序	趋势强度排序	反转信号强度	SelectionMode配置
入场信号	动量得分高	趋势向上突破	反转点确认	EntryRule配置
出场信号	动量排名下降	趋势反转/止损	反转失败/止损	ExitRule配置
持仓数量	3-5只	1-3只	1只	select_num配置
调仓频率	每日/定期	趋势变化时	反转点时	RebalanceFrequency配置

---

三、通用框架设计

3.1 核心架构

┌─────────────────────────────────────────────────────────────┐
│                    QuantStrategyFramework                    │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│  │  DataLayer   │  │  FactorLayer │  │ SignalLayer  │       │
│  │              │  │              │  │              │       │
│  │ - DataSource │  │ - FactorBase │  │ - SignalGen  │       │
│  │ - Router     │  │ - Registry   │  │ - Filter     │       │
│  │ - Cache      │  │ - Combiner   │  │ - Validator  │       │
│  └──────────────┘  └──────────────┘  └──────────────┘       │
│                                                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│  │ ExecutionLayer│ │  RiskLayer   │  │ ReportLayer  │       │
│  │              │  │              │  │              │       │
│  │ - Executor   │  │ - RiskCtrl   │  │ - Generator  │       │
│  │ - Portfolio  │  │ - StopLoss   │  │ - Metrics    │       │
│  │ - Tracker    │  │ - Position   │  │ - Visualizer │       │
│  └──────────────┘  └──────────────┘  └──────────────┘       │
│                                                              │
├─────────────────────────────────────────────────────────────┤
│                     StrategyConfig (YAML)                    │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│  │ RotationStrat│ │ TrendStrat   │  │ ReversalStrat│       │
│  │              │  │              │  │              │       │
│  │ 动量因子     │  │ 趋势因子     │  │ 反转因子     │       │
│  │ Top N选股    │  │ 趋势跟随     │  │ 反转交易     │       │
│  │ 定期调仓     │  │ 趋势止损     │  │ 快速止损     │       │
│  └──────────────┘  └──────────────┘  └──────────────┘       │
│                                                              │
└─────────────────────────────────────────────────────────────┘

3.2 核心抽象接口

# ==================== 因子抽象 ====================
class FactorBase(ABC):
    """因子抽象基类 - 所有策略的因子都继承此接口"""
    
    @abstractmethod
    def compute(self, data: pd.DataFrame) -> pd.Series:
        """计算因子值序列"""
        pass
    
    @property
    @abstractmethod
    def name(self) -> str:
        """因子名称"""
        pass
    
    @property
    def params(self) -> dict:
        """因子参数（可配置）"""
        return {}
    
    def validate(self, data: pd.DataFrame) -> bool:
        """验证数据是否满足计算要求"""
        return len(data) >= self.params.get('min_periods', 20)


class MomentumFactor(FactorBase):
    """动量因子 - 用于轮动策略"""
    
    def __init__(self, n_days: int = 25, weighted: bool = True):
        self._n_days = n_days
        self._weighted = weighted
    
    def compute(self, data):
        if self._weighted:
            # 加权动量得分
            return self._weighted_momentum(data['close'], self._n_days)
        else:
            # 简单动量
            return data['close'].pct_change(self._n_days)
    
    @property
    def name(self):
        return "momentum" if not self._weighted else "weighted_momentum"
    
    @property
    def params(self):
        return {'n_days': self._n_days, 'weighted': self._weighted}


class TrendFactor(FactorBase):
    """趋势因子 - 用于趋势跟踪策略"""
    
    def __init__(self, method: str = 'ma_cross', fast: int = 5, slow: int = 20):
        self._method = method
        self._fast = fast
        self._slow = slow
    
    def compute(self, data):
        if self._method == 'ma_cross':
            fast_ma = data['close'].rolling(self._fast).mean()
            slow_ma = data['close'].rolling(self._slow).mean()
            # 趋势强度 = 快线/慢线偏离度
            return (fast_ma - slow_ma) / slow_ma
        elif self._method == 'macd':
            # MACD趋势强度
            return self._compute_macd(data['close'])
    
    @property
    def name(self):
        return f"trend_{self._method}"
    
    @property
    def params(self):
        return {'method': self._method, 'fast': self._fast, 'slow': self._slow}


class ReversalFactor(FactorBase):
    """反转因子 - 用于反转策略"""
    
    def __init__(self, method: str = 'rsi', period: int = 14):
        self._method = method
        self._period = period
    
    def compute(self, data):
        if self._method == 'rsi':
            # RSI反转信号（RSI偏离度）
            rsi = self._compute_rsi(data['close'], self._period)
            # 超买超卖偏离度
            return np.where(rsi > 70, -(rsi - 70)/30,  # 超买→负值（反转向下）
                           np.where(rsi < 30, (30 - rsi)/30,  # 超卖→正值（反转向上）
                                   0))
        elif self._method == 'kdj':
            return self._compute_kdj(data)
    
    @property
    def name(self):
        return f"reversal_{self._method}"
    
    @property
    def params(self):
        return {'method': self._method, 'period': self._period}


# ==================== 因子注册器 ====================
class FactorRegistry:
    """因子注册器 - 支持动态注册和组合"""
    
    _factors = {}
    
    @classmethod
    def register(cls, factor_cls: FactorBase):
        """注册因子"""
        instance = factor_cls()
        cls._factors[instance.name] = factor_cls
    
    @classmethod
    def get(cls, name: str, **params) -> FactorBase:
        """获取因子实例"""
        factor_cls = cls._factors.get(name)
        if factor_cls:
            return factor_cls(**params)
        raise ValueError(f"Unknown factor: {name}")
    
    @classmethod
    def list(cls) -> List[str]:
        """列出所有注册因子"""
        return list(cls._factors.keys())


class FactorCombiner:
    """因子组合器 - 支持多因子加权"""
    
    def __init__(self, factors: List[FactorBase], weights: List[float] = None):
        self._factors = factors
        self._weights = weights or [1.0] * len(factors)
    
    def compute(self, data: pd.DataFrame) -> pd.Series:
        """计算组合因子值"""
        results = []
        for factor, weight in zip(self._factors, self._weights):
            factor_values = factor.compute(data)
            results.append(factor_values * weight)
        
        # 加权平均
        return pd.concat(results, axis=1).sum(axis=1)


# ==================== 信号生成抽象 ====================
class SignalGenerator(ABC):
    """信号生成器抽象基类"""
    
    @abstractmethod
    def generate(self, factor_values: pd.DataFrame) -> pd.DataFrame:
        """生成交易信号"""
        pass
    
    @property
    @abstractmethod
    def mode(self) -> str:
        """信号生成模式"""
        pass


class TopNSelector(SignalGenerator):
    """Top N选股器 - 用于轮动策略"""
    
    def __init__(self, select_num: int = 3, group_by: str = None):
        self._select_num = select_num
        self._group_by = group_by
    
    def generate(self, factor_values):
        # 按因子值排序
        if self._group_by:
            # 分组选股：每大类选Top1，再全局选Top3
            return self._grouped_selection(factor_values)
        else:
            # 全局Top N
            return self._global_top_n(factor_values)
    
    @property
    def mode(self):
        return "rotation_top_n"


class TrendFollower(SignalGenerator):
    """趋势跟随器 - 用于趋势跟踪策略"""
    
    def __init__(self, entry_threshold: float = 0.02, exit_threshold: float = -0.02):
        self._entry_threshold = entry_threshold
        self._exit_threshold = exit_threshold
    
    def generate(self, factor_values):
        # 趋势强度 > 阈值 → 入场
        # 趋势强度 < 阈值 → 出场
        signals = pd.DataFrame(index=factor_values.index)
        for code in factor_values.columns:
            trend_strength = factor_values[code]
            signals[code + '_entry'] = trend_strength > self._entry_threshold
            signals[code + '_exit'] = trend_strength < self._exit_threshold
        return signals
    
    @property
    def mode(self):
        return "trend_follow"


class ReversalTrader(SignalGenerator):
    """反转交易器 - 用于反转策略"""
    
    def __init__(self, overbought: float = 70, oversold: float = 30):
        self._overbought = overbought
        self._oversold = oversold
    
    def generate(self, factor_values):
        # 超买区域 → 反转向下信号
        # 超卖区域 → 反转向上信号
        signals = pd.DataFrame(index=factor_values.index)
        for code in factor_values.columns:
            reversal_signal = factor_values[code]
            # 正值 → 反转向上（买入）
            signals[code + '_buy'] = reversal_signal > 0
            # 负值 → 反转向下（卖出）
            signals[code + '_sell'] = reversal_signal < 0
        return signals
    
    @property
    def mode(self):
        return "reversal"


# ==================== 策略抽象 ====================
class StrategyBase(ABC):
    """策略抽象基类 - 所有策略的核心接口"""
    
    @abstractmethod
    def run(self, data: pd.DataFrame) -> pd.DataFrame:
        """运行策略"""
        pass
    
    @property
    @abstractmethod
    def name(self) -> str:
        """策略名称"""
        pass
    
    @property
    def config(self) -> dict:
        """策略配置"""
        return {}
    
    def validate_config(self) -> bool:
        """验证配置有效性"""
        return True


class QuantStrategy(StrategyBase):
    """量化策略通用实现 - 配置驱动"""
    
    def __init__(self, config: dict):
        self._config = config
        
        # 初始化因子
        self._factors = self._init_factors()
        
        # 初始化信号生成器
        self._signal_gen = self._init_signal_generator()
        
        # 初始化风控
        self._risk_ctrls = self._init_risk_controls()
    
    def _init_factors(self) -> FactorCombiner:
        """根据配置初始化因子组合"""
        factor_configs = self._config.get('factors', [])
        factors = []
        weights = []
        
        for fc in factor_configs:
            factor = FactorRegistry.get(fc['name'], **fc.get('params', {}))
            factors.append(factor)
            weights.append(fc.get('weight', 1.0))
        
        return FactorCombiner(factors, weights)
    
    def _init_signal_generator(self) -> SignalGenerator:
        """根据配置初始化信号生成器"""
        signal_config = self._config.get('signal', {})
        mode = signal_config.get('mode', 'top_n')
        
        if mode == 'top_n':
            return TopNSelector(
                select_num=signal_config.get('select_num', 3),
                group_by=signal_config.get('group_by')
            )
        elif mode == 'trend_follow':
            return TrendFollower(
                entry_threshold=signal_config.get('entry_threshold', 0.02),
                exit_threshold=signal_config.get('exit_threshold', -0.02)
            )
        elif mode == 'reversal':
            return ReversalTrader(
                overbought=signal_config.get('overbought', 70),
                oversold=signal_config.get('oversold', 30)
            )
        else:
            raise ValueError(f"Unknown signal mode: {mode}")
    
    def _init_risk_controls(self) -> List[RiskControl]:
        """根据配置初始化风控"""
        risk_configs = self._config.get('risk', [])
        controls = []
        
        for rc in risk_configs:
            if rc['type'] == 'stop_loss':
                controls.append(StopLossControl(threshold=rc.get('threshold', 0.05)))
            elif rc['type'] == 'position_limit':
                controls.append(PositionLimitControl(
                    max_position=rc.get('max_position', 0.33)
                ))
        
        return controls
    
    def run(self, data: pd.DataFrame) -> pd.DataFrame:
        """执行策略"""
        # 1. 计算因子
        factor_values = self._factors.compute(data)
        
        # 2. 生成信号
        signals = self._signal_gen.generate(factor_values)
        
        # 3. 应用风控
        for risk_ctrl in self._risk_ctrls:
            signals = risk_ctrl.apply(signals)
        
        return signals
    
    @property
    def name(self):
        return self._config.get('name', 'unknown')
    
    @property
    def config(self):
        return self._config

---

四、三种策略配置化实现

4.1 轮动策略配置

# config/strategies/rotation.yaml

strategy:
  name: "rotation"
  type: "rotation"
  
# 因子配置
factors:
  - name: "weighted_momentum"
    weight: 1.0
    params:
      n_days: 25
      crash_filter: true
      
# 信号配置
signal:
  mode: "top_n"  # Top N选股
  select_num: 3
  group_by: "market"  # 按大类分组
  
# 调仓配置
rebalance:
  frequency: "daily"
  min_holding_days: 1
  cost: 0.001
  
# 风控配置
risk:
  - type: "position_limit"
    max_position: 0.33  # 单品种最大仓位
    
# 执行配置
execution:
  mode: "backtest"
  start_date: "2019-01-01"
  benchmark: "000300.SH"

4.2 趋势跟踪策略配置

# config/strategies/trend_follow.yaml

strategy:
  name: "trend_follow"
  type: "trend"
  
# 因子配置
factors:
  - name: "trend_ma_cross"
    weight: 0.7
    params:
      fast: 5
      slow: 20
      
  - name: "trend_macd"
    weight: 0.3
    params:
      fast: 12
      slow: 26
      
# 信号配置
signal:
  mode: "trend_follow"
  entry_threshold: 0.02  # 趋势强度 > 2% 入场
  exit_threshold: -0.02  # 趋势强度 < -2% 出场
  
# 持仓配置
position:
  max_holdings: 1  # 单标的持仓
  
# 风控配置
risk:
  - type: "stop_loss"
    threshold: 0.05  # 5%止损
    
  - type: "trailing_stop"
    threshold: 0.03  # 3%跟踪止损
    
# 执行配置
execution:
  mode: "backtest"
  start_date: "2020-01-01"

4.3 反转策略配置

# config/strategies/reversal.yaml

strategy:
  name: "reversal"
  type: "reversal"
  
# 因子配置
factors:
  - name: "reversal_rsi"
    weight: 0.6
    params:
      period: 14
      
  - name: "reversal_kdj"
    weight: 0.4
    params:
      period: 9
      
# 信号配置
signal:
  mode: "reversal"
  overbought: 70  # 超买阈值
  oversold: 30    # 超卖阈值
  
# 持仓配置
position:
  max_holdings: 1
  holding_period: 5  # 最大持仓天数
  
# 风控配置
risk:
  - type: "stop_loss"
    threshold: 0.03  # 3%快速止损
    
  - type: "time_stop"
    max_days: 5  # 5天内必须出场
    
# 执行配置
execution:
  mode: "backtest"
  start_date: "2020-01-01"

---

五、通用框架优势

5.1 代码复用率

模块	原实现（各策略独立）	通用框架	复用率提升
数据层	3套独立实现	1套通用实现	67%
因子层	3套因子代码	因子注册+组合	50%
信号层	3套信号逻辑	3种SignalGenerator	33%
执行层	3套回测引擎	1套Executor	67%
风控层	3套风控逻辑	风控组件库	67%
报告层	3套报告代码	1套报告生成	67%

5.2 扩展能力

通用框架支持：

• 新因子类型：只需继承FactorBase并注册
• 新信号逻辑：只需继承SignalGenerator
• 新风控组件：只需继承RiskControl
• 新策略类型：只需编写配置文件

示例：添加新因子

# 新因子只需继承FactorBase
class VolatilityFactor(FactorBase):
    """波动率因子"""
    
    def __init__(self, period: int = 20):
        self._period = period
    
    def compute(self, data):
        return data['close'].rolling(self._period).std()
    
    @property
    def name(self):
        return "volatility"

# 注册因子
FactorRegistry.register(VolatilityFactor)

# 在配置中使用
factors:
  - name: "volatility"
    weight: 0.2
    params:
      period: 20

---

六、实施建议

6.1 分阶段实施

阶段	任务	预估工作量	优先级
阶段一	因子层抽象 + 注册器	1-2天	P1
阶段二	信号生成器抽象	1天	P1
阶段三	风控模块独立	1天	P2
阶段四	策略配置驱动	1天	P2
阶段五	趋势/反转策略实现	2-3天	P3

6.2 立即可实施

现有轮动策略重构路径：

1. 将momentum.py中的因子计算封装为MomentumFactor类
2. 创建FactorRegistry注册动量因子
3. 将选股逻辑封装为TopNSelector
4. 将风控逻辑封装为风险控制组件
5. 通过配置驱动运行策略

---

七、总结

7.1 可抽象程度评估

策略类型	可抽象比例	说明
轮动策略	85%	因子计算、选股、执行、报告均可通用化
趋势跟踪	80%	因子不同，但信号生成、执行、风控可通用
反转策略	75%	因子差异较大，风控需求更强，其他可通用

7.2 核心结论

✅ 三种策略可以抽象出通用框架，核心设计：

1. 因子层：抽象接口 + 注册器 + 组合器
2. 信号层：策略模式（Strategy Pattern）
3. 风控层：组件化风控库
4. 配置驱动：YAML定义策略参数

通用化收益：

• 代码复用率提升50%-67%
• 新策略开发时间缩短50%
• 因子/风控组件可跨策略共享
• 维护成本大幅降低

---

文档版本：V1.0 生成时间：2026-05-08 适用范围：量化策略通用框架设计