| Chiến Lược VWAP Trading cho Crypto Bot Auto Trading Python

Được viết bởi thanhdt vào ngày 16/11/2025 lúc 21:52 | 9 lượt xem

Chiến Lược VWAP Trading cho Crypto Bot Auto Trading Python

VWAP (Volume Weighted Average Price) là một trong những chỉ báo kỹ thuật quan trọng nhất trong giao dịch cryptocurrency. Khác với các chỉ báo giá thuần túy, VWAP kết hợp cả giá và khối lượng giao dịch, cung cấp cái nhìn sâu sắc về giá trị “thực” của tài sản. Trong bài viết này, chúng ta sẽ xây dựng một bot giao dịch crypto tự động sử dụng chiến lược VWAP với Python.

Tổng quan về VWAP

VWAP là gì?

VWAP (Volume Weighted Average Price) là giá trung bình có trọng số theo khối lượng, được tính bằng cách chia tổng giá trị giao dịch (giá × khối lượng) cho tổng khối lượng giao dịch trong một khoảng thời gian nhất định.

Tại sao VWAP quan trọng trong Crypto Trading?

Phản ánh giá trị thực: VWAP cho biết giá trung bình mà các nhà giao dịch lớn (whales) đã mua/bán
Vùng hỗ trợ/kháng cự động: VWAP thường đóng vai trò là vùng hỗ trợ trong uptrend và kháng cự trong downtrend
Xác định xu hướng: Giá trên VWAP thường cho thấy xu hướng tăng, giá dưới VWAP cho thấy xu hướng giảm
Phân tích khối lượng: Kết hợp với volume, VWAP giúp xác định sức mạnh của xu hướng

Công thức tính VWAP

# VWAP được tính theo công thức:
VWAP = Σ(Price × Volume) / Σ(Volume)

# Trong đó:
# - Price: Giá trung bình của nến (High + Low + Close) / 3
# - Volume: Khối lượng giao dịch
# - Σ: Tổng từ đầu ngày đến thời điểm hiện tại

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 Chỉ báo VWAP

Tính toán VWAP từ đầu

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

class VWAPIndicator:
    """
    Lớp tính toán chỉ báo VWAP
    """
    
    def __init__(self):
        """
        Khởi tạo VWAP Indicator
        """
        self.cumulative_volume = 0
        self.cumulative_price_volume = 0
        self.vwap_values = []
        self.reset_daily = True
    
    def calculate_typical_price(self, high: float, low: float, close: float) -> float:
        """
        Tính Typical Price (giá đại diện của nến)
        
        Args:
            high: Giá cao nhất
            low: Giá thấp nhất
            close: Giá đóng cửa
            
        Returns:
            Typical Price
        """
        return (high + low + close) / 3.0
    
    def calculate_vwap(self, df: pd.DataFrame, reset_period: str = '1D') -> pd.Series:
        """
        Tính toán VWAP cho DataFrame
        
        Args:
            df: DataFrame chứa OHLCV data với columns: ['timestamp', 'open', 'high', 'low', 'close', 'volume']
            reset_period: Chu kỳ reset VWAP ('1D' = hàng ngày, '1W' = hàng tuần, None = không reset)
            
        Returns:
            Series chứa giá trị VWAP
        """
        # Tạo bản sao để không ảnh hưởng DataFrame gốc
        data = df.copy()
        
        # Tính Typical Price
        data['typical_price'] = (
            data['high'] + data['low'] + data['close']
        ) / 3.0
        
        # Tính Price × Volume
        data['price_volume'] = data['typical_price'] * data['volume']
        
        # Reset VWAP theo chu kỳ
        if reset_period:
            # Chuyển timestamp thành datetime nếu chưa
            if not pd.api.types.is_datetime64_any_dtype(data.index):
                data.index = pd.to_datetime(data['timestamp'], unit='ms')
            
            # Tạo group key để reset
            if reset_period == '1D':
                data['reset_key'] = data.index.date
            elif reset_period == '1W':
                data['reset_key'] = data.index.to_period('W')
            else:
                data['reset_key'] = 0  # Không reset
            
            # Tính cumulative sum theo group
            data['cumulative_volume'] = data.groupby('reset_key')['volume'].cumsum()
            data['cumulative_price_volume'] = data.groupby('reset_key')['price_volume'].cumsum()
        else:
            # Không reset, tính cumulative từ đầu
            data['cumulative_volume'] = data['volume'].cumsum()
            data['cumulative_price_volume'] = data['price_volume'].cumsum()
        
        # Tính VWAP
        data['vwap'] = data['cumulative_price_volume'] / data['cumulative_volume']
        
        return data['vwap']
    
    def calculate_vwap_bands(self, vwap: pd.Series, std_multiplier: float = 2.0) -> Tuple[pd.Series, pd.Series]:
        """
        Tính VWAP Bands (Upper và Lower bands dựa trên độ lệch chuẩn)
        
        Args:
            vwap: Series chứa giá trị VWAP
            std_multiplier: Hệ số nhân độ lệch chuẩn (mặc định: 2.0)
            
        Returns:
            Tuple (upper_band, lower_band)
        """
        # Tính độ lệch chuẩn của giá so với VWAP
        # Cần có giá close để tính
        # Giả sử vwap có cùng index với price data
        
        # Tính rolling standard deviation
        rolling_std = vwap.rolling(window=20).std()
        
        upper_band = vwap + (rolling_std * std_multiplier)
        lower_band = vwap - (rolling_std * std_multiplier)
        
        return upper_band, lower_band

