| Tính Toán Chỉ Báo Trong Forex MT5 Với Python Cho Bot Auto Trading

Được viết bởi thanhdt vào ngày 15/11/2025 lúc 21:50 | 7 lượt xem

Hướng dẫn chi tiết cách tính toán các chỉ báo kỹ thuật trong Forex MT5 với Python để sử dụng trong bot auto trading

Tính Toán Chỉ Báo Trong Forex MT5 Với Python Cho Bot Auto Trading

Tính toán chỉ báo kỹ thuật là bước quan trọng trong việc xây dựng bot auto trading. Bài viết này sẽ hướng dẫn bạn cách tính toán các chỉ báo kỹ thuật từ dữ liệu MT5 bằng Python để sử dụng trong bot giao dịch tự động.

Chuẩn Bị

import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime

# Kết nối với MT5
if not mt5.initialize():
    print("Khởi tạo MT5 thất bại")
    exit()

# Hàm lấy dữ liệu
def get_rates(symbol, timeframe, count=1000):
    """Lấy dữ liệu OHLC từ MT5"""
    rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, count)
    if rates is None:
        return None
    df = pd.DataFrame(rates)
    df['time'] = pd.to_datetime(df['time'], unit='s')
    return df

Moving Average (MA)

Simple Moving Average (SMA)

def calculate_sma(data, period):
    """
    Tính Simple Moving Average

    Args:
        data: Series hoặc array chứa giá
        period: Chu kỳ

    Returns:
        Series: SMA values
    """
    return data.rolling(window=period).mean()

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    rates['SMA20'] = calculate_sma(rates['close'], 20)
    rates['SMA50'] = calculate_sma(rates['close'], 50)
    print(rates[['time', 'close', 'SMA20', 'SMA50']].tail())

Exponential Moving Average (EMA)

def calculate_ema(data, period):
    """
    Tính Exponential Moving Average

    Args:
        data: Series hoặc array chứa giá
        period: Chu kỳ

    Returns:
        Series: EMA values
    """
    return data.ewm(span=period, adjust=False).mean()

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    rates['EMA12'] = calculate_ema(rates['close'], 12)
    rates['EMA26'] = calculate_ema(rates['close'], 26)
    print(rates[['time', 'close', 'EMA12', 'EMA26']].tail())

Weighted Moving Average (WMA)

def calculate_wma(data, period):
    """
    Tính Weighted Moving Average

    Args:
        data: Series hoặc array chứa giá
        period: Chu kỳ

    Returns:
        Series: WMA values
    """
    weights = np.arange(1, period + 1)
    return data.rolling(window=period).apply(
        lambda x: np.dot(x, weights) / weights.sum(), raw=True
    )

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    rates['WMA20'] = calculate_wma(rates['close'], 20)

RSI (Relative Strength Index)

def calculate_rsi(data, period=14):
    """
    Tính RSI (Relative Strength Index)

    Args:
        data: Series chứa giá close
        period: Chu kỳ (mặc định 14)

    Returns:
        Series: RSI values (0-100)
    """
    delta = data.diff()

    gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()

    rs = gain / loss
    rsi = 100 - (100 / (1 + rs))

    return rsi

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    rates['RSI'] = calculate_rsi(rates['close'], period=14)

    # Tín hiệu giao dịch
    rates['RSI_Overbought'] = rates['RSI'] > 70
    rates['RSI_Oversold'] = rates['RSI'] < 30

    print(rates[['time', 'close', 'RSI', 'RSI_Overbought', 'RSI_Oversold']].tail())

MACD (Moving Average Convergence Divergence)

def calculate_macd(data, fast_period=12, slow_period=26, signal_period=9):
    """
    Tính MACD

    Args:
        data: Series chứa giá close
        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ỳ signal line (mặc định 9)

    Returns:
        dict: Dictionary chứa MACD line, signal line, và histogram
    """
    ema_fast = calculate_ema(data, fast_period)
    ema_slow = calculate_ema(data, slow_period)

    macd_line = ema_fast - ema_slow
    signal_line = calculate_ema(macd_line, signal_period)
    histogram = macd_line - signal_line

    return {
        'macd': macd_line,
        'signal': signal_line,
        'histogram': histogram
    }

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    macd_data = calculate_macd(rates['close'])
    rates['MACD'] = macd_data['macd']
    rates['MACD_Signal'] = macd_data['signal']
    rates['MACD_Histogram'] = macd_data['histogram']

    # Tín hiệu giao dịch
    rates['MACD_Buy'] = (rates['MACD'] > rates['MACD_Signal']) & \
                        (rates['MACD'].shift(1) <= rates['MACD_Signal'].shift(1))
    rates['MACD_Sell'] = (rates['MACD'] < rates['MACD_Signal']) & \
                         (rates['MACD'].shift(1) >= rates['MACD_Signal'].shift(1))

    print(rates[['time', 'close', 'MACD', 'MACD_Signal', 'MACD_Buy', 'MACD_Sell']].tail())

