Tháng 3/2025, một quỹ hedge fund tại Singapore ghi nhận lợi nhuận 340% trong vòng 6 tháng nhờ chiến lược Statistical Arbitrage (Stat Arb) được tối ưu hóa bằng AI. Trong khi đó, nhiều nhà giao dịch retail Việt Nam vẫn đang vật lộn với các chiến lược mean-reversion cổ điển hay grid trading không hiệu quả. Bài viết này sẽ hướng dẫn bạn xây dựng hệ thống Statistical Arbitrage từ con số 0, tích hợp AI để phân tích và ra quyết định nhanh hơn 50 lần so với phương pháp thủ công.
Statistical Arbitrage Là Gì?
Statistical Arbitrage là chiến lược giao dịch định lượng dựa trên việc khai thác các bất hiệu quả về giá ngắn hạn giữa các tài sản có mối quan hệ tương quan. Khác với arbitrage thuần túy (đòi hỏi lợi nhuận không rủi ro), Stat Arb chấp nhận rủi ro thống kê nhưng tìm cách tối đa hóa Sharpe ratio thông qua:
- Pair Trading: Giao dịch chênh lệch giá giữa hai cổ phiếu có correlation cao
- Mean Reversion: Đặt cược rằng giá sẽ quay về giá trị trung bình lịch sử
- Cointegration Analysis: Tìm cặp cointegrated để xác định spread có tính mean-reversion
- Machine Learning: Sử dụng mô hình ML để dự đoán độ lệch giá và xác suất mean reversion
Kiến Trúc Hệ Thống Statistical Arbitrage
Hệ thống Stat Arb hiệu quả cần có 4 thành phần chính:
1. Data Layer - Thu Thập và Xử Lý Dữ Liệu
#!/usr/bin/env python3
"""
Statistical Arbitrage Data Collection System
Author: HolySheep AI Research Team
"""
import pandas as pd
import numpy as np
import requests
from datetime import datetime, timedelta
import time
class DataCollector:
"""
Thu thập dữ liệu giá tài sản từ nhiều nguồn
Tích hợp HolySheep AI để xử lý và phân tích dữ liệu
"""
HOLYSHEEP_API_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
})
def get_historical_prices(self, symbols: list, days: int = 252) -> pd.DataFrame:
"""
Lấy dữ liệu giá lịch sử cho các cặp symbol
Ví dụ: ['AAPL', 'MSFT', 'GOOGL']
"""
# Giả lập dữ liệu cho demo
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
data = {}
for symbol in symbols:
np.random.seed(hash(symbol) % 2**32)
dates = pd.date_range(start=start_date, end=end_date, freq='D')
# Sinh giá với volatility và mean-reversion characteristic
base_price = 100 + np.random.randn() * 50
returns = np.random.randn(len(dates)) * 0.02
prices = base_price * np.exp(np.cumsum(returns))
data[symbol] = prices
df = pd.DataFrame(data, index=dates)
return df.dropna()
def get_market_features(self, symbol: str) -> dict:
"""
Sử dụng HolySheep AI để phân tích đặc điểm thị trường
Trả về: volatility, momentum, correlation signals
"""
prompt = f"""Phân tích dữ liệu thị trường cho {symbol} và trả về JSON:
{{
"volatility_score": 0-100,
"momentum": "bullish/bearish/neutral",
"liquidity_rating": "high/medium/low",
"sentiment": "positive/negative/neutral",
"recommended_position_size": 0.0-1.0
}}"""
response = self.session.post(
f"{self.HOLYSHEEP_API_URL}/chat/completions",
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"max_tokens": 200
}
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content']
return {"error": "API call failed"}
def calculate_correlation_matrix(self, prices: pd.DataFrame) -> pd.DataFrame:
"""Tính ma trận tương quan giữa các tài sản"""
return prices.pct_change().corr()
==================== SỬ DỤNG ====================
if __name__ == "__main__":
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
collector = DataCollector(API_KEY)
# Thu thập dữ liệu 5 cổ phiếu tech
symbols = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'META']
prices = collector.get_historical_prices(symbols, days=365)
print("Dữ liệu giá thu thập thành công:")
print(f"Shape: {prices.shape}")
print(prices.tail())
# Tính ma trận tương quan
corr_matrix = collector.calculate_correlation_matrix(prices)
print("\nMa trận tương quan:")
print(corr_matrix.round(3))2. Pair Selection Engine - Chọn Cặp Giao Dịch
#!/usr/bin/env python3
"""
Statistical Arbitrage Pair Selection Engine
Tìm các cặp cointegrated có potential arbitrage
"""
import pandas as pd
import numpy as np
from statsmodels.tsa.stattools import coint, adfuller
from itertools import combinations
class PairSelector:
"""
Engine chọn cặp giao dịch tối ưu dựa trên:
- Cointegration test
- Hurst exponent
- Half-life of mean reversion
- Correlation coefficient
"""
def __init__(self, min_correlation: float = 0.7,
p_value_threshold: float = 0.05):
self.min_corr = min_correlation
self.p_value_thresh = p_value_threshold
def find_cointegrated_pairs(self, prices: pd.DataFrame) -> list:
"""
Tìm tất cả các cặp cointegrated từ danh sách asset
Trả về list of tuples: (symbol1, symbol2, coint_stats)
"""
symbols = prices.columns.tolist()
pairs = []
print(f"Đang kiểm tra {len(symbols)} assets...")
for sym1, sym2 in combinations(symbols, 2):
# 1. Kiểm tra correlation trước
corr = prices[sym1].corr(prices[sym2])
if abs(corr) < self.min_corr:
continue
# 2. Kiểm tra cointegration
score, p_value, _ = coint(prices[sym1], prices[sym2])
if p_value < self.p_value_thresh:
# Tính additional metrics
spread = prices[sym1] - corr * prices[sym2]
hurst = self._calculate_hurst(spread)
half_life = self._calculate_half_life(spread)
pairs.append({
'symbol1': sym1,
'symbol2': sym2,
'correlation': corr,
'cointegration_pvalue': p_value,
'coint_score': score,
'hurst_exponent': hurst,
'half_life_days': half_life,
'spread_std': spread.std(),
'spread_mean': spread.mean()
})
# Sắp xếp theo độ mạnh cointegration
pairs.sort(key=lambda x: x['cointegration_pvalue'])
return pairs
def _calculate_hurst(self, spread: pd.Series, max_lag: int = 100) -> float:
"""
Tính Hurst Exponent - đo lường tính mean-reversion
H < 0.5: Mean-reverting
H = 0.5: Random walk
H > 0.5: Trending
"""
lags = range(2, min(max_lag, len(spread)//2))
tau = []
for lag in lags:
diff = spread.diff(lag).dropna()
tau.append(np.std(diff))
if len(tau) < 10 or any(t == 0 for t in tau):
return 0.5
poly = np.polyfit(np.log(list(lags)), np.log(tau), 1)
return float(poly[0])
def _calculate_half_life(self, spread: pd.Series) -> float:
"""
Tính half-life của spread - thời gian để spread quay về mean
Sử dụng Ornstein-Uhlenbeck process
"""
spread_lag = spread.shift(1).dropna()
spread_diff = spread.diff().dropna()
# Regress spread_diff on spread_lag
spread_diff = spread_diff.loc[spread_lag.index]
beta = np.polyfit(spread_lag, spread_diff, 1)[0]
alpha = -beta
if alpha <= 0:
return float('inf')
half_life = np.log(2) / alpha
return half_life
def rank_pairs(self, pairs: list) -> pd.DataFrame:
"""
Xếp hạng các cặp theo scoring function
Score = f(correlation, half_life, hurst, spread_volatility)
"""
if not pairs:
return pd.DataFrame()
df = pd.DataFrame(pairs)
# Scoring function
# Lower half-life = better (faster mean reversion)
# Hurst < 0.5 = better (mean reverting)
# Higher correlation = better
# Lower spread volatility = better
df['half_life_score'] = 1 / (1 + df['half_life_days'])
df['hurst_score'] = np.where(df['hurst_exponent'] < 0.5,
1 - df['hurst_exponent'],
df['hurst_exponent'])
df['corr_score'] = df['correlation'].abs()
df['vol_score'] = 1 / (1 + df['spread_std'])
df['total_score'] = (
df['half_life_score'] * 0.3 +
df['hurst_score'] * 0.25 +
df['corr_score'] * 0.25 +
df['vol_score'] * 0.2
)
return df.sort_values('total_score', ascending=False)
==================== DEMO ====================
if __name__ == "__main__":
# Tạo dữ liệu giả với một số cặp cointegrated
np.random.seed(42)
dates = pd.date_range('2024-01-01', '2025-01-01', freq='D')
# Cặp 1: Có cointegration (A và B)
base = 100 + np.cumsum(np.random.randn(len(dates)) * 0.01)
aapl = base + np.random.randn(len(dates)) * 2
msft = 1.2 * base + np.random.randn(len(dates)) * 2.5
# Cặp 2: Không có cointegration (C và D)
goog = 150 + np.cumsum(np.random.randn(len(dates)) * 0.02)
amzn = 80 + np.cumsum(np.random.randn(len(dates)) * 0.015)
# Random (E)
meta = 200 + np.cumsum(np.random.randn(len(dates)) * 0.018)
prices = pd.DataFrame({
'AAPL': aapl, 'MSFT': msft,
'GOOG': goog, 'AMZN': amzn,
'META': meta
}, index=dates)
# Tìm cặp
selector = PairSelector(min_correlation=0.6, p_value_threshold=0.1)
pairs = selector.find_cointegrated_pairs(prices)
if pairs:
ranked = selector.rank_pairs(pairs)
print("Top 5 cặp giao dịch:")
print(ranked[['symbol1', 'symbol2', 'correlation',
'half_life_days', 'hurst_exponent', 'total_score']].head())3. Trading Signal Generator - Tạo Tín Hiệu Giao Dịch
#!/usr/bin/env python3
"""
Statistical Arbitrage Trading Signal Generator
Sử dụng HolySheep AI để phân tích và tạo tín hiệu
"""
import pandas as pd
import numpy as np
import requests
import json
from typing import Dict, Tuple, Optional
from scipy import stats
class TradingSignalGenerator:
"""
Generator tín hiệu giao dịch Statistical Arbitrage
Sử dụng multiple signals và AI để filter
"""
HOLYSHEEP_API = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str,
z_entry: float = 2.0,
z_exit: float = 0.5,
lookback: int = 60):
self.api_key = api_key
self.z_entry = z_entry
self.z_exit = z_exit
self.lookback = lookback
self.session = requests.Session()
self.session.headers['Authorization'] = f'Bearer {api_key}'
def calculate_spread_zscore(self, prices: pd.DataFrame,
sym1: str, sym2: str,
hedge_ratio: float = None) -> pd.Series:
"""
Tính z-score của spread giữa 2 asset
"""
if hedge_ratio is None:
hedge_ratio = self._calculate_hedge_ratio(prices[sym1], prices[sym2])
spread = prices[sym1] - hedge_ratio * prices[sym2]
# Rolling z-score
mean = spread.rolling(window=self.lookback).mean()
std = spread.rolling(window=self.lookback).std()
zscore = (spread - mean) / std
return zscore
def _calculate_hedge_ratio(self, price1: pd.Series,
price2: pd.Series) -> float:
"""Tính hedge ratio bằng OLS"""
return np.polyfit(price1, price2, 1)[0]
def generate_signals(self, prices: pd.DataFrame,
sym1: str, sym2: str) -> pd.DataFrame:
"""
Tạo tín hiệu giao dịch cho cặp
Returns DataFrame với columns: signal, position, spread, zscore
"""
hedge_ratio = self._calculate_hedge_ratio(
prices[sym1], prices[sym2]
)
spread = prices[sym1] - hedge_ratio * prices[sym2]
zscore = self.calculate_spread_zscore(prices, sym1, sym2, hedge_ratio)
signals = pd.DataFrame(index=prices.index)
signals['spread'] = spread
signals['zscore'] = zscore
signals['hedge_ratio'] = hedge_ratio
# Signal logic:
# zscore > z_entry: Short spread (short sym1, long sym2)
# zscore < -z_entry: Long spread (long sym1, short sym2)
# |zscore| < z_exit: Exit
signals['raw_signal'] = 0
signals.loc[signals['zscore'] > self.z_entry, 'raw_signal'] = -1 # Short spread
signals.loc[signals['zscore'] < -self.z_entry, 'raw_signal'] = 1 # Long spread
# Position tracking (hold until exit signal)
signals['position'] = signals['raw_signal'].replace(0, np.nan)
signals['position'] = signals['position'].ffill().fillna(0)
# Exit signals
exit_long = (signals['position'] == 1) & (signals['zscore'] > -self.z_exit)
exit_short = (signals['position'] == -1) & (signals['zscore'] < self.z_exit)
signals.loc[exit_long | exit_short, 'position'] = 0
return signals
def get_ai_signal_filter(self, pair_data: Dict) -> Dict:
"""
Sử dụng HolySheep AI để filter tín hiệu
AI phân tích context thị trường và đưa ra recommendation
"""
prompt = f"""Bạn là chuyên gia Statistical Arbitrage. Phân tích:
Cặp giao dịch: {pair_data['sym1']} vs {pair_data['sym2']}
Z-score hiện tại: {pair_data['zscore']:.2f}
Spread: {pair_data['spread']:.2f}
Hedge ratio: {pair_data['hedge_ratio']:.4f}
Half-life: {pair_data.get('half_life', 'N/A')} ngày
Trả về JSON:
{{
"action": "LONG_SPREAD/SHORT_SPREAD/NO_POSITION",
"confidence": 0.0-1.0,
"risk_level": "low/medium/high",
"position_size": 0.0-1.0,
"stop_loss_zscore": float,
"reasoning": "Giải thích ngắn"
}}
"""
try:
response = self.session.post(
f"{self.HOLYSHEEP_API}/chat/completions",
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.2,
"max_tokens": 300
}
)
if response.status_code == 200:
content = response.json()['choices'][0]['message']['content']
return json.loads(content)
except Exception as e:
print(f"AI Filter Error: {e}")
return {"action": "NO_POSITION", "confidence": 0, "risk_level": "high"}
def backtest_signals(self, prices: pd.DataFrame,
signals: pd.DataFrame,
initial_capital: float = 100000,
commission: float = 0.001) -> Dict:
"""
Backtest chiến lược với signals đã tạo
"""
sym1, sym2 = prices.columns[0], prices.columns[1]
# Tính returns
ret1 = prices[sym1].pct_change()
ret2 = prices[sym2].pct_change()
# Spread return = ret1 - hedge_ratio * ret2
hedge_ratio = signals['hedge_ratio'].iloc[-1]
spread_returns = ret1 - hedge_ratio * ret2
# Strategy returns
signals['strategy_returns'] = signals['position'].shift(1) * spread_returns
# Trừ commission
trades = signals['position'].diff().abs()
signals['strategy_returns'] -= trades * commission
# Cumulative returns
signals['cumulative'] = (1 + signals['strategy_returns'].fillna(0)).cumprod()
signals['equity'] = initial_capital * signals['cumulative']
# Performance metrics
total_return = signals['cumulative'].iloc[-1] - 1
annual_return = (1 + total_return) ** (252/len(signals)) - 1
volatility = signals['strategy_returns'].std() * np.sqrt(252)
sharpe = annual_return / volatility if volatility > 0 else 0
# Max drawdown
rolling_max = signals['equity'].cummax()
drawdown = (signals['equity'] - rolling_max) / rolling_max
max_drawdown = drawdown.min()
return {
'total_return': total_return,
'annual_return': annual_return,
'volatility': volatility,
'sharpe_ratio': sharpe,
'max_drawdown': max_drawdown,
'num_trades': int(trades.sum() / 2),
'signals': signals
}
==================== SỬ DỤNG ====================
if __name__ == "__main__":
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
generator = TradingSignalGenerator(
api_key=API_KEY,
z_entry=2.0,
z_exit=0.5,
lookback=60
)
# Tạo dữ liệu demo
np.random.seed(123)
dates = pd.date_range('2024-01-01', '2025-06-01', freq='D')
price1 = 100 + np.cumsum(np.random.randn(len(dates)) * 0.5)
price2 = 80 + np.cumsum(np.random.randn(len(dates)) * 0.48)
prices = pd.DataFrame({'ASSET1': price1, 'ASSET2': price2}, index=dates)
# Generate signals
signals = generator.generate_signals(prices, 'ASSET1', 'ASSET2')
# Backtest
results = generator.backtest_signals(prices, signals)
print("=== BACKTEST RESULTS ===")
print(f"Total Return: {results['total_return']*100:.2f}%")
print(f"Annual Return: {results['annual_return']*100:.2f}%")
print(f"Sharpe Ratio: {results['sharpe_ratio']:.3f}")
print(f"Max Drawdown: {results['max_drawdown']*100:.2f}%")
print(f"Number of Trades: {results['num_trades']}")4. Risk Management System - Quản Lý Rủi Ro
#!/usr/bin/env python3
"""
Statistical Arbitrage Risk Management System
Quản lý rủi ro đa chiều với AI-powered risk assessment
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Optional
import requests
class RiskManager:
"""
Hệ thống quản lý rủi ro cho Statistical Arbitrage
Features:
- Position sizing thích ứng
- Drawdown protection
- Correlation monitoring
- AI-powered risk alerts
"""
HOLYSHEEP_API = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str,
max_portfolio_risk: float = 0.02, # 2% max risk per trade
max_correlation: float = 0.7,
max_drawdown: float = 0.15,
var_confidence: float = 0.95):
self.api_key = api_key
self.max_portfolio_risk = max_portfolio_risk
self.max_correlation = max_correlation
self.max_drawdown = max_drawdown
self.var_confidence = var_confidence
self.session = requests.Session()
self.session.headers['Authorization'] = f'Bearer {api_key}'
# Portfolio state
self.positions = {}
self.equity_curve = []
self.peak_equity = 100000
def calculate_position_size(self, pair_info: Dict,
account_equity: float,
volatility: float) -> float:
"""
Tính position size tối ưu sử dụng Kelly Criterion
và risk parity approach
"""
# Kelly fraction
win_rate = pair_info.get('win_rate', 0.55)
avg_win = pair_info.get('avg_win', 0.01)
avg_loss = pair_info.get('avg_loss', 0.008)
kelly = (win_rate * avg_win - (1 - win_rate) * avg_loss) / (avg_win + avg_loss)
kelly_fraction = max(0, min(kelly / 2, 0.25)) # Half-Kelly for safety
# Risk-based sizing
risk_amount = account_equity * self.max_portfolio_risk
volatility_adjusted_size = risk_amount / (volatility * 2) # 2 std dev stop
# Final size = minimum of Kelly and risk-based
position_size = min(kelly_fraction * account_equity,
volatility_adjusted_size,
account_equity * 0.1) # Max 10% per position
return position_size
def check_drawdown(self, current_equity: float) -> bool:
"""
Kiểm tra drawdown limit
Nếu vượt max_drawdown, dừng tất cả positions
"""
self.peak_equity = max(self.peak_equity, current_equity)
current_drawdown = (self.peak_equity - current_equity) / self.peak_equity
if current_drawdown > self.max_drawdown:
print(f"⚠️ ALERT: Drawdown {current_drawdown*100:.2f}% vượt limit!")
return True # Should close all positions
return False
def calculate_var(self, returns: pd.Series) -> float:
"""
Tính Value at Risk (VaR)
"""
if len(returns) < 30:
return 0.02 # Default 2%
return np.percentile(returns, (1 - self.var_confidence) * 100)
def get_risk_assessment(self, portfolio: Dict) -> Dict:
"""
Sử dụng HolySheep AI để đánh giá rủi ro portfolio
"""
prompt = f"""Đánh giá rủi ro portfolio Statistical Arbitrage:
Số lượng positions: {len(portfolio.get('positions', []))}
Tổng exposure: {portfolio.get('total_exposure', 0)*100:.1f}%
Portfolio volatility: {portfolio.get('volatility', 0)*100:.2f}%
Current drawdown: {portfolio.get('drawdown', 0)*100:.2f}%
Sharpe ratio hiện tại: {portfolio.get('sharpe', 0):.2f}
Trả về JSON:
{{
"risk_level": "low/medium/high/critical",
"recommendations": ["Khuyến nghị 1", "Khuyến nghị 2"],
"should_reduce_exposure": true/false,
"suggested_exposure_reduction": 0.0-1.0,
"alerts": ["Cảnh báo 1"]
}}
"""
try:
response = self.session.post(
f"{self.HOLYSHEEP_API}/chat/completions",
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 250
}
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content']
except Exception as e:
print(f"Risk AI Error: {e}")
return {"risk_level": "medium", "recommendations": []}
def dynamic_position_adjustment(self, zscore: float,
base_size: float,
market_regime: str = 'normal') -> float:
"""
Điều chỉnh position size theo zscore và market regime
"""
# Z-score adjustment
if abs(zscore) > 3:
# Extreme deviation - reduce size
zscore_multiplier = 0.5
elif abs(zscore) > 2.5:
zscore_multiplier = 0.75
else:
zscore_multiplier = 1.0
# Regime adjustment
regime_multipliers = {
'normal': 1.0,
'volatile': 0.7,
'crisis': 0.3,
'trending': 0.8
}
regime_multiplier = regime_multipliers.get(market_regime, 1.0)
return base_size * zscore_multiplier * regime_multiplier
def correlation_risk_check(self, new_pair_correlation: float,
existing_correlations: List[float]) -> bool:
"""
Kiểm tra correlation risk
Trả về True nếu được phép add position
"""
# Check max correlation with existing positions
if existing_correlations:
max_corr = max(existing_correlations)
if max_corr > self.max_correlation:
print(f"⚠️ Correlation với position hiện tại: {max_corr:.2f}")
return False
return True
def generate_risk_report(self, positions: List[Dict],
account_equity: float) -> pd.DataFrame:
"""
Tạo báo cáo rủi ro chi tiết
"""
report = []
for pos in positions:
exposure = pos['size'] * pos['price'] / account_equity
var = pos.get('var', 0.01)
report.append({
'pair': pos['pair'],
'size': pos['size'],
'exposure_pct': exposure * 100,
'entry_zscore': pos.get('entry_zscore', 0),
'current_zscore': pos.get('current_zscore', 0),
'unrealized_pnl': pos.get('pnl', 0),
'var_95': var * pos['size'],
'risk_score': exposure * 2 # Simplified
})
return pd.DataFrame(report).sort_values('risk_score', ascending=False)
==================== DEMO ====================
if __name__ == "__main__":
risk_manager = RiskManager(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_portfolio_risk=0.02,
max_drawdown=0.15
)
# Test position sizing
pair_info = {
'win_rate': 0.58,
'avg_win': 0.012,
'avg_loss': 0.008
}
size = risk_manager.calculate_position_size(
pair_info=pair_info,
account_equity=100000,
volatility=0.015
)
print(f"Recommended position size: ${size:,.2f}")
# Test drawdown check
should_stop = risk_manager.check_drawdown(92000)
print(f"Should stop trading: {should_stop}")
# Dynamic adjustment
adjusted_size = risk_manager.dynamic_position_adjustment(
zscore=2.8,
base_size=10000,
market_regime='volatile'
)
print(f"Adjusted size for zscore=2.8 in volatile regime: ${adjusted_size:,.2f}")Triển Khai Hoàn Chỉnh Với HolySheep AI
Điểm mấu chốt để chiến lược Stat Arb hiệu quả là