aszer/etf
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs: 添加策略框架调研与设计方案

Browse Source 
包含4份核心文档：
- 轮动策略系统架构分析报告
- 量化策略通用框架抽象设计
- Freqtrade架构调研与对比分析
- ETF轮动策略通用化重构方案

调研结论：三种策略可抽象通用框架
设计决策：因子注册器风格 + 5个核心回调钩子 + TopN/Trend/Reversal信号生成器
This commit is contained in:
aszerW
2026-05-11 22:17:41 +08:00
parent baeeb13c34
commit c54ba442ad
4 changed files with 2243 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

387
docs/ETF轮动策略通用化重构方案.md Normal file
View File

@@ -0,0 +1,387 @@
# ETF轮动策略通用化重构方案
## 一、重构目标与范围
### 1.1 核心目标
**将现有轮动策略重构为通用量化框架**，支持：
- ✅ 轮动策略（动量选股 + Top N）
- ✅ 趋势跟踪策略（趋势因子 + 信号跟随）
- ✅ 反转策略（反转因子 + 超买超卖）
- ✅ 自定义策略扩展
### 1.2 重构范围
| 模块 | 是否重构 | 重构内容 | 保持不变 |
|------|----------|----------|----------|
| **数据层** | ✅ 部分重构 | 添加DataSource抽象接口 | 混合数据源逻辑保留 |
| **因子层** | ✅ 重构 | FactorBase抽象 + 注册器 + 组合器 | 因子计算逻辑保留 |
| **信号层** | ✅ 重构 | SignalGenerator抽象（TopN/Trend/Reversal） | 选股逻辑迁移 |
| **风控层** | ✅ 新增 | 回调钩子 + 风控组件库 | 无（原为分散化） |
| **执行层** | ⚠️ 保持 | 回测引擎不变 | 回测逻辑完整保留 |
| **报告层** | ⚠️ 保持 | KPI计算不变 | 可视化保留 |
| **配置层** | ✅ 重构 | YAML配置驱动 | 参数结构保留 |
---
## 二、核心设计决策（需确认）
### 决策点1：因子抽象方式
**选项A：Freqtrade风格（方法内定义）**
```python
def populate_indicators(self, dataframe):
dataframe['momentum'] = calculate_momentum(dataframe, 25)
dataframe['rsi'] = ta.RSI(dataframe, 14)
return dataframe
```
- 优点：简单直观，与Freqtrade一致
- 缺点：因子不可复用，难以组合
**选项B：注册器风格（模块化）**
```python
class MomentumFactor(FactorBase):
def compute(self, data):
return calculate_momentum(data, self.n_days)
FactorRegistry.register(MomentumFactor)
# 在策略中使用
factors = FactorCombiner([
FactorRegistry.get('momentum', n_days=25),
FactorRegistry.get('volatility', period=20)
])
```
- 优点：因子可复用、可组合、可插拔
- 缺点：复杂度略高
**建议**：**选择B（注册器风格）**，保持我们的模块化优势。
---
### 决策点2：回调钩子范围
**参考Freqtrade的回调设计**：
| 回调钩子 | 用途 | 是否需要 |
|----------|------|----------|
| `before_entry` | 入场前检查（溢价、崩盘） | ✅ 需要 |
| `after_entry` | 入场后记录 | ⚠️ 可选 |
| `before_exit` | 出场前检查 | ⚠️ 可选 |
| `after_exit` | 出场后记录 | ⚠️ 可选 |
| `dynamic_stoploss` | 动态止损 | ✅ 需要 |
| `custom_exit` | 自定义出场条件 | ✅ 需要 |
| `position_adjust` | DCA/部分止盈 | ❌ 不需要（轮动不适用） |
**建议**：**实现5个核心回调**（before_entry, dynamic_stoploss, custom_exit, after_entry, after_exit）。
---
### 册策点3：参数装饰器设计
**Freqtrade风格**：
```python
class RotationStrategy:
n_days = IntParameter(10, 60, default=25, space='factor')
select_num = IntParameter(1, 5, default=3, space='signal')
```
**简化风格（无Hyperopt）**：
```python
class RotationStrategy:
n_days = 25 # 类属性，可被配置覆盖
select_num = 3
```
**建议**：**先实现简化版（类属性 + 配置覆盖）**，后续再添加参数装饰器 + Hyperopt。
---
### 决策点4：信号生成抽象
**选项A：统一SignalGenerator接口**
```python
class SignalGenerator(ABC):
def generate(self, factor_data: DataFrame) -> DataFrame
class TopNSelector(SignalGenerator):
def generate(self, factor_data):
# Top N选股逻辑
class TrendFollower(SignalGenerator):
def generate(self, factor_data):
# 趋势跟随逻辑
```
**选项B：策略直接生成信号**
```python
class RotationStrategy:
def generate_signals(self, data):
# 直接在策略内生成
```
**建议**：**选择A（统一SignalGenerator）**，信号逻辑可复用、可插拔。
---
### 决策点5：执行模式支持
**Freqtrade支持**：backtest / hyperopt / dry / live
**当前系统支持**：backtest only
**需要添加**：
- ✅ dry-run（模拟盘）：验证策略逻辑
- ⚠️ live（实盘）：需券商API对接
- ❌ hyperopt：后期添加
**建议**：**仅添加dry-run模式**，实盘执行器延后。
---
## 三、分阶段实施路径
### 阶段一：因子层抽象（P1，预估2天）
**目标**：因子可插拔、可组合
**实施内容**：
1. 创建`FactorBase`抽象基类
2. 创建`FactorRegistry`注册器
3. 创建`FactorCombiner`组合器
4. 重构现有动量因子为`MomentumFactor`
5. 添加技术因子（`TrendFactor`, `ReversalFactor`）
**验收标准**：
- 因子可注册、可获取、可组合
- 现有轮动策略功能不变（回归测试）
---
### 阶段二：信号层抽象（P1，预估1天）
**目标**：信号生成可插拔
**实施内容**：
1. 创建`SignalGenerator`抽象基类
2. 创建`TopNSelector`实现（迁移现有选股逻辑）
3. 创建`TrendFollower`实现
4. 创建`ReversalTrader`实现
**验收标准**：
- 三种信号生成器可独立运行
- TopNSelector与原逻辑一致
---
### 阶段三：回调钩子机制（P1，预估1天）
**目标**：生命周期可控
**实施内容**：
1. 在`StrategyBase`中添加回调钩子定义
2. 实现`before_entry`回调（溢价过滤）
3. 实现`dynamic_stoploss`回调
4. 实现`custom_exit`回调
**验收标准**：
- 回调可正常触发
- 溢价过滤生效
- 动态止损生效
---
### 阶段四：配置驱动重构（P2，预估1天）
**目标**：策略参数可配置
**实施内容**：
1. YAML配置结构设计
2. 配置加载与验证
3. 配置覆盖类属性
4. 策略工厂（根据配置创建策略）
**验收标准**：
- 可通过YAML配置运行策略
- 配置可覆盖策略默认参数
---
### 阶段五：趋势/反转策略实现（P3，预估2-3天）
**目标**：验证框架扩展性
**实施内容**：
1. 实现`TrendFollowStrategy`
2. 实现`ReversalStrategy`
3. 回测验证
**验收标准**：
- 新策略可正常运行
- 回测报告正常生成
---
## 四、预期收益与风险
### 4.1 预期收益
| 维度 | 当前状态 | 重构后 | 提升 |
|------|----------|--------|------|
| **代码复用** | 因子/信号耦合 | 模块化复用 | 50%+ |
| **策略扩展** | 仅轮动策略 | 支持3+策略类型 | - |
| **参数调优** | 手动修改代码 | 配置驱动 | 效率提升 |
| **风控灵活性** | 固定分散化 | 回调动态控制 | - |
| **维护成本** | 多处修改 | 单点修改 | 降低50% |
### 4.2 风险评估
| 风险 | 等级 | 应对措施 |
|------|------|----------|
| **回归风险**：重构导致原有逻辑变化 | 🔴高 | 每阶段完成后运行回测验证 |
| **性能风险**：抽象层增加性能损耗 | 🟡中 | 保持向量化计算，避免循环 |
| **复杂度风险**：抽象层增加理解成本 | 🟡中 | 完善文档 + 示例代码 |
| **时间风险**：预估时间不准 | 🟢低 | 分阶段验收，及时调整 |
---
## 五、关键架构图
### 5.1 最终架构
```
┌─────────────────────────────────────────────────────────────┐
│ QuantStrategyFramework │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ DataSource │ │ FactorLayer │ │ SignalLayer │ │
│ │ │ │ │ │ │ │
│ │ - HybridSrc │ │ - FactorBase │ │ - SignalGen │ │
│ │ - Tushare │ │ - Registry │ │ - TopN │ │
│ │ - YFinance │ │ - Combiner │ │ - Trend │ │
│ │ │ │ │ │ - Reversal │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ StrategyBase │ │ RiskLayer │ │ ExecLayer │ │
│ │ │ │ │ │ │ │
│ │ - populate_* │ │ - Callbacks │ │ - Backtest │ │
│ │ - 回调钩子 │ │ - StopLoss │ │ - DryRun │ │
│ │ │ │ - Position │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
├─────────────────────────────────────────────────────────────┤
│ Config (YAML驱动) │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │RotationStrat │ │ TrendStrat │ │ ReversalStrat│ │
│ │ │ │ │ │ │ │
│ │ 动量因子 │ │ 趋势因子 │ │ 反转因子 │ │
│ │ TopN选股 │ │ 趋势跟随 │ │ 反转交易 │ │
│ │ 回调过滤 │ │ 动态止损 │ │ 快速止损 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
```
### 5.2 因子层设计
```python
# 因子基类
class FactorBase(ABC):
@abstractmethod
def compute(self, data: DataFrame) -> Series
@property
@abstractmethod
def name(self) -> str
# 因子注册
FactorRegistry.register(MomentumFactor)
FactorRegistry.register(TrendFactor)
# 因子组合
factors = FactorCombiner([
FactorRegistry.get('momentum', n_days=25, weight=0.7),
FactorRegistry.get('volatility', period=20, weight=0.3)
])
```
### 5.3 信号层设计
```python
# 信号生成器
class TopNSelector(SignalGenerator):
def generate(self, factor_data):
# 按因子值排序，选Top N
class TrendFollower(SignalGenerator):
def generate(self, factor_data):
# 趋势强度 > 阈值 → 入场
class ReversalTrader(SignalGenerator):
def generate(self, factor_data):
# 超买超卖 → 反转信号
```
### 5.4 回调钩子设计
```python
class StrategyBase:
# 入场前回调
def before_entry(self, code: str, price: float, **kwargs) -> bool:
"""可拒绝交易"""
if kwargs.get('premium', 0) > 0.10:
return False
return True
# 动态止损回调
def dynamic_stoploss(self, position: Position) -> float:
"""根据持仓时间调整止损"""
if position.holding_days > 10:
return -0.03
return -0.10
# 自定义出场回调
def custom_exit(self, position: Position) -> bool:
"""反转信号强制出场"""
return False
```
---
## 六、需确认事项清单
| 序号 | 决策点 | 选项 | 建议 | 你的选择 |
|------|--------|------|------|----------|
| 1 | 因子抽象方式 | A:方法内 / B:注册器 | **B** | ? |
| 2 | 回调钩子范围 | 5个 / 7个 / 全部 | **5个核心** | ? |
| 3 | 参数装饰器 | 简化版 / Freqtrade风格 | **先简化版** | ? |
| 4 | 信号生成抽象 | 统一接口 / 策略内生成 | **统一接口** | ? |
| 5 | 执行模式支持 | 仅backtest / +dry-run / +live | **+dry-run** | ? |
| 6 | 实施优先级 | 全部实施 / 分阶段 | **分阶段** | ? |
| 7 | 时间安排 | 立即开始 / 等待确认 | **等待确认** | ? |
---
## 七、下一步行动
**确认方案后，我将开始**：
1. 创建`FactorBase`抽象基类
2. 创建`FactorRegistry`注册器
3. 重构现有动量因子
4. 运行回测验证功能不变
**请确认以下内容**：
- ✅ 重构范围是否合理
- ✅ 设计决策是否认可
- ✅ 实施路径是否可行
- ✅ 时间安排是否同意
---
*方案版本：V1.0*
*生成时间：2026-05-08*
*等待用户确认*