Bollinger Bands

def calculate_bollinger_bands(data, period=20, num_std=2):
    """
    Tính Bollinger Bands

    Args:
        data: Series chứa giá close
        period: Chu kỳ (mặc định 20)
        num_std: Số độ lệch chuẩn (mặc định 2)

    Returns:
        dict: Dictionary chứa upper band, middle band (SMA), và lower band
    """
    sma = calculate_sma(data, period)
    std = data.rolling(window=period).std()

    upper_band = sma + (num_std * std)
    lower_band = sma - (num_std * std)

    return {
        'upper': upper_band,
        'middle': sma,
        'lower': lower_band
    }

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    bb = calculate_bollinger_bands(rates['close'], period=20, num_std=2)
    rates['BB_Upper'] = bb['upper']
    rates['BB_Middle'] = bb['middle']
    rates['BB_Lower'] = bb['lower']

    # Tín hiệu giao dịch
    rates['BB_Buy'] = rates['close'] < rates['BB_Lower']  # Giá chạm lower band
    rates['BB_Sell'] = rates['close'] > rates['BB_Upper']  # Giá chạm upper band

    print(rates[['time', 'close', 'BB_Upper', 'BB_Middle', 'BB_Lower', 'BB_Buy', 'BB_Sell']].tail())

Stochastic Oscillator

def calculate_stochastic(high, low, close, k_period=14, d_period=3):
    """
    Tính Stochastic Oscillator

    Args:
        high: Series chứa giá high
        low: Series chứa giá low
        close: Series chứa giá close
        k_period: Chu kỳ %K (mặc định 14)
        d_period: Chu kỳ %D (mặc định 3)

    Returns:
        dict: Dictionary chứa %K và %D
    """
    lowest_low = low.rolling(window=k_period).min()
    highest_high = high.rolling(window=k_period).max()

    k_percent = 100 * ((close - lowest_low) / (highest_high - lowest_low))
    d_percent = k_percent.rolling(window=d_period).mean()

    return {
        'k': k_percent,
        'd': d_percent
    }

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    stoch = calculate_stochastic(rates['high'], rates['low'], rates['close'])
    rates['Stoch_K'] = stoch['k']
    rates['Stoch_D'] = stoch['d']

    # Tín hiệu giao dịch
    rates['Stoch_Buy'] = (rates['Stoch_K'] < 20) & (rates['Stoch_K'] > rates['Stoch_D'])
    rates['Stoch_Sell'] = (rates['Stoch_K'] > 80) & (rates['Stoch_K'] < rates['Stoch_D'])

    print(rates[['time', 'close', 'Stoch_K', 'Stoch_D', 'Stoch_Buy', 'Stoch_Sell']].tail())

ATR (Average True Range)

def calculate_atr(high, low, close, period=14):
    """
    Tính ATR (Average True Range)

    Args:
        high: Series chứa giá high
        low: Series chứa giá low
        close: Series chứa giá close
        period: Chu kỳ (mặc định 14)

    Returns:
        Series: ATR values
    """
    high_low = high - low
    high_close = np.abs(high - close.shift())
    low_close = np.abs(low - close.shift())

    true_range = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
    atr = true_range.rolling(window=period).mean()

    return atr

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    rates['ATR'] = calculate_atr(rates['high'], rates['low'], rates['close'], period=14)

    # Sử dụng ATR để tính stop loss
    current_price = rates['close'].iloc[-1]
    current_atr = rates['ATR'].iloc[-1]
    stop_loss_long = current_price - (2 * current_atr)
    stop_loss_short = current_price + (2 * current_atr)

    print(f"Current Price: {current_price:.5f}")
    print(f"ATR: {current_atr:.5f}")
    print(f"Stop Loss Long: {stop_loss_long:.5f}")
    print(f"Stop Loss Short: {stop_loss_short:.5f}")

