3.1 FreqTrade-策略结构详解

3.1 策略结构详解

学习目标

• 理解Freqtrade策略的基本结构和组件
• 掌握IStrategy接口的核心方法和属性
• 了解策略的生命周期和执行流程
• 学会策略开发的最佳实践

---

3.1.1 策略基础架构

IStrategy接口概述

Freqtrade中的所有策略都必须继承自IStrategy基类：

from freqtrade.strategy import IStrategy
from pandas import DataFrame
import talib.abstract as ta
from freqtrade.strategy import DecimalParameter, IntParameter, CategoricalParameter

class MyFirstStrategy(IStrategy):
    """
    这是一个示例策略，展示Freqtrade策略的基本结构
    """

    # 策略元信息
    INTERFACE_VERSION = 3

    # 策略参数
    minimal_roi = {
        "60": 0.01,
        "30": 0.02,
        "0": 0.04
    }

    stoploss = -0.10
    timeframe = '5m'

    # 可选参数
    can_short: bool = False
    startup_candle_count: int = 30
    process_only_new_candles = True
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # 必需的核心方法
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """添加技术指标"""
        pass

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

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

策略组件详解

1. 元信息和版本控制

class VersionedStrategy(IStrategy):
    """
    策略版本控制和元信息
    """

    # 接口版本（必需）
    INTERFACE_VERSION = 3

    # 策略元信息
    __author__ = "Your Name"
    __version__ = "1.0.0" 
    __date__ = "2024-03-15"
    __description__ = "基于RSI和MACD的趋势跟随策略"

    # 兼容性标记
    minimal_roi = {"0": 0.1}
    stoploss = -0.1
    timeframe = '5m'

2. 策略参数定义

def strategy_parameters_example():
    """
    策略参数类型和定义方式
    """

    class ParameterizedStrategy(IStrategy):
        # 基础参数
        minimal_roi = {
            "60": 0.01,    # 60分钟后最小1%收益
            "30": 0.02,    # 30分钟后最小2%收益  
            "20": 0.03,    # 20分钟后最小3%收益
            "0": 0.04      # 立即4%收益
        }

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

        # 交易控制参数
        can_short = False         # 不允许做空
        startup_candle_count = 50 # 启动所需K线数
        max_open_trades = 5       # 最大开仓数

        # 信号控制
        use_exit_signal = True                # 使用出场信号
        exit_profit_only = False             # 不仅在盈利时出场
        exit_profit_offset = 0.0             # 出场盈利偏移
        ignore_roi_if_entry_signal = False   # ROI时不忽略入场信号

        # 订单控制
        order_types = {
            'entry': 'limit',
            'exit': 'limit', 
            'stoploss': 'market',
            'stoploss_on_exchange': False,
            'stoploss_on_exchange_interval': 60
        }

        # 超参数定义（用于优化）
        buy_rsi_threshold = IntParameter(20, 40, default=30, space="buy")
        sell_rsi_threshold = IntParameter(60, 80, default=70, space="sell")
        roi_space = DecimalParameter(0.01, 0.10, default=0.04, space="roi")

    return ParameterizedStrategy

---

3.1.2 核心方法详解

必需方法实现

1. populate_indicators方法

def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    """
    添加技术指标到数据框

    Args:
        dataframe: OHLCV数据框
        metadata: 包含币对信息的字典

    Returns:
        包含技术指标的数据框
    """

    # 基础指标
    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['ema_26'] = ta.EMA(dataframe, timeperiod=26)

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

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

    # 布林带
    bollinger = qtpylib.bollinger_bands(
        qtpylib.typical_price(dataframe), window=20, stds=2
    )
    dataframe['bb_lowerband'] = bollinger['lower']
    dataframe['bb_middleband'] = bollinger['mid']
    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(20).mean()

    # 自定义指标
    dataframe['price_change'] = dataframe['close'].pct_change()
    dataframe['volatility'] = dataframe['price_change'].rolling(20).std()

    return dataframe

