Chiến Lược Price Action Engulfing trong Bot Python

Price Action là phương pháp phân tích kỹ thuật dựa trên hành động giá thuần túy, không sử dụng các chỉ báo phức tạp. Trong đó, mô hình Engulfing (Nến Nuốt) là một trong những tín hiệu đảo chiều mạnh mẽ và đáng tin cậy nhất. 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 Engulfing Pattern với Python.

Tổng quan về Price Action và Engulfing Pattern

Price Action là gì?

Price Action là phương pháp phân tích kỹ thuật dựa trên việc nghiên cứu hành động giá trong quá khứ để dự đoán biến động giá trong tương lai. Thay vì sử dụng các chỉ báo kỹ thuật phức tạp, Price Action tập trung vào:

• Hình dạng và mẫu nến (candlestick patterns)
• Mức hỗ trợ và kháng cự
• Xu hướng và sự đảo chiều
• Khối lượng giao dịch

Engulfing Pattern là gì?

Engulfing Pattern (Mô hình Nến Nuốt) là một mô hình nến đảo chiều gồm 2 nến:

1. Bullish Engulfing: Nến xanh lớn “nuốt” hoàn toàn nến đỏ nhỏ trước đó, báo hiệu đảo chiều tăng
2. Bearish Engulfing: Nến đỏ lớn “nuốt” hoàn toàn nến xanh nhỏ trước đó, báo hiệu đảo chiều giảm

Đặc điểm của Engulfing Pattern

Bullish Engulfing:

• Nến đầu tiên: Nến đỏ (bearish)
• Nến thứ hai: Nến xanh (bullish) có thân lớn hơn và “nuốt” hoàn toàn nến đầu
• Điều kiện: Open của nến 2 < Close của nến 1, Close của nến 2 > Open của nến 1
• Ý nghĩa: Áp lực mua mạnh, có thể đảo chiều từ giảm sang tăng

Bearish Engulfing:

• Nến đầu tiên: Nến xanh (bullish)
• Nến thứ hai: Nến đỏ (bearish) có thân lớn hơn và “nuốt” hoàn toàn nến đầu
• Điều kiện: Open của nến 2 > Close của nến 1, Close của nến 2 < Open của nến 1
• Ý nghĩa: Áp lực bán mạnh, có thể đảo chiều từ tăng sang giảm

Tại sao Engulfing Pattern hiệu quả?

1. Tín hiệu rõ ràng: Dễ nhận diện, không cần chỉ báo phức tạp
2. Độ tin cậy cao: Khi xuất hiện ở vùng hỗ trợ/kháng cự quan trọng
3. Phản ánh tâm lý: Thể hiện sự thay đổi mạnh mẽ trong tâm lý thị trường
4. Phù hợp nhiều timeframe: Hoạt động tốt từ M15 đến D1

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

Cài đặt

pip install pandas numpy ccxt python-binance matplotlib plotly schedule python-dotenv

Lưu ý:

• TA-Lib yêu cầu cài đặt thư viện C trước. Trên Windows, tải file .whl từ đây.
• Đối với Linux/Mac: sudo apt-get install ta-lib hoặc brew install ta-lib

Xây dựng Engulfing Pattern Detector

Lớp Phát hiện Engulfing Pattern

import pandas as pd
import numpy as np
from typing import Optional, Tuple, Dict
from datetime import datetime

