| Cách Viết Hàm Để Lấy Dữ Liệu Thị Trường Forex MT5 Bằng Python

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

Hướng dẫn chi tiết cách viết hàm để lấy dữ liệu thị trường Forex từ MetaTrader 5 bằng Python, bao gồm giá real-time, dữ liệu lịch sử và tick data.

Cách Viết Hàm Để Lấy Dữ Liệu Thị Trường Forex MT5 Bằng Python

Lấy dữ liệu thị trường là bước quan trọng trong phân tích và giao dịch tự động. Bài viết này sẽ hướng dẫn bạn cách viết các hàm để lấy dữ liệu thị trường Forex từ MetaTrader 5 bằng Python.

Cài Đặt và Kết Nối

import MetaTrader5 as mt5
import pandas as pd
from datetime import datetime, timedelta

# Kết nối với MT5
def connect_mt5():
    """Kết nối với MetaTrader 5"""
    if not mt5.initialize():
        print("Khởi tạo MT5 thất bại, error code =", mt5.last_error())
        return False
    return True

# Sử dụng
if not connect_mt5():
    exit()

Lấy Giá Real-Time

Hàm Lấy Giá Bid/Ask

def get_symbol_info(symbol):
    """
    Lấy thông tin symbol

    Args:
        symbol: Tên symbol (ví dụ: "EURUSD")

    Returns:
        dict: Dictionary chứa thông tin symbol hoặc None nếu lỗi
    """
    symbol_info = mt5.symbol_info(symbol)

    if symbol_info is None:
        print(f"Không tìm thấy symbol {symbol}, error code =", mt5.last_error())
        return None

    if not symbol_info.visible:
        print(f"Symbol {symbol} không hiển thị, đang thử kích hoạt...")
        if not mt5.symbol_select(symbol, True):
            print(f"Không thể kích hoạt symbol {symbol}")
            return None

    return {
        'name': symbol_info.name,
        'bid': symbol_info.bid,
        'ask': symbol_info.ask,
        'spread': symbol_info.ask - symbol_info.bid,
        'point': symbol_info.point,
        'digits': symbol_info.digits,
        'volume_min': symbol_info.volume_min,
        'volume_max': symbol_info.volume_max,
        'volume_step': symbol_info.volume_step,
        'trade_mode': symbol_info.trade_mode,
        'trade_stops_level': symbol_info.trade_stops_level,
        'trade_freeze_level': symbol_info.trade_freeze_level
    }

# Sử dụng
eurusd_info = get_symbol_info("EURUSD")
if eurusd_info:
    print(f"EURUSD - Bid: {eurusd_info['bid']}, Ask: {eurusd_info['ask']}")
    print(f"Spread: {eurusd_info['spread']}")

Hàm Lấy Giá Tick

def get_tick(symbol):
    """
    Lấy tick giá mới nhất

    Args:
        symbol: Tên symbol

    Returns:
        dict: Dictionary chứa thông tin tick hoặc None nếu lỗi
    """
    tick = mt5.symbol_info_tick(symbol)

    if tick is None:
        print(f"Không thể lấy tick cho {symbol}, error code =", mt5.last_error())
        return None

    return {
        'time': datetime.fromtimestamp(tick.time),
        'bid': tick.bid,
        'ask': tick.ask,
        'last': tick.last,
        'volume': tick.volume,
        'time_msc': tick.time_msc,
        'flags': tick.flags,
        'time_digits': tick.time_digits
    }

# Sử dụng
tick = get_tick("EURUSD")
if tick:
    print(f"Time: {tick['time']}")
    print(f"Bid: {tick['bid']}, Ask: {tick['ask']}")

Hàm Lấy Nhiều Tick

