Chiến Lược Mean Reversion Cho Bot Giao Dịch Chứng Khoán Python

Mean Reversion — giá sẽ quay trở về trung bình — là một trong những chiến lược giao dịch được kiểm chứng nhiều nhất trong lịch sử tài chính. Đặc biệt hiệu quả với chứng khoán Việt Nam vì thị trường thường bị "oversell" hoặc "overbuy" cục bộ do tâm lý bầy đàn.

Mean Reversion Là Gì?

Nguyên lý đơn giản: khi giá cổ phiếu lệch quá xa khỏi giá trị "bình thường" của nó — dù do tin tức, cảm xúc hay thanh khoản mỏng — nó có xu hướng quay trở về trung bình.

Điều kiện thị trường phù hợp:

• Sideway market hoặc range-bound
• Cổ phiếu có thanh khoản tốt, fundamental ổn định
• Không có sự kiện bất thường (M&A, phá sản, scandal)

Phương Pháp 1: Bollinger Bands

import pandas as pd
import numpy as np

def bollinger_mean_reversion(df: pd.DataFrame,
                               window: int = 20,
                               num_std: float = 2.0) -> pd.DataFrame:
    df = df.copy()
    df['ma'] = df['close'].rolling(window).mean()
    df['std'] = df['close'].rolling(window).std()
    df['upper'] = df['ma'] + num_std * df['std']
    df['lower'] = df['ma'] - num_std * df['std']

    # Tín hiệu
    df['signal'] = 0
    # Buy khi giá chạm lower band (quá bán)
    df.loc[df['close'] <= df['lower'], 'signal'] = 1
    # Sell khi giá chạm upper band (quá mua)
    df.loc[df['close'] >= df['upper'], 'signal'] = -1
    # Exit khi giá về MA
    df.loc[(df['close'] >= df['ma']) & (df['signal'].shift(1) == 1), 'signal'] = 0
    df.loc[(df['close'] <= df['ma']) & (df['signal'].shift(1) == -1), 'signal'] = 0

    return df

Phương Pháp 2: Z-Score

Z-score đo lường mức độ lệch của giá so với trung bình lịch sử:

def zscore_strategy(df: pd.DataFrame, window: int = 20,
                    entry_z: float = 2.0, exit_z: float = 0.5) -> pd.DataFrame:
    df = df.copy()
    df['zscore'] = ((df['close'] - df['close'].rolling(window).mean()) /
                    df['close'].rolling(window).std())

    df['signal'] = 0
    df.loc[df['zscore'] < -entry_z, 'signal'] = 1   # Oversold → Buy
    df.loc[df['zscore'] > entry_z, 'signal'] = -1   # Overbought → Sell
    # Exit khi z-score về gần 0
    df.loc[abs(df['zscore']) < exit_z, 'signal'] = 0

    return df

Phương Pháp 3: Pairs Trading

Cổ phiếu cùng ngành thường có giá tương quan cao. Khi spread giữa chúng lệch xa, đặt cược chúng hội tụ:

def find_cointegrated_pairs(stocks_df: pd.DataFrame, p_value_threshold: float = 0.05):
    """Tìm các cặp cổ phiếu có cointegration"""
    from statsmodels.tsa.stattools import coint

    n = len(stocks_df.columns)
    pairs = []

    for i in range(n):
        for j in range(i+1, n):
            s1 = stocks_df.columns[i]
            s2 = stocks_df.columns[j]

            score, p_value, _ = coint(stocks_df[s1], stocks_df[s2])

            if p_value < p_value_threshold:
                pairs.append({'pair': (s1, s2), 'p_value': p_value})

    return sorted(pairs, key=lambda x: x['p_value'])

def pairs_trading_signal(s1: pd.Series, s2: pd.Series,
                          window: int = 60, entry_z: float = 2.0):
    """Tạo tín hiệu pairs trading"""
    # Tính hedge ratio bằng linear regression
    from sklearn.linear_model import LinearRegression
    model = LinearRegression()
    model.fit(s2.values.reshape(-1, 1), s1.values)
    hedge_ratio = model.coef_[0]

    # Spread = s1 - hedge_ratio * s2
    spread = s1 - hedge_ratio * s2
    zscore = ((spread - spread.rolling(window).mean()) /
              spread.rolling(window).std())

    signals = pd.DataFrame(index=s1.index)
    signals['zscore'] = zscore
    # Short s1, Long s2 khi spread quá cao
    signals['s1_position'] = np.where(zscore > entry_z, -1,
                             np.where(zscore < -entry_z, 1, 0))
    signals['s2_position'] = -signals['s1_position'] * hedge_ratio

    return signals

# Ví dụ: pairs trading VCB và TCB (2 ngân hàng tương quan cao)

Backtest Mean Reversion Trên VNIndex

from backtesting import Backtest, Strategy

class MeanReversionBB(Strategy):
    window = 20
    num_std = 2.0

    def init(self):
        self.upper = self.I(lambda x: (pd.Series(x).rolling(self.window).mean() +
                           self.num_std * pd.Series(x).rolling(self.window).std()),
                           self.data.Close)
        self.lower = self.I(lambda x: (pd.Series(x).rolling(self.window).mean() -
                           self.num_std * pd.Series(x).rolling(self.window).std()),
                           self.data.Close)
        self.ma = self.I(lambda x: pd.Series(x).rolling(self.window).mean(),
                         self.data.Close)

    def next(self):
        if self.data.Close[-1] <= self.lower[-1] and not self.position:
            self.buy()
        elif self.data.Close[-1] >= self.upper[-1] and self.position.is_long:
            self.position.close()

bt = Backtest(df, MeanReversionBB, cash=100_000_000, commission=0.0025)
stats = bt.run()
print(f"Return: {stats['Return [%]']:.1f}%")
print(f"Sharpe: {stats['Sharpe Ratio']:.2f}")
print(f"Drawdown: {stats['Max. Drawdown [%]']:.1f}%")

Khi Nào Mean Reversion Thất Bại?

Mean reversion thất bại khi có trending market mạnh. Nếu VNIndex đang trong xu hướng giảm dài hạn, bot sẽ liên tục mua vào và thua lỗ.

Giải pháp: Thêm trend filter — chỉ buy khi giá nằm trên EMA(200), chỉ sell khi nằm dưới EMA(200).

df['ema200'] = df['close'].ewm(span=200).mean()
# Chỉ buy khi giá trên EMA200 (trending up environment)
df.loc[df['close'] < df['ema200'], 'signal'] = 0  # Tắt buy signal trong downtrend

Học phân tích định lượng và xây dựng bot chứng khoán thực chiến tại Hướng Nghiệp Dữ Liệu.