2. populate_entry_trend方法

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

    通过设置'enter_long'和'enter_short'列来标记入场信号
    """

    # 做多入场条件
    dataframe.loc[
        (
            # 趋势条件：短期均线上穿长期均线
            (dataframe['ema_12'] > dataframe['ema_26']) &
            (dataframe['ema_12'].shift(1) <= dataframe['ema_26'].shift(1)) &

            # 动量条件：RSI从超卖区域上升
            (dataframe['rsi'] > 30) &
            (dataframe['rsi'].shift(1) <= 30) &

            # MACD确认
            (dataframe['macd'] > dataframe['macdsignal']) &

            # 成交量确认
            (dataframe['volume'] > dataframe['volume_sma']) &

            # 价格位置
            (dataframe['close'] > dataframe['bb_middleband'])
        ),
        'enter_long'] = 1

    # 做空入场条件（如果允许做空）
    if self.can_short:
        dataframe.loc[
            (
                # 趋势条件：短期均线下穿长期均线
                (dataframe['ema_12'] < dataframe['ema_26']) &
                (dataframe['ema_12'].shift(1) >= dataframe['ema_26'].shift(1)) &

                # 动量条件：RSI从超买区域下降
                (dataframe['rsi'] < 70) &
                (dataframe['rsi'].shift(1) >= 70) &

                # MACD确认
                (dataframe['macd'] < dataframe['macdsignal']) &

                # 成交量确认
                (dataframe['volume'] > dataframe['volume_sma']) &

                # 价格位置
                (dataframe['close'] < dataframe['bb_middleband'])
            ),
            'enter_short'] = 1

    return dataframe

3. populate_exit_trend方法

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

    通过设置'exit_long'和'exit_short'列来标记出场信号
    """

    # 做多出场条件
    dataframe.loc[
        (
            # 趋势反转：短期均线下穿长期均线
            (dataframe['ema_12'] < dataframe['ema_26']) &
            (dataframe['ema_12'].shift(1) >= dataframe['ema_26'].shift(1))
        ) |
        (
            # 超买出场：RSI进入超买区
            (dataframe['rsi'] > 70)
        ) |
        (
            # MACD背离
            (dataframe['macd'] < dataframe['macdsignal']) &
            (dataframe['macd'].shift(1) >= dataframe['macdsignal'].shift(1))
        ),
        'exit_long'] = 1

    # 做空出场条件
    if self.can_short:
        dataframe.loc[
            (
                # 趋势反转：短期均线上穿长期均线
                (dataframe['ema_12'] > dataframe['ema_26']) &
                (dataframe['ema_12'].shift(1) <= dataframe['ema_26'].shift(1))
            ) |
            (
                # 超卖出场：RSI进入超卖区
                (dataframe['rsi'] < 30)
            ) |
            (
                # MACD背离
                (dataframe['macd'] > dataframe['macdsignal']) &
                (dataframe['macd'].shift(1) <= dataframe['macdsignal'].shift(1))
            ),
            'exit_short'] = 1

    return dataframe

---

3.1.3 高级方法和回调

可选的高级方法

