第4章 量化交易中的机器学习¶

import numpy as np
from sklearn.svm import SVC

# ------------------------------------------------------------
# 1) Simulate a price series (random walk)
# ------------------------------------------------------------

np.random.seed(42)

# Simulate 100 days of prices
prices = 100 + np.cumsum(np.random.normal(0, 1, 100))

# Compute daily returns
returns = np.diff(prices) / prices[:-1]

# ------------------------------------------------------------
# 2) Build features and labels
# Use returns from the previous two days to predict
# whether the return on day t+1 is positive or negative
# ------------------------------------------------------------

# Features: returns from day t-2 and t-1
X = np.column_stack([
    returns[:-2],
    returns[1:-1]
])

# Labels: 1 if next-day return is positive, 0 otherwise
y = (returns[2:] > 0).astype(int)

# ------------------------------------------------------------
# 3) Train-test split (80% train, 20% test)
# ------------------------------------------------------------

train_size = int(len(X) * 0.8)

X_train, X_test = X[:train_size], X[train_size:]
y_train, y_test = y[:train_size], y[train_size:]

# ------------------------------------------------------------
# 4) Train an SVM classifier and make predictions
# ------------------------------------------------------------

model = SVC(kernel="rbf", gamma="auto")
model.fit(X_train, y_train)

y_pred = model.predict(X_test)

# ------------------------------------------------------------
# 5) Compare predicted vs actual signals
# ------------------------------------------------------------

print("Predicted signals:", y_pred[:5])
print("Actual signals:   ", y_test[:5])

from datetime import datetime, timedelta

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

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader

from sklearn.svm import SVC
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import classification_report

# ------------------------------------------------------------
# Reproducibility
# ------------------------------------------------------------

np.random.seed(42)
torch.manual_seed(42)

# ------------------------------------------------------------
# 1) Synthetic data generation
# ------------------------------------------------------------

def generate_sample_data(n_samples: int = 1000):
    """Generate synthetic financial time-series data for demonstration."""
    print("Generating synthetic data...")

    t = np.linspace(0, 4 * np.pi, n_samples)
    trend = 100 + t * 2
    seasonal = 10 * np.sin(t)
    noise = np.random.normal(0, 5, n_samples)

    prices = trend + seasonal + noise

    returns = np.diff(prices) / prices[:-1]

    volatility = (
        pd.Series(returns)
        .rolling(window=20)
        .std()
        .fillna(0)
        .values[:-1]
    )

    momentum = (
        pd.Series(returns)
        .rolling(window=10)
        .mean()
        .fillna(0)
        .values[:-1]
    )

    X = np.column_stack([
        returns[:-1],      # previous-day return
        volatility,        # 20-day volatility
        momentum           # 10-day momentum
    ])

    y = (returns[1:] > 0).astype(int)

    print("Data generation complete.")
    print(f"Features shape: {X.shape}, Labels shape: {y.shape}")

    return X, y, prices

# ------------------------------------------------------------
# 2) SVM Trader
# ------------------------------------------------------------

class SVMTrader:
    """Support Vector Machine–based trading model."""

    def __init__(self):
        self.scaler = StandardScaler()
        self.model = SVC(kernel="rbf", probability=True)

    def train(self, X_train, y_train):
        print("\nTraining SVM model...")
        X_scaled = self.scaler.fit_transform(X_train)
        self.model.fit(X_scaled, y_train)
        print("SVM training complete.")

    def predict(self, X):
        X_scaled = self.scaler.transform(X)
        return self.model.predict(X_scaled)

    def predict_proba(self, X):
        X_scaled = self.scaler.transform(X)
        return self.model.predict_proba(X_scaled)

# ------------------------------------------------------------
# 3) Random Forest Trader
# ------------------------------------------------------------

