🇨🇳 简体中文
🇺🇸 English
🇯🇵 日本語

Strategy 模块架构文档

本文档描述 Strategy 模块的架构设计、组件结构、接口定义和配置说明。

概述

Strategy 模块是 Fire 量化交易平台的决策核心，负责分析市场数据、生成交易信号和管理策略生命周期。系统采用插件化架构，支持技术分析、统计套利、机器学习等多种策略类型，通过元数据驱动的注册系统实现策略的自动发现和管理。

核心功能

快速开始

# 基本使用示例
from backend.core.trading.strategies import StrategyEngine, StrategyRegistry
from backend.core.trading.strategies.metadata import StrategyMetadata, StrategyCategory

# 获取可用策略
available_strategies = StrategyRegistry.list_strategies()

# 初始化策略引擎
strategy_engine = StrategyEngine(
    session_id="session_123",
    user_id="user_456",
    symbols=["AAPL", "GOOGL"],
    strategies_config={
        "momentum_breakout": {"rsi_period": 14, "volume_multiplier": 2.0},
        "ma_crossover": {"fast_period": 10, "slow_period": 30}
    }
)

# 初始化并启动
strategy_engine.initialize()
strategy_engine.start()

# 处理市场数据
market_data = get_market_data()
signals = strategy_engine.process_market_data(market_data)

组件

组件架构图

graph TD
    subgraph "策略核心"
        BS[BaseStrategy<br/>策略基类]
        SR[StrategyRegistry<br/>策略注册表]
        SF[StrategyFactory<br/>策略工厂]
        SM[StrategyMetadata<br/>策略元数据]
    end

    subgraph "策略引擎"
        SE[StrategyEngine<br/>策略引擎]
        SC[StrategyContext<br/>策略上下文]
        SL[StrategyLoader<br/>策略加载器]
        WS[WarmupService<br/>预热服务]
    end

    subgraph "策略实现"
        MOM[Momentum<br/>动量策略]
        TRD[Trend<br/>趋势策略]
        VOL[Volatility<br/>波动率策略]
        VLM[Volume<br/>成交量策略]
    end

    subgraph "指标系统"
        PTW[PandasTAWrapper<br/>指标包装器]
        IC[IndicatorCache<br/>指标缓存]
        L1[L1Cache<br/>全局缓存]
        L2[L2Cache<br/>实例缓存]
    end

    subgraph "信号处理"
        SG[SignalGenerator<br/>信号生成]
        VS[VotingService<br/>投票服务]
        SA[SignalAggregator<br/>信号聚合]
    end

    SE --> SC
    SC --> BS
    BS --> MOM
    BS --> TRD
    BS --> VOL
    BS --> VLM

    SR --> SF
    SF --> BS
    SM --> SR

    BS --> PTW
    PTW --> IC
    IC --> L1
    IC --> L2

    SE --> WS
    BS --> SG
    SG --> VS
    VS --> SA

    style BS fill:#f9f,stroke:#333,stroke-width:2px
    style SE fill:#bbf,stroke:#333,stroke-width:2px
    style PTW fill:#bfb,stroke:#333,stroke-width:2px
    style SR fill:#fbb,stroke:#333,stroke-width:2px

核心组件说明

BaseStrategy (策略基类)

StrategyEngine (策略引擎)

StrategyRegistry (策略注册表)

PandasTAWrapper (指标包装器)

IndicatorCache (指标缓存)

依赖关系

内部依赖

模块名 版本 用途 是否必需
core.trading.models - 数据模型定义 是
core.trading.engines - 交易引擎接口 是
core.data_source - 市场数据获取 是
core.repositories - 策略配置持久化 是
infrastructure.cache - Redis 缓存支持 是
utils.logging - 日志记录 是
utils.metrics - 性能指标 否

外部依赖

库名 版本要求 用途
pandas ≥2.3.0 数据处理
numpy ≥1.24.0 数值计算
pandas-ta 0.4.71b0 技术指标
pydantic ≥2.6.0 数据验证
asyncio 标准库 异步支持

数据流

策略执行流程图