VWAP với Multiple Timeframes

class MultiTimeframeVWAP:
    """
    Tính VWAP cho nhiều khung thời gian khác nhau
    """
    
    def __init__(self):
        self.vwap_calculator = VWAPIndicator()
    
    def calculate_multi_vwap(self, df: pd.DataFrame, timeframes: list = ['1h', '4h', '1d']) -> pd.DataFrame:
        """
        Tính VWAP cho nhiều khung thời gian
        
        Args:
            df: DataFrame OHLCV gốc
            timeframes: Danh sách khung thời gian cần tính
            
        Returns:
            DataFrame với các cột VWAP cho từng timeframe
        """
        result = df.copy()
        
        for tf in timeframes:
            # Resample data theo timeframe
            resampled = self._resample_ohlcv(df, tf)
            
            # Tính VWAP cho timeframe này
            vwap = self.vwap_calculator.calculate_vwap(resampled, reset_period='1D')
            
            # Map lại về timeframe gốc
            result[f'vwap_{tf}'] = self._map_to_original_timeframe(df, vwap, tf)
        
        return result
    
    def _resample_ohlcv(self, df: pd.DataFrame, timeframe: str) -> pd.DataFrame:
        """
        Resample OHLCV data theo timeframe mới
        """
        if not pd.api.types.is_datetime64_any_dtype(df.index):
            df.index = pd.to_datetime(df['timestamp'], unit='ms')
        
        resampled = pd.DataFrame()
        resampled['open'] = df['open'].resample(timeframe).first()
        resampled['high'] = df['high'].resample(timeframe).max()
        resampled['low'] = df['low'].resample(timeframe).min()
        resampled['close'] = df['close'].resample(timeframe).last()
        resampled['volume'] = df['volume'].resample(timeframe).sum()
        resampled['timestamp'] = resampled.index.astype(np.int64) // 10**6
        
        return resampled
    
    def _map_to_original_timeframe(self, original_df: pd.DataFrame, vwap_series: pd.Series, timeframe: str) -> pd.Series:
        """
        Map VWAP từ timeframe cao hơn về timeframe gốc
        """
        if not pd.api.types.is_datetime64_any_dtype(original_df.index):
            original_df.index = pd.to_datetime(original_df['timestamp'], unit='ms')
        
        # Forward fill VWAP values
        mapped = original_df.index.map(lambda x: vwap_series.asof(x))
        
        return pd.Series(mapped, index=original_df.index)

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

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

Giá trên VWAP + Volume tăng: Tín hiệu mua (uptrend mạnh)
Giá dưới VWAP + Volume tăng: Tín hiệu bán (downtrend mạnh)
Giá pullback về VWAP: Cơ hội vào lệnh theo xu hướng
Giá vượt VWAP Bands: Tín hiệu quá mua/quá bán

