Xây dựng Bot Auto Trading với dữ liệu YFinance

| Xây dựng Bot Auto Trading với dữ liệu YFinance

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

Xây dựng Bot Auto Trading với dữ liệu YFinance bằng Python

YFinance (Yahoo Finance) là một thư viện Python mạnh mẽ cho phép lấy dữ liệu thị trường chứng khoán miễn phí từ Yahoo Finance. Trong bài viết này, chúng ta sẽ học cách sử dụng yfinance để xây dựng một bot giao dịch tự động hoàn chỉnh.

YFinance là gì?

YFinance là một thư viện Python không chính thức để tải dữ liệu từ Yahoo Finance. Nó cung cấp:

Dữ liệu giá real-time và lịch sử: Cổ phiếu, ETF, chỉ số, tiền điện tử
Dữ liệu tài chính: Báo cáo tài chính, phân tích kỹ thuật
Dữ liệu thị trường: Volume, market cap, P/E ratio
Hoàn toàn miễn phí: Không cần API key

Cài đặt YFinance

pip install yfinance pandas numpy matplotlib

Hoặc với conda:

conda install -c conda-forge yfinance

Lấy dữ liệu cơ bản với YFinance

1. Lấy dữ liệu giá cổ phiếu

import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta

# Lấy dữ liệu cho một cổ phiếu
ticker = yf.Ticker("AAPL")  # Apple Inc.

# Lấy dữ liệu lịch sử
data = ticker.history(period="1y")  # 1 năm
print(data.head())

# Hoặc chỉ định khoảng thời gian cụ thể
start_date = datetime.now() - timedelta(days=365)
end_date = datetime.now()
data = ticker.history(start=start_date, end=end_date)

# Lấy dữ liệu với interval khác nhau
data_1d = ticker.history(period="1mo", interval="1d")  # 1 tháng, mỗi ngày
data_1h = ticker.history(period="5d", interval="1h")  # 5 ngày, mỗi giờ
data_1m = ticker.history(period="1d", interval="1m")  # 1 ngày, mỗi phút

2. Lấy thông tin công ty

# Lấy thông tin chi tiết về công ty
info = ticker.info
print(f"Tên công ty: {info['longName']}")
print(f"Ngành: {info['sector']}")
print(f"Market Cap: {info['marketCap']}")
print(f"P/E Ratio: {info.get('trailingPE', 'N/A')}")
print(f"Dividend Yield: {info.get('dividendYield', 'N/A')}")

# Lấy dữ liệu tài chính
financials = ticker.financials
quarterly_financials = ticker.quarterly_financials
balance_sheet = ticker.balance_sheet
cashflow = ticker.cashflow

3. Lấy dữ liệu nhiều cổ phiếu cùng lúc

# Lấy dữ liệu cho nhiều cổ phiếu
tickers = ["AAPL", "GOOGL", "MSFT", "AMZN"]
data = yf.download(tickers, period="1y", interval="1d")

# Dữ liệu sẽ có cấu trúc MultiIndex
print(data.head())

Xây dựng Bot Auto Trading với YFinance

1. Bot cơ bản với Moving Average

import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime
import time

