文章目录

1. 为什么风控必须独立出来？
2. 第一步：在系统里为风控预留“插口”
3. 第二步：实现最常用的风控机制——最大回撤限制
4. 第三步：改造 Backtester，使风控接入主流程
5. 第四步：在 main.py 中绘制三条曲线进行对比
6. 收尾：风控是系统的“生存开关”

在前两篇文章中，我们完成了两个关键目标：

• 构建了一个最小可运行的量化交易系统
• 掌握了将策略拆解为 Feature / Signal / Position 三层结构的方法，像搭积木一样灵活组合

此时的系统已不再是一个孤立脚本，而是一辆真正能够上路行驶的“车”。

但随之而来的问题也浮现了：

这辆车确实能跑，可你敢在崎岖山路上踩满油门吗？

复杂策略的一大特征就是：

盈利时收益惊人，崩溃时损失惨重。

• 在趋势明朗的行情中，可能大幅跑赢大盘
• 但在极端市场环境下，一周内的回撤就可能吞噬掉你三年积累的本金

一个既能带来高收益、又随时可能导致破产的策略，并非资产，而是

一枚定时炸弹。

因此，本文的核心任务是为系统添加一个至关重要的组件：

风控模块（Risk Manager）——给你的策略系上安全带。

一、为何要将风控独立成单独模块？

很多人第一反应可能是：

“我直接在策略代码里加个 if 判断不就行了？”

on_bar

例如：

• 若账户回撤超过10%，立即清仓
• 若波动率过高，则禁止加仓

逻辑看似合理，但从系统架构角度看，这种做法属于典型的职责混淆。

我们需要明确两个核心角色的分工：

策略（Strategy） 的职责是：
利用数据寻找 Alpha，回答的是——
“在当前市场条件下，我希望持有多大的仓位？”

风控（Risk Manager） 的职责是：
管理系统的生存概率，回答的是——
“基于当前风险水平，我允许你持有多大的仓位？”

如果把这两者混在同一段代码中，最终结果只有一个：

你无法分辨自己是在主动进攻，还是被动防御。

理想的设计应是一条清晰的数据流水线：

策略 → 风控 → 最终仓位 → 交易执行

1. 策略输出“期望仓位”（proposed_weight）
2. 风控模块对其进行审核、调整或否决，生成“最终仓位”（target_weight）
3. 回测器或执行引擎仅根据最终确认的仓位发出指令

本文的目标，就是补全这条关键链条。

二、第一步：为风控模块预留系统接口

从功能定位来看，风控本质上是一个后置过滤器（post-processor）：

• 它不主动生成仓位建议
• 只对策略已经提出的仓位进行干预和修正

因此，我们为其设计一个简洁且边界清晰的接口：

• 输入：策略提议的仓位 + 当前账户状态
• 输出：经过风控调整后的实际执行仓位

新建文件：

quant/risk.py

首先定义一个基类：

# quant/risk.py
from typing import Dict, Any

class RiskManager:
    """
    风控模块的基类。
    职责：依据策略提议与当前投资组合状态，调整或否决目标仓位。
    """
    
    def adjust_position(
        self,
        proposed_weight: float,
        portfolio_state: Dict[str, Any],
    ) -> float:
        """
        调整目标仓位。

        Args:
            proposed_weight (float): 策略提议的目标仓位权重（范围：0~1）
            portfolio_state (Dict[str, Any]): 当前投资组合状态，
                示例：{"equity_curve": [...], "date": "2023-01-01"}

        Returns:
            float: 经风控调整后的最终目标仓位权重。
        """
        raise NotImplementedError

这个设计有意避免引入具体金融术语，也不让风控知晓其所面对的具体策略类型。

它只需关注两件事：

1. 策略想做什么（proposed_weight）
2. 当前账户处于何种状态（portfolio_state）

这些信息已足够支撑其做出决策。

三、第二步：实现常用风控功能——最大回撤控制

我们从一个最基础但极其实用的风控规则入手：

当账户净值从历史最高点回落超过 X% 时，强制清仓并进入冷静期。

同时加入两个贴近实战的小细节：

• 触发后启动冷静期（cooldown）：在此期间无论策略信号多强，一律禁止建仓，仓位锁定为0
• 冷静期结束后重置高水位线：接受本次亏损事实，以新的净值起点重新计算未来回撤

这套机制正是许多专业机构所采用的基础风控逻辑。

继续在同一个文件中扩展：

quant/risk.py

class MaxDrawdownRiskManager(RiskManager):
    """
    带“冷静期”与“重置”机制的最大回撤风控模块。
    
    - 当净值从高水位线下跌超过 max_drawdown_pct 时：
      1. 清空所有仓位
      2. 进入 cooldown_days 天的冷静期（暂停交易）
      3. 将当前净值设为新的高水位线
    """
    
    def __init__(self, max_drawdown_pct: float = 0.15, cooldown_days: int = 20):
        if not 0 < max_drawdown_pct < 1:
            raise ValueError("max_drawdown_pct 必须介于 0 和 1 之间")

