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 Multi-Timeframe (M15 + H1) trong Bot Auto Trading
Được viết bởi thanhdt vào ngày 17/11/2025 lúc 17:22 | 67 lượt xem
Chiến lược Multi-Timeframe (M15 + H1) trong Bot Auto Trading: Hướng dẫn Python
Multi-Timeframe Analysis là một trong những phương pháp trading hiệu quả nhất, đặc biệt khi kết hợp M15 (15 phút) và H1 (1 giờ). Chiến lược này cho phép bạn xác định xu hướng chính trên khung thời gian lớn (H1) và tìm điểm vào lệnh tối ưu trên khung thời gian nhỏ (M15). Trong bài viết này, chúng ta sẽ tìm hiểu cách triển khai chiến lược Multi-Timeframe M15 + H1 hiệu quả bằng Python.
1. Hiểu về Multi-Timeframe Analysis
Tại sao sử dụng Multi-Timeframe?
- Xác định xu hướng chính: Khung thời gian lớn (H1) cho biết xu hướng tổng thể
- Tìm điểm vào tối ưu: Khung thời gian nhỏ (M15) cho điểm vào lệnh chính xác
- Giảm false signals: Chỉ trade theo hướng xu hướng chính
- Tăng win rate: Kết hợp cả hai khung thời gian tăng độ chính xác
Quy tắc Multi-Timeframe cơ bản:
- Trend trên H1: Xác định xu hướng chính (uptrend/downtrend)
- Entry trên M15: Tìm điểm vào lệnh theo hướng xu hướng H1
- Confirmation: Cả hai khung thời gian phải đồng thuận
Tỷ lệ khung thời gian:
- H1 : M15 = 4 : 1 (1 giờ = 4 nến 15 phút)
- Đây là tỷ lệ lý tưởng để phân tích multi-timeframe
import pandas as pd
import numpy as np
import ccxt
import pandas_ta as ta
from datetime import datetime
def get_multiple_timeframes(exchange, symbol, timeframes=['15m', '1h'], limit=100):
"""
Lấy dữ liệu từ nhiều khung thời gian
Parameters:
-----------
exchange : ccxt.Exchange
Exchange object
symbol : str
Trading pair
timeframes : list
Danh sách khung thời gian
limit : int
Số nến cần lấy
Returns:
--------
dict: Dictionary chứa DataFrame cho mỗi timeframe
"""
data = {}
for tf in timeframes:
ohlcv = exchange.fetch_ohlcv(symbol, tf, 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]
data[tf] = df
return data
def determine_trend(df, method='ema'):
"""
Xác định xu hướng trên khung thời gian
Parameters:
-----------
df : pd.DataFrame
Dữ liệu OHLCV
method : str
Phương pháp xác định trend ('ema', 'sma', 'price_action')
Returns:
--------
str: 'uptrend', 'downtrend', hoặc 'sideways'
"""
if method == 'ema':
# Sử dụng EMA
ema_fast = df['Close'].ewm(span=20, adjust=False).mean()
ema_slow = df['Close'].ewm(span=50, adjust=False).mean()
current_fast = ema_fast.iloc[-1]
current_slow = ema_slow.iloc[-1]
prev_fast = ema_fast.iloc[-2]
prev_slow = ema_slow.iloc[-2]
# Uptrend: EMA fast trên EMA slow và đang tăng
if current_fast > current_slow and current_fast > prev_fast:
return 'uptrend'
# Downtrend: EMA fast dưới EMA slow và đang giảm
elif current_fast < current_slow and current_fast < prev_fast:
return 'downtrend'
else:
return 'sideways'
elif method == 'sma':
# Sử dụng SMA
sma_fast = df['Close'].rolling(window=20).mean()
sma_slow = df['Close'].rolling(window=50).mean()
if sma_fast.iloc[-1] > sma_slow.iloc[-1]:
return 'uptrend'
elif sma_fast.iloc[-1] < sma_slow.iloc[-1]:
return 'downtrend'
else:
return 'sideways'
elif method == 'price_action':
# Sử dụng price action (higher highs, lower lows)
recent_highs = df['High'].tail(20)
recent_lows = df['Low'].tail(20)
# Uptrend: Higher highs và higher lows
if (recent_highs.iloc[-1] > recent_highs.iloc[-10] and
recent_lows.iloc[-1] > recent_lows.iloc[-10]):
return 'uptrend'
# Downtrend: Lower highs và lower lows
elif (recent_highs.iloc[-1] < recent_highs.iloc[-10] and
recent_lows.iloc[-1] < recent_lows.iloc[-10]):
return 'downtrend'
else:
return 'sideways'
2. Các chiến lược Multi-Timeframe M15 + H1 hiệu quả
2.1. Chiến lược Trend Following (M15 + H1)
Đặc điểm:
- Xác định trend trên H1
- Tìm entry trên M15 theo hướng trend H1
- Đơn giản, dễ triển khai
Quy tắc:
- Mua: H1 uptrend + M15 pullback về support
- Bán: H1 downtrend + M15 pullback về resistance
class TrendFollowingMTFStrategy:
"""Chiến lược Trend Following Multi-Timeframe"""
def __init__(self, h1_ema_fast=20, h1_ema_slow=50,
m15_ema_fast=20, m15_ema_slow=50):
"""
Parameters:
-----------
h1_ema_fast : int
Period EMA nhanh cho H1
h1_ema_slow : int
Period EMA chậm cho H1
m15_ema_fast : int
Period EMA nhanh cho M15
m15_ema_slow : int
Period EMA chậm cho M15
"""
self.h1_ema_fast = h1_ema_fast
self.h1_ema_slow = h1_ema_slow
self.m15_ema_fast = m15_ema_fast
self.m15_ema_slow = m15_ema_slow
def analyze_h1_trend(self, df_h1):
"""Phân tích xu hướng trên H1"""
ema_fast = df_h1['Close'].ewm(span=self.h1_ema_fast, adjust=False).mean()
ema_slow = df_h1['Close'].ewm(span=self.h1_ema_slow, adjust=False).mean()
current_fast = ema_fast.iloc[-1]
current_slow = ema_slow.iloc[-1]
prev_fast = ema_fast.iloc[-2]
if current_fast > current_slow and current_fast > prev_fast:
return 'uptrend'
elif current_fast < current_slow and current_fast < prev_fast:
return 'downtrend'
else:
return 'sideways'
def find_m15_entry(self, df_m15, h1_trend):
"""
Tìm điểm vào lệnh trên M15
Parameters:
-----------
df_m15 : pd.DataFrame
Dữ liệu M15
h1_trend : str
Xu hướng trên H1
Returns:
--------
int: 1 = Mua, -1 = Bán, 0 = Giữ
"""
ema_fast = df_m15['Close'].ewm(span=self.m15_ema_fast, adjust=False).mean()
ema_slow = df_m15['Close'].ewm(span=self.m15_ema_slow, adjust=False).mean()
current_price = df_m15['Close'].iloc[-1]
current_fast = ema_fast.iloc[-1]
current_slow = ema_slow.iloc[-1]
prev_fast = ema_fast.iloc[-2]
# Chỉ trade theo hướng trend H1
if h1_trend == 'uptrend':
# Tìm pullback về support (EMA slow) và bounce
if (current_price > current_slow and # Giá trên EMA slow
prev_fast <= ema_slow.iloc[-2] and # EMA fast vừa cắt lên EMA slow
current_fast > current_slow): # EMA fast trên EMA slow
return 1 # Tín hiệu mua
elif h1_trend == 'downtrend':
# Tìm pullback về resistance (EMA slow) và rejection
if (current_price < current_slow and # Giá dưới EMA slow
prev_fast >= ema_slow.iloc[-2] and # EMA fast vừa cắt xuống EMA slow
current_fast < current_slow): # EMA fast dưới EMA slow
return -1 # Tín hiệu bán
return 0
def generate_signals(self, exchange, symbol):
"""
Tạo tín hiệu giao dịch
Returns:
--------
dict: Chứa trend H1, signal M15, và các thông tin khác
"""
# Lấy dữ liệu từ cả hai khung thời gian
data = get_multiple_timeframes(exchange, symbol, ['15m', '1h'])
df_h1 = data['1h']
df_m15 = data['15m']
# Phân tích trend trên H1
h1_trend = self.analyze_h1_trend(df_h1)
# Tìm entry trên M15
m15_signal = self.find_m15_entry(df_m15, h1_trend)
return {
'h1_trend': h1_trend,
'm15_signal': m15_signal,
'h1_price': df_h1['Close'].iloc[-1],
'm15_price': df_m15['Close'].iloc[-1],
'timestamp': datetime.now()
}
2.2. Chiến lược RSI Multi-Timeframe (Hiệu quả cao)
Đặc điểm:
- RSI trên H1 xác định xu hướng
- RSI trên M15 tìm điểm vào
- Kết hợp oversold/overbought trên cả hai khung
Quy tắc:
- Mua: RSI(H1) > 50 (uptrend) + RSI(M15) < 40 (oversold recovery)
- Bán: RSI(H1) < 50 (downtrend) + RSI(M15) > 60 (overbought rejection)
class RSIMultiTimeframeStrategy:
"""Chiến lược RSI Multi-Timeframe"""
def __init__(self, rsi_period=14, h1_oversold=40, h1_overbought=60,
m15_oversold=30, m15_overbought=70):
"""
Parameters:
-----------
rsi_period : int
Period cho RSI
h1_oversold : float
Ngưỡng oversold cho H1
h1_overbought : float
Ngưỡng overbought cho H1
m15_oversold : float
Ngưỡng oversold cho M15
m15_overbought : float
Ngưỡng overbought cho M15
"""
self.rsi_period = rsi_period
self.h1_oversold = h1_oversold
self.h1_overbought = h1_overbought
self.m15_oversold = m15_oversold
self.m15_overbought = m15_overbought
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 analyze_h1_rsi(self, df_h1):
"""Phân tích RSI trên H1"""
rsi = self.calculate_rsi(df_h1['Close'])
current_rsi = rsi.iloc[-1]
if current_rsi > 50:
return 'bullish' # Uptrend
elif current_rsi < 50:
return 'bearish' # Downtrend
else:
return 'neutral'
def find_m15_rsi_entry(self, df_m15, h1_bias):
"""
Tìm entry dựa trên RSI M15
Parameters:
-----------
h1_bias : str
'bullish', 'bearish', hoặc 'neutral'
"""
rsi = self.calculate_rsi(df_m15['Close'])
current_rsi = rsi.iloc[-1]
prev_rsi = rsi.iloc[-2]
# Chỉ trade theo hướng H1
if h1_bias == 'bullish':
# Tìm oversold recovery trên M15
if (current_rsi < self.m15_overbought and # Không quá overbought
current_rsi > self.m15_oversold and # Đang recovery từ oversold
current_rsi > prev_rsi): # RSI đang tăng
return 1 # Tín hiệu mua
elif h1_bias == 'bearish':
# Tìm overbought rejection trên M15
if (current_rsi > self.m15_oversold and # Không quá oversold
current_rsi < self.m15_overbought and # Đang rejection từ overbought
current_rsi < prev_rsi): # RSI đang giảm
return -1 # Tín hiệu bán
return 0
def generate_signals(self, exchange, symbol):
"""Tạo tín hiệu giao dịch"""
data = get_multiple_timeframes(exchange, symbol, ['15m', '1h'])
df_h1 = data['1h']
df_m15 = data['15m']
# Phân tích RSI trên H1
h1_bias = self.analyze_h1_rsi(df_h1)
# Tìm entry trên M15
m15_signal = self.find_m15_rsi_entry(df_m15, h1_bias)
# Tính RSI cho cả hai khung
rsi_h1 = self.calculate_rsi(df_h1['Close']).iloc[-1]
rsi_m15 = self.calculate_rsi(df_m15['Close']).iloc[-1]
return {
'h1_bias': h1_bias,
'h1_rsi': rsi_h1,
'm15_signal': m15_signal,
'm15_rsi': rsi_m15,
'h1_price': df_h1['Close'].iloc[-1],
'm15_price': df_m15['Close'].iloc[-1],
'timestamp': datetime.now()
}
2.3. Chiến lược MACD Multi-Timeframe (Nâng cao – Rất hiệu quả)
Đặc điểm:
- MACD trên H1 xác định xu hướng chính
- MACD trên M15 tìm điểm vào
- Kết hợp MACD crossover và histogram
Quy tắc:
- Mua: MACD(H1) bullish + MACD(M15) cắt lên Signal
- Bán: MACD(H1) bearish + MACD(M15) cắt xuống Signal
class MACDMultiTimeframeStrategy:
"""Chiến lược MACD Multi-Timeframe"""
def __init__(self, macd_fast=12, macd_slow=26, macd_signal=9):
"""
Parameters:
-----------
macd_fast : int
Period EMA nhanh cho MACD
macd_slow : int
Period EMA chậm cho MACD
macd_signal : int
Period Signal line cho MACD
"""
self.macd_fast = macd_fast
self.macd_slow = macd_slow
self.macd_signal = macd_signal
def calculate_macd(self, prices):
"""Tính MACD"""
macd = ta.macd(prices, fast=self.macd_fast,
slow=self.macd_slow, signal=self.macd_signal)
if macd is None:
return None
return pd.DataFrame({
'MACD': macd.iloc[:, 0],
'Signal': macd.iloc[:, 1],
'Histogram': macd.iloc[:, 2]
})
def analyze_h1_macd(self, df_h1):
"""Phân tích MACD trên H1"""
macd_data = self.calculate_macd(df_h1['Close'])
if macd_data is None:
return 'neutral'
current_macd = macd_data['MACD'].iloc[-1]
current_signal = macd_data['Signal'].iloc[-1]
current_histogram = macd_data['Histogram'].iloc[-1]
# Bullish: MACD trên Signal và histogram dương
if current_macd > current_signal and current_histogram > 0:
return 'bullish'
# Bearish: MACD dưới Signal và histogram âm
elif current_macd < current_signal and current_histogram < 0:
return 'bearish'
else:
return 'neutral'
def find_m15_macd_entry(self, df_m15, h1_bias):
"""Tìm entry dựa trên MACD M15"""
macd_data = self.calculate_macd(df_m15['Close'])
if macd_data is None:
return 0
current_macd = macd_data['MACD'].iloc[-1]
current_signal = macd_data['Signal'].iloc[-1]
prev_macd = macd_data['MACD'].iloc[-2]
prev_signal = macd_data['Signal'].iloc[-2]
# Chỉ trade theo hướng H1
if h1_bias == 'bullish':
# MACD cắt lên Signal
if (current_macd > current_signal and
prev_macd <= prev_signal):
return 1 # Tín hiệu mua
elif h1_bias == 'bearish':
# MACD cắt xuống Signal
if (current_macd < current_signal and
prev_macd >= prev_signal):
return -1 # Tín hiệu bán
return 0
def generate_signals(self, exchange, symbol):
"""Tạo tín hiệu giao dịch"""
data = get_multiple_timeframes(exchange, symbol, ['15m', '1h'])
df_h1 = data['1h']
df_m15 = data['15m']
# Phân tích MACD trên H1
h1_bias = self.analyze_h1_macd(df_h1)
# Tìm entry trên M15
m15_signal = self.find_m15_macd_entry(df_m15, h1_bias)
return {
'h1_bias': h1_bias,
'm15_signal': m15_signal,
'h1_price': df_h1['Close'].iloc[-1],
'm15_price': df_m15['Close'].iloc[-1],
'timestamp': datetime.now()
}
2.4. Chiến lược Support/Resistance Multi-Timeframe (Rất hiệu quả)
Đặc điểm:
- Xác định S/R trên H1
- Tìm entry khi giá chạm S/R trên M15
- Kết hợp với các chỉ báo khác
Quy tắc:
- Mua: Giá chạm support trên H1 + Bounce trên M15
- Bán: Giá chạm resistance trên H1 + Rejection trên M15
class SupportResistanceMTFStrategy:
"""Chiến lược Support/Resistance Multi-Timeframe"""
def __init__(self, lookback=50, tolerance=0.001):
"""
Parameters:
-----------
lookback : int
Số nến để xác định S/R
tolerance : float
Tolerance % để xác định chạm S/R
"""
self.lookback = lookback
self.tolerance = tolerance
def identify_support_resistance(self, df):
"""Xác định support và resistance"""
recent_data = df.tail(self.lookback)
# Tìm các đỉnh và đáy
from scipy.signal import find_peaks
highs = recent_data['High'].values
lows = recent_data['Low'].values
# Tìm đỉnh (resistance)
peaks, _ = find_peaks(highs, distance=5)
if len(peaks) > 0:
resistance = recent_data['High'].iloc[peaks].max()
else:
resistance = recent_data['High'].max()
# Tìm đáy (support)
troughs, _ = find_peaks(-lows, distance=5)
if len(troughs) > 0:
support = recent_data['Low'].iloc[troughs].min()
else:
support = recent_data['Low'].min()
return support, resistance
def check_price_near_sr(self, price, support, resistance):
"""Kiểm tra giá có gần S/R không"""
# Kiểm tra gần support
if abs(price - support) / support < self.tolerance:
return 'near_support'
# Kiểm tra gần resistance
elif abs(price - resistance) / resistance < self.tolerance:
return 'near_resistance'
else:
return 'none'
def find_m15_bounce_rejection(self, df_m15, sr_level, sr_type):
"""
Tìm bounce (support) hoặc rejection (resistance) trên M15
Parameters:
-----------
sr_type : str
'near_support' hoặc 'near_resistance'
"""
current_candle = df_m15.iloc[-1]
prev_candle = df_m15.iloc[-2]
if sr_type == 'near_support':
# Bounce: Giá chạm support và đóng cửa trên
if (current_candle['Low'] <= sr_level * (1 + self.tolerance) and
current_candle['Close'] > sr_level and
current_candle['Close'] > prev_candle['Close']):
return 1 # Tín hiệu mua
elif sr_type == 'near_resistance':
# Rejection: Giá chạm resistance và đóng cửa dưới
if (current_candle['High'] >= sr_level * (1 - self.tolerance) and
current_candle['Close'] < sr_level and
current_candle['Close'] < prev_candle['Close']):
return -1 # Tín hiệu bán
return 0
def generate_signals(self, exchange, symbol):
"""Tạo tín hiệu giao dịch"""
data = get_multiple_timeframes(exchange, symbol, ['15m', '1h'])
df_h1 = data['1h']
df_m15 = data['15m']
# Xác định S/R trên H1
support, resistance = self.identify_support_resistance(df_h1)
# Kiểm tra giá M15 có gần S/R không
m15_price = df_m15['Close'].iloc[-1]
sr_position = self.check_price_near_sr(m15_price, support, resistance)
# Tìm bounce/rejection trên M15
if sr_position == 'near_support':
signal = self.find_m15_bounce_rejection(df_m15, support, 'near_support')
elif sr_position == 'near_resistance':
signal = self.find_m15_bounce_rejection(df_m15, resistance, 'near_resistance')
else:
signal = 0
return {
'h1_support': support,
'h1_resistance': resistance,
'm15_signal': signal,
'sr_position': sr_position,
'h1_price': df_h1['Close'].iloc[-1],
'm15_price': m15_price,
'timestamp': datetime.now()
}
3. Bot Auto Trading Multi-Timeframe hoàn chỉnh
3.1. Bot với Quản lý Rủi ro và Multi-Timeframe Analysis
import ccxt
import pandas as pd
import numpy as np
import time
from datetime import datetime
from typing import Dict, Optional
class MultiTimeframeTradingBot:
"""Bot auto trading sử dụng Multi-Timeframe Analysis"""
def __init__(self, exchange_name: str, api_key: str, api_secret: str,
strategy_type: str = 'trend_following'):
"""
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 ('trend_following', 'rsi', 'macd', 'sr')
"""
# 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.h1_trend = None
# Cài đặt rủi ro
self.max_position_size = 0.1 # 10% vốn
self.stop_loss_pct = 0.02 # 2%
self.take_profit_pct = 0.04 # 4%
self.risk_reward_ratio = 2.0
def _init_strategy(self, strategy_type: str):
"""Khởi tạo chiến lược"""
if strategy_type == 'trend_following':
return TrendFollowingMTFStrategy()
elif strategy_type == 'rsi':
return RSIMultiTimeframeStrategy()
elif strategy_type == 'macd':
return MACDMultiTimeframeStrategy()
elif strategy_type == 'sr':
return SupportResistanceMTFStrategy()
else:
raise ValueError(f"Unknown strategy type: {strategy_type}")
def calculate_position_size(self, balance: float, 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(price - stop_loss)
if risk_per_unit == 0:
return 0
position_size = risk_amount / risk_per_unit
return position_size
def calculate_stop_loss_take_profit(self, entry_price: float, side: str):
"""Tính stop loss và take profit"""
if side == 'long':
stop_loss = entry_price * (1 - self.stop_loss_pct)
risk = entry_price - stop_loss
take_profit = entry_price + (risk * self.risk_reward_ratio)
else: # short
stop_loss = entry_price * (1 + self.stop_loss_pct)
risk = stop_loss - entry_price
take_profit = entry_price - (risk * self.risk_reward_ratio)
return stop_loss, take_profit
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
def check_h1_trend_change(self, new_h1_trend):
"""Kiểm tra xem H1 trend có thay đổi không"""
if self.position and self.h1_trend:
# Nếu trend đảo chiều, đóng vị thế
if self.position == 'long' and new_h1_trend == 'downtrend':
print(f"[{datetime.now()}] H1 Trend changed to downtrend, closing long position")
return True
elif self.position == 'short' and new_h1_trend == 'uptrend':
print(f"[{datetime.now()}] H1 Trend changed to uptrend, closing short position")
return True
return False
def open_position(self, symbol: str, side: str, price: float, amount: float, h1_trend: str):
"""Mở vị thế"""
order = self.place_order(symbol, side, amount)
if order:
self.position = side
self.entry_price = price
self.h1_trend = h1_trend
# Đặt stop loss và take profit
self.stop_loss, self.take_profit = self.calculate_stop_loss_take_profit(
price, side
)
print(f"[{datetime.now()}] Position opened: {side} @ {price}")
print(f"H1 Trend: {h1_trend}")
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.h1_trend = None
def run(self, symbol: str, check_interval: int = 300):
"""
Chạy bot
Parameters:
-----------
symbol : str
Trading pair
check_interval : int
Thời gian chờ giữa các lần kiểm tra (giây)
"""
print(f"[{datetime.now()}] Bot started for {symbol}")
print(f"Strategy: {type(self.strategy).__name__}")
while True:
try:
# Lấy giá hiện tại (M15)
data_m15 = get_multiple_timeframes(self.exchange, symbol, ['15m'], limit=1)
current_price = data_m15['15m']['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 từ strategy
signals = self.strategy.generate_signals(self.exchange, symbol)
# Kiểm tra H1 trend change
h1_trend = signals.get('h1_trend') or signals.get('h1_bias')
if h1_trend:
if self.check_h1_trend_change(h1_trend):
self.close_position(current_price)
time.sleep(check_interval)
continue
# Lấy tín hiệu M15
m15_signal = signals.get('m15_signal', 0)
# Xử lý tín hiệu
if m15_signal == 1 and self.position != 'long':
# Tín hiệu mua
if h1_trend in ['uptrend', 'bullish']: # Chỉ mua khi H1 uptrend
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')
amount = self.calculate_position_size(
available_balance, current_price, stop_loss
)
if amount > 0:
self.open_position(symbol, 'long', current_price, amount, h1_trend)
elif m15_signal == -1 and self.position == 'long':
# Tín hiệu bán
self.close_position(current_price)
# Log thông tin
print(f"[{datetime.now()}] H1 Trend: {h1_trend}, M15 Signal: {m15_signal}, Price: {current_price}")
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 Multi-Timeframe
4.1. Hàm Backtest
def backtest_multitimeframe_strategy(df_h1, df_m15, strategy, initial_capital=10000):
"""
Backtest chiến lược Multi-Timeframe
Parameters:
-----------
df_h1 : pd.DataFrame
Dữ liệu H1
df_m15 : pd.DataFrame
Dữ liệu M15
strategy : Strategy object
Đối tượng chiến lược
initial_capital : float
Vốn ban đầu
Returns:
--------
dict: Kết quả backtest
"""
# Đồng bộ dữ liệu M15 với H1
# Mỗi nến H1 = 4 nến M15
capital = initial_capital
position = 0
entry_price = 0
trades = []
stop_loss_pct = 0.02
take_profit_pct = 0.04
# Lặp qua từng nến H1
for h1_idx in range(50, len(df_h1)):
h1_candle = df_h1.iloc[h1_idx]
h1_timestamp = h1_candle.name
# Tìm các nến M15 tương ứng với nến H1 này
m15_start_idx = h1_idx * 4
m15_end_idx = min(m15_start_idx + 4, len(df_m15))
if m15_end_idx <= m15_start_idx:
continue
window_h1 = df_h1.iloc[:h1_idx+1]
window_m15 = df_m15.iloc[:m15_end_idx]
# Phân tích H1 trend
if isinstance(strategy, TrendFollowingMTFStrategy):
h1_trend = strategy.analyze_h1_trend(window_h1)
m15_signal = strategy.find_m15_entry(window_m15, h1_trend)
elif isinstance(strategy, RSIMultiTimeframeStrategy):
h1_bias = strategy.analyze_h1_rsi(window_h1)
m15_signal = strategy.find_m15_rsi_entry(window_m15, h1_bias)
else:
continue
# Xử lý tín hiệu
current_price = window_m15['Close'].iloc[-1]
# Kiểm tra stop loss và take profit
if position > 0:
if current_price <= entry_price * (1 - stop_loss_pct):
capital = position * current_price
pnl = ((current_price - entry_price) / entry_price) * 100
trades[-1]['exit_price'] = current_price
trades[-1]['pnl'] = pnl
trades[-1]['exit_reason'] = 'stop_loss'
position = 0
elif current_price >= entry_price * (1 + take_profit_pct):
capital = position * current_price
pnl = ((current_price - entry_price) / entry_price) * 100
trades[-1]['exit_price'] = current_price
trades[-1]['pnl'] = pnl
trades[-1]['exit_reason'] = 'take_profit'
position = 0
# Xử lý entry signal
if m15_signal == 1 and position == 0:
if h1_trend == 'uptrend' or h1_bias == 'bullish':
position = capital / current_price
entry_price = current_price
trades.append({
'type': 'buy',
'date': h1_timestamp,
'entry_price': current_price,
'h1_trend': h1_trend if 'h1_trend' in locals() else h1_bias,
'capital': capital
})
elif m15_signal == -1 and position > 0:
capital = position * current_price
pnl = ((current_price - entry_price) / entry_price) * 100
trades[-1]['exit_price'] = current_price
trades[-1]['pnl'] = pnl
trades[-1]['exit_reason'] = 'signal'
position = 0
# Đóng vị thế cuối cùng
if position > 0:
final_price = df_m15['Close'].iloc[-1]
capital = position * final_price
if trades:
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 H1 và M15
# Lưu ý: yfinance không hỗ trợ M15 trực tiếp, cần sử dụng API khác hoặc resample
data_h1 = yf.download('BTC-USD', period='6mo', interval='1h')
df_h1 = pd.DataFrame(data_h1)
df_h1.columns = [col.lower() for col in df_h1.columns]
# Resample từ 1h xuống 15m (giả lập)
data_m15 = yf.download('BTC-USD', period='6mo', interval='5m')
df_m15 = pd.DataFrame(data_m15)
df_m15.columns = [col.lower() for col in df_m15.columns]
# Resample 5m thành 15m
df_m15 = df_m15.resample('15T').agg({
'open': 'first',
'high': 'max',
'low': 'min',
'close': 'last',
'volume': 'sum'
}).dropna()
# Chạy backtest
strategy = TrendFollowingMTFStrategy()
results = backtest_multitimeframe_strategy(df_h1, df_m15, 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ố Multi-Timeframe Strategy
5.1. Tìm tham số tối ưu
from itertools import product
def optimize_multitimeframe_parameters(df_h1, df_m15, strategy_class, param_ranges):
"""
Tối ưu hóa tham số Multi-Timeframe 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_multitimeframe_strategy(df_h1, df_m15, 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 = {
'h1_ema_fast': [15, 20, 25],
'h1_ema_slow': [45, 50, 55],
'm15_ema_fast': [15, 20, 25],
'm15_ema_slow': [45, 50, 55]
}
optimization_results = optimize_multitimeframe_parameters(
df_h1, df_m15, TrendFollowingMTFStrategy, 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 Multi-Timeframe
6.1. Dynamic Stop Loss dựa trên H1 volatility
class MultiTimeframeRiskManager:
"""Quản lý rủi ro cho chiến lược Multi-Timeframe"""
def __init__(self, max_risk_per_trade=0.01, atr_period=14):
self.max_risk_per_trade = max_risk_per_trade
self.atr_period = atr_period
def calculate_stop_loss_from_h1(self, df_h1, entry_price, side='long'):
"""
Tính stop loss dựa trên ATR của H1
Stop loss rộng hơn khi H1 volatility cao
"""
atr = calculate_atr(df_h1, self.atr_period)
current_atr = atr.iloc[-1]
if side == 'long':
# Stop loss = Entry - (ATR * 2)
stop_loss = entry_price - (current_atr * 2)
else: # short
stop_loss = entry_price + (current_atr * 2)
return stop_loss
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
7. Kết luận: Chiến lược Multi-Timeframe nào hiệu quả nhất?
Đánh giá các chiến lược:
- Trend Following (M15 + H1)
- ✅ Đơn giản, dễ triển khai
- ✅ Phù hợp với thị trường có xu hướng
- ⭐ Hiệu quả: 4/5
- RSI Multi-Timeframe
- ✅ Tín hiệu rõ ràng, dễ theo dõi
- ✅ Kết hợp oversold/overbought hiệu quả
- ⭐ Hiệu quả: 4.5/5
- MACD Multi-Timeframe
- ✅ Tín hiệu mạnh, độ chính xác cao
- ✅ Kết hợp momentum và trend
- ⭐ Hiệu quả: 4.5/5
- Support/Resistance Multi-Timeframe
- ✅ Tín hiệu đáng tin cậy nhất
- ✅ Phù hợp với range và trend market
- ⭐ Hiệu quả: 5/5
Khuyến nghị:
- Cho người mới bắt đầu: Trend Following Multi-Timeframe
- Cho trader có kinh nghiệm: RSI hoặc MACD Multi-Timeframe
- Cho swing trading: Support/Resistance Multi-Timeframe
Lưu ý quan trọng:
- Luôn xác nhận H1 trend: Chỉ trade theo hướng xu hướng H1
- Entry trên M15: Sử dụng M15 để tìm điểm vào tối ưu
- Quản lý rủi ro: Đặt stop loss dựa trên H1 volatility
- Backtest kỹ lưỡng: Kiểm tra chiến lược trên nhiều thị trường
- Theo dõi trend change: Đóng vị thế khi H1 trend đảo chiều
- Tránh over-trading: Chỉ trade khi cả hai khung thời gian đồng thuận
8. Tài liệu tham khảo
- Multi-Timeframe Analysis – Investopedia
- Technical Analysis of the Financial Markets – John J. Murphy
- Python for Finance – Yves Hilpisch
- CCXT Documentation
Lưu ý: Trading có rủi ro. Multi-Timeframe Analysis giúp tăng độ chính xác nhưng không đảm bảo lợi nhuận. 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ư.