MetaTrader 5 Machine Learning Blueprint (Part 6): Engineering a Production-Grade Caching System

• Introduction
• Part I: Architectural Foundations
• Part II: The Robust Cacheable Decorator
• Part III: Advanced Caching Patterns
• Part IV: Performance Monitoring and Optimization
• Part V: Integration with Existing Projects
• Conclusion: From Research Velocity to Execution Speed
• Code Repository
• Module Files

Introduction

In our previous installments of the Machine Learning Blueprint series, we’ve built a robust pipeline for financial machine learning—from ensuring data integrity against look-ahead bias to implementing sophisticated labeling methods like Triple-Barrier and Trend-Scanning. However, as our strategies or ML models—as with sequentially bootstrapped random forests—grow more complex, we face a critical challenge: computational bottlenecks that prevent rapid iteration.

You've built a promising mean reversion strategy. Your backtest shows a Sharpe ratio of 1.8, consistent profits across market regimes, and clean equity curves. You're ready to optimize parameters, test different lookback periods, and validate with walk-forward analysis.

Then reality hits.  

Each parameter combination takes 6 minutes to compute. You want to test 50 variations. That's 5 hours of waiting. Change your feature engineering? Another 5 hours. Add a new indicator? You get the idea.

The real cost isn't just time—it's lost opportunities. While you wait for computations, you can't iterate, can't test new ideas, can't improve your edge. Your development velocity grinds to a halt.

This is the problem that killed my early trading strategies. I would spend entire weekends running backtests, only to realize Monday morning that I'd made a simple mistake in my code. More waiting. More frustration.

There had to be a better way.

This article shows you how to eliminate this bottleneck using intelligent caching. By the end, you'll understand how to:

• Reduce strategy optimization time from hours to minutes
• Test 50+ parameter combinations in the time it used to take for 5
• Iterate on features and models without recomputing everything

Let's start with the problem you're facing right now.

The Computational Pipeline Visualization

Before implementing caching, it's important to understand how much time each operation in the ML pipeline actually takes. The diagram below shows the full path from loading ticks to the final backtest—and how many seconds are spent on each step without caching.

Now let's look at how the same sequence of operations works with AFML caching enabled. Note that most steps are no longer recalculated but loaded instantly. The difference in speed is evident in the diagram below.

Why Generic Caching Fails for Financial ML

Let's see why the obvious solution doesn't work.

It might seem like a no-brainer to just use the standard lru_cache. Below is a simple example demonstrating that Python's built-in cache completely breaks down when working with financial data structures like Pandas Series and DataFrame.

# This is USELESS for financial data. Don't do this.
from functools import lru_cache

@lru_cache(maxsize=128)
def compute_rsi(prices, period=14):
    delta = prices.diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)
    
    avg_gain = gain.ewm(alpha=1/period, adjust=False).mean()
    avg_loss = loss.ewm(alpha=1/period, adjust=False).mean()
    
    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    return rsi

compute_rsi(bb_df.close, period=14)

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
Cell In[25], line 18
     15     rsi = 100 - (100 / (1 + rs))
     16     return rsi
---> 18 compute_rsi(bb_df.close, period=14)

TypeError: unhashable type: 'Series'

While Python offers functools.lru_cache, it’s fundamentally inadequate for financial ML:

1. Memory-only storage—cache disappears when Python exits.
2. No persistence across sessions—rerun everything after each restart.
3. No automatic invalidation—stale cache when code changes.
4. Poor handling of NumPy/Pandas—hashing issues with arrays.
5. No distributed caching—can’t share cache across processes.
6. No financial data awareness—doesn’t understand timestamps or look-ahead bias.

Part I: Architectural Foundations

The Core Design Principles

Our AFML caching system is built on three fundamental pillars:

Let’s understand why each component exists and how they work together.

Challenge #1: Persistent Storage

Let's start with the first and most obvious limitation—the standard cache lives only in RAM. As soon as you restart Python, all results disappear. Here's a short demonstration of what the "normal way" looks like and why it's useless in real-world ML pipelines.

# Traditional approach - memory only
from functools import lru_cache