ADX (Average Directional Index)

def calculate_adx(high, low, close, period=14):
    """
    Tính ADX (Average Directional Index)

    Args:
        high: Series chứa giá high
        low: Series chứa giá low
        close: Series chứa giá close
        period: Chu kỳ (mặc định 14)

    Returns:
        dict: Dictionary chứa ADX, +DI, và -DI
    """
    # Tính True Range
    atr = calculate_atr(high, low, close, period)

    # Tính +DM và -DM
    plus_dm = high.diff()
    minus_dm = -low.diff()

    plus_dm[plus_dm < 0] = 0
    minus_dm[minus_dm < 0] = 0

    # Tính +DI và -DI
    plus_di = 100 * (plus_dm.rolling(window=period).mean() / atr)
    minus_di = 100 * (minus_dm.rolling(window=period).mean() / atr)

    # Tính DX
    dx = 100 * np.abs(plus_di - minus_di) / (plus_di + minus_di)

    # Tính ADX
    adx = dx.rolling(window=period).mean()

    return {
        'adx': adx,
        'plus_di': plus_di,
        'minus_di': minus_di
    }

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    adx_data = calculate_adx(rates['high'], rates['low'], rates['close'])
    rates['ADX'] = adx_data['adx']
    rates['Plus_DI'] = adx_data['plus_di']
    rates['Minus_DI'] = adx_data['minus_di']

    # Tín hiệu giao dịch (ADX > 25 = xu hướng mạnh)
    rates['ADX_Strong'] = rates['ADX'] > 25
    rates['ADX_Buy'] = (rates['ADX'] > 25) & (rates['Plus_DI'] > rates['Minus_DI'])
    rates['ADX_Sell'] = (rates['ADX'] > 25) & (rates['Plus_DI'] < rates['Minus_DI'])

    print(rates[['time', 'close', 'ADX', 'Plus_DI', 'Minus_DI', 'ADX_Buy', 'ADX_Sell']].tail())

CCI (Commodity Channel Index)

def calculate_cci(high, low, close, period=20):
    """
    Tính CCI (Commodity Channel Index)

    Args:
        high: Series chứa giá high
        low: Series chứa giá low
        close: Series chứa giá close
        period: Chu kỳ (mặc định 20)

    Returns:
        Series: CCI values
    """
    typical_price = (high + low + close) / 3
    sma_tp = typical_price.rolling(window=period).mean()
    mean_deviation = typical_price.rolling(window=period).apply(
        lambda x: np.mean(np.abs(x - x.mean())), raw=True
    )

    cci = (typical_price - sma_tp) / (0.015 * mean_deviation)

    return cci

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    rates['CCI'] = calculate_cci(rates['high'], rates['low'], rates['close'], period=20)

    # Tín hiệu giao dịch
    rates['CCI_Buy'] = rates['CCI'] < -100  # Oversold
    rates['CCI_Sell'] = rates['CCI'] > 100   # Overbought

    print(rates[['time', 'close', 'CCI', 'CCI_Buy', 'CCI_Sell']].tail())

Williams %R

def calculate_williams_r(high, low, close, period=14):
    """
    Tính Williams %R

    Args:
        high: Series chứa giá high
        low: Series chứa giá low
        close: Series chứa giá close
        period: Chu kỳ (mặc định 14)

    Returns:
        Series: Williams %R values (-100 to 0)
    """
    highest_high = high.rolling(window=period).max()
    lowest_low = low.rolling(window=period).min()

    williams_r = -100 * ((highest_high - close) / (highest_high - lowest_low))

    return williams_r

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    rates['Williams_R'] = calculate_williams_r(rates['high'], rates['low'], rates['close'])

    # Tín hiệu giao dịch
    rates['Williams_Buy'] = rates['Williams_R'] < -80  # Oversold
    rates['Williams_Sell'] = rates['Williams_R'] > -20  # Overbought

    print(rates[['time', 'close', 'Williams_R', 'Williams_Buy', 'Williams_Sell']].tail())

Class Quản Lý Chỉ Báo

