Spaces:
Sleeping
Sleeping
File size: 8,452 Bytes
3e48f76 626bad7 90ba9c1 626bad7 3e48f76 03a3525 3e48f76 626bad7 3e48f76 626bad7 3e48f76 626bad7 3e48f76 626bad7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 | import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
os.environ['TF_ENABLE_ONEDNN_OPTS'] = '0'
os.environ['CUDA_VISIBLE_DEVICES'] = '-1'
import tensorflow as tf
import numpy as np
import pickle
import copy
from sklearn.preprocessing import StandardScaler
class SimpleProgress(tf.keras.callbacks.Callback):
def on_epoch_end(self, epoch, logs=None):
logs = logs or {}
loss = logs.get("loss", 0)
mae = logs.get("mae", 0)
print(f"loss: {loss:.4f}, mae: {mae:.4f}", flush=True)
class TransferLearningModel:
def __init__(self, input_dim=7, seq_length=30, hidden_layers=(64, 32), learning_rate=0.001, dropout_rate=0.2):
"""
Initializes the Deep Neural Network using TensorFlow/Keras.
Args:
input_dim (int): Number of input features (default 7: Return, RSI, MA, etc.)
seq_length (int): Number of time steps in the input sequence.
hidden_layers (tuple): Structure of hidden layers (e.g., (64, 32) neurons).
learning_rate (float): How fast the model updates its weights.
dropout_rate (float): Fraction of the input units to drop to prevent overfitting.
"""
self.input_dim = input_dim
self.seq_length = seq_length
self.hidden_layers = hidden_layers
self.learning_rate = learning_rate
self.dropout_rate = dropout_rate
self.model = self._build_model()
self.scaler = StandardScaler()
def _build_model(self):
"""
Constructs the neural network architecture.
"""
# Sequential: A linear stack of layers
model = tf.keras.Sequential()
# 1. Input Layer
# Set the shape to natively accept 3D sequential data
model.add(tf.keras.layers.InputLayer(shape=(self.seq_length, self.input_dim)))
# 2. Hidden Layers
for i, units in enumerate(self.hidden_layers):
# LSTM: Long Short-Term Memory layer for sequential patterns.
# return_sequences must be True for all LSTM layers except the last one.
return_seq = i < len(self.hidden_layers) - 1
model.add(tf.keras.layers.LSTM(units, return_sequences=return_seq, name=f'lstm_layer_{i+1}'))
if self.dropout_rate > 0:
model.add(tf.keras.layers.Dropout(self.dropout_rate, name=f'dropout_layer_{i+1}'))
# 3. Output Layer
# 1 unit: We want a single prediction (Percentage Rise).
# Activation 'linear': Allows outputting any real number (positive or negative).
model.add(tf.keras.layers.Dense(1, activation='linear', name='output_layer'))
# 4. Compile
# Optimizer: Adam is the standard algorithm for adjusting weights.
# Loss: Mean Squared Error (MSE) is standard for regression.
model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=self.learning_rate),
loss='mean_squared_error',
metrics=['mae'])
return model
def _scale_data(self, X, fit=False):
"""Helper to properly scale 3D data by temporarily flattening to 2D."""
if X.ndim == 3:
samples, timesteps, features = X.shape
X_2d = X.reshape(-1, features)
if fit:
X_scaled = self.scaler.fit_transform(X_2d)
else:
X_scaled = self.scaler.transform(X_2d)
return X_scaled.reshape(samples, timesteps, features)
else:
if fit:
return self.scaler.fit_transform(X)
else:
return self.scaler.transform(X)
def update_scaler(self, X):
"""Incrementally updates the scaler's mean and std using partial_fit."""
X = np.array(X, dtype=np.float32)
if X.ndim == 3:
X_2d = X.reshape(-1, X.shape[2])
self.scaler.partial_fit(X_2d)
else:
self.scaler.partial_fit(X)
def train(self, X, y, epochs=50, batch_size=32, fit_scaler=False, patience=5, verbose=1):
"""
Trains the model on the provided data.
Transfer Learning Note:
In TensorFlow, calling .fit() repeatedly on the same model instance
does NOT reset the weights. It continues training from where it left off.
This is exactly what we need for Stage 1 -> Stage 2 -> Stage 3.
"""
# Ensure inputs are numpy arrays of float32 (standard for TF)
X = np.array(X, dtype=np.float32)
y = np.array(y, dtype=np.float32)
X = self._scale_data(X, fit=fit_scaler)
callbacks = []
if verbose > 0:
callbacks.append(SimpleProgress())
if patience > 0:
# Stop training if 'loss' doesn't improve for 'patience' epochs
early_stopping = tf.keras.callbacks.EarlyStopping(
monitor='loss',
patience=patience,
restore_best_weights=True,
verbose=0
)
callbacks.append(early_stopping)
# Set verbose=0 to suppress Keras default progress bar
history = self.model.fit(X, y, epochs=epochs, batch_size=batch_size, verbose=0, callbacks=callbacks)
return history
def predict(self, X):
"""
Returns binary class predictions (0 or 1).
"""
X = np.array(X, dtype=np.float32)
X = self._scale_data(X, fit=False)
# Return raw continuous values (predicted percentage return)
return self.model.predict(X, verbose=0)
def evaluate(self, X, y):
"""
Returns the Mean Absolute Error (MAE) of the model on the given data.
"""
X = np.array(X, dtype=np.float32)
y = np.array(y, dtype=np.float32)
X = self._scale_data(X, fit=False)
loss, mae = self.model.evaluate(X, y, verbose=0)
return mae
def save(self, filepath):
"""
Saves the entire model (architecture, weights, and optimizer state) to a file.
"""
filepath = str(filepath) # Ensure filepath is a string for Windows compatibility
self.model.save(filepath)
# Save the scaler alongside the model
scaler_path = filepath + ".scaler"
with open(scaler_path, "wb") as f:
pickle.dump(self.scaler, f)
def load(self, filepath):
"""
Loads a saved model from a file.
"""
filepath = str(filepath) # Ensure filepath is a string
self.model = tf.keras.models.load_model(filepath)
# After loading, update the wrapper's attributes to match the model's architecture
try:
# input_shape is typically (None, seq_length, input_dim) for LSTM
input_shape = self.model.input_shape
if len(input_shape) == 3:
self.seq_length = input_shape[1]
self.input_dim = input_shape[2]
elif len(input_shape) == 2: # Fallback for non-sequential models
self.input_dim = input_shape[1]
except Exception as e:
print(f"Warning: Could not auto-determine input shape from loaded model: {e}")
# Load the scaler
scaler_path = filepath + ".scaler"
if os.path.exists(scaler_path):
with open(scaler_path, "rb") as f:
self.scaler = pickle.load(f)
else:
print(f"Warning: Scaler file not found at {scaler_path}. Model will use uninitialized scaler!")
def copy(self):
"""
Creates a deep copy of the model in memory.
Useful for Transfer Learning where we want to branch off a base model.
"""
new_instance = TransferLearningModel(
input_dim=self.input_dim,
seq_length=self.seq_length,
hidden_layers=self.hidden_layers,
learning_rate=self.learning_rate,
dropout_rate=self.dropout_rate
)
new_instance.model.set_weights(self.model.get_weights())
new_instance.scaler = copy.deepcopy(self.scaler)
return new_instance
@staticmethod
def exists(filepath):
"""
Checks if the model and its scaler exist at the given filepath.
"""
filepath = str(filepath)
scaler_path = filepath + ".scaler"
return os.path.exists(filepath) and os.path.exists(scaler_path) |