flowchart LR
    Start([市场数据到达]) --> SE[策略引擎]
    SE --> Check{需要预热?}
    Check -->|是| Warmup[预热服务]
    Check -->|否| Distribute[分发数据]

    Warmup --> Load[加载历史数据]
    Load --> Feed[喂入策略]
    Feed --> Ready{预热完成?}
    Ready -->|是| Distribute
    Ready -->|否| Feed

    Distribute --> Strategy[各策略实例]
    Strategy --> Calc[计算指标]
    Calc --> Cache{缓存命中?}
    Cache -->|是| Use[使用缓存]
    Cache -->|否| Compute[计算新值]
    Compute --> Store[存储缓存]
    Store --> Use

    Use --> Generate[生成信号]
    Generate --> Validate{验证信号}
    Validate -->|有效| Collect[收集信号]
    Validate -->|无效| Skip[跳过]

    Collect --> Vote[投票聚合]
    Vote --> Final[最终信号]
    Final --> End([返回信号])

    style Start fill:#e1f5fe
    style End fill:#e1f5fe
    style Skip fill:#ffebee
    style Final fill:#c8e6c9

策略注册流程

sequenceDiagram
    participant Dev as 开发者
    participant Reg as StrategyRegistry
    participant Meta as StrategyMetadata
    participant Factory as StrategyFactory

    Dev->>Meta: 定义元数据
    Dev->>Reg: @register装饰器
    Reg->>Reg: 验证元数据
    Reg->>Reg: 检查ID唯一性
    Reg->>Reg: 注册策略类
    Reg->>Factory: 更新工厂

    Note over Reg: 运行时
    Dev->>Reg: 列出策略
    Reg-->>Dev: 返回策略列表
    Dev->>Factory: 创建实例
    Factory->>Reg: 获取策略类
    Reg-->>Factory: 返回类定义
    Factory->>Factory: 实例化策略
    Factory-->>Dev: 返回实例

输入数据格式

{
  "symbol": "AAPL",
  "timestamp": "2025-01-01T09:30:00Z",
  "open": 149.50,
  "high": 151.20,
  "low": 149.30,
  "close": 150.25,
  "volume": 5000000,
  "indicators": {
    "sma_20": 148.75,
    "rsi_14": 65.5,
    "macd": {
      "macd": 1.25,
      "signal": 1.10,
      "histogram": 0.15
    }
  }
}

输出数据格式

{
  "signal": {
    "strategy_id": "momentum_breakout",
    "symbol": "AAPL",
    "timestamp": "2025-01-01T09:30:00Z",
    "direction": "BUY",
    "confidence": 0.85,
    "suggested_quantity": 100,
    "stop_loss": 148.50,
    "take_profit": 153.00,
    "metadata": {
      "rsi": 72.5,
      "volume_spike": 2.3,
      "breakout_level": 150.00,
      "indicators_used": ["RSI", "Volume", "ATR"]
    }
  }
}

接口

REST API 端点

GET /api/v1/strategies/metadata/all

获取所有策略的完整元数据（用于Flow编辑器）。

响应示例:

{
  "strategies": [
    {
      "strategy_id": "momentum_breakout",
      "display_name": "Momentum Breakout Strategy",
      "category": "Momentum",
      "risk_level": "High",
      "description": "价格突破 + 成交量确认",
      "parameters": [
        {
          "name": "rsi_period",
          "display_name": "RSI周期",
          "type": "int",
          "default": 14,
          "min": 5,
          "max": 30,
          "description": "RSI指标计算周期"
        },
        {
          "name": "volume_threshold",
          "display_name": "成交量倍数",
          "type": "float",
          "default": 1.5,
          "min": 1.0,
          "max": 3.0,
          "description": "相对平均成交量的倍数"
        }
      ]
    }
  ]
}

实现: backend.api.v1.endpoints.strategies.get_all_metadata()

用途: 前端Flow编辑器获取策略列表和参数定义

技术指标 API 端点

GET /api/v1/indicators

获取所有可用的技术指标列表。

响应示例:

