| 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 | 15 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 np

Tạ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ày

Truy 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

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.

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