class MT5IndicatorCalculator:
    """Class quản lý tính toán chỉ báo từ dữ liệu MT5"""

    def __init__(self, symbol, timeframe):
        """
        Khởi tạo

        Args:
            symbol: Tên symbol
            timeframe: Khung thời gian
        """
        self.symbol = symbol
        self.timeframe = timeframe
        self.data = None

    def load_data(self, count=1000):
        """Tải dữ liệu từ MT5"""
        self.data = get_rates(self.symbol, self.timeframe, count=count)
        return self.data is not None

    def calculate_all_indicators(self):
        """Tính tất cả các chỉ báo"""
        if self.data is None:
            print("Chưa có dữ liệu. Gọi load_data() trước.")
            return

        # Moving Averages
        self.data['SMA20'] = calculate_sma(self.data['close'], 20)
        self.data['SMA50'] = calculate_sma(self.data['close'], 50)
        self.data['EMA12'] = calculate_ema(self.data['close'], 12)
        self.data['EMA26'] = calculate_ema(self.data['close'], 26)

        # RSI
        self.data['RSI'] = calculate_rsi(self.data['close'], 14)

        # MACD
        macd_data = calculate_macd(self.data['close'])
        self.data['MACD'] = macd_data['macd']
        self.data['MACD_Signal'] = macd_data['signal']
        self.data['MACD_Histogram'] = macd_data['histogram']

        # Bollinger Bands
        bb = calculate_bollinger_bands(self.data['close'])
        self.data['BB_Upper'] = bb['upper']
        self.data['BB_Middle'] = bb['middle']
        self.data['BB_Lower'] = bb['lower']

        # Stochastic
        stoch = calculate_stochastic(self.data['high'], self.data['low'], self.data['close'])
        self.data['Stoch_K'] = stoch['k']
        self.data['Stoch_D'] = stoch['d']

        # ATR
        self.data['ATR'] = calculate_atr(self.data['high'], self.data['low'], self.data['close'])

        # ADX
        adx_data = calculate_adx(self.data['high'], self.data['low'], self.data['close'])
        self.data['ADX'] = adx_data['adx']
        self.data['Plus_DI'] = adx_data['plus_di']
        self.data['Minus_DI'] = adx_data['minus_di']

        # CCI
        self.data['CCI'] = calculate_cci(self.data['high'], self.data['low'], self.data['close'])

        # Williams %R
        self.data['Williams_R'] = calculate_williams_r(self.data['high'], self.data['low'], self.data['close'])

    def get_signals(self):
        """Lấy tín hiệu giao dịch từ các chỉ báo"""
        if self.data is None:
            return None

        latest = self.data.iloc[-1]

        signals = {
            'buy_signals': 0,
            'sell_signals': 0,
            'indicators': {}
        }

        # RSI
        if latest['RSI'] < 30:
            signals['buy_signals'] += 1
            signals['indicators']['RSI'] = 'BUY'
        elif latest['RSI'] > 70:
            signals['sell_signals'] += 1
            signals['indicators']['RSI'] = 'SELL'

        # MACD
        if latest['MACD'] > latest['MACD_Signal']:
            signals['buy_signals'] += 1
            signals['indicators']['MACD'] = 'BUY'
        else:
            signals['sell_signals'] += 1
            signals['indicators']['MACD'] = 'SELL'

        # Bollinger Bands
        if latest['close'] < latest['BB_Lower']:
            signals['buy_signals'] += 1
            signals['indicators']['BB'] = 'BUY'
        elif latest['close'] > latest['BB_Upper']:
            signals['sell_signals'] += 1
            signals['indicators']['BB'] = 'SELL'

        # Stochastic
        if latest['Stoch_K'] < 20 and latest['Stoch_K'] > latest['Stoch_D']:
            signals['buy_signals'] += 1
            signals['indicators']['Stoch'] = 'BUY'
        elif latest['Stoch_K'] > 80 and latest['Stoch_K'] < latest['Stoch_D']:
            signals['sell_signals'] += 1
            signals['indicators']['Stoch'] = 'SELL'

        return signals

# Sử dụng
calculator = MT5IndicatorCalculator("EURUSD", mt5.TIMEFRAME_H1)
if calculator.load_data(count=200):
    calculator.calculate_all_indicators()
    signals = calculator.get_signals()
    print(f"Buy Signals: {signals['buy_signals']}")
    print(f"Sell Signals: {signals['sell_signals']}")
    print(f"Indicators: {signals['indicators']}")

Tích Hợp Vào Bot Auto Trading

Hàm Tạo Tín Hiệu Giao Dịch