[
  {
    "id": "rsi",
    "name": "RSI",
    "display_name": "相对强弱指标",
    "category": "momentum",
    "description": "衡量价格变动速度和变化幅度的动量振荡器",
    "parameters_count": 3
  },
  {
    "id": "macd",
    "name": "MACD",
    "display_name": "移动平均收敛/发散",
    "category": "trend",
    "description": "显示两条移动平均线之间关系的趋势跟踪动量指标",
    "parameters_count": 4
  }
]

实现: backend.api.v1.endpoints.indicators.list_indicators()

GET /api/v1/indicators/{indicator_id}

获取单个技术指标的详细信息。

路径参数:

响应示例:

{
  "id": "rsi",
  "name": "RSI",
  "display_name": "相对强弱指标",
  "english_name": "Relative Strength Index",
  "category": "momentum",
  "description": "RSI 是一种动量振荡器，用于衡量价格变动的速度和变化幅度...",
  "parameters": [
    {
      "name": "length",
      "type": "int",
      "default": 14,
      "min": 2,
      "max": 100,
      "description": "计算周期"
    },
    {
      "name": "scalar",
      "type": "float",
      "default": 100.0,
      "description": "缩放因子"
    }
  ],
  "pandas_ta": {
    "function": "ta.rsi(close, length=14, scalar=100, ...)",
    "returns": "pd.Series",
    "example": "df.ta.rsi(length=14)"
  },
  "interpretation": {
    "overbought": "RSI > 70 表示超买",
    "oversold": "RSI < 30 表示超卖",
    "divergence": "价格与RSI背离可能预示反转"
  }
}

实现: backend.api.v1.endpoints.indicators.get_indicator()

技术指标元数据存储

指标元数据以 JSON 文件形式存储在 backend/core/trading/strategies/indicators/metadata/ 目录下：

backend/core/trading/strategies/indicators/metadata/
├── _template.md          # 指标 JSON 模板说明
├── rsi.json              # RSI 指标元数据
├── macd.json             # MACD 指标元数据
├── bollinger_bands.json  # 布林带指标元数据
├── sma.json              # 简单移动平均线
├── ema.json              # 指数移动平均线
├── atr.json              # 真实波幅
├── stoch.json            # 随机指标
├── obv.json              # 能量潮
├── vwap.json             # 成交量加权平均价
└── adx.json              # 平均方向指数

JSON 文件格式:

{
  "id": "indicator_id",
  "name": "INDICATOR",
  "display_name": "指标中文名",
  "english_name": "English Name",
  "category": "category_type",
  "description": "详细描述...",
  "parameters": [...],
  "pandas_ta": {...},
  "interpretation": {...}
}

GET /api/v1/strategies

获取可用策略列表（简化版本）。

请求参数：

参数名 类型 必需 说明 示例
category string 否 策略类别 “MOMENTUM”
enabled boolean 否 是否启用 true

响应示例：

{
  "code": 0,
  "message": "success",
  "data": {
    "strategies": [
      {
        "strategy_id": "momentum_breakout",
        "display_name": "动量突破策略",
        "category": "MOMENTUM",
        "risk_level": "HIGH",
        "description": "基于RSI和成交量的动量突破策略"
      }
    ],
    "total": 10
  }
}

POST /api/v1/strategies/backtest

执行策略回测。

请求参数：

参数名 类型 必需 说明 示例
strategy_id string 是 策略ID “momentum_breakout”
symbol string 是 股票代码 “AAPL”
start_date string 是 开始日期 “2024-01-01”
end_date string 是 结束日期 “2024-12-31”
initial_capital float 是 初始资金 100000
parameters object 否 策略参数 {“rsi_period”: 14}

请求示例：

curl -X POST "http://localhost:8000/api/v1/strategies/backtest" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer your_token_here" \
  -d '{
    "strategy_id": "momentum_breakout",
    "symbol": "AAPL",
    "start_date": "2024-01-01",
    "end_date": "2024-12-31",
    "initial_capital": 100000,
    "parameters": {
      "rsi_period": 14,
      "volume_multiplier": 2.0
    }
  }'

响应格式：

{
  "code": 0,
  "message": "success",
  "data": {
    "metrics": {
      "total_return": 0.25,
      "sharpe_ratio": 1.8,
      "max_drawdown": 0.12,
      "win_rate": 0.62,
      "total_trades": 150
    },
    "equity_curve": [
      {"date": "2024-01-01", "value": 100000},
      {"date": "2024-01-02", "value": 100500}
    ]
  }
}

