Fire 策略与风控

市场状态

Director 将市场分为 6 种状态，每种状态对应不同的仓位约束：

状态	英文标识	允许做多	允许做空	典型场景
强多	`strong_long`	100%	0%	明确上升趋势
弱多	`weak_long`	50%	0%	上升趋势减弱
高位纠缠	`high_consolidation`	30%	30%	高位震荡整理
低位纠缠	`low_consolidation`	30%	30%	低位震荡整理
弱空	`weak_short`	0%	50%	下降趋势减弱
强空	`strong_short`	0%	100%	明确下降趋势

策略体系

Fire 采用 Director-Executor 双层策略架构：大周期定方向，小周期找时机。

MarketData → Director (4h) → 市场状态 → PositionManager
                                            ↓
MarketData → Executor (15m) → 入场信号 → PositionManager → 最终决策

角色	策略数量	时间框架	职责
Director	4	大周期 (4h, 1d)	判断市场状态和方向
Executor	4	小周期 (15m, 1h)	寻找精确入场时机

Director 策略

策略 ID	核心指标	详细文档
`MACDDirector`	MACD 柱状图	MACD 策略
`MACrossDirector`	快慢均线	均线交叉策略
`TrendChannelDirector`	ATR, Donchian	趋势通道策略
`LowVolatilityDirector`	ATR	低波动率策略

Executor 策略

策略 ID	核心指标	详细文档
`RSIMomentumExecutor`	RSI, EMA	RSI 动量策略
`MomentumBreakoutExecutor`	BBANDS, Volume	动量突破策略
`VolumeSpikeExecutor`	Volume MA	成交量异动策略

策略注册系统

策略通过 @register 装饰器自动注册：

from core.trading.mtf.base_strategy import DirectorStrategy
from core.trading.mtf.registry import register

@register(
    strategy_id="macd_director",
    display_name="MACD 方向判定",
    category="trend",
    role="director",
    parameters=[
        {"name": "fast_period", "type": "int", "default": 12},
        {"name": "slow_period", "type": "int", "default": 26},
    ]
)
class MACDDirector(DirectorStrategy):
    def analyze(self, ohlcv: pd.DataFrame) -> StrategyVote:
        pass

策略元数据

参数	类型	说明
`category`	`StrategyCategory`	策略分类（MOMENTUM/TREND_FOLLOWING 等）
`description`	`str`	策略描述
`min_history_bars`	`int`	最少历史数据条数，默认 100
`supports_realtime`	`bool`	是否支持实时交易，默认 True
`supports_backtest`	`bool`	是否支持回测，默认 True

策略生命周期

创建 → 初始化 → 预热（积累历史数据）→ 运行（生成信号）→ 停止

预热阶段：策略通过 get_required_history_length() 声明所需数据量，预热期间不生成信号
运行阶段：处理实时数据，通过 _set_signal() 生成交易信号

指标计算

Fire 集成 pandas-ta 库，通过 PandasTAWrapper 提供常用技术指标：

趋势类：SMA、EMA、MACD、ADX
动量类：RSI、Stochastic
波动类：ATR、Bollinger Bands
成交量类：OBV、VWAP

两级缓存机制：L1 全局缓存（跨策略共享）+ L2 实例缓存，避免重复计算。

仓位管理

PositionManager 约束层叠

PositionManager 按优先级顺序应用 8 层约束：

优先级	约束类型	触发条件
1	止损	价格触及止损线
2	Director 反转	Director 方向反转
3	止盈	价格触及止盈目标
4	移动止损	价格回撤超过阈值
5	Director 约束	超过 max_long/max_short
6	做空模式	short_mode=disabled
7	互斥约束	互斥标的已有仓位
8	SizingPolicy	接受建议仓位