class YFinanceTradingBot:
    """Bot giao dịch sử dụng dữ liệu từ YFinance"""

    def __init__(self, symbol, initial_capital=10000):
        """
        Khởi tạo bot

        Args:
            symbol: Mã cổ phiếu (ví dụ: "AAPL", "TSLA")
            initial_capital: Vốn ban đầu
        """
        self.symbol = symbol
        self.ticker = yf.Ticker(symbol)
        self.capital = initial_capital
        self.shares = 0
        self.positions = []  # Lưu lịch sử giao dịch

    def get_current_price(self):
        """Lấy giá hiện tại"""
        try:
            data = self.ticker.history(period="1d", interval="1m")
            if not data.empty:
                return data['Close'].iloc[-1]
            else:
                # Fallback: lấy giá đóng cửa gần nhất
                data = self.ticker.history(period="1d", interval="1d")
                return data['Close'].iloc[-1]
        except Exception as e:
            print(f"Error getting price: {e}")
            return None

    def get_historical_data(self, period="1mo", interval="1d"):
        """Lấy dữ liệu lịch sử"""
        try:
            data = self.ticker.history(period=period, interval=interval)
            return data
        except Exception as e:
            print(f"Error getting historical data: {e}")
            return pd.DataFrame()

    def calculate_indicators(self, data):
        """Tính toán các chỉ báo kỹ thuật"""
        df = data.copy()

        # Simple Moving Average (SMA)
        df['SMA_20'] = df['Close'].rolling(window=20).mean()
        df['SMA_50'] = df['Close'].rolling(window=50).mean()

        # Exponential Moving Average (EMA)
        df['EMA_12'] = df['Close'].ewm(span=12, adjust=False).mean()
        df['EMA_26'] = df['Close'].ewm(span=26, adjust=False).mean()

        # MACD
        df['MACD'] = df['EMA_12'] - df['EMA_26']
        df['MACD_Signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
        df['MACD_Hist'] = df['MACD'] - df['MACD_Signal']

        # RSI
        delta = df['Close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
        rs = gain / loss
        df['RSI'] = 100 - (100 / (1 + rs))

        # Bollinger Bands
        df['BB_Middle'] = df['Close'].rolling(window=20).mean()
        df['BB_Std'] = df['Close'].rolling(window=20).std()
        df['BB_Upper'] = df['BB_Middle'] + (df['BB_Std'] * 2)
        df['BB_Lower'] = df['BB_Middle'] - (df['BB_Std'] * 2)

        return df

    def generate_signals(self, df):
        """
        Tạo tín hiệu giao dịch dựa trên Moving Average Crossover

        Returns:
            'buy': Tín hiệu mua
            'sell': Tín hiệu bán
            'hold': Giữ nguyên
        """
        if len(df) < 2:
            return 'hold'

        latest = df.iloc[-1]
        prev = df.iloc[-2]

        # Tín hiệu mua: SMA ngắn cắt lên SMA dài
        buy_signal = (
            latest['SMA_20'] > latest['SMA_50'] and
            prev['SMA_20'] <= prev['SMA_50'] and
            latest['RSI'] < 70  # Không quá overbought
        )

        # Tín hiệu bán: SMA ngắn cắt xuống SMA dài
        sell_signal = (
            latest['SMA_20'] < latest['SMA_50'] and
            prev['SMA_20'] >= prev['SMA_50'] and
            latest['RSI'] > 30  # Không quá oversold
        )

        if buy_signal:
            return 'buy'
        elif sell_signal:
            return 'sell'
        else:
            return 'hold'

    def execute_buy(self, price, amount=None):
        """Thực hiện lệnh mua"""
        if amount is None:
            # Mua với toàn bộ số tiền có
            amount = self.capital
        else:
            amount = min(amount, self.capital)

        shares_to_buy = amount / price
        cost = shares_to_buy * price

        if cost <= self.capital:
            self.shares += shares_to_buy
            self.capital -= cost

            trade = {
                'timestamp': datetime.now(),
                'action': 'BUY',
                'price': price,
                'shares': shares_to_buy,
                'cost': cost,
                'capital_remaining': self.capital
            }
            self.positions.append(trade)

            print(f"[BUY] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - "
                  f"Price: ${price:.2f}, Shares: {shares_to_buy:.4f}, "
                  f"Cost: ${cost:.2f}, Capital: ${self.capital:.2f}")
            return True
        else:
            print(f"Insufficient capital. Need ${cost:.2f}, have ${self.capital:.2f}")
            return False

    def execute_sell(self, price, shares=None):
        """Thực hiện lệnh bán"""
        if shares is None:
            shares = self.shares
        else:
            shares = min(shares, self.shares)

        if shares > 0:
            revenue = shares * price
            self.shares -= shares
            self.capital += revenue

            trade = {
                'timestamp': datetime.now(),
                'action': 'SELL',
                'price': price,
                'shares': shares,
                'revenue': revenue,
                'capital_remaining': self.capital
            }
            self.positions.append(trade)

            print(f"[SELL] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - "
                  f"Price: ${price:.2f}, Shares: {shares:.4f}, "
                  f"Revenue: ${revenue:.2f}, Capital: ${self.capital:.2f}")
            return True
        else:
            print("No shares to sell")
            return False

    def get_portfolio_value(self, current_price):
        """Tính giá trị danh mục hiện tại"""
        return self.capital + (self.shares * current_price)

    def run(self, check_interval=300):
        """
        Chạy bot giao dịch

        Args:
            check_interval: Khoảng thời gian kiểm tra (giây), mặc định 5 phút
        """
        print(f"Starting trading bot for {self.symbol}")
        print(f"Initial capital: ${self.capital:.2f}")

        while True:
            try:
                # Lấy dữ liệu mới nhất
                data = self.get_historical_data(period="3mo", interval="1d")

                if data.empty:
                    print("No data available, waiting...")
                    time.sleep(check_interval)
                    continue

                # Tính toán chỉ báo
                df = self.calculate_indicators(data)

                # Tạo tín hiệu
                signal = self.generate_signals(df)

                # Lấy giá hiện tại
                current_price = self.get_current_price()

                if current_price is None:
                    print("Could not get current price, waiting...")
                    time.sleep(check_interval)
                    continue

                # Thực hiện giao dịch
                if signal == 'buy' and self.capital > 0:
                    self.execute_buy(current_price)
                elif signal == 'sell' and self.shares > 0:
                    self.execute_sell(current_price)

                # Hiển thị trạng thái
                portfolio_value = self.get_portfolio_value(current_price)
                print(f"[STATUS] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - "
                      f"Price: ${current_price:.2f}, Signal: {signal.upper()}, "
                      f"Shares: {self.shares:.4f}, "
                      f"Portfolio Value: ${portfolio_value:.2f}")

                # Chờ trước khi kiểm tra lại
                time.sleep(check_interval)

            except KeyboardInterrupt:
                print("\nStopping bot...")
                break
            except Exception as e:
                print(f"Error in main loop: {e}")
                time.sleep(check_interval)

# Sử dụng bot
if __name__ == "__main__":
    bot = YFinanceTradingBot("AAPL", initial_capital=10000)
    # bot.run(check_interval=300)  # Kiểm tra mỗi 5 phút

2. Bot với chiến lược MACD

class MACDTradingBot(YFinanceTradingBot):
    """Bot giao dịch sử dụng chiến lược MACD"""

    def generate_signals(self, df):
        """Tạo tín hiệu dựa trên MACD"""
        if len(df) < 2:
            return 'hold'

        latest = df.iloc[-1]
        prev = df.iloc[-2]

        # Tín hiệu mua: MACD cắt lên Signal line
        buy_signal = (
            latest['MACD'] > latest['MACD_Signal'] and
            prev['MACD'] <= prev['MACD_Signal'] and
            latest['MACD_Hist'] > 0
        )

        # Tín hiệu bán: MACD cắt xuống Signal line
        sell_signal = (
            latest['MACD'] < latest['MACD_Signal'] and
            prev['MACD'] >= prev['MACD_Signal'] and
            latest['MACD_Hist'] < 0
        )

        if buy_signal:
            return 'buy'
        elif sell_signal:
            return 'sell'
        else:
            return 'hold'

3. Bot với chiến lược RSI + Bollinger Bands

class RSIBollingerBot(YFinanceTradingBot):
    """Bot giao dịch kết hợp RSI và Bollinger Bands"""

    def generate_signals(self, df):
        """Tạo tín hiệu dựa trên RSI và Bollinger Bands"""
        if len(df) < 2:
            return 'hold'

        latest = df.iloc[-1]

        # Tín hiệu mua: Giá chạm dưới BB Lower và RSI < 30
        buy_signal = (
            latest['Close'] < latest['BB_Lower'] and
            latest['RSI'] < 30
        )

        # Tín hiệu bán: Giá chạm trên BB Upper và RSI > 70
        sell_signal = (
            latest['Close'] > latest['BB_Upper'] and
            latest['RSI'] > 70
        )

        if buy_signal:
            return 'buy'
        elif sell_signal:
            return 'sell'
        else:
            return 'hold'

Backtesting với YFinance

Backtesting là quá trình kiểm tra chiến lược trên dữ liệu lịch sử để đánh giá hiệu quả.

class Backtester:
    """Lớp backtesting cho chiến lược giao dịch"""

    def __init__(self, symbol, initial_capital=10000):
        self.symbol = symbol
        self.initial_capital = initial_capital
        self.ticker = yf.Ticker(symbol)

    def backtest_strategy(self, strategy_func, start_date, end_date, interval="1d"):
        """
        Backtest một chiến lược

        Args:
            strategy_func: Hàm tạo tín hiệu giao dịch
            start_date: Ngày bắt đầu
            end_date: Ngày kết thúc
            interval: Khoảng thời gian (1d, 1h, etc.)
        """
        # Lấy dữ liệu lịch sử
        data = self.ticker.history(start=start_date, end=end_date, interval=interval)

        if data.empty:
            print("No data available for backtesting")
            return None

        # Tính toán chỉ báo
        bot = YFinanceTradingBot(self.symbol, self.initial_capital)
        df = bot.calculate_indicators(data)

        # Khởi tạo biến
        capital = self.initial_capital
        shares = 0
        trades = []
        equity_curve = []

        # Chạy backtest
        for i in range(1, len(df)):
            current_data = df.iloc[:i+1]
            signal = strategy_func(current_data)
            current_price = df.iloc[i]['Close']

            # Thực hiện giao dịch
            if signal == 'buy' and capital > 0:
                shares_to_buy = capital / current_price
                cost = shares_to_buy * current_price
                if cost <= capital:
                    shares += shares_to_buy
                    capital -= cost
                    trades.append({
                        'date': df.index[i],
                        'action': 'BUY',
                        'price': current_price,
                        'shares': shares_to_buy
                    })

            elif signal == 'sell' and shares > 0:
                revenue = shares * current_price
                capital += revenue
                trades.append({
                    'date': df.index[i],
                    'action': 'SELL',
                    'price': current_price,
                    'shares': shares
                })
                shares = 0

            # Tính giá trị danh mục
            portfolio_value = capital + (shares * current_price)
            equity_curve.append({
                'date': df.index[i],
                'value': portfolio_value
            })

        # Tính toán kết quả
        final_value = capital + (shares * df.iloc[-1]['Close'])
        total_return = ((final_value - self.initial_capital) / self.initial_capital) * 100

        results = {
            'initial_capital': self.initial_capital,
            'final_value': final_value,
            'total_return': total_return,
            'total_trades': len(trades),
            'trades': trades,
            'equity_curve': pd.DataFrame(equity_curve)
        }

        return results

    def print_results(self, results):
        """In kết quả backtesting"""
        print("\n" + "="*50)
        print("BACKTESTING RESULTS")
        print("="*50)
        print(f"Symbol: {self.symbol}")
        print(f"Initial Capital: ${results['initial_capital']:,.2f}")
        print(f"Final Value: ${results['final_value']:,.2f}")
        print(f"Total Return: {results['total_return']:.2f}%")
        print(f"Total Trades: {results['total_trades']}")
        print("="*50)

# Sử dụng backtester
def moving_average_strategy(df):
    """Chiến lược Moving Average"""
    if len(df) < 2:
        return 'hold'

    latest = df.iloc[-1]
    prev = df.iloc[-2]

    buy_signal = (
        latest['SMA_20'] > latest['SMA_50'] and
        prev['SMA_20'] <= prev['SMA_50']
    )

    sell_signal = (
        latest['SMA_20'] < latest['SMA_50'] and
        prev['SMA_20'] >= prev['SMA_50']
    )

    if buy_signal:
        return 'buy'
    elif sell_signal:
        return 'sell'
    else:
        return 'hold'

# Chạy backtest
backtester = Backtester("AAPL", initial_capital=10000)
results = backtester.backtest_strategy(
    strategy_func=moving_average_strategy,
    start_date="2023-01-01",
    end_date="2024-01-01",
    interval="1d"
)

if results:
    backtester.print_results(results)

Visualizing Results

import matplotlib.pyplot as plt

def plot_backtest_results(results, data):
    """Vẽ biểu đồ kết quả backtesting"""
    fig, axes = plt.subplots(2, 1, figsize=(14, 10))

    # Biểu đồ giá và tín hiệu
    ax1 = axes[0]
    ax1.plot(data.index, data['Close'], label='Price', linewidth=2)

    # Đánh dấu các điểm mua/bán
    buy_trades = [t for t in results['trades'] if t['action'] == 'BUY']
    sell_trades = [t for t in results['trades'] if t['action'] == 'SELL']

    if buy_trades:
        buy_dates = [t['date'] for t in buy_trades]
        buy_prices = [t['price'] for t in buy_trades]
        ax1.scatter(buy_dates, buy_prices, color='green', marker='^', 
                   s=100, label='Buy', zorder=5)

    if sell_trades:
        sell_dates = [t['date'] for t in sell_trades]
        sell_prices = [t['price'] for t in sell_trades]
        ax1.scatter(sell_dates, sell_prices, color='red', marker='v', 
                   s=100, label='Sell', zorder=5)

    ax1.set_title(f'Price Chart with Trading Signals')
    ax1.set_xlabel('Date')
    ax1.set_ylabel('Price ($)')
    ax1.legend()
    ax1.grid(True, alpha=0.3)

    # Biểu đồ equity curve
    ax2 = axes[1]
    equity_df = results['equity_curve']
    ax2.plot(equity_df['date'], equity_df['value'], label='Portfolio Value', 
            linewidth=2, color='blue')
    ax2.axhline(y=results['initial_capital'], color='red', 
               linestyle='--', label='Initial Capital')
    ax2.set_title('Equity Curve')
    ax2.set_xlabel('Date')
    ax2.set_ylabel('Portfolio Value ($)')
    ax2.legend()
    ax2.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()

# Vẽ kết quả
if results:
    data = backtester.ticker.history(start="2023-01-01", end="2024-01-01")
    plot_backtest_results(results, data)

Giao dịch nhiều cổ phiếu cùng lúc

class MultiStockBot:
    """Bot giao dịch nhiều cổ phiếu cùng lúc"""

    def __init__(self, symbols, initial_capital=10000):
        """
        Args:
            symbols: Danh sách mã cổ phiếu (ví dụ: ["AAPL", "GOOGL", "MSFT"])
            initial_capital: Vốn ban đầu
        """
        self.symbols = symbols
        self.initial_capital = initial_capital
        self.capital_per_stock = initial_capital / len(symbols)
        self.bots = {}

        # Tạo bot cho mỗi cổ phiếu
        for symbol in symbols:
            self.bots[symbol] = YFinanceTradingBot(
                symbol, 
                initial_capital=self.capital_per_stock
            )

    def run_all(self, check_interval=300):
        """Chạy tất cả các bot"""
        import threading

        threads = []
        for symbol, bot in self.bots.items():
            thread = threading.Thread(
                target=bot.run, 
                args=(check_interval,),
                daemon=True
            )
            thread.start()
            threads.append(thread)

        # Chờ tất cả threads
        for thread in threads:
            thread.join()

    def get_total_portfolio_value(self):
        """Tính tổng giá trị danh mục"""
        total = 0
        for symbol, bot in self.bots.items():
            current_price = bot.get_current_price()
            if current_price:
                total += bot.get_portfolio_value(current_price)
        return total

# Sử dụng
multi_bot = MultiStockBot(["AAPL", "GOOGL", "MSFT"], initial_capital=30000)
# multi_bot.run_all(check_interval=300)

Lưu ý quan trọng

1. Giới hạn của YFinance

Dữ liệu có độ trễ: YFinance không phải real-time, có độ trễ vài phút
Rate limiting: Yahoo Finance có thể giới hạn số lượng request
Không phù hợp cho day trading: Chỉ phù hợp cho swing trading hoặc long-term

2. Paper Trading trước

Luôn test bot trên paper trading (giao dịch giả) trước khi dùng tiền thật:

class PaperTradingBot(YFinanceTradingBot):
    """Bot paper trading - không dùng tiền thật"""

    def execute_buy(self, price, amount=None):
        """Ghi nhận lệnh mua nhưng không thực sự mua"""
        # Chỉ log, không thực sự mua
        print(f"[PAPER BUY] Would buy at ${price:.2f}")
        return super().execute_buy(price, amount)

    def execute_sell(self, price, shares=None):
        """Ghi nhận lệnh bán nhưng không thực sự bán"""
        print(f"[PAPER SELL] Would sell at ${price:.2f}")
        return super().execute_sell(price, shares)

3. Quản lý rủi ro

Diversification: Đa dạng hóa danh mục
Position sizing: Không đầu tư quá nhiều vào một cổ phiếu
Stop loss: Luôn đặt stop loss để giới hạn thua lỗ

Kết luận

YFinance là công cụ tuyệt vời để bắt đầu với bot trading vì:

Miễn phí: Không cần API key
Dễ sử dụng: API đơn giản, trực quan
Dữ liệu phong phú: Nhiều loại dữ liệu tài chính
Phù hợp cho học tập: Lý tưởng để học và thực hành

Tuy nhiên, cần nhớ rằng:

YFinance không phù hợp cho day trading real-time
Luôn backtest kỹ trước khi giao dịch thật
Bắt đầu với paper trading
Quản lý rủi ro cẩn thận

Bài tập thực hành

Tạo bot đơn giản: Xây dựng bot với chiến lược Moving Average
Backtesting: Test chiến lược trên dữ liệu 1 năm
So sánh chiến lược: So sánh hiệu quả của MACD, RSI, và Moving Average
Multi-stock bot: Xây dựng bot giao dịch nhiều cổ phiếu
Visualization: Vẽ biểu đồ kết quả backtesting

Tác giả: Hướng Nghiệp Lập Trình
Ngày đăng: 16/03/2025
Chuyên mục: Lập trình Bot Auto Trading, Python Nâng cao

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