Hiểu và sử dụng thành thạo Python cho phân tích dữ liệu tài chính
· 17 phút để đọc
Python đã trở thành ngôn ngữ lập trình phổ biến nhất trong lĩnh vực phân tích dữ liệu tài chính nhờ vào các thư viện mạnh mẽ và cộng đồng hỗ trợ lớn. Trong bài viết này, chúng ta sẽ tìm hiểu cách sử dụng Python để phân tích dữ liệu tài chính một cách hiệu quả.
1. Các thư viện Python cần thiết
Để bắt đầu với phân tích dữ liệu tài chính, bạn cần cài đặt các thư viện sau:
pip install pandas numpy matplotlib seaborn yfinance scikit-learn ta-lib ccxt backtrader
2. Thu thập dữ liệu tài chính
2.1. Sử dụng yfinance
import yfinance as yf
# Lấy dữ liệu cổ phiếu Apple
aapl = yf.Ticker("AAPL")
hist = aapl.history(period="1y")
# Hiển thị 5 dòng đầu tiên
print(hist.head())
3. Phân tích dữ liệu cơ bản
3.1. Tính toán các chỉ số kỹ thuật
import pandas as pd
import numpy as np
# Tính toán đường trung bình động 20 ngày
hist['MA20'] = hist['Close'].rolling(window=20).mean()
# Tính toán RSI (Relative Strength Index)
def calculate_rsi(data, periods=14):
delta = data.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=periods).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=periods).mean()
rs = gain / loss
return 100 - (100 / (1 + rs))
hist['RSI'] = calculate_rsi(hist['Close'])
4. Trực quan hóa dữ liệu
4.1. Biểu đồ giá và khối lượng
import matplotlib.pyplot as plt
import seaborn as sns
plt.figure(figsize=(12, 6))
plt.plot(hist.index, hist['Close'], label='Giá đóng cửa')
plt.plot(hist.index, hist['MA20'], label='MA20', linestyle='--')
plt.title('Biểu đồ giá cổ phiếu Apple')
plt.xlabel('Ngày')
plt.ylabel('Giá (USD)')
plt.legend()
plt.show()
5. Phân tích rủi ro
5.1. Tính toán độ biến động (Volatility)
# Tính toán độ biến động hàng ngày
daily_returns = hist['Close'].pct_change()
volatility = daily_returns.std() * np.sqrt(252) # Độ biến động hàng năm
print(f"Độ biến động hàng năm: {volatility:.2%}")
6. Xây dựng danh mục đầu tư
6.1. Tối ưu hóa danh mục đầu tư
from scipy.optimize import minimize
def portfolio_volatility(weights, returns):
return np.sqrt(np.dot(weights.T, np.dot(returns.cov() * 252, weights)))
# Tối ưu hóa danh mục đầu tư
def optimize_portfolio(returns):
num_assets = len(returns.columns)
constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
bounds = tuple((0, 1) for asset in range(num_assets))
result = minimize(portfolio_volatility,
x0=np.array([1/num_assets] * num_assets),
args=(returns,),
method='SLSQP',
bounds=bounds,
constraints=constraints)
return result.x
7. Kết luận
Python cung cấp một bộ công cụ mạnh mẽ cho phân tích dữ liệu tài chính. Với các thư viện như pandas, numpy, và matplotlib, bạn có thể thực hiện các phân tích phức tạp và trực quan hóa dữ liệu một cách hiệu quả. Hãy tiếp tục học hỏi và thực hành để nâng cao kỹ năng phân tích tài chính của bạn.
Tài liệu tham khảo
8. Làm chủ các thư viện phân tích dữ liệu và giao dịch tự động
Phân tích dữ liệu tài chính hiện đại không chỉ dừng lại ở việc trực quan hóa hay tính toán các chỉ số cơ bản, mà còn mở rộng sang các kỹ thuật thống kê nâng cao và giao dịch tự động. Dưới đây là các thư viện mạnh mẽ giúp bạn làm chủ lĩnh vực này.
8.1. Thư viện phân tích dữ liệu nâng cao
- statsmodels: Phân tích thống kê, kiểm định giả thuyết, mô hình hồi quy.
- ta-lib: Tính toán hơn 150 chỉ báo kỹ thuật chuyên sâu (MA, RSI, MACD, Bollinger Bands, v.v.).
- scikit-learn: Học máy cho dự báo tài chính, phân loại, hồi quy, clustering.
Cài đặt nhanh:
pip install statsmodels ta-lib scikit-learn
Ví dụ: Tính chỉ báo kỹ thuật với ta-lib
import talib
import numpy as np
import pandas as pd
# Dữ liệu mẫu giá đóng cửa
close = np.random.random(100) * 100
# Tính RSI và MACD
rsi = talib.RSI(close, timeperiod=14)
macd, macdsignal, macdhist = talib.MACD(close, fastperiod=12, slowperiod=26, signalperiod=9)
print('RSI:', rsi[-5:])
print('MACD:', macd[-5:])
8.2. Thư viện giao dịch tự động
- ccxt: Kết nối API các sàn giao dịch tiền số (Binance, Coinbase, v.v.).
- backtrader: Xây dựng, kiểm thử chiến lược giao dịch tự động.
- pyalgotrade: Phân tích và kiểm thử chiến lược giao dịch.
Cài đặt nhanh:
pip install ccxt backtrader pyalgotrade
Ví dụ: Kết nối API sàn với ccxt
import ccxt
# Kết nối tới sàn Binance
exchange = ccxt.binance()
ticker = exchange.fetch_ticker('BTC/USDT')
print(ticker)
Ví dụ: Backtest chiến lược với backtrader
import backtrader as bt
class TestStrategy(bt.Strategy):
def __init__(self):
self.sma = bt.indicators.SimpleMovingAverage(self.datas[0], period=15)
def next(self):
if self.datas[0].close[0] > self.sma[0]:
self.buy()
elif self.datas[0].close[0] < self.sma[0]:
self.sell()
cerebro = bt.Cerebro()
data = bt.feeds.YahooFinanceData(dataname='AAPL', fromdate=pd.Timestamp('2022-01-01'), todate=pd.Timestamp('2023-01-01'))
cerebro.adddata(data)
cerebro.addstrategy(TestStrategy)
cerebro.run()
cerebro.plot()
8.3. Tổng kết
Việc kết hợp các thư viện phân tích dữ liệu nâng cao và giao dịch tự động giúp bạn xây dựng hệ thống phân tích và giao dịch tài chính toàn diện, từ phân tích, dự báo đến thực thi chiến lược tự động.
Nếu bạn gặp lỗi khi cài ta-lib, hãy gửi nội dung lỗi để tôi hướng dẫn chi tiết hơn!
Bạn muốn tôi kiểm tra giúp bước nào, hay gửi hướng dẫn cài ta-lib chi tiết cho Windows?
dir static\img
9. Phân tích dữ liệu thị trường và xây dựng chiến lược giao dịch tự động
9.1. Phân tích dữ liệu thị trường
a. Phân tích xu hướng (Trend Analysis)
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
def analyze_trend(data, window=20):
# Tính toán các đường trung bình động
data['SMA20'] = data['Close'].rolling(window=window).mean()
data['SMA50'] = data['Close'].rolling(window=50).mean()
data['SMA200'] = data['Close'].rolling(window=200).mean()
# Xác định xu hướng
data['Trend'] = np.where(data['SMA20'] > data['SMA50'], 'Uptrend', 'Downtrend')
return data
# Ví dụ sử dụng
df = pd.DataFrame({
'Close': np.random.random(300) * 100
})
df = analyze_trend(df)
print(df['Trend'].value_counts())
b. Phân tích khối lượng (Volume Analysis)
def analyze_volume(data):
# Tính toán khối lượng trung bình
data['Volume_MA'] = data['Volume'].rolling(window=20).mean()
# Xác định khối lượng bất thường
data['Volume_Ratio'] = data['Volume'] / data['Volume_MA']
data['Unusual_Volume'] = data['Volume_Ratio'] > 2.0
return data
9.2. Xây dựng chiến lược giao dịch
a. Chiến lược giao dịch theo xu hướng
class TrendFollowingStrategy:
def __init__(self, data):
self.data = data
self.positions = []
def generate_signals(self):
# Tín hiệu mua khi SMA20 cắt lên SMA50
self.data['Signal'] = 0
self.data.loc[self.data['SMA20'] > self.data['SMA50'], 'Signal'] = 1
self.data.loc[self.data['SMA20'] < self.data['SMA50'], 'Signal'] = -1
return self.data
b. Chiến lược giao dịch theo mô hình giá
def identify_patterns(data):
# Xác định mô hình nến
data['Body'] = data['Close'] - data['Open']
data['Upper_Shadow'] = data['High'] - data[['Open', 'Close']].max(axis=1)
data['Lower_Shadow'] = data[['Open', 'Close']].min(axis=1) - data['Low']
# Xác định mô hình Doji
data['Doji'] = abs(data['Body']) < (data['High'] - data['Low']) * 0.1
return data
9.3. Xây dựng bot giao dịch tự động
a. Cấu trúc cơ bản của bot giao dịch
class TradingBot:
def __init__(self, api_key, api_secret):
self.exchange = ccxt.binance({
'apiKey': api_key,
'secret': api_secret
})
self.strategy = None
def set_strategy(self, strategy):
self.strategy = strategy
def execute_trade(self, signal):
if signal == 1: # Tín hiệu mua
# Thực hiện lệnh mua
pass
elif signal == -1: # Tín hiệu bán
# Thực hiện lệnh bán
pass
def run(self):
while True:
# Lấy dữ liệu thị trường
market_data = self.exchange.fetch_ohlcv('BTC/USDT', '1h')
# Phân tích và tạo tín hiệu
signal = self.strategy.analyze(market_data)
# Thực hiện giao dịch
self.execute_trade(signal)
# Đợi đến chu kỳ tiếp theo
time.sleep(3600) # Đợi 1 giờ
b. Quản lý rủi ro
class RiskManager:
def __init__(self, max_position_size=0.1, stop_loss=0.02):
self.max_position_size = max_position_size
self.stop_loss = stop_loss
def calculate_position_size(self, account_balance, current_price):
return min(
account_balance * self.max_position_size,
account_balance / current_price
)
def set_stop_loss(self, entry_price, position_type):
if position_type == 'long':
return entry_price * (1 - self.stop_loss)
else:
return entry_price * (1 + self.stop_loss)
9.4. Kiểm thử và tối ưu hóa chiến lược
a. Backtesting
def backtest_strategy(strategy, historical_data):
results = []
for i in range(len(historical_data)):
signal = strategy.generate_signal(historical_data[:i+1])
if signal != 0:
results.append({
'date': historical_data.index[i],
'signal': signal,
'price': historical_data['Close'][i]
})
return pd.DataFrame(results)
b. Tối ưu hóa tham số
from scipy.optimize import brute
def optimize_parameters(strategy, data, param_ranges):
def objective(params):
strategy.set_parameters(params)
results = backtest_strategy(strategy, data)
return -calculate_sharpe_ratio(results) # Tối ưu hóa tỷ lệ Sharpe
optimal_params = brute(objective, param_ranges)
return optimal_params
9.5. Tổng kết
Việc phân tích dữ liệu thị trường và xây dựng chiến lược giao dịch tự động đòi hỏi sự kết hợp giữa:
- Phân tích kỹ thuật và cơ bản
- Xây dựng chiến lược giao dịch có hệ thống
- Quản lý rủi ro chặt chẽ
- Kiểm thử và tối ưu hóa liên tục
Python cung cấp các công cụ mạnh mẽ để thực hiện tất cả các bước trên, giúp bạn xây dựng hệ thống giao dịch tự động hiệu quả và an toàn.
10. Các kỹ thuật nâng cao trong phân tích dữ liệu tài chính
10.1. Phân tích chuỗi thời gian (Time Series Analysis)
a. Phân tích tính dừng (Stationarity)
from statsmodels.tsa.stattools import adfuller
def check_stationarity(data):
# Kiểm định Augmented Dickey-Fuller
result = adfuller(data)
print('ADF Statistic:', result[0])
print('p-value:', result[1])
print('Critical values:', result[4])
if result[1] <= 0.05:
print("Dữ liệu là dừng (stationary)")
else:
print("Dữ liệu không dừng (non-stationary)")
b. Phân tích tự tương quan (Autocorrelation)
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf
def analyze_correlation(data):
# Vẽ biểu đồ tự tương quan
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8))
plot_acf(data, ax=ax1)
plot_pacf(data, ax=ax2)
plt.tight_layout()
plt.show()
10.2. Dự báo giá với Machine Learning
a. Mô hình LSTM cho dự báo giá
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
def create_lstm_model(input_shape):
model = Sequential([
LSTM(50, return_sequences=True, input_shape=input_shape),
LSTM(50, return_sequences=False),
Dense(25),
Dense(1)
])
model.compile(optimizer='adam', loss='mse')
return model
def prepare_data(data, look_back=60):
X, y = [], []
for i in range(len(data) - look_back):
X.append(data[i:(i + look_back)])
y.append(data[i + look_back])
return np.array(X), np.array(y)
b. Mô hình XGBoost cho dự báo xu hướng
import xgboost as xgb
def create_xgboost_model():
model = xgb.XGBClassifier(
n_estimators=100,
learning_rate=0.1,
max_depth=5
)
return model
def prepare_features(data):
# Tạo các đặc trưng kỹ thuật
data['Returns'] = data['Close'].pct_change()
data['MA5'] = data['Close'].rolling(window=5).mean()
data['MA20'] = data['Close'].rolling(window=20).mean()
data['Volatility'] = data['Returns'].rolling(window=20).std()
# Tạo nhãn (1: tăng, 0: giảm)
data['Target'] = (data['Close'].shift(-1) > data['Close']).astype(int)
return data
10.3. Tối ưu hóa chiến lược giao dịch
a. Tối ưu hóa tham số với Genetic Algorithm
from deap import base, creator, tools, algorithms
def optimize_strategy_ga(strategy, data, ngen=50):
creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", list, fitness=creator.FitnessMax)
toolbox = base.Toolbox()
toolbox.register("attr_float", random.uniform, 0, 1)
toolbox.register("individual", tools.initRepeat, creator.Individual,
toolbox.attr_float, n=3)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
def evaluate(individual):
strategy.set_parameters(individual)
results = backtest_strategy(strategy, data)
return (calculate_sharpe_ratio(results),)
toolbox.register("evaluate", evaluate)
toolbox.register("mate", tools.cxTwoPoint)
toolbox.register("mutate", tools.mutGaussian, mu=0, sigma=0.2, indpb=0.2)
toolbox.register("select", tools.selTournament, tournsize=3)
pop = toolbox.population(n=50)
result, logbook = algorithms.eaSimple(pop, toolbox, cxpb=0.7, mutpb=0.2,
ngen=ngen, verbose=True)
return tools.selBest(result, k=1)[0]
b. Tối ưu hóa danh mục đầu tư với Modern Portfolio Theory
def optimize_portfolio_modern(returns, risk_free_rate=0.02):
n_assets = len(returns.columns)
returns_mean = returns.mean()
returns_cov = returns.cov()
def portfolio_stats(weights):
portfolio_return = np.sum(returns_mean * weights)
portfolio_std = np.sqrt(np.dot(weights.T, np.dot(returns_cov, weights)))
sharpe_ratio = (portfolio_return - risk_free_rate) / portfolio_std
return -sharpe_ratio
constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
bounds = tuple((0, 1) for asset in range(n_assets))
result = minimize(portfolio_stats,
x0=np.array([1/n_assets] * n_assets),
method='SLSQP',
bounds=bounds,
constraints=constraints)
return result.x
10.4. Xử lý dữ liệu thời gian thực
a. Xử lý dữ liệu streaming
import websocket
import json
import threading
class MarketDataStream:
def __init__(self, symbol, callback):
self.symbol = symbol
self.callback = callback
self.ws = None
def on_message(self, ws, message):
data = json.loads(message)
self.callback(data)
def on_error(self, ws, error):
print(f"Error: {error}")
def on_close(self, ws, close_status_code, close_msg):
print("WebSocket Connection Closed")
def on_open(self, ws):
print("WebSocket Connection Opened")
ws.send(json.dumps({
"method": "SUBSCRIBE",
"params": [f"{self.symbol.lower()}@trade"],
"id": 1
}))
def start(self):
websocket.enableTrace(True)
self.ws = websocket.WebSocketApp(
f"wss://stream.binance.com:9443/ws",
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
wst = threading.Thread(target=self.ws.run_forever)
wst.daemon = True
wst.start()
b. Xử lý dữ liệu với Apache Kafka
from confluent_kafka import Consumer, Producer
class KafkaMarketData:
def __init__(self, bootstrap_servers):
self.producer = Producer({'bootstrap.servers': bootstrap_servers})
self.consumer = Consumer({
'bootstrap.servers': bootstrap_servers,
'group.id': 'market_data_group',
'auto.offset.reset': 'earliest'
})
def produce_data(self, topic, data):
self.producer.produce(topic, json.dumps(data).encode('utf-8'))
self.producer.flush()
def consume_data(self, topic, callback):
self.consumer.subscribe([topic])
while True:
msg = self.consumer.poll(1.0)
if msg is None:
continue
if msg.error():
print(f"Consumer error: {msg.error()}")
continue
data = json.loads(msg.value().decode('utf-8'))
callback(data)
10.5. Tổng kết
Các kỹ thuật nâng cao trong phân tích dữ liệu tài chính và giao dịch tự động bao gồm:
- Phân tích chuỗi thời gian và dự báo
- Machine Learning cho dự báo giá và xu hướng
- Tối ưu hóa chiến lược với các thuật toán tiên tiến
- Xử lý dữ liệu thời gian thực
Đi sâu: bốn kỹ thuật nâng cao
1. Phân tích chuỗi thời gian — kiểm tra tính dừng trước khi làm gì khác
Hầu hết model chuỗi thời gian giả định dữ liệu dừng (stationary) — nghĩa là phân phối không đổi theo thời gian. Giá cổ phiếu không dừng; lợi nhuận thường dừng. Đây là bước phải làm trước tiên.
import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import adfuller
def kiem_tra_dung(chuoi: pd.Series, ten: str) -> dict:
"""Kiểm định Augmented Dickey-Fuller cho tính dừng.
H0: chuỗi có nghiệm đơn vị (KHÔNG dừng)
p < 0,05 → bác bỏ H0 → chuỗi DỪNG
"""
chuoi = chuoi.dropna()
if len(chuoi) < 50:
return {"ten": ten, "loi": "Quá ít dữ liệu"}
ket_qua = adfuller(chuoi, autolag="AIC")
p_value = ket_qua[1]
return {
"ten": ten,
"p_value": round(p_value, 5),
"dung": p_value < 0.05,
"ket_luan": "✅ Dừng" if p_value < 0.05 else "❌ Không dừng — cần lấy sai phân",
}
# Ví dụ minh hoạ: giá không dừng, lợi nhuận log thì dừng
gia = pd.Series(np.cumsum(np.random.default_rng(42).normal(0, 1, 1000)) + 100)
loi_nhuan = np.log(gia / gia.shift(1)).dropna()
print(kiem_tra_dung(gia, "Giá"))
print(kiem_tra_dung(loi_nhuan, "Log return"))
Kết quả điển hình: giá có p-value ~0,9 (không dừng), log return có p-value ~0,0001 (dừng). Điều này giải thích vì sao mọi phân tích tài chính nghiêm túc đều làm việc trên lợi nhuận, không phải giá.
⚠️ Hệ quả thực tế: nếu bạn đưa giá thô vào model ARIMA hoặc hồi quy, kết quả sẽ có R² rất cao (0,95+) nhưng hoàn toàn vô nghĩa — model chỉ học được xu hướng thời gian, không học được quan hệ nhân quả. Đây là hồi quy giả (spurious regression), một trong những lỗi nghiêm trọng nhất trong phân tích tài chính.
2. Kiểm soát overfitting trong tài chính
Tài chính là lĩnh vực dễ overfit nhất vì tỷ lệ tín hiệu/nhiễu cực thấp. Đây là cách phòng ngừa:
| Kỹ thuật | Mô tả | Hiệu quả |
|---|---|---|
| Walk-forward | Chia nhiều cửa sổ, train/tets luân phiên theo thời gian | ⭐⭐⭐⭐⭐ |
| Purged K-Fold | Loại bỏ dữ liệu gần biên giữa train và test | ⭐⭐⭐⭐ |
| Cross-validation theo thời gian | TimeSeriesSplit thay vì KFold | ⭐⭐⭐⭐⭐ |
| Giới hạn số tham số | Mỗi tham số cần ≥ 100 mẫu | ⭐⭐⭐⭐ |
| Kiểm định ngoài mẫu | Giữ riêng dữ liệu, chạm một lần duy nhất | ⭐⭐⭐⭐⭐ |
import numpy as np
import pandas as pd
from sklearn.model_selection import TimeSeriesSplit
def walk_forward(model_fn, X: pd.DataFrame, y: pd.Series,
so_cua_so: int = 5) -> pd.DataFrame:
"""Walk-forward validation — cách kiểm định đúng cho dữ liệu tài chính."""
tscv = TimeSeriesSplit(n_splits=so_cua_so)
ban_ghi = []
for i, (idx_train, idx_test) in enumerate(tscv.split(X), 1):
X_tr, X_te = X.iloc[idx_train], X.iloc[idx_test]
y_tr, y_te = y.iloc[idx_train], y.iloc[idx_test]
model = model_fn()
model.fit(X_tr, y_tr)
diem_train = model.score(X_tr, y_tr)
diem_test = model.score(X_te, y_te)
ban_ghi.append({
"cua_so": i,
"n_train": len(idx_train),
"n_test": len(idx_test),
"diem_train": round(diem_train, 4),
"diem_test": round(diem_test, 4),
"sut_giam": round(diem_train - diem_test, 4),
})
return pd.DataFrame(ban_ghi)
Cách đọc kết quả:
| Mẫu | Diễn giải |
|---|---|
| train 0,72 / test 0,70 | ✅ Tốt — model tổng quát hoá được |
| train 0,95 / test 0,55 | ❌ Overfit nặng — không dùng được |
| train 0,55 / test 0,54 | ⚠️ Không overfit nhưng cũng không học được gì |
3. Vì sao lợi nhuận tài chính gần như không dự báo được
Đây là sự thật khó chấp nhận nhưng cần biết trước khi dành nhiều thời gian cho ML:
Tỷ lệ tín hiệu / nhiễu trong dữ liệu tài chính:
Ảnh y tế (phát hiện ung thư) ~ 1:1 → ML rất hiệu quả
Nhận diện chữ viết ~ 5:1 → ML rất hiệu quả
Dự báo thời tiết ngắn hạn ~ 1:10 → ML khá
Dự báo lợi nhuận cổ phiếu ngày ~ 1:100 → ML rất khó
Dự báo tick-by-tick ~ 1:1000 → Cần hạ tầng HFT
Hệ quả: một model đạt accuracy 53% trong dự báo hướng giá ngày là kết quả TỐT. Nếu bạn thấy model đạt 75%, gần như chắc chắn có data leakage.
# Kiểm tra số nhanh: model có tốt hơn "luôn đoán tăng" không?
from sklearn.dummy import DummyClassifier
# baseline = luôn dự đoán lớp đa số
def so_voi_baseline(model, X_train, y_train, X_test, y_test) -> dict:
baseline = DummyClassifier(strategy="most_frequent").fit(X_train, y_train)
model.fit(X_train, y_train)
diem_baseline = baseline.score(X_test, y_test)
diem_model = model.score(X_test, y_test)
return {
"baseline": f"{diem_baseline:.4f}",
"model": f"{diem_model:.4f}",
"chenh_lech": f"{diem_model - diem_baseline:+.4f}",
"ket_luan": ("✅ Có giá trị" if diem_model > diem_baseline + 0.02
else "❌ Không tốt hơn đoán bừa"),
}
💡 Phép thử quan trọng nhất: nếu model của bạn không vượt baseline ít nhất 2% thì nó chưa có giá trị. Và 2% cũng chưa đủ bù chi phí giao dịch — bạn cần thêm biên độ để trang trải spread và phí.
4. Xử lý dữ liệu thời gian thực
import asyncio
import httpx
from collections import deque
class BoDemGia:
"""Lưu trữ giá realtime với cửa sổ trượt — không dùng list vô hạn."""
def __init__(self, kich_thuoc: int = 500):
self.gia = deque(maxlen=kich_thuoc) # tự động bỏ phần tử cũ
self.thoi_gian = deque(maxlen=kich_thuoc)
def them(self, gia: float, thoi_diem: float) -> None:
self.gia.append(gia)
self.thoi_gian.append(thoi_diem)
def gia_moi_nhat(self) -> float | None:
return self.gia[-1] if self.gia else None
def bien_dong(self) -> float | None:
if len(self.gia) < 20:
return None
import numpy as np
return float(np.std(list(self.gia)[-20:]))
async def theo_doi_stream(symbol: str = "btcusdt") -> None:
"""Đọc WebSocket Binance — dữ liệu realtime thật."""
dem = BoDemGia()
url = f"wss://stream.binance.com:9443/ws/{symbol}@trade"
async with httpx.AsyncClient(timeout=None) as client:
async with client.stream("GET", url) as phan_hoi:
async for dong in phan_hoi.aiter_lines():
if not dong:
continue
import json
du_lieu = json.loads(dong)
dem.them(float(du_lieu["p"]), du_lieu["T"])
if len(dem.gia) % 100 == 0:
print(f"{symbol}: {dem.gia_moi_nhat():,.2f} | "
f"biến động {dem.bien_dong():.4f}")
# asyncio.run(theo_doi_stream())
⚠️ Cảnh báo về dữ liệu realtime: đừng bao giờ lưu tick vào list không giới hạn. Sau vài giờ, chương trình của bạn sẽ hết RAM và chết. Luôn dùng
deque(maxlen=...)hoặc ghi ra file/DB theo lô.
Bốn hiểu lầm phổ biến về ML trong tài chính
- "Nhiều dữ liệu hơn thì model tốt hơn." Không đúng nếu dữ liệu không có tín hiệu. 10 năm dữ liệu nhiễu vẫn là nhiễu.
- "Model phức tạp hơn thì chính xác hơn." Trong tài chính, model đơn giản thường tổng quát hoá tốt hơn. Phần lốn ML được giải thích tốt nhất bằng hồi quy logistic.
- "Accuracy cao nghĩa là model tốt." Với dữ liệu mất cân bằng, accuracy vô nghĩa. Dùng precision/recall/PR-AUC.
- "Backtest tốt thì chạy thật tốt." Không. Backtest tốt là điều kiện cần, không phải điều kiện đủ. Còn cần kiểm định ngoài mẫu, kiểm tra chi phí, và theo dõi thực tế.
Tóm tắt lộ trình học
| Giai đoạn | Nội dung | Thời lượng |
|---|---|---|
| 1 | pandas, numpy, xử lý dữ liệu thiếu | 3–4 tuần |
| 2 | Trực quan hoá, khám phá dữ liệu | 2 tuần |
| 3 | Thống kê: phân phối, kiểm định giả thuyết | 3 tuần |
| 4 | Chuỗi thời gian: tính dừng, ARIMA | 4 tuần |
| 5 | ML cơ bản: hồi quy, cây, đánh giá model | 4 tuần |
| 6 | Kỹ thuật tài chính chuyên biệt: walk-forward, backtest | 4 tuần |
| 7 | Dự án thực tế đủ sâu | 4–6 tuần |