Cấu Trúc Dữ Liệu và Xử Lý Dữ Liệu Tài Chính với Python

2 tháng 10, 2026 · 12 phút để đọc

Trong phân tích dữ liệu tài chính, việc hiểu và sử dụng hiệu quả các cấu trúc dữ liệu là vô cùng quan trọng. Python cung cấp nhiều cấu trúc dữ liệu mạnh mẽ giúp chúng ta xử lý và phân tích dữ liệu tài chính một cách hiệu quả.

1. Các Cấu Trúc Dữ Liệu Cơ Bản

1.1. List (Danh sách)

List là cấu trúc dữ liệu linh hoạt nhất trong Python, cho phép lưu trữ nhiều phần tử có thể thay đổi.

# Tạo list giá cổ phiếu
stock_prices = [100.5, 102.3, 98.7, 105.2, 103.8]

# Thêm giá mới
stock_prices.append(107.5)

# Truy cập phần tử
first_price = stock_prices[0]  # 100.5
last_price = stock_prices[-1]  # 107.5

# Tính toán
average_price = sum(stock_prices) / len(stock_prices)
max_price = max(stock_prices)
min_price = min(stock_prices)

print(f"Giá trung bình: ${average_price:.2f}")
print(f"Giá cao nhất: ${max_price:.2f}")
print(f"Giá thấp nhất: ${min_price:.2f}")

1.2. Tuple (Bộ)

Tuple tương tự như list nhưng không thể thay đổi sau khi tạo.

# Tạo tuple thông tin cổ phiếu
stock_info = ("AAPL", "Apple Inc.", 150.25, 1000000)

# Truy cập thông tin
symbol = stock_info[0]      # "AAPL"
company = stock_info[1]     # "Apple Inc."
price = stock_info[2]       # 150.25
volume = stock_info[3]      # 1000000

# Tuple không thể thay đổi
# stock_info[2] = 155.50  # Lỗi!

1.3. Dictionary (Từ điển)

Dictionary lưu trữ dữ liệu dưới dạng cặp key-value, rất hữu ích cho việc lưu trữ thông tin có cấu trúc.

# Tạo dictionary thông tin danh mục đầu tư
portfolio = {
    "AAPL": {
        "name": "Apple Inc.",
        "shares": 100,
        "price": 150.25
    },
    "GOOGL": {
        "name": "Alphabet Inc.",
        "shares": 50,
        "price": 2800.75
    }
}

# Truy cập thông tin
apple_info = portfolio["AAPL"]
google_info = portfolio["GOOGL"]

# Tính tổng giá trị danh mục
total_value = sum(stock["shares"] * stock["price"] 
                 for stock in portfolio.values())

print(f"Tổng giá trị danh mục: ${total_value:,.2f}")

1.4. Set (Tập hợp)

Set lưu trữ các phần tử duy nhất, hữu ích cho việc loại bỏ trùng lặp và thực hiện các phép toán tập hợp.

# Tạo set các mã cổ phiếu
tech_stocks = {"AAPL", "GOOGL", "MSFT", "AMZN"}
finance_stocks = {"JPM", "BAC", "GS", "MSFT"}

# Phép toán tập hợp
all_stocks = tech_stocks | finance_stocks  # Hợp
common_stocks = tech_stocks & finance_stocks  # Giao
only_tech = tech_stocks - finance_stocks  # Hiệu

print(f"Tất cả cổ phiếu: {all_stocks}")
print(f"Cổ phiếu chung: {common_stocks}")
print(f"Chỉ cổ phiếu công nghệ: {only_tech}")

2. Vòng Lặp và Điều Kiện

2.1. Vòng lặp for

# Duyệt qua danh sách giá cổ phiếu
prices = [100.5, 102.3, 98.7, 105.2, 103.8]
for price in prices:
    print(f"Giá: ${price:.2f}")

# Duyệt qua dictionary
portfolio = {
    "AAPL": {"shares": 100, "price": 150.25},
    "GOOGL": {"shares": 50, "price": 2800.75}
}

for symbol, info in portfolio.items():
    value = info["shares"] * info["price"]
    print(f"{symbol}: {info['shares']} cổ phiếu, giá trị ${value:,.2f}")

2.2. Vòng lặp while