Python SDK

from backend.core.trading.strategies.client import StrategyClient

# 初始化客户端
client = StrategyClient(
    api_key="your_api_key_here",
    base_url="http://localhost:8000"
)

# 获取策略列表
strategies = client.list_strategies(category="MOMENTUM")

# 创建策略实例
strategy = client.create_strategy(
    strategy_id="momentum_breakout",
    parameters={
        "rsi_period": 14,
        "volume_multiplier": 2.0
    }
)

# 执行回测
backtest_result = client.backtest(
    strategy_id="momentum_breakout",
    symbol="AAPL",
    start_date="2024-01-01",
    end_date="2024-12-31",
    initial_capital=100000
)

# 获取策略信号
signal = strategy.analyze(symbol="AAPL", market_data=data)

扩展点

自定义策略实现

通过继承 BaseStrategy 基类来创建自定义策略：

from backend.core.trading.strategies import BaseStrategy, register
from backend.core.trading.strategies.metadata import StrategyMetadata, StrategyCategory

@register(metadata=StrategyMetadata(
    strategy_id="custom_strategy",
    display_name="自定义策略",
    category=StrategyCategory.CUSTOM,
    description="自定义策略实现",
    risk_level="MEDIUM",
    parameters={
        "lookback_period": {"type": "int", "default": 20, "min": 5, "max": 100},
        "threshold": {"type": "float", "default": 0.02, "min": 0.01, "max": 0.1}
    }
))
class CustomStrategy(BaseStrategy):
    """自定义策略实现"""

    def initialize(self, context: StrategyContext) -> None:
        """初始化策略"""
        super().initialize(context)
        self.lookback_period = self.config.get("lookback_period", 20)
        self.threshold = self.config.get("threshold", 0.02)
        self._init_warmup_state()

    def get_required_history_length(self) -> int:
        """声明预热所需数据"""
        return self.lookback_period + 10

    def analyze(self, symbol: str, market_data: pd.DataFrame) -> Signal:
        """分析市场数据生成信号"""
        # 实现策略分析逻辑
        # 参考具体策略文档: docs/strategies/
        pass

自定义指标

from backend.core.trading.strategies.indicators import register_indicator

@register_indicator("custom_oscillator")
def custom_oscillator(df: pd.DataFrame, period: int = 14) -> pd.Series:
    """自定义振荡指标"""
    # 实现指标计算逻辑
    # 返回 pd.Series 格式的指标值
    pass

钩子函数

支持以下生命周期钩子：

@strategy.hook("before_analyze")
def validate_data(market_data: pd.DataFrame) -> bool:
    """分析前验证数据完整性"""
    required_columns = ['open', 'high', 'low', 'close', 'volume']
    return all(col in market_data.columns for col in required_columns)

@strategy.hook("on_signal_generated")
def log_signal(signal: Signal):
    """记录生成的信号"""
    logger.info(f"Signal generated: {signal}")
    metrics.record_signal(signal)

配置

实际配置方式

Strategy 模块的配置通过以下方式管理：

  1. 策略元数据：通过 @register 装饰器定义的 StrategyMetadata
  2. 运行时参数：通过 API 调用时传入的 strategies_config
  3. 环境变量：通过 infrastructure/config/settings.py

策略参数配置系统

参数定义位置

所有策略参数必须在 backend/core/models/strategy_config.py 中定义。这是唯一的参数配置来源。

from core.models.strategy_config import StrategyParameter, ParameterType, StrategyConfigTemplate

