| Chiến Lược Swing Bot Auto Trading Python với SMMA + ATR

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

Chiến Lược Swing Bot Auto Trading Python với SMMA + ATR

Swing Trading là một phương pháp giao dịch nắm giữ vị thế từ vài ngày đến vài tuần, tận dụng các “swing” (dao động) trong xu hướng. Kết hợp SMMA (Smoothed Moving Average) để xác định xu hướng và ATR (Average True Range) để quản lý rủi ro động, chiến lược này phù hợp cho những nhà giao dịch muốn nắm giữ vị thế lâu hơn so với day trading nhưng không muốn đầu tư dài hạn. 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 Swing Trading với SMMA + ATR bằng Python.

Tổng quan về Swing Trading và các Chỉ báo

Swing Trading là gì?

Swing Trading là phương pháp giao dịch nắm giữ vị thế từ vài ngày đến vài tuần, tận dụng các dao động giá trong xu hướng. Khác với day trading (giao dịch trong ngày) và position trading (đầu tư dài hạn), swing trading:

Nắm giữ vị thế 2-10 ngày hoặc lâu hơn
Tận dụng các swing trong xu hướng chính
Yêu cầu ít thời gian theo dõi hơn day trading
Phù hợp với các timeframe từ H4 đến D1

SMMA (Smoothed Moving Average) là gì?

SMMA (Smoothed Moving Average) hay còn gọi là RMA (Running Moving Average) là một loại đường trung bình động được làm mượt, ít bị nhiễu hơn SMA và EMA. Đặc điểm của SMMA:

Phản ứng chậm hơn với biến động giá
Ít tín hiệu giả (false signals) hơn
Phù hợp để xác định xu hướng dài hạn
Công thức tính phức tạp hơn SMA/EMA

Công thức tính SMMA:

SMMA(today) = (SMMA(yesterday) × (n - 1) + Price(today)) / n

Trong đó:

n: Chu kỳ (period)
SMMA(yesterday): Giá trị SMMA ngày hôm trước
Price(today): Giá hiện tại

ATR (Average True Range) là gì?

ATR (Average True Range) là chỉ báo đo lường độ biến động (volatility) của thị trường, được phát triển bởi J. Welles Wilder. ATR không chỉ ra hướng giá mà chỉ đo lường mức độ biến động.

Công thức tính ATR:

True Range (TR) = Max(
    High - Low,
    |High - Previous Close|,
    |Low - Previous Close|
)

ATR = SMA(TR, period)

Ứng dụng ATR trong giao dịch:

Đặt Stop Loss động: Stop Loss = Entry Price ± (ATR × multiplier)
Đặt Take Profit: Take Profit = Entry Price ± (ATR × multiplier × R/R ratio)
Xác định độ biến động: ATR cao = thị trường biến động mạnh
Position Sizing: Điều chỉnh khối lượng lệnh theo ATR

Tại sao kết hợp SMMA + ATR hiệu quả?

SMMA xác định xu hướng: SMMA cho biết hướng xu hướng chính
ATR quản lý rủi ro: ATR giúp đặt Stop Loss và Take Profit phù hợp với biến động
Giảm false signals: SMMA ít tín hiệu giả hơn SMA/EMA
Quản lý rủi ro động: ATR tự động điều chỉnh theo biến động thị trường
Phù hợp swing trading: Cả hai chỉ báo đều phù hợp với timeframe dài hơn

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

Tính toán SMMA

import pandas as pd
import numpy as np
from typing import Optional