# Mô phỏng theo dõi giá cổ phiếu
current_price = 100.0
target_price = 110.0
days = 0

while current_price < target_price:
    # Giả lập biến động giá
    change = np.random.normal(0, 2)
    current_price += change
    days += 1
    
    print(f"Ngày {days}: ${current_price:.2f}")
    
print(f"Đạt mục tiêu sau {days} ngày")

2.3. Điều kiện

# Phân tích giá cổ phiếu
def analyze_price(price, ma20, ma50):
    if price > ma20 and price > ma50:
        return "Xu hướng tăng mạnh"
    elif price > ma20:
        return "Xu hướng tăng"
    elif price < ma20 and price < ma50:
        return "Xu hướng giảm mạnh"
    else:
        return "Xu hướng giảm"

# Sử dụng
current_price = 150.25
ma20 = 145.50
ma50 = 140.75

trend = analyze_price(current_price, ma20, ma50)
print(f"Phân tích: {trend}")

3. Hàm và Module

3.1. Định nghĩa và sử dụng hàm

def calculate_returns(prices):
    """Tính toán tỷ suất lợi nhuận từ chuỗi giá"""
    returns = []
    for i in range(1, len(prices)):
        daily_return = (prices[i] - prices[i-1]) / prices[i-1]
        returns.append(daily_return)
    return returns

def calculate_volatility(returns, window=20):
    """Tính toán độ biến động"""
    return np.std(returns[-window:]) * np.sqrt(252)

# Sử dụng
prices = [100, 102, 98, 105, 103]
returns = calculate_returns(prices)
volatility = calculate_volatility(returns)

print(f"Tỷ suất lợi nhuận: {[f'{r:.2%}' for r in returns]}")
print(f"Độ biến động: {volatility:.2%}")

3.2. Tổ chức code với module

# financial_analysis.py
import numpy as np
import pandas as pd

def calculate_technical_indicators(prices):
    """Tính toán các chỉ báo kỹ thuật"""
    df = pd.DataFrame(prices, columns=['Close'])
    df['MA20'] = df['Close'].rolling(window=20).mean()
    df['MA50'] = df['Close'].rolling(window=50).mean()
    df['RSI'] = calculate_rsi(df['Close'])
    return df

def calculate_rsi(prices, period=14):
    """Tính toán chỉ số RSI"""
    delta = prices.diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
    rs = gain / loss
    return 100 - (100 / (1 + rs))

# main.py
from financial_analysis import calculate_technical_indicators

# Sử dụng module
prices = [100, 102, 98, 105, 103, 107, 104, 108]
indicators = calculate_technical_indicators(prices)
print(indicators)

4. Bài Tập Thực Hành: Xử Lý Dữ Liệu Giá Cổ Phiếu

4.1. Phân tích dữ liệu giá cổ phiếu

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

def analyze_stock_data(symbol, start_date, end_date):
    """Phân tích dữ liệu cổ phiếu"""
    # Tạo dữ liệu mẫu
    dates = pd.date_range(start=start_date, end=end_date)
    np.random.seed(42)
    prices = np.random.normal(100, 5, len(dates)).cumsum() + 1000
    
    # Tạo DataFrame
    df = pd.DataFrame({
        'Close': prices,
        'Volume': np.random.randint(1000000, 5000000, len(dates))
    }, index=dates)
    
    # Tính toán các chỉ số
    df['MA20'] = df['Close'].rolling(window=20).mean()
    df['MA50'] = df['Close'].rolling(window=50).mean()
    df['Returns'] = df['Close'].pct_change()
    df['Volatility'] = df['Returns'].rolling(window=20).std() * np.sqrt(252)
    
    return df

def plot_stock_analysis(df, symbol):
    """Vẽ biểu đồ phân tích"""
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
    
    # Biểu đồ giá và MA
    ax1.plot(df.index, df['Close'], label='Giá đóng cửa')
    ax1.plot(df.index, df['MA20'], label='MA20')
    ax1.plot(df.index, df['MA50'], label='MA50')
    ax1.set_title(f'Phân tích giá cổ phiếu {symbol}')
    ax1.set_ylabel('Giá')
    ax1.legend()
    
    # Biểu đồ độ biến động
    ax2.plot(df.index, df['Volatility'], label='Độ biến động')
    ax2.set_xlabel('Ngày')
    ax2.set_ylabel('Độ biến động')
    ax2.legend()
    
    plt.tight_layout()
    plt.show()