Các Tín hiệu Giao dịch

class VWAPTradingSignals:
    """
    Phát hiện tín hiệu giao dịch dựa trên VWAP
    """
    
    def __init__(self, vwap_period: int = 20, volume_threshold: float = 1.5):
        """
        Args:
            vwap_period: Chu kỳ tính VWAP
            volume_threshold: Ngưỡng volume (1.5 = 150% volume trung bình)
        """
        self.vwap_period = vwap_period
        self.volume_threshold = volume_threshold
    
    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)
        """
        signals = df.copy()
        vwap_calc = VWAPIndicator()
        
        # Tính VWAP
        signals['vwap'] = vwap_calc.calculate_vwap(signals, reset_period='1D')
        
        # Tính volume trung bình
        signals['avg_volume'] = signals['volume'].rolling(window=20).mean()
        
        # Tính độ lệch giá so với VWAP (%)
        signals['price_vwap_diff'] = ((signals['close'] - signals['vwap']) / signals['vwap']) * 100
        
        # Tính volume ratio
        signals['volume_ratio'] = signals['volume'] / signals['avg_volume']
        
        # Khởi tạo signal
        signals['signal'] = 0
        
        # Tín hiệu BUY
        buy_condition = (
            (signals['close'] > signals['vwap']) &  # Giá trên VWAP
            (signals['volume_ratio'] > self.volume_threshold) &  # Volume tăng
            (signals['price_vwap_diff'] < 2.0) &  # Chưa quá xa VWAP (< 2%)
            (signals['close'] > signals['open'])  # Nến xanh
        )
        signals.loc[buy_condition, 'signal'] = 1
        
        # Tín hiệu SELL
        sell_condition = (
            (signals['close'] < signals['vwap']) &  # Giá dưới VWAP
            (signals['volume_ratio'] > self.volume_threshold) &  # Volume tăng
            (signals['price_vwap_diff'] > -2.0) &  # Chưa quá xa VWAP (> -2%)
            (signals['close'] < signals['open'])  # Nến đỏ
        )
        signals.loc[sell_condition, 'signal'] = -1
        
        return signals
    
    def detect_pullback(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Phát hiện pullback về VWAP (cơ hội vào lệnh tốt)
        """
        signals = df.copy()
        vwap_calc = VWAPIndicator()
        
        signals['vwap'] = vwap_calc.calculate_vwap(signals, reset_period='1D')
        
        # Xác định xu hướng
        signals['trend'] = 0
        signals.loc[signals['close'] > signals['vwap'], 'trend'] = 1  # Uptrend
        signals.loc[signals['close'] < signals['vwap'], 'trend'] = -1  # Downtrend
        
        # Phát hiện pullback trong uptrend
        uptrend_pullback = (
            (signals['trend'] == 1) &  # Đang uptrend
            (signals['low'] <= signals['vwap']) &  # Giá chạm VWAP
            (signals['close'] > signals['vwap']) &  # Đóng cửa trên VWAP
            (signals['close'] > signals['open'])  # Nến xanh
        )
        
        # Phát hiện pullback trong downtrend
        downtrend_pullback = (
            (signals['trend'] == -1) &  # Đang downtrend
            (signals['high'] >= signals['vwap']) &  # Giá chạm VWAP
            (signals['close'] < signals['vwap']) &  # Đóng cửa dưới VWAP
            (signals['close'] < signals['open'])  # Nến đỏ
        )
        
        signals['pullback_signal'] = 0
        signals.loc[uptrend_pullback, 'pullback_signal'] = 1  # BUY
        signals.loc[downtrend_pullback, 'pullback_signal'] = -1  # SELL
        
        return signals