@lru_cache(maxsize=128)
def compute_features(data):
    # Expensive computation
    return features  # Lost on restart!

AFML solves this problem simply: we move the cache to disk. Joblib persists calculation results between Python runs, so even after a restart, you instantly get the previously calculated data. An example implementation is below.

# AFML approach - persistent
from joblib import Memory
from appdirs import user_cache_dir


memory = Memory(location=user_cache_dir("afml"), verbose=0)

@memory.cache
def compute_features(data):
    # Expensive computation
    return features  # Saved to disk automatically!

Challenge #2: Hashing Complex Financial Data Structures

And then there's an even bigger problem—financial data can't be hashed using standard methods. Pandas DataFrames, NumPy arrays, and dates are all "unhashable", and a regular cache simply won't work. Here's an example of how this breaks down.

# This fails!
import pandas as pd
from functools import lru_cache

@lru_cache
def bad_cache_example(df: pd.DataFrame):
    return df.mean()

df = pd.DataFrame({'price': [100, 101, 102]})
bad_cache_example(df)  # TypeError: unhashable type: 'DataFrame'

To ensure the cache works correctly, AFML creates its own key generator. It can "understand" DataFrames: their structure, data types, columns, indexes, and, most importantly, time range. Here's what a custom hashing implementation looks like.

# afml/cache/robust_cache_keys.py

