安全与风控

本章详细介绍 Freqtrade 交易中的安全管理与风险控制措施，涵盖资金管理规则、最大回撤控制、仓位大小计算、黑天鹅应对策略以及日志审计，帮助读者构建安全的量化交易体系。

资金管理规则

核心资金管理原则

{
    "max_open_trades": 3,
    "stake_amount": 50,
    "tradable_balance_ratio": 0.95,
    "dry_run_wallet": 10000,
    "stake_currency": "USDT"
}

凯利公式计算仓位

from freqtrade.strategy import IStrategy

class KellyPositionStrategy(IStrategy):
    """
    使用凯利公式计算最优仓位
    凯利公式: f = (p * b - q) / b
    其中 f = 最优仓位比例, p = 胜率, q = 败率, b = 盈亏比
    """

    def custom_stake_amount(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_stake: float,
        min_stake: float,
        max_stake: float,
        entry_tag: Optional[str],
        side: str,
        **kwargs
    ) -> float:
        # 获取历史交易数据
        trades = Trade.get_trades_proxy(pair=pair, is_open=False)

        if len(trades) < 20:
            # 数据不足时使用默认仓位
            return proposed_stake * 0.5

        # 计算胜率和盈亏比
        wins = [t for t in trades if t.calc_profit_ratio() > 0]
        losses = [t for t in trades if t.calc_profit_ratio() <= 0]

        win_rate = len(wins) / len(trades)
        avg_win = sum(t.calc_profit_ratio() for t in wins) / len(wins) if wins else 0
        avg_loss = abs(sum(t.calc_profit_ratio() for t in losses) / len(losses)) if losses else 0

        # 盈亏比
        profit_loss_ratio = avg_win / avg_loss if avg_loss > 0 else 1

        # 凯利公式
        kelly_fraction = (win_rate * profit_loss_ratio - (1 - win_rate)) / profit_loss_ratio

        # 保守化：使用半凯利
        kelly_fraction = max(0, min(kelly_fraction * 0.5, 0.25))

        return proposed_stake * kelly_fraction

最大回撤控制

配置级回撤控制

{
    "max_open_trades": 3,
    "stoploss": -0.10,
    "trailing_stop": true,
    "trailing_stop_positive": -0.02,
    "trailing_stop_positive_offset": 0.03,
    "trailing_only_offset_is_reached": true
}

代码级回撤保护

from freqtrade.strategy import IStrategy
from freqtrade.persistence import Trade

class DrawdownProtectedStrategy(IStrategy):
    """
    回撤保护策略：当总体回撤超过阈值时降低风险
    """
    # 最大允许回撤
    max_drawdown = 0.15  # 15%

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        return dataframe

    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time: datetime,
        entry_tag: Optional[str],
        side: str,
        **kwargs
    ) -> bool:
        # 计算当前总回撤
        current_drawdown = self._get_current_drawdown()

        # 如果当前回撤超过限制，拒绝开仓
        if current_drawdown > self.max_drawdown:
            self.logger.warning(
                f"当前回撤 {current_drawdown:.2%} 超过限制 {self.max_drawdown:.2%}，拒绝开仓"
            )
            return False

        # 回撤接近限制时，减少仓位
        if current_drawdown > self.max_drawdown * 0.8:
            self.logger.info(f"回撤警告: {current_drawdown:.2%}，谨慎开仓")

        return True

    def _get_current_drawdown(self) -> float:
        """计算当前账户回撤"""
        try:
            # 获取总利润
            trades = Trade.get_trades_proxy(is_open=False)
            if not trades:
                return 0.0

            profits = []
            for trade in trades:
                profit = trade.calc_profit_ratio()
                profits.append(profit)

            # 计算累计利润和峰值
            cumulative = 0
            peak = 0
            max_drawdown = 0

            for p in profits:
                cumulative += p
                if cumulative > peak:
                    peak = cumulative
                drawdown = (peak - cumulative) / (1 + peak) if peak > 0 else 0
                max_drawdown = max(max_drawdown, drawdown)

            return max_drawdown

        except Exception as e:
            self.logger.error(f"计算回撤失败: {e}")
            return 0.0

每日亏损限额

def confirm_trade_entry(self, pair, ...) -> bool:
    """每日亏损限额控制"""
    today = current_time.date()

    # 获取今日已平仓交易的盈亏
    trades_today = Trade.get_trades_proxy(is_open=False)
    daily_pnl = sum(
        t.calc_profit_ratio()
        for t in trades_today
        if t.close_date and t.close_date.date() == today
    )

    # 每日亏损超过 5% 时停止交易
    max_daily_loss = -0.05
    if daily_pnl < max_daily_loss:
        self.logger.warning(
            f"今日亏损 {daily_pnl:.2%} 超过限额 {max_daily_loss:.2%}，停止开仓"
        )
        return False

    return True

