| Chiến Lược Momentum Bot Auto Trading

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

Chiến Lược Momentum Trading bằng Python

Momentum Trading 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 “xu hướng là bạn của bạn” (trend is your friend). Chiến lược này giả định rằng các tài sản đang tăng giá sẽ tiếp tục tăng, và các tài sản đang giảm giá sẽ tiếp tục giảm. 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 Momentum Trading với Python.

Tổng quan về Momentum Trading

Momentum Trading là gì?

Momentum Trading là phương pháp giao dịch dựa trên giả định rằng xu hướng hiện tại sẽ tiếp tục trong tương lai. Nguyên tắc cơ bản:

Mua khi giá đang tăng mạnh (momentum tăng)
Bán khi giá đang giảm mạnh (momentum giảm)
Nắm giữ vị thế cho đến khi momentum yếu đi
Thoát lệnh khi momentum đảo chiều

Tại sao Momentum Trading hiệu quả?

Xu hướng có tính bền vững: Xu hướng mạnh thường tiếp tục trong một khoảng thời gian
Phù hợp thị trường trending: Hoạt động tốt khi có xu hướng rõ ràng
Lợi nhuận cao: Có thể bắt được toàn bộ xu hướng
Tín hiệu rõ ràng: Các chỉ báo momentum dễ nhận diện
Có thể tự động hóa: Dễ dàng lập trình thành bot

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

MACD (Moving Average Convergence Divergence): Đo lường sự thay đổi momentum
RSI (Relative Strength Index): Xác định momentum và quá mua/quá bán
Rate of Change (ROC): Tốc độ thay đổi giá
Momentum Indicator: Chênh lệch giá giữa hai thời điểm
Stochastic Oscillator: So sánh giá đóng cửa với phạm vi giá
ADX (Average Directional Index): Đo lường sức mạnh xu hướng

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 Momentum

Tính toán MACD

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

class MACDIndicator:
    """
    Lớp tính toán MACD
    """
    
    def __init__(self, fast_period: int = 12, slow_period: int = 26, signal_period: int = 9):
        """
        Khởi tạo MACD Indicator
        
        Args:
            fast_period: Chu kỳ EMA nhanh (mặc định: 12)
            slow_period: Chu kỳ EMA chậm (mặc định: 26)
            signal_period: Chu kỳ EMA cho Signal Line (mặc định: 9)
        """
        self.fast_period = fast_period
        self.slow_period = slow_period
        self.signal_period = signal_period
    
    def calculate_ema(self, data: pd.Series, period: int) -> pd.Series:
        """
        Tính toán Exponential Moving Average
        
        Args:
            data: Series chứa giá
            period: Chu kỳ EMA
            
        Returns:
            Series chứa giá trị EMA
        """
        return data.ewm(span=period, adjust=False).mean()
    
    def calculate(self, data: pd.Series) -> pd.DataFrame:
        """
        Tính toán MACD, Signal và Histogram
        
        Args:
            data: Series chứa giá (thường là close price)
            
        Returns:
            DataFrame với các cột: macd, signal, histogram
        """
        # Tính EMA nhanh và chậm
        ema_fast = self.calculate_ema(data, self.fast_period)
        ema_slow = self.calculate_ema(data, self.slow_period)
        
        # Tính MACD Line
        macd_line = ema_fast - ema_slow
        
        # Tính Signal Line
        signal_line = self.calculate_ema(macd_line, self.signal_period)
        
        # Tính Histogram
        histogram = macd_line - signal_line
        
        return pd.DataFrame({
            'macd': macd_line,
            'signal': signal_line,
            'histogram': histogram
        })
    
    def get_signal(self, macd: float, signal: float, histogram: float) -> int:
        """
        Xác định tín hiệu từ MACD
        
        Args:
            macd: Giá trị MACD
            signal: Giá trị Signal
            histogram: Giá trị Histogram
            
        Returns:
            1: Bullish, -1: Bearish, 0: Neutral
        """
        # Bullish: MACD cắt lên Signal và Histogram > 0
        if macd > signal and histogram > 0:
            return 1
        # Bearish: MACD cắt xuống Signal và Histogram < 0
        elif macd < signal and histogram < 0:
            return -1
        return 0

Tính toán Rate of Change (ROC)

class ROCIndicator:
    """
    Lớp tính toán Rate of Change
    """
    
    def __init__(self, period: int = 10):
        """
        Khởi tạo ROC Indicator
        
        Args:
            period: Chu kỳ ROC (mặc định: 10)
        """
        self.period = period
    
    def calculate(self, data: pd.Series) -> pd.Series:
        """
        Tính toán Rate of Change
        
        ROC = ((Price(today) - Price(n periods ago)) / Price(n periods ago)) × 100
        
        Args:
            data: Series chứa giá
            
        Returns:
            Series chứa giá trị ROC (%)
        """
        roc = ((data - data.shift(self.period)) / data.shift(self.period)) * 100
        return roc
    
    def is_strong_momentum(self, roc: float, threshold: float = 2.0) -> bool:
        """
        Kiểm tra momentum có mạnh không
        
        Args:
            roc: Giá trị ROC
            threshold: Ngưỡng momentum mạnh (%)
            
        Returns:
            True nếu momentum mạnh
        """
        return abs(roc) >= threshold