1. 自定义止损

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

    Returns:
        float: 止损价格比例（负数）
    """

    # 获取交易持续时间（分钟）
    trade_duration = (current_time - trade.open_date_utc).total_seconds() / 60

    # 时间基础的动态止损
    if trade_duration < 60:
        # 前1小时：固定10%止损
        return -0.10
    elif trade_duration < 180:
        # 1-3小时：减少到8%
        return -0.08
    elif trade_duration < 360:
        # 3-6小时：减少到5%
        return -0.05
    else:
        # 6小时后：减少到3%
        return -0.03

def trailing_stoploss_example(self, pair: str, trade: 'Trade', current_time: 'datetime',
                             current_rate: float, current_profit: float, **kwargs) -> float:
    """
    跟踪止损示例
    """

    # 如果盈利超过5%，启用跟踪止损
    if current_profit > 0.05:
        # 跟踪止损设为盈利的50%
        return current_profit * -0.5

    # 否则使用固定止损
    return -0.10

2. 自定义仓位大小

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:
    """
    自定义仓位大小

    Returns:
        float: 实际投入的资金量
    """

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

    # 基于RSI调整仓位大小
    rsi = current_candle['rsi']

    if rsi < 25:
        # RSI极度超卖，增加仓位
        return proposed_stake * 1.5
    elif rsi < 35:
        # RSI超卖，正常仓位
        return proposed_stake
    elif rsi > 75:
        # RSI极度超买，减少仓位
        return proposed_stake * 0.5
    elif rsi > 65:
        # RSI超买，小幅减少仓位
        return proposed_stake * 0.8
    else:
        # 中性区域，正常仓位
        return proposed_stake

def volatility_based_sizing(self, pair: str, **kwargs) -> float:
    """
    基于波动率的仓位调整
    """
    dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
    current_candle = dataframe.iloc[-1].squeeze()

    # 获取近期波动率
    volatility = current_candle['volatility']

    # 根据波动率调整仓位
    if volatility > 0.05:  # 高波动
        return kwargs['proposed_stake'] * 0.5
    elif volatility > 0.03:  # 中等波动
        return kwargs['proposed_stake'] * 0.8
    else:  # 低波动
        return kwargs['proposed_stake'] * 1.2

3. 入场和出场确认

def confirm_trade_entry(self, pair: str, order_type: str, amount: float,
                       rate: float, time_in_force: str, current_time: 'datetime',
                       entry_tag: str, side: str, **kwargs) -> bool:
    """
    入场确认 - 在实际下单前的最后检查

    Returns:
        bool: True表示确认入场，False表示取消
    """

    # 获取最新数据
    dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
    last_candle = dataframe.iloc[-1].squeeze()

    # 流动性检查
    if last_candle['volume'] < last_candle['volume_sma'] * 0.5:
        # 成交量过低，取消交易
        return False

    # 价格剧烈波动检查
    price_change = abs(last_candle['price_change'])
    if price_change > 0.05:  # 单根K线涨跌超过5%
        # 价格波动过大，等待稳定
        return False

    # 市场时间检查（避免在低流动性时段交易）
    current_hour = current_time.hour
    if 22 <= current_hour or current_hour <= 6:  # UTC时间晚10点到早6点
        return False

    return True

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:
    """
    出场确认
    """

    # 如果是止损出场，立即执行
    if exit_reason in ['stop_loss', 'stoploss_on_exchange']:
        return True

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

    # 如果当前盈利很小且RSI显示可能反转，暂缓出场
    current_profit = trade.calc_profit_ratio(rate)
    if (current_profit < 0.02 and 
        ((trade.is_open and last_candle['rsi'] < 35) or
         (not trade.is_open and last_candle['rsi'] > 65))):
        return False

    return True

---

3.1.4 策略生命周期

执行流程

def strategy_lifecycle_explanation():
    """
    策略执行生命周期说明
    """

    lifecycle_stages = {
        "1. 初始化阶段": {
            "description": "策略类实例化",
            "actions": [
                "读取策略参数",
                "验证配置",
                "初始化数据提供者"
            ]
        },

        "2. 数据准备阶段": {
            "description": "准备历史数据",
            "actions": [
                "加载startup_candle_count数量的历史K线",
                "调用populate_indicators添加指标",
                "数据框准备完成"
            ]
        },

        "3. 交易循环阶段": {
            "description": "主要交易逻辑执行",
            "actions": [
                "获取新的K线数据",
                "更新技术指标",
                "调用populate_entry_trend检查入场信号",
                "调用populate_exit_trend检查出场信号",
                "执行风险管理检查",
                "生成交易订单"
            ]
        },

        "4. 订单执行阶段": {
            "description": "处理订单和交易",
            "actions": [
                "调用confirm_trade_entry确认入场",
                "调用custom_stake_amount计算仓位",
                "提交订单到交易所",
                "监控订单状态",
                "调用confirm_trade_exit确认出场"
            ]
        },

        "5. 风险管理阶段": {
            "description": "持续风险监控",
            "actions": [
                "调用custom_stoploss检查止损",
                "检查ROI目标",
                "应用保护机制",
                "更新交易状态"
            ]
        }
    }

    return lifecycle_stages

# 策略执行时序图
execution_flow = """
Time    │ Action
──────────────────────────────────────
00:00   │ 加载历史数据
00:01   │ populate_indicators()
00:02   │ ├── 计算SMA, EMA
00:03   │ ├── 计算RSI, MACD
00:04   │ └── 计算布林带
00:05   │ populate_entry_trend()
00:06   │ ├── 检查趋势条件
00:07   │ ├── 检查动量指标
00:08   │ └── 生成入场信号
00:09   │ populate_exit_trend()
00:10   │ ├── 检查出场条件
00:11   │ └── 生成出场信号
00:12   │ confirm_trade_entry()
00:13   │ custom_stake_amount()
00:14   │ 提交买单
00:15   │ 监控订单状态
...     │ ...
01:00   │ custom_stoploss()
...     │ ...
02:00   │ confirm_trade_exit()
02:01   │ 提交卖单
"""

数据流转

def data_flow_in_strategy():
    """
    策略中的数据流转
    """

    data_flow = {
        "输入数据": {
            "OHLCV": "开高低收成交量数据",
            "metadata": "币对信息字典",
            "trade": "当前交易对象（某些方法中）",
            "current_time": "当前时间"
        },

        "处理过程": {
            "1. 原始数据": "交易所提供的OHLCV数据",
            "2. 技术指标": "通过populate_indicators添加",
            "3. 信号生成": "通过populate_entry/exit_trend生成",
            "4. 信号过滤": "通过confirm方法过滤",
            "5. 订单执行": "最终生成交易订单"
        },

        "输出结果": {
            "entry_signals": "入场信号列",
            "exit_signals": "出场信号列", 
            "orders": "实际交易订单",
            "trades": "交易记录"
        }
    }

    return data_flow

---

3.1.5 策略开发最佳实践

代码结构和组织

1. 策略模板

"""
策略开发模板 - 包含所有必要组件
"""

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

logger = logging.getLogger(__name__)

class TemplateStrategy(IStrategy):
    """
    策略模板 - 包含完整的策略结构
    """

    # =============================================================================
    # 策略元信息
    # =============================================================================
    INTERFACE_VERSION = 3
    __author__ = "Your Name"
    __version__ = "1.0.0"
    __strategy_name__ = "TemplateStrategy"

    # =============================================================================
    # 策略参数
    # =============================================================================

    # 基础参数
    minimal_roi = {
        "60": 0.01,
        "30": 0.02, 
        "20": 0.03,
        "0": 0.04
    }

    stoploss = -0.10
    timeframe = '5m'

    # 交易控制
    can_short = False
    startup_candle_count = 50
    process_only_new_candles = True
    use_exit_signal = True

    # 超参数定义
    buy_rsi_threshold = IntParameter(20, 40, default=30, space="buy")
    sell_rsi_threshold = IntParameter(60, 80, default=70, space="sell")

    # =============================================================================
    # 辅助方法
    # =============================================================================

    def informative_pairs(self) -> List[Tuple[str, str]]:
        """
        定义额外需要的数据对
        """
        return [
            ("BTC/USDT", "1h"),  # 比特币1小时数据作为大盘参考
        ]

    def get_btc_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        获取比特币趋势作为市场情绪参考
        """
        btc_info = self.dp.get_pair_dataframe("BTC/USDT", "1h")
        btc_info['btc_sma_20'] = ta.SMA(btc_info, timeperiod=20)
        btc_info['btc_trend'] = btc_info['close'] > btc_info['btc_sma_20']

        # 将比特币趋势信息合并到主数据框
        dataframe = merge_informative_pair(
            dataframe, btc_info, self.timeframe, "1h", ffill=True
        )

        return dataframe

    # =============================================================================
    # 核心策略方法
    # =============================================================================

    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['ema_26'] = ta.EMA(dataframe, timeperiod=26)

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

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

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

        # 获取比特币趋势
        dataframe = self.get_btc_trend(dataframe, metadata)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        入场信号定义
        """

        dataframe.loc[
            (
                # 趋势过滤
                (dataframe['close'] > dataframe['sma_20']) &
                (dataframe['sma_20'] > dataframe['sma_50']) &

                # 动量确认
                (dataframe['rsi'] > self.buy_rsi_threshold.value) &
                (dataframe['rsi'] < 50) &

                # MACD确认
                (dataframe['macd'] > dataframe['macdsignal']) &

                # 成交量确认
                (dataframe['volume'] > dataframe['volume_sma']) &

                # 大盘情绪（比特币趋势）
                (dataframe['btc_trend_1h'] == True)
            ),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        出场信号定义
        """

        dataframe.loc[
            (
                # RSI超买
                (dataframe['rsi'] > self.sell_rsi_threshold.value)
            ) |
            (
                # 趋势反转
                (dataframe['close'] < dataframe['sma_20']) &
                (dataframe['close'].shift(1) >= dataframe['sma_20'].shift(1))
            ) |
            (
                # MACD死叉
                (dataframe['macd'] < dataframe['macdsignal']) &
                (dataframe['macd'].shift(1) >= dataframe['macdsignal'].shift(1))
            ),
            'exit_long'] = 1

        return dataframe