def get_ticks(symbol, count=1000):
    """
    Lấy nhiều tick gần nhất

    Args:
        symbol: Tên symbol
        count: Số lượng tick cần lấy

    Returns:
        DataFrame: DataFrame chứa tick data hoặc None nếu lỗi
    """
    ticks = mt5.copy_ticks_from(symbol, datetime.now(), count, mt5.COPY_TICKS_ALL)

    if ticks is None:
        print(f"Không thể lấy ticks cho {symbol}, error code =", mt5.last_error())
        return None

    # Chuyển đổi sang DataFrame
    ticks_df = pd.DataFrame(ticks)
    ticks_df['time'] = pd.to_datetime(ticks_df['time'], unit='s')

    return ticks_df

# Sử dụng
ticks = get_ticks("EURUSD", count=100)
if ticks is not None:
    print(ticks.head())

Lấy Dữ Liệu Lịch Sử (OHLC)

Hàm Lấy Dữ Liệu Theo Khung Thời Gian

def get_rates(symbol, timeframe, count=1000, start_time=None, end_time=None):
    """
    Lấy dữ liệu OHLC theo khung thời gian

    Args:
        symbol: Tên symbol
        timeframe: Khung thời gian (mt5.TIMEFRAME_M1, M5, M15, H1, D1, etc.)
        count: Số lượng nến cần lấy (nếu không có start_time/end_time)
        start_time: Thời gian bắt đầu (datetime)
        end_time: Thời gian kết thúc (datetime)

    Returns:
        DataFrame: DataFrame chứa OHLC data hoặc None nếu lỗi
    """
    if start_time and end_time:
        # Lấy dữ liệu theo khoảng thời gian
        rates = mt5.copy_rates_range(symbol, timeframe, start_time, end_time)
    elif start_time:
        # Lấy dữ liệu từ start_time đến hiện tại
        rates = mt5.copy_rates_from(symbol, timeframe, start_time, count)
    else:
        # Lấy dữ liệu gần nhất
        rates = mt5.copy_rates_from_pos(symbol, timeframe, 0, count)

    if rates is None:
        print(f"Không thể lấy rates cho {symbol}, error code =", mt5.last_error())
        return None

    # Chuyển đổi sang DataFrame
    rates_df = pd.DataFrame(rates)
    rates_df['time'] = pd.to_datetime(rates_df['time'], unit='s')

    return rates_df

# Sử dụng
# Lấy 100 nến H1 gần nhất
eurusd_h1 = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=100)
if eurusd_h1 is not None:
    print(eurusd_h1.head())
    print(f"\nTổng số nến: {len(eurusd_h1)}")

# Lấy dữ liệu theo khoảng thời gian
start = datetime(2024, 1, 1)
end = datetime(2024, 1, 31)
eurusd_range = get_rates("EURUSD", mt5.TIMEFRAME_D1, start_time=start, end_time=end)

Hàm Lấy Dữ Liệu Nhiều Symbol

def get_multiple_symbols_rates(symbols, timeframe, count=100):
    """
    Lấy dữ liệu cho nhiều symbol cùng lúc

    Args:
        symbols: List các symbol
        timeframe: Khung thời gian
        count: Số lượng nến

    Returns:
        dict: Dictionary với key là symbol, value là DataFrame
    """
    data = {}

    for symbol in symbols:
        rates = get_rates(symbol, timeframe, count=count)
        if rates is not None:
            data[symbol] = rates

    return data

# Sử dụng
symbols = ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD"]
multi_data = get_multiple_symbols_rates(symbols, mt5.TIMEFRAME_H1, count=100)
for symbol, df in multi_data.items():
    print(f"{symbol}: {len(df)} nến")

Lấy Thông Tin Symbol

Hàm Lấy Tất Cả Symbol

def get_all_symbols(group="*"):
    """
    Lấy danh sách tất cả symbol

    Args:
        group: Filter theo group (ví dụ: "EUR*", "*USD", "*")

    Returns:
        list: List các symbol
    """
    symbols = mt5.symbols_get(group)

    if symbols is None:
        print("Không thể lấy danh sách symbol, error code =", mt5.last_error())
        return []

    return [symbol.name for symbol in symbols]

# Sử dụng
all_symbols = get_all_symbols()
print(f"Tổng số symbol: {len(all_symbols)}")

