Bài viết gần đây

| Chiến Lược Mean Reversion Bot Auto Trading

Được viết bởi thanhdt vào ngày 17/11/2025 lúc 12:04 | 69 lượt xem

Chiến Lược Mean Reversion Bot Python

Mean Reversion (Hồi quy về giá trị trung bình) là một trong những chiến lược giao dịch phổ biến nhất, dựa trên nguyên tắc rằng giá sẽ có xu hướng quay trở lại mức trung bình sau khi lệch xa. Chiến lược này đặc biệt hiệu quả trong thị trường sideways (đi ngang) và có thể tạo ra lợi nhuận ổn định khi được thực hiện đúng cách. Trong bài viết này, chúng ta sẽ xây dựng một bot giao dịch tự động sử dụng chiến lược Mean Reversion với Python.

Tổng quan về Mean Reversion

Mean Reversion là gì?

Mean Reversion là hiện tượng giá của một tài sản có xu hướng quay trở lại mức trung bình (mean) sau khi di chuyển quá xa khỏi nó. Nguyên tắc cơ bản:

  • Giá tạm thời lệch xa khỏi mức trung bình
  • Áp lực mua/bán sẽ đưa giá trở lại mức trung bình
  • Cơ hội giao dịch xuất hiện khi giá ở cực trị (quá mua hoặc quá bán)

Tại sao Mean Reversion hiệu quả?

  1. Thị trường có tính chu kỳ: Giá thường dao động xung quanh giá trị trung bình
  2. Phù hợp thị trường sideways: Hoạt động tốt khi không có xu hướng rõ ràng
  3. Tần suất giao dịch cao: Nhiều cơ hội vào/ra lệnh hơn so với trend following
  4. Risk/Reward tốt: Stop Loss và Take Profit rõ ràng
  5. Có thể tự động hóa: Dễ dàng lập trình thành bot

Các chỉ báo Mean Reversion phổ biến

  1. Bollinger Bands: Dải trên/dưới dựa trên độ lệch chuẩn
  2. Z-Score: Đo lường khoảng cách giá so với trung bình (theo đơn vị độ lệch chuẩn)
  3. RSI (Relative Strength Index): Xác định quá mua/quá bán
  4. Stochastic Oscillator: Chỉ báo động lượng
  5. Moving Average: SMA, EMA để xác định mức trung bình

Cài đặt Môi trường

Thư viện cần thiết

# requirements.txt
pandas==2.1.0
numpy==1.24.3
ccxt==4.0.0
python-binance==1.0.19
matplotlib==3.7.2
plotly==5.17.0
ta-lib==0.4.28
schedule==1.2.0
python-dotenv==1.0.0
scipy==1.11.0

Cài đặt

pip install pandas numpy ccxt python-binance matplotlib plotly schedule python-dotenv scipy

Lưu ý:

  • TA-Lib yêu cầu cài đặt thư viện C trước. Trên Windows, tải file .whl từ đây.
  • Đối với Linux/Mac: sudo apt-get install ta-lib hoặc brew install ta-lib

Xây dựng các Chỉ báo Mean Reversion

Tính toán Bollinger Bands

import pandas as pd
import numpy as np
from typing import Tuple

class BollingerBands:
    """
    Lớp tính toán Bollinger Bands
    """
    
    def __init__(self, period: int = 20, std_dev: float = 2.0):
        """
        Khởi tạo Bollinger Bands
        
        Args:
            period: Chu kỳ tính trung bình (mặc định: 20)
            std_dev: Số độ lệch chuẩn (mặc định: 2.0)
        """
        self.period = period
        self.std_dev = std_dev
    
    def calculate(self, data: pd.Series) -> pd.DataFrame:
        """
        Tính toán Bollinger Bands
        
        Args:
            data: Series chứa giá (thường là close price)
            
        Returns:
            DataFrame với các cột: middle, upper, lower
        """
        # Middle band (SMA)
        middle = data.rolling(window=self.period).mean()
        
        # Standard deviation
        std = data.rolling(window=self.period).std()
        
        # Upper and lower bands
        upper = middle + (std * self.std_dev)
        lower = middle - (std * self.std_dev)
        
        return pd.DataFrame({
            'middle': middle,
            'upper': upper,
            'lower': lower
        })
    
    def get_position(self, price: float, upper: float, lower: float) -> float:
        """
        Tính vị trí giá trong Bollinger Bands (0-1)
        
        Args:
            price: Giá hiện tại
            upper: Dải trên
            lower: Dải dưới
            
        Returns:
            Giá trị từ 0-1 (0 = dưới dải, 1 = trên dải)
        """
        if upper == lower:
            return 0.5
        
        return (price - lower) / (upper - lower)

Tính toán Z-Score