639
docs/Freqtrade架构调研与对比分析.md Normal file
View File

@@ -0,0 +1,639 @@
# Freqtrade架构调研与对比分析报告
## 一、Freqtrade架构概览
### 1.1 项目定位
Freqtrade是一个开源的加密货币量化交易框架，核心特点：
- **成熟度高**：2017年开源，持续维护，社区活跃
- **生产就绪**：支持回测、模拟盘、实盘三种模式
- **配置驱动**：策略、参数、风控均可通过配置控制
- **扩展性强**：支持自定义因子、策略、损失函数、交易所
### 1.2 核心架构图
```
┌─────────────────────────────────────────────────────────────────────┐
│ FreqtradeBot (主控制器) │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Strategy │ │ Exchange │ │ DataProvider │ │
│ │ (IStrategy)│ │ (CCXT) │ │ │ │
│ │ │ │ │ │ │ │
│ │ - 指标计算 │ │ - OHLCV获取 │ │ - 数据缓存 │ │
│ │ - 入场信号 │ │ - 交易执行 │ │ - 实时推送 │ │
│ │ - 出场信号 │ │ - 余额查询 │ │ │ │
│ │ - 回调函数 │ │ │ │ │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Portfolio │ │ RPC/Notify│ │ Hyperopt │ │
│ │ Manager │ │ │ │ (Optuna) │ │
│ │ │ │ - Telegram │ │ │ │
│ │ - 持仓跟踪 │ │ - Discord │ │ - 参数优化 │ │
│ │ - 余额管理 │ │ - WebUI │ │ - 损失函数 │ │
│ │ │ │ │ │ │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
├─────────────────────────────────────────────────────────────────────┤
│ Configuration (JSON/YAML) │
└─────────────────────────────────────────────────────────────────────┘
```
---
## 二、核心模块设计
### 2.1 策略抽象层（IStrategy）
**设计理念**：策略是核心，一切围绕策略接口设计
```python
class IStrategy:
"""策略抽象接口"""
# ===== 类属性（配置化参数） =====
INTERFACE_VERSION = 3 # 接口版本控制
timeframe = '5m' # K线周期
can_short = False # 是否允许做空
minimal_roi = { # 最小ROI目标
"0": 0.10, # 立即获利10%
"60": 0.05, # 60分钟后获利5%
}
stoploss = -0.10 # 止损比例
# ===== 核心抽象方法（必须实现） =====
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""计算技术指标"""
# 用户自定义指标逻辑
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
dataframe['macd'] = ta.MACD(dataframe)['macd']
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""生成入场信号"""
dataframe.loc[
(dataframe['rsi'] < 30) & # 条件组合
(dataframe['macd'] > 0),
'enter_long'
] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""生成出场信号"""
dataframe.loc[
(dataframe['rsi'] > 70),
'exit_long'
] = 1
return dataframe
# ===== 回调函数（可选实现） =====
def confirm_trade_entry(self, pair, order_type, amount, rate, **kwargs) -> bool:
"""入场前确认回调 - 可拒绝交易"""
return True
def custom_stake_amount(self, pair, current_time, current_rate, **kwargs) -> float:
"""动态仓位调整"""
return self.stake_amount
def custom_stoploss(self, pair, trade, current_time, current_rate, **kwargs) -> float:
"""动态止损"""
return self.stoploss
def custom_exit(self, pair, trade, current_time, current_rate, **kwargs) -> bool:
"""自定义出场条件"""
return False
def adjust_trade_position(self, pair, trade, current_time, current_rate, **kwargs) -> float:
"""动态仓位调整（DCA/部分止盈）"""
return None
```
**关键设计点**：
| 设计点 | 说明 | 优势 |
|--------|------|------|
| **接口版本控制** | `INTERFACE_VERSION` | 向后兼容，平滑升级 |
| **类属性配置** | `timeframe`, `stoploss` | 参数可从配置文件覆盖 |
| **DataFrame驱动** | 所有计算基于DataFrame | 向量化计算，高性能 |
| **回调钩子** | 多生命周期钩子 | 灵活控制交易细节 |
| **信号列名约定** | `enter_long`, `exit_long` | 统一信号格式 |
### 2.2 数据层抽象（Exchange + DataProvider）
**Exchange接口**：基于CCXT库，统一封装100+交易所
```python
class Exchange:
"""交易所抽象（基于CCXT）"""
def __init__(self, config: dict):
# 初始化CCXT交易所实例
self._api = ccxt.binance({
'apiKey': config['key'],
'secret': config['secret'],
'enableRateLimit': True,
})
# ===== 数据获取 =====
def fetch_ohlcv(self, pair: str, timeframe: str, since: int) -> List:
"""获取K线数据"""
return self._api.fetch_ohlcv(pair, timeframe, since)
def fetch_ticker(self, pair: str) -> dict:
"""获取行情"""
return self._api.fetch_ticker(pair)
def fetch_order_book(self, pair: str, limit: int) -> dict:
"""获取订单簿"""
return self._api.fetch_order_book(pair, limit)
# ===== 交易执行 =====
def create_order(self, pair: str, type: str, side: str, amount: float, price: float) -> dict:
"""创建订单"""
return self._api.create_order(pair, type, side, amount, price)
def cancel_order(self, order_id: str, pair: str) -> dict:
"""取消订单"""
return self._api.cancel_order(order_id, pair)
# ===== 账户查询 =====
def fetch_balance(self) -> dict:
"""查询余额"""
return self._api.fetch_balance()
def fetch_orders(self, pair: str) -> List:
"""查询订单"""
return self._api.fetch_orders(pair)
```
**DataProvider**：数据缓存与统一访问
```python
class DataProvider:
"""数据提供者 - 缓存管理"""
def __init__(self, exchange: Exchange, config: dict):
self._exchange = exchange
self._cached_pairs = {}
def ohlcv(self, pair: str, timeframe: str, since: int) -> DataFrame:
"""获取K线数据（带缓存）"""
cache_key = f"{pair}_{timeframe}"
if cache_key not in self._cached_pairs:
# 从交易所获取
data = self._exchange.fetch_ohlcv(pair, timeframe, since)
# 转换为DataFrame
df = pd.DataFrame(data, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
self._cached_pairs[cache_key] = df
return self._cached_pairs[cache_key]
def refresh(self, pair: str, timeframe: str):
"""刷新缓存"""
# 只更新最新数据，不重新下载全部
pass
```
### 2.3 配置驱动设计
**配置层级**：
```json
{
// 系统配置
"max_open_trades": 3,
"stake_currency": "USDT",
"stake_amount": "unlimited",
"dry_run": true,
"dry_run_wallet": 1000,
// 交易所配置
"exchange": {
"name": "binance",
"key": "your_key",
"secret": "your_secret",
"ccxt_config": {
"enableRateLimit": true
},
"pair_whitelist": ["BTC/USDT", "ETH/USDT"],
"pair_blacklist": ["BNB/USDT"]
},
// 策略配置（可覆盖策略类属性）
"strategy": "SampleStrategy",
"timeframe": "5m",
"minimal_roi": {
"0": 0.10
},
"stoploss": -0.10,
// 通知配置
"telegram": {
"enabled": true,
"token": "your_token",
"chat_id": "your_chat_id"
}
}
```
**配置优先级**：
```
命令行参数 > 环境变量 > 配置文件 > 策略默认值
```
### 2.4 Hyperopt参数优化
**参数定义**（在策略中）：
```python
class MyStrategy(IStrategy):
# 定义可优化参数
buy_rsi = IntParameter(10, 40, default=30, space='buy')
sell_rsi = IntParameter(60, 90, default=70, space='sell')
stoploss = DecimalParameter(-0.05, -0.20, default=-0.10, space='stoploss')
def populate_entry_trend(self, dataframe, metadata):
dataframe.loc[
(dataframe['rsi'] < self.buy_rsi.value), # 使用参数
'enter_long'
] = 1
return dataframe
```
**优化空间**：
- `buy` / `enter`: 入场参数
- `sell` / `exit`: 出场参数
- `roi`: ROI表优化
- `stoploss`: 止损优化
- `trailing`: 跟踪止损优化
- `protection`: 保护策略优化
**损失函数**：
```python
class SharpeHyperOptLoss(IHyperOptLoss):
"""Sharpe比率损失函数"""
@staticmethod
def hyperopt_loss_function(results, trade_count, min_date, max_date, *args) -> float:
# 计算Sharpe比率
returns = results['profit_ratio']
sharpe = returns.mean() / returns.std() * np.sqrt(365)
return -sharpe # 返回负值（优化最大化）
```
---
## 三、Freqtrade设计优势
### 3.1 核心设计亮点
| 设计点 | Freqtrade实现 | 优势 |
|--------|---------------|------|
| **策略接口** | `IStrategy`抽象类 + 回调钩子 | 灵活度高，生命周期完整 |
| **数据抽象** | CCXT统一封装100+交易所 | 数据源可插拔 |
| **配置驱动** | JSON配置 + 参数覆盖 | 无需修改代码即可调整 |
| **参数优化** | Optuna集成 + 参数装饰器 | 自动化参数搜索 |
| **执行模式** | backtest/dry/live统一接口 | 平滑切换 |
| **风控回调** | `custom_stoploss`, `custom_exit` | 灵活风控逻辑 |
### 3.2 向量化设计
**DataFrame驱动**：
- 所有指标计算基于DataFrame
- 避免`iloc[-1]`，使用`shift()`
- 批量计算，高性能
- 防止前视偏差（lookahead bias）
```python
# 错误：使用iloc（可能引入前视偏差）
if dataframe['rsi'].iloc[-1] > 70:
dataframe['exit_long'] = 1
# 正确：使用shift()
dataframe.loc[
(dataframe['rsi'].shift(1) > 70), # 使用前一周期数据
'exit_long'
] = 1
```
---
## 四、与我们的框架对比
### 4.1 设计差异对比
| 维度 | Freqtrade | 我们的框架 | 建议 |
|------|-----------|------------|------|
| **策略接口** | `IStrategy`类 + 回调钩子 | 配置驱动策略模式 | **借鉴回调钩子设计** |
| **因子层** | `populate_indicators`方法内定义 | `FactorBase`抽象 + 注册器 | **我们的设计更模块化** |
| **信号层** | `populate_entry_trend/exit_trend` | `SignalGenerator`抽象 | **两者设计相似** |
| **数据层** | CCXT统一封装 | 混合数据源（Tushare + YFinance） | **借鉴交易所抽象接口** |
| **风控层** | 回调函数（`custom_stoploss`） | 风控组件库 | **回调更灵活，组件更复用** |
| **配置驱动** | JSON配置 + 类属性覆盖 | YAML配置 | **两者相似** |
| **参数优化** | Optuna集成 + 参数装饰器 | 无 | **需要添加Hyperopt模块** |
### 4.2 可借鉴设计
#### （1）回调钩子机制
```python
# Freqtrade的回调设计
class IStrategy:
def confirm_trade_entry(self, pair, order_type, amount, rate, **kwargs) -> bool:
"""入场前确认 - 可拒绝交易"""
# 检查溢价率、市场状态等
if self.check_premium(pair) > 0.10:
return False # 拒绝交易
return True
def custom_stoploss(self, pair, trade, current_time, current_rate, **kwargs) -> float:
"""动态止损"""
# 根据持仓时间调整止损
if trade.open_date < current_time - timedelta(days=5):
return -0.05 # 5天后收紧止损
return -0.10
def custom_exit(self, pair, trade, current_time, current_rate, **kwargs) -> bool:
"""自定义出场条件"""
# 检查市场反转信号
if self.detect_reversal(pair):
return True # 强制出场
return False
```
**借鉴方案**：
```python
class StrategyBase:
"""我们的策略基类 + 回调钩子"""
# 入场回调
def before_entry(self, code: str, price: float, **kwargs) -> bool:
"""入场前检查（溢价率、崩盘检测）"""
return True
def after_entry(self, trade: Trade, **kwargs):
"""入场后记录"""
pass
# 出场回调
def before_exit(self, trade: Trade, **kwargs) -> bool:
"""出场前检查"""
return True
def after_exit(self, trade: Trade, **kwargs):
"""出场后记录"""
pass
# 风控回调
def dynamic_stoploss(self, position: Position, **kwargs) -> float:
"""动态止损"""
return self.config.get('stoploss', -0.05)
def position_adjust(self, portfolio: Portfolio, **kwargs) -> dict:
"""仓位调整（DCA/部分止盈）"""
return {}
```
#### （2）参数装饰器
```python
# Freqtrade的参数装饰器
from freqtrade.strategy import IntParameter, DecimalParameter, CategoricalParameter
class MyStrategy(IStrategy):
# 可优化参数定义
buy_rsi_threshold = IntParameter(10, 40, default=30, space='buy', optimize=True)
sell_rsi_threshold = IntParameter(60, 90, default=70, space='sell', optimize=True)
timeframe = CategoricalParameter(['1m', '5m', '15m'], default='5m', space='buy')
```
**借鉴方案**：
```python
# 我们的参数装饰器设计
class Parameter:
"""参数基类"""
def __init__(self, low, high, default, space='default', optimize=True):
self.low = low
self.high = high
self.default = default
self.space = space
self.optimize = optimize
self._value = default
def __get__(self, obj, objtype):
return self._value
def __set__(self, obj, value):
self._value = value
class IntParameter(Parameter):
"""整数参数"""
pass
class FloatParameter(Parameter):
"""浮点参数"""
pass
# 使用示例
class RotationStrategy(StrategyBase):
n_days = IntParameter(10, 60, default=25, space='factor')
select_num = IntParameter(1, 5, default=3, space='signal')
stoploss = FloatParameter(0.01, 0.10, default=0.05, space='risk')
```
#### （3）接口版本控制
```python
# Freqtrade的接口版本
class IStrategy:
INTERFACE_VERSION = 3 # 当前版本
# 不同版本的参数命名不同
# v2: buy, sell
# v3: enter_long, exit_long
```
**借鉴方案**：
```python
class StrategyBase:
"""策略基类 + 版本控制"""
INTERFACE_VERSION = 1
@classmethod
def validate_version(cls, config: dict) -> bool:
"""验证配置与版本匹配"""
if config.get('version', 1) != cls.INTERFACE_VERSION:
raise ValueError(f"Strategy version mismatch")
return True
```
---
## 五、融合设计方案
### 5.1 最佳实践融合
**融合Freqtrade优势 + 我们的模块化设计**：
```python
class QuantStrategy(StrategyBase):
"""量化策略基类 - 融合设计"""
INTERFACE_VERSION = 1
# ===== 类属性（配置可覆盖） =====
timeframe = '1d'
select_num = 3
stoploss = -0.05
# ===== 可优化参数（装饰器） =====
n_days = IntParameter(10, 60, default=25, space='factor')
# ===== 因子模块（模块化） =====
def init_factors(self) -> FactorCombiner:
"""初始化因子组合"""
factors = [
FactorRegistry.get('momentum', n_days=self.n_days.value),
FactorRegistry.get('volatility', period=20),
]
return FactorCombiner(factors, weights=[0.7, 0.3])
# ===== 核心方法（必须实现） =====
def populate_indicators(self, data: DataFrame, metadata: dict) -> DataFrame:
"""计算因子"""
factor_values = self._factors.compute(data)
data = pd.concat([data, factor_values], axis=1)
return data
def generate_signals(self, data: DataFrame, metadata: dict) -> DataFrame:
"""生成信号"""
return self._signal_gen.generate(data)
# ===== 回调钩子（可选实现） =====
def before_entry(self, code: str, price: float, **kwargs) -> bool:
"""入场前检查"""
# 溢价率过滤
if kwargs.get('premium', 0) > 0.10:
print(f"溢价过高，拒绝入场: {code}")
return False
# 崩盘检测
if self._detect_crash(code):
return False
return True
def dynamic_stoploss(self, position: Position, **kwargs) -> float:
"""动态止损"""
# 持仓时间越长，止损越紧
holding_days = position.holding_days
if holding_days > 10:
return -0.03 # 10天后收紧止损
elif holding_days > 5:
return -0.05
return -0.10
def custom_exit(self, position: Position, **kwargs) -> bool:
"""自定义出场"""
# 反转信号检测
if self._detect_reversal(position.code):
return True
return False
```
### 5.2 配置驱动融合
```yaml
# config/strategies/rotation.yaml（融合设计）
strategy:
name: "rotation"
version: 1
# 参数（可覆盖类属性）
parameters:
n_days: 25 # 可优化参数
select_num: 3
stoploss: -0.05
# 因子配置
factors:
- name: "momentum"
weight: 0.7
params:
n_days: "${n_days}" # 引用参数
- name: "volatility"
weight: 0.3
# 信号配置
signal:
mode: "top_n"
select_num: "${select_num}"
group_by: "market"
# 回调配置
callbacks:
before_entry:
- type: "premium_filter"
threshold: 0.10
- type: "crash_filter"
dynamic_stoploss:
enabled: true
schedule:
- days: 5
stoploss: -0.05
- days: 10
stoploss: -0.03
# Hyperopt配置
hyperopt:
enabled: true
spaces: ["factor", "signal"]
loss: "SharpeLoss"
epochs: 100
```
---
## 六、总结与建议
### 6.1 Freqtrade设计精髓
| 设计精髓 | 说明 |
|----------|------|
| **策略为中心** | 所有模块围绕`IStrategy`接口设计 |
| **回调钩子** | 生命周期完整控制，灵活度高 |
| **配置覆盖** | 类属性 → 配置文件 → 命令行 |
| **参数装饰器** | 声明式参数定义，自动优化 |
| **向量化计算** | DataFrame驱动，高性能 + 防前视偏差 |
| **接口版本** | 平滑升级，向后兼容 |
### 6.2 建议借鉴内容
| 优先级 | 借鉴内容 | 预估工作量 |
|--------|----------|------------|
| **P1** | 回调钩子机制（`before_entry`, `custom_stoploss`） | 1天 |
| **P1** | 参数装饰器（`IntParameter`, `FloatParameter`） | 1天 |
| **P2** | 接口版本控制 | 0.5天 |
| **P2** | Hyperopt模块（Optuna集成） | 2天 |
| **P3** | DataFrame向量化检查工具 | 1天 |
### 6.3 保持我们优势
| 我们的设计优势 | 说明 |
|----------------|------|
| **因子模块化** | `FactorBase` + `FactorRegistry` 更易于复用 |
| **信号模式** | `SignalGenerator` 抽象更清晰 |
| **风控组件库** | 组件化设计更易于组合 |
| **混合数据源** | Tushare + YFinance 跨市场支持 |
---
*文档版本：V1.0*
*调研时间：2026-05-08*
*参考项目：Freqtrade (freqtrade/freqtrade)*