# Lấy chỉ các cặp EUR
eur_symbols = get_all_symbols("EUR*")
print(f"Các cặp EUR: {eur_symbols[:10]}")

Hàm Lấy Thông Tin Chi Tiết Symbol

def get_symbol_details(symbol):
    """
    Lấy thông tin chi tiết của symbol

    Args:
        symbol: Tên symbol

    Returns:
        dict: Dictionary chứa thông tin chi tiết
    """
    symbol_info = mt5.symbol_info(symbol)

    if symbol_info is None:
        return None

    return {
        'name': symbol_info.name,
        'description': symbol_info.description,
        'currency_base': symbol_info.currency_base,
        'currency_profit': symbol_info.currency_profit,
        'currency_margin': symbol_info.currency_margin,
        'bid': symbol_info.bid,
        'ask': symbol_info.ask,
        'spread': symbol_info.ask - symbol_info.bid,
        'point': symbol_info.point,
        'digits': symbol_info.digits,
        'volume_min': symbol_info.volume_min,
        'volume_max': symbol_info.volume_max,
        'volume_step': symbol_info.volume_step,
        'volume_limit': symbol_info.volume_limit,
        'trade_mode': symbol_info.trade_mode,
        'trade_stops_level': symbol_info.trade_stops_level,
        'trade_freeze_level': symbol_info.trade_freeze_level,
        'swap_mode': symbol_info.swap_mode,
        'swap_long': symbol_info.swap_long,
        'swap_short': symbol_info.swap_short,
        'margin_initial': symbol_info.margin_initial,
        'margin_maintenance': symbol_info.margin_maintenance
    }

# Sử dụng
details = get_symbol_details("EURUSD")
if details:
    print(f"Symbol: {details['name']}")
    print(f"Base Currency: {details['currency_base']}")
    print(f"Profit Currency: {details['currency_profit']}")
    print(f"Min Volume: {details['volume_min']}")
    print(f"Max Volume: {details['volume_max']}")

Tính Toán Chỉ Báo Kỹ Thuật

Hàm Tính Moving Average

def calculate_ma(rates_df, period=20, ma_type='SMA'):
    """
    Tính Moving Average

    Args:
        rates_df: DataFrame chứa OHLC data
        period: Chu kỳ
        ma_type: Loại MA ('SMA', 'EMA', 'WMA')

    Returns:
        Series: Series chứa giá trị MA
    """
    if ma_type == 'SMA':
        return rates_df['close'].rolling(window=period).mean()
    elif ma_type == 'EMA':
        return rates_df['close'].ewm(span=period, adjust=False).mean()
    elif ma_type == 'WMA':
        weights = np.arange(1, period + 1)
        return rates_df['close'].rolling(window=period).apply(
            lambda x: np.dot(x, weights) / weights.sum(), raw=True
        )
    else:
        return rates_df['close'].rolling(window=period).mean()

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=200)
if rates is not None:
    rates['MA20'] = calculate_ma(rates, period=20, ma_type='SMA')
    rates['EMA12'] = calculate_ma(rates, period=12, ma_type='EMA')
    print(rates[['time', 'close', 'MA20', 'EMA12']].tail())

Hàm Tính RSI

