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 Liquidity Grab trong Bot Auto Trading
Được viết bởi thanhdt vào ngày 16/11/2025 lúc 22:47 | 10 lượt xem
Chiến lược Liquidity Grab trong Bot Auto Trading: Hướng dẫn Python
Liquidity Grab là một chiến lược trading nâng cao dựa trên lý thuyết Smart Money Concepts (SMC). Chiến lược này tập trung vào việc phát hiện các vùng thanh khoản (liquidity zones) nơi các nhà giao dịch tổ chức thường “grab” (lấy) thanh khoản từ các trader nhỏ lẻ trước khi đảo chiều giá. Trong bài viết này, chúng ta sẽ tìm hiểu cách triển khai chiến lược Liquidity Grab hiệu quả bằng Python.
1. Hiểu về Liquidity Grab
Liquidity Grab xảy ra khi giá phá vỡ một mức hỗ trợ hoặc kháng cự quan trọng, kích hoạt các lệnh stop loss của retail traders, sau đó giá nhanh chóng đảo chiều. Đây là một kỹ thuật được các tổ chức tài chính lớn sử dụng để thu thập thanh khoản trước khi di chuyển giá theo hướng mong muốn.
Đặc điểm của Liquidity Grab:
- False Breakout: Giá phá vỡ mức nhưng không tiếp tục theo hướng phá vỡ
- Wick Rejection: Nến có wick dài (bóng nến) sau khi phá vỡ
- Quick Reversal: Giá đảo chiều nhanh chóng sau khi grab liquidity
- Volume Spike: Thường có volume tăng đột biến khi grab xảy ra
Các loại Liquidity Zones:
- Equal Highs/Lows: Nhiều đỉnh/đáy ở cùng mức giá
- Previous High/Low: Đỉnh/đáy trước đó
- Order Blocks: Vùng có nhiều lệnh chờ (pending orders)
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from scipy.signal import find_peaks, find_peaks_inverse
def identify_liquidity_zones(df, lookback=50, min_touches=2):
"""
Xác định các vùng liquidity (equal highs/lows)
Parameters:
-----------
df : pd.DataFrame
DataFrame chứa OHLCV data
lookback : int
Số nến để xem lại
min_touches : int
Số lần chạm tối thiểu để coi là liquidity zone
Returns:
--------
dict: Chứa các liquidity zones
"""
recent_data = df.tail(lookback)
# Tìm các đỉnh và đáy
high_peaks, _ = find_peaks(recent_data['High'].values, distance=5)
low_peaks, _ = find_peaks(-recent_data['Low'].values, distance=5)
# Nhóm các đỉnh/đáy gần nhau
tolerance = 0.001 # 0.1% tolerance
liquidity_zones = {
'equal_highs': [],
'equal_lows': [],
'resistance': [],
'support': []
}
# Xử lý equal highs
if len(high_peaks) >= min_touches:
high_values = recent_data['High'].iloc[high_peaks].values
high_indices = recent_data.index[high_peaks]
# Nhóm các đỉnh có giá gần nhau
for i, high_val in enumerate(high_values):
similar_highs = [high_val]
similar_indices = [high_indices[i]]
for j, other_high in enumerate(high_values):
if i != j and abs(high_val - other_high) / high_val < tolerance:
similar_highs.append(other_high)
similar_indices.append(high_indices[j])
if len(similar_highs) >= min_touches:
avg_high = np.mean(similar_highs)
liquidity_zones['equal_highs'].append({
'price': avg_high,
'touches': len(similar_highs),
'indices': similar_indices
})
# Xử lý equal lows
if len(low_peaks) >= min_touches:
low_values = recent_data['Low'].iloc[low_peaks].values
low_indices = recent_data.index[low_peaks]
# Nhóm các đáy có giá gần nhau
for i, low_val in enumerate(low_values):
similar_lows = [low_val]
similar_indices = [low_indices[i]]
for j, other_low in enumerate(low_values):
if i != j and abs(low_val - other_low) / low_val < tolerance:
similar_lows.append(other_low)
similar_indices.append(low_indices[j])
if len(similar_lows) >= min_touches:
avg_low = np.mean(similar_lows)
liquidity_zones['equal_lows'].append({
'price': avg_low,
'touches': len(similar_lows),
'indices': similar_indices
})
return liquidity_zones
2. Các chiến lược Liquidity Grab hiệu quả
2.1. Chiến lược Equal Highs/Lows Grab
Đặc điểm:
- Phát hiện khi giá phá vỡ equal highs/lows
- Chờ tín hiệu rejection (wick rejection)
- Vào lệnh theo hướng đảo chiều
Quy tắc:
- Mua: Giá phá vỡ equal lows, tạo wick rejection, sau đó đảo chiều tăng
- Bán: Giá phá vỡ equal highs, tạo wick rejection, sau đó đảo chiều giảm
class EqualHighsLowsGrabStrategy:
"""Chiến lược Liquidity Grab với Equal Highs/Lows"""
def __init__(self, lookback=50, min_touches=2, wick_ratio=0.6):
"""
Parameters:
-----------
lookback : int
Số nến để xem lại
min_touches : int
Số lần chạm tối thiểu
wick_ratio : float
Tỷ lệ wick tối thiểu (0.6 = 60% body)
"""
self.lookback = lookback
self.min_touches = min_touches
self.wick_ratio = wick_ratio
def identify_liquidity_zones(self, df):
"""Xác định liquidity zones"""
return identify_liquidity_zones(df, self.lookback, self.min_touches)
def detect_wick_rejection(self, df, index, zone_price, is_resistance=True):
"""
Phát hiện wick rejection
Parameters:
-----------
df : pd.DataFrame
Dữ liệu OHLCV
index : int
Chỉ số nến hiện tại
zone_price : float
Giá của liquidity zone
is_resistance : bool
True nếu là resistance zone
"""
if index >= len(df):
return False
candle = df.iloc[index]
body_size = abs(candle['Close'] - candle['Open'])
candle_range = candle['High'] - candle['Low']
if candle_range == 0:
return False
if is_resistance:
# Wick rejection ở resistance: giá phá vỡ lên nhưng đóng cửa dưới
upper_wick = candle['High'] - max(candle['Open'], candle['Close'])
wick_ratio = upper_wick / candle_range
return (candle['High'] > zone_price and
candle['Close'] < zone_price and
wick_ratio >= self.wick_ratio)
else:
# Wick rejection ở support: giá phá vỡ xuống nhưng đóng cửa trên
lower_wick = min(candle['Open'], candle['Close']) - candle['Low']
wick_ratio = lower_wick / candle_range
return (candle['Low'] < zone_price and
candle['Close'] > zone_price and
wick_ratio >= self.wick_ratio)
def generate_signals(self, df):
"""
Tạo tín hiệu giao dịch
Returns:
--------
pd.Series: 1 = Mua, -1 = Bán, 0 = Giữ
"""
df = df.copy()
df['Signal'] = 0
for i in range(self.lookback, len(df)):
window_df = df.iloc[i-self.lookback:i+1]
liquidity_zones = self.identify_liquidity_zones(window_df.iloc[:-1])
current_candle = df.iloc[i]
# Kiểm tra grab ở equal lows (bullish)
for zone in liquidity_zones['equal_lows']:
zone_price = zone['price']
# Kiểm tra xem giá có phá vỡ zone không
if (current_candle['Low'] < zone_price * 0.999 and # Phá vỡ xuống
self.detect_wick_rejection(df, i, zone_price, is_resistance=False)):
# Xác nhận đảo chiều: nến tiếp theo tăng
if i < len(df) - 1:
next_candle = df.iloc[i+1]
if next_candle['Close'] > current_candle['Close']:
df.iloc[i+1, df.columns.get_loc('Signal')] = 1
break
# Kiểm tra grab ở equal highs (bearish)
for zone in liquidity_zones['equal_highs']:
zone_price = zone['price']
# Kiểm tra xem giá có phá vỡ zone không
if (current_candle['High'] > zone_price * 1.001 and # Phá vỡ lên
self.detect_wick_rejection(df, i, zone_price, is_resistance=True)):
# Xác nhận đảo chiều: nến tiếp theo giảm
if i < len(df) - 1:
next_candle = df.iloc[i+1]
if next_candle['Close'] < current_candle['Close']:
df.iloc[i+1, df.columns.get_loc('Signal')] = -1
break
return df['Signal']
2.2. Chiến lược Previous High/Low Grab (Hiệu quả cao)
Đặc điểm:
- Phát hiện grab ở previous high/low
- Kết hợp với volume để xác nhận
- Tín hiệu mạnh và đáng tin cậy
Quy tắc:
- Mua: Giá phá vỡ previous low, có volume spike, sau đó đảo chiều
- Bán: Giá phá vỡ previous high, có volume spike, sau đó đảo chiều
class PreviousHighLowGrabStrategy:
"""Chiến lược Liquidity Grab với Previous High/Low"""
def __init__(self, lookback=100, volume_multiplier=1.5, confirmation_candles=2):
"""
Parameters:
-----------
lookback : int
Số nến để tìm previous high/low
volume_multiplier : float
Hệ số volume (volume hiện tại > avg * multiplier)
confirmation_candles : int
Số nến xác nhận đảo chiều
"""
self.lookback = lookback
self.volume_multiplier = volume_multiplier
self.confirmation_candles = confirmation_candles
def find_previous_high_low(self, df, current_index):
"""Tìm previous high và low"""
if current_index < self.lookback:
return None, None
window_df = df.iloc[current_index-self.lookback:current_index]
previous_high = window_df['High'].max()
previous_low = window_df['Low'].min()
return previous_high, previous_low
def check_volume_spike(self, df, index):
"""Kiểm tra volume spike"""
if index < 20:
return False
current_volume = df.iloc[index]['Volume']
avg_volume = df.iloc[index-20:index]['Volume'].mean()
return current_volume > avg_volume * self.volume_multiplier
def confirm_reversal(self, df, index, direction):
"""
Xác nhận đảo chiều
Parameters:
-----------
direction : str
'bullish' hoặc 'bearish'
"""
if index + self.confirmation_candles >= len(df):
return False
confirmation_window = df.iloc[index+1:index+1+self.confirmation_candles]
if direction == 'bullish':
# Xác nhận tăng: các nến sau đóng cửa cao hơn
return all(confirmation_window['Close'].iloc[i] >
confirmation_window['Close'].iloc[i-1]
for i in range(1, len(confirmation_window)))
else: # bearish
# Xác nhận giảm: các nến sau đóng cửa thấp hơn
return all(confirmation_window['Close'].iloc[i] <
confirmation_window['Close'].iloc[i-1]
for i in range(1, len(confirmation_window)))
def generate_signals(self, df):
"""Tạo tín hiệu giao dịch"""
df = df.copy()
df['Signal'] = 0
for i in range(self.lookback, len(df) - self.confirmation_candles):
previous_high, previous_low = self.find_previous_high_low(df, i)
if previous_high is None or previous_low is None:
continue
current_candle = df.iloc[i]
# Bullish grab: Phá vỡ previous low
if (current_candle['Low'] < previous_low * 0.999 and
self.check_volume_spike(df, i) and
current_candle['Close'] > previous_low):
if self.confirm_reversal(df, i, 'bullish'):
# Vào lệnh ở nến xác nhận
entry_index = i + self.confirmation_candles
if entry_index < len(df):
df.iloc[entry_index, df.columns.get_loc('Signal')] = 1
# Bearish grab: Phá vỡ previous high
if (current_candle['High'] > previous_high * 1.001 and
self.check_volume_spike(df, i) and
current_candle['Close'] < previous_high):
if self.confirm_reversal(df, i, 'bearish'):
# Vào lệnh ở nến xác nhận
entry_index = i + self.confirmation_candles
if entry_index < len(df):
df.iloc[entry_index, df.columns.get_loc('Signal')] = -1
return df['Signal']
2.3. Chiến lược Order Block Grab (Nâng cao – Rất hiệu quả)
Đặc điểm:
- Phát hiện order blocks (vùng có nhiều lệnh chờ)
- Kết hợp với market structure
- Tín hiệu mạnh, độ chính xác cao
Quy tắc:
- Mua: Grab liquidity ở order block bearish, sau đó đảo chiều tăng
- Bán: Grab liquidity ở order block bullish, sau đó đảo chiều giảm
class OrderBlockGrabStrategy:
"""Chiến lược Liquidity Grab với Order Blocks"""
def __init__(self, lookback=50, order_block_candles=3):
"""
Parameters:
-----------
lookback : int
Số nến để xem lại
order_block_candles : int
Số nến để xác định order block
"""
self.lookback = lookback
self.order_block_candles = order_block_candles
def identify_order_blocks(self, df):
"""
Xác định order blocks
Order block là vùng giá nơi có nến mạnh (strong candle)
trước khi đảo chiều xu hướng
"""
order_blocks = []
for i in range(self.order_block_candles, len(df) - 1):
# Kiểm tra order block bearish (trước khi tăng)
block_candles = df.iloc[i-self.order_block_candles:i]
# Tìm nến mạnh giảm
strong_bearish = block_candles[
(block_candles['Close'] < block_candles['Open']) &
((block_candles['Close'] - block_candles['Open']) /
(block_candles['High'] - block_candles['Low']) > 0.7)
]
if len(strong_bearish) > 0:
# Kiểm tra xem có đảo chiều tăng sau đó không
next_candles = df.iloc[i:i+3]
if all(next_candles['Close'] > next_candles['Close'].shift(1).fillna(0)):
# Đây là order block bearish (sẽ là support sau này)
ob_low = strong_bearish['Low'].min()
ob_high = strong_bearish['High'].max()
order_blocks.append({
'type': 'bearish', # Order block bearish = support
'low': ob_low,
'high': ob_high,
'index': i
})
# Tìm order block bullish (trước khi giảm)
strong_bullish = block_candles[
(block_candles['Close'] > block_candles['Open']) &
((block_candles['Close'] - block_candles['Open']) /
(block_candles['High'] - block_candles['Low']) > 0.7)
]
if len(strong_bullish) > 0:
# Kiểm tra xem có đảo chiều giảm sau đó không
next_candles = df.iloc[i:i+3]
if all(next_candles['Close'] < next_candles['Close'].shift(1).fillna(0)):
# Đây là order block bullish (sẽ là resistance sau này)
ob_low = strong_bullish['Low'].min()
ob_high = strong_bullish['High'].max()
order_blocks.append({
'type': 'bullish', # Order block bullish = resistance
'low': ob_low,
'high': ob_high,
'index': i
})
return order_blocks
def detect_liquidity_grab_at_order_block(self, df, order_blocks, current_index):
"""Phát hiện liquidity grab ở order block"""
current_candle = df.iloc[current_index]
for ob in order_blocks:
# Chỉ xem các order block gần đây
if current_index - ob['index'] > 100:
continue
if ob['type'] == 'bearish': # Order block bearish = support
# Grab: Giá phá vỡ xuống order block nhưng đảo chiều
if (current_candle['Low'] < ob['low'] * 0.999 and
current_candle['Close'] > ob['low']):
# Có wick rejection
wick_size = min(current_candle['Open'], current_candle['Close']) - current_candle['Low']
candle_range = current_candle['High'] - current_candle['Low']
if candle_range > 0 and wick_size / candle_range > 0.4:
return 'bullish' # Tín hiệu mua
elif ob['type'] == 'bullish': # Order block bullish = resistance
# Grab: Giá phá vỡ lên order block nhưng đảo chiều
if (current_candle['High'] > ob['high'] * 1.001 and
current_candle['Close'] < ob['high']):
# Có wick rejection
wick_size = current_candle['High'] - max(current_candle['Open'], current_candle['Close'])
candle_range = current_candle['High'] - current_candle['Low']
if candle_range > 0 and wick_size / candle_range > 0.4:
return 'bearish' # Tín hiệu bán
return None
def generate_signals(self, df):
"""Tạo tín hiệu giao dịch"""
df = df.copy()
df['Signal'] = 0
# Xác định order blocks
order_blocks = self.identify_order_blocks(df)
for i in range(self.lookback, len(df)):
signal = self.detect_liquidity_grab_at_order_block(df, order_blocks, i)
if signal == 'bullish':
df.iloc[i, df.columns.get_loc('Signal')] = 1
elif signal == 'bearish':
df.iloc[i, df.columns.get_loc('Signal')] = -1
return df['Signal']
2.4. Chiến lược Liquidity Grab với Market Structure (Rất hiệu quả)
Đặc điểm:
- Kết hợp liquidity grab với market structure
- Phân tích higher highs/lower lows
- Tín hiệu đáng tin cậy nhất
Quy tắc:
- Mua: Grab ở lower low trong uptrend, sau đó tạo higher low
- Bán: Grab ở higher high trong downtrend, sau đó tạo lower high
class MarketStructureLiquidityGrabStrategy:
"""Chiến lược Liquidity Grab với Market Structure"""
def __init__(self, lookback=100, swing_period=10):
"""
Parameters:
-----------
lookback : int
Số nến để phân tích
swing_period : int
Period để xác định swing high/low
"""
self.lookback = lookback
self.swing_period = swing_period
def identify_swing_points(self, df):
"""Xác định swing highs và swing lows"""
swing_highs = []
swing_lows = []
for i in range(self.swing_period, len(df) - self.swing_period):
# Swing high: điểm cao nhất trong window
window_highs = df.iloc[i-self.swing_period:i+self.swing_period+1]['High']
if df.iloc[i]['High'] == window_highs.max():
swing_highs.append({
'index': i,
'price': df.iloc[i]['High'],
'date': df.index[i]
})
# Swing low: điểm thấp nhất trong window
window_lows = df.iloc[i-self.swing_period:i+self.swing_period+1]['Low']
if df.iloc[i]['Low'] == window_lows.min():
swing_lows.append({
'index': i,
'price': df.iloc[i]['Low'],
'date': df.index[i]
})
return swing_highs, swing_lows
def determine_market_structure(self, swing_highs, swing_lows):
"""
Xác định market structure
Returns:
--------
str: 'uptrend', 'downtrend', hoặc 'range'
"""
if len(swing_highs) < 2 or len(swing_lows) < 2:
return 'range'
# Kiểm tra higher highs và higher lows (uptrend)
recent_highs = sorted(swing_highs, key=lambda x: x['index'])[-3:]
recent_lows = sorted(swing_lows, key=lambda x: x['index'])[-3:]
if len(recent_highs) >= 2 and len(recent_lows) >= 2:
# Higher highs
hh = recent_highs[-1]['price'] > recent_highs[-2]['price']
# Higher lows
hl = recent_lows[-1]['price'] > recent_lows[-2]['price']
if hh and hl:
return 'uptrend'
# Lower highs
lh = recent_highs[-1]['price'] < recent_highs[-2]['price']
# Lower lows
ll = recent_lows[-1]['price'] < recent_lows[-2]['price']
if lh and ll:
return 'downtrend'
return 'range'
def detect_liquidity_grab_with_structure(self, df, swing_highs, swing_lows,
market_structure, current_index):
"""Phát hiện liquidity grab dựa trên market structure"""
current_candle = df.iloc[current_index]
if market_structure == 'uptrend':
# Trong uptrend, tìm grab ở lower low
recent_lows = [s for s in swing_lows if s['index'] < current_index]
if len(recent_lows) >= 2:
previous_low = recent_lows[-1]['price']
# Grab: Giá phá vỡ previous low nhưng đảo chiều
if (current_candle['Low'] < previous_low * 0.999 and
current_candle['Close'] > previous_low):
# Xác nhận: nến sau tạo higher low
if current_index < len(df) - 3:
next_lows = df.iloc[current_index+1:current_index+4]['Low']
if next_lows.min() > previous_low:
return 1 # Tín hiệu mua
elif market_structure == 'downtrend':
# Trong downtrend, tìm grab ở higher high
recent_highs = [s for s in swing_highs if s['index'] < current_index]
if len(recent_highs) >= 2:
previous_high = recent_highs[-1]['price']
# Grab: Giá phá vỡ previous high nhưng đảo chiều
if (current_candle['High'] > previous_high * 1.001 and
current_candle['Close'] < previous_high):
# Xác nhận: nến sau tạo lower high
if current_index < len(df) - 3:
next_highs = df.iloc[current_index+1:current_index+4]['High']
if next_highs.max() < previous_high:
return -1 # Tín hiệu bán
return 0
def generate_signals(self, df):
"""Tạo tín hiệu giao dịch"""
df = df.copy()
df['Signal'] = 0
# Xác định swing points
swing_highs, swing_lows = self.identify_swing_points(df)
for i in range(self.lookback, len(df)):
# Xác định market structure
market_structure = self.determine_market_structure(swing_highs, swing_lows)
# Phát hiện liquidity grab
signal = self.detect_liquidity_grab_with_structure(
df, swing_highs, swing_lows, market_structure, i
)
if signal != 0:
df.iloc[i, df.columns.get_loc('Signal')] = signal
return df['Signal']
3. Bot Auto Trading Liquidity Grab hoàn chỉnh
3.1. Bot với Quản lý Rủi ro và Entry Management
import ccxt
import pandas as pd
import numpy as np
import time
from datetime import datetime
from typing import Dict, Optional
class LiquidityGrabTradingBot:
"""Bot auto trading sử dụng chiến lược Liquidity Grab"""
def __init__(self, exchange_name: str, api_key: str, api_secret: str,
strategy_type: str = 'previous_high_low'):
"""
Khởi tạo bot
Parameters:
-----------
exchange_name : str
Tên sàn (binance, coinbase, etc.)
api_key : str
API key
api_secret : str
API secret
strategy_type : str
Loại chiến lược ('equal_highs_lows', 'previous_high_low',
'order_block', 'market_structure')
"""
# Kết nối exchange
exchange_class = getattr(ccxt, exchange_name)
self.exchange = exchange_class({
'apiKey': api_key,
'secret': api_secret,
'enableRateLimit': True,
})
# Chọn chiến lược
self.strategy = self._init_strategy(strategy_type)
# Quản lý vị thế
self.position = None
self.entry_price = None
self.stop_loss = None
self.take_profit = None
self.liquidity_zone = None
# Cài đặt rủi ro
self.max_position_size = 0.1 # 10% vốn
self.risk_reward_ratio = 2.0 # Risk:Reward = 1:2
def _init_strategy(self, strategy_type: str):
"""Khởi tạo chiến lược"""
if strategy_type == 'equal_highs_lows':
return EqualHighsLowsGrabStrategy()
elif strategy_type == 'previous_high_low':
return PreviousHighLowGrabStrategy()
elif strategy_type == 'order_block':
return OrderBlockGrabStrategy()
elif strategy_type == 'market_structure':
return MarketStructureLiquidityGrabStrategy()
else:
raise ValueError(f"Unknown strategy type: {strategy_type}")
def get_market_data(self, symbol: str, timeframe: str = '1h', limit: int = 200):
"""Lấy dữ liệu thị trường"""
ohlcv = self.exchange.fetch_ohlcv(symbol, 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)
df.columns = [col.capitalize() for col in df.columns]
return df
def calculate_stop_loss_take_profit(self, entry_price: float, side: str,
liquidity_zone: float):
"""
Tính stop loss và take profit dựa trên liquidity zone
Parameters:
-----------
entry_price : float
Giá vào lệnh
side : str
'long' hoặc 'short'
liquidity_zone : float
Giá của liquidity zone đã bị grab
"""
if side == 'long':
# Stop loss dưới liquidity zone
stop_loss = liquidity_zone * 0.998
risk = entry_price - stop_loss
take_profit = entry_price + (risk * self.risk_reward_ratio)
else: # short
# Stop loss trên liquidity zone
stop_loss = liquidity_zone * 1.002
risk = stop_loss - entry_price
take_profit = entry_price - (risk * self.risk_reward_ratio)
return stop_loss, take_profit
def calculate_position_size(self, balance: float, entry_price: float,
stop_loss: float) -> float:
"""Tính toán kích thước vị thế dựa trên rủi ro"""
risk_amount = balance * 0.01 # Risk 1% mỗi lệnh
risk_per_unit = abs(entry_price - stop_loss)
if risk_per_unit == 0:
return 0
position_size = risk_amount / risk_per_unit
return position_size
def place_order(self, symbol: str, side: str, amount: float,
order_type: str = 'market'):
"""Đặt lệnh giao dịch"""
try:
if side == 'buy':
order = self.exchange.create_market_buy_order(symbol, amount)
else:
order = self.exchange.create_market_sell_order(symbol, amount)
print(f"[{datetime.now()}] {side.upper()} {amount} {symbol} @ {order['price']}")
return order
except Exception as e:
print(f"Error placing order: {e}")
return None
def check_stop_loss_take_profit(self, current_price: float):
"""Kiểm tra stop loss và take profit"""
if self.position is None:
return
if self.position == 'long':
if current_price <= self.stop_loss:
print(f"[{datetime.now()}] Stop Loss triggered @ {current_price}")
self.close_position(current_price)
return
if current_price >= self.take_profit:
print(f"[{datetime.now()}] Take Profit triggered @ {current_price}")
self.close_position(current_price)
return
elif self.position == 'short':
if current_price >= self.stop_loss:
print(f"[{datetime.now()}] Stop Loss triggered @ {current_price}")
self.close_position(current_price)
return
if current_price <= self.take_profit:
print(f"[{datetime.now()}] Take Profit triggered @ {current_price}")
self.close_position(current_price)
return
def open_position(self, symbol: str, side: str, price: float, amount: float,
liquidity_zone: float):
"""Mở vị thế"""
order = self.place_order(symbol, side, amount)
if order:
self.position = side
self.entry_price = price
self.liquidity_zone = liquidity_zone
# Đặt stop loss và take profit
self.stop_loss, self.take_profit = self.calculate_stop_loss_take_profit(
price, side, liquidity_zone
)
print(f"[{datetime.now()}] Position opened: {side} @ {price}")
print(f"Stop Loss: {self.stop_loss}, Take Profit: {self.take_profit}")
def close_position(self, price: float):
"""Đóng vị thế"""
if self.position:
if self.position == 'long':
pnl_pct = ((price - self.entry_price) / self.entry_price) * 100
else: # short
pnl_pct = ((self.entry_price - price) / self.entry_price) * 100
print(f"[{datetime.now()}] Position closed. P&L: {pnl_pct:.2f}%")
self.position = None
self.entry_price = None
self.stop_loss = None
self.take_profit = None
self.liquidity_zone = None
def find_liquidity_zone(self, df, signal_index, side):
"""Tìm liquidity zone đã bị grab"""
if signal_index < 10:
return None
# Tìm liquidity zone gần nhất
window_df = df.iloc[signal_index-50:signal_index]
if side == 'long':
# Tìm previous low đã bị phá vỡ
lows = window_df['Low'].values
current_low = df.iloc[signal_index]['Low']
# Tìm low gần nhất bị phá vỡ
for i in range(len(lows)-1, -1, -1):
if lows[i] > current_low:
return lows[i]
else: # short
# Tìm previous high đã bị phá vỡ
highs = window_df['High'].values
current_high = df.iloc[signal_index]['High']
# Tìm high gần nhất bị phá vỡ
for i in range(len(highs)-1, -1, -1):
if highs[i] < current_high:
return highs[i]
return None
def run(self, symbol: str, timeframe: str = '1h', check_interval: int = 300):
"""Chạy bot"""
print(f"[{datetime.now()}] Bot started for {symbol}")
while True:
try:
# Lấy dữ liệu thị trường
df = self.get_market_data(symbol, timeframe)
current_price = df['Close'].iloc[-1]
# Kiểm tra stop loss và take profit
if self.position:
self.check_stop_loss_take_profit(current_price)
if self.position is None:
time.sleep(check_interval)
continue
# Tạo tín hiệu
signals = self.strategy.generate_signals(df)
signal = signals.iloc[-1]
signal_index = len(df) - 1
# Xử lý tín hiệu
if signal == 1 and self.position != 'long':
# Tín hiệu mua
liquidity_zone = self.find_liquidity_zone(df, signal_index, 'long')
if liquidity_zone:
balance = self.exchange.fetch_balance()
available_balance = balance['USDT']['free'] if 'USDT' in balance else balance['total']['USDT']
stop_loss, _ = self.calculate_stop_loss_take_profit(
current_price, 'long', liquidity_zone
)
amount = self.calculate_position_size(
available_balance, current_price, stop_loss
)
if amount > 0:
self.open_position(symbol, 'long', current_price, amount, liquidity_zone)
elif signal == -1 and self.position != 'short':
# Tín hiệu bán
liquidity_zone = self.find_liquidity_zone(df, signal_index, 'short')
if liquidity_zone:
balance = self.exchange.fetch_balance()
available_balance = balance['USDT']['free'] if 'USDT' in balance else balance['total']['USDT']
stop_loss, _ = self.calculate_stop_loss_take_profit(
current_price, 'short', liquidity_zone
)
amount = self.calculate_position_size(
available_balance, current_price, stop_loss
)
if amount > 0:
self.open_position(symbol, 'short', current_price, amount, liquidity_zone)
time.sleep(check_interval)
except KeyboardInterrupt:
print(f"[{datetime.now()}] Bot stopped by user")
break
except Exception as e:
print(f"[{datetime.now()}] Error: {e}")
time.sleep(check_interval)
4. Backtesting Chiến lược Liquidity Grab
4.1. Hàm Backtest
def backtest_liquidity_grab_strategy(df, strategy, initial_capital=10000):
"""
Backtest chiến lược Liquidity Grab
Parameters:
-----------
df : pd.DataFrame
Dữ liệu OHLCV
strategy : Strategy object
Đối tượng chiến lược
initial_capital : float
Vốn ban đầu
Returns:
--------
dict: Kết quả backtest
"""
# Tạo tín hiệu
signals = strategy.generate_signals(df.copy())
df['Signal'] = signals
# Tính toán vị thế và lợi nhuận
capital = initial_capital
position = 0
entry_price = 0
trades = []
risk_reward_ratio = 2.0
for i in range(len(df)):
price = df['Close'].iloc[i]
signal = df['Signal'].iloc[i]
if signal == 1 and position == 0: # Mua
position = capital / price
entry_price = price
# Tìm liquidity zone để tính stop loss
window_df = df.iloc[max(0, i-50):i]
if len(window_df) > 0:
liquidity_zone = window_df['Low'].min()
stop_loss = liquidity_zone * 0.998
risk = entry_price - stop_loss
take_profit = entry_price + (risk * risk_reward_ratio)
else:
stop_loss = entry_price * 0.98
take_profit = entry_price * 1.04
trades.append({
'type': 'buy',
'date': df.index[i],
'entry_price': price,
'stop_loss': stop_loss,
'take_profit': take_profit,
'capital': capital
})
elif signal == -1 and position > 0: # Bán
capital = position * price
pnl = ((price - entry_price) / entry_price) * 100
if trades:
trades[-1]['exit_price'] = price
trades[-1]['pnl'] = pnl
trades[-1]['capital'] = capital
position = 0
# Kiểm tra stop loss và take profit
if position > 0 and trades:
last_trade = trades[-1]
if 'exit_price' not in last_trade:
if price <= last_trade['stop_loss']:
capital = position * price
pnl = ((price - entry_price) / entry_price) * 100
last_trade['exit_price'] = price
last_trade['pnl'] = pnl
last_trade['exit_reason'] = 'stop_loss'
position = 0
elif price >= last_trade['take_profit']:
capital = position * price
pnl = ((price - entry_price) / entry_price) * 100
last_trade['exit_price'] = price
last_trade['pnl'] = pnl
last_trade['exit_reason'] = 'take_profit'
position = 0
# Đóng vị thế cuối cùng nếu còn
if position > 0:
final_price = df['Close'].iloc[-1]
capital = position * final_price
if trades and 'exit_price' not in trades[-1]:
pnl = ((final_price - entry_price) / entry_price) * 100
trades[-1]['exit_price'] = final_price
trades[-1]['pnl'] = pnl
trades[-1]['exit_reason'] = 'end_of_data'
# Tính toán metrics
completed_trades = [t for t in trades if 'pnl' in t]
total_return = ((capital - initial_capital) / initial_capital) * 100
winning_trades = [t for t in completed_trades if t.get('pnl', 0) > 0]
losing_trades = [t for t in completed_trades if t.get('pnl', 0) < 0]
win_rate = len(winning_trades) / len(completed_trades) * 100 if completed_trades else 0
avg_win = np.mean([t['pnl'] for t in winning_trades]) if winning_trades else 0
avg_loss = np.mean([t['pnl'] for t in losing_trades]) if losing_trades else 0
return {
'initial_capital': initial_capital,
'final_capital': capital,
'total_return': total_return,
'total_trades': len(completed_trades),
'winning_trades': len(winning_trades),
'losing_trades': len(losing_trades),
'win_rate': win_rate,
'avg_win': avg_win,
'avg_loss': avg_loss,
'profit_factor': abs(avg_win / avg_loss) if avg_loss != 0 else 0,
'trades': trades
}
# Ví dụ sử dụng
import yfinance as yf
# Lấy dữ liệu
data = yf.download('BTC-USD', period='1y', interval='1h')
df = pd.DataFrame(data)
df.columns = [col.lower() for col in df.columns]
# Chạy backtest
strategy = PreviousHighLowGrabStrategy(lookback=100, volume_multiplier=1.5)
results = backtest_liquidity_grab_strategy(df, strategy, initial_capital=10000)
print(f"Total Return: {results['total_return']:.2f}%")
print(f"Win Rate: {results['win_rate']:.2f}%")
print(f"Total Trades: {results['total_trades']}")
print(f"Profit Factor: {results['profit_factor']:.2f}")
5. Tối ưu hóa tham số Liquidity Grab Strategy
5.1. Tìm tham số tối ưu
from itertools import product
def optimize_liquidity_grab_parameters(df, strategy_class, param_ranges):
"""
Tối ưu hóa tham số Liquidity Grab Strategy
"""
best_params = None
best_score = -float('inf')
best_results = None
param_names = list(param_ranges.keys())
param_values = list(param_ranges.values())
for params in product(*param_values):
param_dict = dict(zip(param_names, params))
try:
strategy = strategy_class(**param_dict)
results = backtest_liquidity_grab_strategy(df, strategy)
# Đánh giá: kết hợp return, win rate và profit factor
score = (results['total_return'] * 0.4 +
results['win_rate'] * 0.3 +
results['profit_factor'] * 10 * 0.3)
if score > best_score:
best_score = score
best_params = param_dict
best_results = results
except:
continue
return {
'best_params': best_params,
'best_score': best_score,
'results': best_results
}
# Ví dụ tối ưu hóa
param_ranges = {
'lookback': [50, 100, 150],
'volume_multiplier': [1.3, 1.5, 1.8],
'confirmation_candles': [1, 2, 3]
}
optimization_results = optimize_liquidity_grab_parameters(
df, PreviousHighLowGrabStrategy, param_ranges
)
print("Best Parameters:", optimization_results['best_params'])
print("Best Score:", optimization_results['best_score'])
6. Quản lý rủi ro với Liquidity Grab
6.1. Risk Management nâng cao
class LiquidityGrabRiskManager:
"""Quản lý rủi ro cho chiến lược Liquidity Grab"""
def __init__(self, max_risk_per_trade=0.01, max_daily_loss=0.05):
self.max_risk_per_trade = max_risk_per_trade
self.max_daily_loss = max_daily_loss
self.daily_pnl = 0
self.trades_today = 0
def can_trade(self, account_balance):
"""Kiểm tra xem có thể trade không"""
# Kiểm tra daily loss limit
if abs(self.daily_pnl) >= account_balance * self.max_daily_loss:
return False
return True
def calculate_position_size(self, account_balance, entry_price, stop_loss):
"""Tính toán kích thước vị thế"""
risk_amount = account_balance * self.max_risk_per_trade
risk_per_unit = abs(entry_price - stop_loss)
if risk_per_unit == 0:
return 0
position_size = risk_amount / risk_per_unit
return position_size
def update_daily_pnl(self, pnl):
"""Cập nhật P&L trong ngày"""
self.daily_pnl += pnl
self.trades_today += 1
7. Kết luận: Chiến lược Liquidity Grab nào hiệu quả nhất?
Đánh giá các chiến lược:
- Equal Highs/Lows Grab
- ✅ Đơn giản, dễ triển khai
- ❌ Cần nhiều touches để xác định zone
- ⭐ Hiệu quả: 3.5/5
- Previous High/Low Grab
- ✅ Tín hiệu rõ ràng, dễ phát hiện
- ✅ Kết hợp volume để xác nhận
- ⭐ Hiệu quả: 4.5/5
- Order Block Grab
- ✅ Tín hiệu mạnh, độ chính xác cao
- ❌ Phức tạp hơn, cần hiểu order blocks
- ⭐ Hiệu quả: 4.5/5
- Market Structure Liquidity Grab
- ✅ Tín hiệu đáng tin cậy nhất
- ✅ Kết hợp với market structure
- ⭐ Hiệu quả: 5/5
Khuyến nghị:
- Cho người mới bắt đầu: Previous High/Low Grab Strategy
- Cho trader có kinh nghiệm: Market Structure Liquidity Grab
- Cho scalping: Order Block Grab với khung thời gian ngắn (M15, M30)
Lưu ý quan trọng:
- Xác nhận Grab: Luôn chờ xác nhận đảo chiều sau khi grab
- Quản lý rủi ro: Luôn đặt stop loss dưới/trên liquidity zone
- Risk:Reward: Tỷ lệ Risk:Reward nên từ 1:2 trở lên
- Backtest kỹ lưỡng: Kiểm tra chiến lược trên nhiều thị trường khác nhau
- Theo dõi market structure: Chỉ trade khi market structure rõ ràng
- Tránh trade trong tin tức: Liquidity grab có thể bị ảnh hưởng bởi tin tức
8. Tài liệu tham khảo
- Smart Money Concepts – TradingView
- Liquidity in Trading – Investopedia
- Order Blocks Trading Strategy
- Market Structure Analysis
Lưu ý: Trading có rủi ro cao. Liquidity Grab là chiến lược nâng cao, cần hiểu rõ về Smart Money Concepts. Hãy luôn backtest kỹ lưỡng và bắt đầu với số vốn nhỏ. Bài viết này chỉ mang tính chất giáo dục, không phải lời khuyên đầu tư.