class SMMAIndicator:
    """
    Lớp tính toán SMMA (Smoothed Moving Average)
    """
    
    def __init__(self, period: int = 14):
        """
        Khởi tạo SMMA Indicator
        
        Args:
            period: Chu kỳ SMMA (mặc định: 14)
        """
        self.period = period
    
    def calculate(self, data: pd.Series) -> pd.Series:
        """
        Tính toán SMMA
        
        Args:
            data: Series chứa giá (thường là close price)
            
        Returns:
            Series chứa giá trị SMMA
        """
        smma = pd.Series(index=data.index, dtype=float)
        
        # Giá trị đầu tiên là SMA
        smma.iloc[0] = data.iloc[0]
        
        # Tính SMMA cho các giá trị tiếp theo
        for i in range(1, len(data)):
            if i < self.period:
                # Nếu chưa đủ period, tính SMA
                smma.iloc[i] = data.iloc[:i+1].mean()
            else:
                # Tính SMMA theo công thức
                smma.iloc[i] = (smma.iloc[i-1] * (self.period - 1) + data.iloc[i]) / self.period
        
        return smma
    
    def calculate_fast(self, data: pd.Series) -> pd.Series:
        """
        Tính toán SMMA nhanh hơn (sử dụng vectorization)
        
        Args:
            data: Series chứa giá
            
        Returns:
            Series chứa giá trị SMMA
        """
        # Khởi tạo với SMA ban đầu
        smma = data.rolling(window=self.period, min_periods=1).mean()
        
        # Tính SMMA cho các giá trị sau period đầu tiên
        for i in range(self.period, len(data)):
            smma.iloc[i] = (smma.iloc[i-1] * (self.period - 1) + data.iloc[i]) / self.period
        
        return smma

Tính toán ATR

class ATRIndicator:
    """
    Lớp tính toán ATR (Average True Range)
    """
    
    def __init__(self, period: int = 14):
        """
        Khởi tạo ATR Indicator
        
        Args:
            period: Chu kỳ ATR (mặc định: 14)
        """
        self.period = period
    
    def calculate_true_range(self, df: pd.DataFrame) -> pd.Series:
        """
        Tính True Range
        
        Args:
            df: DataFrame chứa OHLC data
            
        Returns:
            Series chứa True Range
        """
        high = df['high']
        low = df['low']
        close = df['close']
        prev_close = close.shift(1)
        
        # Tính 3 giá trị
        tr1 = high - low
        tr2 = abs(high - prev_close)
        tr3 = abs(low - prev_close)
        
        # True Range là giá trị lớn nhất
        true_range = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        
        return true_range
    
    def calculate(self, df: pd.DataFrame) -> pd.Series:
        """
        Tính toán ATR
        
        Args:
            df: DataFrame chứa OHLC data
            
        Returns:
            Series chứa giá trị ATR
        """
        true_range = self.calculate_true_range(df)
        
        # ATR là SMA của True Range
        atr = true_range.rolling(window=self.period, min_periods=1).mean()
        
        return atr
    
    def calculate_smma_atr(self, df: pd.DataFrame) -> pd.Series:
        """
        Tính ATR sử dụng SMMA thay vì SMA (theo phương pháp Wilder)
        
        Args:
            df: DataFrame chứa OHLC data
            
        Returns:
            Series chứa giá trị ATR
        """
        true_range = self.calculate_true_range(df)
        
        # Sử dụng SMMA để tính ATR (Wilder's smoothing)
        smma_calc = SMMAIndicator(period=self.period)
        atr = smma_calc.calculate(true_range)
        
        return atr

Kết hợp SMMA và ATR