def generate_trading_signals(symbol, timeframe):
    """
    Tạo tín hiệu giao dịch từ các chỉ báo

    Args:
        symbol: Tên symbol
        timeframe: Khung thời gian

    Returns:
        dict: Dictionary chứa tín hiệu giao dịch
    """
    calculator = MT5IndicatorCalculator(symbol, timeframe)

    if not calculator.load_data(count=200):
        return None

    calculator.calculate_all_indicators()
    signals = calculator.get_signals()

    latest = calculator.data.iloc[-1]

    # Quyết định giao dịch
    action = 'HOLD'
    confidence = 0

    if signals['buy_signals'] > signals['sell_signals']:
        action = 'BUY'
        confidence = signals['buy_signals'] / (signals['buy_signals'] + signals['sell_signals'])
    elif signals['sell_signals'] > signals['buy_signals']:
        action = 'SELL'
        confidence = signals['sell_signals'] / (signals['buy_signals'] + signals['sell_signals'])

    return {
        'symbol': symbol,
        'action': action,
        'confidence': confidence,
        'price': latest['close'],
        'signals': signals,
        'indicators': {
            'RSI': latest['RSI'],
            'MACD': latest['MACD'],
            'ADX': latest['ADX'],
            'ATR': latest['ATR']
        }
    }

# Sử dụng
signal = generate_trading_signals("EURUSD", mt5.TIMEFRAME_H1)
if signal:
    print(f"Symbol: {signal['symbol']}")
    print(f"Action: {signal['action']}")
    print(f"Confidence: {signal['confidence']:.2%}")
    print(f"Price: {signal['price']:.5f}")
    print(f"RSI: {signal['indicators']['RSI']:.2f}")
    print(f"ADX: {signal['indicators']['ADX']:.2f}")

Hàm Tính Stop Loss và Take Profit

def calculate_stop_loss_take_profit(price, atr, direction='BUY', risk_reward_ratio=2):
    """
    Tính Stop Loss và Take Profit dựa trên ATR

    Args:
        price: Giá vào lệnh
        atr: Giá trị ATR
        direction: Hướng giao dịch ('BUY' hoặc 'SELL')
        risk_reward_ratio: Tỷ lệ Risk/Reward (mặc định 2)

    Returns:
        dict: Dictionary chứa stop loss và take profit
    """
    if direction == 'BUY':
        stop_loss = price - (2 * atr)
        take_profit = price + (2 * atr * risk_reward_ratio)
    else:  # SELL
        stop_loss = price + (2 * atr)
        take_profit = price - (2 * atr * risk_reward_ratio)

    return {
        'stop_loss': stop_loss,
        'take_profit': take_profit,
        'risk': abs(price - stop_loss),
        'reward': abs(take_profit - price)
    }

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    rates['ATR'] = calculate_atr(rates['high'], rates['low'], rates['close'])

    current_price = rates['close'].iloc[-1]
    current_atr = rates['ATR'].iloc[-1]

    sl_tp = calculate_stop_loss_take_profit(current_price, current_atr, direction='BUY')
    print(f"Price: {current_price:.5f}")
    print(f"Stop Loss: {sl_tp['stop_loss']:.5f}")
    print(f"Take Profit: {sl_tp['take_profit']:.5f}")
    print(f"Risk: {sl_tp['risk']:.5f}")
    print(f"Reward: {sl_tp['reward']:.5f}")
    print(f"Risk/Reward Ratio: {sl_tp['risk']/sl_tp['reward']:.2f}")

Kết Luận

Bài viết đã hướng dẫn cách tính toán các chỉ báo kỹ thuật trong Forex MT5 với Python để sử dụng trong bot auto trading. Các chỉ báo bao gồm:

Moving Averages (SMA, EMA, WMA)
RSI (Relative Strength Index)
MACD (Moving Average Convergence Divergence)
Bollinger Bands
Stochastic Oscillator
ATR (Average True Range)
ADX (Average Directional Index)
CCI (Commodity Channel Index)
Williams %R

Lưu ý quan trọng:

Luôn kết hợp nhiều chỉ báo để tăng độ chính xác
Sử dụng ATR để tính stop loss và take profit động
Backtest chiến lược trước khi sử dụng trên tài khoản thật
Quản lý rủi ro cẩn thận

Tài Liệu Tham Khảo

Bài viết được biên soạn bởi CoinGetMarket – Nền tảng giáo dục về crypto và trading bot.

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

Cấu trúc thư mục của một dự án Flutter

So sánh Flutter và React Native