aszerW
c54ba442ad
包含4份核心文档： - 轮动策略系统架构分析报告 - 量化策略通用框架抽象设计 - Freqtrade架构调研与对比分析 - ETF轮动策略通用化重构方案 调研结论：三种策略可抽象通用框架 设计决策：因子注册器风格 + 5个核心回调钩子 + TopN/Trend/Reversal信号生成器
682 lines
22 KiB
Markdown
682 lines
22 KiB
Markdown
# 量化策略通用框架抽象设计
|
||
|
||
## 一、三种策略核心能力对比
|
||
|
||
| 能力维度 | 轮动策略 | 趋势跟踪策略 | 反转策略 |
|
||
|----------|----------|--------------|----------|
|
||
| **核心逻辑** | 动量排序，选出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*
|
||
*适用范围：量化策略通用框架设计* |