# Sử dụng
df = analyze_stock_data('AAPL', '2023-01-01', '2023-12-31')
plot_stock_analysis(df, 'AAPL')

# In thống kê
print("\nThống kê cơ bản:")
print(f"Giá trung bình: ${df['Close'].mean():.2f}")
print(f"Độ biến động trung bình: {df['Volatility'].mean():.2%}")
print(f"Tỷ suất lợi nhuận: {df['Returns'].mean():.2%}")

4.2. Phân tích danh mục đầu tư

def analyze_portfolio(portfolio_data):
    """Phân tích danh mục đầu tư"""
    # Tính toán giá trị và tỷ trọng
    total_value = sum(stock['shares'] * stock['price'] 
                     for stock in portfolio_data.values())
    
    portfolio_analysis = {}
    for symbol, data in portfolio_data.items():
        value = data['shares'] * data['price']
        weight = value / total_value
        portfolio_analysis[symbol] = {
            'value': value,
            'weight': weight
        }
    
    return portfolio_analysis

# Sử dụng
portfolio = {
    "AAPL": {"shares": 100, "price": 150.25},
    "GOOGL": {"shares": 50, "price": 2800.75},
    "MSFT": {"shares": 75, "price": 300.50}
}

analysis = analyze_portfolio(portfolio)

print("\nPhân tích danh mục đầu tư:")
for symbol, data in analysis.items():
    print(f"{symbol}:")
    print(f"  Giá trị: ${data['value']:,.2f}")
    print(f"  Tỷ trọng: {data['weight']:.2%}")

5. Tổng Kết

Trong bài viết này, chúng ta đã tìm hiểu:

Các cấu trúc dữ liệu cơ bản trong Python và cách sử dụng chúng
Vòng lặp và điều kiện trong xử lý dữ liệu tài chính
Cách tổ chức code với hàm và module
Bài tập thực hành phân tích dữ liệu giá cổ phiếu

Các cấu trúc dữ liệu và kỹ thuật xử lý này là nền tảng quan trọng cho việc phân tích dữ liệu tài chính. Trong các bài tiếp theo, chúng ta sẽ đi sâu vào:

Phân tích dữ liệu với pandas
Trực quan hóa dữ liệu với matplotlib
Xây dựng chiến lược giao dịch
Và nhiều chủ đề thú vị khác!

Chọn cấu trúc dữ liệu nào cho việc gì?

Đây là câu hỏi quyết định hiệu năng của mọi hệ thống tài chính. Bảng dưới là kinh nghiệm thực tế:

Dữ liệu	Cấu trúc nên dùng	Lý do
Chuỗi giá OHLCV	`pandas.DataFrame` với `DatetimeIndex`	Hỗ trợ resample, rolling, alignment theo thời gian
Số dư tài khoản, cấu hình	`dict`	Tra cứu O(1), linh hoạt
Danh sách mã đã mua	`set`	Kiểm tra tồn tại O(1), tự loại trùng
Sổ lệnh (order book)	`dict` + `heapq`	Cần lấy min/max liên tục
Ma trận giá nhiều mã	`numpy.ndarray` 2 chiều	Tính toán vector hoá, dùng lại được cho ML
Dữ liệu tick khối lượng lớn	`numpy` array hoặc Arrow/Parquet	`DataFrame` quá tốn bộ nhớ
Hàng đợi lệnh chờ xử lý	`collections.deque`	Thêm/xoá hai đầu O(1)
Lịch sử giao dịch có tìm kiếm	`dict` theo `order_id`	Truy cập trực tiếp

Time series — kỹ năng quan trọng nhất với dữ liệu tài chính

Dữ liệu tài chính là chuỗi thời gian, và pandas được sinh ra cho việc này.

import pandas as pd
import numpy as np

# --- Tạo DatetimeIndex đúng cách ---
ngay = pd.date_range("2026-01-01", periods=500, freq="B")     # B = business days
gia = pd.Series(np.random.default_rng(42).normal(100, 2, 500).cumsum(), index=ngay)

# --- Resample: đổi tần suất ---
from_ngay_sang_tuan = gia.resample("W").agg(["first", "max", "min", "last"])
print(from_ngay_sang_tuan.head())