# 在 STRATEGY_TEMPLATES 字典中定义策略参数
"MomentumBreakoutStrategy": StrategyConfigTemplate(
    strategy_name="MomentumBreakoutStrategy",
    display_name="动量突破策略",
    description="检测价格突破历史波动区间，结合成交量确认",
    version="1.0.0",
    author="系统",
    parameters=[
        # 整数参数 - 周期类
        StrategyParameter(
            name="lookback_period",
            type=ParameterType.INTEGER,  # 整数类型
            default_value=20,
            description="回溯周期（用于计算统计值的历史条数）",
            min_value=5,
            max_value=100,
            category="technical",
        ),
        # 浮点数参数 - 比率类
        StrategyParameter(
            name="breakout_threshold",
            type=ParameterType.FLOAT,  # 浮点数类型
            default_value=2.0,
            description="突破阈值（标准差倍数）",
            min_value=1.0,
            max_value=5.0,
            category="technical",
        ),
        StrategyParameter(
            name="stop_loss",
            type=ParameterType.FLOAT,
            default_value=0.02,
            description="止损比例",
            min_value=0.01,
            max_value=0.1,
            category="risk",
        ),
    ],
    # ... 其他配置
)

参数类型系统

Fire 支持以下参数类型,前端会自动渲染对应的UI组件:

参数类型 Python枚举 前端UI 步长 精度 使用场景
整数 ParameterType.INTEGER Slider/InputNumber 1 0 周期、数量、索引
浮点数 ParameterType.FLOAT Slider/InputNumber 0.01 2 比率、百分比、价格
布尔 ParameterType.BOOLEAN Switch - - 开关标志
字符串 ParameterType.STRING Input - - 文本输入
枚举 ParameterType.STRING + allowed_values Select - - 选项列表

参数类型选择指南

参数命名模式 推荐类型 示例 原因
*_period, *_days, lookback_* INTEGER fast_period=12 K线数量/天数必须是整数
*_threshold, *_ratio, *_factor FLOAT breakout_threshold=2.0 比率值通常是小数
stop_*, take_* FLOAT stop_loss=0.03 止损止盈为价格变化百分比
*_multiplier FLOAT volume_multiplier=1.5 乘数可以是小数
max_*, min_* (数量) INTEGER max_positions=5 数量限制通常是整数
use_*, enable_* BOOLEAN use_ema=True 开关标志
*_type, *_mode STRING + allowed_values ma_type="EMA" 固定选项列表

前端自动化处理

定义好参数类型后,前端会自动处理:

  1. 类型区分: INTEGER vs FLOAT 在UI层面完全区分
  2. 步长控制: 整数 step=1 (不允许小数), 浮点数 step=0.01
  3. 精度显示: 整数显示 “12”, 浮点数显示 “0.03”
  4. 验证规则: 根据 min_value 和 max_value 自动验证

无需前端额外配置,只需在后端正确定义 ParameterType 即可!

运行时配置示例

# 运行时配置 (从前端或API传入)
strategy_engine = StrategyEngine(
    session_id="session_123",
    strategies_config={
        "MomentumBreakoutStrategy": {
            "lookback_period": 20,  # 整数参数
            "breakout_threshold": 2.0,  # 浮点数参数
            "stop_loss": 0.02  # 浮点数参数
        }
    }
)

参考文档

支持的环境变量

基于实际的 settings.py：

环境变量 说明 默认值
LOG_LEVEL 日志级别 INFO
REDIS_HOST Redis缓存主机 localhost
REDIS_PORT Redis端口 6379
DEBUG 调试模式 False

已实现策略

Fire 系统当前提供 8 个生产级策略，涵盖动量、趋势、波动率和成交量四大类别。所有策略基于 pandas-ta 库实现，确保指标计算的准确性和可靠性。

策略分类总览

分类 策略ID 核心指标 预热要求 风险特征
Momentum (动量)        
  MomentumBreakoutStrategy BBANDS, SMA, STD 50条 中等风险
  RSIMomentumStrategy RSI, SMA 50条 中等风险
Trend (趋势)        
  MACDStrategy MACD (EMA派生) 17条 中等风险
  MACrossStrategy SMA/EMA 50条 低风险
  TrendChannelStrategy ATR, SMA 50条 高风险
Mean Reversion (均值回归)        
  BollingerBandsStrategy BBANDS (SMA + STD) 50条 中等风险
Volume (成交量)        
  VolumeSpikeStrategy SMA (成交量) 30条 高风险
Volatility (波动率)        
  LowVolatilityStrategy ATR 50条 低风险

动量策略 (Momentum)