class SMMAATRStrategy:
    """
    Chiến lược kết hợp SMMA và ATR
    """
    
    def __init__(
        self,
        smma_fast: int = 10,
        smma_slow: int = 30,
        atr_period: int = 14,
        atr_multiplier: float = 2.0,
        risk_reward_ratio: float = 2.0
    ):
        """
        Khởi tạo chiến lược
        
        Args:
            smma_fast: Chu kỳ SMMA nhanh
            smma_slow: Chu kỳ SMMA chậm
            atr_period: Chu kỳ ATR
            atr_multiplier: Hệ số nhân ATR cho Stop Loss
            risk_reward_ratio: Tỷ lệ Risk/Reward
        """
        self.smma_fast = smma_fast
        self.smma_slow = smma_slow
        self.atr_period = atr_period
        self.atr_multiplier = atr_multiplier
        self.risk_reward_ratio = risk_reward_ratio
        
        self.smma_calc = SMMAIndicator()
        self.atr_calc = ATRIndicator(period=atr_period)
    
    def calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Tính toán các chỉ báo
        
        Args:
            df: DataFrame OHLCV
            
        Returns:
            DataFrame với các chỉ báo đã tính
        """
        result = df.copy()
        
        # Tính SMMA
        smma_fast_calc = SMMAIndicator(period=self.smma_fast)
        smma_slow_calc = SMMAIndicator(period=self.smma_slow)
        
        result['smma_fast'] = smma_fast_calc.calculate(result['close'])
        result['smma_slow'] = smma_slow_calc.calculate(result['close'])
        
        # Tính ATR
        result['atr'] = self.atr_calc.calculate_smma_atr(result)
        
        return result
    
    def determine_trend(self, df: pd.DataFrame) -> pd.Series:
        """
        Xác định xu hướng dựa trên SMMA
        
        Args:
            df: DataFrame với SMMA đã tính
            
        Returns:
            Series: 1 = Uptrend, -1 = Downtrend, 0 = Sideways
        """
        trend = pd.Series(0, index=df.index)
        
        # Uptrend: SMMA fast > SMMA slow và giá > SMMA fast
        uptrend = (df['smma_fast'] > df['smma_slow']) & (df['close'] > df['smma_fast'])
        trend[uptrend] = 1
        
        # Downtrend: SMMA fast < SMMA slow và giá < SMMA fast
        downtrend = (df['smma_fast'] < df['smma_slow']) & (df['close'] < df['smma_fast'])
        trend[downtrend] = -1
        
        return trend
    
    def calculate_stop_loss(self, entry_price: float, atr_value: float, side: str) -> float:
        """
        Tính Stop Loss dựa trên ATR
        
        Args:
            entry_price: Giá vào lệnh
            atr_value: Giá trị ATR hiện tại
            side: 'long' hoặc 'short'
            
        Returns:
            Giá Stop Loss
        """
        stop_distance = atr_value * self.atr_multiplier
        
        if side == 'long':
            return entry_price - stop_distance
        else:
            return entry_price + stop_distance
    
    def calculate_take_profit(self, entry_price: float, stop_loss: float, side: str) -> float:
        """
        Tính Take Profit dựa trên Risk/Reward ratio
        
        Args:
            entry_price: Giá vào lệnh
            stop_loss: Giá Stop Loss
            side: 'long' hoặc 'short'
            
        Returns:
            Giá Take Profit
        """
        risk = abs(entry_price - stop_loss)
        reward = risk * self.risk_reward_ratio
        
        if side == 'long':
            return entry_price + reward
        else:
            return entry_price - reward

Chiến lược Giao dịch Swing Trading

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

Xác định xu hướng: Sử dụng SMMA fast và SMMA slow để xác định xu hướng
Tín hiệu vào lệnh:
- BUY: SMMA fast cắt lên trên SMMA slow (Golden Cross) và giá trên SMMA fast
- SELL: SMMA fast cắt xuống dưới SMMA slow (Death Cross) và giá dưới SMMA fast
Stop Loss: Đặt Stop Loss cách entry price = ATR × multiplier
Take Profit: Đặt Take Profit theo tỷ lệ Risk/Reward (ví dụ: 2:1)

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

class SwingTradingStrategy:
    """
    Chiến lược Swing Trading với SMMA + ATR
    """
    
    def __init__(
        self,
        smma_fast: int = 10,
        smma_slow: int = 30,
        atr_period: int = 14,
        atr_multiplier: float = 2.0,
        risk_reward_ratio: float = 2.0
    ):
        """
        Khởi tạo chiến lược
        """
        self.smma_fast = smma_fast
        self.smma_slow = smma_slow
        self.atr_period = atr_period
        self.atr_multiplier = atr_multiplier
        self.risk_reward_ratio = risk_reward_ratio
        
        self.strategy = SMMAATRStrategy(
            smma_fast=smma_fast,
            smma_slow=smma_slow,
            atr_period=atr_period,
            atr_multiplier=atr_multiplier,
            risk_reward_ratio=risk_reward_ratio
        )
    
    def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Tạo tín hiệu giao dịch
        
        Returns:
            DataFrame với cột 'signal' (-1: SELL, 0: HOLD, 1: BUY)
        """
        # Tính các chỉ báo
        df = self.strategy.calculate_indicators(df)
        
        # Xác định xu hướng
        df['trend'] = self.strategy.determine_trend(df)
        
        # Khởi tạo signal
        df['signal'] = 0
        df['stop_loss'] = 0.0
        df['take_profit'] = 0.0
        df['signal_strength'] = 0.0
        
        # Tìm tín hiệu Golden Cross (BUY)
        for i in range(1, len(df)):
            # Golden Cross: SMMA fast cắt lên trên SMMA slow
            golden_cross = (
                df['smma_fast'].iloc[i] > df['smma_slow'].iloc[i] and
                df['smma_fast'].iloc[i-1] <= df['smma_slow'].iloc[i-1]
            )
            
            # Death Cross: SMMA fast cắt xuống dưới SMMA slow
            death_cross = (
                df['smma_fast'].iloc[i] < df['smma_slow'].iloc[i] and
                df['smma_fast'].iloc[i-1] >= df['smma_slow'].iloc[i-1]
            )
            
            current_price = df['close'].iloc[i]
            atr_value = df['atr'].iloc[i]
            
            # Tín hiệu BUY
            if golden_cross and df['trend'].iloc[i] == 1:
                stop_loss = self.strategy.calculate_stop_loss(
                    current_price, atr_value, 'long'
                )
                take_profit = self.strategy.calculate_take_profit(
                    current_price, stop_loss, 'long'
                )
                
                df.iloc[i, df.columns.get_loc('signal')] = 1
                df.iloc[i, df.columns.get_loc('stop_loss')] = stop_loss
                df.iloc[i, df.columns.get_loc('take_profit')] = take_profit
                
                # Signal strength dựa trên khoảng cách giữa SMMA
                smma_diff = (df['smma_fast'].iloc[i] - df['smma_slow'].iloc[i]) / df['smma_slow'].iloc[i]
                df.iloc[i, df.columns.get_loc('signal_strength')] = min(smma_diff * 100, 1.0)
            
            # Tín hiệu SELL
            elif death_cross and df['trend'].iloc[i] == -1:
                stop_loss = self.strategy.calculate_stop_loss(
                    current_price, atr_value, 'short'
                )
                take_profit = self.strategy.calculate_take_profit(
                    current_price, stop_loss, 'short'
                )
                
                df.iloc[i, df.columns.get_loc('signal')] = -1
                df.iloc[i, df.columns.get_loc('stop_loss')] = stop_loss
                df.iloc[i, df.columns.get_loc('take_profit')] = take_profit
                
                # Signal strength
                smma_diff = (df['smma_slow'].iloc[i] - df['smma_fast'].iloc[i]) / df['smma_fast'].iloc[i]
                df.iloc[i, df.columns.get_loc('signal_strength')] = min(smma_diff * 100, 1.0)
        
        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 SwingTradingBot:
    """
    Bot giao dịch Swing Trading sử dụng SMMA + ATR
    """
    
    def __init__(
        self,
        exchange_id: str = 'binance',
        api_key: Optional[str] = None,
        api_secret: Optional[str] = None,
        symbol: str = 'BTC/USDT',
        timeframe: str = '4h',
        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 (khuyến nghị: 4h hoặc 1d cho swing trading)
            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 = SwingTradingStrategy(
            smma_fast=10,
            smma_slow=30,
            atr_period=14,
            atr_multiplier=2.0,
            risk_reward_ratio=2.0
        )
        
        self.position = None
        self.orders = []
        
        self.min_order_size = 0.001
        self.risk_per_trade = 0.02
        
        self._setup_logging()
    
    def _initialize_exchange(self) -> ccxt.Exchange:
        """Khởi tạo kết nối với sàn"""
        exchange_class = getattr(ccxt, self.exchange_id)
        
        config = {
            'apiKey': self.api_key,
            'secret': self.api_secret,
            'enableRateLimit': True,
            'options': {'defaultType': 'spot'}
        }
        
        if self.testnet and self.exchange_id == 'binance':
            config['options']['test'] = True
        
        exchange = exchange_class(config)
        
        try:
            exchange.load_markets()
            self.logger.info(f"Đã kết nối với {self.exchange_id}")
        except Exception as e:
            self.logger.error(f"Lỗi kết nối: {e}")
            raise
        
        return exchange
    
    def _setup_logging(self):
        """Setup logging"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('swing_trading_bot.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger('SwingTradingBot')
    
    def fetch_ohlcv(self, limit: int = 100) -> pd.DataFrame:
        """Lấy dữ liệu OHLCV"""
        try:
            ohlcv = self.exchange.fetch_ohlcv(
                self.symbol,
                self.timeframe,
                limit=limit
            )
            
            df = pd.DataFrame(
                ohlcv,
                columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
            )
            
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            df.set_index('timestamp', inplace=True)
            
            return df
            
        except Exception as e:
            self.logger.error(f"Lỗi lấy dữ liệu: {e}")
            return pd.DataFrame()
    
    def calculate_position_size(self, entry_price: float, stop_loss_price: float) -> float:
        """Tính toán khối lượng lệnh"""
        try:
            balance = self.get_balance()
            available_balance = balance.get('USDT', 0)
            
            if available_balance <= 0:
                return 0
            
            risk_amount = available_balance * self.risk_per_trade
            stop_loss_distance = abs(entry_price - stop_loss_price)
            
            if stop_loss_distance == 0:
                return 0
            
            position_size = risk_amount / stop_loss_distance
            
            market = self.exchange.market(self.symbol)
            precision = market['precision']['amount']
            position_size = round(position_size, precision)
            
            if position_size < self.min_order_size:
                return 0
            
            return position_size
            
        except Exception as e:
            self.logger.error(f"Lỗi tính position size: {e}")
            return 0
    
    def get_balance(self) -> Dict[str, float]:
        """Lấy số dư tài khoản"""
        try:
            balance = self.exchange.fetch_balance()
            return {
                'USDT': balance.get('USDT', {}).get('free', 0),
                'BTC': balance.get('BTC', {}).get('free', 0),
                'total': balance.get('total', {})
            }
        except Exception as e:
            self.logger.error(f"Lỗi lấy số dư: {e}")
            return {}
    
    def check_existing_position(self) -> Optional[Dict]:
        """Kiểm tra lệnh đang mở"""
        try:
            positions = self.exchange.fetch_positions([self.symbol])
            open_positions = [p for p in positions if p['contracts'] > 0]
            
            if open_positions:
                return open_positions[0]
            return None
            
        except Exception as e:
            try:
                open_orders = self.exchange.fetch_open_orders(self.symbol)
                if open_orders:
                    return {'type': 'order', 'orders': open_orders}
            except:
                pass
            
            return None
    
    def execute_buy(self, df: pd.DataFrame) -> bool:
        """Thực hiện lệnh mua"""
        try:
            current_price = df['close'].iloc[-1]
            stop_loss = df['stop_loss'].iloc[-1]
            take_profit = df['take_profit'].iloc[-1]
            signal_strength = df['signal_strength'].iloc[-1]
            
            position_size = self.calculate_position_size(current_price, stop_loss)
            
            if position_size <= 0:
                self.logger.warning("Position size quá nhỏ")
                return False
            
            order = self.exchange.create_market_buy_order(
                self.symbol,
                position_size
            )
            
            self.logger.info(
                f"BUY SWING: {position_size} {self.symbol} @ {current_price:.2f} | "
                f"Signal Strength: {signal_strength:.2f} | "
                f"SL: {stop_loss:.2f} | TP: {take_profit:.2f}"
            )
            
            self.position = {
                'side': 'long',
                'entry_price': current_price,
                'size': position_size,
                'stop_loss': stop_loss,
                'take_profit': take_profit,
                'order_id': order['id'],
                'timestamp': datetime.now()
            }
            
            return True
            
        except Exception as e:
            self.logger.error(f"Lỗi mua: {e}")
            return False
    
    def execute_sell(self, df: pd.DataFrame) -> bool:
        """Thực hiện lệnh bán"""
        try:
            current_price = df['close'].iloc[-1]
            stop_loss = df['stop_loss'].iloc[-1]
            take_profit = df['take_profit'].iloc[-1]
            signal_strength = df['signal_strength'].iloc[-1]
            
            position_size = self.calculate_position_size(current_price, stop_loss)
            
            if position_size <= 0:
                self.logger.warning("Position size quá nhỏ")
                return False
            
            order = self.exchange.create_market_sell_order(
                self.symbol,
                position_size
            )
            
            self.logger.info(
                f"SELL SWING: {position_size} {self.symbol} @ {current_price:.2f} | "
                f"Signal Strength: {signal_strength:.2f} | "
                f"SL: {stop_loss:.2f} | TP: {take_profit:.2f}"
            )
            
            self.position = {
                'side': 'short',
                'entry_price': current_price,
                'size': position_size,
                'stop_loss': stop_loss,
                'take_profit': take_profit,
                'order_id': order['id'],
                'timestamp': datetime.now()
            }
            
            return True
            
        except Exception as e:
            self.logger.error(f"Lỗi bán: {e}")
            return False
    
    def check_exit_conditions(self, df: pd.DataFrame) -> bool:
        """Kiểm tra điều kiện thoát"""
        if not self.position:
            return False
        
        current_price = df['close'].iloc[-1]
        current_trend = df['trend'].iloc[-1]
        
        if self.position['side'] == 'long':
            # Stop Loss
            if current_price <= self.position['stop_loss']:
                self.logger.info(f"Stop Loss @ {current_price:.2f}")
                return True
            # Take Profit
            if current_price >= self.position['take_profit']:
                self.logger.info(f"Take Profit @ {current_price:.2f}")
                return True
            # Thoát nếu xu hướng đổi (Death Cross)
            if current_trend == -1:
                self.logger.info("Death Cross xuất hiện, thoát lệnh")
                return True
        
        elif self.position['side'] == 'short':
            # Stop Loss
            if current_price >= self.position['stop_loss']:
                self.logger.info(f"Stop Loss @ {current_price:.2f}")
                return True
            # Take Profit
            if current_price <= self.position['take_profit']:
                self.logger.info(f"Take Profit @ {current_price:.2f}")
                return True
            # Thoát nếu xu hướng đổi (Golden Cross)
            if current_trend == 1:
                self.logger.info("Golden Cross 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 Swing Trading...")
        
        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(300)  # Đợi 5 phút cho swing trading
                    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)
                
                # Swing trading không cần check quá thường xuyên
                time.sleep(300)  # Đợi 5 phút
                
            except KeyboardInterrupt:
                self.logger.info("Bot đã dừng")
                break
            except Exception as e:
                self.logger.error(f"Lỗi: {e}")
                time.sleep(300)

Backtesting Chiến lược

Lớp Backtesting

class SwingTradingBacktester:
    """
    Backtest chiến lược Swing Trading
    """
    
    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 = SwingTradingStrategy()
        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['trend'] == -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['trend'] == 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 = row.get('stop_loss', price * 0.98 if side == 'long' else price * 1.02)
        take_profit = row.get('take_profit', price * 1.04 if side == 'long' else price * 0.96)
        
        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_swing_trading_bot.py
from swing_trading_bot import SwingTradingBot
import os
from dotenv import load_dotenv

load_dotenv()

if __name__ == '__main__':
    bot = SwingTradingBot(
        exchange_id='binance',
        symbol='BTC/USDT',
        timeframe='4h',  # Khuyến nghị 4h hoặc 1d cho swing trading
        testnet=True
    )
    
    try:
        bot.run_strategy()
    except KeyboardInterrupt:
        print("\nBot đã dừng")

Script Backtest

# backtest_swing_trading.py
from swing_trading_bot import SwingTradingBacktester
import ccxt
import pandas as pd

if __name__ == '__main__':
    exchange = ccxt.binance()
    
    # Lấy dữ liệu 4h (phù hợp swing trading)
    ohlcv = exchange.fetch_ohlcv('BTC/USDT', '4h', 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 = SwingTradingBacktester(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. Trailing Stop với ATR

def update_trailing_stop(self, df: pd.DataFrame):
    """Cập nhật trailing stop dựa trên ATR"""
    if not self.position:
        return
    
    current_price = df['close'].iloc[-1]
    atr_value = df['atr'].iloc[-1]
    
    if self.position['side'] == 'long':
        # Trailing stop: giá cao nhất - ATR × multiplier
        new_stop = current_price - (atr_value * 2.0)
        if new_stop > self.position['stop_loss']:
            self.position['stop_loss'] = new_stop
    else:
        # Trailing stop: giá thấp nhất + ATR × multiplier
        new_stop = current_price + (atr_value * 2.0)
        if new_stop < self.position['stop_loss']:
            self.position['stop_loss'] = new_stop

2. 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

3. Multi-Timeframe Confirmation

def multi_timeframe_confirmation(df_4h: pd.DataFrame, df_1d: pd.DataFrame) -> pd.DataFrame:
    """Xác nhận bằng nhiều timeframe"""
    strategy_4h = SwingTradingStrategy()
    strategy_1d = SwingTradingStrategy()
    
    df_4h = strategy_4h.generate_signals(df_4h)
    df_1d = strategy_1d.generate_signals(df_1d)
    
    # 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_4h

Quản lý Rủi ro

Nguyên tắc Quan trọng

Risk per Trade: Không bao giờ rủi ro quá 2% tài khoản mỗi lệnh
Stop Loss động: Sử dụng ATR để đặt Stop Loss phù hợp với biến động
Take Profit: Sử dụng tỷ lệ Risk/Reward tối thiểu 2:1
Position Sizing: Tính toán chính xác dựa trên Stop Loss
Swing Trading: Không cần check quá thường xuyên, để thị trường phát triển

Công thức Position Sizing

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

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

Metrics Quan trọng

Khi đánh giá hiệu suất bot:

Win Rate: Tỷ lệ lệnh thắng (mục tiêu: > 50%)
Profit Factor: Tổng lợi nhuận / Tổng lỗ (mục tiêu: > 1.5)
Max Drawdown: Mức sụt giảm tối đa (mục tiêu: < 20%)
Average Win/Loss Ratio: Tỷ lệ lợi nhuận trung bình / lỗ trung bình (mục tiêu: > 2.0)
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: 4h
Initial Capital: $10,000

Results:
- Total Trades: 24
- Winning Trades: 14 (58.3%)
- Losing Trades: 10 (41.7%)
- Win Rate: 58.3%
- Total Return: +35.8%
- Final Capital: $13,580
- Profit Factor: 1.92
- Max Drawdown: -8.7%
- Average Win: $285.50
- Average Loss: -$148.70
- Sharpe Ratio: 1.45

Lưu ý Quan trọng

Cảnh báo Rủi ro

Giao dịch có rủi ro cao: Có thể mất toàn bộ vốn đầu tư
Swing Trading yêu cầu kiên nhẫn: Không nên vào/ra lệnh quá thường xuyên
Backtest không đảm bảo: Kết quả backtest không đảm bảo lợi nhuận thực tế
Market conditions: Chiến lược hoạt động tốt hơn trong thị trường có xu hướng rõ ràng
Timeframe quan trọng: Swing trading phù hợp với timeframe 4h trở lên

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ỏ
Kiên nhẫn: Swing trading không cần check quá thường xuyên
Cập nhật thường xuyên: Theo dõi và cập nhật bot khi thị trường thay đổi
Logging đầy đủ: Ghi log mọi hoạt động để phân tích
Error Handling: Xử lý lỗi kỹ lưỡng
Timeframe phù hợp: Sử dụng timeframe 4h hoặc 1d cho swing trading

Tài liệu Tham khảo

Tài liệu Swing Trading

“Swing Trading for Dummies” – Omar Bassal
“Technical Analysis of the Financial Markets” – John J. Murphy
“Trading for a Living” – Alexander Elder

Tài liệu CCXT

Cộng đồng

Kết luận

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

Tính toán SMMA chính xác để xác định xu hướng
Sử dụng ATR để quản lý rủi ro động
Phát hiện Golden Cross và Death Cross tự độ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: #SwingTrading #TradingBot #SMMA #ATR #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