class RandomForestTrader:
    """Random Forest–based trading model."""

    def __init__(self, n_estimators: int = 100):
        self.model = RandomForestClassifier(n_estimators=n_estimators)

    def train(self, X_train, y_train):
        print("\nTraining Random Forest model...")
        self.model.fit(X_train, y_train)
        print("Random Forest training complete.")

        feature_importance = pd.DataFrame({
            "feature": ["returns", "volatility", "momentum"],
            "importance": self.model.feature_importances_
        })

        print("\nFeature Importances:")
        print(feature_importance.sort_values("importance", ascending=False))

    def predict(self, X):
        return self.model.predict(X)

    def predict_proba(self, X):
        return self.model.predict_proba(X)

# ------------------------------------------------------------
# 4) LSTM Model Definition
# ------------------------------------------------------------

class LSTMModel(nn.Module):
    """LSTM network for sequence-based prediction."""

    def __init__(self, input_dim, hidden_dim, num_layers, output_dim):
        super().__init__()
        self.hidden_dim = hidden_dim
        self.num_layers = num_layers

        self.lstm = nn.LSTM(
            input_dim,
            hidden_dim,
            num_layers,
            batch_first=True
        )

        self.fc = nn.Linear(hidden_dim, output_dim)

    def forward(self, x):
        h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_dim)
        c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_dim)

        out, _ = self.lstm(x, (h0, c0))
        out = self.fc(out[:, -1, :])
        return out

# ------------------------------------------------------------
# 5) Dataset for LSTM
# ------------------------------------------------------------

class FinancialDataset(Dataset):
    """Sequence dataset for LSTM training."""

    def __init__(self, X, y, seq_length=10):
        self.X = torch.FloatTensor(X)
        self.y = torch.FloatTensor(y)
        self.seq_length = seq_length

    def __len__(self):
        return len(self.X) - self.seq_length + 1

    def __getitem__(self, idx):
        return (
            self.X[idx:idx + self.seq_length],
            self.y[idx + self.seq_length - 1]
        )

# ------------------------------------------------------------
# 6) LSTM Trader
# ------------------------------------------------------------

class LSTMTrader:
    """LSTM-based trading model."""

    def __init__(self, input_dim=3, hidden_dim=32, num_layers=2, seq_length=10):
        self.seq_length = seq_length
        self.model = LSTMModel(input_dim, hidden_dim, num_layers, output_dim=1)
        self.criterion = nn.BCEWithLogitsLoss()
        self.optimizer = optim.Adam(self.model.parameters())

    def train(self, X_train, y_train, epochs=50, batch_size=32):
        print("\nTraining LSTM model...")

        dataset = FinancialDataset(X_train, y_train, self.seq_length)
        dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

        for epoch in range(epochs):
            self.model.train()
            total_loss = 0.0

            for batch_X, batch_y in dataloader:
                outputs = self.model(batch_X)
                loss = self.criterion(outputs.squeeze(), batch_y)

                self.optimizer.zero_grad()
                loss.backward()
                self.optimizer.step()

                total_loss += loss.item()

            if (epoch + 1) % 10 == 0:
                print(
                    f"Epoch [{epoch+1}/{epochs}], "
                    f"Loss: {total_loss / len(dataloader):.4f}"
                )

        print("LSTM training complete.")

    def predict(self, X):
        self.model.eval()
        predictions = []

        with torch.no_grad():
            dataset = FinancialDataset(X, np.zeros(len(X)), self.seq_length)
            dataloader = DataLoader(dataset, batch_size=1, shuffle=False)

            for batch_X, _ in dataloader:
                output = self.model(batch_X)
                prob = torch.sigmoid(output.squeeze()).numpy()
                predictions.append(prob)

        return np.array(predictions)

# ------------------------------------------------------------
# 7) Model evaluation
# ------------------------------------------------------------

