etf/docs/轮动策略系统架构分析与通用化设计.md

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 策略核心流程

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 核心接口设计

# ==================== 数据源接口 ====================
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 配置驱动设计

# 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轮动策略系统架构分析与通用化设计