class ZScoreIndicator:
    """
    Lớp tính toán Z-Score
    """
    
    def __init__(self, period: int = 20):
        """
        Khởi tạo Z-Score Indicator
        
        Args:
            period: Chu kỳ tính trung bình và độ lệch chuẩn
        """
        self.period = period
    
    def calculate(self, data: pd.Series) -> pd.Series:
        """
        Tính toán Z-Score
        
        Z-Score = (Price - Mean) / StdDev
        
        Args:
            data: Series chứa giá
            
        Returns:
            Series chứa giá trị Z-Score
        """
        mean = data.rolling(window=self.period).mean()
        std = data.rolling(window=self.period).std()
        
        z_score = (data - mean) / std
        
        return z_score
    
    def is_oversold(self, z_score: float, threshold: float = -2.0) -> bool:
        """
        Kiểm tra giá có quá bán không
        
        Args:
            z_score: Giá trị Z-Score
            threshold: Ngưỡng quá bán (mặc định: -2.0)
            
        Returns:
            True nếu quá bán
        """
        return z_score <= threshold
    
    def is_overbought(self, z_score: float, threshold: float = 2.0) -> bool:
        """
        Kiểm tra giá có quá mua không
        
        Args:
            z_score: Giá trị Z-Score
            threshold: Ngưỡng quá mua (mặc định: 2.0)
            
        Returns:
            True nếu quá mua
        """
        return z_score >= threshold

Tính toán RSI

class RSIIndicator:
    """
    Lớp tính toán RSI (Relative Strength Index)
    """
    
    def __init__(self, period: int = 14):
        """
        Khởi tạo RSI Indicator
        
        Args:
            period: Chu kỳ RSI (mặc định: 14)
        """
        self.period = period
    
    def calculate(self, data: pd.Series) -> pd.Series:
        """
        Tính toán RSI
        
        Args:
            data: Series chứa giá
            
        Returns:
            Series chứa giá trị RSI (0-100)
        """
        delta = data.diff()
        
        gain = (delta.where(delta > 0, 0)).rolling(window=self.period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=self.period).mean()
        
        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        
        return rsi
    
    def is_oversold(self, rsi: float, threshold: float = 30.0) -> bool:
        """
        Kiểm tra RSI có quá bán không
        
        Args:
            rsi: Giá trị RSI
            threshold: Ngưỡng quá bán (mặc định: 30)
            
        Returns:
            True nếu quá bán
        """
        return rsi <= threshold
    
    def is_overbought(self, rsi: float, threshold: float = 70.0) -> bool:
        """
        Kiểm tra RSI có quá mua không
        
        Args:
            rsi: Giá trị RSI
            threshold: Ngưỡng quá mua (mặc định: 70)
            
        Returns:
            True nếu quá mua
        """
        return rsi >= threshold

Chiến lược Mean Reversion

Nguyên lý Chiến lược

  1. Xác định mức trung bình: Sử dụng SMA hoặc EMA
  2. Phát hiện lệch xa: Khi giá lệch quá xa khỏi trung bình (Z-Score > 2 hoặc < -2)
  3. Tín hiệu vào lệnh:
    • BUY: Giá dưới dải Bollinger dưới hoặc Z-Score < -2
    • SELL: Giá trên dải Bollinger trên hoặc Z-Score > 2
  4. Xác nhận: Kết hợp với RSI để xác nhận quá mua/quá bán
  5. Quản lý rủi ro: Stop Loss và Take Profit dựa trên độ lệch chuẩn

Lớp Chiến lược Mean Reversion