调试和测试

1. 策略调试工具

def debug_strategy_signals(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
    """
    调试策略信号的辅助方法
    """

    # 添加调试信息
    dataframe['debug_trend'] = (
        (dataframe['close'] > dataframe['sma_20']) &
        (dataframe['sma_20'] > dataframe['sma_50'])
    )

    dataframe['debug_momentum'] = (
        (dataframe['rsi'] > 30) & (dataframe['rsi'] < 50)
    )

    dataframe['debug_volume'] = (
        dataframe['volume'] > dataframe['volume_sma']
    )

    # 计算信号强度
    dataframe['signal_strength'] = (
        dataframe['debug_trend'].astype(int) +
        dataframe['debug_momentum'].astype(int) +
        dataframe['debug_volume'].astype(int)
    )

    # 记录调试信息
    if metadata['pair'] == 'BTC/USDT':  # 只为BTC/USDT记录日志
        last_candle = dataframe.iloc[-1]
        logger.info(f"调试信息 - 趋势:{last_candle['debug_trend']}, "
                   f"动量:{last_candle['debug_momentum']}, "
                   f"成交量:{last_candle['debug_volume']}, "
                   f"信号强度:{last_candle['signal_strength']}")

    return dataframe

2. 单元测试框架

import unittest
from unittest.mock import Mock, patch
import pandas as pd

class TestTemplateStrategy(unittest.TestCase):
    """
    策略单元测试
    """

    def setUp(self):
        """测试初始化"""
        self.strategy = TemplateStrategy()

        # 创建测试数据
        self.test_data = pd.DataFrame({
            'open': [100, 101, 102, 103, 104],
            'high': [101, 102, 103, 104, 105],
            'low': [99, 100, 101, 102, 103],
            'close': [100.5, 101.5, 102.5, 103.5, 104.5],
            'volume': [1000, 1100, 1200, 1300, 1400]
        })

    def test_populate_indicators(self):
        """测试指标计算"""
        result = self.strategy.populate_indicators(self.test_data, {'pair': 'BTC/USDT'})

        # 验证指标存在
        self.assertIn('sma_20', result.columns)
        self.assertIn('rsi', result.columns)
        self.assertIn('macd', result.columns)

        # 验证指标值合理性
        self.assertFalse(result['sma_20'].isna().all())

    def test_entry_signals(self):
        """测试入场信号"""
        # 先计算指标
        dataframe = self.strategy.populate_indicators(self.test_data, {'pair': 'BTC/USDT'})

        # 手动设置一些指标值以触发信号
        dataframe.loc[0, 'rsi'] = 35
        dataframe.loc[0, 'close'] = 105
        dataframe.loc[0, 'sma_20'] = 100

        result = self.strategy.populate_entry_trend(dataframe, {'pair': 'BTC/USDT'})

        # 验证信号列存在
        self.assertIn('enter_long', result.columns)

    def test_parameter_ranges(self):
        """测试参数范围"""
        self.assertGreaterEqual(self.strategy.buy_rsi_threshold.value, 20)
        self.assertLessEqual(self.strategy.buy_rsi_threshold.value, 40)

if __name__ == '__main__':
    unittest.main()

---

本节总结

策略结构是Freqtrade量化交易的核心，理解和掌握策略的各个组件对于开发有效的交易策略至关重要。

关键要点:

1. 继承IStrategy接口并实现必需方法
2. 合理设置策略参数和超参数
3. 使用高级方法实现复杂逻辑
4. 理解策略执行生命周期
5. 遵循最佳实践提高代码质量

最佳实践:

• 模块化代码结构，便于维护和测试
• 使用有意义的变量和方法命名
• 添加充分的注释和文档
• 实现调试和测试功能
• 考虑策略的可扩展性

---

实践练习

练习3.1.1: 基础策略框架

# 创建一个完整的策略框架，包含:
# 1. 基础策略结构
# 2. 必需方法实现
# 3. 参数定义
# 4. 调试功能

练习3.1.2: 高级方法实现

# 实现以下高级方法:
# 1. 自定义止损逻辑
# 2. 动态仓位大小
# 3. 入场出场确认
# 4. 多时间框架分析

练习3.1.3: 策略测试

# 为策略添加测试功能:
# 1. 单元测试
# 2. 集成测试  
# 3. 性能测试
# 4. 回归测试

---

下一节：3.2 编写简单策略