Danh mục: Tin tức
Bài viết gần đây
| NumPy Được Sử Dụng Trong Phân Tích Dữ Liệu Tài Chính
Được viết bởi thanhdt vào ngày 15/11/2025 lúc 20:00 | 158 lượt xem
Hướng dẫn chi tiết cách sử dụng NumPy trong phân tích dữ liệu tài chính với các ví dụ thực tế về tính toán giá, chỉ báo kỹ thuật và phân tích rủi ro.
NumPy Được Sử Dụng Trong Phân Tích Dữ Liệu Tài Chính
NumPy (Numerical Python) là thư viện cơ bản và quan trọng nhất trong phân tích dữ liệu tài chính với Python. Bài viết này sẽ hướng dẫn bạn cách sử dụng NumPy để phân tích dữ liệu tài chính một cách hiệu quả.
import numpy as npTại Sao Sử Dụng NumPy?
NumPy cung cấp:
- Hiệu suất cao: Tính toán nhanh hơn Python thuần 10-100 lần
- Mảng đa chiều: Xử lý dữ liệu tài chính dễ dàng
- Hàm toán học phong phú: Từ cơ bản đến nâng cao
- Tích hợp tốt: Làm nền tảng cho Pandas, SciPy, Matplotlib
- Tối ưu bộ nhớ: Xử lý dữ liệu lớn hiệu quả
1. Tạo và Xử Lý Mảng Dữ Liệu Giá
Tạo Mảng Giá Từ Dữ Liệu
import numpy as np
# Dữ liệu giá đóng cửa (Close prices)
prices = np.array([100, 102, 101, 105, 103, 107, 106, 108, 110, 109])
print("Giá đóng cửa:", prices)
print("Kiểu dữ liệu:", prices.dtype)
print("Kích thước:", prices.shape)
# Tạo mảng 2D cho OHLC (Open, High, Low, Close)
ohlc_data = np.array([
[100, 103, 99, 102], # Ngày 1
[102, 104, 101, 101], # Ngày 2
[101, 105, 100, 105], # Ngày 3
[105, 107, 104, 103], # Ngày 4
[103, 108, 102, 107] # Ngày 5
])
print("\nDữ liệu OHLC:")
print(ohlc_data)
print("Kích thước:", ohlc_data.shape) # (5, 4) - 5 ngày, 4 giá trị mỗi ngàyTruy Cập Dữ Liệu
# Lấy cột giá đóng cửa (cột cuối cùng)
close_prices = ohlc_data[:, -1]
print("Giá đóng cửa:", close_prices)
# Lấy giá cao nhất
high_prices = ohlc_data[:, 1]
print("Giá cao nhất:", high_prices)
# Lấy giá thấp nhất
low_prices = ohlc_data[:, 2]
print("Giá thấp nhất:", low_prices)2. Tính Toán Thay Đổi Giá
Tính Phần Trăm Thay Đổi
# Tính phần trăm thay đổi giá
price_changes = np.diff(prices) / prices[:-1] * 100
print("Thay đổi giá (%):", price_changes)
# Tính log return (thường dùng trong tài chính)
log_returns = np.diff(np.log(prices)) * 100
print("Log returns (%):", log_returns)
# Tính tổng lợi nhuận tích lũy
cumulative_returns = np.cumsum(log_returns)
print("Lợi nhuận tích lũy (%):", cumulative_returns)Tính Giá Trung Bình và Độ Lệch
# Giá trung bình
mean_price = np.mean(prices)
print(f"Giá trung bình: ${mean_price:.2f}")
# Giá trung vị
median_price = np.median(prices)
print(f"Giá trung vị: ${median_price:.2f}")
# Độ lệch chuẩn
std_price = np.std(prices)
print(f"Độ lệch chuẩn: ${std_price:.2f}")
# Phương sai
variance = np.var(prices)
print(f"Phương sai: ${variance:.2f}")3. Tính Toán Chỉ Báo Kỹ Thuật
Moving Average (Trung Bình Động)
def moving_average(prices, window):
"""Tính trung bình động"""
return np.convolve(prices, np.ones(window)/window, mode='valid')
# Tính SMA 5 ngày
sma_5 = moving_average(prices, 5)
print("SMA 5 ngày:", sma_5)
# Tính SMA 10 ngày
sma_10 = moving_average(prices, 10)
print("SMA 10 ngày:", sma_10)
# Cách khác: Sử dụng cumsum
def moving_average_cumsum(prices, window):
"""Tính MA bằng cumsum (nhanh hơn)"""
cumsum = np.cumsum(prices)
cumsum[window:] = cumsum[window:] - cumsum[:-window]
return cumsum[window - 1:] / window
sma_fast = moving_average_cumsum(prices, 5)
print("SMA 5 (cách nhanh):", sma_fast)Exponential Moving Average (EMA)
def exponential_moving_average(prices, alpha):
"""
Tính Exponential Moving Average
Args:
prices: Mảng giá
alpha: Hệ số làm mịn (0 < alpha <= 1)
"""
ema = np.zeros_like(prices)
ema[0] = prices[0]
for i in range(1, len(prices)):
ema[i] = alpha * prices[i] + (1 - alpha) * ema[i-1]
return ema
# Tính EMA với alpha = 0.2 (tương đương 9 ngày)
ema_9 = exponential_moving_average(prices, 0.2)
print("EMA 9 ngày:", ema_9)RSI (Relative Strength Index)
def calculate_rsi(prices, period=14):
"""
Tính RSI (Relative Strength Index)
Args:
prices: Mảng giá
period: Chu kỳ (mặc định 14)
"""
# Tính thay đổi giá
deltas = np.diff(prices)
# Tách lợi nhuận và thua lỗ
gains = np.where(deltas > 0, deltas, 0)
losses = np.where(deltas < 0, -deltas, 0)
# Tính trung bình lợi nhuận và thua lỗ
avg_gains = moving_average_cumsum(gains, period)
avg_losses = moving_average_cumsum(losses, period)
# Tránh chia cho 0
rs = np.where(avg_losses != 0, avg_gains / avg_losses, 0)
rsi = 100 - (100 / (1 + rs))
return rsi
# Tính RSI 14 ngày
rsi = calculate_rsi(prices, period=14)
print("RSI:", rsi)MACD (Moving Average Convergence Divergence)
def calculate_macd(prices, fast_period=12, slow_period=26, signal_period=9):
"""
Tính MACD
Args:
prices: Mảng giá
fast_period: Chu kỳ EMA nhanh (mặc định 12)
slow_period: Chu kỳ EMA chậm (mặc định 26)
signal_period: Chu kỳ signal line (mặc định 9)
"""
# Tính alpha cho EMA
fast_alpha = 2 / (fast_period + 1)
slow_alpha = 2 / (slow_period + 1)
signal_alpha = 2 / (signal_period + 1)
# Tính EMA nhanh và chậm
ema_fast = exponential_moving_average(prices, fast_alpha)
ema_slow = exponential_moving_average(prices, slow_alpha)
# MACD line
macd_line = ema_fast - ema_slow
# Signal line (EMA của MACD line)
signal_line = exponential_moving_average(macd_line, signal_alpha)
# Histogram
histogram = macd_line - signal_line
return macd_line, signal_line, histogram
# Tính MACD
macd, signal, hist = calculate_macd(prices)
print("MACD line:", macd[-5:])
print("Signal line:", signal[-5:])
print("Histogram:", hist[-5:])Bollinger Bands
def bollinger_bands(prices, window=20, num_std=2):
"""
Tính Bollinger Bands
Args:
prices: Mảng giá
window: Cửa sổ (mặc định 20)
num_std: Số độ lệch chuẩn (mặc định 2)
"""
# Tính SMA
sma = moving_average_cumsum(prices, window)
# Tính độ lệch chuẩn
std = np.zeros(len(sma))
for i in range(len(sma)):
std[i] = np.std(prices[i:i+window])
# Upper và Lower bands
upper_band = sma + (num_std * std)
lower_band = sma - (num_std * std)
return sma, upper_band, lower_band
# Tính Bollinger Bands
middle, upper, lower = bollinger_bands(prices, window=5, num_std=2)
print("Middle band (SMA):", middle)
print("Upper band:", upper)
print("Lower band:", lower)4. Phân Tích Rủi Ro
Tính Toán Volatility (Độ Biến Động)
# Tính log returns
log_returns = np.diff(np.log(prices))
# Volatility (độ lệch chuẩn của log returns)
volatility = np.std(log_returns) * np.sqrt(252) # Annualized (252 ngày giao dịch)
print(f"Volatility hàng năm: {volatility * 100:.2f}%")
# Volatility theo ngày
daily_volatility = np.std(log_returns)
print(f"Volatility hàng ngày: {daily_volatility * 100:.2f}%")Value at Risk (VaR)
def calculate_var(returns, confidence_level=0.05):
"""
Tính Value at Risk (VaR)
Args:
returns: Mảng lợi nhuận
confidence_level: Mức tin cậy (mặc định 5%)
"""
# VaR = percentile của phân phối lợi nhuận
var = np.percentile(returns, confidence_level * 100)
return var
# Tính VaR 5%
returns = np.diff(prices) / prices[:-1]
var_5 = calculate_var(returns, 0.05)
print(f"VaR 5%: {var_5 * 100:.2f}%")Maximum Drawdown
def maximum_drawdown(prices):
"""
Tính Maximum Drawdown
Args:
prices: Mảng giá
"""
# Tính cumulative returns
cumulative = np.cumprod(1 + np.diff(prices) / prices[:-1])
# Tính running maximum
running_max = np.maximum.accumulate(cumulative)
# Tính drawdown
drawdown = (cumulative - running_max) / running_max
# Maximum drawdown
max_dd = np.min(drawdown)
return max_dd, drawdown
# Tính Maximum Drawdown
max_dd, dd = maximum_drawdown(prices)
print(f"Maximum Drawdown: {max_dd * 100:.2f}%")5. Phân Tích Tương Quan
Ma Trận Tương Quan
# Dữ liệu giá của nhiều cổ phiếu/coin
stock1_prices = np.array([100, 102, 101, 105, 103])
stock2_prices = np.array([50, 51, 50, 52, 51])
stock3_prices = np.array([200, 201, 200, 203, 202])
# Tính log returns
returns1 = np.diff(np.log(stock1_prices))
returns2 = np.diff(np.log(stock2_prices))
returns3 = np.diff(np.log(stock3_prices))
# Tạo ma trận returns
returns_matrix = np.column_stack([returns1, returns2, returns3])
# Tính ma trận tương quan
correlation_matrix = np.corrcoef(returns_matrix.T)
print("Ma trận tương quan:")
print(correlation_matrix)
# Tính ma trận hiệp phương sai
covariance_matrix = np.cov(returns_matrix.T)
print("\nMa trận hiệp phương sai:")
print(covariance_matrix)6. Tối Ưu Hóa Danh Mục Đầu Tư
Tính Lợi Nhuận Kỳ Vọng và Rủi Ro
# Lợi nhuận kỳ vọng (trung bình)
expected_returns = np.mean(returns_matrix, axis=0)
print("Lợi nhuận kỳ vọng:", expected_returns)
# Rủi ro (độ lệch chuẩn)
risks = np.std(returns_matrix, axis=0)
print("Rủi ro:", risks)
# Sharpe Ratio (giả sử risk-free rate = 0)
sharpe_ratios = expected_returns / risks
print("Sharpe Ratio:", sharpe_ratios)Tối Ưu Hóa Trọng Số Danh Mục
def portfolio_optimization(expected_returns, covariance_matrix, risk_aversion=1.0):
"""
Tối ưu hóa danh mục đầu tư (Markowitz)
Args:
expected_returns: Lợi nhuận kỳ vọng
covariance_matrix: Ma trận hiệp phương sai
risk_aversion: Hệ số ngại rủi ro
"""
# Số lượng tài sản
n = len(expected_returns)
# Ma trận nghịch đảo
inv_cov = np.linalg.inv(covariance_matrix)
# Vector ones
ones = np.ones(n)
# Tính trọng số tối ưu
# w = (1/λ) * inv(Σ) * μ
# Với λ là risk aversion
optimal_weights = (1 / risk_aversion) * inv_cov @ expected_returns
# Chuẩn hóa để tổng = 1
optimal_weights = optimal_weights / np.sum(optimal_weights)
return optimal_weights
# Tối ưu hóa danh mục
weights = portfolio_optimization(expected_returns, covariance_matrix, risk_aversion=1.0)
print("Trọng số tối ưu:", weights)
print("Tổng trọng số:", np.sum(weights))7. Xử Lý Dữ Liệu Thiếu và Ngoại Lai
Xử Lý Dữ Liệu Thiếu
# Dữ liệu có giá trị NaN
prices_with_nan = np.array([100, 102, np.nan, 105, 103, np.nan, 107])
# Kiểm tra NaN
has_nan = np.isnan(prices_with_nan)
print("Có NaN:", has_nan)
# Thay thế NaN bằng giá trị trung bình
mean_price = np.nanmean(prices_with_nan)
prices_filled = np.where(np.isnan(prices_with_nan), mean_price, prices_with_nan)
print("Sau khi điền:", prices_filled)
# Hoặc forward fill
prices_ffill = np.array(prices_with_nan)
for i in range(1, len(prices_ffill)):
if np.isnan(prices_ffill[i]):
prices_ffill[i] = prices_ffill[i-1]
print("Forward fill:", prices_ffill)Phát Hiện và Xử Lý Ngoại Lai
def detect_outliers(prices, threshold=3):
"""
Phát hiện ngoại lai sử dụng Z-score
Args:
prices: Mảng giá
threshold: Ngưỡng Z-score (mặc định 3)
"""
# Tính Z-score
mean = np.mean(prices)
std = np.std(prices)
z_scores = np.abs((prices - mean) / std)
# Xác định ngoại lai
outliers = z_scores > threshold
return outliers, z_scores
# Phát hiện ngoại lai
prices_test = np.array([100, 102, 101, 500, 103, 107, 106]) # 500 là ngoại lai
outliers, z_scores = detect_outliers(prices_test, threshold=2)
print("Ngoại lai:", outliers)
print("Z-scores:", z_scores)
# Loại bỏ ngoại lai
prices_clean = prices_test[~outliers]
print("Sau khi loại bỏ:", prices_clean)8. Tính Toán Hiệu Suất
Tính Các Chỉ Số Hiệu Suất
def calculate_performance_metrics(prices):
"""
Tính các chỉ số hiệu suất
Args:
prices: Mảng giá
"""
# Tính returns
returns = np.diff(prices) / prices[:-1]
# Tổng lợi nhuận
total_return = (prices[-1] - prices[0]) / prices[0]
# Lợi nhuận trung bình
mean_return = np.mean(returns)
# Volatility
volatility = np.std(returns)
# Sharpe Ratio (giả sử risk-free = 0)
sharpe = mean_return / volatility if volatility > 0 else 0
# Maximum Drawdown
max_dd, _ = maximum_drawdown(prices)
# Win rate
win_rate = np.sum(returns > 0) / len(returns)
return {
'total_return': total_return,
'mean_return': mean_return,
'volatility': volatility,
'sharpe_ratio': sharpe,
'max_drawdown': max_dd,
'win_rate': win_rate
}
# Tính hiệu suất
metrics = calculate_performance_metrics(prices)
print("Chỉ số hiệu suất:")
for key, value in metrics.items():
if isinstance(value, float):
print(f" {key}: {value * 100:.2f}%")
else:
print(f" {key}: {value:.4f}")9. Tối Ưu Hóa Hiệu Suất
Vectorization vs Loops
import time
# Dữ liệu lớn
large_prices = np.random.randn(100000) * 10 + 100
# Cách 1: Sử dụng loop (chậm)
start = time.time()
result_loop = []
for i in range(len(large_prices) - 1):
result_loop.append(large_prices[i+1] - large_prices[i])
time_loop = time.time() - start
# Cách 2: Sử dụng NumPy vectorization (nhanh)
start = time.time()
result_vectorized = np.diff(large_prices)
time_vectorized = time.time() - start
print(f"Loop: {time_loop:.4f} giây")
print(f"Vectorized: {time_vectorized:.4f} giây")
print(f"Tăng tốc: {time_loop / time_vectorized:.1f}x")10. Ví Dụ Thực Tế: Phân Tích Dữ Liệu Crypto
import numpy as np
# Giả sử dữ liệu giá BTC trong 30 ngày
btc_prices = np.array([
45000, 45200, 44800, 45500, 46000, 45800, 46200, 46500,
46300, 46800, 47000, 47200, 47500, 47300, 47800, 48000,
48200, 48500, 48800, 49000, 49200, 49500, 49800, 50000,
50200, 50500, 50800, 51000, 51200, 51500
])
# 1. Tính các chỉ báo
sma_10 = moving_average_cumsum(btc_prices, 10)
ema_12 = exponential_moving_average(btc_prices, 2/13)
rsi = calculate_rsi(btc_prices, 14)
# 2. Phân tích rủi ro
returns = np.diff(btc_prices) / btc_prices[:-1]
volatility = np.std(returns) * np.sqrt(252)
max_dd, _ = maximum_drawdown(btc_prices)
# 3. Tính hiệu suất
total_return = (btc_prices[-1] - btc_prices[0]) / btc_prices[0]
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252)
print("=== Phân Tích BTC ===")
print(f"Giá hiện tại: ${btc_prices[-1]:,.2f}")
print(f"Tổng lợi nhuận: {total_return * 100:.2f}%")
print(f"Volatility: {volatility * 100:.2f}%")
print(f"Sharpe Ratio: {sharpe:.2f}")
print(f"Max Drawdown: {max_dd * 100:.2f}%")
print(f"RSI: {rsi[-1]:.2f}")Kết Luận
NumPy là công cụ không thể thiếu trong phân tích dữ liệu tài chính với Python. Nó cung cấp:
- Tính toán nhanh: Vectorization thay vì loops
- Hàm toán học phong phú: Từ cơ bản đến nâng cao
- Xử lý mảng hiệu quả: OHLC, returns, indicators
- Phân tích rủi ro: VaR, drawdown, volatility
- Tối ưu hóa: Portfolio optimization, performance metrics
Best Practices:
- Luôn sử dụng vectorization thay vì loops
- Kiểm tra dữ liệu NaN và ngoại lai
- Sử dụng các hàm NumPy có sẵn thay vì tự viết
- Tối ưu hóa bộ nhớ với các kiểu dữ liệu phù hợp
Tài Liệu Tham Khảo
- NumPy Official Documentation
- NumPy User Guide
- Python Phân Tích Định Lượng SMA
- Làm Bot Giao Dịch Backtest Với Pandas
- Các Thu Viện Python Phổ Biến Trong Giao Dịch Định Lượng
Bài viết được biên soạn bởi CoinGetMarket – Nền tảng giáo dục về crypto và trading bot.