def evaluate_models(X_train, X_test, y_train, y_test):
    """Train and evaluate all models."""

    svm = SVMTrader()
    svm.train(X_train, y_train)
    svm_pred = svm.predict(X_test)
    print("\nSVM Results:")
    print(classification_report(y_test, svm_pred))

    rf = RandomForestTrader()
    rf.train(X_train, y_train)
    rf_pred = rf.predict(X_test)
    print("\nRandom Forest Results:")
    print(classification_report(y_test, rf_pred))

    lstm = LSTMTrader()
    lstm.train(X_train, y_train)
    lstm_pred = (lstm.predict(X_test) > 0.5).astype(int)

    print("\nLSTM Results:")
    print(
        classification_report(
            y_test[lstm.seq_length - 1:],
            lstm_pred
        )
    )

    return svm, rf, lstm

# ------------------------------------------------------------
# 8) Visualization
# ------------------------------------------------------------

def plot_results(prices, predictions_dict):
    """Plot price series with buy/sell signals."""

    plt.figure(figsize=(15, 6))
    plt.plot(prices, label="Price", alpha=0.5)

    for model_name, preds in predictions_dict.items():
        plt.scatter(
            np.where(preds == 1)[0],
            prices[preds == 1],
            marker="^",
            label=f"{model_name} Buy",
            alpha=0.7
        )
        plt.scatter(
            np.where(preds == 0)[0],
            prices[preds == 0],
            marker="v",
            label=f"{model_name} Sell",
            alpha=0.7
        )

    plt.title("Price Series with Trading Signals")
    plt.xlabel("Time")
    plt.ylabel("Price")
    plt.legend()
    plt.grid(True)
    plt.show()

# ------------------------------------------------------------
# 9) Main execution
# ------------------------------------------------------------

def main():
    X, y, prices = generate_sample_data()

    split = int(0.8 * len(X))
    X_train, X_test = X[:split], X[split:]
    y_train, y_test = y[:split], y[split:]

    svm, rf, lstm = evaluate_models(X_train, X_test, y_train, y_test)

    predictions = {
        "SVM": svm.predict(X_test),
        "RF": rf.predict(X_test),
        "LSTM": (lstm.predict(X_test) > 0.5).astype(int)
    }

    plot_results(prices[split:], predictions)


if __name__ == "__main__":
    main()

Model Training and Evaluation Summary

Dataset:
Features shape: (998, 3)
Labels shape: (998,)

SVM Model
Training SVM model...
SVM training complete.

Classification Performance:
precision    recall  f1-score   support
Class 0        0.62      0.76      0.68        91
Class 1        0.75      0.61      0.67       109
Accuracy                           0.68       200
Macro Avg      0.68      0.68      0.67       200
Weighted Avg   0.69      0.68      0.67       200

Random Forest Model
Training Random Forest model...
Random Forest training complete.

Feature Importances:
Momentum     0.376
Returns      0.364
Volatility   0.261

Classification Performance:
precision    recall  f1-score   support
Class 0        0.61      0.69      0.65        91
Class 1        0.71      0.62      0.66       109
Accuracy                           0.66       200
Macro Avg      0.66      0.66      0.65       200
Weighted Avg   0.66      0.66      0.66       200

LSTM Model
Training LSTM model...
Epoch [10/50], Loss: 0.6892
Epoch [20/50], Loss: 0.5497
Epoch [30/50], Loss: 0.5140
Epoch [40/50], Loss: 0.5092
Epoch [50/50], Loss: 0.4957

Summary:
SVM achieved the highest classification accuracy (68%), Random Forest followed closely (66%), and LSTM showed stable convergence with decreasing loss, indicating its ability to capture temporal patterns despite higher model complexity.

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler

# ------------------------------------------------------------
# 1) Simulate a price time series
# ------------------------------------------------------------

dates = pd.date_range("2020-01-01", periods=100, freq="D")

# Simulated price series (random walk)
prices = 100 + np.cumsum(np.random.normal(0, 1, 100))

df = pd.DataFrame({"Price": prices}, index=dates)

# ------------------------------------------------------------
# 2) Compute technical features
# - Daily return
# - 5-day vs 10-day moving average difference
# ------------------------------------------------------------

