TensorFlow للتعلم العميق
إطار Google للتعلم العميق. بناء شبكات عصبية للتنبؤ بالأسعار واكتشاف الأنماط وتحليل المشاعر.
الصعوبة: متقدم
الفئة: تعلم آلي
🧠 تعلم عميق
التثبيت
# Install TensorFlow (CPU)
$ pip install tensorflow
# Install with GPU support (CUDA required)
$ pip install tensorflow[and-cuda]
# Verify GPU
$ python -c "import tensorflow as tf; print(tf.config.list_physical_devices('GPU'))"
الميزات الرئيسية
Keras API
واجهة عالية المستوى لبناء وتدريب الشبكات العصبية بسهولة.
GPU مُسرّع
تسريع التدريب باستخدام CUDA GPU للمجموعات الكبيرة.
نماذج LSTM
شبكات ذاكرة طويلة قصيرة المدى للتنبؤ بالسلاسل الزمنية.
نشر الإنتاج
TensorFlow Serving و TFLite للنشر في بيئات الإنتاج.
أمثلة الكود
Setup and Data Preparation
Prepare financial data for deep learning
Python
import tensorflow as tf
import numpy as np
import pandas as pd
import yfinance as yf
from sklearn.preprocessing import MinMaxScaler
# Download data
df = yf.download('EURUSD=X', start='2020-01-01', end='2024-01-01')
prices = df['Close'].values.reshape(-1, 1)
# Scale data
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_prices = scaler.fit_transform(prices)
# Create sequences for LSTM
def create_sequences(data, seq_length):
X, y = [], []
for i in range(len(data) - seq_length):
X.append(data[i:i+seq_length])
y.append(data[i+seq_length])
return np.array(X), np.array(y)
SEQ_LENGTH = 60
X, y = create_sequences(scaled_prices, SEQ_LENGTH)
# Split data
split = int(len(X) * 0.8)
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]
print(f"Training samples: {len(X_train)}")
print(f"Testing samples: {len(X_test)}")
print(f"Sequence shape: {X_train.shape}")
LSTM Price Prediction Model
Build an LSTM network for time series
Python
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint
# Build LSTM model
model = Sequential([
LSTM(50, return_sequences=True, input_shape=(SEQ_LENGTH, 1)),
Dropout(0.2),
LSTM(50, return_sequences=True),
Dropout(0.2),
LSTM(50),
Dropout(0.2),
Dense(25),
Dense(1)
])
model.compile(
optimizer='adam',
loss='mse',
metrics=['mae']
)
model.summary()
# Callbacks
callbacks = [
EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True),
ModelCheckpoint('best_model.keras', monitor='val_loss', save_best_only=True)
]
# Train
history = model.fit(
X_train, y_train,
epochs=100,
batch_size=32,
validation_split=0.2,
callbacks=callbacks,
verbose=1
)
# Evaluate
loss, mae = model.evaluate(X_test, y_test)
print(f"Test Loss: {loss:.6f}")
print(f"Test MAE: {mae:.6f}")
CNN for Pattern Recognition
Detect chart patterns with convolutional networks
Python
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv1D, MaxPooling1D, Flatten, Dense, Dropout
def create_pattern_dataset(prices, window_size=30):
"""Create pattern classification dataset"""
X, y = [], []
for i in range(len(prices) - window_size - 5):
window = prices[i:i+window_size]
future_return = (prices[i+window_size+5] - prices[i+window_size]) / prices[i+window_size]
# Simple label: 1 if price goes up > 0.5%, 0 otherwise
label = 1 if future_return > 0.005 else 0
X.append(window)
y.append(label)
return np.array(X), np.array(y)
X, y = create_pattern_dataset(prices.flatten())
X = X.reshape(-1, 30, 1)
# CNN Model
cnn_model = Sequential([
Conv1D(64, 3, activation='relu', input_shape=(30, 1)),
MaxPooling1D(2),
Conv1D(128, 3, activation='relu'),
MaxPooling1D(2),
Conv1D(64, 3, activation='relu'),
Flatten(),
Dense(64, activation='relu'),
Dropout(0.3),
Dense(1, activation='sigmoid')
])
cnn_model.compile(
optimizer='adam',
loss='binary_crossentropy',
metrics=['accuracy']
)
# Train
history = cnn_model.fit(
X, y,
epochs=50,
batch_size=64,
validation_split=0.2,
callbacks=[EarlyStopping(patience=5)]
)
print(f"Validation Accuracy: {max(history.history['val_accuracy']):.2%}")
Transformer for Time Series
Modern transformer architecture for predictions
Python
import tensorflow as tf
from tensorflow.keras.layers import Layer, Dense, Dropout, MultiHeadAttention, LayerNormalization
class TransformerBlock(Layer):
def __init__(self, embed_dim, num_heads, ff_dim, rate=0.1):
super().__init__()
self.att = MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
self.ffn = tf.keras.Sequential([
Dense(ff_dim, activation="relu"),
Dense(embed_dim),
])
self.layernorm1 = LayerNormalization(epsilon=1e-6)
self.layernorm2 = LayerNormalization(epsilon=1e-6)
self.dropout1 = Dropout(rate)
self.dropout2 = Dropout(rate)
def call(self, inputs, training):
attn_output = self.att(inputs, inputs)
attn_output = self.dropout1(attn_output, training=training)
out1 = self.layernorm1(inputs + attn_output)
ffn_output = self.ffn(out1)
ffn_output = self.dropout2(ffn_output, training=training)
return self.layernorm2(out1 + ffn_output)
# Build Transformer model
inputs = tf.keras.Input(shape=(SEQ_LENGTH, 1))
x = Dense(32)(inputs)
x = TransformerBlock(32, 2, 64)(x)
x = TransformerBlock(32, 2, 64)(x)
x = tf.keras.layers.GlobalAveragePooling1D()(x)
x = Dense(20, activation="relu")(x)
x = Dropout(0.1)(x)
outputs = Dense(1)(x)
transformer_model = tf.keras.Model(inputs=inputs, outputs=outputs)
transformer_model.compile(optimizer="adam", loss="mse", metrics=["mae"])
transformer_model.fit(X_train, y_train, epochs=50, batch_size=32, validation_split=0.2)
Multi-Feature Input Model
Use OHLCV and indicators as features
Python
import pandas_ta as ta
# Calculate technical indicators
df['RSI'] = ta.rsi(df['Close'], length=14)
df['MACD'] = ta.macd(df['Close'])['MACD_12_26_9']
df['ATR'] = ta.atr(df['High'], df['Low'], df['Close'], length=14)
df['SMA_20'] = ta.sma(df['Close'], length=20)
df['SMA_50'] = ta.sma(df['Close'], length=50)
# Prepare features
features = ['Close', 'Volume', 'RSI', 'MACD', 'ATR', 'SMA_20', 'SMA_50']
df_features = df[features].dropna()
# Scale all features
scaler = MinMaxScaler()
scaled_features = scaler.fit_transform(df_features)
# Create multi-feature sequences
def create_multi_sequences(data, seq_length):
X, y = [], []
for i in range(len(data) - seq_length):
X.append(data[i:i+seq_length])
y.append(data[i+seq_length, 0]) # Predict Close
return np.array(X), np.array(y)
X, y = create_multi_sequences(scaled_features, 60)
# Model with multiple features
model = Sequential([
LSTM(100, return_sequences=True, input_shape=(60, len(features))),
Dropout(0.2),
LSTM(100),
Dropout(0.2),
Dense(50),
Dense(1)
])
model.compile(optimizer='adam', loss='mse')
model.fit(X, y, epochs=50, batch_size=32, validation_split=0.2)
Make Predictions and Trade
Generate trading signals from predictions
Python
def predict_next_price(model, scaler, recent_data, seq_length=60):
"""Predict next price from recent data"""
# Scale input
scaled = scaler.transform(recent_data.reshape(-1, 1))
# Create sequence
X = scaled[-seq_length:].reshape(1, seq_length, 1)
# Predict
scaled_pred = model.predict(X, verbose=0)
# Inverse transform
predicted_price = scaler.inverse_transform(scaled_pred)[0, 0]
return predicted_price
def generate_signals(model, scaler, prices, seq_length=60, threshold=0.001):
"""Generate trading signals based on predictions"""
signals = []
for i in range(seq_length, len(prices)):
recent = prices[i-seq_length:i]
current_price = prices[i-1]
predicted_price = predict_next_price(model, scaler, recent, seq_length)
expected_return = (predicted_price - current_price) / current_price
if expected_return > threshold:
signals.append('BUY')
elif expected_return < -threshold:
signals.append('SELL')
else:
signals.append('HOLD')
return signals
# Generate signals
signals = generate_signals(model, scaler, prices.flatten())
print(f"Buy signals: {signals.count('BUY')}")
print(f"Sell signals: {signals.count('SELL')}")
print(f"Hold signals: {signals.count('HOLD')}")
Save and Load Models
Persist trained models for production
Python
# Save model
model.save('price_prediction_model.keras')
# Save model weights only
model.save_weights('model_weights.weights.h5')
# Save scaler
import joblib
joblib.dump(scaler, 'scaler.pkl')
# Load model
from tensorflow.keras.models import load_model
loaded_model = load_model('price_prediction_model.keras')
# Load scaler
loaded_scaler = joblib.load('scaler.pkl')
# Verify
test_pred = loaded_model.predict(X_test[:1])
print(f"Prediction works: {test_pred[0, 0]:.6f}")
# Convert to TensorFlow Lite for mobile/edge
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()
with open('model.tflite', 'wb') as f:
f.write(tflite_model)
حالات الاستخدام الشائعة
التنبؤ بالأسعار
اكتشاف الأنماط
تحليل المشاعر
كشف الشذوذ
التعلم المعزز
تصنيف السوق
التنبؤ بالتقلبات
اختيار المحفظة
نمذجة السلاسل الزمنية
التداول الآلي
Best Practices & Common Pitfalls
تطبيع البيانات
طبّع بيانات الأسعار دائماً (0-1 أو Z-score) قبل التدريب.
تجنب الإفراط في التعلم
استخدم Early Stopping و Dropout و Validation Split.
اختيار الميزات
ابدأ بميزات بسيطة ذات معنى قبل إضافة تعقيد.
المشي للأمام
استخدم تقسيم زمني (Walk-Forward) وليس عشوائي.
استخدام GPU
استخدم GPU للتدريب — أسرع 10-50 مرة من CPU.
حفظ النماذج
احفظ النماذج وأوزانها بانتظام أثناء التدريب.
موارد إضافية
Trading-Specific
- Time Series Forecasting
- Financial ML Papers
- Quantitative Finance Resources
الخطوات التالية
Explore other ML frameworks or complement with traditional ML: