Bài viết gần đây
-
-
So sánh Flutter và React Native
Tháng mười một 17, 2025 -
So sánh Flutter và React Native
Tháng mười một 17, 2025 -
Chiến lược RSI 30–70 trong Bot Auto Trading Python
Tháng mười một 17, 2025 -
Chiến Lược Giao Dịch News Filter sử dụng API Python
Tháng mười một 17, 2025
| Chiến lược RSI 30–70 trong Bot Auto Trading Python
Được viết bởi thanhdt vào ngày 17/11/2025 lúc 19:39 | 8 lượt xem
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