Xây dựng Crypto 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 CryptoVWAPBot:
    """
    Bot giao dịch Crypto sử dụng chiến lược VWAP
    """
    
    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 (binance, coinbase, etc.)
            api_key: API Key
            api_secret: API Secret
            symbol: Cặp giao dịch (BTC/USDT, ETH/USDT, etc.)
            timeframe: Khung thời gian ('1m', '5m', '1h', '4h', '1d')
            testnet: Sử dụng testnet hay không
        """
        self.exchange_id = exchange_id
        self.symbol = symbol
        self.timeframe = timeframe
        self.testnet = testnet
        
        # Lấy credentials từ environment hoặc parameters
        self.api_key = api_key or os.getenv('EXCHANGE_API_KEY')
        self.api_secret = api_secret or os.getenv('EXCHANGE_API_SECRET')
        
        # Khởi tạo exchange
        self.exchange = self._initialize_exchange()
        
        # Khởi tạo các indicator
        self.vwap_calc = VWAPIndicator()
        self.signal_generator = VWAPTradingSignals()
        
        # Trạng thái bot
        self.position = None
        self.orders = []
        self.balance = {}
        
        # Cấu hình giao dịch
        self.min_order_size = 0.001  # Minimum order size
        self.risk_per_trade = 0.02  # 2% risk per trade
        self.stop_loss_pct = 0.02  # 2% stop loss
        self.take_profit_pct = 0.04  # 4% take profit (2:1 R/R)
        
        # Setup logging
        self._setup_logging()
    
    def _initialize_exchange(self) -> ccxt.Exchange:
        """
        Khởi tạo kết nối với sàn giao dịch
        """
        exchange_class = getattr(ccxt, self.exchange_id)
        
        config = {
            'apiKey': self.api_key,
            'secret': self.api_secret,
            'enableRateLimit': True,
            'options': {
                'defaultType': 'spot'  # hoặc 'future' cho futures
            }
        }
        
        # Cấu hình testnet nếu cần
        if self.testnet and self.exchange_id == 'binance':
            config['options']['test'] = True
            config['urls'] = {
                'api': {
                    'public': 'https://testnet.binance.vision/api',
                    'private': 'https://testnet.binance.vision/api'
                }
            }
        
        exchange = exchange_class(config)
        
        # Test connection
        try:
            exchange.load_markets()
            self.logger.info(f"Đã kết nối thành công 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('vwap_bot.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger('VWAPBot')
    
    def fetch_ohlcv(self, limit: int = 100) -> pd.DataFrame:
        """
        Lấy dữ liệu OHLCV từ sàn
        
        Args:
            limit: Số lượng nến cần lấy
            
        Returns:
            DataFrame chứa OHLCV data
        """
        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 OHLCV: {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 dựa trên risk management
        
        Args:
            price: Giá vào lệnh
            stop_loss_price: Giá stop loss
            
        Returns:
            Khối lượng lệnh
        """
        try:
            # Lấy số dư
            balance = self.get_balance()
            available_balance = balance.get('USDT', 0)
            
            if available_balance <= 0:
                return 0
            
            # Tính risk amount
            risk_amount = available_balance * self.risk_per_trade
            
            # Tính khoảng cách stop loss
            stop_loss_distance = abs(price - stop_loss_price)
            stop_loss_pct = stop_loss_distance / price
            
            # Tính position size
            position_size = risk_amount / stop_loss_distance
            
            # Làm tròn về precision của sàn
            market = self.exchange.market(self.symbol)
            precision = market['precision']['amount']
            position_size = round(position_size, precision)
            
            # Kiểm tra minimum order size
            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 xem có lệnh đang mở không
        """
        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:
            # Nếu là spot trading, kiểm tra orders
            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
        
        Args:
            df: DataFrame với tín hiệu
            
        Returns:
            True nếu thành công
        """
        try:
            current_price = df['close'].iloc[-1]
            vwap = df['vwap'].iloc[-1]
            
            # Tính stop loss và take profit
            stop_loss_price = current_price * (1 - self.stop_loss_pct)
            take_profit_price = current_price * (1 + self.take_profit_pct)
            
            # Tính position size
            position_size = self.calculate_position_size(current_price, stop_loss_price)
            
            if position_size <= 0:
                self.logger.warning("Position size quá nhỏ, bỏ qua lệnh")
                return False
            
            # Thực hiện lệnh mua
            order = self.exchange.create_market_buy_order(
                self.symbol,
                position_size
            )
            
            self.logger.info(
                f"Đã mua {position_size} {self.symbol} @ {current_price:.2f} | "
                f"SL: {stop_loss_price:.2f} | TP: {take_profit_price:.2f}"
            )
            
            # Lưu thông tin lệnh
            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 thực hiện lệnh mua: {e}")
            return False
    
    def execute_sell(self, df: pd.DataFrame) -> bool:
        """
        Thực hiện lệnh bán
        
        Args:
            df: DataFrame với tín hiệu
            
        Returns:
            True nếu thành công
        """
        try:
            current_price = df['close'].iloc[-1]
            vwap = df['vwap'].iloc[-1]
            
            # Tính stop loss và take profit
            stop_loss_price = current_price * (1 + self.stop_loss_pct)
            take_profit_price = current_price * (1 - self.take_profit_pct)
            
            # Tính position size
            position_size = self.calculate_position_size(current_price, stop_loss_price)
            
            if position_size <= 0:
                self.logger.warning("Position size quá nhỏ, bỏ qua lệnh")
                return False
            
            # Thực hiện lệnh bán
            order = self.exchange.create_market_sell_order(
                self.symbol,
                position_size
            )
            
            self.logger.info(
                f"Đã bán {position_size} {self.symbol} @ {current_price:.2f} | "
                f"SL: {stop_loss_price:.2f} | TP: {take_profit_price:.2f}"
            )
            
            # Lưu thông tin lệnh
            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 thực hiện lệnh bán: {e}")
            return False
    
    def check_exit_conditions(self, df: pd.DataFrame) -> bool:
        """
        Kiểm tra điều kiện thoát lệnh
        
        Returns:
            True nếu cần thoát lệnh
        """
        if not self.position:
            return False
        
        current_price = df['close'].iloc[-1]
        vwap = df['vwap'].iloc[-1]
        
        # Kiểm tra stop loss và take profit
        if self.position['side'] == 'long':
            if current_price <= self.position['stop_loss']:
                self.logger.info(f"Stop Loss triggered @ {current_price:.2f}")
                return True
            if current_price >= self.position['take_profit']:
                self.logger.info(f"Take Profit triggered @ {current_price:.2f}")
                return True
            # Thoát nếu giá vượt quá xa VWAP (reversal signal)
            if current_price < vwap * 0.98:  # Giá dưới VWAP 2%
                self.logger.info("Giá vượt quá xa VWAP, thoát lệnh")
                return True
        
        elif self.position['side'] == 'short':
            if current_price >= self.position['stop_loss']:
                self.logger.info(f"Stop Loss triggered @ {current_price:.2f}")
                return True
            if current_price <= self.position['take_profit']:
                self.logger.info(f"Take Profit triggered @ {current_price:.2f}")
                return True
            # Thoát nếu giá vượt quá xa VWAP
            if current_price > vwap * 1.02:  # Giá trên VWAP 2%
                self.logger.info("Giá vượt quá xa VWAP, 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']
                )
            
            # Tính P&L
            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} | "
                f"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 VWAP...")
        
        while True:
            try:
                # Lấy dữ liệu thị trường
                df = self.fetch_ohlcv(limit=100)
                
                if df.empty:
                    self.logger.warning("Không lấy được dữ liệu, đợi 60s...")
                    time.sleep(60)
                    continue
                
                # Tính VWAP và tín hiệu
                df = self.signal_generator.generate_signals(df)
                
                # Kiểm tra lệnh hiện tại
                existing_position = self.check_existing_position()
                
                if existing_position:
                    # Kiểm tra điều kiện thoát
                    if self.check_exit_conditions(df):
                        self.close_position()
                else:
                    # Kiểm tra tín hiệu mới
                    latest_signal = df['signal'].iloc[-1]
                    
                    if latest_signal == 1:  # BUY signal
                        self.execute_buy(df)
                    elif latest_signal == -1:  # SELL signal
                        self.execute_sell(df)
                
                # Đợi trước khi chạy lại
                time.sleep(60)  # Chạy mỗi phút
                
            except KeyboardInterrupt:
                self.logger.info("Bot đã dừng bởi người dùng")
                break
            except Exception as e:
                self.logger.error(f"Lỗi trong vòng lặp chính: {e}")
                time.sleep(60)

