diff --git a/docs/ETF轮动策略通用化重构方案.md b/docs/ETF轮动策略通用化重构方案.md new file mode 100644 index 0000000..25cf1fe --- /dev/null +++ b/docs/ETF轮动策略通用化重构方案.md @@ -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* +*等待用户确认* \ No newline at end of file diff --git a/docs/Freqtrade架构调研与对比分析.md b/docs/Freqtrade架构调研与对比分析.md new file mode 100644 index 0000000..0a12308 --- /dev/null +++ b/docs/Freqtrade架构调研与对比分析.md @@ -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)* \ No newline at end of file diff --git a/docs/轮动策略系统架构分析与通用化设计.md b/docs/轮动策略系统架构分析与通用化设计.md new file mode 100644 index 0000000..d79e5bb --- /dev/null +++ b/docs/轮动策略系统架构分析与通用化设计.md @@ -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轮动策略系统架构分析与通用化设计* \ No newline at end of file diff --git a/docs/量化策略通用框架抽象设计.md b/docs/量化策略通用框架抽象设计.md new file mode 100644 index 0000000..59bd839 --- /dev/null +++ b/docs/量化策略通用框架抽象设计.md @@ -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* +*适用范围：量化策略通用框架设计* \ No newline at end of file