# --- Rolling: tính chỉ báo ---
gia_df = pd.DataFrame({"close": gia})
gia_df["sma20"] = gia_df["close"].rolling(20).mean()
gia_df["std20"] = gia_df["close"].rolling(20).std()
gia_df["bollinger_tren"] = gia_df["sma20"] + 2 * gia_df["std20"]
gia_df["bollinger_duoi"] = gia_df["sma20"] - 2 * gia_df["std20"]

# --- Lợi nhuận log — luôn dùng log return cho phân tích ---
gia_df["log_return"] = np.log(gia_df["close"] / gia_df["close"].shift(1))

# --- Biến động lịch sử (annualized) ---
vol_ngay = gia_df["log_return"].rolling(30).std()
vol_nam = vol_ngay * np.sqrt(252)
print(f"Volatility 30 ngày: {vol_nam.iloc[-1]:.2%}")

# --- Dịch chuyển thời gian ---
gia_df["gia_hom_qua"] = gia_df["close"].shift(1)
gia_df["gia_tuan_truoc"] = gia_df["close"].shift(5)
gia_df["sma20_ky_truoc"] = gia_df["sma20"].shift(1)

# --- Tránh look-ahead bias: chỉ dùng dữ liệu quá khứ ---
# SAI: dùng sma20 của chính nến đang xét để quyết định tại nến đó
gia_df["tin_hieu_sai"] = (gia_df["close"] > gia_df["sma20"]).astype(int)
# ĐÚNG: quyết định dựa trên thông tin đã biết ở nến trước
gia_df["tin_hieu_dung"] = (gia_df["close"].shift(1) > gia_df["sma20"].shift(1)).astype(int)

⚠️ Look-ahead bias là lỗi nghiêm trọng nhất trong phân tích tài chính. Nó khiến backtest báo lợi nhuận 200%/năm, còn thực tế thì lỗ. Nguyên tắc bất di bất dịch: mọi giá trị dùng để ra quyết định tại thời điểm t phải được biết ở thời điểm t. Khi viết code, hãy luôn tự hỏi: "Dữ liệu này có thực sự có sẵn lúc đó không?"

Cấu trúc sổ lệnh (Order Book)

import heapq
from dataclasses import dataclass, field


@dataclass
class SoLenh:
    """Sổ lệnh đơn giản — lấy best bid/ask trong O(log n)."""

    bids: list = field(default_factory=list)      # max-heap (lưu giá âm)
    asks: list = field(default_factory=list)      # min-heap

    def cap_nhat_bid(self, gia: float, khoi_luong: float) -> None:
        if khoi_luong <= 0:
            return                                  # lệnh bị huỷ
        heapq.heappush(self.bids, (-gia, khoi_luong))

    def cap_nhat_ask(self, gia: float, khoi_luong: float) -> None:
        if khoi_luong <= 0:
            return
        heapq.heappush(self.asks, (gia, khoi_luong))

    @property
    def gia_mua_tot_nhat(self) -> float | None:
        return -self.bids[0][0] if self.bids else None

    @property
    def gia_ban_tot_nhat(self) -> float | None:
        return self.asks[0][0] if self.asks else None

    @property
    def spread(self) -> float | None:
        if not self.bids or not self.asks:
            return None
        return self.gia_ban_tot_nhat - self.gia_mua_tot_nhat


so = SoLenh()
for gia, kl in [(100.0, 1.5), (100.2, 2.0), (99.8, 0.8), (100.5, 3.0)]:
    so.cap_nhat_bid(gia, kl)
for gia, kl in [(100.6, 1.0), (100.4, 2.5), (100.9, 0.7)]:
    so.cap_nhat_ask(gia, kl)

print(f"Best bid: {so.gia_mua_tot_nhat} | Best ask: {so.gia_ban_tot_nhat} | "
      f"Spread: {so.spread:.2f}")

heapq là lựa chọn đúng ở đây vì ta liên tục cần phần tử nhỏ nhất / lớn nhất. Nếu dùng list và sorted() mỗi lần cập nhật, chi phí sẽ là O(n log n) cho mỗi tick — không thể chịu được với dữ liệu realtime.