self.max_drawdown_pct = max_drawdown_pct
self.cooldown_days = cooldown_days
self.high_water_mark = None
self.cooldown_counter = 0

def adjust_position(
    self,
    proposed_weight: float,
    portfolio_state: Dict[str, Any],
) -> float:
    equity_curve = portfolio_state.get("equity_curve", [])
    
    # 若无历史净值数据，直接通过策略建议
    if not equity_curve:
        return proposed_weight

    current_equity = equity_curve[-1]

    # 只有在非冷静期时才更新高水位线
    if self.cooldown_counter == 0:
        if self.high_water_mark is None or current_equity > self.high_water_mark:
            self.high_water_mark = current_equity

    # 处于冷静期内：强制空仓，并递减冷静天数
    if self.cooldown_counter > 0:
        self.cooldown_counter -= 1
        return 0.0

    # 计算当前回撤幅度
    drawdown = (self.high_water_mark - current_equity) / self.high_water_mark

    # 触发最大回撤风控条件
    if drawdown > self.max_drawdown_pct:
        print(f"--- RISK MANAGER TRIGGERED ---")
        print(f"Date: {portfolio_state.get('date', 'N/A')}, "
              f"Drawdown: {drawdown:.2%}, Equity: {current_equity:.2f}")
        print(f"Action: Liquidating position, starting {self.cooldown_days}-day cooldown, "
              f"and resetting high-water mark.")

        # 启动三步操作：
        # 1. 设定冷静期
        self.cooldown_counter = self.cooldown_days
        # 2. 将当前净值设为新的高水位线
        self.high_water_mark = current_equity
        # 3. 强制清仓，返回零仓位
        return 0.0

    # 未触发风控机制，允许执行原策略建议
    return proposed_weight

你可以将这个类视作一位极其严格的风控监管员。他每天持续监控账户的净值表现：一旦发现亏损触及预设阈值，便会立即强制平仓，并启动一段固定长度的“冷静期”。在此期间，无论市场如何波动，交易行为都被完全禁止。待冷静期结束后，才允许重新参与交易，但所有回撤计算将以新的净值起点为准。

这一机制虽然逻辑简洁，却具备极强的实战适应性，足以支撑多数实盘场景下的基础风控需求。

四、第三步：集成风控模块至回测系统主流程

接下来的目标明确：在不修改原有策略逻辑与数据处理模块的前提下，仅对回测器进行轻量级扩展，实现风控系统的无缝接入。

这相当于在现有回测架构中新增一块可插拔的功能组件：

for 每个 bar:
    策略 → 提出目标仓位
    回测器 → 按这个仓位算股票数量、计算交易、更新现金和持仓

我们需要在这个标准回测循环的关键节点插入风控判断环节：

策略提出仓位 → 风控调整仓位 → 回测器执行

具体操作是修改以下文件：

quant/backtester.py

为其中的

run

方法添加一个新的可选参数：

risk_manager

并在每次执行交易决策前调用该风控模块进行校验。

# quant/backtester.py
import numpy as np
import pandas as pd
from typing import Optional
from .data import DataFeed
from .risk import RiskManager

class Backtester:
    def __init__(self, fee_rate: float = 0.0005, slippage: float = 0.0005):
        self.fee_rate = fee_rate
        self.slippage = slippage

    def run(
        self,
        data_feed: DataFeed,
        strategy,
        initial_state,
        initial_capital: float = 100_000,
        risk_manager: Optional[RiskManager] = None,  # <-- 新增风控插槽
    ):
        cash = initial_capital
        position_shares = 0.0
        equity_curve = []
        dates = []
        trades = []
        state = initial_state
        last_price = None

equity_df, trades_df = run_backtest(strategy_config)

Strategy

RiskManager

Backtester

五、第四步：在 main.py 中实现三组策略对比实验

quant/main.py

# quant/main.py
import matplotlib.pyplot as plt
from .backtester import Backtester
from .data import YFinanceDataFeed
from .risk import MaxDrawdownRiskManager
from .strategy.modular_strategy import (
    AdvancedStrategy,
    AdvancedStrategyState,
    MomentumSignal,
    MovingAverageFeature,
    SimplePositionSizer,
    VolatilityFeature,
    VolatilityFilter,
)

def main():
    symbol = "AAPL"
    start_date = "2018-01-01"
    end_date = "2024-01-01"
    initial_capital = 100_000

    # --- 1. 纯动量策略 ---
    print("--- 运行: 仅动量 ---")

