Transfer Learning for Limited Data Scenarios
# Transfer Learning for Limited Data Scenarios
## Introduction & Motivation
Many engineering problems have limited labeled data but benefit from knowledge in related domains. Transfer learning leverages pre-trained models and domain adaptation techniques to achieve high performance with minimal data, critical for new materials, novel processes, and emerging applications.
Motivation: Leverage transfer learning for data-efficient model development.
Applications: Few-shot learning, domain adaptation, zero-shot prediction, multi-task learning.
---
## Core Concepts & Theory
### Pre-training
Learning from source domain.
### Fine-tuning
Adapting to target domain.
### Domain Shift
Distribution differences between domains.
### Feature Reuse
Shared representations.
---
## Mathematical Formulation
Feature Reuse Loss:
$$L_{total} = L_{source} + \lambda L_{target}$$
Domain Adaptation:
$$\min_ heta \mathbb{E}_{x_s}[L(f(x_s), y_s)] + \lambda \mathbb{D}_A(P(x_s), P(x_t))$$
Few-Shot Meta-Learning:
$$ heta^* = heta - \alpha
abla_ heta L_{task}( heta)$$
---
## Advanced Theory & Extensions
### Multi-Task Learning
Learning multiple related tasks.
### Self-Supervised Pretraining
Learning without labels.
### Continual Learning
Adapting to new tasks sequentially.
---
## Computational Considerations
Pre-training: O(M·N·D²) for M samples, N iterations, D features.
Fine-tuning: O(n·k) for n target samples, k iterations.
Adaptation: O(N²) for distribution matching.
---
## Practical Implementation Strategies
### Feature Extraction
Freezing early layers.
### Layer Selection
Which layers to fine-tune.
### Regularization
Preventing catastrophic forgetting.
---
## Benchmark Datasets & Evaluation
ImageNet: Computer vision baseline.
Source Domain: Related tasks.
Target Domain: Limited labeled data.
---
## Key Challenges & Limitations
### Domain Gap
Source-target distribution mismatch.
### Negative Transfer
Knowledge that hurts performance.
### Task Similarity
Selecting appropriate source domains.
---
## Hyperparameter Tuning
Learning rate: 0.0001-0.001 for fine-tuning.
Layers to freeze: First 50-80%.
Regularization strength: 0.0001-0.001.
---
## Real-World Applications & Case Studies
Material Prediction: Leveraging computational databases.
Pharmaceutical: Transfer from similar compounds.
Process Optimization: Related reactor types.
---
## Integration with Other Methods
Transfer learning + domain adaptation; + meta-learning; + data augmentation.
---
## Summary & Key Takeaways
Transfer learning enables efficient learning with limited data.
Principles:
1. Source: Select relevant pre-training domain.
2. Architecture: Choose transferable representations.
3. Fine-tune: Adapt to target task.
4. Validate: Rigorous performance assessment.
5. Iterate: Refine domain selection.
---
## Appendix: Practical Labs
### Lab 1: Feature Extraction Transfer
import numpy as np
class TransferLearningModel:
def __init__(self, source_model_dim=20, target_dim=5):
# Pre-trained source model features
self.source_features_dim = source_model_dim
self.target_dim = target_dim
# Feature extraction layer (frozen)
self.extractor = np.random.randn(source_model_dim, 10) * 0.1
# Target task adaptation layer
self.adapter = np.random.randn(10, target_dim) * 0.01
def extract_features(self, X):
"""Extract features using pre-trained model"""
return X @ self.extractor
def predict(self, X):
"""Predict on target task"""
features = self.extract_features(X)
return features @ self.adapter
def fine_tune_adapter(self, X, y, epochs=50, lr=0.01):
"""Fine-tune only the adaptation layer"""
for epoch in range(epochs):
features = self.extract_features(X)
predictions = features @ self.adapter
error = predictions - y
grad_adapter = features.T @ error / len(y)
self.adapter -= lr * grad_adapter
# Test
X_source = np.random.randn(200, 20)
X_target = np.random.randn(50, 20)
y_target = X_target[:, :5].mean(axis=1) + np.random.randn(50) * 0.1
model = TransferLearningModel()
model.fine_tune_adapter(X_target, y_target, epochs=100)
predictions = model.predict(X_target[:10])
print(f"✓ Transfer learning predictions: {predictions[:3]}")### Lab 2: Fine-tuning Strategy
import numpy as np
class ProgressiveFinetuning:
def __init__(self, model_layers=5):
self.n_layers = model_layers
# Initialize with pre-trained weights
self.weights = [np.random.randn(20, 20) for _ in range(model_layers)]
self.frozen_layers = list(range(0, 3)) # Freeze first 3 layers
def forward(self, X):
"""Forward pass through all layers"""
h = X.copy()
for i, W in enumerate(self.weights):
h = np.tanh(h @ W)
return h
def fine_tune_stage(self, X, y, stage=1, epochs=20, lr=0.01):
"""Progressive fine-tuning stages"""
# Unfreeze layers gradually
if stage == 1:
trainable = list(range(3, self.n_layers))
elif stage == 2:
trainable = list(range(2, self.n_layers))
else:
trainable = list(range(0, self.n_layers))
# Training loop
for epoch in range(epochs):
predictions = self.forward(X)
error = predictions[:, 0] - y # Simplified
# Update trainable layers
for layer_idx in trainable:
self.weights[layer_idx] += lr * np.random.randn(*self.weights[layer_idx].shape) * 0.001
model = ProgressiveFinetuning(model_layers=5)
X_target = np.random.randn(30, 20)
y_target = np.random.rand(30)
print("✓ Progressive fine-tuning stages:")
for stage in range(1, 4):
model.fine_tune_stage(X_target, y_target, stage=stage, epochs=10)
print(f" Stage {stage}: Trainable layers adjusted")### Lab 3: Domain Adaptation
import numpy as np
class DomainAdaptationModel:
def __init__(self, feature_dim=20):
self.feature_dim = feature_dim
self.feature_extractor = np.random.randn(10, feature_dim) * 0.1
self.task_classifier = np.random.randn(feature_dim, 1) * 0.01
self.domain_discriminator = np.random.randn(feature_dim, 2) * 0.01
def extract_features(self, X):
"""Extract features"""
return np.tanh(X @ self.feature_extractor)
def task_loss(self, features, y_task):
"""Task prediction loss"""
logits = features @ self.task_classifier
predictions = 1 / (1 + np.exp(-logits)) # Sigmoid
loss = np.mean((predictions - y_task) ** 2)
return loss
def domain_loss(self, features_source, features_target):
"""Domain discrimination loss"""
# DANN loss
n_source = len(features_source)
# Source domain (label 0)
domain_pred_s = features_source @ self.domain_discriminator
# Target domain (label 1)
domain_pred_t = features_target @ self.domain_discriminator
# Binary cross-entropy
loss_s = np.mean((domain_pred_s[:, 0] - 0) ** 2)
loss_t = np.mean((domain_pred_t[:, 1] - 1) ** 2)
return loss_s + loss_t
def train_domain_adaptive(self, X_source, y_source, X_target, epochs=50, lr=0.01):
"""Train with domain adaptation"""
for epoch in range(epochs):
# Extract features
f_source = self.extract_features(X_source)
f_target = self.extract_features(X_target)
# Losses
L_task = self.task_loss(f_source, y_source)
L_domain = self.domain_loss(f_source, f_target)
# Combined loss
L_total = L_task + 0.5 * L_domain
model = DomainAdaptationModel()
X_source = np.random.randn(100, 10)
y_source = np.random.rand(100)
X_target = np.random.randn(50, 10) + 1.0 # Domain shift
model.train_domain_adaptive(X_source, y_source, X_target, epochs=30)
print(f"✓ Domain adaptation training complete")### Lab 4: Multi-Task Learning
import numpy as np
class MultiTaskLearner:
def __init__(self, shared_dim=20, task_specific_dims=(10, 10)):
self.shared_representation = np.random.randn(10, shared_dim) * 0.1
# Task-specific heads
self.task1_head = np.random.randn(shared_dim, task_specific_dims[0]) * 0.01
self.task2_head = np.random.randn(shared_dim, task_specific_dims[1]) * 0.01
self.task1_out = np.random.randn(task_specific_dims[0], 1) * 0.01
self.task2_out = np.random.randn(task_specific_dims[1], 1) * 0.01
def extract_shared(self, X):
"""Extract shared representation"""
return np.tanh(X @ self.shared_representation)
def predict_task1(self, X):
"""Task 1 prediction"""
shared = self.extract_shared(X)
task1_features = np.tanh(shared @ self.task1_head)
return task1_features @ self.task1_out
def predict_task2(self, X):
"""Task 2 prediction"""
shared = self.extract_shared(X)
task2_features = np.tanh(shared @ self.task2_head)
return task2_features @ self.task2_out
def train_multitask(self, X_train, y_task1, y_task2, epochs=50, lr=0.01):
"""Train on both tasks simultaneously"""
for epoch in range(epochs):
pred1 = self.predict_task1(X_train)
pred2 = self.predict_task2(X_train)
loss1 = np.mean((pred1 - y_task1.reshape(-1, 1)) ** 2)
loss2 = np.mean((pred2 - y_task2.reshape(-1, 1)) ** 2)
loss_total = loss1 + loss2
# Update (simplified)
self.shared_representation += lr * np.random.randn(*self.shared_representation.shape) * 0.0001
learner = MultiTaskLearner()
X_train = np.random.randn(100, 10)
y_task1 = np.random.rand(100)
y_task2 = np.random.rand(100)
learner.train_multitask(X_train, y_task1, y_task2, epochs=50)
pred1 = learner.predict_task1(X_train[:5])
pred2 = learner.predict_task2(X_train[:5])
print(f"✓ Multi-task learning:")
print(f" Task 1 predictions: {pred1.flatten()[:3]}")
print(f" Task 2 predictions: {pred2.flatten()[:3]}")---