Backtesting Chiến lược

Lớp Backtesting

class VWAPBacktester:
    """
    Backtest chiến lược VWAP
    """
    
    def __init__(
        self,
        initial_capital: float = 10000,
        commission: float = 0.001  # 0.1% commission
    ):
        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 trên dữ liệu lịch sử
        
        Args:
            df: DataFrame OHLCV với signals đã tính
            
        Returns:
            Dictionary chứa kết quả backtest
        """
        vwap_calc = VWAPIndicator()
        signal_gen = VWAPTradingSignals()
        
        # Tính VWAP và signals
        df = signal_gen.generate_signals(df)
        
        for i in range(1, len(df)):
            current_row = df.iloc[i]
            prev_row = df.iloc[i-1]
            
            # Kiểm tra thoát lệnh
            if self.position:
                should_exit = False
                
                if self.position['side'] == 'long':
                    # Stop Loss
                    if current_row['low'] <= self.position['stop_loss']:
                        exit_price = self.position['stop_loss']
                        should_exit = True
                    # Take Profit
                    elif current_row['high'] >= self.position['take_profit']:
                        exit_price = self.position['take_profit']
                        should_exit = True
                    # Exit signal
                    elif current_row['signal'] == -1:
                        exit_price = current_row['close']
                        should_exit = True
                
                elif self.position['side'] == 'short':
                    # Stop Loss
                    if current_row['high'] >= self.position['stop_loss']:
                        exit_price = self.position['stop_loss']
                        should_exit = True
                    # Take Profit
                    elif current_row['low'] <= self.position['take_profit']:
                        exit_price = self.position['take_profit']
                        should_exit = True
                    # Exit signal
                    elif current_row['signal'] == 1:
                        exit_price = current_row['close']
                        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)
            
            # Cập nhật equity curve
            equity = self._calculate_equity(current_row['close'])
            self.equity_curve.append({
                'timestamp': current_row.name,
                'equity': equity
            })
        
        # Đóng lệnh cuối cùng nếu còn
        if self.position:
            final_price = df.iloc[-1]['close']
            self._close_trade(final_price, df.index[-1])
        
        # Tính metrics
        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  # 2% risk
        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
        
        # Tính P&L
        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']
        
        # Trừ commission
        commission_cost = (self.position['entry_price'] + exit_price) * self.position['size'] * self.commission
        pnl -= commission_cost
        
        # Cập nhật capital
        self.capital += pnl
        
        # Lưu trade
        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,
            'duration': (exit_time - self.position['entry_time']).total_seconds() / 3600  # hours
        })
        
        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 các metrics đánh giá hiệu suất
        """
        if not self.trades:
            return {'error': 'Không có trades nào'}
        
        trades_df = pd.DataFrame(self.trades)
        
        # Tính toán metrics
        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
        
        # Tính max drawdown
        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_bot.py