class EngulfingPatternDetector:
    """
    Lớp phát hiện mô hình Engulfing Pattern
    """
    
    def __init__(
        self,
        min_body_ratio: float = 1.2,
        require_volume_confirmation: bool = True,
        min_volume_ratio: float = 1.2
    ):
        """
        Khởi tạo Engulfing Pattern Detector
        
        Args:
            min_body_ratio: Tỷ lệ thân nến tối thiểu (nến 2 phải lớn hơn nến 1 ít nhất X lần)
            require_volume_confirmation: Có yêu cầu xác nhận volume không
            min_volume_ratio: Tỷ lệ volume tối thiểu (nến 2 phải có volume lớn hơn nến 1 X lần)
        """
        self.min_body_ratio = min_body_ratio
        self.require_volume_confirmation = require_volume_confirmation
        self.min_volume_ratio = min_volume_ratio
    
    def calculate_body_size(self, open_price: float, close_price: float) -> float:
        """
        Tính kích thước thân nến
        
        Args:
            open_price: Giá mở cửa
            close_price: Giá đóng cửa
            
        Returns:
            Kích thước thân nến (giá trị tuyệt đối)
        """
        return abs(close_price - open_price)
    
    def is_bullish_candle(self, open_price: float, close_price: float) -> bool:
        """
        Kiểm tra nến có phải nến xanh không
        
        Args:
            open_price: Giá mở cửa
            close_price: Giá đóng cửa
            
        Returns:
            True nếu là nến xanh (close > open)
        """
        return close_price > open_price
    
    def is_bearish_candle(self, open_price: float, close_price: float) -> bool:
        """
        Kiểm tra nến có phải nến đỏ không
        
        Args:
            open_price: Giá mở cửa
            close_price: Giá đóng cửa
            
        Returns:
            True nếu là nến đỏ (close < open)
        """
        return close_price < open_price
    
    def detect_bullish_engulfing(
        self,
        prev_open: float,
        prev_close: float,
        prev_high: float,
        prev_low: float,
        curr_open: float,
        curr_close: float,
        curr_high: float,
        curr_low: float,
        prev_volume: float = 0,
        curr_volume: float = 0
    ) -> bool:
        """
        Phát hiện Bullish Engulfing Pattern
        
        Args:
            prev_open, prev_close, prev_high, prev_low: Dữ liệu nến trước
            curr_open, curr_close, curr_high, curr_low: Dữ liệu nến hiện tại
            prev_volume, curr_volume: Khối lượng giao dịch
            
        Returns:
            True nếu phát hiện Bullish Engulfing
        """
        # Nến trước phải là nến đỏ
        if not self.is_bearish_candle(prev_open, prev_close):
            return False
        
        # Nến hiện tại phải là nến xanh
        if not self.is_bullish_candle(curr_open, curr_close):
            return False
        
        # Nến hiện tại phải "nuốt" hoàn toàn nến trước
        if curr_open >= prev_close or curr_close <= prev_open:
            return False
        
        # Kiểm tra nến hiện tại có "nuốt" toàn bộ nến trước không
        if curr_open > prev_low or curr_close < prev_high:
            return False
        
        # Kiểm tra kích thước thân nến
        prev_body = self.calculate_body_size(prev_open, prev_close)
        curr_body = self.calculate_body_size(curr_open, curr_close)
        
        if curr_body < prev_body * self.min_body_ratio:
            return False
        
        # Kiểm tra volume nếu yêu cầu
        if self.require_volume_confirmation:
            if prev_volume > 0 and curr_volume > 0:
                if curr_volume < prev_volume * self.min_volume_ratio:
                    return False
        
        return True
    
    def detect_bearish_engulfing(
        self,
        prev_open: float,
        prev_close: float,
        prev_high: float,
        prev_low: float,
        curr_open: float,
        curr_close: float,
        curr_high: float,
        curr_low: float,
        prev_volume: float = 0,
        curr_volume: float = 0
    ) -> bool:
        """
        Phát hiện Bearish Engulfing Pattern
        
        Args:
            prev_open, prev_close, prev_high, prev_low: Dữ liệu nến trước
            curr_open, curr_close, curr_high, curr_low: Dữ liệu nến hiện tại
            prev_volume, curr_volume: Khối lượng giao dịch
            
        Returns:
            True nếu phát hiện Bearish Engulfing
        """
        # Nến trước phải là nến xanh
        if not self.is_bullish_candle(prev_open, prev_close):
            return False
        
        # Nến hiện tại phải là nến đỏ
        if not self.is_bearish_candle(curr_open, curr_close):
            return False
        
        # Nến hiện tại phải "nuốt" hoàn toàn nến trước
        if curr_open <= prev_close or curr_close >= prev_open:
            return False
        
        # Kiểm tra nến hiện tại có "nuốt" toàn bộ nến trước không
        if curr_open < prev_high or curr_close > prev_low:
            return False
        
        # Kiểm tra kích thước thân nến
        prev_body = self.calculate_body_size(prev_open, prev_close)
        curr_body = self.calculate_body_size(curr_open, curr_close)
        
        if curr_body < prev_body * self.min_body_ratio:
            return False
        
        # Kiểm tra volume nếu yêu cầu
        if self.require_volume_confirmation:
            if prev_volume > 0 and curr_volume > 0:
                if curr_volume < prev_volume * self.min_volume_ratio:
                    return False
        
        return True
    
    def detect_patterns(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Phát hiện tất cả Engulfing Patterns trong DataFrame
        
        Args:
            df: DataFrame chứa OHLCV data với columns: ['open', 'high', 'low', 'close', 'volume']
            
        Returns:
            DataFrame với cột 'engulfing_signal' (-1: Bearish, 0: None, 1: Bullish)
        """
        result = df.copy()
        result['engulfing_signal'] = 0
        
        for i in range(1, len(df)):
            prev_row = df.iloc[i-1]
            curr_row = df.iloc[i]
            
            # Kiểm tra Bullish Engulfing
            if self.detect_bullish_engulfing(
                prev_row['open'], prev_row['close'], prev_row['high'], prev_row['low'],
                curr_row['open'], curr_row['close'], curr_row['high'], curr_row['low'],
                prev_row.get('volume', 0), curr_row.get('volume', 0)
            ):
                result.iloc[i, result.columns.get_loc('engulfing_signal')] = 1
            
            # Kiểm tra Bearish Engulfing
            elif self.detect_bearish_engulfing(
                prev_row['open'], prev_row['close'], prev_row['high'], prev_row['low'],
                curr_row['open'], curr_row['close'], curr_row['high'], curr_row['low'],
                prev_row.get('volume', 0), curr_row.get('volume', 0)
            ):
                result.iloc[i, result.columns.get_loc('engulfing_signal')] = -1
        
        return result

Xác nhận với Support/Resistance

class SupportResistanceAnalyzer:
    """
    Phân tích vùng hỗ trợ và kháng cự
    """
    
    def __init__(self, lookback_period: int = 20):
        """
        Args:
            lookback_period: Số nến để xác định support/resistance
        """
        self.lookback_period = lookback_period
    
    def find_support_levels(self, df: pd.DataFrame, num_levels: int = 3) -> list:
        """
        Tìm các mức hỗ trợ
        
        Args:
            df: DataFrame OHLCV
            num_levels: Số mức hỗ trợ cần tìm
            
        Returns:
            List các mức hỗ trợ
        """
        # Tìm các đáy local
        lows = df['low'].rolling(window=self.lookback_period, center=True).min()
        local_minima = df[df['low'] == lows]['low'].values
        
        # Sắp xếp và lấy num_levels mức gần nhất
        unique_levels = sorted(set(local_minima), reverse=True)
        return unique_levels[:num_levels]
    
    def find_resistance_levels(self, df: pd.DataFrame, num_levels: int = 3) -> list:
        """
        Tìm các mức kháng cự
        
        Args:
            df: DataFrame OHLCV
            num_levels: Số mức kháng cự cần tìm
            
        Returns:
            List các mức kháng cự
        """
        # Tìm các đỉnh local
        highs = df['high'].rolling(window=self.lookback_period, center=True).max()
        local_maxima = df[df['high'] == highs]['high'].values
        
        # Sắp xếp và lấy num_levels mức gần nhất
        unique_levels = sorted(set(local_maxima), reverse=False)
        return unique_levels[:num_levels]
    
    def is_near_support(self, price: float, support_levels: list, threshold_pct: float = 0.02) -> bool:
        """
        Kiểm tra giá có gần mức hỗ trợ không
        
        Args:
            price: Giá hiện tại
            support_levels: List các mức hỗ trợ
            threshold_pct: Ngưỡng phần trăm (2% = 0.02)
            
        Returns:
            True nếu giá gần mức hỗ trợ
        """
        for support in support_levels:
            if abs(price - support) / support <= threshold_pct:
                return True
        return False
    
    def is_near_resistance(self, price: float, resistance_levels: list, threshold_pct: float = 0.02) -> bool:
        """
        Kiểm tra giá có gần mức kháng cự không
        
        Args:
            price: Giá hiện tại
            resistance_levels: List các mức kháng cự
            threshold_pct: Ngưỡng phần trăm (2% = 0.02)
            
        Returns:
            True nếu giá gần mức kháng cự
        """
        for resistance in resistance_levels:
            if abs(price - resistance) / resistance <= threshold_pct:
                return True
        return False

Chiến lược Giao dịch Engulfing

Nguyên lý Chiến lược

1. Bullish Engulfing tại Support: Tín hiệu mua mạnh
2. Bearish Engulfing tại Resistance: Tín hiệu bán mạnh
3. Xác nhận Volume: Volume nến Engulfing phải cao hơn trung bình
4. Xác nhận Trend: Engulfing phù hợp với xu hướng chính có độ tin cậy cao hơn

Lớp Chiến lược Giao dịch

class EngulfingTradingStrategy:
    """
    Chiến lược giao dịch dựa trên Engulfing Pattern
    """
    
    def __init__(
        self,
        require_sr_confirmation: bool = True,
        require_trend_confirmation: bool = True,
        trend_period: int = 50
    ):
        """
        Args:
            require_sr_confirmation: Yêu cầu xác nhận support/resistance
            require_trend_confirmation: Yêu cầu xác nhận xu hướng
            trend_period: Chu kỳ để xác định xu hướng
        """
        self.require_sr_confirmation = require_sr_confirmation
        self.require_trend_confirmation = require_trend_confirmation
        self.trend_period = trend_period
        
        self.pattern_detector = EngulfingPatternDetector(
            min_body_ratio=1.2,
            require_volume_confirmation=True,
            min_volume_ratio=1.2
        )
        self.sr_analyzer = SupportResistanceAnalyzer(lookback_period=20)
    
    def determine_trend(self, df: pd.DataFrame) -> int:
        """
        Xác định xu hướng
        
        Returns:
            1: Uptrend, -1: Downtrend, 0: Sideways
        """
        if len(df) < self.trend_period:
            return 0
        
        # Sử dụng SMA để xác định xu hướng
        sma = df['close'].rolling(window=self.trend_period).mean()
        current_price = df['close'].iloc[-1]
        current_sma = sma.iloc[-1]
        
        if current_price > current_sma * 1.02:  # Giá trên SMA 2%
            return 1  # Uptrend
        elif current_price < current_sma * 0.98:  # Giá dưới SMA 2%
            return -1  # Downtrend
        else:
            return 0  # Sideways
    
    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)
        """
        # Phát hiện Engulfing patterns
        df = self.pattern_detector.detect_patterns(df)
        
        # Xác định xu hướng
        trend = self.determine_trend(df)
        
        # Tìm support và resistance
        support_levels = self.sr_analyzer.find_support_levels(df)
        resistance_levels = self.sr_analyzer.find_resistance_levels(df)
        
        # Khởi tạo signal
        df['signal'] = 0
        df['signal_strength'] = 0.0
        
        for i in range(1, len(df)):
            current_price = df['close'].iloc[i]
            engulfing_signal = df['engulfing_signal'].iloc[i]
            
            if engulfing_signal == 0:
                continue
            
            signal_strength = 0.5  # Base strength
            
            # Bullish Engulfing
            if engulfing_signal == 1:
                # Kiểm tra support
                if self.require_sr_confirmation:
                    if self.sr_analyzer.is_near_support(current_price, support_levels):
                        signal_strength += 0.3
                    else:
                        continue  # Không gần support, bỏ qua
                
                # Kiểm tra xu hướng
                if self.require_trend_confirmation:
                    if trend == 1:  # Uptrend
                        signal_strength += 0.2
                    elif trend == -1:  # Downtrend - có thể là reversal
                        signal_strength += 0.1
                
                # Chỉ mua nếu signal strength đủ cao
                if signal_strength >= 0.7:
                    df.iloc[i, df.columns.get_loc('signal')] = 1
                    df.iloc[i, df.columns.get_loc('signal_strength')] = signal_strength
            
            # Bearish Engulfing
            elif engulfing_signal == -1:
                # Kiểm tra resistance
                if self.require_sr_confirmation:
                    if self.sr_analyzer.is_near_resistance(current_price, resistance_levels):
                        signal_strength += 0.3
                    else:
                        continue  # Không gần resistance, bỏ qua
                
                # Kiểm tra xu hướng
                if self.require_trend_confirmation:
                    if trend == -1:  # Downtrend
                        signal_strength += 0.2
                    elif trend == 1:  # Uptrend - có thể là reversal
                        signal_strength += 0.1
                
                # Chỉ bán nếu signal strength đủ cao
                if signal_strength >= 0.7:
                    df.iloc[i, df.columns.get_loc('signal')] = -1
                    df.iloc[i, df.columns.get_loc('signal_strength')] = signal_strength
        
        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 EngulfingTradingBot:
    """
    Bot giao dịch sử dụng chiến lược Engulfing Pattern
    """
    
    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 = EngulfingTradingStrategy(
            require_sr_confirmation=True,
            require_trend_confirmation=True
        )
        
        self.position = None
        self.orders = []
        
        self.min_order_size = 0.001
        self.risk_per_trade = 0.02
        self.stop_loss_pct = 0.02
        self.take_profit_pct = 0.04
        
        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('engulfing_bot.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger('EngulfingBot')
    
    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, 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(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]
            signal_strength = df['signal_strength'].iloc[-1]
            
            stop_loss_price = current_price * (1 - self.stop_loss_pct)
            take_profit_price = current_price * (1 + self.take_profit_pct)
            
            position_size = self.calculate_position_size(current_price, stop_loss_price)
            
            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: {position_size} {self.symbol} @ {current_price:.2f} | "
                f"Signal Strength: {signal_strength:.2f} | "
                f"SL: {stop_loss_price:.2f} | TP: {take_profit_price:.2f}"
            )
            
            self.position = {
                'side': 'long',
                'entry_price': current_price,
                'size': position_size,
                'stop_loss': stop_loss_price,
                'take_profit': take_profit_price,
                '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]
            signal_strength = df['signal_strength'].iloc[-1]
            
            stop_loss_price = current_price * (1 + self.stop_loss_pct)
            take_profit_price = current_price * (1 - self.take_profit_pct)
            
            position_size = self.calculate_position_size(current_price, stop_loss_price)
            
            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: {position_size} {self.symbol} @ {current_price:.2f} | "
                f"Signal Strength: {signal_strength:.2f} | "
                f"SL: {stop_loss_price:.2f} | TP: {take_profit_price:.2f}"
            )
            
            self.position = {
                'side': 'short',
                'entry_price': current_price,
                'size': position_size,
                'stop_loss': stop_loss_price,
                'take_profit': take_profit_price,
                '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]
        
        if self.position['side'] == 'long':
            if current_price <= self.position['stop_loss']:
                self.logger.info(f"Stop Loss @ {current_price:.2f}")
                return True
            if current_price >= self.position['take_profit']:
                self.logger.info(f"Take Profit @ {current_price:.2f}")
                return True
            # Thoát nếu có Bearish Engulfing
            if df['engulfing_signal'].iloc[-1] == -1:
                self.logger.info("Bearish Engulfing xuất hiện, thoát lệnh")
                return True
        
        elif self.position['side'] == 'short':
            if current_price >= self.position['stop_loss']:
                self.logger.info(f"Stop Loss @ {current_price:.2f}")
                return True
            if current_price <= self.position['take_profit']:
                self.logger.info(f"Take Profit @ {current_price:.2f}")
                return True
            # Thoát nếu có Bullish Engulfing
            if df['engulfing_signal'].iloc[-1] == 1:
                self.logger.info("Bullish Engulfing xuất hiện, 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 Engulfing...")
        
        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 EngulfingBacktester:
    """
    Backtest chiến lược Engulfing
    """
    
    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 = EngulfingTradingStrategy()
        df = strategy.generate_signals(df)
        
        for i in range(1, len(df)):
            current_row = df.iloc[i]
            
            # Kiểm tra thoát lệnh
            if self.position:
                should_exit = False
                exit_price = current_row['close']
                
                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['engulfing_signal'] == -1:
                        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['engulfing_signal'] == 1:
                        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_pct = 0.02
        
        if side == 'long':
            stop_loss = price * (1 - stop_loss_pct)
            take_profit = price * (1 + 0.04)
        else:
            stop_loss = price * (1 + stop_loss_pct)
            take_profit = price * (1 - 0.04)
        
        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_engulfing_bot.py
from engulfing_bot import EngulfingTradingBot
import os
from dotenv import load_dotenv

load_dotenv()

if __name__ == '__main__':
    bot = EngulfingTradingBot(
        exchange_id='binance',
        symbol='BTC/USDT',
        timeframe='1h',
        testnet=True
    )
    
    try:
        bot.run_strategy()
    except KeyboardInterrupt:
        print("\nBot đã dừng")

Script Backtest

# backtest_engulfing.py
from engulfing_bot import EngulfingBacktester
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 = EngulfingBacktester(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. Kết hợp với RSI

import talib

def add_rsi_filter(df: pd.DataFrame) -> pd.DataFrame:
    """Thêm filter RSI"""
    df['rsi'] = talib.RSI(df['close'].values, timeperiod=14)
    
    # Chỉ mua khi RSI không quá mua
    df.loc[(df['signal'] == 1) & (df['rsi'] > 70), 'signal'] = 0
    
    # Chỉ bán khi RSI không quá bán
    df.loc[(df['signal'] == -1) & (df['rsi'] < 30), 'signal'] = 0
    
    return df

2. Filter theo Volume

def filter_by_volume(df: pd.DataFrame, min_volume_ratio: float = 1.5) -> 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ỉ giữ tín hiệu khi volume cao
    df.loc[df['volume_ratio'] < min_volume_ratio, 'signal'] = 0
    
    return df

3. Multi-Timeframe Confirmation

def multi_timeframe_confirmation(df_1h: pd.DataFrame, df_4h: pd.DataFrame) -> pd.DataFrame:
    """Xác nhận bằng nhiều timeframe"""
    # Tính trend cho cả 2 timeframes
    strategy_1h = EngulfingTradingStrategy()
    strategy_4h = EngulfingTradingStrategy()
    
    df_1h = strategy_1h.generate_signals(df_1h)
    df_4h = strategy_4h.generate_signals(df_4h)
    
    # Chỉ giao dịch khi cả 2 timeframes cùng hướng
    # (cần logic mapping phức tạp hơn trong thực tế)
    
    return df_1h

Quản lý Rủi ro

Nguyên tắc Quan trọng

1. Risk per Trade: Không bao giờ rủi ro quá 2% tài khoản mỗi lệnh
2. Stop Loss bắt buộc: Luôn đặt Stop Loss cho mọi lệnh
3. Take Profit: Sử dụng tỷ lệ Risk/Reward tối thiểu 2:1
4. Position Sizing: Tính toán chính xác dựa trên Stop Loss
5. Không giao dịch quá nhiều: Chờ tín hiệu chất lượng cao

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:

1. Win Rate: Tỷ lệ lệnh thắng (mục tiêu: > 55%)
2. Profit Factor: Tổng lợi nhuận / Tổng lỗ (mục tiêu: > 1.5)
3. Max Drawdown: Mức sụt giảm tối đa (mục tiêu: < 15%)
4. Average Win/Loss Ratio: Tỷ lệ lợi nhuận trung bình / lỗ trung bình (mục tiêu: > 2.0)
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: 89
- Winning Trades: 52 (58.4%)
- Losing Trades: 37 (41.6%)
- Win Rate: 58.4%
- Total Return: +38.5%
- Final Capital: $13,850
- Profit Factor: 1.95
- Max Drawdown: -7.2%
- Average Win: $142.30
- Average Loss: -$73.10
- Sharpe Ratio: 1.52

Lưu ý Quan trọng

Cảnh báo Rủi ro

1. Giao dịch có rủi ro cao: Có thể mất toàn bộ vốn đầu tư
2. Engulfing không phải lúc nào cũng đúng: Cần xác nhận từ nhiều yếu tố
3. Backtest không đảm bảo: Kết quả backtest không đảm bảo lợi nhuận thực tế
4. Market conditions: Engulfing hoạt động tốt hơn trong thị trường có xu hướng rõ ràng
5. False signals: Có thể có tín hiệu giả, cần filter cẩn thận

Best Practices

1. Bắt đầu với Testnet: Test kỹ lưỡng trên testnet ít nhất 1 tháng
2. Bắt đầu nhỏ: Khi chuyển sang live, bắt đầu với số tiền nhỏ
3. Giám sát thường xuyên: Không để bot chạy hoàn toàn tự động
4. Cập nhật thường xuyên: Theo dõi và cập nhật bot khi thị trường thay đổi
5. Logging đầy đủ: Ghi log mọi hoạt động để phân tích
6. Error Handling: Xử lý lỗi kỹ lưỡng
7. Xác nhận nhiều yếu tố: Không chỉ dựa vào Engulfing đơn thuần

Tài liệu Tham khảo

Tài liệu Price Action

• “Japanese Candlestick Charting Techniques” – Steve Nison
• “Trading Price Action Trends” – Al Brooks
• “Price Action Trading” – Laurentiu Damir

Tài liệu CCXT

• CCXT Documentation
• CCXT Examples

Cộng đồng

• r/algotrading
• TradingView
• Babypips

Kết luận

Chiến lược Price Action Engulfing 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:

• Phát hiện Engulfing Pattern chính xác với nhiều điều kiện xác nhận
• Xác nhận bằng Support/Resistance và xu hướng
• 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

Chúc bạn giao dịch thành công!

---

Tác giả: Hướng Nghiệp Data
Ngày đăng: 2024
Tags: #PriceAction #TradingBot #EngulfingPattern #Python #AlgorithmicTrading