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ì:

1. Miễn phí: Không cần API key
2. Dễ sử dụng: API đơn giản, trực quan
3. Dữ liệu phong phú: Nhiều loại dữ liệu tài chính
4. 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

1. Tạo bot đơn giản: Xây dựng bot với chiến lược Moving Average
2. Backtesting: Test chiến lược trên dữ liệu 1 năm
3. So sánh chiến lược: So sánh hiệu quả của MACD, RSI, và Moving Average
4. Multi-stock bot: Xây dựng bot giao dịch nhiều cổ phiếu
5. 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