仓位大小计算

基于波动率的仓位调整

def custom_stake_amount(
    self,
    pair: str,
    current_time: datetime,
    current_rate: float,
    proposed_stake: float,
    min_stake: float,
    max_stake: float,
    entry_tag: Optional[str],
    side: str,
    **kwargs
) -> float:
    """基于 ATR 的动态仓位"""
    dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

    if dataframe is None or len(dataframe) < 30:
        return proposed_stake * 0.5

    last_candle = dataframe.iloc[-1]
    atr = last_candle.get("atr", 0)
    close = last_candle.get("close", current_rate)

    if atr == 0 or close == 0:
        return proposed_stake * 0.5

    # ATR 百分比
    atr_pct = atr / close

    # 目标风险为单笔亏损不超过总资金的 1%
    risk_per_trade = 0.01
    account_balance = self.wallets.get_total_stake_amount()

    # 计算仓位：风险 / (ATR% * 止损%)
    stoploss_pct = abs(self.stoploss) if self.stoploss else 0.05
    position_size = (account_balance * risk_per_trade) / (atr_pct * stoploss_pct)

    # 限制在配置范围内
    position_size = min(position_size, proposed_stake)
    position_size = max(position_size, min_stake)

    return position_size

仓位大小计算对比

黑天鹅应对

极端行情防护

def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    # 检测异常波动
    returns = dataframe["close"].pct_change()
    dataframe["returns_std"] = returns.rolling(20).std()

    # Z-score 异常检测
    latest_return = returns.iloc[-1] if len(returns) > 0 else 0
    mean_return = returns.mean()
    std_return = returns.std()

    if std_return > 0:
        z_score = (latest_return - mean_return) / std_return

        # 极端波动时触发保护
        if abs(z_score) > 3:
            self.logger.warning(
                f"检测到极端行情! Z-score: {z_score:.2f}, "
                f"涨跌幅: {latest_return:.4f}"
            )
            # 可以在此处发送告警

    # 黑天鹅检测：单根 K 线跌幅超过 10%
    dataframe["black_swan"] = (dataframe["close"] / dataframe["open"] - 1) < -0.10

    return dataframe

def confirm_trade_entry(self, pair, ...) -> bool:
    """黑天鹅期间暂停开仓"""
    dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)

    if dataframe is not None and len(dataframe) > 10:
        recent_swans = dataframe["black_swan"].tail(10).sum()
        if recent_swans > 0:
            self.logger.warning(f"近期检测到黑天鹅事件，暂停 {pair} 的开仓")
            return False

    return True

熔断机制

class CircuitBreakerStrategy(IStrategy):
    """
    熔断机制：
    - 连续 N 笔亏损后暂停交易
    - 暂停一段时间后自动恢复
    """
    consecutive_loss_limit = 3  # 连续亏损笔数限制
    cooldown_period = 24        # 冷却时间（小时）

    def __init__(self, config: dict) -> None:
        super().__init__(config)
        self.consecutive_losses = {}
        self.cooldown_until = {}

    def confirm_trade_exit(
        self,
        pair: str,
        trade: Trade,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        exit_reason: str,
        current_time: datetime,
        **kwargs
    ) -> bool:
        """在平仓时记录盈亏"""
        profit_ratio = trade.calc_profit_ratio(rate)

        if profit_ratio < 0:
            # 记录亏损
            self.consecutive_losses[pair] = self.consecutive_losses.get(pair, 0) + 1

            # 触发熔断
            if self.consecutive_losses[pair] >= self.consecutive_loss_limit:
                cooldown_end = current_time + timedelta(hours=self.cooldown_period)
                self.cooldown_until[pair] = cooldown_end
                self.logger.warning(
                    f"{pair} 连续 {self.consecutive_losses[pair]} 笔亏损，"
                    f"熔断至 {cooldown_end}"
                )
        else:
            # 盈利时重置计数
            self.consecutive_losses[pair] = 0

        return True

    def confirm_trade_entry(self, pair, ...) -> bool:
        """检查是否在熔断期"""
        cooldown_end = self.cooldown_until.get(pair)

        if cooldown_end and current_time < cooldown_end:
            remaining = (cooldown_end - current_time).total_seconds() / 3600
            self.logger.info(
                f"{pair} 仍在熔断期，剩余 {remaining:.1f} 小时"
            )
            return False

        return True

紧急停止脚本

#!/bin/bash
# emergency_stop.sh
# 紧急停止所有交易操作

