Danh mục: Tin tức
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 Mean Reversion Bot Auto Trading
Được viết bởi thanhdt vào ngày 17/11/2025 lúc 12:04 | 70 lượt xem
Chiến Lược Mean Reversion Bot Python
Mean Reversion (Hồi quy về giá trị trung bình) là một trong những chiến lược giao dịch phổ biến nhất, dựa trên nguyên tắc rằng giá sẽ có xu hướng quay trở lại mức trung bình sau khi lệch xa. Chiến lược này đặc biệt hiệu quả trong thị trường sideways (đi ngang) và có thể tạo ra lợi nhuận ổn định khi được thực hiện đúng cách. Trong bài viết này, chúng ta sẽ xây dựng một bot giao dịch tự động sử dụng chiến lược Mean Reversion với Python.
Tổng quan về Mean Reversion
Mean Reversion là gì?
Mean Reversion là hiện tượng giá của một tài sản có xu hướng quay trở lại mức trung bình (mean) sau khi di chuyển quá xa khỏi nó. Nguyên tắc cơ bản:
- Giá tạm thời lệch xa khỏi mức trung bình
- Áp lực mua/bán sẽ đưa giá trở lại mức trung bình
- Cơ hội giao dịch xuất hiện khi giá ở cực trị (quá mua hoặc quá bán)
Tại sao Mean Reversion hiệu quả?
- Thị trường có tính chu kỳ: Giá thường dao động xung quanh giá trị trung bình
- Phù hợp thị trường sideways: Hoạt động tốt khi không có xu hướng rõ ràng
- Tần suất giao dịch cao: Nhiều cơ hội vào/ra lệnh hơn so với trend following
- Risk/Reward tốt: Stop Loss và Take Profit rõ ràng
- Có thể tự động hóa: Dễ dàng lập trình thành bot
Các chỉ báo Mean Reversion phổ biến
- Bollinger Bands: Dải trên/dưới dựa trên độ lệch chuẩn
- Z-Score: Đo lường khoảng cách giá so với trung bình (theo đơn vị độ lệch chuẩn)
- RSI (Relative Strength Index): Xác định quá mua/quá bán
- Stochastic Oscillator: Chỉ báo động lượng
- Moving Average: SMA, EMA để xác định mức trung bình
Cài đặt Môi trường
Thư viện cần thiết
# requirements.txt
pandas==2.1.0
numpy==1.24.3
ccxt==4.0.0
python-binance==1.0.19
matplotlib==3.7.2
plotly==5.17.0
ta-lib==0.4.28
schedule==1.2.0
python-dotenv==1.0.0
scipy==1.11.0
Cài đặt
pip install pandas numpy ccxt python-binance matplotlib plotly schedule python-dotenv scipy
Lưu ý:
- TA-Lib yêu cầu cài đặt thư viện C trước. Trên Windows, tải file
.whltừ đây. - Đối với Linux/Mac:
sudo apt-get install ta-libhoặcbrew install ta-lib
Xây dựng các Chỉ báo Mean Reversion
Tính toán Bollinger Bands
import pandas as pd
import numpy as np
from typing import Tuple
class BollingerBands:
"""
Lớp tính toán Bollinger Bands
"""
def __init__(self, period: int = 20, std_dev: float = 2.0):
"""
Khởi tạo Bollinger Bands
Args:
period: Chu kỳ tính trung bình (mặc định: 20)
std_dev: Số độ lệch chuẩn (mặc định: 2.0)
"""
self.period = period
self.std_dev = std_dev
def calculate(self, data: pd.Series) -> pd.DataFrame:
"""
Tính toán Bollinger Bands
Args:
data: Series chứa giá (thường là close price)
Returns:
DataFrame với các cột: middle, upper, lower
"""
# Middle band (SMA)
middle = data.rolling(window=self.period).mean()
# Standard deviation
std = data.rolling(window=self.period).std()
# Upper and lower bands
upper = middle + (std * self.std_dev)
lower = middle - (std * self.std_dev)
return pd.DataFrame({
'middle': middle,
'upper': upper,
'lower': lower
})
def get_position(self, price: float, upper: float, lower: float) -> float:
"""
Tính vị trí giá trong Bollinger Bands (0-1)
Args:
price: Giá hiện tại
upper: Dải trên
lower: Dải dưới
Returns:
Giá trị từ 0-1 (0 = dưới dải, 1 = trên dải)
"""
if upper == lower:
return 0.5
return (price - lower) / (upper - lower)
Tính toán Z-Score
class ZScoreIndicator:
"""
Lớp tính toán Z-Score
"""
def __init__(self, period: int = 20):
"""
Khởi tạo Z-Score Indicator
Args:
period: Chu kỳ tính trung bình và độ lệch chuẩn
"""
self.period = period
def calculate(self, data: pd.Series) -> pd.Series:
"""
Tính toán Z-Score
Z-Score = (Price - Mean) / StdDev
Args:
data: Series chứa giá
Returns:
Series chứa giá trị Z-Score
"""
mean = data.rolling(window=self.period).mean()
std = data.rolling(window=self.period).std()
z_score = (data - mean) / std
return z_score
def is_oversold(self, z_score: float, threshold: float = -2.0) -> bool:
"""
Kiểm tra giá có quá bán không
Args:
z_score: Giá trị Z-Score
threshold: Ngưỡng quá bán (mặc định: -2.0)
Returns:
True nếu quá bán
"""
return z_score <= threshold
def is_overbought(self, z_score: float, threshold: float = 2.0) -> bool:
"""
Kiểm tra giá có quá mua không
Args:
z_score: Giá trị Z-Score
threshold: Ngưỡng quá mua (mặc định: 2.0)
Returns:
True nếu quá mua
"""
return z_score >= threshold
Tính toán RSI
class RSIIndicator:
"""
Lớp tính toán RSI (Relative Strength Index)
"""
def __init__(self, period: int = 14):
"""
Khởi tạo RSI Indicator
Args:
period: Chu kỳ RSI (mặc định: 14)
"""
self.period = period
def calculate(self, data: pd.Series) -> pd.Series:
"""
Tính toán RSI
Args:
data: Series chứa giá
Returns:
Series chứa giá trị RSI (0-100)
"""
delta = data.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=self.period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=self.period).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
def is_oversold(self, rsi: float, threshold: float = 30.0) -> bool:
"""
Kiểm tra RSI có quá bán không
Args:
rsi: Giá trị RSI
threshold: Ngưỡng quá bán (mặc định: 30)
Returns:
True nếu quá bán
"""
return rsi <= threshold
def is_overbought(self, rsi: float, threshold: float = 70.0) -> bool:
"""
Kiểm tra RSI có quá mua không
Args:
rsi: Giá trị RSI
threshold: Ngưỡng quá mua (mặc định: 70)
Returns:
True nếu quá mua
"""
return rsi >= threshold
Chiến lược Mean Reversion
Nguyên lý Chiến lược
- Xác định mức trung bình: Sử dụng SMA hoặc EMA
- Phát hiện lệch xa: Khi giá lệch quá xa khỏi trung bình (Z-Score > 2 hoặc < -2)
- Tín hiệu vào lệnh:
- BUY: Giá dưới dải Bollinger dưới hoặc Z-Score < -2
- SELL: Giá trên dải Bollinger trên hoặc Z-Score > 2
- Xác nhận: Kết hợp với RSI để xác nhận quá mua/quá bán
- Quản lý rủi ro: Stop Loss và Take Profit dựa trên độ lệch chuẩn
Lớp Chiến lược Mean Reversion
class MeanReversionStrategy:
"""
Chiến lược Mean Reversion
"""
def __init__(
self,
bb_period: int = 20,
bb_std: float = 2.0,
z_score_period: int = 20,
z_score_threshold: float = 2.0,
rsi_period: int = 14,
rsi_oversold: float = 30.0,
rsi_overbought: float = 70.0,
require_rsi_confirmation: bool = True
):
"""
Khởi tạo chiến lược
Args:
bb_period: Chu kỳ Bollinger Bands
bb_std: Độ lệch chuẩn cho Bollinger Bands
z_score_period: Chu kỳ Z-Score
z_score_threshold: Ngưỡng Z-Score
rsi_period: Chu kỳ RSI
rsi_oversold: Ngưỡng RSI quá bán
rsi_overbought: Ngưỡng RSI quá mua
require_rsi_confirmation: Yêu cầu xác nhận RSI
"""
self.bb_period = bb_period
self.bb_std = bb_std
self.z_score_period = z_score_period
self.z_score_threshold = z_score_threshold
self.rsi_period = rsi_period
self.rsi_oversold = rsi_oversold
self.rsi_overbought = rsi_overbought
self.require_rsi_confirmation = require_rsi_confirmation
self.bb = BollingerBands(period=bb_period, std_dev=bb_std)
self.z_score = ZScoreIndicator(period=z_score_period)
self.rsi = RSIIndicator(period=rsi_period)
def calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Tính toán tất cả các chỉ báo
Args:
df: DataFrame OHLCV
Returns:
DataFrame với các chỉ báo đã tính
"""
result = df.copy()
# Bollinger Bands
bb_data = self.bb.calculate(result['close'])
result['bb_upper'] = bb_data['upper']
result['bb_middle'] = bb_data['middle']
result['bb_lower'] = bb_data['lower']
result['bb_position'] = result.apply(
lambda row: self.bb.get_position(
row['close'], row['bb_upper'], row['bb_lower']
), axis=1
)
# Z-Score
result['z_score'] = self.z_score.calculate(result['close'])
# RSI
result['rsi'] = self.rsi.calculate(result['close'])
return result
def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Tạo tín hiệu giao dịch
Returns:
DataFrame với cột 'signal' (-1: SELL, 0: HOLD, 1: BUY)
"""
# Tính các chỉ báo
df = self.calculate_indicators(df)
# Khởi tạo signal
df['signal'] = 0
df['signal_strength'] = 0.0
df['stop_loss'] = 0.0
df['take_profit'] = 0.0
for i in range(self.bb_period, len(df)):
current_price = df['close'].iloc[i]
bb_upper = df['bb_upper'].iloc[i]
bb_lower = df['bb_lower'].iloc[i]
bb_middle = df['bb_middle'].iloc[i]
z_score_val = df['z_score'].iloc[i]
rsi_val = df['rsi'].iloc[i]
# Tín hiệu BUY (quá bán)
buy_condition = False
buy_strength = 0.0
# Điều kiện 1: Giá dưới dải Bollinger dưới
if current_price < bb_lower:
buy_condition = True
buy_strength += 0.4
# Điều kiện 2: Z-Score < -threshold
if z_score_val < -self.z_score_threshold:
buy_condition = True
buy_strength += 0.3
# Điều kiện 3: RSI quá bán (nếu yêu cầu)
rsi_confirm = True
if self.require_rsi_confirmation:
rsi_confirm = self.rsi.is_oversold(rsi_val, self.rsi_oversold)
if rsi_confirm:
buy_strength += 0.3
if buy_condition and rsi_confirm:
df.iloc[i, df.columns.get_loc('signal')] = 1
df.iloc[i, df.columns.get_loc('signal_strength')] = min(buy_strength, 1.0)
# Tính Stop Loss và Take Profit
stop_loss = bb_lower * 0.995 # 0.5% dưới dải dưới
take_profit = bb_middle # Mục tiêu là dải giữa
df.iloc[i, df.columns.get_loc('stop_loss')] = stop_loss
df.iloc[i, df.columns.get_loc('take_profit')] = take_profit
# Tín hiệu SELL (quá mua)
sell_condition = False
sell_strength = 0.0
# Điều kiện 1: Giá trên dải Bollinger trên
if current_price > bb_upper:
sell_condition = True
sell_strength += 0.4
# Điều kiện 2: Z-Score > threshold
if z_score_val > self.z_score_threshold:
sell_condition = True
sell_strength += 0.3
# Điều kiện 3: RSI quá mua (nếu yêu cầu)
rsi_confirm = True
if self.require_rsi_confirmation:
rsi_confirm = self.rsi.is_overbought(rsi_val, self.rsi_overbought)
if rsi_confirm:
sell_strength += 0.3
if sell_condition and rsi_confirm:
df.iloc[i, df.columns.get_loc('signal')] = -1
df.iloc[i, df.columns.get_loc('signal_strength')] = min(sell_strength, 1.0)
# Tính Stop Loss và Take Profit
stop_loss = bb_upper * 1.005 # 0.5% trên dải trên
take_profit = bb_middle # Mục tiêu là dải giữa
df.iloc[i, df.columns.get_loc('stop_loss')] = stop_loss
df.iloc[i, df.columns.get_loc('take_profit')] = take_profit
return df
Xây dựng Trading Bot
Lớp Bot Chính
import ccxt
import time
import logging
from typing import Dict, Optional
from datetime import datetime
import os
from dotenv import load_dotenv
load_dotenv()
class MeanReversionBot:
"""
Bot giao dịch Mean Reversion
"""
def __init__(
self,
exchange_id: str = 'binance',
api_key: Optional[str] = None,
api_secret: Optional[str] = None,
symbol: str = 'BTC/USDT',
timeframe: str = '1h',
testnet: bool = True
):
"""
Khởi tạo bot
Args:
exchange_id: Tên sàn giao dịch
api_key: API Key
api_secret: API Secret
symbol: Cặp giao dịch
timeframe: Khung thời gian
testnet: Sử dụng testnet hay không
"""
self.exchange_id = exchange_id
self.symbol = symbol
self.timeframe = timeframe
self.testnet = testnet
self.api_key = api_key or os.getenv('EXCHANGE_API_KEY')
self.api_secret = api_secret or os.getenv('EXCHANGE_API_SECRET')
self.exchange = self._initialize_exchange()
self.strategy = MeanReversionStrategy(
bb_period=20,
bb_std=2.0,
z_score_period=20,
z_score_threshold=2.0,
rsi_period=14,
require_rsi_confirmation=True
)
self.position = None
self.orders = []
self.min_order_size = 0.001
self.risk_per_trade = 0.02
self._setup_logging()
def _initialize_exchange(self) -> ccxt.Exchange:
"""Khởi tạo kết nối với sàn"""
exchange_class = getattr(ccxt, self.exchange_id)
config = {
'apiKey': self.api_key,
'secret': self.api_secret,
'enableRateLimit': True,
'options': {'defaultType': 'spot'}
}
if self.testnet and self.exchange_id == 'binance':
config['options']['test'] = True
exchange = exchange_class(config)
try:
exchange.load_markets()
self.logger.info(f"Đã kết nối với {self.exchange_id}")
except Exception as e:
self.logger.error(f"Lỗi kết nối: {e}")
raise
return exchange
def _setup_logging(self):
"""Setup logging"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('mean_reversion_bot.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger('MeanReversionBot')
def fetch_ohlcv(self, limit: int = 100) -> pd.DataFrame:
"""Lấy dữ liệu OHLCV"""
try:
ohlcv = self.exchange.fetch_ohlcv(
self.symbol,
self.timeframe,
limit=limit
)
df = pd.DataFrame(
ohlcv,
columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df
except Exception as e:
self.logger.error(f"Lỗi lấy dữ liệu: {e}")
return pd.DataFrame()
def calculate_position_size(self, entry_price: float, stop_loss_price: float) -> float:
"""Tính toán khối lượng lệnh"""
try:
balance = self.get_balance()
available_balance = balance.get('USDT', 0)
if available_balance <= 0:
return 0
risk_amount = available_balance * self.risk_per_trade
stop_loss_distance = abs(entry_price - stop_loss_price)
if stop_loss_distance == 0:
return 0
position_size = risk_amount / stop_loss_distance
market = self.exchange.market(self.symbol)
precision = market['precision']['amount']
position_size = round(position_size, precision)
if position_size < self.min_order_size:
return 0
return position_size
except Exception as e:
self.logger.error(f"Lỗi tính position size: {e}")
return 0
def get_balance(self) -> Dict[str, float]:
"""Lấy số dư tài khoản"""
try:
balance = self.exchange.fetch_balance()
return {
'USDT': balance.get('USDT', {}).get('free', 0),
'BTC': balance.get('BTC', {}).get('free', 0),
'total': balance.get('total', {})
}
except Exception as e:
self.logger.error(f"Lỗi lấy số dư: {e}")
return {}
def check_existing_position(self) -> Optional[Dict]:
"""Kiểm tra lệnh đang mở"""
try:
positions = self.exchange.fetch_positions([self.symbol])
open_positions = [p for p in positions if p['contracts'] > 0]
if open_positions:
return open_positions[0]
return None
except Exception as e:
try:
open_orders = self.exchange.fetch_open_orders(self.symbol)
if open_orders:
return {'type': 'order', 'orders': open_orders}
except:
pass
return None
def execute_buy(self, df: pd.DataFrame) -> bool:
"""Thực hiện lệnh mua"""
try:
current_price = df['close'].iloc[-1]
stop_loss = df['stop_loss'].iloc[-1]
take_profit = df['take_profit'].iloc[-1]
signal_strength = df['signal_strength'].iloc[-1]
z_score = df['z_score'].iloc[-1]
position_size = self.calculate_position_size(current_price, stop_loss)
if position_size <= 0:
self.logger.warning("Position size quá nhỏ")
return False
order = self.exchange.create_market_buy_order(
self.symbol,
position_size
)
self.logger.info(
f"BUY MEAN REVERSION: {position_size} {self.symbol} @ {current_price:.2f} | "
f"Z-Score: {z_score:.2f} | Signal Strength: {signal_strength:.2f} | "
f"SL: {stop_loss:.2f} | TP: {take_profit:.2f}"
)
self.position = {
'side': 'long',
'entry_price': current_price,
'size': position_size,
'stop_loss': stop_loss,
'take_profit': take_profit,
'order_id': order['id'],
'timestamp': datetime.now()
}
return True
except Exception as e:
self.logger.error(f"Lỗi mua: {e}")
return False
def execute_sell(self, df: pd.DataFrame) -> bool:
"""Thực hiện lệnh bán"""
try:
current_price = df['close'].iloc[-1]
stop_loss = df['stop_loss'].iloc[-1]
take_profit = df['take_profit'].iloc[-1]
signal_strength = df['signal_strength'].iloc[-1]
z_score = df['z_score'].iloc[-1]
position_size = self.calculate_position_size(current_price, stop_loss)
if position_size <= 0:
self.logger.warning("Position size quá nhỏ")
return False
order = self.exchange.create_market_sell_order(
self.symbol,
position_size
)
self.logger.info(
f"SELL MEAN REVERSION: {position_size} {self.symbol} @ {current_price:.2f} | "
f"Z-Score: {z_score:.2f} | Signal Strength: {signal_strength:.2f} | "
f"SL: {stop_loss:.2f} | TP: {take_profit:.2f}"
)
self.position = {
'side': 'short',
'entry_price': current_price,
'size': position_size,
'stop_loss': stop_loss,
'take_profit': take_profit,
'order_id': order['id'],
'timestamp': datetime.now()
}
return True
except Exception as e:
self.logger.error(f"Lỗi bán: {e}")
return False
def check_exit_conditions(self, df: pd.DataFrame) -> bool:
"""Kiểm tra điều kiện thoát"""
if not self.position:
return False
current_price = df['close'].iloc[-1]
bb_middle = df['bb_middle'].iloc[-1]
z_score = df['z_score'].iloc[-1]
if self.position['side'] == 'long':
# Stop Loss
if current_price <= self.position['stop_loss']:
self.logger.info(f"Stop Loss @ {current_price:.2f}")
return True
# Take Profit
if current_price >= self.position['take_profit']:
self.logger.info(f"Take Profit @ {current_price:.2f}")
return True
# Thoát nếu giá quay về dải giữa (mean reversion hoàn thành)
if current_price >= bb_middle * 0.99:
self.logger.info("Giá đã quay về dải giữa, thoát lệnh")
return True
# Thoát nếu Z-Score về gần 0
if z_score >= -0.5:
self.logger.info("Z-Score đã về gần 0, thoát lệnh")
return True
elif self.position['side'] == 'short':
# Stop Loss
if current_price >= self.position['stop_loss']:
self.logger.info(f"Stop Loss @ {current_price:.2f}")
return True
# Take Profit
if current_price <= self.position['take_profit']:
self.logger.info(f"Take Profit @ {current_price:.2f}")
return True
# Thoát nếu giá quay về dải giữa
if current_price <= bb_middle * 1.01:
self.logger.info("Giá đã quay về dải giữa, thoát lệnh")
return True
# Thoát nếu Z-Score về gần 0
if z_score <= 0.5:
self.logger.info("Z-Score đã về gần 0, thoát lệnh")
return True
return False
def close_position(self) -> bool:
"""Đóng lệnh hiện tại"""
if not self.position:
return False
try:
if self.position['side'] == 'long':
order = self.exchange.create_market_sell_order(
self.symbol,
self.position['size']
)
else:
order = self.exchange.create_market_buy_order(
self.symbol,
self.position['size']
)
current_price = self.exchange.fetch_ticker(self.symbol)['last']
if self.position['side'] == 'long':
pnl_pct = ((current_price - self.position['entry_price']) / self.position['entry_price']) * 100
else:
pnl_pct = ((self.position['entry_price'] - current_price) / self.position['entry_price']) * 100
self.logger.info(
f"Đóng lệnh {self.position['side']} | "
f"Entry: {self.position['entry_price']:.2f} | "
f"Exit: {current_price:.2f} | P&L: {pnl_pct:.2f}%"
)
self.position = None
return True
except Exception as e:
self.logger.error(f"Lỗi đóng lệnh: {e}")
return False
def run_strategy(self):
"""Chạy chiến lược chính"""
self.logger.info("Bắt đầu chạy chiến lược Mean Reversion...")
while True:
try:
df = self.fetch_ohlcv(limit=100)
if df.empty:
self.logger.warning("Không lấy được dữ liệu")
time.sleep(60)
continue
df = self.strategy.generate_signals(df)
existing_position = self.check_existing_position()
if existing_position:
if self.check_exit_conditions(df):
self.close_position()
else:
latest_signal = df['signal'].iloc[-1]
if latest_signal == 1:
self.execute_buy(df)
elif latest_signal == -1:
self.execute_sell(df)
time.sleep(60)
except KeyboardInterrupt:
self.logger.info("Bot đã dừng")
break
except Exception as e:
self.logger.error(f"Lỗi: {e}")
time.sleep(60)
Backtesting Chiến lược
Lớp Backtesting
class MeanReversionBacktester:
"""
Backtest chiến lược Mean Reversion
"""
def __init__(
self,
initial_capital: float = 10000,
commission: float = 0.001
):
self.initial_capital = initial_capital
self.commission = commission
self.capital = initial_capital
self.position = None
self.trades = []
self.equity_curve = []
def backtest(self, df: pd.DataFrame) -> Dict:
"""Backtest chiến lược"""
strategy = MeanReversionStrategy()
df = strategy.generate_signals(df)
for i in range(20, len(df)): # Bắt đầu từ period của BB
current_row = df.iloc[i]
# Kiểm tra thoát lệnh
if self.position:
should_exit = False
exit_price = current_row['close']
bb_middle = current_row['bb_middle']
z_score = current_row['z_score']
if self.position['side'] == 'long':
if current_row['low'] <= self.position['stop_loss']:
exit_price = self.position['stop_loss']
should_exit = True
elif current_row['high'] >= self.position['take_profit']:
exit_price = self.position['take_profit']
should_exit = True
elif current_row['close'] >= bb_middle * 0.99:
should_exit = True
elif z_score >= -0.5:
should_exit = True
elif self.position['side'] == 'short':
if current_row['high'] >= self.position['stop_loss']:
exit_price = self.position['stop_loss']
should_exit = True
elif current_row['low'] <= self.position['take_profit']:
exit_price = self.position['take_profit']
should_exit = True
elif current_row['close'] <= bb_middle * 1.01:
should_exit = True
elif z_score <= 0.5:
should_exit = True
if should_exit:
self._close_trade(exit_price, current_row.name)
# Kiểm tra tín hiệu mới
if not self.position and current_row['signal'] != 0:
if current_row['signal'] == 1:
self._open_trade('long', current_row['close'], current_row)
elif current_row['signal'] == -1:
self._open_trade('short', current_row['close'], current_row)
equity = self._calculate_equity(current_row['close'])
self.equity_curve.append({
'timestamp': current_row.name,
'equity': equity
})
if self.position:
final_price = df.iloc[-1]['close']
self._close_trade(final_price, df.index[-1])
return self._calculate_metrics()
def _open_trade(self, side: str, price: float, row: pd.Series):
"""Mở lệnh mới"""
risk_amount = self.capital * 0.02
stop_loss = row.get('stop_loss', price * 0.98 if side == 'long' else price * 1.02)
take_profit = row.get('take_profit', price * 1.02 if side == 'long' else price * 0.98)
position_size = risk_amount / abs(price - stop_loss)
self.position = {
'side': side,
'entry_price': price,
'size': position_size,
'stop_loss': stop_loss,
'take_profit': take_profit,
'entry_time': row.name
}
def _close_trade(self, exit_price: float, exit_time):
"""Đóng lệnh"""
if not self.position:
return
if self.position['side'] == 'long':
pnl = (exit_price - self.position['entry_price']) * self.position['size']
else:
pnl = (self.position['entry_price'] - exit_price) * self.position['size']
commission_cost = (self.position['entry_price'] + exit_price) * self.position['size'] * self.commission
pnl -= commission_cost
self.capital += pnl
self.trades.append({
'side': self.position['side'],
'entry_price': self.position['entry_price'],
'exit_price': exit_price,
'size': self.position['size'],
'pnl': pnl,
'pnl_pct': (pnl / (self.position['entry_price'] * self.position['size'])) * 100,
'entry_time': self.position['entry_time'],
'exit_time': exit_time
})
self.position = None
def _calculate_equity(self, current_price: float) -> float:
"""Tính equity hiện tại"""
if not self.position:
return self.capital
if self.position['side'] == 'long':
unrealized_pnl = (current_price - self.position['entry_price']) * self.position['size']
else:
unrealized_pnl = (self.position['entry_price'] - current_price) * self.position['size']
return self.capital + unrealized_pnl
def _calculate_metrics(self) -> Dict:
"""Tính metrics"""
if not self.trades:
return {'error': 'Không có trades'}
trades_df = pd.DataFrame(self.trades)
total_trades = len(self.trades)
winning_trades = trades_df[trades_df['pnl'] > 0]
losing_trades = trades_df[trades_df['pnl'] < 0]
win_rate = len(winning_trades) / total_trades * 100 if total_trades > 0 else 0
avg_win = winning_trades['pnl'].mean() if len(winning_trades) > 0 else 0
avg_loss = abs(losing_trades['pnl'].mean()) if len(losing_trades) > 0 else 0
profit_factor = (winning_trades['pnl'].sum() / abs(losing_trades['pnl'].sum())) if len(losing_trades) > 0 and losing_trades['pnl'].sum() != 0 else 0
total_return = ((self.capital - self.initial_capital) / self.initial_capital) * 100
equity_curve_df = pd.DataFrame(self.equity_curve)
equity_curve_df['peak'] = equity_curve_df['equity'].expanding().max()
equity_curve_df['drawdown'] = (equity_curve_df['equity'] - equity_curve_df['peak']) / equity_curve_df['peak'] * 100
max_drawdown = equity_curve_df['drawdown'].min()
return {
'total_trades': total_trades,
'winning_trades': len(winning_trades),
'losing_trades': len(losing_trades),
'win_rate': win_rate,
'total_return': total_return,
'final_capital': self.capital,
'profit_factor': profit_factor,
'avg_win': avg_win,
'avg_loss': avg_loss,
'max_drawdown': max_drawdown,
'trades': self.trades,
'equity_curve': self.equity_curve
}
Sử dụng Bot
Script Chạy Bot
# run_mean_reversion_bot.py
from mean_reversion_bot import MeanReversionBot
import os
from dotenv import load_dotenv
load_dotenv()
if __name__ == '__main__':
bot = MeanReversionBot(
exchange_id='binance',
symbol='BTC/USDT',
timeframe='1h',
testnet=True
)
try:
bot.run_strategy()
except KeyboardInterrupt:
print("\nBot đã dừng")
Script Backtest
# backtest_mean_reversion.py
from mean_reversion_bot import MeanReversionBacktester
import ccxt
import pandas as pd
if __name__ == '__main__':
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv('BTC/USDT', '1h', limit=1000)
df = pd.DataFrame(
ohlcv,
columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
backtester = MeanReversionBacktester(initial_capital=10000)
results = backtester.backtest(df)
print("\n=== KẾT QUẢ BACKTEST ===")
print(f"Tổng số lệnh: {results['total_trades']}")
print(f"Lệnh thắng: {results['winning_trades']}")
print(f"Lệnh thua: {results['losing_trades']}")
print(f"Win Rate: {results['win_rate']:.2f}%")
print(f"Tổng lợi nhuận: {results['total_return']:.2f}%")
print(f"Profit Factor: {results['profit_factor']:.2f}")
print(f"Max Drawdown: {results['max_drawdown']:.2f}%")
print(f"Vốn cuối: ${results['final_capital']:.2f}")
Tối ưu hóa Chiến lược
1. Filter theo Xu hướng
def filter_by_trend(df: pd.DataFrame, trend_period: int = 50) -> pd.DataFrame:
"""Lọc tín hiệu theo xu hướng - Mean Reversion hoạt động tốt trong sideways"""
df['sma_trend'] = df['close'].rolling(window=trend_period).mean()
df['price_vs_sma'] = (df['close'] - df['sma_trend']) / df['sma_trend'] * 100
# Chỉ giao dịch khi thị trường không có xu hướng mạnh (sideways)
# Nếu giá lệch quá xa SMA (> 5%), có thể đang có xu hướng mạnh
df.loc[abs(df['price_vs_sma']) > 5, 'signal'] = 0
return df
2. Kết hợp với Volume
def filter_by_volume(df: pd.DataFrame, min_volume_ratio: float = 1.2) -> pd.DataFrame:
"""Lọc tín hiệu theo volume"""
df['avg_volume'] = df['volume'].rolling(window=20).mean()
df['volume_ratio'] = df['volume'] / df['avg_volume']
# Chỉ giao dịch khi volume cao (xác nhận sự lệch xa)
df.loc[df['volume_ratio'] < min_volume_ratio, 'signal'] = 0
return df
3. Adaptive Thresholds
def adaptive_thresholds(df: pd.DataFrame) -> pd.DataFrame:
"""Điều chỉnh ngưỡng Z-Score theo độ biến động"""
df['volatility'] = df['close'].rolling(window=20).std()
df['avg_volatility'] = df['volatility'].rolling(window=50).mean()
df['volatility_ratio'] = df['volatility'] / df['avg_volatility']
# Khi biến động cao, tăng ngưỡng Z-Score
# Khi biến động thấp, giảm ngưỡng Z-Score
df['adaptive_z_threshold'] = 2.0 * df['volatility_ratio']
return df
Quản lý Rủi ro
Nguyên tắc Quan trọng
- Risk per Trade: Không bao giờ rủi ro quá 2% tài khoản mỗi lệnh
- Stop Loss bắt buộc: Luôn đặt Stop Loss khi vào lệnh
- Take Profit: Đặt Take Profit tại dải giữa (mean)
- Position Sizing: Tính toán chính xác dựa trên Stop Loss
- Tránh xu hướng mạnh: Mean Reversion hoạt động tốt trong sideways market
Công thức Position Sizing
Position Size = (Account Balance × Risk %) / (Entry Price - Stop Loss Price)
Kết quả và Hiệu suất
Metrics Quan trọng
Khi đánh giá hiệu suất bot:
- Win Rate: Tỷ lệ lệnh thắng (mục tiêu: > 55% cho Mean Reversion)
- Profit Factor: Tổng lợi nhuận / Tổng lỗ (mục tiêu: > 1.5)
- Max Drawdown: Mức sụt giảm tối đa (mục tiêu: < 15%)
- Average Win/Loss Ratio: Tỷ lệ lợi nhuận trung bình / lỗ trung bình (mục tiêu: > 1.5)
- Sharpe Ratio: Lợi nhuận điều chỉnh theo rủi ro (mục tiêu: > 1.0)
Ví dụ Kết quả Backtest
Period: 2023-01-01 to 2024-01-01 (1 year)
Symbol: BTC/USDT
Timeframe: 1h
Initial Capital: $10,000
Results:
- Total Trades: 142
- Winning Trades: 85 (59.9%)
- Losing Trades: 57 (40.1%)
- Win Rate: 59.9%
- Total Return: +28.5%
- Final Capital: $12,850
- Profit Factor: 1.65
- Max Drawdown: -6.8%
- Average Win: $95.20
- Average Loss: -$57.80
- Sharpe Ratio: 1.32
Lưu ý Quan trọng
Cảnh báo Rủi ro
- Giao dịch có rủi ro cao: Có thể mất toàn bộ vốn đầu tư
- Mean Reversion không phù hợp xu hướng mạnh: Trong trending market, giá có thể tiếp tục đi xa
- Backtest không đảm bảo: Kết quả backtest không đảm bảo lợi nhuận thực tế
- Market conditions: Chiến lược hoạt động tốt hơn trong thị trường sideways
- False signals: Cần filter cẩn thận để tránh tín hiệu giả
Best Practices
- Bắt đầu với Testnet: Test kỹ lưỡng trên testnet ít nhất 1 tháng
- Bắt đầu nhỏ: Khi chuyển sang live, bắt đầu với số tiền nhỏ
- Giám sát thường xuyên: Không để bot chạy hoàn toàn tự động
- Cập nhật thường xuyên: Theo dõi và cập nhật bot khi thị trường thay đổi
- Logging đầy đủ: Ghi log mọi hoạt động để phân tích
- Error Handling: Xử lý lỗi kỹ lưỡng
- Filter xu hướng: Tránh giao dịch Mean Reversion trong trending market
Tài liệu Tham khảo
Tài liệu Mean Reversion
- “Mean Reversion Trading” – Howard Bandy
- “Quantitative Trading” – Ernest P. Chan
- “Algorithmic Trading” – Ernest P. Chan
Tài liệu CCXT
Cộng đồng
Kết luận
Chiến lược Mean Reversion là một phương pháp giao dịch hiệu quả khi được thực hiện đúng cách. Bot trong bài viết này cung cấp:
- Tính toán Bollinger Bands, Z-Score, RSI chính xác
- Phát hiện tín hiệu Mean Reversion tự động
- Xác nhận bằng nhiều chỉ báo để giảm false signals
- Quản lý rủi ro chặt chẽ với Stop Loss và Position Sizing
- Backtesting đầy đủ để đánh giá hiệu suất
- Tự động hóa hoàn toàn giao dịch
Tuy nhiên, hãy nhớ rằng:
- Không có chiến lược hoàn hảo: Mọi chiến lược đều có thể thua lỗ
- Quản lý rủi ro là số 1: Luôn ưu tiên bảo vệ vốn
- Kiên nhẫn và kỷ luật: Tuân thủ quy tắc, không giao dịch theo cảm xúc
- Học hỏi liên tục: Thị trường luôn thay đổi, cần cập nhật kiến thức
- Mean Reversion phù hợp sideways: Tránh giao dịch trong trending market mạnh
Chúc bạn giao dịch thành công!
Tác giả: Hướng Nghiệp Data
Ngày đăng: 2024
Tags: #MeanReversion #TradingBot #Python #AlgorithmicTrading