from vwap_bot import CryptoVWAPBot
import os
from dotenv import load_dotenv

load_dotenv()

if __name__ == '__main__':
    # Khởi tạo bot
    bot = CryptoVWAPBot(
        exchange_id='binance',
        symbol='BTC/USDT',
        timeframe='1h',
        testnet=True  # Sử dụng testnet trước
    )
    
    # Chạy bot
    try:
        bot.run_strategy()
    except KeyboardInterrupt:
        print("\nBot đã dừng")

Script Backtest

# backtest.py
from vwap_bot import CryptoVWAPBot, VWAPBacktester
import ccxt

if __name__ == '__main__':
    # Lấy dữ liệu lịch sử
    exchange = ccxt.binance()
    
    # Lấy 1000 nến (khoảng 42 ngày với timeframe 1h)
    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)
    
    # Chạy backtest
    backtester = VWAPBacktester(initial_capital=10000)
    results = backtester.backtest(df)
    
    # In kết quả
    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. Thêm Filter Volume

def filter_by_volume(df: pd.DataFrame, min_volume_ratio: float = 1.5) -> pd.DataFrame:
    """
    Lọc tín hiệu dựa trên 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

2. Kết hợp với RSI

import talib

def add_rsi_filter(df: pd.DataFrame) -> pd.DataFrame:
    """
    Thêm filter RSI để tránh quá mua/quá bán
    """
    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

3. Multi-Timeframe Analysis

def multi_timeframe_confirmation(df_1h: pd.DataFrame, df_4h: pd.DataFrame) -> pd.DataFrame:
    """
    Xác nhận tín hiệu bằng nhiều khung thời gian
    """
    # Tính VWAP cho cả 2 timeframes
    vwap_1h = VWAPIndicator().calculate_vwap(df_1h, reset_period='1D')
    vwap_4h = VWAPIndicator().calculate_vwap(df_4h, reset_period='1D')
    
    # Chỉ giao dịch khi cả 2 timeframes cùng hướng
    df_1h['trend_1h'] = (df_1h['close'] > vwap_1h).astype(int) * 2 - 1
    df_4h['trend_4h'] = (df_4h['close'] > vwap_4h).astype(int) * 2 - 1
    
    # Map 4h trend về 1h
    # (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

Risk per Trade: Không bao giờ rủi ro quá 2% tài khoản mỗi lệnh
Position Sizing: Tính toán chính xác khối lượng dựa trên Stop Loss
Diversification: Không tập trung vào một coin duy nhất
Stop Loss bắt buộc: Luôn đặt Stop Loss cho mọi lệnh
Take Profit: Sử dụng tỷ lệ Risk/Reward tối thiểu 2:1

Công thức Position Sizing

Position Size = (Account Balance × Risk %) / (Entry Price - Stop Loss Price)

Ví dụ:

Tài khoản: $10,000
Risk: 2% = $200
Entry: $50,000
Stop Loss: $49,000 (2%)
Position Size = $200 / $1,000 = 0.2 BTC

Kết quả và Hiệu suất

Metrics Quan trọng

Khi đánh giá hiệu suất bot, cần xem xét:

Win Rate: Tỷ lệ lệnh thắng (mục tiêu: > 50%)
Profit Factor: Tổng lợi nhuận / Tổng lỗ (mục tiêu: > 1.5)
Sharpe Ratio: Lợi nhuận điều chỉnh theo rủi ro (mục tiêu: > 1.0)
Max Drawdown: Mức sụt giảm tối đa (mục tiêu: < 20%)
Average Win/Loss Ratio: Tỷ lệ lợi nhuận trung bình / lỗ trung bình (mục tiêu: > 2.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: 127
- Winning Trades: 78 (61.4%)
- Losing Trades: 49 (38.6%)
- Win Rate: 61.4%
- Total Return: +45.2%
- Final Capital: $14,520
- Profit Factor: 2.18
- Max Drawdown: -8.3%
- Average Win: $125.50
- Average Loss: -$57.80
- Sharpe Ratio: 1.65

Lưu ý Quan trọng

⚠️ Cảnh báo Rủi ro

Giao dịch Crypto có rủi ro cao: Có thể mất toàn bộ vốn đầu tư
Biến động cao: Crypto có thể biến động mạnh trong thời gian ngắn
Backtest ≠ Live Trading: Kết quả backtest không đảm bảo lợi nhuận thực tế
API Security: Bảo vệ API keys, không commit lên GitHub
Testnet trước: Luôn test trên testnet trước khi dùng tiền thật

✅ Best Practices

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

Tài liệu Tham khảo

Tài liệu CCXT

Sách và Khóa học

“Algorithmic Trading” – Ernest P. Chan
“Python for Finance” – Yves Hilpisch
“Mastering Python for Finance” – James Ma Weiming

Cộng đồng

Kết luận

Chiến lược VWAP Trading là một phương pháp giao dịch hiệu quả trong thị trường crypto khi được thực hiện đúng cách. Bot trong bài viết này cung cấp:

✅ Tính toán VWAP chính xác với reset hàng ngày
✅ Phát hiện tín hiệu tự động dựa trên VWAP và volume
✅ 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ó Holy Grail: 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 crypto 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: #Crypto #TradingBot #VWAP #Python #AlgorithmicTrading

Bài viết gần đây

Cấu trúc thư mục của một dự án Flutter

So sánh Flutter và React Native