flowchart TD
    Input[输入信号] --> SL{止损?}
    SL -->|是| Force[强制平仓]
    SL -->|否| Rev{Director 反转?}
    Rev -->|是| Clear[清空仓位]
    Rev -->|否| TP{止盈?}
    TP -->|是| Reduce[减仓]
    TP -->|否| Dir{Director 约束?}
    Dir -->|是| Limit[限制仓位]
    Dir -->|否| Accept[接受建议仓位]

    Force --> Output[最终决策]
    Clear --> Output
    Reduce --> Output
    Limit --> Output
    Accept --> Output

    style SL fill:#ff6b6b
    style Force fill:#ff6b6b

状态持久化

PositionManager 状态持久化在 Redis，键格式：pm_state:{session_id}:{symbol}

风险控制

四级风险体系

层级	类型	监控指标	触发动作
1	仓位风险	单股占比 <30%	限制加仓
2	杠杆风险	总杠杆率 <3x	降低杠杆
3	回撤风险	最大回撤 <15%	减仓或停止
4	集中度风险	行业集中度 <50%	分散配置

RiskEngine（预交易检查）

检查项	默认阈值
单股持仓比例	≤10%
单笔订单比例	≤5%
最小现金比例	≥10%
最大杠杆	≤1x

RiskGuard（实时监控）

指标	默认阈值	告警响应
回撤	≤15%	WARNING → ERROR → CRITICAL
日损失	≤2%	限制新仓位 → 自动平仓
持仓集中度	≤30%	记录日志

风险配置示例

risk_config = {
    "max_position_ratio": 0.1,
    "max_drawdown": 0.15,
    "allowed_symbols": ["AAPL", "GOOGL", "MSFT"],
    "stop_loss_ratio": 0.05,
    "take_profit_ratio": 0.15,
}

策略开发指南

自定义 Director 策略

from core.trading.mtf.base_strategy import DirectorStrategy
from core.trading.mtf.registry import register

@register(
    strategy_id="custom_director",
    display_name="自定义方向判定",
    category="custom",
    role="director",
    parameters=[{"name": "period", "type": "int", "default": 20}]
)
class CustomDirector(DirectorStrategy):
    def analyze(self, ohlcv: pd.DataFrame) -> StrategyVote:
        momentum = self._calculate_momentum(ohlcv)
        return StrategyVote(
            strategy_id=self.strategy_id,
            momentum=momentum,
            confidence=0.8,
        )

自定义 Executor 策略

from core.trading.mtf.base_strategy import ExecutorStrategy
from core.trading.mtf.registry import register

@register(
    strategy_id="custom_executor",
    display_name="自定义入场判定",
    category="custom",
    role="executor",
)
class CustomExecutor(ExecutorStrategy):
    def analyze(self, ohlcv: pd.DataFrame) -> StrategyVote:
        target_ratio = self._calculate_entry_signal(ohlcv)
        return StrategyVote(
            strategy_id=self.strategy_id,
            momentum=target_ratio,
            confidence=0.75,
        )

测试策略

测试文件位于 backend/tests/unit/strategies/，验证：策略初始化、信号生成、预热管理、边界条件。

REST API

方法	路径	说明
GET	`/api/v1/strategies/mtf/metadata`	获取所有策略元数据
GET	`/api/v1/strategies/mtf/metadata/{strategy_id}`	获取指定策略元数据

文档	内容
系统概览	系统简介、技术栈、核心概念
工程架构	分层架构、目录结构、数据层设计
交易架构	DEP 核心架构、FlowRuntime、交易数据流
市场数据规范	数据缺失处理规则

Fire Quantitative Trading System

基于现代技术栈的量化交易平台，支持股票交易、策略回测、风险管理等功能

Fire 策略与风控

市场状态

策略体系

Director 策略

Executor 策略

策略注册系统

策略元数据

策略生命周期

指标计算

仓位管理

PositionManager 约束层叠

状态持久化

风险控制

四级风险体系

RiskEngine（预交易检查）

RiskGuard（实时监控）

风险配置示例

策略开发指南

自定义 Director 策略

自定义 Executor 策略

测试策略

REST API

相关文档

具体策略文档

Director 策略

Executor 策略