# --- 2. 动量 + 波动率过滤 ---
print("\n--- 运行: 动量 + 波动率过滤 ---")
feed2 = YFinanceDataFeed(symbol=symbol, start=start_date, end=end_date)
strategy2 = AdvancedStrategy(
    features={
        "ma50": MovingAverageFeature(window=50),
        "vol20": VolatilityFeature(window=20),
    },
    signals={
        "momentum": MomentumSignal(ma_feature_name="ma50"),
        "vol_filter": VolatilityFilter(vol_feature_name="vol20", threshold=2.0),
    },
    position_sizer=SimplePositionSizer(
        momentum_signal_name="momentum",
        vol_filter_name="vol_filter",
        step=0.05,
    ),
)
equity2, _ = Backtester().run(
    data_feed=feed2,
    strategy=strategy2,
    initial_state=AdvancedStrategyState(),
    initial_capital=initial_capital,
)
print(f"最终权益: {equity2['equity'].iloc[-1]:.2f}")

# --- 1. 基础动量策略 ---
feed1 = YFinanceDataFeed(symbol=symbol, start=start_date, end=end_date)
strategy1 = AdvancedStrategy(
    features={"ma50": MovingAverageFeature(window=50)},
    signals={"momentum": MomentumSignal(ma_feature_name="ma50")},
    position_sizer=SimplePositionSizer(momentum_signal_name="momentum", step=0.05),
)
equity1, _ = Backtester().run(
    data_feed=feed1,
    strategy=strategy1,
    initial_state=AdvancedStrategyState(),
    initial_capital=initial_capital,
)
print(f"最终权益: {equity1['equity'].iloc[-1]:.2f}")

# --- 3. 动量 + 波动率过滤 + 最大回撤风控 ---
print("\n--- 运行: 动量 + 波动率过滤 + 最大回撤风控 ---")
feed3 = YFinanceDataFeed(symbol=symbol, start=start_date, end=end_date)
risk_manager = MaxDrawdownRiskManager(max_drawdown_pct=0.10, cooldown_days=20)
equity3, _ = Backtester().run(
    data_feed=feed3,
    strategy=strategy2,  # 使用与策略2相同的策略逻辑，仅增加风控模块
    initial_state=AdvancedStrategyState(),
    initial_capital=initial_capital,
    risk_manager=risk_manager,
)
print(f"最终权益: {equity3['equity'].iloc[-1]:.2f}")

# --- 4. 绘图对比 ---
plt.figure(figsize=(12, 8))
equity1["equity"].plot(label="Momentum Only", legend=True)
equity2["equity"].plot(label="Momentum + Vol Filter", legend=True)
equity3["equity"].plot(label="Momentum + Vol Filter + Risk Mgmt", legend=True)
plt.title(f"Strategy Comparison on {symbol}")
plt.xlabel("Date")
plt.ylabel("Equity")
plt.grid(True)
plt.tight_layout()

plt.show()

if __name__ == "__main__":

运行 main() 函数（特别是在使用 uv 环境的情况下）：

uv run python -m quant.main

你将观察到三条走势风格迥异的曲线，分别代表三种不同的风控策略路径：

纯动量策略

• 净值曲线更为陡峭，上涨时显得极具爆发力
• 但相应的回撤也非常剧烈
• 表面上看“收益惊人”，可一旦市场急转直下，情绪容易崩溃，很可能在最低点恐慌性清仓离场

动量 + 波动率过滤

• 当市场波动剧烈、行情躁动时，系统自动降低仓位以应对不确定性
• 整体曲线相较前者略显平滑
• 然而，在极端下跌行情中，仍可能出现从高点快速连续下滑的情形，难以完全规避风险

动量 + 波动率过滤 + 最大回撤风控

• 一旦累计亏损触及预设的心理底线（例如 10%），系统会立即启动保护机制，强制减仓或暂停交易
• 在急剧下跌阶段，你会看到曲线进入一段“横盘”状态——这正是系统的冷静期
• 虽然最终总收益未必最高，但整体走势更加稳健，尾部风险被显著压缩

不妨认真问自己一个关键问题：

在真实的交易环境中，你更愿意持有哪一条曲线？

六、总结：风控是系统存续的“生命开关”

至此，你的“迷你量化系统”已构建起三大核心支柱：

DataFeed

Strategy

RiskManager

这套结构带来的优势在于：

• 无需改动回测框架，即可自由尝试多种策略组合
• 保持策略逻辑不变的前提下，灵活插拔不同风控模块进行对比测试
• 清晰区分问题来源：究竟是策略本身失效，还是风控设置过松或过严所致

下一篇内容将基于此系统架构，深入探讨以下议题：

• 如何科学评估一个策略的优劣？
• 年化收益率、波动率、最大回撤等核心指标的实际意义
• Sharpe比率、Calmar比率等“性价比”类指标应如何解读？
• 参数敏感性分析、时间切片检验与样本外验证的方法论
• 识别“回测中的艺术品”与“现实中可持续运行的策略”之间的本质差异

届时，你的角色将不再局限于“编写策略”的执行者，而是真正具备能力去

审判策略