Comprehensive Adaptive Estimators Demo
This comprehensive demo showcases the new adaptive estimator system with automatic framework selection, robust error handling, and EEG contamination testing.
Overview
The comprehensive adaptive estimators provide:
Automatic Framework Selection: GPU/JAX, CPU/Numba, or NumPy based on data characteristics
Robust Error Handling: Adaptive parameter selection and progressive fallbacks
Performance Profiling: Built-in performance monitoring and optimization
EEG Contamination Testing: Realistic artifact scenarios for biomedical applications
Mathematical Verification: All estimators verified against theoretical foundations
Basic Usage
from lrdbenchmark.analysis.comprehensive_adaptive_estimators import get_all_comprehensive_adaptive_classical_estimators
from lrdbenchmark import FBMModel
import numpy as np
# Get all comprehensive adaptive estimators
estimators = get_all_comprehensive_adaptive_classical_estimators()
# Generate test data
model = FBMModel(H=0.7, sigma=1.0)
data = model.generate(1000, seed=42)
print("Comprehensive Adaptive Estimators Demo")
print("=" * 50)
print(f"Generated FBM data with true H = 0.7")
print(f"Data length: {len(data)}")
print(f"Data mean: {data.mean():.3f}, std: {data.std():.3f}")
# Test all estimators
results = {}
for name, estimator in estimators.items():
try:
result = estimator.estimate(data)
results[name] = result
print(f"{name:25s}: H={result['hurst_parameter']:.4f}, "
f"Framework={result['framework_used']:15s}, "
f"Time={result['execution_time']:.4f}s")
except Exception as e:
print(f"{name:25s}: Failed - {e}")
# Calculate average error
errors = []
for name, result in results.items():
error = abs(result['hurst_parameter'] - 0.7)
errors.append(error)
print(f"\nAverage absolute error: {np.mean(errors):.4f}")
print(f"Best estimator: {min(results.keys(), key=lambda k: abs(results[k]['hurst_parameter'] - 0.7))}")
Performance Comparison
from lrdbenchmark.analysis.comprehensive_adaptive_estimators import ComprehensiveAdaptiveRS, ComprehensiveAdaptiveDFA
from lrdbenchmark import FBMModel
import numpy as np
import time
# Generate test data
model = FBMModel(H=0.7, sigma=1.0)
data = model.generate(5000, seed=42)
print("Performance Comparison Demo")
print("=" * 40)
print(f"Data length: {len(data)}")
# Test R/S estimator
rs_estimator = ComprehensiveAdaptiveRS(enable_profiling=True)
start_time = time.time()
rs_result = rs_estimator.estimate(data)
rs_time = time.time() - start_time
# Test DFA estimator
dfa_estimator = ComprehensiveAdaptiveDFA(enable_profiling=True)
start_time = time.time()
dfa_result = dfa_estimator.estimate(data)
dfa_time = time.time() - start_time
print(f"R/S Analysis:")
print(f" Hurst estimate: {rs_result['hurst_parameter']:.4f}")
print(f" Framework used: {rs_result['framework_used']}")
print(f" Execution time: {rs_time:.4f}s")
print(f" Framework reasoning: {rs_result['framework_reasoning']}")
print(f"\nDFA Analysis:")
print(f" Hurst estimate: {dfa_result['hurst_parameter']:.4f}")
print(f" Framework used: {dfa_result['framework_used']}")
print(f" Execution time: {dfa_time:.4f}s")
print(f" Framework reasoning: {dfa_result['framework_reasoning']}")
# Get performance information
rs_perf = rs_estimator.get_performance_info()
dfa_perf = dfa_estimator.get_performance_info()
print(f"\nPerformance Summary:")
print(f"R/S total runs: {rs_perf['total_runs']}")
print(f"DFA total runs: {dfa_perf['total_runs']}")
EEG Contamination Testing
from lrdbenchmark.analysis.comprehensive_adaptive_estimators import ComprehensiveAdaptiveDFA
from lrdbenchmark import ContaminationFactory, ConfoundingScenario
from lrdbenchmark import FBMModel
import numpy as np
# Create estimator and contamination factory
estimator = ComprehensiveAdaptiveDFA()
contamination_factory = ContaminationFactory()
# Generate pure data
model = FBMModel(H=0.7, sigma=1.0)
pure_data = model.generate(1000, seed=42)
# Get pure data estimate
pure_result = estimator.estimate(pure_data)
pure_h = pure_result['hurst_parameter']
print("EEG Contamination Testing Demo")
print("=" * 40)
print(f"Pure data H estimate: {pure_h:.4f}")
print(f"Pure data framework: {pure_result['framework_used']}")
# EEG contamination scenarios
eeg_scenarios = [
ConfoundingScenario.EEG_OCULAR_ARTIFACTS,
ConfoundingScenario.EEG_MUSCLE_ARTIFACTS,
ConfoundingScenario.EEG_CARDIAC_ARTIFACTS,
ConfoundingScenario.EEG_ELECTRODE_POPPING,
ConfoundingScenario.EEG_ELECTRODE_DRIFT,
ConfoundingScenario.EEG_60HZ_NOISE,
ConfoundingScenario.EEG_SWEAT_ARTIFACTS,
ConfoundingScenario.EEG_MOVEMENT_ARTIFACTS
]
print(f"\nEEG Contamination Results:")
print(f"{'Scenario':<25s} {'H Estimate':<12s} {'Error':<8s} {'Framework':<15s}")
print("-" * 70)
for scenario in eeg_scenarios:
contaminated_data, description = contamination_factory.apply_confounding(
pure_data, scenario, intensity=0.3
)
result = estimator.estimate(contaminated_data)
h_est = result['hurst_parameter']
error = abs(h_est - pure_h)
framework = result['framework_used']
print(f"{scenario.value:<25s} {h_est:<12.4f} {error:<8.4f} {framework:<15s}")
# Calculate robustness metrics
errors = []
for scenario in eeg_scenarios:
contaminated_data, _ = contamination_factory.apply_confounding(
pure_data, scenario, intensity=0.3
)
result = estimator.estimate(contaminated_data)
error = abs(result['hurst_parameter'] - pure_h)
errors.append(error)
print(f"\nRobustness Summary:")
print(f"Average error: {np.mean(errors):.4f}")
print(f"Max error: {np.max(errors):.4f}")
print(f"Min error: {np.min(errors):.4f}")
print(f"Std error: {np.std(errors):.4f}")
Data Length Robustness Testing
from lrdbenchmark.analysis.comprehensive_adaptive_estimators import ComprehensiveAdaptiveGPH, ComprehensiveAdaptiveCWT
from lrdbenchmark import FBMModel
import numpy as np
# Create estimators
gph_estimator = ComprehensiveAdaptiveGPH()
cwt_estimator = ComprehensiveAdaptiveCWT()
# Generate test data
model = FBMModel(H=0.7, sigma=1.0)
# Test different data lengths
data_lengths = [50, 100, 200, 500, 1000, 2000, 5000]
print("Data Length Robustness Testing")
print("=" * 40)
print(f"{'Length':<8s} {'GPH H':<8s} {'GPH Error':<10s} {'CWT H':<8s} {'CWT Error':<10s}")
print("-" * 50)
for length in data_lengths:
data = model.generate(length, seed=42)
# Test GPH
try:
gph_result = gph_estimator.estimate(data)
gph_h = gph_result['hurst_parameter']
gph_error = abs(gph_h - 0.7)
gph_str = f"{gph_h:.4f}"
gph_err_str = f"{gph_error:.4f}"
except Exception as e:
gph_str = "Failed"
gph_err_str = "N/A"
# Test CWT
try:
cwt_result = cwt_estimator.estimate(data)
cwt_h = cwt_result['hurst_parameter']
cwt_error = abs(cwt_h - 0.7)
cwt_str = f"{cwt_h:.4f}"
cwt_err_str = f"{cwt_error:.4f}"
except Exception as e:
cwt_str = "Failed"
cwt_err_str = "N/A"
print(f"{length:<8d} {gph_str:<8s} {gph_err_str:<10s} {cwt_str:<8s} {cwt_err_str:<10s}")
Framework Selection Analysis
from lrdbenchmark.analysis.comprehensive_adaptive_estimators import ComprehensiveAdaptiveRS
from lrdbenchmark import FBMModel
import numpy as np
# Create estimator with profiling
estimator = ComprehensiveAdaptiveRS(enable_profiling=True)
# Generate test data
model = FBMModel(H=0.7, sigma=1.0)
# Test with different data sizes
data_sizes = [100, 500, 1000, 2000, 5000, 10000]
print("Framework Selection Analysis")
print("=" * 40)
print(f"{'Data Size':<10s} {'Framework':<15s} {'Time (s)':<10s} {'H Estimate':<12s}")
print("-" * 50)
for size in data_sizes:
data = model.generate(size, seed=42)
result = estimator.estimate(data)
print(f"{size:<10d} {result['framework_used']:<15s} "
f"{result['execution_time']:<10.4f} {result['hurst_parameter']:<12.4f}")
# Get performance information
perf_info = estimator.get_performance_info()
print(f"\nPerformance Summary:")
print(f"Total runs: {perf_info['total_runs']}")
print(f"Frameworks used: {perf_info['frameworks_used']}")
Comprehensive Benchmark
from lrdbenchmark.analysis.comprehensive_adaptive_estimators import get_all_comprehensive_adaptive_classical_estimators
from lrdbenchmark import ContaminationFactory, ConfoundingScenario
from lrdbenchmark import FBMModel
from lrdbenchmark import FGNModel
import numpy as np
import pandas as pd
# Get all estimators
estimators = get_all_comprehensive_adaptive_classical_estimators()
contamination_factory = ContaminationFactory()
# Test scenarios
data_models = {
'FBM': FBMModel(H=0.7, sigma=1.0),
'FGN': FGNModel(H=0.7, sigma=1.0)
}
contamination_scenarios = [
ConfoundingScenario.PURE,
ConfoundingScenario.GAUSSIAN_NOISE,
ConfoundingScenario.EEG_OCULAR_ARTIFACTS,
ConfoundingScenario.EEG_MUSCLE_ARTIFACTS
]
data_lengths = [500, 1000, 2000]
print("Comprehensive Benchmark Demo")
print("=" * 40)
# Store results
all_results = []
for model_name, model in data_models.items():
for length in data_lengths:
for scenario in contamination_scenarios:
# Generate data
data = model.generate(length, seed=42)
# Apply contamination
if scenario != ConfoundingScenario.PURE:
data, description = contamination_factory.apply_confounding(
data, scenario, intensity=0.3
)
# Test all estimators
for est_name, estimator in estimators.items():
try:
result = estimator.estimate(data)
all_results.append({
'Model': model_name,
'Length': length,
'Scenario': scenario.value,
'Estimator': est_name,
'H_Estimate': result['hurst_parameter'],
'Framework': result['framework_used'],
'Time': result['execution_time'],
'Success': True
})
except Exception as e:
all_results.append({
'Model': model_name,
'Length': length,
'Scenario': scenario.value,
'Estimator': est_name,
'H_Estimate': np.nan,
'Framework': 'Failed',
'Time': 0,
'Success': False
})
# Create DataFrame
df = pd.DataFrame(all_results)
# Calculate success rate
success_rate = df['Success'].mean() * 100
print(f"Overall success rate: {success_rate:.1f}%")
# Calculate average error (for successful estimates)
successful_df = df[df['Success'] == True]
if len(successful_df) > 0:
# Calculate error for FBM (true H = 0.7)
fbm_df = successful_df[successful_df['Model'] == 'FBM']
if len(fbm_df) > 0:
fbm_df['Error'] = abs(fbm_df['H_Estimate'] - 0.7)
avg_error = fbm_df['Error'].mean()
print(f"Average error (FBM): {avg_error:.4f}")
# Framework usage statistics
framework_usage = df['Framework'].value_counts()
print(f"\nFramework usage:")
for framework, count in framework_usage.items():
print(f" {framework}: {count} ({count/len(df)*100:.1f}%)")
# Estimator performance
estimator_success = df.groupby('Estimator')['Success'].mean() * 100
print(f"\nEstimator success rates:")
for estimator, success_rate in estimator_success.items():
print(f" {estimator}: {success_rate:.1f}%")
Best Practices
Use Comprehensive Adaptive Estimators: They provide the most robust and efficient estimation
Enable Performance Profiling: Monitor framework selection and performance
Test with Contamination: Use realistic contamination scenarios for robustness testing
Compare Multiple Estimators: Different estimators may perform better for different data types
Validate with Known Data: Test on synthetic data with known Hurst parameters
Monitor Framework Selection: Understand which frameworks are being used and why
Note
The comprehensive adaptive estimators automatically handle most edge cases and provide the most reliable way to estimate Hurst parameters from time series data.
Warning
While the adaptive system is robust, very short data (< 50 points) may still produce unreliable results regardless of the framework used.