3.2 FreqTrade编写简单策略

3.2 编写简单策略

学习目标

• 从零开始编写第一个完整的交易策略
• 实现基于经典技术指标的策略逻辑
• 理解策略参数调优的基本方法
• 掌握策略测试和验证流程

---

3.2.1 双均线交叉策略

策略原理

双均线交叉是最经典的趋势跟随策略之一：

• 金叉: 短期均线上穿长期均线，产生买入信号
• 死叉: 短期均线下穿长期均线，产生卖出信号

完整代码实现

# user_data/strategies/DualMAStrategy.py

from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
from pandas import DataFrame
import talib.abstract as ta
from technical import qtpylib
import logging

logger = logging.getLogger(__name__)

class DualMAStrategy(IStrategy):
    """
    双均线交叉策略

    策略逻辑:
    1. 计算短期和长期移动平均线
    2. 短期均线上穿长期均线时买入
    3. 短期均线下穿长期均线时卖出
    4. 添加成交量过滤提高信号质量
    """

    # 策略基本信息
    INTERFACE_VERSION = 3

    # 基础参数
    minimal_roi = {
        "60": 0.01,    # 60分钟后1%收益就卖出
        "30": 0.02,    # 30分钟后2%收益就卖出
        "0": 0.04      # 立即4%收益就卖出
    }

    stoploss = -0.10           # 10%止损
    timeframe = '5m'           # 5分钟K线

    # 策略控制参数
    can_short = False          # 不做空
    startup_candle_count = 60  # 启动需要60根K线
    process_only_new_candles = True
    use_exit_signal = True

    # 超参数定义（用于优化）
    short_ma_period = IntParameter(5, 20, default=10, space="buy")
    long_ma_period = IntParameter(21, 60, default=30, space="buy")
    volume_check = DecimalParameter(0.5, 2.0, default=1.0, space="buy")

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        计算技术指标
        """

        # 计算移动平均线
        dataframe['ma_short'] = ta.SMA(dataframe, timeperiod=self.short_ma_period.value)
        dataframe['ma_long'] = ta.SMA(dataframe, timeperiod=self.long_ma_period.value)

        # 计算成交量均线
        dataframe['volume_ma'] = dataframe['volume'].rolling(window=20).mean()

        # 计算价格变化率（用于过滤异常波动）
        dataframe['price_change'] = dataframe['close'].pct_change()

        # 添加调试信息
        dataframe['ma_diff'] = dataframe['ma_short'] - dataframe['ma_long']
        dataframe['ma_diff_pct'] = (dataframe['ma_short'] / dataframe['ma_long'] - 1) * 100

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        定义买入条件
        """

        # 基础条件：短期均线上穿长期均线（金叉）
        golden_cross = (
            (dataframe['ma_short'] > dataframe['ma_long']) &
            (dataframe['ma_short'].shift(1) <= dataframe['ma_long'].shift(1))
        )

        # 趋势过滤：确保是真正的上升趋势
        uptrend_filter = (
            (dataframe['ma_short'] > dataframe['ma_short'].shift(3)) &  # 短期均线在上升
            (dataframe['ma_long'] > dataframe['ma_long'].shift(5))      # 长期均线也在上升
        )

        # 成交量过滤：确保有足够的交易量
        volume_filter = (
            dataframe['volume'] > (dataframe['volume_ma'] * self.volume_check.value)
        )

        # 价格波动过滤：避免在异常波动时买入
        volatility_filter = (
            abs(dataframe['price_change']) < 0.05  # 单根K线涨跌幅不超过5%
        )

        # 组合所有条件
        dataframe.loc[
            (
                golden_cross &
                uptrend_filter &
                volume_filter &
                volatility_filter
            ),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        定义卖出条件
        """

        # 基础条件：短期均线下穿长期均线（死叉）
        death_cross = (
            (dataframe['ma_short'] < dataframe['ma_long']) &
            (dataframe['ma_short'].shift(1) >= dataframe['ma_long'].shift(1))
        )

        # 趋势确认：确保是真正的下降趋势
        downtrend_confirm = (
            (dataframe['ma_short'] < dataframe['ma_short'].shift(3)) &  # 短期均线在下降
            (dataframe['close'] < dataframe['ma_short'])                # 价格跌破短期均线
        )

        # 组合条件
        dataframe.loc[
            (
                death_cross |  # 死叉信号
                downtrend_confirm  # 或者趋势确认下降
            ),
            'exit_long'] = 1

        return dataframe

    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: 'datetime',
                       current_rate: float, current_profit: float, **kwargs) -> float:
        """
        自定义动态止损
        """

        # 如果已经盈利，使用跟踪止损
        if current_profit > 0.02:  # 盈利超过2%
            return current_profit * -0.5  # 回撤一半利润时止损

        # 否则使用固定止损
        return self.stoploss

策略创建脚本

#!/bin/bash
# create_dual_ma_strategy.sh

echo "创建双均线策略..."

# 确保策略目录存在
mkdir -p user_data/strategies

# 创建策略文件
cat > user_data/strategies/DualMAStrategy.py << 'EOF'
# 在这里插入上面的完整策略代码
EOF

echo "双均线策略创建完成: user_data/strategies/DualMAStrategy.py"

# 验证策略语法
python -m py_compile user_data/strategies/DualMAStrategy.py
if [ $? -eq 0 ]; then
    echo "策略语法检查通过"
else
    echo "策略语法有误，请检查代码"
fi

---

3.2.2 RSI超买超卖策略

策略原理

RSI（相对强弱指标）超买超卖策略是典型的均值回归策略：

• 超卖买入: RSI < 30时买入
• 超买卖出: RSI > 70时卖出

完整代码实现

# user_data/strategies/RSIStrategy.py

from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
from pandas import DataFrame
import talib.abstract as ta
from technical import qtpylib
import logging

logger = logging.getLogger(__name__)

class RSIStrategy(IStrategy):
    """
    RSI超买超卖策略

    策略逻辑:
    1. 使用RSI指标识别超买超卖
    2. RSI从超卖区域向上突破时买入
    3. RSI进入超买区域时卖出
    4. 添加趋势过滤避免逆势交易
    """

    INTERFACE_VERSION = 3

    # 基础参数
    minimal_roi = {
        "120": 0.01,   # 2小时后1%收益
        "60": 0.02,    # 1小时后2%收益
        "30": 0.03,    # 30分钟后3%收益
        "0": 0.05      # 立即5%收益
    }

    stoploss = -0.08           # 8%止损
    timeframe = '15m'          # 15分钟K线

    can_short = False
    startup_candle_count = 50
    process_only_new_candles = True
    use_exit_signal = True

    # 超参数
    rsi_period = IntParameter(10, 20, default=14, space="buy")
    rsi_oversold = IntParameter(20, 35, default=30, space="buy")
    rsi_overbought = IntParameter(65, 80, default=70, space="sell")
    trend_period = IntParameter(20, 50, default=30, space="buy")

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        计算技术指标
        """

        # RSI指标
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)

        # 趋势过滤：简单移动平均线
        dataframe['sma_trend'] = ta.SMA(dataframe, timeperiod=self.trend_period.value)

        # RSI的移动平均（平滑RSI信号）
        dataframe['rsi_ma'] = dataframe['rsi'].rolling(window=3).mean()

        # 价格动量
        dataframe['momentum'] = dataframe['close'] / dataframe['close'].shift(5) - 1

        # 成交量相对强度
        dataframe['volume_sma'] = dataframe['volume'].rolling(window=20).mean()
        dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']

        # 计算支撑阻力位（简化版）
        dataframe['resistance'] = dataframe['high'].rolling(window=20).max()
        dataframe['support'] = dataframe['low'].rolling(window=20).min()

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        定义买入条件
        """

        # RSI超卖条件
        rsi_oversold_condition = (
            (dataframe['rsi'] > self.rsi_oversold.value) &          # RSI从超卖区上升
            (dataframe['rsi'].shift(1) <= self.rsi_oversold.value)   # 前一根K线RSI在超卖区
        )

        # 或者RSI快速反弹
        rsi_bounce_condition = (
            (dataframe['rsi'] < 40) &                               # 当前RSI仍较低
            (dataframe['rsi'] > dataframe['rsi'].shift(1) + 3) &    # RSI快速上升
            (dataframe['rsi'].shift(1) < self.rsi_oversold.value)   # 前期确实超卖
        )

        # 趋势过滤：只在上升趋势中买入
        trend_filter = (
            dataframe['close'] > dataframe['sma_trend']  # 价格在趋势线上方
        )

        # 动量确认
        momentum_filter = (
            dataframe['momentum'] > -0.02  # 近期跌幅不超过2%
        )

        # 成交量确认
        volume_filter = (
            dataframe['volume_ratio'] > 0.8  # 成交量不能太低
        )

        # 距离支撑位不远（增加反弹概率）
        support_filter = (
            (dataframe['close'] - dataframe['support']) / dataframe['support'] < 0.05
        )

        dataframe.loc[
            (
                (rsi_oversold_condition | rsi_bounce_condition) &
                trend_filter &
                momentum_filter &
                volume_filter
            ),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        定义卖出条件
        """

        # RSI超买卖出
        rsi_overbought_exit = (
            dataframe['rsi'] > self.rsi_overbought.value
        )

        # RSI顶背离（价格创新高但RSI没有创新高）
        rsi_divergence = (
            (dataframe['close'] > dataframe['close'].shift(5)) &     # 价格创新高
            (dataframe['rsi'] < dataframe['rsi'].shift(5)) &         # RSI没创新高
            (dataframe['rsi'] > 60)                                  # RSI在高位
        )

        # 跌破趋势线
        trend_break = (
            (dataframe['close'] < dataframe['sma_trend']) &
            (dataframe['close'].shift(1) >= dataframe['sma_trend'].shift(1))
        )

        # 成交量放大的下跌
        volume_dump = (
            (dataframe['close'] < dataframe['close'].shift(1)) &      # 价格下跌
            (dataframe['volume_ratio'] > 1.5) &                      # 放量
            (dataframe['rsi'] < 50)                                   # RSI已经回落
        )

        dataframe.loc[
            (
                rsi_overbought_exit |
                rsi_divergence |
                trend_break |
                volume_dump
            ),
            'exit_long'] = 1

        return dataframe

    def custom_stake_amount(self, pair: str, current_time: 'datetime', current_rate: float,
                          proposed_stake: float, min_stake: float, max_stake: float,
                          leverage: float, entry_tag: str, side: str, **kwargs) -> float:
        """
        基于RSI调整仓位大小
        """

        # 获取当前数据
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        current_candle = dataframe.iloc[-1].squeeze()

        rsi = current_candle['rsi']

        # RSI越低，仓位越大（但有上限）
        if rsi < 25:
            return min(proposed_stake * 1.5, max_stake)
        elif rsi < 30:
            return proposed_stake * 1.2
        elif rsi < 35:
            return proposed_stake
        else:
            return proposed_stake * 0.8

