Phân tích chuỗi thời gian cho dữ liệu tài chính với Python
2024-03-20 — QuantTrade
Giới thiệu
Phân tích chuỗi thời gian là một lĩnh vực quan trọng trong phân tích dữ liệu tài chính, giúp hiểu các mẫu, xu hướng và mối quan hệ trong dữ liệu thay đổi theo thời gian. Dữ liệu tài chính, như giá cổ phiếu, tỷ giá hối đoái, hay lãi suất, đều có tính chất chuỗi thời gian rõ rệt. Việc phân tích hiệu quả các chuỗi thời gian này không chỉ giúp nắm bắt được diễn biến quá khứ mà còn có thể dự đoán xu hướng tương lai.
Python, với hệ sinh thái phong phú các thư viện phân tích dữ liệu, đã trở thành công cụ lý tưởng cho phân tích chuỗi thời gian trong lĩnh vực tài chính. Bài viết này sẽ giới thiệu các phương pháp, kỹ thuật và công cụ Python để phân tích chuỗi thời gian tài chính một cách hiệu quả.
I. Chuẩn bị dữ liệu chuỗi thời gian tài chính
1. Thu thập dữ liệu tài chính
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import yfinance as yf
from datetime import datetime, timedelta
# Thiết lập style cho biểu đồ
plt.style.use('seaborn-v0_8-darkgrid')
sns.set_palette("Set2")
# Định nghĩa khoảng thời gian
end_date = datetime.now()
start_date = end_date - timedelta(days=365*5) # 5 năm dữ liệu
# Tải dữ liệu giá cổ phiếu từ Yahoo Finance
ticker = 'AAPL'
data = yf.download(ticker, start=start_date, end=end_date)
# Kiểm tra dữ liệu
print(data.head())
print(f"Kích thước dữ liệu: {data.shape}")
2. Tiền xử lý dữ liệu chuỗi thời gian
# Kiểm tra giá trị thiếu
print(f"Số giá trị thiếu: \n{data.isnull().sum()}")
# Điền các giá trị thiếu (nếu có)
data = data.fillna(method='ffill') # Dùng giá trị trước đó để điền
# Tính lợi nhuận (returns)
data['Return'] = data['Adj Close'].pct_change()
# Tính lợi nhuận tích lũy
data['Cumulative Return'] = (1 + data['Return']).cumprod() - 1
# Tạo cột ngày, tháng, năm, thứ để phân tích mùa vụ
data['Year'] = data.index.year
data['Month'] = data.index.month
data['Day'] = data.index.day
data['Weekday'] = data.index.weekday # 0 là thứ Hai, 6 là Chủ nhật
# Kiểm tra dữ liệu sau khi xử lý
print(data.tail())
3. Tính toán các chỉ báo kỹ thuật cơ bản
import talib as ta
# Thêm các chỉ báo trung bình động (Moving Averages)
data['SMA_20'] = ta.SMA(data['Adj Close'], timeperiod=20)
data['SMA_50'] = ta.SMA(data['Adj Close'], timeperiod=50)
data['SMA_200'] = ta.SMA(data['Adj Close'], timeperiod=200)
data['EMA_20'] = ta.EMA(data['Adj Close'], timeperiod=20)
# Thêm dải Bollinger Band
data['Upper_Band'], data['Middle_Band'], data['Lower_Band'] = ta.BBANDS(
data['Adj Close'], timeperiod=20, nbdevup=2, nbdevdn=2, matype=0)
# Thêm các chỉ báo dao động (Oscillators)
data['RSI'] = ta.RSI(data['Adj Close'], timeperiod=14)
data['MACD'], data['MACD_Signal'], data['MACD_Hist'] = ta.MACD(
data['Adj Close'], fastperiod=12, slowperiod=26, signalperiod=9)
# Thêm chỉ báo khối lượng (Volume Indicators)
data['OBV'] = ta.OBV(data['Adj Close'], data['Volume'])
II. Phân tích thống kê chuỗi thời gian tài chính
1. Phân tích thống kê mô tả
# Thống kê mô tả cho chuỗi giá và lợi nhuận
print("\nThống kê mô tả cho giá đóng cửa:")
print(data['Adj Close'].describe())
print("\nThống kê mô tả cho lợi nhuận hàng ngày:")
print(data['Return'].describe())
# Tính các chỉ số rủi ro cơ bản
annualized_return = data['Return'].mean() * 252 # 252 ngày giao dịch/năm
annualized_vol = data['Return'].std() * np.sqrt(252)
sharpe_ratio = annualized_return / annualized_vol # Giả sử lãi suất phi rủi ro = 0
print(f"\nLợi nhuận hàng năm: {annualized_return:.4f}")
print(f"Biến động hàng năm: {annualized_vol:.4f}")
print(f"Tỷ lệ Sharpe: {sharpe_ratio:.4f}")
# Phân tích phân phối lợi nhuận
plt.figure(figsize=(12, 6))
sns.histplot(data['Return'].dropna(), kde=True, bins=50)
plt.title(f'Phân phối lợi nhuận hàng ngày của {ticker}')
plt.xlabel('Lợi nhuận hàng ngày')
plt.ylabel('Tần suất')
plt.axvline(x=0, color='r', linestyle='--')
plt.grid(True)
plt.show()
# Kiểm tra tính dừng (stationarity)
from statsmodels.tsa.stattools import adfuller
def test_stationarity(timeseries):
# Thực hiện kiểm định Dickey-Fuller mở rộng
result = adfuller(timeseries.dropna())
# In kết quả
print('Kết quả kiểm định Dickey-Fuller:')
print(f'Giá trị kiểm định ADF: {result[0]:.6f}')
print(f'P-value: {result[1]:.6f}')
print(f'Số lượng độ trễ sử dụng: {result[2]}')
print(f'Số lượng quan sát sử dụng cho kiểm định: {result[3]}')
for key, value in result[4].items():
print(f'Giá trị tới hạn ({key}): {value:.6f}')
# Kết luận về tính dừng
if result[1] <= 0.05:
print("Kết luận: Chuỗi thời gian có tính dừng (p-value <= 0.05)")
else:
print("Kết luận: Chuỗi thời gian không có tính dừng (p-value > 0.05)")
# Kiểm tra tính dừng của giá và lợi nhuận
print("\nKiểm tra tính dừng cho chuỗi giá:")
test_stationarity(data['Adj Close'])
print("\nKiểm tra tính dừng cho chuỗi lợi nhuận:")
test_stationarity(data['Return'])
2. Phân tích tương quan và tự tương quan
# Tính ma trận tương quan giữa các tài sản (nếu có nhiều tài sản)
tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'META']
multi_data = yf.download(tickers, start=start_date, end=end_date)['Adj Close']
returns = multi_data.pct_change().dropna()
# Tính và vẽ ma trận tương quan
correlation_matrix = returns.corr()
plt.figure(figsize=(10, 8))
sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm', linewidths=0.5)
plt.title('Ma trận tương quan giữa các cổ phiếu')
plt.tight_layout()
plt.show()
# Phân tích tự tương quan (autocorrelation)
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf
plt.figure(figsize=(12, 5))
plt.subplot(121)
plot_acf(data['Return'].dropna(), lags=40, ax=plt.gca())
plt.title('Hàm tự tương quan (ACF) của lợi nhuận')
plt.subplot(122)
plot_pacf(data['Return'].dropna(), lags=40, ax=plt.gca())
plt.title('Hàm tự tương quan riêng phần (PACF) của lợi nhuận')
plt.tight_layout()
plt.show()
# Kiểm tra tính ngẫu nhiên của lợi nhuận
from statsmodels.stats.diagnostic import acorr_ljungbox
lb_test = acorr_ljungbox(data['Return'].dropna(), lags=20)
print("\nKiểm định Ljung-Box cho chuỗi lợi nhuận:")
for i, p_value in enumerate(lb_test[1]):
print(f"Lag {i+1}: p-value = {p_value:.6f}")
# Kiểm tra tính "hiệu quả" của thị trường (theo nghĩa yếu)
market_efficiency = all(p > 0.05 for p in lb_test[1])
print(f"\nThị trường hiệu quả (theo nghĩa yếu): {market_efficiency}")
3. Phân tích mùa vụ và chu kỳ
# Phân tích biến động theo ngày trong tuần
weekday_returns = data.groupby('Weekday')['Return'].agg(['mean', 'std'])
weekday_returns.index = ['Thứ 2', 'Thứ 3', 'Thứ 4', 'Thứ 5', 'Thứ 6', 'Thứ 7', 'Chủ nhật']
weekday_returns['mean'] *= 100 # Chuyển sang phần trăm
weekday_returns['std'] *= 100 # Chuyển sang phần trăm
plt.figure(figsize=(12, 6))
weekday_returns['mean'].plot(kind='bar', color='blue', alpha=0.7)
plt.title(f'Lợi nhuận trung bình theo ngày trong tuần của {ticker}')
plt.ylabel('Lợi nhuận trung bình (%)')
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Phân tích biến động theo tháng
monthly_returns = data.groupby('Month')['Return'].agg(['mean', 'std'])
monthly_returns.index = ['T1', 'T2', 'T3', 'T4', 'T5', 'T6', 'T7', 'T8', 'T9', 'T10', 'T11', 'T12']
monthly_returns['mean'] *= 100 # Chuyển sang phần trăm
monthly_returns['std'] *= 100 # Chuyển sang phần trăm
plt.figure(figsize=(12, 6))
monthly_returns['mean'].plot(kind='bar', color='green', alpha=0.7)
plt.title(f'Lợi nhuận trung bình theo tháng của {ticker}')
plt.ylabel('Lợi nhuận trung bình (%)')
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# Phân tích "Hiệu ứng tháng Giêng" (January Effect)
january_effect = data.groupby('Month')['Return'].mean() * 100
is_january_effect = january_effect.iloc[0] > january_effect.mean()
print(f"Hiệu ứng tháng Giêng (lợi nhuận tháng 1 cao hơn trung bình): {is_january_effect}")
print(f"Lợi nhuận trung bình tháng 1: {january_effect.iloc[0]:.4f}%")
print(f"Lợi nhuận trung bình các tháng: {january_effect.mean():.4f}%")
III. Mô hình hóa chuỗi thời gian tài chính
1. Mô hình ARIMA (AutoRegressive Integrated Moving Average)
from statsmodels.tsa.arima.model import ARIMA
from statsmodels.tsa.stattools import arma_order_select_ic
# Tìm bậc tối ưu cho mô hình ARIMA
def find_optimal_arima_order(timeseries, max_p=5, max_d=2, max_q=5):
# Giả sử d = 1 cho chuỗi giá (thường thì chuỗi giá không dừng)
# Hoặc d = 0 cho chuỗi lợi nhuận (thường thì chuỗi lợi nhuận đã dừng)
best_aic = float('inf')
best_order = None
for p in range(max_p + 1):
for d in range(max_d + 1):
for q in range(max_q + 1):
try:
model = ARIMA(timeseries, order=(p, d, q))
results = model.fit()
if results.aic < best_aic:
best_aic = results.aic
best_order = (p, d, q)
except:
continue
return best_order, best_aic
# Tìm bậc tối ưu cho mô hình ARIMA
print("Đang tìm bậc tối ưu cho mô hình ARIMA (có thể mất một lúc)...")
# Sử dụng một phần dữ liệu để tìm bậc tối ưu (để giảm thời gian tính toán)
sample_data = data['Adj Close'][-500:]
optimal_order, aic = find_optimal_arima_order(sample_data, max_p=3, max_d=1, max_q=3)
print(f"Bậc tối ưu cho mô hình ARIMA: {optimal_order}, AIC: {aic:.4f}")
# Xây dựng mô hình ARIMA với bậc tối ưu
model = ARIMA(data['Adj Close'], order=optimal_order)
results = model.fit()
print(results.summary())
# Dự đoán giá trong tương lai
forecast_steps = 30 # Dự đoán 30 ngày tới
forecast = results.forecast(steps=forecast_steps)
forecast_index = pd.date_range(start=data.index[-1] + pd.Timedelta(days=1), periods=forecast_steps)
forecast_series = pd.Series(forecast, index=forecast_index)
# Vẽ biểu đồ kết quả
plt.figure(figsize=(12, 6))
plt.plot(data.index[-200:], data['Adj Close'][-200:], label='Giá thực tế')
plt.plot(forecast_index, forecast_series, color='red', label='Dự đoán')
plt.fill_between(forecast_index,
forecast_series - 1.96 * results.params[-1],
forecast_series + 1.96 * results.params[-1],
color='pink', alpha=0.3)
plt.title(f'Mô hình ARIMA{optimal_order} - Dự đoán giá {ticker}')
plt.xlabel('Ngày')
plt.ylabel('Giá ($)')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# Đánh giá mô hình
from sklearn.metrics import mean_squared_error, mean_absolute_error
# Chia dữ liệu thành tập huấn luyện và kiểm thử
train_size = int(len(data) * 0.8)
train, test = data['Adj Close'][:train_size], data['Adj Close'][train_size:]
# Xây dựng mô hình trên tập huấn luyện
model_train = ARIMA(train, order=optimal_order)
results_train = model_train.fit()
# Dự đoán trên tập kiểm thử
forecast_test = results_train.forecast(steps=len(test))
# Tính các chỉ số đánh giá
mse = mean_squared_error(test, forecast_test)
rmse = np.sqrt(mse)
mae = mean_absolute_error(test, forecast_test)
mape = np.mean(np.abs((test - forecast_test) / test)) * 100
print(f"MSE (Mean Squared Error): {mse:.4f}")
print(f"RMSE (Root Mean Squared Error): {rmse:.4f}")
print(f"MAE (Mean Absolute Error): {mae:.4f}")
print(f"MAPE (Mean Absolute Percentage Error): {mape:.4f}%")
# Vẽ biểu đồ kết quả dự đoán trên tập kiểm thử
plt.figure(figsize=(12, 6))
plt.plot(test.index, test, label='Giá thực tế')
plt.plot(test.index, forecast_test, color='red', label='Dự đoán')
plt.title(f'Đánh giá mô hình ARIMA{optimal_order} trên tập kiểm thử')
plt.xlabel('Ngày')
plt.ylabel('Giá ($)')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
2. Mô hình GARCH (Generalized AutoRegressive Conditional Heteroskedasticity)
from arch import arch_model
# Xây dựng mô hình GARCH cho chuỗi lợi nhuận
returns = data['Return'].dropna() * 100 # Chuyển sang phần trăm
# Mô hình GARCH(1,1)
garch_model = arch_model(returns, vol='Garch', p=1, q=1)
garch_results = garch_model.fit(disp='off')
print(garch_results.summary())
# Dự đoán biến động tương lai
forecast_horizon = 30
garch_forecast = garch_results.forecast(horizon=forecast_horizon)
forecast_vol = np.sqrt(garch_forecast.variance.iloc[-1].values)
# Vẽ biểu đồ biến động dự đoán
plt.figure(figsize=(12, 6))
plt.plot(forecast_vol, marker='o')
plt.title(f'Dự đoán biến động {forecast_horizon} ngày tới với mô hình GARCH(1,1)')
plt.xlabel('Ngày')
plt.ylabel('Biến động dự đoán (% / ngày)')
plt.grid(True)
plt.tight_layout()
plt.show()
# Phân tích biến động lịch sử
conditional_vol = np.sqrt(garch_results.conditional_volatility)
plt.figure(figsize=(12, 8))
plt.subplot(211)
plt.plot(returns.index, returns)
plt.title(f'Lợi nhuận hàng ngày của {ticker} (%)')
plt.ylabel('Lợi nhuận (%)')
plt.grid(True)
plt.subplot(212)
plt.plot(returns.index, conditional_vol)
plt.title('Biến động có điều kiện từ mô hình GARCH(1,1)')
plt.ylabel('Biến động (%)')
plt.grid(True)
plt.tight_layout()
plt.show()
# Vẽ biểu đồ QQ-plot để kiểm tra giả định phân phối của mô hình
from scipy import stats
import statsmodels.api as sm
standardized_residuals = garch_results.resid / garch_results.conditional_volatility
plt.figure(figsize=(10, 6))
sm.qqplot(standardized_residuals, line='s')
plt.title('Q-Q Plot cho phần dư chuẩn hóa của mô hình GARCH')
plt.grid(True)
plt.tight_layout()
plt.show()
3. Mô hình VAR (Vector AutoRegression) cho nhiều chuỗi thời gian
from statsmodels.tsa.api import VAR
# Sử dụng dữ liệu lợi nhuận của nhiều cổ phiếu
var_data = returns.copy()
var_data = var_data.dropna()
# Kiểm tra tính dừng của các chuỗi
for column in var_data.columns:
print(f"Kiểm tra tính dừng của {column}:")
test_stationarity(var_data[column])
print()
# Xây dựng mô hình VAR
var_model = VAR(var_data)
# Chọn độ trễ tối ưu
lag_order_results = var_model.select_order(maxlags=10)
print("Độ trễ tối ưu theo các tiêu chí thông tin:")
print(lag_order_results.summary())
lag_order = lag_order_results.aic # Sử dụng tiêu chí AIC
print(f"Độ trễ tối ưu theo AIC: {lag_order}")
# Xây dựng mô hình VAR với độ trễ tối ưu
var_result = var_model.fit(lag_order)
print(var_result.summary())
# Dự đoán tương lai
forecast_steps = 10
var_forecast = var_result.forecast(var_data.values, steps=forecast_steps)
forecast_index = pd.date_range(start=var_data.index[-1] + pd.Timedelta(days=1), periods=forecast_steps)
forecast_df = pd.DataFrame(data=var_forecast, index=forecast_index, columns=var_data.columns)
# Vẽ biểu đồ dự đoán
plt.figure(figsize=(14, 7))
for i, ticker in enumerate(var_data.columns):
plt.subplot(len(var_data.columns), 1, i+1)
plt.plot(var_data.index[-30:], var_data[ticker][-30:], label='Thực tế')
plt.plot(forecast_index, forecast_df[ticker], color='red', label='Dự đoán')
plt.title(f'Dự đoán lợi nhuận của {ticker}')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# Phân tích xung đáp (Impulse Response Analysis)
irf = var_result.irf(10) # Phân tích trong 10 kỳ
plt.figure(figsize=(14, 10))
irf.plot(orth=False)
plt.suptitle('Phân tích xung đáp (Impulse Response)')
plt.tight_layout()
plt.subplots_adjust(top=0.95)
plt.show()
# Phân tích phân rã phương sai (Variance Decomposition)
fevd = var_result.fevd(10)
fevd.plot()
plt.suptitle('Phân tích phân rã phương sai (Variance Decomposition)')
plt.tight_layout()
plt.subplots_adjust(top=0.95)
plt.show()
IV. Phân tích chuỗi thời gian nâng cao cho dữ liệu tài chính
1. Phát hiện điểm thay đổi (Change Point Detection)
import ruptures as rpt
# Phát hiện điểm thay đổi trong chuỗi giá
prices = data['Adj Close'].values
change_point_detector = rpt.Pelt(model="rbf").fit(prices)
change_points = change_point_detector.predict(pen=10)
# Vẽ biểu đồ với các điểm thay đổi
plt.figure(figsize=(14, 7))
plt.plot(data.index, data['Adj Close'])
for cp in change_points[:-1]: # Bỏ qua điểm cuối cùng
plt.axvline(x=data.index[cp], color='r', linestyle='--', alpha=0.5)
plt.title(f'Phát hiện điểm thay đổi trong giá {ticker}')
plt.xlabel('Ngày')
plt.ylabel('Giá ($)')
plt.grid(True)
plt.tight_layout()
plt.show()
# Phân tích các giai đoạn thị trường
segment_returns = []
segment_dates = []
for i in range(len(change_points) - 1):
start_idx = change_points[i]
end_idx = change_points[i+1]
segment_price = prices[start_idx:end_idx]
segment_return = (segment_price[-1] / segment_price[0] - 1) * 100
segment_dates.append((data.index[start_idx], data.index[end_idx-1]))
segment_returns.append(segment_return)
# In thông tin về các giai đoạn
print("\nPhân tích các giai đoạn thị trường:")
for i, ((start_date, end_date), segment_return) in enumerate(zip(segment_dates, segment_returns)):
print(f"Giai đoạn {i+1}: {start_date.date()} đến {end_date.date()}")
print(f" - Lợi nhuận: {segment_return:.2f}%")
print(f" - Thời gian: {(end_date - start_date).days} ngày")
if segment_return > 0:
print(" - Xu hướng: Tăng")
else:
print(" - Xu hướng: Giảm")
print()
2. Phân tích phổ chuỗi thời gian (Spectral Analysis)
from scipy import signal
# Phân tích phổ cho chuỗi lợi nhuận
returns = data['Return'].dropna().values
f, Pxx = signal.periodogram(returns, fs=1) # fs=1 ngày
plt.figure(figsize=(12, 6))
plt.semilogy(f, Pxx)
plt.xlabel('Tần số (1/ngày)')
plt.ylabel('Mật độ phổ công suất')
plt.title('Phân tích phổ chuỗi lợi nhuận')
plt.grid(True)
plt.tight_layout()
plt.show()
# Biến đổi wavelet liên tục (CWT)
from scipy import signal
import pywt
# Sử dụng CWT
N = len(returns)
scales = np.arange(1, 128)
wavelet = 'morl' # Morlet wavelet
dt = 1
# Tính CWT
coef, freqs = pywt.cwt(returns, scales, wavelet, dt)
power = (np.abs(coef)) ** 2
# Vẽ biểu đồ scalogram
plt.figure(figsize=(12, 8))
plt.imshow(power, aspect='auto', cmap='jet',
extent=[0, N, 1, len(scales)],
vmax=np.percentile(power, 95), vmin=0)
plt.colorbar(label='Công suất')
plt.ylabel('Scale')
plt.xlabel('Thời gian (ngày)')
plt.title('Phân tích Wavelet của chuỗi lợi nhuận')
plt.tight_layout()
plt.show()
3. Phân tích Long Memory và Hurst Exponent
def hurst_exponent(time_series, max_lag=100):
"""
Tính Hurst exponent của chuỗi thời gian
H > 0.5: có tính persistent (xu hướng dài)
H = 0.5: chuyển động Brown (random walk)
H < 0.5: có tính anti-persistent (xu hướng đảo chiều)
"""
lags = range(2, max_lag)
tau = [np.std(np.subtract(time_series[lag:], time_series[:-lag])) for lag in lags]
# Tính Hurst exponent
reg = np.polyfit(np.log(lags), np.log(tau), 1)
hurst = reg[0] / 2.0
return hurst, reg
# Tính Hurst exponent cho chuỗi giá và lợi nhuận
prices = data['Adj Close'].values
returns = data['Return'].dropna().values
hurst_price, reg_price = hurst_exponent(prices)
hurst_return, reg_return = hurst_exponent(returns)
print(f"Hurst exponent của chuỗi giá: {hurst_price:.4f}")
print(f"Hurst exponent của chuỗi lợi nhuận: {hurst_return:.4f}")
# Vẽ biểu đồ của log(R/S) và log(lag)
lags = range(2, 100)
tau_price = [np.std(np.subtract(prices[lag:], prices[:-lag])) for lag in lags]
tau_return = [np.std(np.subtract(returns[lag:], returns[:-lag])) for lag in lags]
plt.figure(figsize=(12, 6))
plt.loglog(lags, tau_price, 'o', markersize=5, label=f'Giá (H = {hurst_price:.4f})')
plt.loglog(lags, tau_return, 'o', markersize=5, label=f'Lợi nhuận (H = {hurst_return:.4f})')
# Thêm đường hồi quy
plt.loglog(lags, np.exp(reg_price[1]) * np.power(lags, reg_price[0]), 'r-', linewidth=2)
plt.loglog(lags, np.exp(reg_return[1]) * np.power(lags, reg_return[0]), 'g-', linewidth=2)
plt.legend()
plt.title('Phân tích Hurst Exponent')
plt.xlabel('Lag (log scale)')
plt.ylabel('Range/Standard deviation (log scale)')
plt.grid(True, which="both", ls="-")
plt.tight_layout()
plt.show()
# Giải thích ý nghĩa
print("\nGiải thích Hurst Exponent:")
if hurst_price > 0.5:
print(f"Chuỗi giá có tính persistent (H = {hurst_price:.4f} > 0.5)")
print("→ Xu hướng hiện tại có khả năng tiếp tục trong tương lai")
elif hurst_price < 0.5:
print(f"Chuỗi giá có tính anti-persistent (H = {hurst_price:.4f} < 0.5)")
print("→ Xu hướng hiện tại có khả năng đảo chiều trong tương lai")
else:
print(f"Chuỗi giá có tính chất chuyển động Brown (H = {hurst_price:.4f} ≈ 0.5)")
print("→ Chuỗi giá thể hiện tính ngẫu nhiên")
if hurst_return > 0.5:
print(f"\nChuỗi lợi nhuận có tính persistent (H = {hurst_return:.4f} > 0.5)")
print("→ Lợi nhuận dương/âm có khả năng tiếp tục trong tương lai")
elif hurst_return < 0.5:
print(f"\nChuỗi lợi nhuận có tính anti-persistent (H = {hurst_return:.4f} < 0.5)")
print("→ Lợi nhuận có khả năng thay đổi dấu trong tương lai")
else:
print(f"\nChuỗi lợi nhuận có tính chất chuyển động Brown (H = {hurst_return:.4f} ≈ 0.5)")
print("→ Chuỗi lợi nhuận thể hiện tính ngẫu nhiên")
V. Học máy cho phân tích chuỗi thời gian tài chính
1. Tiền xử lý dữ liệu và tạo đặc trưng
# Tạo bộ dữ liệu với các đặc trưng
def create_features(df, target_column='Adj Close', window=20, shift=1):
"""
Tạo các đặc trưng cho mô hình học máy từ chuỗi thời gian tài chính
Parameters:
-----------
df : pd.DataFrame
DataFrame chứa dữ liệu gốc
target_column : str
Tên cột chứa giá trị cần dự đoán
window : int
Độ dài cửa sổ dùng để tạo đặc trưng
shift : int
Số ngày dịch chuyển để dự đoán
Returns:
--------
X : pd.DataFrame
DataFrame chứa các đặc trưng
y : pd.Series
Series chứa giá trị cần dự đoán
"""
df = df.copy()
# Tạo mục tiêu
df['Target'] = df[target_column].shift(-shift)
# Tạo đặc trưng kỹ thuật
# 1. Lợi nhuận quá khứ
df['Return_1d'] = df[target_column].pct_change(1)
df['Return_5d'] = df[target_column].pct_change(5)
df['Return_10d'] = df[target_column].pct_change(10)
df['Return_20d'] = df[target_column].pct_change(20)
# 2. Giá tương đối với các MA
df['Price_to_MA20'] = df[target_column] / df['SMA_20']
df['Price_to_MA50'] = df[target_column] / df['SMA_50']
df['Price_to_MA200'] = df[target_column] / df['SMA_200']
# 3. Độ dốc của MA
df['MA20_slope'] = df['SMA_20'].pct_change(5)
df['MA50_slope'] = df['SMA_50'].pct_change(10)
# 4. Khoảng cách giữa các MA
df['MA20_50_spread'] = df['SMA_20'] / df['SMA_50'] - 1
df['MA50_200_spread'] = df['SMA_50'] / df['SMA_200'] - 1
# 5. Đặc trưng chuẩn hóa từ Bollinger Bands
df['Bollinger_width'] = (df['Upper_Band'] - df['Lower_Band']) / df['Middle_Band']
df['Bollinger_z'] = (df[target_column] - df['Middle_Band']) / ((df['Upper_Band'] - df['Lower_Band']) / 2)
# 6. Các đặc trưng từ RSI và MACD
df['RSI_scaled'] = df['RSI'] / 100 # Scale về 0-1
df['MACD_signal_diff'] = df['MACD'] - df['MACD_Signal']
df['MACD_direction'] = np.sign(df['MACD_signal_diff'])
# 7. Đặc trưng về biến động
df['Volatility_20d'] = df['Return_1d'].rolling(window=20).std()
# 8. Đặc trưng về khối lượng
df['Volume_pct_change'] = df['Volume'].pct_change()
df['Volume_to_MA20'] = df['Volume'] / df['Volume'].rolling(window=20).mean()
# 9. Các đặc trưng về thời gian
df['Day_of_week'] = df.index.dayofweek
df['Month'] = df.index.month
df['Quarter'] = df.index.quarter
# Chuyển các đặc trưng về dạng one-hot encoding
dow_dummies = pd.get_dummies(df['Day_of_week'], prefix='DOW')
month_dummies = pd.get_dummies(df['Month'], prefix='Month')
# Nối các đặc trưng vào DataFrame
df = pd.concat([df, dow_dummies, month_dummies], axis=1)
# Loại bỏ các dòng có giá trị NaN
df = df.dropna()
# Tạo X và y
features = [col for col in df.columns if col not in [target_column, 'Target', 'Day_of_week', 'Month', 'Quarter']]
X = df[features]
y = df['Target']
return X, y, df
# Tạo đặc trưng
X, y, processed_data = create_features(data, shift=5) # Dự đoán giá 5 ngày tới
# Chia dữ liệu thành tập huấn luyện và kiểm thử
from sklearn.model_selection import train_test_split
train_size = int(len(X) * 0.8)
X_train, X_test = X[:train_size], X[train_size:]
y_train, y_test = y[:train_size], y[train_size:]
print(f"Kích thước tập huấn luyện: {X_train.shape}")
print(f"Kích thước tập kiểm thử: {X_test.shape}")
# Chuẩn hóa dữ liệu
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
# Xem 10 đặc trưng quan trọng nhất
from sklearn.ensemble import RandomForestRegressor
feature_selector = RandomForestRegressor(n_estimators=100, random_state=42)
feature_selector.fit(X_train, y_train)
# Lấy độ quan trọng của đặc trưng
feature_importance = feature_selector.feature_importances_
features_df = pd.DataFrame({
'Feature': X.columns,
'Importance': feature_importance
}).sort_values('Importance', ascending=False)
print("\n10 đặc trưng quan trọng nhất:")
print(features_df.head(10))
# Vẽ biểu đồ độ quan trọng đặc trưng
plt.figure(figsize=(12, 8))
sns.barplot(x='Importance', y='Feature', data=features_df.head(20))
plt.title('Độ quan trọng của các đặc trưng')
plt.tight_layout()
plt.show()
2. Mô hình dự đoán giá với Random Forest
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
# Xây dựng mô hình Random Forest
rf_model = RandomForestRegressor(n_estimators=100, random_state=42)
rf_model.fit(X_train_scaled, y_train)
# Dự đoán trên tập kiểm thử
y_pred_rf = rf_model.predict(X_test_scaled)
# Đánh giá mô hình
mse_rf = mean_squared_error(y_test, y_pred_rf)
rmse_rf = np.sqrt(mse_rf)
mae_rf = mean_absolute_error(y_test, y_pred_rf)
r2_rf = r2_score(y_test, y_pred_rf)
print("\nKết quả đánh giá mô hình Random Forest:")
print(f"MSE: {mse_rf:.4f}")
print(f"RMSE: {rmse_rf:.4f}")
print(f"MAE: {mae_rf:.4f}")
print(f"R^2: {r2_rf:.4f}")
# Vẽ biểu đồ so sánh giá thực tế và dự đoán
plt.figure(figsize=(12, 6))
plt.plot(y_test.index, y_test.values, label='Giá thực tế')
plt.plot(y_test.index, y_pred_rf, color='red', label='Dự đoán (Random Forest)')
plt.title('So sánh giá thực tế và dự đoán với Random Forest')
plt.xlabel('Ngày')
plt.ylabel('Giá ($)')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# Vẽ biểu đồ scatter plot
plt.figure(figsize=(10, 6))
plt.scatter(y_test, y_pred_rf, alpha=0.5)
plt.plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()], 'r--')
plt.xlabel('Giá thực tế ($)')
plt.ylabel('Giá dự đoán ($)')
plt.title('Scatter Plot: Giá thực tế vs Dự đoán (Random Forest)')
plt.grid(True)
plt.tight_layout()
plt.show()
# Phân tích lỗi dự đoán
errors = y_test - y_pred_rf
plt.figure(figsize=(12, 6))
plt.subplot(121)
plt.hist(errors, bins=50)
plt.title('Phân phối lỗi dự đoán')
plt.xlabel('Lỗi ($)')
plt.ylabel('Tần suất')
plt.grid(True)
plt.subplot(122)
plt.scatter(y_pred_rf, errors, alpha=0.5)
plt.axhline(y=0, color='r', linestyle='--')
plt.title('Lỗi dự đoán vs Giá dự đoán')
plt.xlabel('Giá dự đoán ($)')
plt.ylabel('Lỗi ($)')
plt.grid(True)
plt.tight_layout()
plt.show()
3. Mô hình mạng nơ-ron LSTM cho dự đoán chuỗi thời gian
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from tensorflow.keras.callbacks import EarlyStopping
from sklearn.preprocessing import MinMaxScaler
# Chuẩn bị dữ liệu cho LSTM
def create_sequences(data, target, seq_length):
"""
Tạo chuỗi dữ liệu cho mô hình LSTM
Parameters:
-----------
data : np.array
Dữ liệu đặc trưng
target : np.array
Dữ liệu mục tiêu
seq_length : int
Độ dài chuỗi đầu vào
Returns:
--------
np.array, np.array
X và y cho mô hình LSTM
"""
X, y = [], []
for i in range(len(data) - seq_length):
X.append(data[i:i + seq_length])
y.append(target[i + seq_length])
return np.array(X), np.array(y)
# Chọn các đặc trưng quan trọng
top_features = features_df.head(10)['Feature'].values
X_selected = X[top_features]
# Chuẩn hóa dữ liệu
scaler_X = MinMaxScaler()
scaler_y = MinMaxScaler()
X_scaled = scaler_X.fit_transform(X_selected)
y_scaled = scaler_y.fit_transform(y.values.reshape(-1, 1)).flatten()
# Tạo chuỗi dữ liệu
seq_length = 30 # Độ dài chuỗi 30 ngày
X_seq, y_seq = create_sequences(X_scaled, y_scaled, seq_length)
# Chia dữ liệu thành tập huấn luyện và kiểm thử
train_size = int(len(X_seq) * 0.8)
X_train_seq, X_test_seq = X_seq[:train_size], X_seq[train_size:]
y_train_seq, y_test_seq = y_seq[:train_size], y_seq[train_size:]
print(f"Kích thước X_train cho LSTM: {X_train_seq.shape}")
print(f"Kích thước X_test cho LSTM: {X_test_seq.shape}")
# Xây dựng mô hình LSTM
def build_lstm_model(input_shape):
model = Sequential([
LSTM(50, return_sequences=True, input_shape=input_shape),
Dropout(0.2),
LSTM(50),
Dropout(0.2),
Dense(1)
])
model.compile(optimizer='adam', loss='mse')
return model
lstm_model = build_lstm_model((X_train_seq.shape[1], X_train_seq.shape[2]))
print(lstm_model.summary())
# Huấn luyện mô hình
early_stopping = EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True)
history = lstm_model.fit(
X_train_seq, y_train_seq,
epochs=100,
batch_size=32,
validation_split=0.2,
callbacks=[early_stopping],
verbose=1
)
# Vẽ biểu đồ quá trình huấn luyện
plt.figure(figsize=(10, 6))
plt.plot(history.history['loss'], label='Train Loss')
plt.plot(history.history['val_loss'], label='Validation Loss')
plt.title('Quá trình huấn luyện mô hình LSTM')
plt.xlabel('Epoch')
plt.ylabel('Loss (MSE)')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# Dự đoán trên tập kiểm thử
y_pred_lstm_scaled = lstm_model.predict(X_test_seq)
y_pred_lstm = scaler_y.inverse_transform(y_pred_lstm_scaled).flatten()
y_test_unscaled = scaler_y.inverse_transform(y_test_seq.reshape(-1, 1)).flatten()
# Đánh giá mô hình
mse_lstm = mean_squared_error(y_test_unscaled, y_pred_lstm)
rmse_lstm = np.sqrt(mse_lstm)
mae_lstm = mean_absolute_error(y_test_unscaled, y_pred_lstm)
r2_lstm = r2_score(y_test_unscaled, y_pred_lstm)
print("\nKết quả đánh giá mô hình LSTM:")
print(f"MSE: {mse_lstm:.4f}")
print(f"RMSE: {rmse_lstm:.4f}")
print(f"MAE: {mae_lstm:.4f}")
print(f"R^2: {r2_lstm:.4f}")
# Vẽ biểu đồ so sánh giá thực tế và dự đoán
test_dates = y.index[train_size+seq_length:]
plt.figure(figsize=(12, 6))
plt.plot(test_dates, y_test_unscaled, label='Giá thực tế')
plt.plot(test_dates, y_pred_lstm, color='red', label='Dự đoán (LSTM)')
plt.title('So sánh giá thực tế và dự đoán với LSTM')
plt.xlabel('Ngày')
plt.ylabel('Giá ($)')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# So sánh hiệu suất của Random Forest và LSTM
models = ['Random Forest', 'LSTM']
metrics = {
'MSE': [mse_rf, mse_lstm],
'RMSE': [rmse_rf, rmse_lstm],
'MAE': [mae_rf, mae_lstm],
'R²': [r2_rf, r2_lstm]
}
plt.figure(figsize=(14, 7))
for i, (metric_name, metric_values) in enumerate(metrics.items()):
plt.subplot(2, 2, i+1)
plt.bar(models, metric_values, color=['blue', 'green'])
plt.title(metric_name)
plt.ylabel('Giá trị')
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.suptitle('So sánh hiệu suất các mô hình', fontsize=16)
plt.subplots_adjust(top=0.9)
plt.show()
VI. Chiến lược giao dịch dựa trên phân tích chuỗi thời gian
1. Xây dựng chiến lược giao dịch cơ bản
# Tạo tín hiệu giao dịch dựa trên giao cắt MA
def generate_ma_signals(df, short_window=20, long_window=50):
"""
Tạo tín hiệu giao dịch dựa trên giao cắt MA
Parameters:
-----------
df : pd.DataFrame
DataFrame chứa dữ liệu giá
short_window : int
Độ dài cửa sổ MA ngắn
long_window : int
Độ dài cửa sổ MA dài
Returns:
--------
pd.DataFrame
DataFrame chứa tín hiệu giao dịch
"""
df = df.copy()
# Tính toán MA
df[f'SMA_{short_window}'] = df['Adj Close'].rolling(window=short_window).mean()
df[f'SMA_{long_window}'] = df['Adj Close'].rolling(window=long_window).mean()
# Tạo tín hiệu
df['Signal'] = 0
df['Signal'] = np.where(df[f'SMA_{short_window}'] > df[f'SMA_{long_window}'], 1, 0)
df['Position'] = df['Signal'].diff()
return df
# Tạo tín hiệu giao dịch dựa trên RSI
def generate_rsi_signals(df, rsi_window=14, overbought=70, oversold=30):
"""
Tạo tín hiệu giao dịch dựa trên RSI
Parameters:
-----------
df : pd.DataFrame
DataFrame chứa dữ liệu giá
rsi_window : int
Độ dài cửa sổ RSI
overbought : int
Ngưỡng quá mua
oversold : int
Ngưỡng quá bán
Returns:
--------
pd.DataFrame
DataFrame chứa tín hiệu giao dịch
"""
df = df.copy()
# Tính toán RSI
delta = df['Adj Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=rsi_window).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=rsi_window).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
# Tạo tín hiệu
df['Signal'] = 0
df.loc[df['RSI'] < oversold, 'Signal'] = 1 # Tín hiệu mua khi RSI < oversold
df.loc[df['RSI'] > overbought, 'Signal'] = -1 # Tín hiệu bán khi RSI > overbought
# Tạo vị thế (chỉ thay đổi khi tín hiệu thay đổi)
df['Position'] = 0
# Chỉ mua hoặc bán khi tín hiệu thay đổi
for i in range(1, len(df)):
if df['Signal'].iloc[i] == 1 and df['Signal'].iloc[i-1] != 1:
df['Position'].iloc[i] = 1 # Mua
elif df['Signal'].iloc[i] == -1 and df['Signal'].iloc[i-1] != -1:
df['Position'].iloc[i] = -1 # Bán
return df
# Tạo tín hiệu giao dịch dựa trên MACD
def generate_macd_signals(df, fast=12, slow=26, signal=9):
"""
Tạo tín hiệu giao dịch dựa trên MACD
Parameters:
-----------
df : pd.DataFrame
DataFrame chứa dữ liệu giá
fast : int
Độ dài cửa sổ EMA nhanh
slow : int
Độ dài cửa sổ EMA chậm
signal : int
Độ dài cửa sổ tín hiệu
Returns:
--------
pd.DataFrame
DataFrame chứa tín hiệu giao dịch
"""
df = df.copy()
# Tính MACD
ema_fast = df['Adj Close'].ewm(span=fast, adjust=False).mean()
ema_slow = df['Adj Close'].ewm(span=slow, adjust=False).mean()
df['MACD'] = ema_fast - ema_slow
df['MACD_Signal'] = df['MACD'].ewm(span=signal, adjust=False).mean()
df['MACD_Hist'] = df['MACD'] - df['MACD_Signal']
# Tạo tín hiệu
df['Signal'] = 0
df['Signal'] = np.where(df['MACD'] > df['MACD_Signal'], 1, 0)
df['Position'] = df['Signal'].diff()
return df
# Tính toán hiệu suất chiến lược
def calculate_strategy_returns(df, initial_capital=100000):
"""
Tính toán hiệu suất chiến lược giao dịch
Parameters:
-----------
df : pd.DataFrame
DataFrame chứa tín hiệu giao dịch
initial_capital : float
Vốn ban đầu
Returns:
--------
pd.DataFrame
DataFrame chứa hiệu suất chiến lược
"""
df = df.copy()
# Tính lợi nhuận hàng ngày
df['Market_Return'] = df['Adj Close'].pct_change()
# Tính lợi nhuận chiến lược (dựa trên vị thế ngày trước)
df['Strategy_Return'] = df['Market_Return'] * df['Signal'].shift(1)
# Loại bỏ các dòng có giá trị NaN
df = df.dropna()
# Tính lợi nhuận tích lũy
df['Market_Cumulative_Return'] = (1 + df['Market_Return']).cumprod()
df['Strategy_Cumulative_Return'] = (1 + df['Strategy_Return']).cumprod()
# Tính giá trị danh mục
df['Market_Value'] = initial_capital * df['Market_Cumulative_Return']
df['Strategy_Value'] = initial_capital * df['Strategy_Cumulative_Return']
return df
# Đánh giá chiến lược
def evaluate_strategy(df):
"""
Đánh giá hiệu suất chiến lược giao dịch
Parameters:
-----------
df : pd.DataFrame
DataFrame chứa hiệu suất chiến lược
Returns:
--------
dict
Dictionary chứa các chỉ số đánh giá
"""
# Tính các chỉ số hiệu suất
total_return = df['Strategy_Cumulative_Return'].iloc[-1] - 1
annual_return = total_return / (len(df) / 252)
daily_returns = df['Strategy_Return']
market_returns = df['Market_Return']
sharpe_ratio = np.sqrt(252) * daily_returns.mean() / daily_returns.std()
# Tính Maximum Drawdown
cumulative_returns = df['Strategy_Cumulative_Return']
running_max = cumulative_returns.cummax()
drawdown = (cumulative_returns - running_max) / running_max
max_drawdown = drawdown.min()
# Tính số lượng giao dịch
positions = df['Position']
trades = positions[positions != 0]
num_trades = len(trades)
# Tính tỷ lệ thắng/thua
trade_returns = []
current_position = 0
entry_price = 0
for i in range(len(df)):
# Mở vị thế mới
if df['Position'].iloc[i] == 1: # Tín hiệu mua
current_position = 1
entry_price = df['Adj Close'].iloc[i]
elif df['Position'].iloc[i] == -1: # Tín hiệu bán
if current_position == 1: # Đóng vị thế mua
exit_price = df['Adj Close'].iloc[i]
trade_return = (exit_price / entry_price) - 1
trade_returns.append(trade_return)
current_position = 0
win_rate = sum(1 for r in trade_returns if r > 0) / len(trade_returns) if trade_returns else 0
# Tính tương quan với thị trường
correlation = daily_returns.corr(market_returns)
# Tính Beta
covariance = daily_returns.cov(market_returns)
variance = market_returns.var()
beta = covariance / variance
return {
'Total Return': total_return,
'Annual Return': annual_return,
'Sharpe Ratio': sharpe_ratio,
'Max Drawdown': max_drawdown,
'Number of Trades': num_trades,
'Win Rate': win_rate,
'Correlation': correlation,
'Beta': beta
}
# Áp dụng và đánh giá các chiến lược
# 1. Chiến lược MA Crossover
ma_data = generate_ma_signals(data)
ma_strategy = calculate_strategy_returns(ma_data)
ma_metrics = evaluate_strategy(ma_strategy)
# 2. Chiến lược RSI
rsi_data = generate_rsi_signals(data)
rsi_strategy = calculate_strategy_returns(rsi_data)
rsi_metrics = evaluate_strategy(rsi_strategy)
# 3. Chiến lược MACD
macd_data = generate_macd_signals(data)
macd_strategy = calculate_strategy_returns(macd_data)
macd_metrics = evaluate_strategy(macd_strategy)
# So sánh hiệu suất các chiến lược
strategies = ['MA Crossover', 'RSI', 'MACD', 'Buy & Hold']
metrics_df = pd.DataFrame({
'MA Crossover': ma_metrics,
'RSI': rsi_metrics,
'MACD': macd_metrics,
'Buy & Hold': {
'Total Return': ma_strategy['Market_Cumulative_Return'].iloc[-1] - 1,
'Annual Return': (ma_strategy['Market_Cumulative_Return'].iloc[-1] - 1) / (len(ma_strategy) / 252),
'Sharpe Ratio': np.sqrt(252) * ma_strategy['Market_Return'].mean() / ma_strategy['Market_Return'].std(),
'Max Drawdown': (ma_strategy['Market_Cumulative_Return'] - ma_strategy['Market_Cumulative_Return'].cummax()) / ma_strategy['Market_Cumulative_Return'].cummax(),
'Number of Trades': 1,
'Win Rate': 1 if ma_strategy['Market_Cumulative_Return'].iloc[-1] > 1 else 0,
'Correlation': 1,
'Beta': 1
}
})
print("\nSo sánh hiệu suất các chiến lược:")
print(metrics_df.T)
# Vẽ biểu đồ so sánh lợi nhuận tích lũy
plt.figure(figsize=(12, 6))
plt.plot(ma_strategy.index, ma_strategy['Market_Cumulative_Return'], label='Buy & Hold')
plt.plot(ma_strategy.index, ma_strategy['Strategy_Cumulative_Return'], label='MA Crossover')
plt.plot(rsi_strategy.index, rsi_strategy['Strategy_Cumulative_Return'], label='RSI')
plt.plot(macd_strategy.index, macd_strategy['Strategy_Cumulative_Return'], label='MACD')
plt.title('So sánh lợi nhuận tích lũy của các chiến lược')
plt.xlabel('Ngày')
plt.ylabel('Lợi nhuận tích lũy')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# Vẽ biểu đồ so sánh các chỉ số quan trọng
metrics_to_plot = ['Total Return', 'Sharpe Ratio', 'Max Drawdown', 'Win Rate']
plt.figure(figsize=(14, 10))
for i, metric in enumerate(metrics_to_plot):
plt.subplot(2, 2, i+1)
values = [metrics_df[s][metric] for s in strategies]
if metric == 'Max Drawdown':
values = [abs(v) for v in values]
plt.ylabel('Giá trị (Giá trị tuyệt đối)')
else:
plt.ylabel('Giá trị')
plt.bar(strategies, values)
plt.title(metric)
plt.xticks(rotation=45)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
2. Kết hợp phân tích chuỗi thời gian với chiến lược giao dịch
# Kết hợp nhiều tín hiệu để tạo chiến lược giao dịch toàn diện
def create_combined_strategy(df, ma_short=20, ma_long=50, rsi_window=14,
rsi_oversold=30, rsi_overbought=70,
macd_fast=12, macd_slow=26, macd_signal=9):
"""
Tạo chiến lược kết hợp từ nhiều chỉ báo
Parameters:
-----------
df : pd.DataFrame
DataFrame chứa dữ liệu giá
ma_short, ma_long : int
Độ dài cửa sổ MA ngắn và dài
rsi_window, rsi_oversold, rsi_overbought : int
Tham số RSI
macd_fast, macd_slow, macd_signal : int
Tham số MACD
Returns:
--------
pd.DataFrame
DataFrame chứa tín hiệu giao dịch kết hợp
"""
df = df.copy()
# Tính các chỉ báo
# 1. MA
df[f'SMA_{ma_short}'] = df['Adj Close'].rolling(window=ma_short).mean()
df[f'SMA_{ma_long}'] = df['Adj Close'].rolling(window=ma_long).mean()
df['MA_Signal'] = np.where(df[f'SMA_{ma_short}'] > df[f'SMA_{ma_long}'], 1, -1)
# 2. RSI
delta = df['Adj Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=rsi_window).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=rsi_window).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
df['RSI_Signal'] = 0
df.loc[df['RSI'] < rsi_oversold, 'RSI_Signal'] = 1
df.loc[df['RSI'] > rsi_overbought, 'RSI_Signal'] = -1
# 3. MACD
ema_fast = df['Adj Close'].ewm(span=macd_fast, adjust=False).mean()
ema_slow = df['Adj Close'].ewm(span=macd_slow, adjust=False).mean()
df['MACD'] = ema_fast - ema_slow
df['MACD_Signal_Line'] = df['MACD'].ewm(span=macd_signal, adjust=False).mean()
df['MACD_Hist'] = df['MACD'] - df['MACD_Signal_Line']
df['MACD_Signal'] = np.where(df['MACD'] > df['MACD_Signal_Line'], 1, -1)
# Kết hợp các tín hiệu
df['Combined_Signal'] = 0
# Quy tắc vào lệnh:
# 1. Mua khi: MA Signal = 1 VÀ (RSI Signal = 1 HOẶC MACD Signal = 1)
buy_condition = (df['MA_Signal'] == 1) & ((df['RSI_Signal'] == 1) | (df['MACD_Signal'] == 1))
df.loc[buy_condition, 'Combined_Signal'] = 1
# 2. Bán khi: MA Signal = -1 VÀ (RSI Signal = -1 HOẶC MACD Signal = -1)
sell_condition = (df['MA_Signal'] == -1) & ((df['RSI_Signal'] == -1) | (df['MACD_Signal'] == -1))
df.loc[sell_condition, 'Combined_Signal'] = -1
# Tạo vị thế
df['Position'] = 0
current_position = 0
for i in range(len(df)):
if df['Combined_Signal'].iloc[i] == 1 and current_position <= 0:
df['Position'].iloc[i] = 1 # Mua
current_position = 1
elif df['Combined_Signal'].iloc[i] == -1 and current_position >= 0:
df['Position'].iloc[i] = -1 # Bán
current_position = -1
# Thêm stop loss và take profit
stop_loss_pct = 0.05 # 5%
take_profit_pct = 0.10 # 10%
entry_price = 0
has_position = False
for i in range(1, len(df)):
# Mở vị thế mới
if df['Position'].iloc[i-1] == 1: # Tín hiệu mua
entry_price = df['Adj Close'].iloc[i-1]
has_position = True
# Kiểm tra stop loss và take profit
if has_position:
current_price = df['Adj Close'].iloc[i]
# Stop loss
if current_price < entry_price * (1 - stop_loss_pct):
df['Position'].iloc[i] = -1 # Bán để cắt lỗ
has_position = False
# Take profit
elif current_price > entry_price * (1 + take_profit_pct):
df['Position'].iloc[i] = -1 # Bán để chốt lời
has_position = False
return df
# Áp dụng chiến lược kết hợp
combined_data = create_combined_strategy(data)
combined_strategy = calculate_strategy_returns(combined_data)
combined_metrics = evaluate_strategy(combined_strategy)
# So sánh với các chiến lược trước đó
metrics_df['Combined'] = combined_metrics
print("\nHiệu suất chiến lược kết hợp:")
print(metrics_df[['Combined', 'Buy & Hold']].T)
# Vẽ biểu đồ so sánh
plt.figure(figsize=(12, 6))
plt.plot(combined_strategy.index, combined_strategy['Market_Cumulative_Return'], label='Buy & Hold')
plt.plot(combined_strategy.index, combined_strategy['Strategy_Cumulative_Return'], label='Combined Strategy')
plt.title('Lợi nhuận tích lũy của chiến lược kết hợp')
plt.xlabel('Ngày')
plt.ylabel('Lợi nhuận tích lũy')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# Vẽ biểu đồ giá với tín hiệu giao dịch
plt.figure(figsize=(14, 8))
# Biểu đồ giá và MA
plt.subplot(211)
plt.plot(combined_data.index, combined_data['Adj Close'], label='Giá')
plt.plot(combined_data.index, combined_data[f'SMA_{20}'], label='SMA 20')
plt.plot(combined_data.index, combined_data[f'SMA_{50}'], label='SMA 50')
# Đánh dấu các điểm mua/bán
buy_signals = combined_data[combined_data['Position'] == 1]
sell_signals = combined_data[combined_data['Position'] == -1]
plt.scatter(buy_signals.index, buy_signals['Adj Close'], marker='^', color='green', s=100, label='Mua')
plt.scatter(sell_signals.index, sell_signals['Adj Close'], marker='v', color='red', s=100, label='Bán')
plt.title('Chiến lược giao dịch kết hợp')
plt.ylabel('Giá ($)')
plt.legend()
plt.grid(True)
# Biểu đồ chỉ báo RSI và MACD
plt.subplot(212)
plt.plot(combined_data.index, combined_data['RSI'], label='RSI')
plt.axhline(y=70, color='r', linestyle='--', alpha=0.5)
plt.axhline(y=30, color='g', linestyle='--', alpha=0.5)
plt.fill_between(combined_data.index, 70, 100, color='red', alpha=0.1)
plt.fill_between(combined_data.index, 0, 30, color='green', alpha=0.1)
plt.title('RSI và tín hiệu giao dịch')
plt.ylabel('RSI')
plt.xlabel('Ngày')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
# Vẽ biểu đồ MACD
plt.figure(figsize=(14, 6))
plt.subplot(111)
plt.plot(combined_data.index, combined_data['MACD'], label='MACD')
plt.plot(combined_data.index, combined_data['MACD_Signal_Line'], label='Signal Line')
plt.bar(combined_data.index, combined_data['MACD_Hist'],
color=np.where(combined_data['MACD_Hist'] > 0, 'g', 'r'), alpha=0.5, label='Histogram')
plt.title('MACD và tín hiệu giao dịch')
plt.ylabel('MACD')
plt.xlabel('Ngày')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
VII. Kết luận và các bước tiếp theo
Phân tích chuỗi thời gian cho dữ liệu tài chính là một lĩnh vực rộng lớn với nhiều phương pháp và kỹ thuật khác nhau. Trong bài viết này, chúng ta đã khám phá từ các phương pháp phân tích cơ bản đến các kỹ thuật nâng cao sử dụng Python.
Các kết quả chính từ phân tích của chúng ta:
Đặc trưng thống kê: Dữ liệu tài chính thường không tuân theo phân phối chuẩn và có đuôi dày (fat tails), điều này ảnh hưởng đến việc đánh giá rủi ro.
Tính dừng: Chuỗi giá thường không dừng (non-stationary) trong khi chuỗi lợi nhuận thường dừng (stationary), điều này ảnh hưởng đến việc lựa chọn mô hình phù hợp.
Mùa vụ và chu kỳ: Dữ liệu tài chính thường có tính mùa vụ và chu kỳ, như hiệu ứng tháng Giêng hay biến động theo ngày trong tuần.
Mô hình dự đoán: Các mô hình như ARIMA, GARCH và các mô hình học máy đều có thể áp dụng để dự đoán xu hướng thị trường, nhưng mỗi mô hình có ưu và nhược điểm riêng.
Chiến lược giao dịch: Việc kết hợp nhiều chỉ báo và phương pháp phân tích có thể tạo ra chiến lược giao dịch hiệu quả hơn so với việc chỉ dựa vào một chỉ báo.
Các bước tiếp theo
Để tiếp tục phát triển kỹ năng phân tích chuỗi thời gian tài chính, bạn có thể:
Mở rộng nguồn dữ liệu: Kết hợp dữ liệu giá và khối lượng với dữ liệu từ các nguồn khác như tin tức, mạng xã hội, hay dữ liệu vĩ mô.
Thử nghiệm các mô hình nâng cao: Khám phá các mô hình nâng cao hơn như mô hình Hidden Markov, LSTM đa chiều, hay các mô hình Deep Learning phức tạp hơn.
Áp dụng vào các loại tài sản khác: Mở rộng phân tích sang các loại tài sản khác như tiền điện tử, hàng hóa, hay trái phiếu.
Phát triển hệ thống giao dịch tự động: Tự động hóa chiến lược giao dịch và tích hợp với API của sàn giao dịch.
Áp dụng kỹ thuật tối ưu hóa danh mục đầu tư: Kết hợp phân tích chuỗi thời gian với lý thuyết danh mục hiện đại để xây dựng danh mục đầu tư tối ưu.
Nhớ rằng, dù phân tích kỹ thuật và mô hình hóa chuỗi thời gian có thể cung cấp những hiểu biết hữu ích, chúng không thể dự đoán chính xác tương lai của thị trường tài chính. Luôn kết hợp phân tích kỹ thuật với phân tích cơ bản và quản lý rủi ro thận trọng để đạt được kết quả đầu tư tốt nhất.