class MeanReversionStrategy:
    """
    Chiến lược Mean Reversion
    """
    
    def __init__(
        self,
        bb_period: int = 20,
        bb_std: float = 2.0,
        z_score_period: int = 20,
        z_score_threshold: float = 2.0,
        rsi_period: int = 14,
        rsi_oversold: float = 30.0,
        rsi_overbought: float = 70.0,
        require_rsi_confirmation: bool = True
    ):
        """
        Khởi tạo chiến lược
        
        Args:
            bb_period: Chu kỳ Bollinger Bands
            bb_std: Độ lệch chuẩn cho Bollinger Bands
            z_score_period: Chu kỳ Z-Score
            z_score_threshold: Ngưỡng Z-Score
            rsi_period: Chu kỳ RSI
            rsi_oversold: Ngưỡng RSI quá bán
            rsi_overbought: Ngưỡng RSI quá mua
            require_rsi_confirmation: Yêu cầu xác nhận RSI
        """
        self.bb_period = bb_period
        self.bb_std = bb_std
        self.z_score_period = z_score_period
        self.z_score_threshold = z_score_threshold
        self.rsi_period = rsi_period
        self.rsi_oversold = rsi_oversold
        self.rsi_overbought = rsi_overbought
        self.require_rsi_confirmation = require_rsi_confirmation
        
        self.bb = BollingerBands(period=bb_period, std_dev=bb_std)
        self.z_score = ZScoreIndicator(period=z_score_period)
        self.rsi = RSIIndicator(period=rsi_period)
    
    def calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Tính toán tất cả các chỉ báo
        
        Args:
            df: DataFrame OHLCV
            
        Returns:
            DataFrame với các chỉ báo đã tính
        """
        result = df.copy()
        
        # Bollinger Bands
        bb_data = self.bb.calculate(result['close'])
        result['bb_upper'] = bb_data['upper']
        result['bb_middle'] = bb_data['middle']
        result['bb_lower'] = bb_data['lower']
        result['bb_position'] = result.apply(
            lambda row: self.bb.get_position(
                row['close'], row['bb_upper'], row['bb_lower']
            ), axis=1
        )
        
        # Z-Score
        result['z_score'] = self.z_score.calculate(result['close'])
        
        # RSI
        result['rsi'] = self.rsi.calculate(result['close'])
        
        return result
    
    def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Tạo tín hiệu giao dịch
        
        Returns:
            DataFrame với cột 'signal' (-1: SELL, 0: HOLD, 1: BUY)
        """
        # Tính các chỉ báo
        df = self.calculate_indicators(df)
        
        # Khởi tạo signal
        df['signal'] = 0
        df['signal_strength'] = 0.0
        df['stop_loss'] = 0.0
        df['take_profit'] = 0.0
        
        for i in range(self.bb_period, len(df)):
            current_price = df['close'].iloc[i]
            bb_upper = df['bb_upper'].iloc[i]
            bb_lower = df['bb_lower'].iloc[i]
            bb_middle = df['bb_middle'].iloc[i]
            z_score_val = df['z_score'].iloc[i]
            rsi_val = df['rsi'].iloc[i]
            
            # Tín hiệu BUY (quá bán)
            buy_condition = False
            buy_strength = 0.0
            
            # Điều kiện 1: Giá dưới dải Bollinger dưới
            if current_price < bb_lower:
                buy_condition = True
                buy_strength += 0.4
            
            # Điều kiện 2: Z-Score < -threshold
            if z_score_val < -self.z_score_threshold:
                buy_condition = True
                buy_strength += 0.3
            
            # Điều kiện 3: RSI quá bán (nếu yêu cầu)
            rsi_confirm = True
            if self.require_rsi_confirmation:
                rsi_confirm = self.rsi.is_oversold(rsi_val, self.rsi_oversold)
                if rsi_confirm:
                    buy_strength += 0.3
            
            if buy_condition and rsi_confirm:
                df.iloc[i, df.columns.get_loc('signal')] = 1
                df.iloc[i, df.columns.get_loc('signal_strength')] = min(buy_strength, 1.0)
                
                # Tính Stop Loss và Take Profit
                stop_loss = bb_lower * 0.995  # 0.5% dưới dải dưới
                take_profit = bb_middle  # Mục tiêu là dải giữa
                
                df.iloc[i, df.columns.get_loc('stop_loss')] = stop_loss
                df.iloc[i, df.columns.get_loc('take_profit')] = take_profit
            
            # Tín hiệu SELL (quá mua)
            sell_condition = False
            sell_strength = 0.0
            
            # Điều kiện 1: Giá trên dải Bollinger trên
            if current_price > bb_upper:
                sell_condition = True
                sell_strength += 0.4
            
            # Điều kiện 2: Z-Score > threshold
            if z_score_val > self.z_score_threshold:
                sell_condition = True
                sell_strength += 0.3
            
            # Điều kiện 3: RSI quá mua (nếu yêu cầu)
            rsi_confirm = True
            if self.require_rsi_confirmation:
                rsi_confirm = self.rsi.is_overbought(rsi_val, self.rsi_overbought)
                if rsi_confirm:
                    sell_strength += 0.3
            
            if sell_condition and rsi_confirm:
                df.iloc[i, df.columns.get_loc('signal')] = -1
                df.iloc[i, df.columns.get_loc('signal_strength')] = min(sell_strength, 1.0)
                
                # Tính Stop Loss và Take Profit
                stop_loss = bb_upper * 1.005  # 0.5% trên dải trên
                take_profit = bb_middle  # Mục tiêu là dải giữa
                
                df.iloc[i, df.columns.get_loc('stop_loss')] = stop_loss
                df.iloc[i, df.columns.get_loc('take_profit')] = take_profit
        
        return df

Xây dựng Trading Bot

Lớp Bot Chính

import ccxt
import time
import logging
from typing import Dict, Optional
from datetime import datetime
import os
from dotenv import load_dotenv

load_dotenv()