---

3.2.3 布林带回归策略

策略原理

布林带回归策略基于价格向均值回归的假设：

• 下轨买入: 价格触及布林带下轨时买入
• 上轨卖出: 价格触及布林带上轨时卖出

完整代码实现

# user_data/strategies/BollingerBandsStrategy.py

from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
from pandas import DataFrame
import talib.abstract as ta
from technical import qtpylib
import numpy as np

class BollingerBandsStrategy(IStrategy):
    """
    布林带回归策略

    策略逻辑:
    1. 计算布林带上下轨
    2. 价格触及下轨且有反弹迹象时买入
    3. 价格触及上轨或回到中轨时卖出
    4. 使用带宽判断市场波动性
    """

    INTERFACE_VERSION = 3

    minimal_roi = {
        "180": 0.01,
        "60": 0.02,
        "30": 0.03,
        "0": 0.04
    }

    stoploss = -0.06
    timeframe = '15m'

    can_short = False
    startup_candle_count = 40
    process_only_new_candles = True
    use_exit_signal = True

    # 布林带参数
    bb_period = IntParameter(15, 25, default=20, space="buy")
    bb_std = DecimalParameter(1.5, 2.5, default=2.0, space="buy")

    # 回归参数
    mean_reversion_threshold = DecimalParameter(0.02, 0.10, default=0.05, space="buy")

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        计算技术指标
        """

        # 布林带
        bollinger = qtpylib.bollinger_bands(
            qtpylib.typical_price(dataframe), 
            window=self.bb_period.value, 
            stds=self.bb_std.value
        )
        dataframe['bb_lowerband'] = bollinger['lower']
        dataframe['bb_middleband'] = bollinger['mid']
        dataframe['bb_upperband'] = bollinger['upper']

        # 布林带位置（0-1之间，0是下轨，1是上轨）
        dataframe['bb_percent'] = (
            (dataframe['close'] - dataframe['bb_lowerband']) /
            (dataframe['bb_upperband'] - dataframe['bb_lowerband'])
        )

        # 布林带宽度（衡量波动性）
        dataframe['bb_width'] = (
            (dataframe['bb_upperband'] - dataframe['bb_lowerband']) / 
            dataframe['bb_middleband']
        )

        # 价格相对于中轨的偏离度
        dataframe['bb_deviation'] = (
            (dataframe['close'] - dataframe['bb_middleband']) / 
            dataframe['bb_middleband']
        )

        # RSI辅助判断
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)

        # 成交量指标
        dataframe['volume_sma'] = dataframe['volume'].rolling(window=20).mean()

        # 价格动量
        dataframe['price_momentum'] = dataframe['close'].pct_change(periods=3)

        # 计算最近的高低点
        dataframe['local_high'] = dataframe['high'].rolling(window=10).max()
        dataframe['local_low'] = dataframe['low'].rolling(window=10).min()

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        定义买入条件
        """

        # 基础条件：价格接近或触及下轨
        near_lower_band = (
            dataframe['bb_percent'] < 0.2  # 在布林带下端20%区域
        )

        # 反弹确认：价格开始从下轨反弹
        bounce_confirmation = (
            (dataframe['close'] > dataframe['low']) &  # 当前价格高于最低价
            (dataframe['close'] > dataframe['close'].shift(1)) &  # 价格开始上升
            (dataframe['low'] <= dataframe['bb_lowerband'] * 1.01)  # 曾经接近下轨
        )

        # 超卖确认
        oversold_confirmation = (
            dataframe['rsi'] < 35  # RSI超卖
        )

        # 波动性过滤：避免在极低波动时交易
        volatility_filter = (
            dataframe['bb_width'] > 0.02  # 布林带宽度足够
        )

        # 成交量确认
        volume_confirmation = (
            dataframe['volume'] > dataframe['volume_sma'] * 0.8
        )

        # 避免在快速下跌中买入
        momentum_filter = (
            dataframe['price_momentum'] > -0.03  # 3根K线跌幅不超过3%
        )

        dataframe.loc[
            (
                near_lower_band &
                bounce_confirmation &
                oversold_confirmation &
                volatility_filter &
                volume_confirmation &
                momentum_filter
            ),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        定义卖出条件
        """

        # 主要出场：价格接近上轨
        near_upper_band = (
            dataframe['bb_percent'] > 0.8  # 在布林带上端20%区域
        )

        # 回归中轨出场
        return_to_mean = (
            (dataframe['bb_percent'] > 0.6) &  # 已经超过中位
            (dataframe['close'] < dataframe['bb_middleband']) &  # 价格跌破中轨
            (dataframe['close'].shift(1) >= dataframe['bb_middleband'].shift(1))  # 之前在中轨上方
        )

        # RSI超买出场
        rsi_overbought = (
            dataframe['rsi'] > 70
        )

        # 下降动量出场
        negative_momentum = (
            (dataframe['price_momentum'] < -0.02) &  # 负动量
            (dataframe['bb_percent'] > 0.5)  # 且不在下轨附近
        )

        # 波动性收缩出场（可能预示趋势改变）
        volatility_squeeze = (
            (dataframe['bb_width'] < dataframe['bb_width'].shift(5) * 0.8) &  # 带宽收缩
            (dataframe['bb_percent'] > 0.7)  # 且在上方
        )

        dataframe.loc[
            (
                near_upper_band |
                return_to_mean |
                rsi_overbought |
                negative_momentum |
                volatility_squeeze
            ),
            'exit_long'] = 1

        return dataframe

    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: 'datetime',
                       current_rate: float, current_profit: float, **kwargs) -> float:
        """
        动态止损：基于布林带位置调整
        """

        # 获取当前数据
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        current_candle = dataframe.iloc[-1].squeeze()

        bb_percent = current_candle['bb_percent']

        # 如果价格跌破下轨太多，立即止损
        if bb_percent < -0.1:
            return 0.001  # 立即止损

        # 基于位置调整止损
        if bb_percent < 0.2:
            return -0.03  # 在下轨附近，较小止损
        elif bb_percent < 0.5:
            return -0.05  # 中间位置，正常止损
        else:
            return -0.08  # 在上方，较大止损空间

---

3.2.4 策略测试和验证

基础回测

#!/bin/bash
# test_strategies.sh - 策略测试脚本

CONFIG="user_data/config.json"
TIMERANGE="20240101-20240331"

# 测试策略列表
STRATEGIES=(
    "DualMAStrategy"
    "RSIStrategy" 
    "BollingerBandsStrategy"
)

echo "开始测试策略..."

for strategy in "${STRATEGIES[@]}"; do
    echo "================================"
    echo "测试策略: $strategy"
    echo "================================"

    # 快速回测（1个月数据）
    echo "进行快速回测..."
    freqtrade backtesting \
        --config $CONFIG \
        --strategy $strategy \
        --timerange 20240301-20240331 \
        --breakdown day

    echo "$strategy 快速回测完成"
    echo ""
done

echo "所有策略测试完成！"

性能对比分析

# analyze_strategies.py - 策略性能分析脚本

import pandas as pd
import json
from pathlib import Path

def analyze_strategy_performance():
    """
    分析多个策略的回测结果
    """

    results_dir = Path("user_data/backtest_results")
    strategies = ["DualMAStrategy", "RSIStrategy", "BollingerBandsStrategy"]

    performance_data = []

    for strategy in strategies:
        # 查找最新的回测结果文件
        result_files = list(results_dir.glob(f"*{strategy}*.json"))
        if not result_files:
            continue

        latest_file = max(result_files, key=lambda x: x.stat().st_mtime)

        with open(latest_file, 'r') as f:
            data = json.load(f)

        # 提取关键指标
        strategy_result = data.get('strategy', {}).get(strategy, {})

        performance_data.append({
            'strategy': strategy,
            'total_trades': strategy_result.get('total_trades', 0),
            'profit_total': strategy_result.get('profit_total', 0),
            'profit_total_pct': strategy_result.get('profit_total_pct', 0),
            'wins': strategy_result.get('wins', 0),
            'losses': strategy_result.get('losses', 0),
            'win_rate': strategy_result.get('wins', 0) / max(strategy_result.get('total_trades', 1), 1),
            'avg_profit': strategy_result.get('profit_mean', 0),
            'max_drawdown': strategy_result.get('max_drawdown', 0),
            'sharpe_ratio': strategy_result.get('sharpe', 0),
            'calmar_ratio': strategy_result.get('calmar', 0)
        })

    # 创建对比表
    df = pd.DataFrame(performance_data)

    print("策略性能对比:")
    print("=" * 80)
    print(df.to_string(index=False, float_format='%.4f'))

    # 找出最佳策略
    best_profit = df.loc[df['profit_total_pct'].idxmax()]
    best_sharpe = df.loc[df['sharpe_ratio'].idxmax()]
    best_winrate = df.loc[df['win_rate'].idxmax()]

    print("\n最佳策略:")
    print(f"最高收益: {best_profit['strategy']} ({best_profit['profit_total_pct']:.2f}%)")
    print(f"最佳夏普: {best_sharpe['strategy']} ({best_sharpe['sharpe_ratio']:.4f})")
    print(f"最高胜率: {best_winrate['strategy']} ({best_winrate['win_rate']:.2f}%)")

    return df

if __name__ == "__main__":
    analyze_strategy_performance()

策略优化脚本

#!/bin/bash
# optimize_strategies.sh - 策略优化脚本

CONFIG="user_data/config.json"
TIMERANGE="20240101-20240331"
EPOCHS=100

echo "开始策略优化..."

# 优化双均线策略
echo "优化双均线策略..."
freqtrade hyperopt \
    --config $CONFIG \
    --hyperopt-loss SharpeHyperOptLoss \
    --strategy DualMAStrategy \
    --epochs $EPOCHS \
    --spaces buy \
    --timerange $TIMERANGE

# 优化RSI策略
echo "优化RSI策略..."
freqtrade hyperopt \
    --config $CONFIG \
    --hyperopt-loss SortinoHyperOptLoss \
    --strategy RSIStrategy \
    --epochs $EPOCHS \
    --spaces buy sell \
    --timerange $TIMERANGE

# 优化布林带策略
echo "优化布林带策略..."
freqtrade hyperopt \
    --config $CONFIG \
    --hyperopt-loss CalmarHyperOptLoss \
    --strategy BollingerBandsStrategy \
    --epochs $EPOCHS \
    --spaces buy \
    --timerange $TIMERANGE

echo "策略优化完成！"

# 显示优化结果
echo "优化结果总结:"
freqtrade hyperopt-list --best

---

3.2.5 策略改进和扩展

多时间框架策略

# user_data/strategies/MultiTimeframeStrategy.py

from freqtrade.strategy import IStrategy, merge_informative_pair
from pandas import DataFrame
import talib.abstract as ta

class MultiTimeframeStrategy(IStrategy):
    """
    多时间框架策略示例
    结合多个时间框架的信号提高策略效果
    """

    INTERFACE_VERSION = 3

    minimal_roi = {"0": 0.1}
    stoploss = -0.10
    timeframe = '5m'

    startup_candle_count = 100

    def informative_pairs(self):
        """
        定义需要的额外时间框架数据
        """
        return [
            (pair, '1h') for pair in self.dp.current_whitelist()
        ]

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        计算多时间框架指标
        """

        # 主时间框架指标（5分钟）
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
        dataframe['sma_20'] = ta.SMA(dataframe, timeperiod=20)

        # 获取1小时时间框架数据
        informative_1h = self.dp.get_pair_dataframe(
            pair=metadata['pair'], 
            timeframe='1h'
        )

        # 计算1小时指标
        informative_1h['sma_50_1h'] = ta.SMA(informative_1h, timeperiod=50)
        informative_1h['rsi_1h'] = ta.RSI(informative_1h, timeperiod=14)

        # 确定1小时趋势
        informative_1h['trend_1h'] = (
            informative_1h['close'] > informative_1h['sma_50_1h']
        )

        # 合并1小时数据到5分钟数据框
        dataframe = merge_informative_pair(
            dataframe, informative_1h, self.timeframe, '1h', ffill=True
        )

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        多时间框架买入条件
        """

        dataframe.loc[
            (
                # 1小时趋势向上
                (dataframe['trend_1h'] == True) &

                # 1小时RSI不超买
                (dataframe['rsi_1h'] < 70) &

                # 5分钟RSI超卖反弹
                (dataframe['rsi'] > 30) &
                (dataframe['rsi'].shift(1) <= 30) &

                # 5分钟价格在均线上方
                (dataframe['close'] > dataframe['sma_20'])
            ),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        多时间框架卖出条件
        """

        dataframe.loc[
            (
                # 1小时趋势转向或RSI超买
                (dataframe['trend_1h'] == False) |
                (dataframe['rsi_1h'] > 80) |

                # 5分钟RSI超买
                (dataframe['rsi'] > 70)
            ),
            'exit_long'] = 1

        return dataframe

组合策略框架

# user_data/strategies/ComboStrategy.py

from freqtrade.strategy import IStrategy
from pandas import DataFrame
import talib.abstract as ta

class ComboStrategy(IStrategy):
    """
    组合策略：结合多种技术指标的综合策略
    """

    INTERFACE_VERSION = 3

    minimal_roi = {"0": 0.1}
    stoploss = -0.10
    timeframe = '15m'

    startup_candle_count = 60

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        计算综合技术指标
        """

        # 趋势指标
        dataframe['sma_20'] = ta.SMA(dataframe, timeperiod=20)
        dataframe['sma_50'] = ta.SMA(dataframe, timeperiod=50)
        dataframe['ema_12'] = ta.EMA(dataframe, timeperiod=12)

        # 动量指标
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)

        # MACD
        macd = ta.MACD(dataframe)
        dataframe['macd'] = macd['macd']
        dataframe['macdsignal'] = macd['macdsignal']

        # 布林带
        bollinger = qtpylib.bollinger_bands(
            qtpylib.typical_price(dataframe), window=20, stds=2
        )
        dataframe['bb_lowerband'] = bollinger['lower']
        dataframe['bb_upperband'] = bollinger['upper']
        dataframe['bb_percent'] = (
            (dataframe['close'] - dataframe['bb_lowerband']) /
            (dataframe['bb_upperband'] - dataframe['bb_lowerband'])
        )

        # 成交量
        dataframe['volume_sma'] = dataframe['volume'].rolling(window=20).mean()

        return dataframe

    def calculate_signal_strength(self, dataframe: DataFrame) -> DataFrame:
        """
        计算综合信号强度
        """

        # 趋势信号 (0-3分)
        trend_score = 0
        trend_score += (dataframe['close'] > dataframe['sma_20']).astype(int)
        trend_score += (dataframe['sma_20'] > dataframe['sma_50']).astype(int)
        trend_score += (dataframe['ema_12'] > dataframe['sma_20']).astype(int)

        # 动量信号 (0-2分)
        momentum_score = 0
        momentum_score += (dataframe['rsi'] > 30).astype(int) & (dataframe['rsi'] < 70).astype(int)
        momentum_score += (dataframe['macd'] > dataframe['macdsignal']).astype(int)

        # 超卖反弹信号 (0-2分)
        oversold_score = 0
        oversold_score += ((dataframe['rsi'] > 30) & (dataframe['rsi'].shift(1) <= 30)).astype(int)
        oversold_score += (dataframe['bb_percent'] < 0.2).astype(int)

        # 成交量确认 (0-1分)
        volume_score = (dataframe['volume'] > dataframe['volume_sma']).astype(int)

        # 总分 (0-8分)
        dataframe['signal_strength'] = trend_score + momentum_score + oversold_score + volume_score

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        基于信号强度的买入条件
        """

        # 计算信号强度
        dataframe = self.calculate_signal_strength(dataframe)

        # 强信号买入
        dataframe.loc[
            (
                dataframe['signal_strength'] >= 6  # 至少6分信号强度
            ),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        综合卖出条件
        """

        dataframe.loc[
            (
                # RSI超买
                (dataframe['rsi'] > 70) |

                # 趋势反转
                (dataframe['close'] < dataframe['sma_20']) |

                # MACD死叉
                (dataframe['macd'] < dataframe['macdsignal']) |

                # 布林带上轨
                (dataframe['bb_percent'] > 0.8)
            ),
            'exit_long'] = 1

        return dataframe

---

本节总结

通过编写这些简单策略，我们掌握了策略开发的基本流程和技巧。

关键要点:

1. 从简单策略开始，逐步增加复杂性
2. 每个策略都要有明确的逻辑和规则
3. 合理使用技术指标和过滤条件
4. 重视策略测试和性能分析
5. 通过组合和扩展提高策略效果

策略开发步骤:

1. 确定策略逻辑和原理
2. 选择合适的技术指标
3. 定义入场和出场条件
4. 添加过滤和确认机制
5. 实现风险管理功能
6. 进行回测和优化
7. 分析性能和改进

最佳实践:

• 保持策略逻辑简单清晰
• 使用多重确认避免假信号
• 重视风险管理和止损
• 定期回测和验证策略
• 记录策略表现和改进点

---

实践练习

练习3.2.1: 自定义策略开发

# 开发自己的策略，要求包含:
# 1. 明确的交易逻辑
# 2. 至少3个技术指标
# 3. 入场出场条件
# 4. 风险管理机制

练习3.2.2: 策略优化实验

# 对编写的策略进行优化:
# 1. 参数敏感性分析
# 2. 不同市场环境测试
# 3. 组合不同指标
# 4. 改进风险控制

练习3.2.3: 策略性能分析

# 深入分析策略性能:
# 1. 详细回测报告
# 2. 风险收益分析
# 3. 与基准对比
# 4. 改进建议

---

下一节：3.3 干运行测试