💡 Vì sao lưu giá âm cho max-heap: Python chỉ có min-heap. Để dùng nó như max-heap, ta đảo dấu giá trị. Đây là kỹ thuật kinh điển bạn sẽ gặp nhiều lần.

Hiệu năng — những con số bạn cần biết

Thao tác	`list`	`dict`/`set`
Truy cập theo index	O(1)	—
Truy cập theo khoá	O(n)	O(1)
Kiểm tra tồn tại	O(n)	O(1)
Thêm/xoá ở cuối	O(1)	O(1)
Thêm/xoá ở đầu	O(n)	O(1)

import time

N = 1_000_000
du_lieu_list = list(range(N))
du_lieu_set = set(range(N))

# Kiểm tra tồn tại 1000 lần — đo sự khác biệt
for ten, tap in [("list O(n)", du_lieu_list), ("set O(1)", du_lieu_set)]:
    t = time.perf_counter()
    for i in range(0, N, N // 1000):
        _ = i in tap
    print(f"{ten}: {time.perf_counter() - t:.4f}s")

Với 1 triệu phần tử, set nhanh hơn list khoảng hàng trăm đến hàng nghìn lần. Sự khác biệt này không phải chi tiết nhỏ — trong hệ thống giao dịch, nó là khác biệt giữa bot chạy được và bot không kịp phản ứng.

Tối ưu bộ nhớ — khi dữ liệu vượt RAM

import pandas as pd
import numpy as np

# --- Chọn dtype nhỏ hơn khi có thể ---
df = pd.DataFrame({
    "ma": pd.Series(["BTC", "ETH"] * 500_000),
    "gia": pd.Series(np.random.rand(1_000_000) * 70000),
    "khoi_luong": pd.Series(np.random.rand(1_000_000)),
    "so_lenh": pd.Series(np.random.randint(0, 100, 1_000_000)),
})
truoc = df.memory_usage(deep=True).sum() / 1e6
print(f"Trước tối ưu: {truoc:.1f} MB")

# Chuyển chuỗi lặp lại thành category — tiết kiệm rất nhiều
df["ma"] = df["ma"].astype("category")
# Dùng float32 thay float64 (đủ chính xác cho giá)
df["gia"] = df["gia"].astype("float32")
df["khoi_luong"] = df["khoi_luong"].astype("float32")
# Dùng int8/int16 khi giá trị nhỏ
df["so_lenh"] = df["so_lenh"].astype("int16")

sau = df.memory_usage(deep=True).sum() / 1e6
print(f"Sau tối ưu: {sau:.1f} MB  (giảm {(1 - sau/truoc) * 100:.0f}%)")

Ba kỹ thuật xử lý dữ liệu lớn hơn RAM:

Đọc theo chunk — pd.read_csv(..., chunksize=100_000) rồi xử lý từng phần.
Định dạng cột — Parquet nhỏ hơn CSV 5–10 lần và đọc nhanh hơn nhiều.
DuckDB / Polars — xử lý ngoài bộ nhớ, cú pháp gần giống pandas.

# Chuyển CSV sang Parquet — nên làm ngay khi dữ liệu lớn hơn 100 MB
df.to_parquet("du-lieu.parquet", compression="snappy", index=False)
df_lai = pd.read_parquet("du-lieu.parquet")

1. Các Cấu Trúc Dữ Liệu Cơ Bản​

1.1. List (Danh sách)​

1.2. Tuple (Bộ)​

1.3. Dictionary (Từ điển)​

1.4. Set (Tập hợp)​

2. Vòng Lặp và Điều Kiện​

2.1. Vòng lặp for​

2.2. Vòng lặp while​

2.3. Điều kiện​

3. Hàm và Module​

3.1. Định nghĩa và sử dụng hàm​

3.2. Tổ chức code với module​

4. Bài Tập Thực Hành: Xử Lý Dữ Liệu Giá Cổ Phiếu​

4.1. Phân tích dữ liệu giá cổ phiếu​

4.2. Phân tích danh mục đầu tư​

5. Tổng Kết​

Chọn cấu trúc dữ liệu nào cho việc gì?​

Time series — kỹ năng quan trọng nhất với dữ liệu tài chính​

Cấu trúc sổ lệnh (Order Book)​

Hiệu năng — những con số bạn cần biết​

Tối ưu bộ nhớ — khi dữ liệu vượt RAM​

Tài liệu tham khảo​