echo "========================================"
echo "  紧急停止脚本"
echo "  时间: $(date)"
echo "========================================"

# 1. 通过 API 停止机器人
echo "[1/3] 通过 API 停止交易..."
curl -s -X POST "http://127.0.0.1:8080/api/v1/stop"     -u "admin:password"     -H "Content-Type: application/json"

# 2. 强制平仓所有持仓
echo "[2/3] 强制平仓所有持仓..."
curl -s -X POST "http://127.0.0.1:8080/api/v1/forcesell"     -u "admin:password"     -H "Content-Type: application/json"     -d '{"tradeid": "all"}'

# 3. 停止 Docker 容器
echo "[3/3] 停止 Docker 容器..."
docker compose -f /opt/freqtrade/docker-compose.yml down

echo "========================================"
echo "  紧急停止完成"
echo "========================================"

日志审计

审计日志配置

{
    "loglevel": "info",
    "logfile": "user_data/logs/freqtrade.log",
    "api_server": {
        "enabled": true,
        "verbosity": "info"
    }
}

关键审计事件

Freqtrade 会记录以下关键事件到日志：

# 开仓事件
2024-01-15 08:00:00 - INFO - Buy signal found: BTC/USDT, rate=42000.0, tag=rsi_oversold
2024-01-15 08:00:05 - INFO - Buy order filled: BTC/USDT, rate=42150.0, amount=0.002

# 平仓事件
2024-01-15 14:00:00 - INFO - Sell signal found: BTC/USDT, rate=43500.0, profit=3.2%
2024-01-15 14:00:05 - INFO - Sell order filled: BTC/USDT, rate=43450.0, profit=3.1%

# 止损事件
2024-01-15 10:30:00 - WARNING - Stoploss triggered: BTC/USDT, rate=37800.0, loss=-10.0%

# 风控事件
2024-01-15 09:00:00 - WARNING - Locked pair: ETH/USDT due to 3 consecutive losses
2024-01-15 12:00:00 - INFO - Unlocked pair: ETH/USDT, lock expired
2024-01-15 15:00:00 - ERROR - Exchange API error: binance, timeout after 20s

# 配置变更
2024-01-15 08:00:00 - INFO - Reloading configuration...
2024-01-15 08:00:01 - INFO - Configuration reloaded successfully

审计日志分析脚本

#!/bin/bash
# audit_report.sh
# 生成每日审计报告

LOGFILE="user_data/logs/freqtrade.log"
DATE=$(date +"%Y-%m-%d")
REPORT="user_data/logs/audit_${DATE}.txt"

echo "===== Freqtrade 审计报告 =====" > $REPORT
echo "日期: $DATE" >> $REPORT
echo "" >> $REPORT

# 统计交易数
echo "--- 交易统计 ---" >> $REPORT
grep "Buy order filled" $LOGFILE | grep "$DATE" | wc -l | xargs echo "开仓次数:" >> $REPORT
grep "Sell order filled" $LOGFILE | grep "$DATE" | wc -l | xargs echo "平仓次数:" >> $REPORT

# 统计止损
echo "" >> $REPORT
echo "--- 止损统计 ---" >> $REPORT
grep "Stoploss triggered" $LOGFILE | grep "$DATE" | wc -l | xargs echo "止损次数:" >> $REPORT

# 统计错误
echo "" >> $REPORT
echo "--- 错误统计 ---" >> $REPORT
grep "ERROR" $LOGFILE | grep "$DATE" | wc -l | xargs echo "错误次数:" >> $REPORT

# 统计警告
echo "" >> $REPORT
echo "--- 警告统计 ---" >> $REPORT
grep "WARNING" $LOGFILE | grep "$DATE" | wc -l | xargs echo "警告次数:" >> $REPORT

# 列出最近的错误
echo "" >> $REPORT
echo "--- 最近的错误 ---" >> $REPORT
grep "ERROR" $LOGFILE | grep "$DATE" | tail -5 >> $REPORT

echo "" >> $REPORT
echo "报告生成时间: $(date)" >> $REPORT

echo "审计报告已生成: $REPORT"

安全配置参考

{
    "telegram": {
        "enabled": true,
        "token": "${TELEGRAM_TOKEN}",
        "chat_id": "${TELEGRAM_CHAT_ID}",
        "notification_parameters": {
            "warning": true,
            "startup": true,
            "protections": true
        }
    },
    "api_server": {
        "enabled": true,
        "listen_ip_address": "127.0.0.1",
        "listen_port": 8080,
        "username": "admin",
        "password": "${API_PASSWORD}",
        "jwt_secret_key": "${JWT_SECRET}"
    }
}

风控清单