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

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

| 能力维度 | 轮动策略 | 趋势跟踪策略 | 反转策略 |
|----------|----------|--------------|----------|
| **核心逻辑** | 动量排序，选出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 核心抽象接口

```python
# ==================== 因子抽象 ====================
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 轮动策略配置

```yaml
# 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 趋势跟踪策略配置

```yaml
# 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 反转策略配置

```yaml
# 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`
- 新策略类型：只需编写配置文件

**示例：添加新因子**
```python
# 新因子只需继承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*
*适用范围：量化策略通用框架设计*