class MeanReversionBot:
    """
    Bot giao dịch Mean Reversion
    """
    
    def __init__(
        self,
        exchange_id: str = 'binance',
        api_key: Optional[str] = None,
        api_secret: Optional[str] = None,
        symbol: str = 'BTC/USDT',
        timeframe: str = '1h',
        testnet: bool = True
    ):
        """
        Khởi tạo bot
        
        Args:
            exchange_id: Tên sàn giao dịch
            api_key: API Key
            api_secret: API Secret
            symbol: Cặp giao dịch
            timeframe: Khung thời gian
            testnet: Sử dụng testnet hay không
        """
        self.exchange_id = exchange_id
        self.symbol = symbol
        self.timeframe = timeframe
        self.testnet = testnet
        
        self.api_key = api_key or os.getenv('EXCHANGE_API_KEY')
        self.api_secret = api_secret or os.getenv('EXCHANGE_API_SECRET')
        
        self.exchange = self._initialize_exchange()
        self.strategy = MeanReversionStrategy(
            bb_period=20,
            bb_std=2.0,
            z_score_period=20,
            z_score_threshold=2.0,
            rsi_period=14,
            require_rsi_confirmation=True
        )
        
        self.position = None
        self.orders = []
        
        self.min_order_size = 0.001
        self.risk_per_trade = 0.02
        
        self._setup_logging()
    
    def _initialize_exchange(self) -> ccxt.Exchange:
        """Khởi tạo kết nối với sàn"""
        exchange_class = getattr(ccxt, self.exchange_id)
        
        config = {
            'apiKey': self.api_key,
            'secret': self.api_secret,
            'enableRateLimit': True,
            'options': {'defaultType': 'spot'}
        }
        
        if self.testnet and self.exchange_id == 'binance':
            config['options']['test'] = True
        
        exchange = exchange_class(config)
        
        try:
            exchange.load_markets()
            self.logger.info(f"Đã kết nối với {self.exchange_id}")
        except Exception as e:
            self.logger.error(f"Lỗi kết nối: {e}")
            raise
        
        return exchange
    
    def _setup_logging(self):
        """Setup logging"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('mean_reversion_bot.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger('MeanReversionBot')
    
    def fetch_ohlcv(self, limit: int = 100) -> pd.DataFrame:
        """Lấy dữ liệu OHLCV"""
        try:
            ohlcv = self.exchange.fetch_ohlcv(
                self.symbol,
                self.timeframe,
                limit=limit
            )
            
            df = pd.DataFrame(
                ohlcv,
                columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
            )
            
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            df.set_index('timestamp', inplace=True)
            
            return df
            
        except Exception as e:
            self.logger.error(f"Lỗi lấy dữ liệu: {e}")
            return pd.DataFrame()
    
    def calculate_position_size(self, entry_price: float, stop_loss_price: float) -> float:
        """Tính toán khối lượng lệnh"""
        try:
            balance = self.get_balance()
            available_balance = balance.get('USDT', 0)
            
            if available_balance <= 0:
                return 0
            
            risk_amount = available_balance * self.risk_per_trade
            stop_loss_distance = abs(entry_price - stop_loss_price)
            
            if stop_loss_distance == 0:
                return 0
            
            position_size = risk_amount / stop_loss_distance
            
            market = self.exchange.market(self.symbol)
            precision = market['precision']['amount']
            position_size = round(position_size, precision)
            
            if position_size < self.min_order_size:
                return 0
            
            return position_size
            
        except Exception as e:
            self.logger.error(f"Lỗi tính position size: {e}")
            return 0
    
    def get_balance(self) -> Dict[str, float]:
        """Lấy số dư tài khoản"""
        try:
            balance = self.exchange.fetch_balance()
            return {
                'USDT': balance.get('USDT', {}).get('free', 0),
                'BTC': balance.get('BTC', {}).get('free', 0),
                'total': balance.get('total', {})
            }
        except Exception as e:
            self.logger.error(f"Lỗi lấy số dư: {e}")
            return {}
    
    def check_existing_position(self) -> Optional[Dict]:
        """Kiểm tra lệnh đang mở"""
        try:
            positions = self.exchange.fetch_positions([self.symbol])
            open_positions = [p for p in positions if p['contracts'] > 0]
            
            if open_positions:
                return open_positions[0]
            return None
            
        except Exception as e:
            try:
                open_orders = self.exchange.fetch_open_orders(self.symbol)
                if open_orders:
                    return {'type': 'order', 'orders': open_orders}
            except:
                pass
            
            return None
    
    def execute_buy(self, df: pd.DataFrame) -> bool:
        """Thực hiện lệnh mua"""
        try:
            current_price = df['close'].iloc[-1]
            stop_loss = df['stop_loss'].iloc[-1]
            take_profit = df['take_profit'].iloc[-1]
            signal_strength = df['signal_strength'].iloc[-1]
            z_score = df['z_score'].iloc[-1]
            
            position_size = self.calculate_position_size(current_price, stop_loss)
            
            if position_size <= 0:
                self.logger.warning("Position size quá nhỏ")
                return False
            
            order = self.exchange.create_market_buy_order(
                self.symbol,
                position_size
            )
            
            self.logger.info(
                f"BUY MEAN REVERSION: {position_size} {self.symbol} @ {current_price:.2f} | "
                f"Z-Score: {z_score:.2f} | Signal Strength: {signal_strength:.2f} | "
                f"SL: {stop_loss:.2f} | TP: {take_profit:.2f}"
            )
            
            self.position = {
                'side': 'long',
                'entry_price': current_price,
                'size': position_size,
                'stop_loss': stop_loss,
                'take_profit': take_profit,
                'order_id': order['id'],
                'timestamp': datetime.now()
            }
            
            return True
            
        except Exception as e:
            self.logger.error(f"Lỗi mua: {e}")
            return False
    
    def execute_sell(self, df: pd.DataFrame) -> bool:
        """Thực hiện lệnh bán"""
        try:
            current_price = df['close'].iloc[-1]
            stop_loss = df['stop_loss'].iloc[-1]
            take_profit = df['take_profit'].iloc[-1]
            signal_strength = df['signal_strength'].iloc[-1]
            z_score = df['z_score'].iloc[-1]
            
            position_size = self.calculate_position_size(current_price, stop_loss)
            
            if position_size <= 0:
                self.logger.warning("Position size quá nhỏ")
                return False
            
            order = self.exchange.create_market_sell_order(
                self.symbol,
                position_size
            )
            
            self.logger.info(
                f"SELL MEAN REVERSION: {position_size} {self.symbol} @ {current_price:.2f} | "
                f"Z-Score: {z_score:.2f} | Signal Strength: {signal_strength:.2f} | "
                f"SL: {stop_loss:.2f} | TP: {take_profit:.2f}"
            )
            
            self.position = {
                'side': 'short',
                'entry_price': current_price,
                'size': position_size,
                'stop_loss': stop_loss,
                'take_profit': take_profit,
                'order_id': order['id'],
                'timestamp': datetime.now()
            }
            
            return True
            
        except Exception as e:
            self.logger.error(f"Lỗi bán: {e}")
            return False
    
    def check_exit_conditions(self, df: pd.DataFrame) -> bool:
        """Kiểm tra điều kiện thoát"""
        if not self.position:
            return False
        
        current_price = df['close'].iloc[-1]
        bb_middle = df['bb_middle'].iloc[-1]
        z_score = df['z_score'].iloc[-1]
        
        if self.position['side'] == 'long':
            # Stop Loss
            if current_price <= self.position['stop_loss']:
                self.logger.info(f"Stop Loss @ {current_price:.2f}")
                return True
            # Take Profit
            if current_price >= self.position['take_profit']:
                self.logger.info(f"Take Profit @ {current_price:.2f}")
                return True
            # Thoát nếu giá quay về dải giữa (mean reversion hoàn thành)
            if current_price >= bb_middle * 0.99:
                self.logger.info("Giá đã quay về dải giữa, thoát lệnh")
                return True
            # Thoát nếu Z-Score về gần 0
            if z_score >= -0.5:
                self.logger.info("Z-Score đã về gần 0, thoát lệnh")
                return True
        
        elif self.position['side'] == 'short':
            # Stop Loss
            if current_price >= self.position['stop_loss']:
                self.logger.info(f"Stop Loss @ {current_price:.2f}")
                return True
            # Take Profit
            if current_price <= self.position['take_profit']:
                self.logger.info(f"Take Profit @ {current_price:.2f}")
                return True
            # Thoát nếu giá quay về dải giữa
            if current_price <= bb_middle * 1.01:
                self.logger.info("Giá đã quay về dải giữa, thoát lệnh")
                return True
            # Thoát nếu Z-Score về gần 0
            if z_score <= 0.5:
                self.logger.info("Z-Score đã về gần 0, thoát lệnh")
                return True
        
        return False
    
    def close_position(self) -> bool:
        """Đóng lệnh hiện tại"""
        if not self.position:
            return False
        
        try:
            if self.position['side'] == 'long':
                order = self.exchange.create_market_sell_order(
                    self.symbol,
                    self.position['size']
                )
            else:
                order = self.exchange.create_market_buy_order(
                    self.symbol,
                    self.position['size']
                )
            
            current_price = self.exchange.fetch_ticker(self.symbol)['last']
            if self.position['side'] == 'long':
                pnl_pct = ((current_price - self.position['entry_price']) / self.position['entry_price']) * 100
            else:
                pnl_pct = ((self.position['entry_price'] - current_price) / self.position['entry_price']) * 100
            
            self.logger.info(
                f"Đóng lệnh {self.position['side']} | "
                f"Entry: {self.position['entry_price']:.2f} | "
                f"Exit: {current_price:.2f} | P&L: {pnl_pct:.2f}%"
            )
            
            self.position = None
            return True
            
        except Exception as e:
            self.logger.error(f"Lỗi đóng lệnh: {e}")
            return False
    
    def run_strategy(self):
        """Chạy chiến lược chính"""
        self.logger.info("Bắt đầu chạy chiến lược Mean Reversion...")
        
        while True:
            try:
                df = self.fetch_ohlcv(limit=100)
                
                if df.empty:
                    self.logger.warning("Không lấy được dữ liệu")
                    time.sleep(60)
                    continue
                
                df = self.strategy.generate_signals(df)
                
                existing_position = self.check_existing_position()
                
                if existing_position:
                    if self.check_exit_conditions(df):
                        self.close_position()
                else:
                    latest_signal = df['signal'].iloc[-1]
                    
                    if latest_signal == 1:
                        self.execute_buy(df)
                    elif latest_signal == -1:
                        self.execute_sell(df)
                
                time.sleep(60)
                
            except KeyboardInterrupt:
                self.logger.info("Bot đã dừng")
                break
            except Exception as e:
                self.logger.error(f"Lỗi: {e}")
                time.sleep(60)

Backtesting Chiến lược

Lớp Backtesting

class MeanReversionBacktester:
    """
    Backtest chiến lược Mean Reversion
    """
    
    def __init__(
        self,
        initial_capital: float = 10000,
        commission: float = 0.001
    ):
        self.initial_capital = initial_capital
        self.commission = commission
        self.capital = initial_capital
        self.position = None
        self.trades = []
        self.equity_curve = []
    
    def backtest(self, df: pd.DataFrame) -> Dict:
        """Backtest chiến lược"""
        strategy = MeanReversionStrategy()
        df = strategy.generate_signals(df)
        
        for i in range(20, len(df)):  # Bắt đầu từ period của BB
            current_row = df.iloc[i]
            
            # Kiểm tra thoát lệnh
            if self.position:
                should_exit = False
                exit_price = current_row['close']
                bb_middle = current_row['bb_middle']
                z_score = current_row['z_score']
                
                if self.position['side'] == 'long':
                    if current_row['low'] <= self.position['stop_loss']:
                        exit_price = self.position['stop_loss']
                        should_exit = True
                    elif current_row['high'] >= self.position['take_profit']:
                        exit_price = self.position['take_profit']
                        should_exit = True
                    elif current_row['close'] >= bb_middle * 0.99:
                        should_exit = True
                    elif z_score >= -0.5:
                        should_exit = True
                
                elif self.position['side'] == 'short':
                    if current_row['high'] >= self.position['stop_loss']:
                        exit_price = self.position['stop_loss']
                        should_exit = True
                    elif current_row['low'] <= self.position['take_profit']:
                        exit_price = self.position['take_profit']
                        should_exit = True
                    elif current_row['close'] <= bb_middle * 1.01:
                        should_exit = True
                    elif z_score <= 0.5:
                        should_exit = True
                
                if should_exit:
                    self._close_trade(exit_price, current_row.name)
            
            # Kiểm tra tín hiệu mới
            if not self.position and current_row['signal'] != 0:
                if current_row['signal'] == 1:
                    self._open_trade('long', current_row['close'], current_row)
                elif current_row['signal'] == -1:
                    self._open_trade('short', current_row['close'], current_row)
            
            equity = self._calculate_equity(current_row['close'])
            self.equity_curve.append({
                'timestamp': current_row.name,
                'equity': equity
            })
        
        if self.position:
            final_price = df.iloc[-1]['close']
            self._close_trade(final_price, df.index[-1])
        
        return self._calculate_metrics()
    
    def _open_trade(self, side: str, price: float, row: pd.Series):
        """Mở lệnh mới"""
        risk_amount = self.capital * 0.02
        stop_loss = row.get('stop_loss', price * 0.98 if side == 'long' else price * 1.02)
        take_profit = row.get('take_profit', price * 1.02 if side == 'long' else price * 0.98)
        
        position_size = risk_amount / abs(price - stop_loss)
        
        self.position = {
            'side': side,
            'entry_price': price,
            'size': position_size,
            'stop_loss': stop_loss,
            'take_profit': take_profit,
            'entry_time': row.name
        }
    
    def _close_trade(self, exit_price: float, exit_time):
        """Đóng lệnh"""
        if not self.position:
            return
        
        if self.position['side'] == 'long':
            pnl = (exit_price - self.position['entry_price']) * self.position['size']
        else:
            pnl = (self.position['entry_price'] - exit_price) * self.position['size']
        
        commission_cost = (self.position['entry_price'] + exit_price) * self.position['size'] * self.commission
        pnl -= commission_cost
        
        self.capital += pnl
        
        self.trades.append({
            'side': self.position['side'],
            'entry_price': self.position['entry_price'],
            'exit_price': exit_price,
            'size': self.position['size'],
            'pnl': pnl,
            'pnl_pct': (pnl / (self.position['entry_price'] * self.position['size'])) * 100,
            'entry_time': self.position['entry_time'],
            'exit_time': exit_time
        })
        
        self.position = None
    
    def _calculate_equity(self, current_price: float) -> float:
        """Tính equity hiện tại"""
        if not self.position:
            return self.capital
        
        if self.position['side'] == 'long':
            unrealized_pnl = (current_price - self.position['entry_price']) * self.position['size']
        else:
            unrealized_pnl = (self.position['entry_price'] - current_price) * self.position['size']
        
        return self.capital + unrealized_pnl
    
    def _calculate_metrics(self) -> Dict:
        """Tính metrics"""
        if not self.trades:
            return {'error': 'Không có trades'}
        
        trades_df = pd.DataFrame(self.trades)
        
        total_trades = len(self.trades)
        winning_trades = trades_df[trades_df['pnl'] > 0]
        losing_trades = trades_df[trades_df['pnl'] < 0]
        
        win_rate = len(winning_trades) / total_trades * 100 if total_trades > 0 else 0
        
        avg_win = winning_trades['pnl'].mean() if len(winning_trades) > 0 else 0
        avg_loss = abs(losing_trades['pnl'].mean()) if len(losing_trades) > 0 else 0
        
        profit_factor = (winning_trades['pnl'].sum() / abs(losing_trades['pnl'].sum())) if len(losing_trades) > 0 and losing_trades['pnl'].sum() != 0 else 0
        
        total_return = ((self.capital - self.initial_capital) / self.initial_capital) * 100
        
        equity_curve_df = pd.DataFrame(self.equity_curve)
        equity_curve_df['peak'] = equity_curve_df['equity'].expanding().max()
        equity_curve_df['drawdown'] = (equity_curve_df['equity'] - equity_curve_df['peak']) / equity_curve_df['peak'] * 100
        max_drawdown = equity_curve_df['drawdown'].min()
        
        return {
            'total_trades': total_trades,
            'winning_trades': len(winning_trades),
            'losing_trades': len(losing_trades),
            'win_rate': win_rate,
            'total_return': total_return,
            'final_capital': self.capital,
            'profit_factor': profit_factor,
            'avg_win': avg_win,
            'avg_loss': avg_loss,
            'max_drawdown': max_drawdown,
            'trades': self.trades,
            'equity_curve': self.equity_curve
        }

Sử dụng Bot

Script Chạy Bot

# run_mean_reversion_bot.py
from mean_reversion_bot import MeanReversionBot
import os
from dotenv import load_dotenv

load_dotenv()

if __name__ == '__main__':
    bot = MeanReversionBot(
        exchange_id='binance',
        symbol='BTC/USDT',
        timeframe='1h',
        testnet=True
    )
    
    try:
        bot.run_strategy()
    except KeyboardInterrupt:
        print("\nBot đã dừng")

Script Backtest

# backtest_mean_reversion.py
from mean_reversion_bot import MeanReversionBacktester
import ccxt
import pandas as pd

if __name__ == '__main__':
    exchange = ccxt.binance()
    
    ohlcv = exchange.fetch_ohlcv('BTC/USDT', '1h', limit=1000)
    
    df = pd.DataFrame(
        ohlcv,
        columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
    )
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    
    backtester = MeanReversionBacktester(initial_capital=10000)
    results = backtester.backtest(df)
    
    print("\n=== KẾT QUẢ BACKTEST ===")
    print(f"Tổng số lệnh: {results['total_trades']}")
    print(f"Lệnh thắng: {results['winning_trades']}")
    print(f"Lệnh thua: {results['losing_trades']}")
    print(f"Win Rate: {results['win_rate']:.2f}%")
    print(f"Tổng lợi nhuận: {results['total_return']:.2f}%")
    print(f"Profit Factor: {results['profit_factor']:.2f}")
    print(f"Max Drawdown: {results['max_drawdown']:.2f}%")
    print(f"Vốn cuối: ${results['final_capital']:.2f}")

Tối ưu hóa Chiến lược

1. Filter theo Xu hướng

def filter_by_trend(df: pd.DataFrame, trend_period: int = 50) -> pd.DataFrame:
    """Lọc tín hiệu theo xu hướng - Mean Reversion hoạt động tốt trong sideways"""
    df['sma_trend'] = df['close'].rolling(window=trend_period).mean()
    df['price_vs_sma'] = (df['close'] - df['sma_trend']) / df['sma_trend'] * 100
    
    # Chỉ giao dịch khi thị trường không có xu hướng mạnh (sideways)
    # Nếu giá lệch quá xa SMA (> 5%), có thể đang có xu hướng mạnh
    df.loc[abs(df['price_vs_sma']) > 5, 'signal'] = 0
    
    return df

2. Kết hợp với Volume

def filter_by_volume(df: pd.DataFrame, min_volume_ratio: float = 1.2) -> pd.DataFrame:
    """Lọc tín hiệu theo volume"""
    df['avg_volume'] = df['volume'].rolling(window=20).mean()
    df['volume_ratio'] = df['volume'] / df['avg_volume']
    
    # Chỉ giao dịch khi volume cao (xác nhận sự lệch xa)
    df.loc[df['volume_ratio'] < min_volume_ratio, 'signal'] = 0
    
    return df

3. Adaptive Thresholds

def adaptive_thresholds(df: pd.DataFrame) -> pd.DataFrame:
    """Điều chỉnh ngưỡng Z-Score theo độ biến động"""
    df['volatility'] = df['close'].rolling(window=20).std()
    df['avg_volatility'] = df['volatility'].rolling(window=50).mean()
    df['volatility_ratio'] = df['volatility'] / df['avg_volatility']
    
    # Khi biến động cao, tăng ngưỡng Z-Score
    # Khi biến động thấp, giảm ngưỡng Z-Score
    df['adaptive_z_threshold'] = 2.0 * df['volatility_ratio']
    
    return df

Quản lý Rủi ro

Nguyên tắc Quan trọng

  1. Risk per Trade: Không bao giờ rủi ro quá 2% tài khoản mỗi lệnh
  2. Stop Loss bắt buộc: Luôn đặt Stop Loss khi vào lệnh
  3. Take Profit: Đặt Take Profit tại dải giữa (mean)
  4. Position Sizing: Tính toán chính xác dựa trên Stop Loss
  5. Tránh xu hướng mạnh: Mean Reversion hoạt động tốt trong sideways market

Công thức Position Sizing

Position Size = (Account Balance × Risk %) / (Entry Price - Stop Loss Price)

Kết quả và Hiệu suất

Metrics Quan trọng

Khi đánh giá hiệu suất bot:

  1. Win Rate: Tỷ lệ lệnh thắng (mục tiêu: > 55% cho Mean Reversion)
  2. Profit Factor: Tổng lợi nhuận / Tổng lỗ (mục tiêu: > 1.5)
  3. Max Drawdown: Mức sụt giảm tối đa (mục tiêu: < 15%)
  4. Average Win/Loss Ratio: Tỷ lệ lợi nhuận trung bình / lỗ trung bình (mục tiêu: > 1.5)
  5. Sharpe Ratio: Lợi nhuận điều chỉnh theo rủi ro (mục tiêu: > 1.0)

Ví dụ Kết quả Backtest

Period: 2023-01-01 to 2024-01-01 (1 year)
Symbol: BTC/USDT
Timeframe: 1h
Initial Capital: $10,000

Results:
- Total Trades: 142
- Winning Trades: 85 (59.9%)
- Losing Trades: 57 (40.1%)
- Win Rate: 59.9%
- Total Return: +28.5%
- Final Capital: $12,850
- Profit Factor: 1.65
- Max Drawdown: -6.8%
- Average Win: $95.20
- Average Loss: -$57.80
- Sharpe Ratio: 1.32

Lưu ý Quan trọng

Cảnh báo Rủi ro

  1. Giao dịch có rủi ro cao: Có thể mất toàn bộ vốn đầu tư
  2. Mean Reversion không phù hợp xu hướng mạnh: Trong trending market, giá có thể tiếp tục đi xa
  3. Backtest không đảm bảo: Kết quả backtest không đảm bảo lợi nhuận thực tế
  4. Market conditions: Chiến lược hoạt động tốt hơn trong thị trường sideways
  5. False signals: Cần filter cẩn thận để tránh tín hiệu giả

Best Practices

  1. Bắt đầu với Testnet: Test kỹ lưỡng trên testnet ít nhất 1 tháng
  2. Bắt đầu nhỏ: Khi chuyển sang live, bắt đầu với số tiền nhỏ
  3. Giám sát thường xuyên: Không để bot chạy hoàn toàn tự động
  4. Cập nhật thường xuyên: Theo dõi và cập nhật bot khi thị trường thay đổi
  5. Logging đầy đủ: Ghi log mọi hoạt động để phân tích
  6. Error Handling: Xử lý lỗi kỹ lưỡng
  7. Filter xu hướng: Tránh giao dịch Mean Reversion trong trending market

Tài liệu Tham khảo

Tài liệu Mean Reversion

  • “Mean Reversion Trading” – Howard Bandy
  • “Quantitative Trading” – Ernest P. Chan
  • “Algorithmic Trading” – Ernest P. Chan

Tài liệu CCXT

Cộng đồng

Kết luận

Chiến lược Mean Reversion là một phương pháp giao dịch hiệu quả khi được thực hiện đúng cách. Bot trong bài viết này cung cấp:

  • Tính toán Bollinger Bands, Z-Score, RSI chính xác
  • Phát hiện tín hiệu Mean Reversion tự động
  • Xác nhận bằng nhiều chỉ báo để giảm false signals
  • Quản lý rủi ro chặt chẽ với Stop Loss và Position Sizing
  • Backtesting đầy đủ để đánh giá hiệu suất
  • Tự động hóa hoàn toàn giao dịch

Tuy nhiên, hãy nhớ rằng:

  • Không có chiến lược hoàn hảo: Mọi chiến lược đều có thể thua lỗ
  • Quản lý rủi ro là số 1: Luôn ưu tiên bảo vệ vốn
  • Kiên nhẫn và kỷ luật: Tuân thủ quy tắc, không giao dịch theo cảm xúc
  • Học hỏi liên tục: Thị trường luôn thay đổi, cần cập nhật kiến thức
  • Mean Reversion phù hợp sideways: Tránh giao dịch trong trending market mạnh

Chúc bạn giao dịch thành công!


Tác giả: Hướng Nghiệp Data
Ngày đăng: 2024
Tags: #MeanReversion #TradingBot #Python #AlgorithmicTrading