Author: Mike Amega
Date: November 13, 2025
Version: 1.0
License: MIT License (for reference implementation)
Patent Status: Prior art established through defensive publication
This technical whitepaper provides complete implementation specifications for the EARCP (Ensemble Auto-Régulé par Cohérence et Performance) architecture. This document serves as:
- Prior art establishment - Timestamped public disclosure preventing future patent claims by third parties
- Implementation reference - Complete algorithmic specifications for reproducible implementations
- Technical documentation - Detailed architecture decisions and design rationales
Intellectual Property Notice: This work represents original research by Mike Amega. Public disclosure through this document establishes prior art as of November 13, 2025, protecting against future patent claims while preserving the author's rights to commercialize, license, or patent this technology.
- Architecture Overview
- Mathematical Foundations
- Complete Algorithm Specification
- Implementation Details
- Hyperparameter Configuration
- Performance Optimization
- Extension Points
- Reference Implementation
- Validation Protocol
- IP Protection Strategy
EARCP introduces a novel dual-signal weighting mechanism that combines:
Weight[i,t] = f(Performance[i,t], Coherence[i,t])
Where:
- Performance[i,t]: Expert i's predictive accuracy (exploitation signal)
- Coherence[i,t]: Expert i's agreement with ensemble (exploration signal)
- f(·): Multiplicative update function with exponential transformation
- Dynamic Adaptation: Continuous weight adjustment based on realized performance
- Coherence-Aware: Leverages inter-model agreement as reliability signal
- Theoretical Guarantees: Provable regret bounds O(√(T log M))
- Practical Stability: Multiple stabilization mechanisms (floors, clipping, smoothing)
Input State (x_t)
|
|-----> [Expert 1: CNN] -------> p_1,t
|-----> [Expert 2: LSTM] ------> p_2,t
|-----> [Expert 3: Transformer] -> p_3,t
|-----> [Expert 4: DQN] -------> p_4,t
|
v
[EARCP Weight Computation Module]
|
|---> Performance Tracker (EMA)
|---> Coherence Analyzer (Agreement Matrix)
|---> Score Combiner (β-weighted)
|---> Weight Updater (Exponential + Normalization)
|
v
Weighted Combination: ŷ_t = Σ w_i,t · p_i,t
Given:
- Time horizon T
- M expert models {m₁, ..., m_M}
- Input space 𝒳
- Output space 𝒴
- Loss function L: 𝒴 × 𝒴 → ℝ₊
Objective: Learn time-varying weights w_t ∈ Δ_M (probability simplex) to minimize:
Regret_T = Σₜ L(ŷ_t, y_t) - min_i Σₜ L(p_i,t, y_t)
Exponential moving average of negative loss:
P_i,t = α_P · P_i,t-1 + (1 - α_P) · (-ℓ_i,t)
Properties:
- Higher values indicate better recent performance
- α_P ∈ (0,1) controls memory length
- Initialization: P_i,0 = 0
Design rationale: EMA provides smooth tracking while emphasizing recent performance, crucial for non-stationary environments.
For classification (discrete predictions):
C_i,t = (1/(M-1)) · Σ_{j≠i} 𝟙{argmax(p_i,t) = argmax(p_j,t)}
For regression (continuous predictions):
C_i,t = (1/(M-1)) · Σ_{j≠i} exp(-γ · ||p_i,t - p_j,t||²)
Properties:
- C_i,t ∈ [0,1] measures agreement with other experts
- Higher values indicate consensus
- γ > 0 controls sensitivity to disagreement
Design rationale: Coherence serves as collective wisdom signal - when experts agree, predictions are more reliable.
C̄_i,t = α_C · C̄_i,t-1 + (1 - α_C) · C_i,t
Properties:
- Reduces noise in coherence signal
- α_C ∈ (0,1) controls smoothing
- Initialization: C̄_i,0 = 0.5 (neutral)
Normalize performance and coherence:
P̃_i,t = (P_i,t - min_j P_j,t) / (max_j P_j,t - min_j P_j,t + ε)
C̃_i,t = (C̄_i,t - min_j C̄_j,t) / (max_j C̄_j,t - min_j C̄_j,t + ε)
Combine with β-weighting:
s_i,t = β · P̃_i,t + (1 - β) · C̃_i,t
Properties:
- β ∈ [0,1] controls performance-coherence balance
- ε = 10⁻⁸ prevents division by zero
- s_i,t ∈ [0,1] by construction
Design rationale: Normalization ensures comparable scales; β allows tuning exploitation-exploration trade-off.
Exponential transformation:
w̃_i,t = exp(η_s · clip(s_i,t, -s_max, s_max))
Normalization:
w'_i,t = w̃_i,t / Σ_j w̃_j,t
Floor enforcement:
w_i,t = max(w'_i,t, w_min)
Final renormalization:
w_i,t ← w_i,t / Σ_j w_j,t
Properties:
- η_s > 0 controls sensitivity (typical: 3-7)
- s_max prevents overflow (typical: 10)
- w_min ensures exploration (typical: 0.05)
Design rationale: Exponential amplifies differences while floor prevents premature expert elimination.
Theorem (EARCP Regret Bound):
Under assumptions:
- Bounded losses: ℓ_i,t ∈ [0,1]
- Convex loss function
- Lipschitz continuity
With β=1 (pure performance), α_P=0 (no smoothing), η = √(2log M / T):
Regret_T ≤ √(2T log M)
With 0 < β < 1 (coherence included):
Regret_T ≤ (1/β) · √(2T log M)
Proof sketch: Reduction to multiplicative weights update algorithm (Hedge). The performance component drives convergence while coherence acts as side information, scaling effective learning rate by β.
class EARCP:
"""
Ensemble Auto-Régulé par Cohérence et Performance
Original work by Mike Amega, 2025
Protected by defensive publication
"""
def __init__(self, experts, alpha_P=0.9, alpha_C=0.85, beta=0.7,
eta_s=5.0, w_min=0.05, performance_window=50):
"""
Initialize EARCP ensemble.
Args:
experts: List of M expert models with .predict() method
alpha_P: Performance EMA smoothing (0,1)
alpha_C: Coherence EMA smoothing (0,1)
beta: Performance-coherence balance [0,1]
eta_s: Exponential sensitivity (>0)
w_min: Minimum weight floor (>=0)
performance_window: History window for statistics
"""
self.experts = experts
self.M = len(experts)
# Hyperparameters
self.alpha_P = alpha_P
self.alpha_C = alpha_C
self.beta = beta
self.eta_s = eta_s
self.w_min = w_min
self.performance_window = performance_window
# State variables
self.weights = np.ones(self.M) / self.M # Initialize uniform
self.P = np.zeros(self.M) # Performance scores
self.C_bar = np.full(self.M, 0.5) # Smoothed coherence
# History for normalization
self.P_history = [deque(maxlen=performance_window)
for _ in range(self.M)]
self.C_history = [deque(maxlen=performance_window)
for _ in range(self.M)]
self.t = 0 # Time step counter
def predict(self, x):
"""
Generate ensemble prediction.
Args:
x: Input state
Returns:
Weighted ensemble prediction
"""
# Get predictions from all experts
predictions = np.array([expert.predict(x)
for expert in self.experts])
# Weighted combination
ensemble_pred = np.sum(self.weights[:, np.newaxis] * predictions,
axis=0)
return ensemble_pred, predictions
def update(self, predictions, target):
"""
Update weights based on observed target.
Args:
predictions: Array of expert predictions [M x output_dim]
target: True target value
"""
self.t += 1
# 1. Compute losses
losses = np.array([self.loss_function(pred, target)
for pred in predictions])
# Ensure losses in [0,1]
losses = np.clip(losses, 0, 1)
# 2. Update performance scores (EMA)
self.P = self.alpha_P * self.P + (1 - self.alpha_P) * (-losses)
# Store in history
for i in range(self.M):
self.P_history[i].append(self.P[i])
# 3. Compute coherence
C_raw = self._compute_coherence(predictions)
# 4. Update smoothed coherence (EMA)
self.C_bar = self.alpha_C * self.C_bar + (1 - self.alpha_C) * C_raw
# Store in history
for i in range(self.M):
self.C_history[i].append(self.C_bar[i])
# 5. Normalize scores
P_tilde = self._normalize_scores(self.P)
C_tilde = self._normalize_scores(self.C_bar)
# 6. Combine scores
s = self.beta * P_tilde + (1 - self.beta) * C_tilde
# 7. Clip scores to prevent overflow
s = np.clip(s, -10, 10)
# 8. Exponential transformation
w_tilde = np.exp(self.eta_s * s)
# 9. Normalization
weights = w_tilde / np.sum(w_tilde)
# 10. Enforce floor
weights = np.maximum(weights, self.w_min)
# 11. Final renormalization
self.weights = weights / np.sum(weights)
return {
'losses': losses,
'weights': self.weights.copy(),
'performance': self.P.copy(),
'coherence': self.C_bar.copy()
}
def _compute_coherence(self, predictions):
"""
Compute coherence scores for all experts.
Args:
predictions: Array [M x output_dim]
Returns:
Coherence scores [M]
"""
M = len(predictions)
coherence = np.zeros(M)
# Get predicted classes (argmax for classification)
if predictions.ndim == 2 and predictions.shape[1] > 1:
classes = np.argmax(predictions, axis=1)
# Pairwise agreement
for i in range(M):
agreements = (classes == classes[i])
coherence[i] = (np.sum(agreements) - 1) / (M - 1)
else:
# Regression: use exponential distance
gamma = 1.0 # Sensitivity parameter
for i in range(M):
distances = np.sum((predictions - predictions[i])**2, axis=1)
similarities = np.exp(-gamma * distances)
coherence[i] = (np.sum(similarities) - 1) / (M - 1)
return coherence
def _normalize_scores(self, scores):
"""
Normalize scores to [0,1] using min-max normalization.
Args:
scores: Array of scores [M]
Returns:
Normalized scores [M]
"""
epsilon = 1e-8
min_score = np.min(scores)
max_score = np.max(scores)
if max_score - min_score < epsilon:
return np.ones_like(scores) * 0.5
return (scores - min_score) / (max_score - min_score + epsilon)
def loss_function(self, prediction, target):
"""
Compute loss between prediction and target.
Override this method for custom loss functions.
Args:
prediction: Model prediction
target: True target
Returns:
Loss value in [0,1]
"""
# Default: MSE normalized to [0,1]
mse = np.mean((prediction - target)**2)
return np.clip(mse, 0, 1)
def get_diagnostics(self):
"""
Return diagnostic information for monitoring.
Returns:
Dictionary with diagnostic metrics
"""
return {
'weights': self.weights.copy(),
'performance': self.P.copy(),
'coherence': self.C_bar.copy(),
'weight_entropy': -np.sum(self.weights * np.log(self.weights + 1e-10)),
'max_weight': np.max(self.weights),
'min_weight': np.min(self.weights),
'time_step': self.t
}Issue: Exponential function can overflow for large scores.
Solution: Clip scores before exponentiation:
s = np.clip(s, -10, 10) # Prevents exp() overflowIssue: Division by zero in normalization.
Solution: Add small epsilon:
denominator = max_val - min_val + 1e-8Performance and coherence histories use bounded deques:
self.P_history = [deque(maxlen=window_size) for _ in range(M)]This ensures O(M × window_size) memory rather than O(M × T).
Weights: Start uniform w_i,0 = 1/M to avoid bias.
Performance: Start at P_i,0 = 0 (neutral).
Coherence: Start at C̄_i,0 = 0.5 (neutral midpoint).
All expert models must implement:
class ExpertInterface:
def predict(self, x):
"""
Generate prediction for input x.
Args:
x: Input state/features
Returns:
Prediction (array-like)
"""
raise NotImplementedErrorEARCP supports any loss function L: 𝒴 × 𝒴 → ℝ₊ satisfying:
- Non-negativity: L(ŷ, y) ≥ 0
- Boundedness: L(ŷ, y) ∈ [0, 1] (or can be normalized)
Common choices:
- MSE (regression):
(y - ŷ)² - Cross-entropy (classification):
-Σ y_k log(ŷ_k) - MAE (robust regression):
|y - ŷ|
def coherence_classification(predictions):
classes = np.argmax(predictions, axis=1)
M = len(classes)
coherence = np.zeros(M)
for i in range(M):
agreement = np.sum(classes == classes[i]) - 1
coherence[i] = agreement / (M - 1)
return coherencedef coherence_regression(predictions, gamma=1.0):
M = len(predictions)
coherence = np.zeros(M)
for i in range(M):
distances = np.sum((predictions - predictions[i])**2, axis=1)
similarities = np.exp(-gamma * distances)
coherence[i] = (np.sum(similarities) - 1) / (M - 1)
return coherenceRecommended starting values:
DEFAULT_CONFIG = {
'alpha_P': 0.9, # Performance smoothing
'alpha_C': 0.85, # Coherence smoothing
'beta': 0.7, # Performance-coherence balance
'eta_s': 5.0, # Exponential sensitivity
'w_min': 0.05, # Weight floor
'performance_window': 50 # History window
}- Higher (0.95-0.99): Slow adaptation, stable in stationary environments
- Lower (0.7-0.85): Fast adaptation, responsive to changes
- Recommendation: Start with 0.9, decrease for non-stationary data
- Higher (0.9-0.95): Smooth coherence signal, less noise
- Lower (0.7-0.85): Responsive to agreement changes
- Recommendation: Set slightly lower than alpha_P (0.85)
- High (0.8-1.0): Favor individual performance (exploitation)
- Low (0.0-0.5): Favor ensemble consensus (exploration)
- Recommendation: 0.7 for balanced approach
- High (>7): Aggressive weight updates, risk of instability
- Low (<3): Conservative updates, slow convergence
- Recommendation: 5.0 for standard settings
- Higher (0.1): More uniform weighting, exploration emphasis
- Lower (0.01): Allow strong concentration, exploitation emphasis
- Recommendation: 0.05 for good exploration-exploitation balance
High-frequency data (milliseconds-seconds):
- Lower α_P, α_C (0.7-0.8) for responsiveness
- Higher eta_s (6-8) for decisive updates
Low-frequency data (hours-days):
- Higher α_P, α_C (0.95-0.98) for stability
- Lower eta_s (3-5) for smooth transitions
High-noise environments:
- Higher α_P, α_C for smoothing
- Higher w_min (0.1) to prevent premature elimination
Adversarial/non-stationary:
- Lower α_P (0.7-0.8) for rapid adaptation
- Higher w_min (0.1) to maintain diversity
Profiling results (M=4 experts):
- Expert predictions: ~95% of time (parallelizable)
- Coherence computation: ~3% of time O(M²)
- Weight updates: ~2% of time O(M)
Optimization strategies:
from concurrent.futures import ThreadPoolExecutor
def parallel_predict(self, x):
with ThreadPoolExecutor(max_workers=self.M) as executor:
futures = [executor.submit(expert.predict, x)
for expert in self.experts]
predictions = [f.result() for f in futures]
return np.array(predictions)For large M, sample K << M expert pairs:
def approximate_coherence(self, predictions, K=10):
M = len(predictions)
coherence = np.zeros(M)
for i in range(M):
# Sample K random other experts
j_samples = np.random.choice([j for j in range(M) if j != i],
size=min(K, M-1), replace=False)
agreements = [self._check_agreement(predictions[i], predictions[j])
for j in j_samples]
coherence[i] = np.mean(agreements)
return coherenceReduces complexity from O(M²) to O(M·K).
Cache min/max values for normalization:
class CachedNormalizer:
def __init__(self, window_size=50):
self.window = deque(maxlen=window_size)
self._min = None
self._max = None
def normalize(self, scores):
self.window.extend(scores)
# Update cached min/max only when needed
if self._min is None or len(self.window) % 10 == 0:
self._min = min(self.window)
self._max = max(self.window)
return (scores - self._min) / (self._max - self._min + 1e-8)Override the loss_function method:
class CustomEARCP(EARCP):
def loss_function(self, prediction, target):
# Custom loss (e.g., asymmetric loss for finance)
error = prediction - target
loss = np.where(error >= 0, error**2, 2 * error**2)
return np.clip(np.mean(loss), 0, 1)Override _compute_coherence:
class CustomEARCP(EARCP):
def _compute_coherence(self, predictions):
# Custom coherence (e.g., correlation-based)
M = len(predictions)
coherence = np.zeros(M)
for i in range(M):
correlations = [np.corrcoef(predictions[i], predictions[j])[0,1]
for j in range(M) if j != i]
coherence[i] = np.mean(correlations)
return np.clip(coherence, 0, 1)Organize experts in hierarchy:
class HierarchicalEARCP:
def __init__(self, expert_groups):
"""
Args:
expert_groups: List of lists, each sublist is a group
"""
# First level: EARCP for each group
self.group_ensembles = [EARCP(group) for group in expert_groups]
# Second level: Meta-EARCP combining groups
self.meta_ensemble = EARCP(self.group_ensembles)
def predict(self, x):
return self.meta_ensemble.predict(x)def add_expert(self, new_expert):
"""Add new expert dynamically."""
self.experts.append(new_expert)
self.M += 1
# Initialize state for new expert
new_weight = 1.0 / self.M
self.weights *= (1 - new_weight) # Renormalize existing
self.weights = np.append(self.weights, new_weight)
self.P = np.append(self.P, 0)
self.C_bar = np.append(self.C_bar, 0.5)
self.P_history.append(deque(maxlen=self.performance_window))
self.C_history.append(deque(maxlen=self.performance_window))
def remove_expert(self, expert_idx):
"""Remove expert by index."""
del self.experts[expert_idx]
self.M -= 1
# Remove state
self.weights = np.delete(self.weights, expert_idx)
self.weights /= np.sum(self.weights) # Renormalize
self.P = np.delete(self.P, expert_idx)
self.C_bar = np.delete(self.C_bar, expert_idx)
del self.P_history[expert_idx]
del self.C_history[expert_idx]Complete working example:
import numpy as np
from collections import deque
class SimpleExpert:
"""Example expert implementation."""
def __init__(self, name, predictor_fn):
self.name = name
self.predictor_fn = predictor_fn
def predict(self, x):
return self.predictor_fn(x)
# Create synthetic experts
def expert1_fn(x): return np.array([0.8, 0.2]) # Optimistic
def expert2_fn(x): return np.array([0.3, 0.7]) # Pessimistic
def expert3_fn(x): return np.array([0.5, 0.5]) # Neutral
def expert4_fn(x): return np.array([0.6, 0.4]) # Moderate
experts = [
SimpleExpert("CNN", expert1_fn),
SimpleExpert("LSTM", expert2_fn),
SimpleExpert("Transformer", expert3_fn),
SimpleExpert("DQN", expert4_fn)
]
# Initialize EARCP
ensemble = EARCP(experts, beta=0.7, eta_s=5.0, w_min=0.05)
# Simulation loop
T = 1000
for t in range(T):
# Generate random input
x = np.random.randn(10)
# Get prediction
prediction, expert_preds = ensemble.predict(x)
# Simulate target (could be real data)
target = np.array([0.6, 0.4])
# Update weights
metrics = ensemble.update(expert_preds, target)
# Monitor every 100 steps
if t % 100 == 0:
diag = ensemble.get_diagnostics()
print(f"Step {t}:")
print(f" Weights: {diag['weights']}")
print(f" Entropy: {diag['weight_entropy']:.3f}")import unittest
class TestEARCP(unittest.TestCase):
def setUp(self):
self.experts = [SimpleExpert(f"E{i}", lambda x: np.random.rand(2))
for i in range(4)]
self.ensemble = EARCP(self.experts)
def test_weights_sum_to_one(self):
"""Weights must always sum to 1."""
x = np.random.randn(10)
_, preds = self.ensemble.predict(x)
target = np.random.rand(2)
self.ensemble.update(preds, target)
self.assertAlmostEqual(np.sum(self.ensemble.weights), 1.0, places=6)
def test_weights_above_floor(self):
"""All weights must be >= w_min."""
x = np.random.randn(10)
_, preds = self.ensemble.predict(x)
target = np.random.rand(2)
self.ensemble.update(preds, target)
self.assertTrue(np.all(self.ensemble.weights >= self.ensemble.w_min))
def test_performance_updates(self):
"""Performance scores must update after each step."""
P_before = self.ensemble.P.copy()
x = np.random.randn(10)
_, preds = self.ensemble.predict(x)
target = np.random.rand(2)
self.ensemble.update(preds, target)
self.assertFalse(np.allclose(P_before, self.ensemble.P))
def test_numerical_stability(self):
"""Test with extreme values."""
# Large scores should not cause overflow
self.ensemble.P = np.array([1000, -1000, 500, -500])
x = np.random.randn(10)
_, preds = self.ensemble.predict(x)
target = np.random.rand(2)
# Should not raise exception
self.ensemble.update(preds, target)
# Weights should still be valid
self.assertTrue(np.all(np.isfinite(self.ensemble.weights)))Test on real sequential prediction tasks with known ground truth.
def benchmark_earcp(dataset, experts, num_trials=10):
"""
Benchmark EARCP against baselines.
Args:
dataset: Sequence of (x, y) pairs
experts: List of expert models
num_trials: Number of random seeds
Returns:
Dictionary of metrics
"""
results = {
'earcp': [],
'equal': [],
'best_single': [],
'hedge': []
}
for trial in range(num_trials):
# Test EARCP
ensemble = EARCP(experts)
earcp_loss = run_online_experiment(ensemble, dataset)
results['earcp'].append(earcp_loss)
# Test equal weighting
equal_loss = run_equal_weighting(experts, dataset)
results['equal'].append(equal_loss)
# Test oracle best single expert
best_loss = run_best_single_expert(experts, dataset)
results['best_single'].append(best_loss)
# Test Hedge (beta=1)
hedge = EARCP(experts, beta=1.0)
hedge_loss = run_online_experiment(hedge, dataset)
results['hedge'].append(hedge_loss)
# Compute statistics
summary = {}
for method, losses in results.items():
summary[method] = {
'mean': np.mean(losses),
'std': np.std(losses),
'min': np.min(losses),
'max': np.max(losses)
}
return summaryThis whitepaper serves as defensive publication (also called statutory bar or prior art):
Legal effect:
- Establishes public disclosure date: November 13, 2025
- Prevents third-party patents on disclosed inventions
- Preserves inventor's rights to:
- Commercialize the technology
- License the technology
- File patents within grace period (where applicable)
Critical elements disclosed:
- Complete algorithm with mathematical formulation
- Implementation details and pseudocode
- Key design decisions and rationales
- Extension mechanisms and variations
Code implementations are automatically copyrighted:
- Copyright holder: Mike Amega
- Copyright date: 2025
- Rights: All rights reserved unless explicitly licensed
-
Immediate:
- Upload this document to GitHub with timestamp
- Create release tag with date
- Include LICENSE file (MIT, Apache 2.0, or proprietary)
- Archive on Zenodo or figshare for permanent DOI
-
Within 30 days:
- Submit to arXiv for academic timestamp
- Consider submitting to technical conference/journal
- Register copyright if in jurisdiction requiring registration
-
Ongoing:
- Document all improvements and variations
- Maintain change log with dates
- Keep signed records of development notebooks
Options:
- Pure defensive: Publish everything, prevent others from patenting
- Mixed strategy: Patent core innovations, publish extensions
- Full patent: File provisional patent before publication
Recommendation for Mike Amega: Given GitHub publication goal, pursue defensive publication strategy:
- Publish complete technical details (this document)
- Maintain copyright on code
- Consider commercial licensing for specific use cases
- Reserve option to patent future improvements not yet disclosed
- Initial public release
- Complete algorithm specification
- Reference implementation
- Theoretical guarantees
- Extension mechanisms
If using EARCP in academic work, please cite:
@techreport{amega2025earcp,
title={EARCP: Ensemble Auto-Régulé par Cohérence et Performance - Technical Whitepaper},
author={Amega, Mike},
year={2025},
institution={Independent Research},
url={https://github.com/[username]/earcp},
note={Prior art established through defensive publication}
}
Author: Mike Amega
Email: contact@mikeamega.ca
Location: Windsor, Ontario, Canada
Date: November 13, 2025
License Options:
- Academic/Research: MIT License (free use with attribution)
- Commercial: Contact author for licensing terms
- Patents: All rights reserved for future patent applications
Code Repository: [To be published on GitHub]
INTELLECTUAL PROPERTY NOTICE
This document contains original technical work by Mike Amega. Public disclosure through this whitepaper establishes prior art as of November 13, 2025, preventing third-party patent claims while preserving the author's rights to commercialize, license, or patent this technology.
All algorithms, mathematical formulations, and implementation details herein are protected by copyright and defensive publication. Commercial use requires explicit permission from the author.
For licensing inquiries: contact@mikeamega.ca
END OF WHITEPAPER