import numpy as np

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

    Args:
        rates_df: DataFrame chứa OHLC data
        period: Chu kỳ (mặc định 14)

    Returns:
        Series: Series chứa giá trị RSI
    """
    delta = rates_df['close'].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, period=14)
    print(rates[['time', 'close', 'RSI']].tail())

Lấy Dữ Liệu Real-Time

Hàm Monitor Giá Real-Time

import time

def monitor_price(symbol, interval=1):
    """
    Monitor giá real-time

    Args:
        symbol: Tên symbol
        interval: Khoảng thời gian giữa các lần cập nhật (giây)
    """
    print(f"Bắt đầu monitor {symbol}...")
    print("Nhấn Ctrl+C để dừng\n")

    try:
        while True:
            tick = get_tick(symbol)
            if tick:
                print(f"[{tick['time']}] {symbol} - "
                      f"Bid: {tick['bid']:.5f} | "
                      f"Ask: {tick['ask']:.5f} | "
                      f"Spread: {tick['ask'] - tick['bid']:.5f}")

            time.sleep(interval)

    except KeyboardInterrupt:
        print("\nDừng monitor...")

# Sử dụng
# monitor_price("EURUSD", interval=1)

Hàm Lấy Giá Nhiều Symbol

def get_multiple_prices(symbols):
    """
    Lấy giá của nhiều symbol cùng lúc

    Args:
        symbols: List các symbol

    Returns:
        dict: Dictionary với key là symbol, value là dict chứa bid/ask
    """
    prices = {}

    for symbol in symbols:
        tick = get_tick(symbol)
        if tick:
            prices[symbol] = {
                'bid': tick['bid'],
                'ask': tick['ask'],
                'spread': tick['ask'] - tick['bid'],
                'time': tick['time']
            }

    return prices

# Sử dụng
symbols = ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD"]
prices = get_multiple_prices(symbols)
for symbol, price_info in prices.items():
    print(f"{symbol}: Bid={price_info['bid']:.5f}, Ask={price_info['ask']:.5f}")

Lưu Dữ Liệu

Hàm Lưu Dữ Liệu Vào CSV

def save_to_csv(rates_df, filename):
    """
    Lưu dữ liệu vào file CSV

    Args:
        rates_df: DataFrame cần lưu
        filename: Tên file
    """
    rates_df.to_csv(filename, index=False)
    print(f"Đã lưu dữ liệu vào {filename}")

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=1000)
if rates is not None:
    save_to_csv(rates, "EURUSD_H1.csv")

Hàm Lưu Dữ Liệu Vào Database

import sqlite3

def save_to_database(rates_df, symbol, timeframe, db_name="mt5_data.db"):
    """
    Lưu dữ liệu vào SQLite database

    Args:
        rates_df: DataFrame cần lưu
        symbol: Tên symbol
        timeframe: Khung thời gian
        db_name: Tên database
    """
    conn = sqlite3.connect(db_name)

    table_name = f"{symbol}_{timeframe}".replace(" ", "_")
    rates_df.to_sql(table_name, conn, if_exists='replace', index=False)

    conn.close()
    print(f"Đã lưu dữ liệu vào database: {table_name}")

# Sử dụng
rates = get_rates("EURUSD", mt5.TIMEFRAME_H1, count=1000)
if rates is not None:
    save_to_database(rates, "EURUSD", "H1")

Class Quản Lý Dữ Liệu Thị Trường

class MT5MarketData:
    """Class quản lý dữ liệu thị trường MT5"""

    def __init__(self):
        """Khởi tạo và kết nối với MT5"""
        if not mt5.initialize():
            raise Exception(f"Không thể khởi tạo MT5, error code = {mt5.last_error()}")
        self.connected = True

    def __del__(self):
        """Đóng kết nối khi object bị hủy"""
        if self.connected:
            mt5.shutdown()

    def get_current_price(self, symbol):
        """Lấy giá hiện tại"""
        tick = get_tick(symbol)
        if tick:
            return {'bid': tick['bid'], 'ask': tick['ask']}
        return None

    def get_historical_data(self, symbol, timeframe, count=1000):
        """Lấy dữ liệu lịch sử"""
        return get_rates(symbol, timeframe, count=count)

    def get_ticks_data(self, symbol, count=1000):
        """Lấy tick data"""
        return get_ticks(symbol, count=count)

    def get_symbol_info(self, symbol):
        """Lấy thông tin symbol"""
        return get_symbol_info(symbol)

    def monitor_symbols(self, symbols, callback, interval=1):
        """
        Monitor nhiều symbol

        Args:
            symbols: List các symbol
            callback: Hàm callback được gọi mỗi khi có dữ liệu mới
            interval: Khoảng thời gian (giây)
        """
        import time
        try:
            while True:
                for symbol in symbols:
                    tick = get_tick(symbol)
                    if tick:
                        callback(symbol, tick)
                time.sleep(interval)
        except KeyboardInterrupt:
            print("Dừng monitor...")

# Sử dụng
def price_callback(symbol, tick):
    """Callback function"""
    print(f"{symbol}: Bid={tick['bid']:.5f}, Ask={tick['ask']:.5f}")

# market_data = MT5MarketData()
# market_data.monitor_symbols(["EURUSD", "GBPUSD"], price_callback, interval=1)

Ví Dụ Sử Dụng Thực Tế

Script Thu Thập Dữ Liệu

from datetime import datetime, timedelta

def collect_daily_data(symbols, days=30):
    """
    Thu thập dữ liệu hàng ngày cho nhiều symbol

    Args:
        symbols: List các symbol
        days: Số ngày cần thu thập
    """
    end_time = datetime.now()
    start_time = end_time - timedelta(days=days)

    for symbol in symbols:
        print(f"Đang thu thập dữ liệu cho {symbol}...")
        rates = get_rates(symbol, mt5.TIMEFRAME_D1, 
                         start_time=start_time, end_time=end_time)

        if rates is not None:
            filename = f"{symbol}_D1_{days}days.csv"
            save_to_csv(rates, filename)
            print(f"Đã lưu {len(rates)} nến cho {symbol}")
        else:
            print(f"Không thể lấy dữ liệu cho {symbol}")

# Sử dụng
symbols = ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD", "USDCAD"]
collect_daily_data(symbols, days=90)

Script Phân Tích Xu Hướng

def analyze_trend(symbol, timeframe, ma_period=20):
    """
    Phân tích xu hướng dựa trên Moving Average

    Args:
        symbol: Tên symbol
        timeframe: Khung thời gian
        ma_period: Chu kỳ MA
    """
    rates = get_rates(symbol, timeframe, count=200)

    if rates is None:
        return None

    # Tính MA
    rates['MA'] = calculate_ma(rates, period=ma_period)

    # Xác định xu hướng
    current_price = rates['close'].iloc[-1]
    current_ma = rates['MA'].iloc[-1]
    prev_price = rates['close'].iloc[-2]
    prev_ma = rates['MA'].iloc[-2]

    if current_price > current_ma and prev_price <= prev_ma:
        trend = "UPTREND (Giá vượt lên MA)"
    elif current_price < current_ma and prev_price >= prev_ma:
        trend = "DOWNTREND (Giá xuống dưới MA)"
    elif current_price > current_ma:
        trend = "UPTREND"
    else:
        trend = "DOWNTREND"

    return {
        'symbol': symbol,
        'current_price': current_price,
        'ma': current_ma,
        'trend': trend,
        'distance_from_ma': ((current_price - current_ma) / current_ma) * 100
    }

# Sử dụng
analysis = analyze_trend("EURUSD", mt5.TIMEFRAME_H1, ma_period=20)
if analysis:
    print(f"Symbol: {analysis['symbol']}")
    print(f"Giá hiện tại: {analysis['current_price']:.5f}")
    print(f"MA20: {analysis['ma']:.5f}")
    print(f"Xu hướng: {analysis['trend']}")
    print(f"Khoảng cách từ MA: {analysis['distance_from_ma']:.2f}%")

Kết Luận

Bài viết đã hướng dẫn cách viết các hàm để lấy dữ liệu thị trường Forex MT5 bằng Python. Các hàm này giúp bạn:

Lấy giá real-time (bid/ask)
Lấy dữ liệu lịch sử (OHLC)
Lấy tick data
Lấy thông tin symbol
Tính toán chỉ báo kỹ thuật
Monitor giá real-time
Lưu dữ liệu vào file hoặc database

Lưu ý quan trọng:

Luôn kiểm tra kết quả trả về từ các hàm MT5
Xử lý lỗi đúng cách
Sử dụng pandas để xử lý dữ liệu hiệu quả
Lưu dữ liệu định kỳ để phân tích sau

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