MomentumBreakoutStrategy - 动量突破策略

原理: 基于统计学的突破检测，结合成交量确认捕捉趋势形成初期。

核心逻辑:

  1. 计算价格均值和标准差（通过 BBANDS 指标）
  2. 检测价格突破 ± N倍标准差区间
  3. 成交量确认（当前成交量 ≥ 均值 × 阈值）
  4. 动态置信度评分（突破强度）

关键参数:

信号生成:

适用场景: 趋势启动、突破交易

RSIMomentumStrategy - RSI 动量策略

原理: 结合 RSI 超买超卖区域和趋势过滤器，仅在顺势方向交易。

核心逻辑:

  1. 计算 RSI 识别超买超卖
  2. 使用 SMA 判断趋势方向
  3. 仅在 RSI 信号与趋势一致时交易
  4. 动态置信度（RSI 距离阈值）

关键参数:

信号生成:

适用场景: 趋势市场、避免逆势交易

趋势策略 (Trend)

MACDStrategy - MACD 策略

原理: MACD 是趋势跟踪动量指标，通过快慢 EMA 差值判断买卖时机。

核心逻辑:

  1. 计算 MACD 线（快 EMA - 慢 EMA）
  2. 计算信号线（MACD 的 EMA）
  3. 检测真正的交叉事件（非静态比较）
  4. 强度阈值过滤（柱状图绝对值 > 0.01）

关键参数:

信号生成:

适用场景: 趋势跟踪、多时间框架

MACrossStrategy - 均线交叉策略

原理: 经典趋势跟踪，通过快慢均线交叉判断趋势开始和结束。

核心逻辑:

  1. 计算快速和慢速移动平均线
  2. 检测真正的交叉事件
  3. 基于分离度评估信号强度
  4. 支持 SMA 和 EMA 两种类型

关键参数:

信号生成:

分离度与置信度:

适用场景: 趋势确认、多周期组合

TrendChannelStrategy - 趋势通道策略

原理: 基于 ATR 构建动态通道，价格突破通道时进场，使用 ATR 计算止损止盈。

核心逻辑:

  1. 构建通道（中轨 = SMA，上/下轨 = 中轨 ± multiplier × ATR）
  2. 检测通道突破
  3. 动态止损止盈（基于 ATR 倍数）
  4. 持仓跟踪和平仓管理

关键参数:

信号生成:

适用场景: 突破交易、自适应波动

均值回归策略 (Mean Reversion)

BollingerBandsStrategy - 布林带策略

原理: 基于布林带的均值回归，在价格触及上下轨时交易。

核心逻辑:

  1. 计算布林带（中轨、上轨、下轨）
  2. 检测价格触及边界
  3. 均值回归信号生成
  4. 带宽判断市场状态

关键参数:

信号生成:

适用场景: 震荡市场、区间交易

成交量策略 (Volume)

VolumeSpikeStrategy - 成交量异动策略

原理: 基于”成交量先于价格”理论，捕捉成交量激增并结合价格方向。

核心逻辑:

  1. 计算成交量移动平均作为基准
  2. 检测成交量激增（当前 > MA × 阈值）
  3. 结合价格变化确认方向
  4. 双重确认机制（成交量 + 价格）

关键参数:

信号生成:

适用场景: 事件驱动、主力进场

波动率策略 (Volatility)

LowVolatilityStrategy - 低波动率策略

原理: 基于”低波动率异常”理论，选择波动率较低的股票构建投资组合。

核心逻辑:

  1. 使用 ATR/价格比率标准化波动率
  2. 对波动率进行历史排名
  3. 流动性过滤（成交量阈值）
  4. 动态置信度（波动率 + 排名）

关键参数:

信号生成:

适用场景: 稳健投资、组合构建

数据缺失处理

所有策略必须遵循 市场数据缺失处理规范：

核心原则:

  1. 数据完整性优先: 凑齐 N 个有效数据点，而非 N 个时间单位
  2. 向前查找: 遇到缺失继续往前推，直到凑齐
  3. 区分粒度: 分钟级、小时级、天级有不同处理规则

处理规则:

相关文档

架构文档

开发指南

具体策略文档