Chiến lược RSI 30–70 trong Bot Python - Hướng dẫn chi tiết
Lập trình Bot Auto Trading
5 phút đọc

Hướng Nghiệp Lập Trình
Chuyên gia lập trình
Chiến lược RSI 30–70 trong Bot Python - Hướng dẫn chi tiết
RSI (Relative Strength Index) là một trong những chỉ báo kỹ thuật phổ biến nhất trong phân tích kỹ thuật. Chiến lược RSI 30-70 là một phương pháp giao dịch đơn giản nhưng hiệu quả, sử dụng các ngưỡng 30 (oversold) và 70 (overbought) để tạo tín hiệu mua và bán.
RSI là gì?
RSI (Relative Strength Index) là chỉ báo động lượng được phát triển bởi J. Welles Wilder vào năm 1978. RSI đo lường tốc độ và độ lớn của biến động giá, có giá trị từ 0 đến 100.
Công thức tính RSI
RSI = 100 - (100 / (1 + RS))
Trong đó:
RS = Average Gain / Average Loss
Average Gain = Trung bình của các phiên tăng trong 14 phiên gần nhất
Average Loss = Trung bình của các phiên giảm trong 14 phiên gần nhất
Ý nghĩa của RSI
- RSI > 70: Thị trường được coi là overbought (mua quá mức), có thể sắp giảm
- RSI < 30: Thị trường được coi là oversold (bán quá mức), có thể sắp tăng
- RSI = 50: Vùng trung tính, không có xu hướng rõ ràng
Chiến lược RSI 30-70
Nguyên lý hoạt động
Chiến lược RSI 30-70 dựa trên nguyên tắc:
- Tín hiệu MUA: Khi RSI vượt lên trên 30 (từ vùng oversold), báo hiệu giá có thể tăng
- Tín hiệu BÁN: Khi RSI giảm xuống dưới 70 (từ vùng overbought), báo hiệu giá có thể giảm
Ưu điểm
- Đơn giản: Dễ hiểu và implement
- Rõ ràng: Tín hiệu mua/bán rõ ràng
- Phù hợp nhiều thị trường: Hoạt động tốt với cổ phiếu, crypto, forex
- Giảm false signals: Tránh giao dịch trong vùng trung tính
Nhược điểm
- Chậm phản ứng: RSI có thể chậm trong thị trường trending mạnh
- False signals: Có thể có tín hiệu sai trong thị trường sideway
- Cần kết hợp: Nên kết hợp với các chỉ báo khác để tăng độ chính xác
Tính toán RSI trong Python
Cách 1: Tính toán thủ công
import pandas as pd import numpy as np def calculate_rsi(prices, period=14): """ Tính toán RSI Args: prices: Series giá đóng cửa period: Chu kỳ RSI (mặc định 14) Returns: Series RSI values """ delta = prices.diff() # Tách gain và loss gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() # Tính RS và RSI rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi # Ví dụ sử dụng import yfinance as yf # Lấy dữ liệu ticker = yf.Ticker("AAPL") data = ticker.history(period="6mo", interval="1d") # Tính RSI data['RSI'] = calculate_rsi(data['Close'], period=14) print(data[['Close', 'RSI']].tail(10))
Cách 2: Sử dụng thư viện TA-Lib
# Cài đặt: pip install TA-Lib import talib # Tính RSI với TA-Lib data['RSI'] = talib.RSI(data['Close'].values, timeperiod=14)
Cách 3: Sử dụng pandas_ta
# Cài đặt: pip install pandas_ta import pandas_ta as ta # Tính RSI với pandas_ta data['RSI'] = ta.rsi(data['Close'], length=14)
Xây dựng Bot RSI 30-70
Bot cơ bản
import yfinance as yf import pandas as pd import numpy as np from datetime import datetime import time class RSI30_70Bot: """Bot giao dịch sử dụng chiến lược RSI 30-70""" def __init__(self, symbol, initial_capital=10000, rsi_period=14): """ Khởi tạo bot Args: symbol: Mã cổ phiếu (ví dụ: "AAPL", "BTC-USD") initial_capital: Vốn ban đầu rsi_period: Chu kỳ RSI (mặc định 14) """ self.symbol = symbol self.ticker = yf.Ticker(symbol) self.capital = initial_capital self.shares = 0 self.rsi_period = rsi_period self.positions = [] self.last_rsi = None def calculate_rsi(self, prices): """Tính toán RSI""" delta = prices.diff() gain = (delta.where(delta > 0, 0)).rolling(window=self.rsi_period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=self.rsi_period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi def get_historical_data(self, period="3mo", 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 data: {e}") return pd.DataFrame() 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: 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 generate_signal(self, df): """ Tạo tín hiệu giao dịch dựa trên RSI 30-70 Logic: - MUA: RSI vượt lên trên 30 (từ dưới 30 lên trên 30) - BÁN: RSI giảm xuống dưới 70 (từ trên 70 xuống dưới 70) Returns: 'buy': Tín hiệu mua 'sell': Tín hiệu bán 'hold': Giữ nguyên """ if len(df) < self.rsi_period + 1: return 'hold' # Tính RSI df['RSI'] = self.calculate_rsi(df['Close']) # Lấy RSI hiện tại và trước đó current_rsi = df['RSI'].iloc[-1] prev_rsi = df['RSI'].iloc[-2] # Tín hiệu MUA: RSI vượt lên trên 30 buy_signal = ( current_rsi > 30 and prev_rsi <= 30 ) # Tín hiệu BÁN: RSI giảm xuống dưới 70 sell_signal = ( current_rsi < 70 and prev_rsi >= 70 ) # Lưu RSI hiện tại self.last_rsi = current_rsi 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: 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, 'rsi': self.last_rsi, 'capital_remaining': self.capital } self.positions.append(trade) print(f"[BUY] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - " f"Price: ${price:.2f}, RSI: {self.last_rsi:.2f}, " f"Shares: {shares_to_buy:.4f}, Cost: ${cost:.2f}") return True 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, 'rsi': self.last_rsi, 'capital_remaining': self.capital } self.positions.append(trade) print(f"[SELL] {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - " f"Price: ${price:.2f}, RSI: {self.last_rsi:.2f}, " f"Shares: {shares:.4f}, Revenue: ${revenue:.2f}") return True return False def get_portfolio_value(self, current_price): """Tính giá trị danh mục""" return self.capital + (self.shares * current_price) def run(self, check_interval=300): """ Chạy bot Args: check_interval: Khoảng thời gian kiểm tra (giây) """ print(f"Starting RSI 30-70 Bot for {self.symbol}") print(f"Initial capital: ${self.capital:.2f}") print(f"RSI Period: {self.rsi_period}") print("-" * 60) while True: try: # Lấy dữ liệu data = self.get_historical_data(period="3mo", interval="1d") if data.empty: print("No data available, waiting...") time.sleep(check_interval) continue # Tạo tín hiệu signal = self.generate_signal(data) # 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 # Tính RSI hiện tại để hiển thị data['RSI'] = self.calculate_rsi(data['Close']) current_rsi = data['RSI'].iloc[-1] # 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}, RSI: {current_rsi:.2f}, " f"Signal: {signal.upper()}, " f"Portfolio: ${portfolio_value:.2f}") time.sleep(check_interval) except KeyboardInterrupt: print("\nStopping bot...") break except Exception as e: print(f"Error: {e}") time.sleep(check_interval) # Sử dụng bot if __name__ == "__main__": bot = RSI30_70Bot("AAPL", initial_capital=10000, rsi_period=14) # bot.run(check_interval=300) # Kiểm tra mỗi 5 phút
Biến thể chiến lược RSI 30-70
1. RSI với ngưỡng tùy chỉnh
class CustomRSIBot(RSI30_70Bot): """Bot RSI với ngưỡng tùy chỉnh""" def __init__(self, symbol, initial_capital=10000, rsi_period=14, oversold=30, overbought=70): super().__init__(symbol, initial_capital, rsi_period) self.oversold = oversold self.overbought = overbought def generate_signal(self, df): """Tạo tín hiệu với ngưỡng tùy chỉnh""" if len(df) < self.rsi_period + 1: return 'hold' df['RSI'] = self.calculate_rsi(df['Close']) current_rsi = df['RSI'].iloc[-1] prev_rsi = df['RSI'].iloc[-2] # MUA: RSI vượt lên trên ngưỡng oversold buy_signal = ( current_rsi > self.oversold and prev_rsi <= self.oversold ) # BÁN: RSI giảm xuống dưới ngưỡng overbought sell_signal = ( current_rsi < self.overbought and prev_rsi >= self.overbought ) self.last_rsi = current_rsi if buy_signal: return 'buy' elif sell_signal: return 'sell' else: return 'hold' # Sử dụng với ngưỡng 25-75 bot = CustomRSIBot("AAPL", oversold=25, overbought=75)
2. RSI kết hợp với Moving Average
class RSIWithMABot(RSI30_70Bot): """Bot RSI kết hợp với Moving Average để lọc tín hiệu""" def generate_signal(self, df): """Tạo tín hiệu với filter MA""" if len(df) < self.rsi_period + 1: return 'hold' # Tính RSI df['RSI'] = self.calculate_rsi(df['Close']) # Tính Moving Average df['SMA_20'] = df['Close'].rolling(window=20).mean() df['SMA_50'] = df['Close'].rolling(window=50).mean() current_rsi = df['RSI'].iloc[-1] prev_rsi = df['RSI'].iloc[-2] current_price = df['Close'].iloc[-1] sma_20 = df['SMA_20'].iloc[-1] sma_50 = df['SMA_50'].iloc[-1] # Chỉ mua khi xu hướng tăng (giá > SMA 20 > SMA 50) uptrend = current_price > sma_20 > sma_50 # Chỉ bán khi xu hướng giảm (giá < SMA 20 < SMA 50) downtrend = current_price < sma_20 < sma_50 # Tín hiệu mua: RSI vượt 30 VÀ xu hướng tăng buy_signal = ( current_rsi > 30 and prev_rsi <= 30 and uptrend ) # Tín hiệu bán: RSI giảm xuống 70 VÀ xu hướng giảm sell_signal = ( current_rsi < 70 and prev_rsi >= 70 and downtrend ) self.last_rsi = current_rsi if buy_signal: return 'buy' elif sell_signal: return 'sell' else: return 'hold'
3. RSI với Volume Confirmation
class RSIWithVolumeBot(RSI30_70Bot): """Bot RSI với xác nhận volume""" def generate_signal(self, df): """Tạo tín hiệu với xác nhận volume""" if len(df) < self.rsi_period + 1: return 'hold' df['RSI'] = self.calculate_rsi(df['Close']) # Tính volume trung bình df['Volume_MA'] = df['Volume'].rolling(window=20).mean() current_rsi = df['RSI'].iloc[-1] prev_rsi = df['RSI'].iloc[-2] current_volume = df['Volume'].iloc[-1] avg_volume = df['Volume_MA'].iloc[-1] # Volume tăng mạnh (gấp 1.5 lần trung bình) high_volume = current_volume > avg_volume * 1.5 # Tín hiệu mua: RSI vượt 30 VÀ volume tăng buy_signal = ( current_rsi > 30 and prev_rsi <= 30 and high_volume ) # Tín hiệu bán: RSI giảm xuống 70 VÀ volume tăng sell_signal = ( current_rsi < 70 and prev_rsi >= 70 and high_volume ) self.last_rsi = current_rsi if buy_signal: return 'buy' elif sell_signal: return 'sell' else: return 'hold'
Backtesting chiến lược RSI 30-70
class RSIBacktester: """Backtesting cho chiến lược RSI 30-70""" def __init__(self, symbol, initial_capital=10000, rsi_period=14): self.symbol = symbol self.initial_capital = initial_capital self.rsi_period = rsi_period self.ticker = yf.Ticker(symbol) def calculate_rsi(self, prices): """Tính RSI""" delta = prices.diff() gain = (delta.where(delta > 0, 0)).rolling(window=self.rsi_period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=self.rsi_period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi def backtest(self, start_date, end_date, interval="1d"): """Chạy backtest""" # Lấy dữ liệu data = self.ticker.history(start=start_date, end=end_date, interval=interval) if data.empty: return None # Tính RSI data['RSI'] = self.calculate_rsi(data['Close']) # Khởi tạo capital = self.initial_capital shares = 0 trades = [] equity_curve = [] # Backtest for i in range(self.rsi_period + 1, len(data)): current_rsi = data['RSI'].iloc[i] prev_rsi = data['RSI'].iloc[i-1] current_price = data['Close'].iloc[i] # Tín hiệu mua if current_rsi > 30 and prev_rsi <= 30 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': data.index[i], 'action': 'BUY', 'price': current_price, 'rsi': current_rsi, 'shares': shares_to_buy }) # Tín hiệu bán elif current_rsi < 70 and prev_rsi >= 70 and shares > 0: revenue = shares * current_price capital += revenue trades.append({ 'date': data.index[i], 'action': 'SELL', 'price': current_price, 'rsi': current_rsi, 'shares': shares }) shares = 0 # Tính giá trị danh mục portfolio_value = capital + (shares * current_price) equity_curve.append({ 'date': data.index[i], 'value': portfolio_value, 'rsi': current_rsi }) # Tính kết quả final_value = capital + (shares * data['Close'].iloc[-1]) total_return = ((final_value - self.initial_capital) / self.initial_capital) * 100 # Tính số lệnh thắng/thua winning_trades = 0 losing_trades = 0 total_profit = 0 i = 0 while i < len(trades) - 1: if trades[i]['action'] == 'BUY' and trades[i+1]['action'] == 'SELL': profit = (trades[i+1]['price'] - trades[i]['price']) * trades[i]['shares'] total_profit += profit if profit > 0: winning_trades += 1 else: losing_trades += 1 i += 2 else: i += 1 results = { 'initial_capital': self.initial_capital, 'final_value': final_value, 'total_return': total_return, 'total_trades': len(trades), 'winning_trades': winning_trades, 'losing_trades': losing_trades, 'win_rate': (winning_trades / (winning_trades + losing_trades) * 100) if (winning_trades + losing_trades) > 0 else 0, 'total_profit': total_profit, 'trades': trades, 'equity_curve': pd.DataFrame(equity_curve) } return results def print_results(self, results): """In kết quả backtest""" print("\n" + "="*60) print("RSI 30-70 BACKTESTING RESULTS") print("="*60) 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(f"Winning Trades: {results['winning_trades']}") print(f"Losing Trades: {results['losing_trades']}") print(f"Win Rate: {results['win_rate']:.2f}%") print(f"Total Profit: ${results['total_profit']:,.2f}") print("="*60) # Chạy backtest backtester = RSIBacktester("AAPL", initial_capital=10000, rsi_period=14) results = backtester.backtest("2023-01-01", "2024-01-01", interval="1d") if results: backtester.print_results(results)
Visualization
import matplotlib.pyplot as plt def plot_rsi_strategy(data, results): """Vẽ biểu đồ chiến lược RSI""" fig, axes = plt.subplots(3, 1, figsize=(14, 12)) # Biểu đồ giá và tín hiệu ax1 = axes[0] ax1.plot(data.index, data['Close'], label='Price', linewidth=2, color='black') # Đánh dấu 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 Signal', 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 Signal', zorder=5) ax1.set_title('Price Chart with RSI 30-70 Signals') ax1.set_ylabel('Price ($)') ax1.legend() ax1.grid(True, alpha=0.3) # Biểu đồ RSI ax2 = axes[1] ax2.plot(data.index, data['RSI'], label='RSI', linewidth=2, color='blue') ax2.axhline(y=70, color='red', linestyle='--', label='Overbought (70)') ax2.axhline(y=30, color='green', linestyle='--', label='Oversold (30)') ax2.fill_between(data.index, 30, 70, alpha=0.1, color='gray') ax2.set_ylabel('RSI') ax2.set_ylim(0, 100) ax2.legend() ax2.grid(True, alpha=0.3) # Equity curve ax3 = axes[2] equity_df = results['equity_curve'] ax3.plot(equity_df['date'], equity_df['value'], label='Portfolio Value', linewidth=2, color='blue') ax3.axhline(y=results['initial_capital'], color='red', linestyle='--', label='Initial Capital') ax3.set_title('Equity Curve') ax3.set_xlabel('Date') ax3.set_ylabel('Portfolio Value ($)') ax3.legend() ax3.grid(True, alpha=0.3) plt.tight_layout() plt.show() # Vẽ biểu đồ if results: data = backtester.ticker.history(start="2023-01-01", end="2024-01-01") data['RSI'] = backtester.calculate_rsi(data['Close']) plot_rsi_strategy(data, results)
Best Practices
1. Tối ưu hóa tham số RSI
def optimize_rsi_period(symbol, start_date, end_date, periods=[10, 14, 20, 30]): """Tìm chu kỳ RSI tối ưu""" best_period = None best_return = -float('inf') for period in periods: backtester = RSIBacktester(symbol, rsi_period=period) results = backtester.backtest(start_date, end_date) if results and results['total_return'] > best_return: best_return = results['total_return'] best_period = period print(f"Period {period}: Return = {results['total_return']:.2f}%") print(f"\nBest Period: {best_period} with Return: {best_return:.2f}%") return best_period # Tối ưu hóa optimize_rsi_period("AAPL", "2023-01-01", "2024-01-01")
2. Kết hợp với Stop Loss
class RSIWithStopLossBot(RSI30_70Bot): """Bot RSI với Stop Loss""" def __init__(self, symbol, initial_capital=10000, rsi_period=14, stop_loss_pct=2.0): super().__init__(symbol, initial_capital, rsi_period) self.stop_loss_pct = stop_loss_pct self.entry_price = None def check_stop_loss(self, current_price): """Kiểm tra stop loss""" if self.shares > 0 and self.entry_price: loss_pct = ((current_price - self.entry_price) / self.entry_price) * 100 if loss_pct <= -self.stop_loss_pct: print(f"Stop Loss triggered at {loss_pct:.2f}%") self.execute_sell(current_price) self.entry_price = None return True return False def execute_buy(self, price, amount=None): """Ghi nhận giá entry khi mua""" if super().execute_buy(price, amount): self.entry_price = price return True return False
Kết luận
Chiến lược RSI 30-70 là một phương pháp giao dịch đơn giản và hiệu quả:
- Dễ implement: Code đơn giản, dễ hiểu
- Rõ ràng: Tín hiệu mua/bán rõ ràng
- Linh hoạt: Có thể tùy chỉnh ngưỡng và kết hợp với chỉ báo khác
- Phù hợp nhiều thị trường: Hoạt động tốt với cổ phiếu, crypto, forex
Lưu ý quan trọng:
- Luôn backtest trước khi giao dịch thật
- Kết hợp với quản lý rủi ro (stop loss, position sizing)
- Không nên chỉ dựa vào RSI, nên kết hợp với các chỉ báo khác
- Test trên paper trading trước
Bài tập thực hành
- Tạo bot RSI cơ bản: Implement bot RSI 30-70 đơn giản
- Backtesting: Test chiến lược trên nhiều cổ phiếu khác nhau
- Tối ưu hóa: Tìm chu kỳ RSI và ngưỡng tối ưu
- Kết hợp chỉ báo: Thêm Moving Average hoặc Volume filter
- So sánh: So sánh hiệu quả của RSI 30-70 với các chiến lược khác
Tác giả: Hướng Nghiệp Lập Trình
Ngày đăng: 17/03/2025
Chuyên mục: Lập trình Bot Auto Trading, Python Nâng cao