535
docs/轮动策略系统架构分析与通用化设计.md Normal file
View File

@@ -0,0 +1,535 @@
# ETF轮动策略系统架构分析与通用化设计报告
## 一、现有系统架构分析
### 1.1 整体架构概览
```
etf/
├── config/ # 配置中心
│ └── strategies/
│ └── rotation.yaml # 策略参数配置
├── core/ # 核心基础设施
│ ├── common/ # 公共工具（utils、db、notify）
│ ├── datasource/ # 数据源层（混合数据源、统一fetcher）
│ └── factors/ # 因子计算层（动量、技术指标）
├── strategies/ # 策略层
│ ├── base.py # 策略抽象基类
│ └── rotation/ # 轮动策略实现
│ ├── engine.py # 策略引擎（核心逻辑）
│ ├── portfolio.py # 持仓跟踪与交易记录
│ └── report.py # 绩效报告生成
├── scripts/ # 运行入口
│ └── run_rotation.py # 回测脚本
├── visualization/ # 可视化层
│ └── report_generator/ # HTML报告生成器
└── tests/ # 测试与实验
```
### 1.2 核心模块职责
| 模块 | 职责 | 关键文件 |
|------|------|----------|
| **数据源层** | 多源数据获取、SSH隧道、缓存管理 | `hybrid_source.py`, `universal_fetcher.py` |
| **因子层** | 动量得分计算、崩盘过滤、ATR动态周期 | `momentum.py` |
| **策略引擎** | 信号生成、选股逻辑、调仓控制 | `engine.py` |
| **持仓跟踪** | 交易记录、持仓统计、汇总 | `portfolio.py` |
| **报告层** | KPI计算、图表绘制、HTML生成 | `report.py`, `generate_report.py` |
### 1.3 策略核心流程
```mermaid
graph TB
A[配置加载] --> B[数据获取]
B --> C[因子计算]
C --> D[信号生成]
D --> E[回测执行]
E --> F[持仓跟踪]
F --> G[报告生成]
B --> B1[Tushare A股]
B --> B2[YFinance 港美股]
B --> B3[SSH隧道代理]
C --> C1[加权动量得分]
C --> C2[崩盘过滤]
C --> C3[溢价控制]
D --> D1[类内竞争Top1]
D --> D2[跨类排序Top3]
D --> D3[T+1执行]
```
---
## 二、现有实现特点分析
### 2.1 数据层设计
**优点**：
- 混合数据源架构：Tushare(A股) + YFinance(港美股)
- SSH隧道统一治理：解决网络受限环境问题
- 双轨数据架构：指数价格（信号源）+ ETF价格（收益计算）
- 智能路由：根据代码格式自动选择数据源
**问题**：
- ETF列名与指数代码不匹配导致收益计算使用错误数据源
- 缓存策略过于简单，缺乏版本控制
### 2.2 因子层设计
**优点**：
- 三种因子类型：momentum、slope_r2、weighted_momentum
- 崩盘过滤机制：防止追接下跌飞刀
- ATR动态周期：自适应市场波动
- 加权线性回归：近期权重更高，对趋势变化敏感
**问题**：
- 因子类型固定，扩展需修改源码
- 缺乏因子组合/加权机制
### 2.3 策略层设计
**优点**：
- 分散化选股：每大类Top1→全局Top3，强制跨资产配置
- T+1执行机制：信号基于T日数据，T+1日执行
- 溢价控制：跨境ETF溢价过滤
- 持仓跟踪：完整的交易记录和净值贡献计算
**问题**：
- 调仓逻辑与因子计算耦合
- 缺乏风控模块（止损、仓位管理）
- 单一策略类型，无法扩展其他策略
### 2.4 报告层设计
**优点**：
- 完整的KPI指标体系
- 可视化报告：净值曲线、月度收益、盈亏分布、品种排行
- 数据一致性：严格基于回测结果文件
**问题**：
- 报告与策略耦合
- 缺乏实盘监控接口
---
## 三、通用交易系统抽象设计
### 3.1 核心抽象原则
| 原则 | 说明 |
|------|------|
| **数据独立** | 数据获取与策略逻辑分离，支持多源扩展 |
| **因子可插拔** | 因子计算模块化，支持自定义因子组合 |
| **策略可配置** | 策略逻辑通过配置驱动，无需修改代码 |
| **执行可扩展** | 支持回测、模拟盘、实盘多种执行模式 |
| **风控独立** | 风控模块独立，支持止损、仓位、敞口控制 |
### 3.2 建议架构设计
```
trading_system/
├── core/
│ ├── data/
│ │ ├── sources/ # 数据源接口（可插拔）
│ │ │ ├── base.py # 数据源抽象基类
│ │ │ ├── tushare.py
│ │ │ ├── yfinance.py
│ │ │ └── ccxt.py
│ │ ├── router.py # 数据源路由
│ │ └── cache.py # 缓存管理（版本控制）
│ │
│ ├── factors/
│ │ ├── base.py # 因子抽象基类
│ │ ├── registry.py # 因子注册器
│ │ ├── momentum.py
│ │ ├── technical.py
│ │ └── fundamental.py
│ │
│ ├── execution/
│ │ ├── base.py # 执行器抽象基类
│ │ ├── backtest.py # 回测执行器
│ │ ├── simulator.py # 模拟盘执行器
│ │ └── live.py # 实盘执行器
│ │
│ └── risk/
│ │ ├── base.py # 风控抽象基类
│ │ ├── stop_loss.py # 止损控制
│ │ ├── position.py # 仓位控制
│ │ └── exposure.py # 敞口控制
│
├── strategies/
│ ├── base.py # 策略抽象基类
│ ├── registry.py # 策略注册器
│ ├── rotation/ # 轮动策略
│ ├── momentum/ # 动量策略
│ └── screener/ # 篩选策略
│
├── portfolio/
│ ├── manager.py # 持仓管理器
│ ├── tracker.py # 交易跟踪
│ ├── rebalancer.py # 调仓执行器
│
├── report/
│ ├── generator.py # 报告生成器
│ ├── metrics.py # KPI计算
│ └── visualizer.py # 可视化
│
├── config/
│ ├── system.yaml # 系统配置
│ ├── strategies/ # 策略配置
│ │ ├── rotation.yaml
│ │ └── momentum.yaml
│ └── factors/ # 因子配置
│
└── scheduler/
├── engine.py # 调度引擎
├── triggers.py # 触发器（定时、事件）
└── notifier.py # 通知器
```
### 3.3 核心接口设计
```python
# ==================== 数据源接口 ====================
class DataSource(ABC):
"""数据源抽象基类"""
@abstractmethod
def fetch(self, code: str, start: str, end: str) -> pd.DataFrame:
"""获取OHLCV数据"""
pass
@abstractmethod
def get_trading_calendar(self, market: str) -> List[datetime]:
"""获取交易日历"""
pass
@abstractmethod
def get_supported_codes(self) -> List[str]:
"""获取支持的代码列表"""
pass
# ==================== 因子接口 ====================
class Factor(ABC):
"""因子抽象基类"""
@abstractmethod
def compute(self, data: pd.DataFrame) -> pd.Series:
"""计算因子值"""
pass
@abstractmethod
def get_name(self) -> str:
"""获取因子名称"""
pass
@property
def params(self) -> dict:
"""因子参数"""
return {}
class FactorRegistry:
"""因子注册器"""
_factors = {}
@classmethod
def register(cls, name: str, factor_cls: Factor):
cls._factors[name] = factor_cls
@classmethod
def get(cls, name: str) -> Factor:
return cls._factors.get(name)
@classmethod
def list(cls) -> List[str]:
return list(cls._factors.keys())
# ==================== 执行器接口 ====================
class Executor(ABC):
"""执行器抽象基类"""
@abstractmethod
def execute(self, signals: pd.DataFrame, portfolio: Portfolio) -> Portfolio:
"""执行信号"""
pass
@abstractmethod
def get_mode(self) -> str:
"""获取执行模式"""
pass
class BacktestExecutor(Executor):
"""回测执行器"""
def execute(self, signals, portfolio):
# 处理调仓成本
# 计算收益率
# 更新净值
pass
def get_mode(self) -> str:
return "backtest"
class LiveExecutor(Executor):
"""实盘执行器"""
def execute(self, signals, portfolio):
# 检查交易权限
# 发送交易指令
# 记录交易结果
pass
def get_mode(self) -> str:
return "live"
# ==================== 风控接口 ====================
class RiskControl(ABC):
"""风控抽象基类"""
@abstractmethod
def check(self, portfolio: Portfolio, signal: Signal) -> bool:
"""风控检查"""
pass
@abstractmethod
def apply(self, portfolio: Portfolio) -> Portfolio:
"""应用风控"""
pass
class StopLossControl(RiskControl):
"""止损控制"""
def __init__(self, threshold: float = 0.05):
self.threshold = threshold
def check(self, portfolio, signal):
# 检查是否触发止损
for position in portfolio.positions:
if position.loss > self.threshold:
return False
return True
def apply(self, portfolio):
# 强制平仓止损
pass
# ==================== 策略接口 ====================
class Strategy(ABC):
"""策略抽象基类"""
@abstractmethod
def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame:
"""生成信号"""
pass
@abstractmethod
def get_name(self) -> str:
"""获取策略名称"""
pass
@abstractmethod
def validate_config(self, config: dict) -> bool:
"""验证配置"""
pass
# ==================== 持仓管理接口 ====================
class Portfolio:
"""持仓管理器"""
def __init__(self):
self.positions = {} # 当前持仓
self.cash = 0.0 # 现金
self.nav = 1.0 # 净值
self.trades = [] # 交易记录
def rebalance(self, target_positions: dict):
"""调仓"""
pass
def get_nav(self) -> float:
"""获取净值"""
pass
def get_positions(self) -> dict:
"""获取持仓"""
return self.positions
```
### 3.4 配置驱动设计
```yaml
# config/strategies/rotation.yaml（通用化后）
# ==================== 基础配置 ====================
strategy:
name: "rotation"
version: "2.0"
mode: "backtest" # backtest | simulator | live
# ==================== 候选池配置 ====================
universe:
type: "static" # static | dynamic
codes:
# ...（与现有配置相同）
mapping:
signal_to_trade: true # 信号源→交易标的映射
# ==================== 因子配置 ====================
factors:
- name: "weighted_momentum"
weight: 1.0
params:
n_days: 25
crash_filter: true
# 可扩展多个因子
# - name: "volatility"
# weight: 0.3
# params:
# period: 20
# ==================== 选股配置 ====================
selection:
mode: "diversified" # top_n | diversified | custom
select_num: 3
group_by: "market" # 按大类分组
top_per_group: 1
# ==================== 调仓配置 ====================
rebalance:
frequency: "daily"
min_holding_days: 1
threshold: 0.0
cost: 0.001
# ==================== 风控配置 ====================
risk:
stop_loss:
enabled: true
threshold: 0.05
position_limit:
max_position: 0.5
max_single: 0.33
premium_control:
enabled: true
threshold: 0.10
# ==================== 执行配置 ====================
execution:
mode: "backtest"
start_date: "2019-01-01"
benchmark: "000300.SH"
# ==================== 数据源配置 ====================
data:
sources:
- name: "tushare"
markets: ["A"]
- name: "yfinance"
markets: ["US", "HK", "JP", "EU"]
proxy:
type: "ssh"
host: "8.218.167.69"
port: 22
```
---
## 四、迁移路径建议
### 4.1 阶段一：数据层重构
**目标**：修复ETF列名问题，建立可扩展数据源架构
**任务**：
1. 修复`compute_factors()`中的ETF数据路由逻辑
2. 创建`DataSource`抽象基类
3. 统一数据缓存管理（添加版本控制）
### 4.2 阶段二：因子层抽象
**目标**：因子计算可插拔、可组合
**任务**：
1. 创建`Factor`抽象基类
2. 建立`FactorRegistry`注册机制
3. 支持因子组合和加权
### 4.3 阶段三：风控模块独立
**目标**：止损、仓位、敞口控制
**任务**：
1. 创建`RiskControl`抽象基类
2. 实现止损控制
3. 实现仓位限制
### 4.4 阶段四：执行器分离
**目标**：支持回测、模拟盘、实盘三种模式
**任务**：
1. 创建`Executor`抽象基类
2. 实现回测执行器（迁移现有逻辑）
3. 实现模拟盘执行器（模拟交易）
4. 实现实盘执行器（连接券商API）
### 4.5 阶段五：配置驱动
**目标**：策略逻辑通过配置驱动
**任务**：
1. 策略配置YAML重构
2. 配置验证机制
3. 热加载配置
---
## 五、总结与建议
### 5.1 现有系统评价
| 维度 | 评分 | 说明 |
|------|------|------|
| 数据层 | 3.5/5 | 多源架构优秀，但有列名匹配bug |
| 因子层 | 4/5 | 加权动量设计合理，崩盘过滤有效 |
| 策略层 | 3/5 | 分散化选股创新，但缺乏扩展性 |
| 执行层 | 2/5 | 仅支持回测，缺乏模拟/实盘 |
| 风控层 | 1/5 | 仅溢价控制，缺乏止损/仓位管理 |
| 报告层 | 4.5/5 | KPI完整，可视化丰富 |
### 5.2 通用化优先级
| 优先级 | 模块 | 紧急程度 | 预估工作量 |
|--------|------|----------|------------|
| P0 | 修复ETF数据路由bug | 🔴紧急 | 0.5天 |
| P1 | 风控模块独立 | 🟡重要 | 2天 |
| P1 | 因子层抽象 | 🟡重要 | 1天 |
| P2 | 数据源可插拔 | 🟢一般 | 2天 |
| P2 | 执行器分离 | 🟢一般 | 3天 |
| P3 | 配置驱动 | 🔵低 | 1天 |
### 5.3 关键改进建议
1. **立即修复**：ETF数据列名匹配问题（影响回测真实性）
2. **短期改进**：添加止损控制（保护实盘资金）
3. **中期重构**：数据源/因子/执行器抽象（提升扩展性）
4. **长期规划**：实盘执行器、调度系统（支持自动化运行）
---
*文档版本：V1.0*
*生成时间：2026-05-08*
*适用范围：ETF轮动策略系统架构分析与通用化设计*

682
docs/量化策略通用框架抽象设计.md Normal file
View File

@@ -0,0 +1,682 @@
# 量化策略通用框架抽象设计
## 一、三种策略核心能力对比
| 能力维度 | 轮动策略 | 趋势跟踪策略 | 反转策略 |
|----------|----------|--------------|----------|
| **核心逻辑** | 动量排序，选出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*
*适用范围：量化策略通用框架设计*