class CacheKeyGenerator:
    """Generate collision-resistant cache keys for ML data structures."""
    
    @staticmethod
    def _hash_dataframe(df: pd.DataFrame, name: str) -> str:
        """
        Hash DataFrame with attention to:
        1. Shape and structure
        2. Column names and types
        3. Index (especially DatetimeIndex)
        4. Actual data content
        """
        parts = [
            f"shape_{df.shape}",
            f"cols_{hashlib.md5(str(tuple(df.columns)).encode()).hexdigest()[:8]}",
            f"dtypes_{hashlib.md5(str(tuple(df.dtypes)).encode()).hexdigest()[:8]}",
        ]
        
        # Special handling for DatetimeIndex (critical for financial data!)
        if isinstance(df.index, pd.DatetimeIndex):
            # Hash: start date, end date, and length
            # This catches both data changes AND temporal shifts
            parts.append(f"idx_dt_{df.index[0]}_{df.index[-1]}_{len(df.index)}")
        else:
            idx_hash = hashlib.md5(str(tuple(df.index)).encode()).hexdigest()[:8]
            parts.append(f"idx_{idx_hash}")
        
        # For large DataFrames, sample for performance
        if df.size > 10000:
            # Sample ~100 rows evenly distributed
            sample_rows = df.iloc[::max(1, len(df) // 100)]
            content_hash = hashlib.md5(sample_rows.values.tobytes()).hexdigest()[:8]
        else:
            # Hash full content for small DataFrames
            content_hash = hashlib.md5(df.values.tobytes()).hexdigest()[:8]
        
        parts.append(f"data_{content_hash}")
        
        return f"{name}_df_{'_'.join(parts)}"

To understand the value of custom hashing, let's compare how standard Python caches data and how AFML does it. Below is an illustration of why the standard approach leads to look-ahead bias, while AFML does not.

Challenge #3: Automatic Cache Invalidation When Code Changes

Even a perfect cache becomes useless if you edit your code and old, stale results remain in the cache. AFML automatically tracks any changes to functions and flushes only the portion of the cache that is out of date. Let's walk through how this is implemented.

AFML stores a hash of the function's source code and the file's last modification date. If anything changes, the cache for that function is automatically reset. Here's how this mechanism works internally.

# afml/cache/selective_cleaner.py
class FunctionTracker:
    """
    Tracks function signatures and source code hashes.
    Automatically detects when functions change.
    """
    
    def track_function(self, func) -> bool:
        """
        Returns True if function has changed since last tracking.
        """
        func_name = f"{func.__module__}.{func.__qualname__}"
        
        # Get current function metadata
        current_hash = self._get_function_hash(func)      # Hash source code
        current_mtime = self._get_file_mtime(func)        # File modification time
        
        # Compare with stored metadata
        stored = self.tracked_functions.get(func_name, {})
        stored_hash = stored.get("hash")
        stored_mtime = stored.get("mtime")
        
        # Function changed if EITHER hash or mtime differs
        has_changed = (
            current_hash != stored_hash or 
            current_mtime != stored_mtime or 
            stored_hash is None  # New function
        )
        
        if has_changed:
            # Update tracking data
            self.tracked_functions[func_name] = {
                "hash": current_hash,
                "mtime": current_mtime,
                "module": func.__module__,
            }
            self._save_tracking_data()
            
        return has_changed
    
    def _get_function_hash(self, func) -> Optional[str]:
        """Hash function source code."""
        try:
            source = inspect.getsource(func)
            return hashlib.md5(source.encode()).hexdigest()
        except (OSError, TypeError):
            return None

To see this in action, consider a real-world scenario: a function initially contains an error, then we fix it—and AFML automatically detects the change, clearing only the relevant portion of the cache.

Part II: The Robust Cacheable Decorator

All AFML capabilities are combined in a single powerful factory decorator. It creates the desired caching type: regular, temporary, data-tracking, code-tracking, and so on. Here's the source code for this factory.

# afml/cache/robust_cache_keys.py

def create_robust_cacheable(
    track_data_access: bool = False,
    dataset_name: Optional[str] = None,
    purpose: Optional[str] = None,
    use_time_awareness: bool = False,
):
    """
    Factory function to create robust cacheable decorators.
    This is where all the magic comes together.
    """
    from functools import wraps
    from . import cache_stats, memory
    from .cache_monitoring import get_cache_monitor
    
    def decorator(func):
        func_name = f"{func.__module__}.{func.__qualname__}"
        cached_func = memory.cache(func)  # Use joblib for persistence
        seen_signatures = set()           # Track cache hits/misses
        monitor = get_cache_monitor()     # Performance monitoring
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Step 1: Generate cache key
            try:
                if use_time_awareness:
                    cache_key = TimeSeriesCacheKey.generate_key_with_time_range(
                        func, args, kwargs
                    )
                else:
                    cache_key = CacheKeyGenerator.generate_key(func, args, kwargs)
                
                # Step 2: Track hit/miss
                if cache_key in seen_signatures:
                    cache_stats.record_hit(func_name)
                    is_hit = True
                else:
                    cache_stats.record_miss(func_name)
                    seen_signatures.add(cache_key)
                    is_hit = False
                    
            except Exception as e:
                logger.warning(f"Cache key generation failed: {e}")
                cache_stats.record_miss(func_name)
                cache_key = None
                is_hit = False
            
            # Step 3: Track data access if requested (prevent look-ahead bias)
            if track_data_access:
                try:
                    from .data_access_tracker import get_data_tracker
                    _track_dataframe_access(
                        get_data_tracker(), args, kwargs, dataset_name, purpose
                    )
                except Exception as e:
                    logger.warning(f"Data tracking failed: {e}")
            
            # Step 4: Track access time (for monitoring)
            monitor.track_access(func_name)
            
            # Step 5: Execute function with timing
            start_time = time.time()
            try:
                result = cached_func(*args, **kwargs)
                
                # Track computation time for misses
                if not is_hit:
                    computation_time = time.time() - start_time
                    monitor.track_computation_time(func_name, computation_time)
                
                return result
                
            except (EOFError, pickle.PickleError, OSError) as e:
                # Handle cache corruption gracefully
                logger.warning(f"Cache corruption: {type(e).__name__} - recomputing")
                
                # Clear corrupted cache
                if cache_key is not None:
                    _clear_corrupted_cache(cached_func, cache_key)
                
                # Execute function directly
                return func(*args, **kwargs)
                
        wrapper._afml_cacheable = True
        return wrapper
    
    return decorator

# Standard decorators
robust_cacheable = create_robust_cacheable(use_time_awareness=False)
time_aware_cacheable = create_robust_cacheable(use_time_awareness=True)

# Data tracking decorators
data_tracking_cacheable = lambda dataset_name, purpose: create_robust_cacheable(
    track_data_access=True, dataset_name=dataset_name, purpose=purpose, use_time_awareness=False
)

time_aware_data_tracking_cacheable = lambda dataset_name, purpose: create_robust_cacheable(
    track_data_access=True, dataset_name=dataset_name, purpose=purpose, use_time_awareness=True
)

Part III: Advanced Caching Patterns

Pattern #1: Time-Aware Caching for Walk-Forward Analysis

In walk-forward validation, it's critical that the cache distinguishes between different time periods. Without this, collisions and incorrect results can easily occur. The diagram below shows how this works in practice.

Time Series Data (2024):
─────────────────────────────────────────────────────────────────────
Jan  Feb  Mar  Apr  May  Jun  Jul  Aug  Sep  Oct  Nov  Dec
├────┼────┼────┼────┼────┼────┼────┼────┼────┼────┼────┼────┤
│    │    │    │    │    │    │    │    │    │    │    │    │
▼────▼────▼────▼────▼────▼────▼────▼────▼────▼────▼────▼────▼

Walk-Forward Splits:
─────────────────────────────────────────────────────────────────────

Split 1:
        Train Period: Jan-Mar              Test Period: Apr
        ├─────────────────────────────┤    ├───────┤

Split 2:
            Train Period: Feb-Apr              Test Period: May
            ├─────────────────────────────┤    ├───────┤

Split 3:
                Train Period: Mar-May              Test Period: Jun
                ├─────────────────────────────┤    ├───────┤

The example below illustrates how using regular caching with two different time periods can produce the same hash. This leads to the model being trained on the wrong data, and look-ahead bias occurs.

@robust_cacheable  # Generic caching
def train_model(data, params):
    return model

Split 1: data = data['2024-01':'2024-03']
         Cache Key: hash(data.values)
         └──> "a7f3e8d92c1b..."        ◄──┐
                                          │
Split 2: data = data['2024-02':'2024-04'] │
         Cache Key: hash(data.values)     │
         └──> "a7f3e8d92c1b..."        ◄──┘ COLLISION! 
              └──> Cache hit on DIFFERENT time period
              └──> Model trained on wrong data!

Time-aware cache adds a time range during key generation, completely eliminating such collisions. See how the cache now correctly distinguishes between periods.

@time_aware_cacheable  # Includes temporal info
def train_model(data, params):
    return model

Split 1: data = data['2024-01':'2024-03']
         Cache Key: hash(data.values + "2024-01_2024-03")
         └──> "a7f3_time_2024-01_2024-03"
         
Split 2: data = data['2024-02':'2024-04']
         Cache Key: hash(data.values + "2024-02_2024-04")
         └──> "b9e4_time_2024-02_2024-04"  ← Different!
              └──> Cache miss (correct)
              └──> Train new model for this period ✓

Split 3: data = data['2024-03':'2024-05']
         Cache Key: hash(data.values + "2024-03_2024-05")
         └──> "c7d1_time_2024-03_2024-05"  ← Different!
              └──> Cache miss (correct)
              └──> Train new model for this period ✓

Here is an implementation of a class that adds timestamps to cache keys.

class TimeSeriesCacheKey(CacheKeyGenerator):
    """Extended cache key generator with time-series awareness."""
    
    @staticmethod
    def generate_key_with_time_range(
        func, 
        args: tuple, 
        kwargs: dict, 
        time_range: Tuple[pd.Timestamp, pd.Timestamp] = None
    ) -> str:
        """
        Generate cache key that includes time range information.
        Critical for preventing temporal data leakage.
        """
        base_key = CacheKeyGenerator.generate_key(func, args, kwargs)
        
        if time_range is None:
            # Try to extract time range from data
            time_range = TimeSeriesCacheKey._extract_time_range(args, kwargs)
        
        if time_range:
            start, end = time_range
            time_hash = f"time_{start}_{end}"
            return f"{base_key}_{time_hash}"
        
        return base_key

# Usage in walk-forward validation
@time_aware_cacheable
def train_on_period(data: pd.DataFrame, params: dict) -> Model:
    """
    Train model on specific time period.
    Cache is automatically keyed by time range!
    """
    model = RandomForestClassifier(**params)
    model.fit(data.drop('target', axis=1), data['target'])
    return model

# Walk-forward loop
for train_start, train_end, test_start, test_end in walk_forward_splits:
    # Each period is cached independently
    train_data = data.loc[train_start:train_end]
    model = train_on_period(train_data, params)  # Cached per period
    
    test_data = data.loc[test_start:test_end]
    predictions = model.predict(test_data)

Pattern #2: Cross-Validation Caching with Sklearn Estimators

Cross-validation caching is a separate issue. Sklearn classifiers contain internal state that can't be hashed directly. AFML solves this by hashing only the model type and its parameters. The key function is below.

# afml/cache/cv_cache.py

def _hash_classifier(clf: BaseEstimator) -> str:
    """
    Generate stable hash for sklearn classifier.
    
    KEY INSIGHT: Hash the type + parameters, NOT the trained state!
    """
    try:
        clf_type = type(clf).__name__
        params = clf.get_params(deep=True)
        
        # Filter out non-serializable params
        serializable_params = {}
        for k, v in params.items():
            try:
                json.dumps(v)  # Test if JSON serializable
                serializable_params[k] = v
            except (TypeError, ValueError):
                # Use type name for non-serializable params
                serializable_params[k] = f"<{type(v).__name__}>"
        
        # Create stable hash
        param_str = json.dumps(serializable_params, sort_keys=True)
        combined = f"{clf_type}_{param_str}"
        return hashlib.md5(combined.encode()).hexdigest()[:12]
        
    except Exception as e:
        logger.debug(f"Failed to hash classifier: {e}")
        return f"clf_{type(clf).__name__}_{id(clf)}"

Here's what a decorator for caching cross-validation results looks like. It saves a tremendous amount of time when selecting hyperparameters.

@cv_cacheable
def ml_cross_val_score(
    classifier: BaseEstimator,
    X: pd.DataFrame,
    y: pd.Series,
    cv_gen,  # PurgedKFold, TimeSeriesSplit, etc.
    sample_weight_train: Optional[np.ndarray] = None,
    scoring: str = 'neg_log_loss'
) -> np.ndarray:
    """
    Cross-validation with proper caching.
    
    Caches based on:
    - Classifier type and parameters (not trained state)
    - Data content (X, y)
    - CV generator configuration
    - Sample weights
    """
    scores = []
    
    for train_idx, test_idx in cv_gen.split(X):
        X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
        y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
        
        # Train fresh model (not from cache)
        model = clone(classifier)
        
        if sample_weight_train is not None:
            model.fit(X_train, y_train, 
                     sample_weight=sample_weight_train[train_idx])
        else:
            model.fit(X_train, y_train)
        
        score = model.score(X_test, y_test)
        scores.append(score)
    
    return np.array(scores)

Pattern #3: Preventing Test Set Contamination

In ML development, one of the most hidden mistakes is accidentally using test data during training. AFML monitors every access to the dataset and alerts you to any leaks.

Here's how AFML records every data access, capturing the time range, purpose (train, test, validate), and source of the call. Based on these logs, the system then generates a report on potential leaks.

# afml/cache/data_access_tracker.py

class DataAccessTracker:
    """
    Track every data access to detect test set contamination.
    
    This is CRITICAL for preventing data snooping bias.
    """
    
    def log_access(
        self,
        dataset_name: str,
        start_date: pd.Timestamp,
        end_date: pd.Timestamp,
        purpose: str,  # 'train', 'test', 'validate', 'optimize'
        data_shape: Optional[Tuple[int, int]] = None,
    ):
        """Log a dataset access with full temporal metadata."""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "dataset": dataset_name,
            "start_date": str(start_date),
            "end_date": str(end_date),
            "purpose": purpose,
            "data_shape": str(data_shape) if data_shape else None,
            "caller": self._get_caller_info(),  # Stack trace
        }
        
        self.access_log.append(entry)
        logger.debug(
            f"Logged access: {dataset_name} [{start_date} to {end_date}] "
            f"for {purpose}"
        )

# Usage with caching decorator
@time_aware_data_tracking_cacheable(dataset_name="eur_usd_2024", purpose="test")
def evaluate_on_test_set(test_data: pd.DataFrame, model) -> dict:
    """
    Evaluate model on test set.
    Access is logged automatically!
    """
    predictions = model.predict(test_data)
    metrics = calculate_metrics(predictions, test_data['target'])
    return metrics

To check for contamination during our model development process we call print_contamination_report() as shown:

# Later, check for contamination
from afml.cache import print_contamination_report

print_contamination_report()

Part IV: Performance Monitoring and Optimization

Cache Health Monitoring

AFML also provides built-in cache health monitoring tracking:

• which functions are most frequently cached
• where cache misses occur
• what the cache size is
• whether there are any suspicious situations

from afml.cache import print_cache_health


print_cache_health()

Part V: Integration with Existing Projects

To start using the system, simply import the module and use the desired decorator. No complicated configuration required—here's an example of what it looks like in a real project.

Run the code below to clone from your terminal, or download cache.zip:

git clone https://github.com/pnjoroge54/Machine-Learning-Blueprint.git
cd Machine-Learning-Blueprint/afml/cache

my_package/
├── cache/
│   ├── __init__.py
│   ├── backtest_cache.py
│   ├── cache_monitoring.py
│   ├── cv_cache.py
│   ├── data_access_tracker.py
│   ├── robust_cache_keys.py
│   ├── selective_cleaner.py
│   └── mql5_bridge.py

For full details on implementing this caching system with your project see user_guide.py.

Conclusion: From Research Velocity to Execution Speed

The AFML caching system fundamentally transforms financial ML development by addressing the two critical bottlenecks that separate research from production: iteration speed and execution latency.

Research Velocity: The Foundation of Alpha Generation

In the research phase, our caching architecture enables unprecedented exploration:

• Rapid Feature Engineering: Test dozens of technical indicator combinations, RSI periods, and volatility calculations without recomputing base transformations. The system intelligently caches intermediate results, allowing you to iterate on feature selection rather than waiting for computation.

• Exhaustive Strategy Validation: Run PurgedKFold cross-validation across hundreds of parameter combinations. Each data fold’s features and labels remain cached, enabling rigorous backtesting at scale without the temporal data leakage that plagues traditional approaches.

• Multi-Timeframe Analysis: Experiment with complex feature interactions across 1-minute, 5-minute, and hourly timeframes. The time-aware caching ensures each period’s computations remain independent and reproducible.

• Labeling Strategy Optimization: Compare Triple-Barrier, Trend-Scanning, and Meta-Labeling approaches without recalculating expensive bar sampling operations.

The Complete Pipeline: Research to Execution

Our caching architecture creates a seamless transition from experimental research to production trading:

1. Research Phase: Use Python’s rich ecosystem with AFML caching for rapid experimentation and validation
2. Model Export: Convert validated models to ONNX format for dependency-free deployment
3. Feature Pipeline Migration: Implement critical feature computations natively in MQL5 with parallel caching
4. Production Deployment: Execute strategies with microsecond latency while maintaining research integrity

Data Integrity: The Unseen Advantage

Beyond performance, the system’s automatic data access tracking prevents the subtle contamination that undermines production ML systems. Complete audit trails ensure you never accidentally optimize on test data, while temporal awareness guarantees walk-forward validation integrity.

Looking Ahead: The Final Frontier

In our next installment, we complete the pipeline with:

1. ONNX Model Deployment: Export scikit-learn and custom models to run natively in MQL5 without Python dependencies
2. MQL5 Inference Engine: Build ultra-low-latency prediction systems that operate in microseconds
3. Hybrid Feature Management: Design intelligent systems where complex features compute in Python during research, then migrate to MQL5 for production
4. Real-Time Model Monitoring: Implement drift detection and performance tracking within the MQL5 execution environment

The result is a complete ecosystem where you can research with Python’s flexibility, deploy with MQL5’s performance, and maintain scientific rigor throughout the entire lifecycle. This isn’t just about faster computation—it’s about creating a framework where sophisticated ML strategies can actually work in live markets.

Code Repository

All code from this article is available in the Machine-Learning-Blueprint repository. Run the code below to clone from your terminal:

git clone https://github.com/pnjoroge54/Machine-Learning-Blueprint.git
cd Machine-Learning-Blueprint/afml/cache

Module Files