Tính toán Momentum Indicator

class MomentumIndicator:
    """
    Lớp tính toán Momentum Indicator
    """
    
    def __init__(self, period: int = 10):
        """
        Khởi tạo Momentum Indicator
        
        Args:
            period: Chu kỳ Momentum (mặc định: 10)
        """
        self.period = period
    
    def calculate(self, data: pd.Series) -> pd.Series:
        """
        Tính toán Momentum
        
        Momentum = Price(today) - Price(n periods ago)
        
        Args:
            data: Series chứa giá
            
        Returns:
            Series chứa giá trị Momentum
        """
        momentum = data - data.shift(self.period)
        return momentum
    
    def get_signal(self, momentum: float, prev_momentum: float) -> int:
        """
        Xác định tín hiệu từ Momentum
        
        Args:
            momentum: Giá trị Momentum hiện tại
            prev_momentum: Giá trị Momentum trước đó
            
        Returns:
            1: Bullish, -1: Bearish, 0: Neutral
        """
        # Momentum tăng và dương
        if momentum > 0 and momentum > prev_momentum:
            return 1
        # Momentum giảm và âm
        elif momentum < 0 and momentum < prev_momentum:
            return -1
        return 0

Tính toán ADX (Average Directional Index)

class ADXIndicator:
    """
    Lớp tính toán ADX (Average Directional Index)
    """
    
    def __init__(self, period: int = 14):
        """
        Khởi tạo ADX Indicator
        
        Args:
            period: Chu kỳ ADX (mặc định: 14)
        """
        self.period = period
    
    def calculate_true_range(self, df: pd.DataFrame) -> pd.Series:
        """Tính True Range"""
        high = df['high']
        low = df['low']
        close = df['close']
        prev_close = close.shift(1)
        
        tr1 = high - low
        tr2 = abs(high - prev_close)
        tr3 = abs(low - prev_close)
        
        true_range = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        return true_range
    
    def calculate_directional_movement(self, df: pd.DataFrame) -> Tuple[pd.Series, pd.Series]:
        """Tính Directional Movement"""
        high = df['high']
        low = df['low']
        prev_high = high.shift(1)
        prev_low = low.shift(1)
        
        # Plus Directional Movement (+DM)
        plus_dm = high - prev_high
        plus_dm[plus_dm < 0] = 0
        plus_dm[(high - prev_high) < (prev_low - low)] = 0
        
        # Minus Directional Movement (-DM)
        minus_dm = prev_low - low
        minus_dm[minus_dm < 0] = 0
        minus_dm[(prev_low - low) < (high - prev_high)] = 0
        
        return plus_dm, minus_dm
    
    def calculate(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Tính toán ADX, +DI, -DI
        
        Args:
            df: DataFrame OHLCV
            
        Returns:
            DataFrame với các cột: adx, plus_di, minus_di
        """
        # Tính True Range
        tr = self.calculate_true_range(df)
        atr = tr.rolling(window=self.period).mean()
        
        # Tính Directional Movement
        plus_dm, minus_dm = self.calculate_directional_movement(df)
        
        # Tính Directional Indicators
        plus_di = 100 * (plus_dm.rolling(window=self.period).mean() / atr)
        minus_di = 100 * (minus_dm.rolling(window=self.period).mean() / atr)
        
        # Tính DX
        dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di)
        
        # Tính ADX (SMA của DX)
        adx = dx.rolling(window=self.period).mean()
        
        return pd.DataFrame({
            'adx': adx,
            'plus_di': plus_di,
            'minus_di': minus_di
        })
    
    def is_strong_trend(self, adx: float, threshold: float = 25.0) -> bool:
        """
        Kiểm tra xu hướng có mạnh không
        
        Args:
            adx: Giá trị ADX
            threshold: Ngưỡng xu hướng mạnh (mặc định: 25)
            
        Returns:
            True nếu xu hướng mạnh
        """
        return adx >= 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

Tính toán ATR

class ATRIndicator:
    """
    Lớp tính toán ATR (Average True Range)
    """
    
    def __init__(self, period: int = 14):
        """
        Khởi tạo ATR Indicator
        
        Args:
            period: Chu kỳ ATR (mặc định: 14)
        """
        self.period = period
    
    def calculate_true_range(self, df: pd.DataFrame) -> pd.Series:
        """Tính True Range"""
        high = df['high']
        low = df['low']
        close = df['close']
        prev_close = close.shift(1)
        
        tr1 = high - low
        tr2 = abs(high - prev_close)
        tr3 = abs(low - prev_close)
        
        true_range = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        return true_range
    
    def calculate_smma_atr(self, df: pd.DataFrame) -> pd.Series:
        """
        Tính ATR sử dụng SMMA (Wilder's smoothing)
        
        Args:
            df: DataFrame chứa OHLC data
            
        Returns:
            Series chứa giá trị ATR
        """
        true_range = self.calculate_true_range(df)
        
        # Sử dụng SMMA để tính ATR
        atr = true_range.ewm(alpha=1.0/self.period, adjust=False).mean()
        
        return atr