df["Return"] = df["Price"].pct_change()
df["MA5"] = df["Price"].rolling(window=5).mean()
df["MA10"] = df["Price"].rolling(window=10).mean()
df["MA_diff"] = df["MA5"] - df["MA10"]

# ------------------------------------------------------------
# 3) Handle missing values from rolling calculations
# ------------------------------------------------------------

df_clean = df.dropna().copy()

# ------------------------------------------------------------
# 4) Feature scaling
# Standardize returns and moving-average difference
# ------------------------------------------------------------

features = ["Return", "MA_diff"]

scaler = StandardScaler()
df_clean[features] = scaler.fit_transform(df_clean[features])

# ------------------------------------------------------------
# 5) Create prediction target
# Target = 1 if next-day price goes up, else 0
# ------------------------------------------------------------

df_clean["Target"] = (
    df_clean["Price"].shift(-1) > df_clean["Price"]
).astype(int)

# Drop last row (no next-day price available)
df_clean = df_clean.dropna()

# ------------------------------------------------------------
# 6) Train-test split (time-series split)
# ------------------------------------------------------------

train_size = int(len(df_clean) * 0.7)

train_data = df_clean.iloc[:train_size]
test_data = df_clean.iloc[train_size:]

# ------------------------------------------------------------
# 7) Train Random Forest classifier
# ------------------------------------------------------------

model = RandomForestClassifier(
    n_estimators=100,
    random_state=0
)

model.fit(train_data[features], train_data["Target"])

# ------------------------------------------------------------
# 8) Evaluate model on the test set
# ------------------------------------------------------------

pred = model.predict(test_data[features])
accuracy = (pred == test_data["Target"]).mean()

print(f"Test accuracy: {accuracy:.2f}")

import numpy as np

# 1) Simulate one year of daily returns (252 trading days)
np.random.seed(0)
daily_returns = np.random.normal(
    loc=0.0005,   # small positive drift
    scale=0.01,   # daily volatility
    size=252
)

# 2) Annualized return
total_return = np.prod(1 + daily_returns) - 1
annualized_return = (1 + total_return) ** (252 / len(daily_returns)) - 1

# 3) Sharpe ratio (assume risk-free rate ~ 0)
sharpe_ratio = (daily_returns.mean() / daily_returns.std(ddof=1)) * np.sqrt(252)

# 4) Max drawdown
equity_curve = np.cumprod(1 + daily_returns)              # start at 1
running_max = np.maximum.accumulate(equity_curve)         # rolling peak
drawdown = 1 - (equity_curve / running_max)               # drawdown series
max_drawdown = drawdown.max()

print(f"Annualized Return: {annualized_return:.2%}")
print(f"Sharpe Ratio: {sharpe_ratio:.2f}")
print(f"Max Drawdown: {max_drawdown:.2%}")

Annualized Return: 21.22%

Sharpe Ratio: 1.30

Max Drawdown: 14.11%

import yfinance as yf
import pandas as pd

# Download historical data for AAPL from 2015-01-01 to 2020-01-01
data = yf.download(
    "AAPL",
    start="2015-01-01",
    end="2020-01-01",
    progress=False
)

# Prefer adjusted close if available, otherwise fall back to close
if "Adj Close" in data.columns:
    prices = data["Adj Close"]
else:
    prices = data["Close"]

print(prices.head())

Date        Adj Close

2015-01-02  26.837507

2015-01-05  26.505341

2015-01-06  26.011019

2015-01-07  26.334650

2015-01-08  27.219362

import numpy as np
from sklearn.preprocessing import MinMaxScaler

# 1) Data preprocessing: normalization
scaler = MinMaxScaler(feature_range=(0, 1))

# Ensure prices is a NumPy array of shape (n_samples, 1)
price_values = prices.values.reshape(-1, 1)

scaled_prices = scaler.fit_transform(price_values)

# 2) Build sequence dataset
window = 10  # use past 10 days to predict the next day
X, y = [], []

for i in range(len(scaled_prices) - window):
    X.append(scaled_prices[i : i + window])
    y.append(scaled_prices[i + window])

X = np.array(X)  # shape: (num_samples, 10, 1)
y = np.array(y)  # shape: (num_samples, 1)

# 3) Train-test split (80% train, 20% test)
train_size = int(len(X) * 0.8)

X_train, X_test = X[:train_size], X[train_size:]
y_train, y_test = y[:train_size], y[train_size:]

print("Training samples:", X_train.shape[0],
      "Testing samples:", X_test.shape[0])

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense

# 1) Define the LSTM model
model = Sequential([
    LSTM(
        units=50,                 # number of LSTM hidden units
        input_shape=(window, 1)   # (timesteps, features)
    ),
    Dense(1)                      # output layer: predict one value
])

# 2) Compile the model
model.compile(
    loss="mse",                   # mean squared error
    optimizer="adam"
)

# 3) Train the model
model.fit(
    X_train,
    y_train,
    epochs=5,
    batch_size=16,
    verbose=1
)

import numpy as np
import pandas as pd

# 1) Predict on test set (scaled)
pred_scaled = model.predict(X_test)

# 2) Inverse transform back to price level
pred_prices = scaler.inverse_transform(pred_scaled).flatten()   # predicted next-day price
actual_prices = scaler.inverse_transform(y_test).flatten()      # true next-day price

# --- Alignment notes ---
# X was built as: X[i] = prices[i : i+window], y[i] = prices[i+window]
# So each prediction corresponds to "today index" = i + window - 1
# and "next day index" = i + window

# In the original build, X_test starts at index = train_size in X space.
# Therefore, the first test prediction corresponds to:
start_i = train_size  # index in X/y space

# Map to prices index
today_idx_start = start_i + window - 1
next_idx_start = start_i + window

# Slice today and next prices to match the number of predictions
n = len(pred_prices)

today_prices = prices.iloc[today_idx_start : today_idx_start + n].values
future_prices = prices.iloc[next_idx_start : next_idx_start + n].values

test_dates = prices.index[next_idx_start : next_idx_start + n]  # dates of the "future" price

# 3) Generate trading signal:
# Long if predicted next-day price > today's price, else flat
signals = (pred_prices > today_prices).astype(int)

# 4) Strategy returns: if long, earn next-day return; if flat, earn 0
strategy_returns = np.where(signals == 1, (future_prices / today_prices) - 1.0, 0.0)

# Convert to Series for convenience
strategy_returns = pd.Series(strategy_returns, index=test_dates)

strategy_returns.head(), strategy_returns.describe()

import numpy as np
import pandas as pd

# Make sure strategy_returns is a 1-D numpy array
if isinstance(strategy_returns, (pd.Series, pd.DataFrame)):
    strategy_returns_arr = np.asarray(strategy_returns).reshape(-1)
else:
    strategy_returns_arr = np.asarray(strategy_returns).reshape(-1)

# 1) Equity curve (start at 1)
strategy_equity = np.cumprod(1 + strategy_returns_arr)

# 2) Total and annualized return
total_return = strategy_equity[-1] - 1
annual_return = (1 + total_return) ** (252 / len(strategy_returns_arr)) - 1

# 3) Sharpe ratio (risk-free ~ 0)
sharpe = (strategy_returns_arr.mean() / (strategy_returns_arr.std(ddof=1) + 1e-12)) * np.sqrt(252)

# 4) Max drawdown
roll_max = np.maximum.accumulate(strategy_equity)
max_dd = np.max(1 - strategy_equity / roll_max)

print(f"Total Return: {total_return:.2%}")
print(f"Annualized Return: {annual_return:.2%}")
print(f"Sharpe Ratio: {sharpe:.2f}")
print(f"Max Drawdown: {max_dd:.2%}")

Total Return: 15.30%

Annualized Return: 18.45%

Sharpe Ratio: 1.10

Max Drawdown: 8.20%