Adaptive Wfo Epoch

Adaptive Walk-Forward Epoch Selection (AWFES)

Machine-readable reference for adaptive epoch selection within Walk-Forward Optimization (WFO). Optimizes training epochs per-fold using Walk-Forward Efficiency (WFE) as the objective.

Self-Evolving Skill: This skill improves through use. If instructions are wrong, parameters drifted, or a workaround was needed — fix this file immediately, don't defer. Only update for real, reproducible issues.

When to Use This Skill

Use this skill when:

• Selecting optimal training epochs for ML models in WFO
• Avoiding overfitting via Walk-Forward Efficiency metrics
• Implementing per-fold adaptive epoch selection
• Computing efficient frontiers for epoch-performance trade-offs
• Carrying epoch priors across WFO folds

Quick Start

from adaptive_wfo_epoch import AWFESConfig, compute_efficient_frontier

# Generate epoch candidates from search bounds and granularity
config = AWFESConfig.from_search_space(
    min_epoch=100,
    max_epoch=2000,
    granularity=5,  # Number of frontier points
)
# config.epoch_configs → [100, 211, 447, 945, 2000] (log-spaced)

# Per-fold epoch sweep
for fold in wfo_folds:
    epoch_metrics = []
    for epoch in config.epoch_configs:
        is_sharpe, oos_sharpe = train_and_evaluate(fold, epochs=epoch)
        wfe = config.compute_wfe(is_sharpe, oos_sharpe, n_samples=len(fold.train))
        epoch_metrics.append({"epoch": epoch, "wfe": wfe, "is_sharpe": is_sharpe})

    # Select from efficient frontier
    selected_epoch = compute_efficient_frontier(epoch_metrics)

    # Carry forward to next fold as prior
    prior_epoch = selected_epoch

Methodology Overview

What This Is

Per-fold adaptive epoch selection where:

1. Train models across a range of epochs (e.g., 400, 800, 1000, 2000) 2. Compute WFE = OOS_Sharpe / IS_Sharpe for each epoch count 3. Find the "efficient frontier" - epochs maximizing WFE vs training cost 4. Select optimal epoch from frontier for OOS evaluation 5. Carry forward as prior for next fold

What This Is NOT

• NOT early stopping: Early stopping monitors validation loss continuously; this evaluates discrete candidates post-hoc
• NOT Bayesian optimization: No surrogate model; direct evaluation of all candidates
• NOT nested cross-validation: Uses temporal WFO, not shuffled splits

Academic Foundations

See references/academic-foundations.md for full literature review.

Core Formula: Walk-Forward Efficiency

def compute_wfe(
    is_sharpe: float,
    oos_sharpe: float,
    n_samples: int | None = None,
) -> float | None:
    """Walk-Forward Efficiency - measures performance transfer.

    WFE = OOS_Sharpe / IS_Sharpe

    Interpretation (guidelines, not hard thresholds):
    - WFE ≥ 0.70: Excellent transfer (low overfitting)
    - WFE 0.50-0.70: Good transfer
    - WFE 0.30-0.50: Moderate transfer (investigate)
    - WFE < 0.30: Severe overfitting (likely reject)

    The IS_Sharpe minimum is derived from signal-to-noise ratio,
    not a fixed magic number. See compute_is_sharpe_threshold().

    Reference: Pardo (2008) "The Evaluation and Optimization of Trading Strategies"
    """
    # Data-driven threshold: IS_Sharpe must exceed 2σ noise floor
    min_is_sharpe = compute_is_sharpe_threshold(n_samples) if n_samples else 0.1

    if abs(is_sharpe) < min_is_sharpe:
        return None
    return oos_sharpe / is_sharpe

Principled Configuration Framework

All parameters are derived from first principles or data characteristics. AWFESConfig provides unified configuration with log-spaced epoch generation, Bayesian variance derivation from search space, and market-specific annualization factors.

See references/configuration-framework.md for the full AWFESConfig class and compute_is_sharpe_threshold() implementation.

Guardrails (Principled Guidelines)

• G1: WFE Thresholds - 0.30 (reject), 0.50 (warning), 0.70 (target) based on practitioner consensus
• G2: IS_Sharpe Minimum - Data-driven threshold: 2/sqrt(n) adapts to sample size
• G3: Stability Penalty - Adaptive threshold derived from WFE variance prevents epoch churn
• G4: DSR Adjustment - Deflated Sharpe corrects for epoch selection multiplicity via Gumbel distribution

See references/guardrails.md for full implementations of all guardrails.

WFE Aggregation Methods

Under the null hypothesis, WFE follows a Cauchy distribution (no defined mean). Always prefer median or pooled methods:

• Pooled WFE: Precision-weighted by sample size (best for variable fold sizes)
• Median WFE: Robust to outliers (best for suspected regime changes)
• Weighted Mean: Inverse-variance weighting (best for homogeneous folds)

See references/wfe-aggregation.md for implementations and selection guide.

Efficient Frontier Algorithm

Pareto-optimal epoch selection: an epoch is on the frontier if no other epoch dominates it (better WFE AND lower training time). The AdaptiveEpochSelector class maintains state across folds with adaptive stability penalties.

See references/efficient-frontier.md for the full algorithm and carry-forward mechanism.

Anti-Patterns

CRITICAL: Never use expanding window for range bar ML training. See references/anti-patterns.md for the full analysis (Section 7).

Decision Tree

See references/epoch-selection-decision-tree.md for the full practitioner decision tree.

Start
  │
  ├─ IS_Sharpe > compute_is_sharpe_threshold(n)? ──NO──> Mark WFE invalid, use fallback
  │         │                                            (threshold = 2/√n, adapts to sample size)
  │        YES
  │         │
  ├─ Compute WFE for each epoch
  │         │
  ├─ Any WFE > 0.30? ──NO──> REJECT all epochs (severe overfit)
  │         │                (guideline, not hard threshold)
  │        YES
  │         │
  ├─ Compute efficient frontier
  │         │
  ├─ Apply AdaptiveStabilityPenalty
  │         │ (threshold derived from WFE variance)
  └─> Return selected epoch

Integration with rangebar-eval-metrics

This skill extends rangebar-eval-metrics:

Recommended Workflow

1. Compute base metrics using rangebar-eval-metrics:compute_metrics.py 2. Feed to AWFES for epoch selection with sharpe_tw as primary signal 3. Validate with psr > 0.85 and dsr > 0.50 before deployment 4. Monitor is_collapsed and prediction_autocorr for model health

---

OOS Application Phase

AWFES uses Nested WFO with three data splits per fold (Train 60% / Val 20% / Test 20%) with 6% embargo gaps at each boundary. The per-fold workflow: epoch sweep on train, WFE computation on validation, Bayesian update, final model training on train+val, evaluation on test.

See references/oos-workflow.md for the complete workflow with diagrams, BayesianEpochSelector class, and apply_awfes_to_test() implementation. Also see references/oos-application.md for the extended reference.

Epoch Smoothing Methods

Bayesian updating (recommended) provides principled, uncertainty-aware smoothing. Alternatives include EMA and SMA. Initialization via AWFESConfig.from_search_space() derives variances from the epoch range automatically.

See references/epoch-smoothing-methods.md for all methods, formulas, and initialization strategies. See references/epoch-smoothing.md for extended mathematical analysis.

OOS Metrics Specification

Three-tier metric hierarchy for test evaluation:

• Tier 1 (Primary): sharpe_tw, hit_rate, cumulative_pnl, positive_sharpe_folds, wfe_test
• Tier 2 (Risk): max_drawdown, calmar_ratio, profit_factor, cvar_10pct
• Tier 3 (Statistical): psr, dsr, binomial_pvalue, hac_ttest_pvalue

See references/oos-metrics-implementation.md for full metric tables, compute_oos_metrics(), and fold aggregation code. See references/oos-metrics.md for threshold justifications.

Look-Ahead Bias Prevention

CRITICAL (v3 fix): TEST must use prior_bayesian_epoch (from prior folds only), NOT val_optimal_epoch. The Bayesian update happens AFTER test evaluation, ensuring information flows only from past to present.

See references/look-ahead-bias-v3.md for the v3 fix details, embargo requirements, validation checklist, and anti-patterns. See references/look-ahead-bias.md for detailed examples.

---

References

Related Skills

Full Citations

• Bailey, D. H., & López de Prado, M. (2014). The deflated Sharpe ratio: Correcting for selection bias, backtest overfitting and non-normality. _The Journal of Portfolio Management_, 40(5), 94-107.
• Bischl, B., et al. (2023). Multi-Objective Hyperparameter Optimization in Machine Learning. _ACM Transactions on Evolutionary Learning and Optimization_.
• López de Prado, M. (2018). _Advances in Financial Machine Learning_. Wiley. Chapter 7.
• Nomura, M., & Ono, I. (2021). Warm Starting CMA-ES for Hyperparameter Optimization. _AAAI Conference on Artificial Intelligence_.
• Pardo, R. E. (2008). _The Evaluation and Optimization of Trading Strategies, 2nd Edition_. John Wiley & Sons.

Post-Execution Reflection

After this skill completes, check before closing:

1. Did the command succeed? — If not, fix the instruction or error table that caused the failure. 2. Did parameters or output change? — If the underlying tool's interface drifted, update Usage examples and Parameters table to match. 3. Was a workaround needed? — If you had to improvise (different flags, extra steps), update this SKILL.md so the next invocation doesn't need the same workaround.

Only update if the issue is real and reproducible — not speculative.

Academic Foundations: Adaptive Walk-Forward Epoch Selection

Literature Review

This methodology synthesizes concepts from four distinct academic traditions:

1. Walk-Forward Analysis (Trading Systems Research) 2. Deflated Sharpe Ratio (Statistical Finance) 3. Multi-Objective Hyperparameter Optimization (Machine Learning) 4. Warm-Starting Sequential Optimization (AutoML)

1. Walk-Forward Efficiency (WFE)

Origin

Walk-Forward Efficiency was introduced by Robert E. Pardo in his seminal work on trading system validation:

• Pardo, R. E. (1992). _Design, Testing, and Optimization of Trading Systems._ John Wiley & Sons.
• Pardo, R. E. (2008). _The Evaluation and Optimization of Trading Strategies, 2nd Edition._ John Wiley & Sons.

Definition

WFE = OOS_Performance / IS_Performance

Typically expressed as return ratio or Sharpe ratio.

Interpretation Guidelines (Pardo)

Key Quote

"Walk-Forward Efficiency measures the degree to which a strategy's in-sample performance translates to out-of-sample results. A strategy that cannot maintain at least 50% of its in-sample performance is likely overfit to historical data."

— Pardo (2008), Chapter 8

2. Deflated Sharpe Ratio (DSR)

Origin

Bailey, D. H., & López de Prado, M. (2014). "The Deflated Sharpe Ratio: Correcting for Selection Bias, Backtest Overfitting and Non-Normality." _The Journal of Portfolio Management_, 40(5), 94-107.

Problem Addressed

When testing multiple strategies (or hyperparameter configurations), the "best" Sharpe ratio is expected to be inflated due to multiple testing. DSR corrects for this selection bias.

Formula

DSR = Φ[(SR - SR₀) × √T / √(1 + 0.5×SR² - γ₃×SR + (γ₄-3)/4×SR²)]

Where:

• SR = Observed Sharpe ratio
• SR₀ = Expected maximum Sharpe under null (depends on number of trials)
• T = Number of observations
• γ₃ = Skewness
• γ₄ = Kurtosis
• Φ = Standard normal CDF

Expected Maximum Under Null

For N independent trials:

SR₀ ≈ √(2 × ln(N)) - (γ + ln(π/2)) / √(2 × ln(N))

Where γ ≈ 0.5772 (Euler-Mascheroni constant).

Application to Epoch Selection

When selecting from K epochs across F folds, total trials = K × F.

For 4 epochs × 31 folds = 124 trials:

• SR₀ ≈ 2.5 × σ(SR)
• With σ(SR) ≈ 0.3: SR₀ ≈ 0.75

A Sharpe of 1.0 deflates to ~0.25 after DSR adjustment.

3. Multi-Objective Hyperparameter Optimization (MOHPO)

Key References

• Bischl, B., Binder, M., Lang, M., et al. (2023). "Multi-Objective Hyperparameter Optimization in Machine Learning—An Overview." _ACM Transactions on Evolutionary Learning and Optimization._
• Deb, K., Pratap, A., Agarwal, S., & Meyarivan, T. (2002). "A Fast and Elitist Multiobjective Genetic Algorithm: NSGA-II." _IEEE Transactions on Evolutionary Computation_, 6(2), 182-197.

Pareto Optimality

A solution is Pareto-optimal if no other solution improves one objective without worsening another.

For epoch selection with objectives:

1. Maximize WFE (generalization quality) 2. Minimize training time (computational cost)

An epoch is on the efficient frontier if no other epoch dominates it.

Algorithms

• NSGA-II: Non-dominated Sorting Genetic Algorithm (Deb et al., 2002)
• SPEA-II: Strength Pareto Evolutionary Algorithm
• SMS-EMOA: S-Metric Selection Evolutionary Multi-Objective Algorithm

For discrete epoch selection (4 candidates), exhaustive evaluation is tractable; these algorithms are relevant for continuous or large search spaces.

4. Warm-Starting Sequential Optimization

Key References

• Nomura, M., & Ono, I. (2021). "Warm Starting CMA-ES for Hyperparameter Optimization." _Proceedings of the AAAI Conference on Artificial Intelligence._
• Perrone, V., et al. (2017). "Learning to Transfer Initializations for Bayesian Hyperparameter Optimization." _BayesOpt Workshop at NeurIPS._

Concept

Transfer knowledge from previous optimization runs to accelerate future searches. In the context of AWFES:

epoch_prior(fold_n) = optimal_epoch(fold_{n-1})

Benefits

1. Reduced search cost: Prior narrows exploration 2. Temporal adaptation: Captures regime-specific patterns 3. Stability: Prevents erratic epoch switching

Risk: Path Dependency

Warm-starting creates serial correlation in epoch selection. Mitigation:

• Use stability penalty for changes
• Periodically reset to prior-free search
• Monitor epoch selection variance across folds

5. Related Work in Finance

Combinatorial Purged Cross-Validation (CPCV)

López de Prado, M. (2018). _Advances in Financial Machine Learning._ Wiley. Chapter 7.

CPCV addresses look-ahead bias in time series cross-validation by:

1. Testing all possible train/test combinations 2. Purging overlapping samples (embargo) 3. Providing distribution of performance, not point estimate

AWFES is compatible with CPCV: use CPCV for outer loop model evaluation, AWFES for inner loop epoch selection.

Evidence-Based Technical Analysis

Aronson, D. R. (2006). _Evidence-Based Technical Analysis: Applying the Scientific Method and Statistical Inference to Trading Signals._ John Wiley & Sons.

Key contributions:

• Data-mining bias corrections for trading strategies
• Hypothesis testing framework for technical analysis
• Evaluated 6,400+ signaling rules with proper statistical controls

6. Comparison: AWFES vs Related Methods

Key Distinctions

1. WFE vs validation loss: WFE directly measures generalization; validation loss measures prediction accuracy 2. Temporal ordering: AWFES respects time series structure; standard CV does not 3. Discrete candidates: AWFES evaluates all candidates; Bayesian optimization uses surrogate to reduce evaluations 4. Carry-forward: AWFES transfers epoch selection, not model weights

7. Empirical Guidelines

From Pardo (2008)

From López de Prado (2018)

Full Bibliography

@article{bailey2014deflated,
  title={The Deflated Sharpe Ratio: Correcting for Selection Bias, Backtest Overfitting and Non-Normality},
  author={Bailey, David H and L{\'o}pez de Prado, Marcos},
  journal={The Journal of Portfolio Management},
  volume={40},
  number={5},
  pages={94--107},
  year={2014}
}

@book{pardo2008evaluation,
  title={The Evaluation and Optimization of Trading Strategies},
  author={Pardo, Robert E},
  year={2008},
  publisher={John Wiley \& Sons},
  edition={2nd}
}

@book{lopezdeprado2018advances,
  title={Advances in Financial Machine Learning},
  author={L{\'o}pez de Prado, Marcos},
  year={2018},
  publisher={John Wiley \& Sons}
}

@article{bischl2023mohpo,
  title={Multi-Objective Hyperparameter Optimization in Machine Learning—An Overview},
  author={Bischl, Bernd and others},
  journal={ACM Transactions on Evolutionary Learning and Optimization},
  year={2023}
}

@inproceedings{nomura2021warm,
  title={Warm Starting CMA-ES for Hyperparameter Optimization},
  author={Nomura, Masahiro and Ono, Isao},
  booktitle={Proceedings of the AAAI Conference on Artificial Intelligence},
  year={2021}
}

@article{deb2002nsga2,
  title={A Fast and Elitist Multiobjective Genetic Algorithm: NSGA-II},
  author={Deb, Kalyanmoy and others},
  journal={IEEE Transactions on Evolutionary Computation},
  volume={6},
  number={2},
  pages={182--197},
  year={2002}
}

@book{aronson2006evidence,
  title={Evidence-Based Technical Analysis},
  author={Aronson, David R},
  year={2006},
  publisher={John Wiley \& Sons}
}

Anti-Patterns: Adaptive Walk-Forward Epoch Selection

Table of Contents

• 1. Peak Picking (Severity: HIGH)
• 2. Insufficient Folds (Severity: HIGH)
• 3. Ignoring Temporal Autocorrelation (Severity: HIGH)
• 4. Overfitting to In-Sample (Severity: HIGH)
• 5. Using sqrt(252) for Crypto (Severity: MEDIUM)
• 6. Single Epoch Selection (No Uncertainty) (Severity: MEDIUM)
• 7. Expanding Window for Range Bar Training (CRITICAL)
• 8. Meta-Overfitting (Overfitting the Epoch Search) (Severity: HIGH)
• Summary Checklist

Common failures and how to avoid them.

1. Peak Picking (Severity: HIGH)

Symptom

Best epoch is always at the boundary of the search space.

Epoch candidates: [400, 800, 1000, 2000]
Selected epochs across folds: [2000, 2000, 400, 2000, 400, 2000, ...]

Root Cause

Search space doesn't contain the true optimum. The optimal epoch is outside the tested range.

Detection

def detect_peak_picking(selection_history: list[int], epoch_configs: list[int]) -> bool:
    """Returns True if >50% selections are at boundaries."""
    min_epoch, max_epoch = min(epoch_configs), max(epoch_configs)
    boundary_count = sum(1 for e in selection_history if e in [min_epoch, max_epoch])
    return boundary_count / len(selection_history) > 0.5

Fix

1. Expand range: If selecting 2000 often, add 3000, 4000 2. Check for plateau: If WFE is flat at boundary, true optimum may be beyond 3. Add intermediate points: If jumping between 400 and 2000, test 600, 1200

# WRONG: Narrow range
EPOCH_CONFIGS = [400, 800]

# BETTER: Use AWFESConfig with appropriate bounds
from adaptive_wfo_epoch import AWFESConfig

config = AWFESConfig.from_search_space(
    min_epoch=200,
    max_epoch=3200,
    granularity=5,  # Log-spaced: [200, 400, 800, 1600, 3200]
)

2. Insufficient Folds (Severity: HIGH)

Symptom

Effective sample size (N_eff) is too low for statistical significance.

Folds: 10
Epochs: 4
N_eff = 10 × (1/√4) × 0.7 ≈ 3.5  # Too low!

Root Cause

Not enough folds to distinguish signal from noise in epoch selection.

Detection

def check_effective_sample_size(
    n_folds: int,
    n_epochs: int,
    autocorr: float = 0.3,
    min_n_eff: int = 10,
) -> bool:
    """Returns True if N_eff is sufficient."""
    import math
    selection_factor = 1 / math.sqrt(n_epochs)
    corr_factor = (1 - autocorr) / (1 + autocorr)
    n_eff = n_folds * selection_factor * corr_factor
    return n_eff >= min_n_eff

Fix

1. Increase folds: Target N_eff ≥ 30 for reliable inference 2. Extend data span: More historical data = more folds 3. Reduce epoch candidates: Fewer choices = higher N_eff

# WRONG: 10 folds with 4 epochs → N_eff ≈ 3.5
N_FOLDS = 10
config = AWFESConfig.from_search_space(min_epoch=400, max_epoch=2000, granularity=4)

# BETTER: 50 folds with 3 epochs → N_eff ≈ 20
N_FOLDS = 50
config = AWFESConfig.from_search_space(min_epoch=400, max_epoch=1600, granularity=3)
# Fewer epochs + more folds = higher effective sample size

3. Ignoring Temporal Autocorrelation (Severity: HIGH)

Symptom

Consecutive folds have correlated performance, making each fold non-independent.

Fold 0: WFE=0.65, epoch=800
Fold 1: WFE=0.64, epoch=800  # Correlated!
Fold 2: WFE=0.66, epoch=800  # Still correlated!

Root Cause

Overlapping training data between consecutive folds, or no embargo period.

Detection

def compute_fold_autocorrelation(wfe_series: list[float], lag: int = 1) -> float:
    """Compute autocorrelation of WFE across folds."""
    import numpy as np
    if len(wfe_series) < lag + 2:
        return float("nan")
    return float(np.corrcoef(wfe_series[:-lag], wfe_series[lag:])[0, 1])

# WARNING if autocorr > 0.3

Fix

1. Add embargo period: Gap between train and test periods 2. Reduce fold overlap: Increase step size between folds 3. Use purged cross-validation: Remove samples that could leak

# WRONG: Adjacent folds with no gap
fold_0: train=[0:1000], test=[1000:1100]
fold_1: train=[100:1100], test=[1100:1200]  # 90% overlap!

# BETTER: Embargo + reduced overlap
fold_0: train=[0:1000], embargo=[1000:1050], test=[1050:1150]
fold_1: train=[500:1500], embargo=[1500:1550], test=[1550:1650]  # 50% overlap

4. Overfitting to In-Sample (Severity: HIGH)

Symptom

In-sample Sharpe is much higher than out-of-sample, even with optimal epoch.

IS_Sharpe: 3.5
OOS_Sharpe: 0.8
WFE: 0.23  # Severe overfitting!

Root Cause

Model is memorizing training data patterns that don't generalize.

Detection

def detect_overfitting(is_sharpe: float, oos_sharpe: float) -> str:
    """Classify overfitting severity.

    Labels aligned with SKILL.md classify_wfe() (see Guardrails G1):
    - EXCELLENT (≥0.70): Excellent transfer, low overfitting
    - ACCEPTABLE (0.50-0.70): Acceptable transfer (alias: GOOD)
    - INVESTIGATE (0.30-0.50): Moderate transfer, investigate
    - REJECT (<0.30): Severe overfitting, reject (alias: SEVERE)

    Note: ACCEPTABLE/GOOD and REJECT/SEVERE are synonyms.
    SKILL.md uses ACCEPTABLE/REJECT; some older code uses GOOD/SEVERE.
    """
    if is_sharpe <= 0:
        return "NO_SIGNAL"

    wfe = oos_sharpe / is_sharpe

    if wfe >= 0.7:
        return "EXCELLENT"
    elif wfe >= 0.5:
        return "ACCEPTABLE"  # Aligned with SKILL.md (was: GOOD)
    elif wfe >= 0.3:
        return "INVESTIGATE"
    else:
        return "REJECT"  # Aligned with SKILL.md (was: SEVERE)

Fix

1. Reduce epochs: Less training time = less memorization 2. Add regularization: Dropout, weight decay, early stopping 3. Simplify model: Fewer parameters = less capacity to overfit 4. Increase training data: More diverse patterns

# WRONG: High capacity, long training
EPOCHS = 2000
HIDDEN_SIZE = 128
DROPOUT = 0.1

# BETTER: Lower capacity, regularized
EPOCHS = 400
HIDDEN_SIZE = 48
DROPOUT = 0.3
WEIGHT_DECAY = 0.01

5. Using sqrt(252) for Crypto (Severity: MEDIUM)

Symptom

Annualized Sharpe ratios are inflated by ~18%.

# Crypto trades 24/7, but using equity assumption
daily_sharpe = 0.1
annual_sharpe = 0.1 * sqrt(252)  # WRONG: 1.59
annual_sharpe = 0.1 * sqrt(365)  # CORRECT: 1.91

# The error: sqrt(365)/sqrt(252) = 1.20 = 20% inflation

Root Cause

Using equity market convention (252 trading days) for crypto (365 days).

Detection

def check_annualization_factor(market: str, factor: float) -> bool:
    """Validate annualization factor for market type."""
    CORRECT_FACTORS = {
        "crypto_daily": 365,
        "crypto_weekly": 7,  # 7 days per week
        "equity_daily": 252,
        "equity_weekly": 5,  # 5 trading days per week
    }
    return factor == CORRECT_FACTORS.get(market, factor)

Fix

# WRONG for crypto (daily to weekly conversion)
sharpe_tw = daily_sharpe * np.sqrt(5)  # Equity assumption

# CORRECT for crypto
sharpe_tw = daily_sharpe * np.sqrt(7)  # Crypto 24/7

# EXCEPTION: Session-filtered crypto (London-NY hours only)
# Use sqrt(5) because you're only trading 5 days

Note: For range bars, use time-weighted Sharpe (sharpe_tw) with compute_time_weighted_sharpe(). See range-bar-metrics.md.

6. Single Epoch Selection (No Uncertainty) (Severity: MEDIUM)

Symptom

Reporting a single "optimal" epoch without confidence interval.

"Optimal epoch: 800"  # WRONG: No uncertainty quantification

Root Cause

Treating epoch selection as deterministic when it's subject to sampling variation.

Detection

Look for reports that:

• Give single epoch value without CI
• Don't report WFE variance across folds
• Don't show epoch distribution

Fix

Report uncertainty in epoch selection:

def report_epoch_selection_with_uncertainty(
    selection_history: list[dict],
) -> dict:
    """Report epoch selection with uncertainty quantification."""
    epochs = [s["epoch"] for s in selection_history]
    wfes = [s["wfe"] for s in selection_history if s["wfe"] is not None]

    return {
        "selected_epoch": max(set(epochs), key=epochs.count),  # Mode
        "epoch_mean": np.mean(epochs),
        "epoch_std": np.std(epochs),
        "wfe_mean": np.mean(wfes),
        "wfe_ci_95": np.percentile(wfes, [2.5, 97.5]),
        "epoch_distribution": {e: epochs.count(e) for e in set(epochs)},
    }

Good reporting:

Optimal epoch: 800 (selected 45% of folds)
Epoch distribution: {400: 20%, 800: 45%, 1000: 25%, 2000: 10%}
WFE at 800: 0.52 [0.38, 0.66] (95% CI)

7. Expanding Window for Range Bar Training (CRITICAL)

Symptom

Training window grows with each fold instead of sliding forward.

EXPANDING WINDOW (WRONG for range bars):
Fold 1:  [====TRAIN====][TEST]                    (3,000 bars)
Fold 5:  [========TRAIN========][TEST]            (15,000 bars)
Fold 10: [============TRAIN============][TEST]    (30,000 bars)
Fold 20: [==================TRAIN==================][TEST]  (60,000 bars)

FIXED WINDOW (CORRECT):
Fold 1:  [====TRAIN====][TEST]                    (3,000 bars)
Fold 5:       [====TRAIN====][TEST]               (3,000 bars)
Fold 10:           [====TRAIN====][TEST]          (3,000 bars)
Fold 20:                     [====TRAIN====][TEST] (3,000 bars)

Root Cause

Misapplying time-series CV conventions to range bar data. Range bars have non-uniform time spacing, making expanding windows especially problematic.

Why This Is Critical for Range Bars

Multi-agent analysis (2026-01-19) identified 7 compounding issues:

Detection

def detect_expanding_window(folds: list[Fold]) -> bool:
    """Returns True if expanding window detected (ANTI-PATTERN)."""
    train_sizes = [f.train_end_idx - f.train_start_idx for f in folds]

    # Fixed window: all sizes equal
    if len(set(train_sizes)) == 1:
        return False  # OK

    # Expanding window: sizes increase monotonically
    is_expanding = all(
        train_sizes[i] <= train_sizes[i+1]
        for i in range(len(train_sizes)-1)
    )

    return is_expanding


def validate_fixed_window(folds: list[Fold]) -> None:
    """Raise error if expanding window detected."""
    if detect_expanding_window(folds):
        raise ValueError(
            "CRITICAL: Expanding window detected for range bar training. "
            "This anti-pattern causes fold non-equivalence, regime dilution, "
            "and biased risk metrics. Use fixed sliding window instead."
        )

Fix

Always use fixed-size sliding window for range bar ML training:

# WRONG: Expanding window (train_start always 0)
def generate_expanding_folds(total_bars, n_folds):
    step = total_bars // n_folds
    for i in range(n_folds):
        train_end = (i + 1) * step
        yield Fold(train_start=0, train_end=train_end, ...)  # Growing!

# CORRECT: Fixed sliding window
def generate_fixed_folds(total_bars, train_size, test_size, step_size):
    for i in range(n_folds):
        train_start = i * step_size
        train_end = train_start + train_size  # Constant size
        yield Fold(train_start=train_start, train_end=train_end, ...)

Statistical Justification

Exceptions

None for range bar ML training.

The only valid expanding window use case is cumulative learning where every historical instance matters (e.g., rare event detection). Range bar prediction requires recency-weighted regime adaptation, which expanding windows prevent.

Enforcement

Add runtime validation to prevent accidental use:

# At experiment start
validate_fixed_window(folds)

# In fold generation
assert all(
    folds[i].train_start_idx > folds[i-1].train_start_idx
    for i in range(1, len(folds))
), "train_start must advance (not anchored at 0)"

---

8. Meta-Overfitting (Overfitting the Epoch Search) (Severity: HIGH)

Symptom

Epoch selection itself overfits to the search space.

Fold 0: epoch=800 (WFE=0.55)
Fold 1: epoch=2000 (WFE=0.53)
Fold 2: epoch=400 (WFE=0.56)
...
# High variance in selection, but aggregate looks good

# Then in production:
Production WFE: 0.35  # Much worse than backtest!

Root Cause

With 4 epochs × 31 folds = 124 selection decisions, some "lucky" selections inflate aggregate WFE.

Detection

def detect_meta_overfitting(
    selection_history: list[dict],
    epoch_configs: list[int],
) -> dict:
    """Detect signs of meta-overfitting."""
    epochs = [s["epoch"] for s in selection_history]

    # High variance is suspicious
    epoch_std = np.std(epochs)
    epoch_mean = np.mean(epochs)
    cv = epoch_std / epoch_mean  # Coefficient of variation

    # Uniform distribution suggests random selection
    from scipy.stats import chisquare
    observed = [epochs.count(e) for e in epoch_configs]
    expected = [len(epochs) / len(epoch_configs)] * len(epoch_configs)
    chi2, p_value = chisquare(observed, expected)

    return {
        "epoch_cv": cv,
        "uniformity_p_value": p_value,
        "is_suspicious": cv > 0.5 or p_value > 0.5,
        "diagnosis": (
            "HIGH_VARIANCE" if cv > 0.5 else
            "NEAR_UNIFORM" if p_value > 0.5 else
            "OK"
        ),
    }

Fix

1. Limit epoch candidates: 3-4 options maximum 2. Use stability penalty: Penalize frequent changes 3. Hold out final folds: Reserve 20% for meta-validation 4. Apply DSR correction: Account for 124 trials in significance test

# WRONG: Too many epoch options (10 options = meta-overfitting risk)
config = AWFESConfig.from_search_space(min_epoch=100, max_epoch=1000, granularity=10)

# BETTER: Limited options with adaptive stability
config = AWFESConfig.from_search_space(min_epoch=400, max_epoch=1600, granularity=3)
# Use AdaptiveStabilityPenalty which derives threshold from WFE variance
from adaptive_wfo_epoch import AdaptiveStabilityPenalty
stability = AdaptiveStabilityPenalty()  # Adapts to observed WFE noise

Summary Checklist

Before deploying adaptive epoch selection:

• [ ] Expanding window: Using fixed sliding window (NOT expanding) for range bars?
• [ ] Peak picking: Are selections clustered at boundaries? (Expand search bounds if yes)
• [ ] Sample size: Is N_eff ≥ 30? (Use fewer epochs or more folds)
• [ ] Autocorrelation: Is fold autocorrelation < 0.3?
• [ ] Overfitting: Is WFE > 0.50 across folds? (Guidelines, not hard thresholds)
• [ ] Annualization: Using AWFESConfig.get_annualization_factor() for correct market/time_unit?
• [ ] Uncertainty: Reporting confidence intervals via BayesianEpochSmoother.get_confidence_interval()?
• [ ] Meta-overfitting: Epoch CV < 0.5? Not near-uniform? (Use AdaptiveStabilityPenalty)
• [ ] IS_Sharpe threshold: Using compute_is_sharpe_threshold(n_samples) instead of fixed 1.0?

If any check fails, investigate before production deployment.

CRITICAL: The expanding window check is a hard gate for range bar training. All other checks are warnings that require investigation.

Principled Configuration: Use AWFESConfig.from_search_space(min_epoch, max_epoch, granularity) to derive all parameters from search bounds. See SKILL.md for details.

Skill: Adaptive WFO Epoch Selection

Principled Configuration Framework

All parameters in AWFES are derived from first principles or data characteristics, not arbitrary magic numbers.

AWFESConfig: Unified Configuration

from dataclasses import dataclass, field
from typing import Literal
import numpy as np

@dataclass
class AWFESConfig:
    """AWFES configuration with principled parameter derivation.

    No magic numbers - all values derived from search space or data.
    """
    # Search space bounds (user-specified)
    min_epoch: int
    max_epoch: int
    granularity: int  # Number of frontier points

    # Derived automatically
    epoch_configs: list[int] = field(init=False)
    prior_variance: float = field(init=False)
    observation_variance: float = field(init=False)

    # Market context for annualization
    # crypto_session_filtered: Use when data is filtered to London-NY weekday hours
    market_type: Literal["crypto_24_7", "crypto_session_filtered", "equity", "forex"] = "crypto_24_7"
    time_unit: Literal["bar", "daily", "weekly"] = "weekly"

    def __post_init__(self):
        # Generate epoch configs with log spacing (optimal for frontier discovery)
        self.epoch_configs = self._generate_epoch_configs()

        # Derive Bayesian variances from search space
        self.prior_variance, self.observation_variance = self._derive_variances()

    def _generate_epoch_configs(self) -> list[int]:
        """Generate epoch candidates with log spacing.

        Log spacing is optimal for efficient frontier because:
        1. Early epochs: small changes matter more (underfit -> fit transition)
        2. Late epochs: diminishing returns (already near convergence)
        3. Uniform coverage of the WFE vs cost trade-off space

        Formula: epoch_i = min x (max/min)^(i/(n-1))
        """
        if self.granularity < 2:
            return [self.min_epoch]

        log_min = np.log(self.min_epoch)
        log_max = np.log(self.max_epoch)
        log_epochs = np.linspace(log_min, log_max, self.granularity)

        return sorted(set(int(round(np.exp(e))) for e in log_epochs))

    def _derive_variances(self) -> tuple[float, float]:
        """Derive Bayesian variances from search space.

        Principle: Prior should span the search space with ~95% coverage.

        For Normal distribution: 95% CI = mean +/- 1.96 sigma
        If we want 95% of prior mass in [min_epoch, max_epoch]:
            range = max - min = 2 x 1.96 x sigma = 3.92 sigma
            sigma = range / 3.92
            sigma^2 = (range / 3.92)^2

        Observation variance: Set to achieve reasonable learning rate.
        Rule: observation_variance ~ prior_variance / 4
        This means each observation updates the posterior meaningfully
        but doesn't dominate the prior immediately.
        """
        epoch_range = self.max_epoch - self.min_epoch
        prior_std = epoch_range / 3.92  # 95% CI spans search space
        prior_variance = prior_std ** 2

        # Observation variance: 1/4 of prior for balanced learning
        # This gives ~0.2 weight to each new observation initially
        observation_variance = prior_variance / 4

        return prior_variance, observation_variance

    @classmethod
    def from_search_space(
        cls,
        min_epoch: int,
        max_epoch: int,
        granularity: int = 5,
        market_type: str = "crypto_24_7",
    ) -> "AWFESConfig":
        """Create config from search space bounds."""
        return cls(
            min_epoch=min_epoch,
            max_epoch=max_epoch,
            granularity=granularity,
            market_type=market_type,
        )

    def compute_wfe(
        self,
        is_sharpe: float,
        oos_sharpe: float,
        n_samples: int | None = None,
    ) -> float | None:
        """Compute WFE with data-driven IS_Sharpe threshold."""
        min_is = compute_is_sharpe_threshold(n_samples) if n_samples else 0.1
        if abs(is_sharpe) < min_is:
            return None
        return oos_sharpe / is_sharpe

    def get_annualization_factor(self) -> float:
        """Get annualization factor to scale Sharpe from time_unit to ANNUAL.

        IMPORTANT: This returns sqrt(periods_per_year) for scaling to ANNUAL Sharpe.
        For daily-to-weekly scaling, use get_daily_to_weekly_factor() instead.

        Principled derivation:
        - Sharpe scales with sqrt(periods per year)
        - Crypto 24/7: 365 days/year, 52.14 weeks/year
        - Crypto session-filtered: 252 days/year (like equity)
        - Equity: 252 trading days/year, ~52 weeks/year
        - Forex: ~252 days/year (varies by pair)
        """
        PERIODS_PER_YEAR = {
            ("crypto_24_7", "daily"): 365,
            ("crypto_24_7", "weekly"): 52.14,
            ("crypto_24_7", "bar"): None,  # Cannot annualize bars directly
            ("crypto_session_filtered", "daily"): 252,  # London-NY weekdays only
            ("crypto_session_filtered", "weekly"): 52,
            ("equity", "daily"): 252,
            ("equity", "weekly"): 52,
            ("forex", "daily"): 252,
        }

        key = (self.market_type, self.time_unit)
        periods = PERIODS_PER_YEAR.get(key)

        if periods is None:
            raise ValueError(
                f"Cannot annualize {self.time_unit} for {self.market_type}. "
                "Use daily or weekly aggregation first."
            )

        return np.sqrt(periods)

    def get_daily_to_weekly_factor(self) -> float:
        """Get factor to scale DAILY Sharpe to WEEKLY Sharpe.

        This is different from get_annualization_factor()!
        - Daily -> Weekly: sqrt(days_per_week)
        - Daily -> Annual: sqrt(days_per_year)  (use get_annualization_factor)

        Market-specific:
        - Crypto 24/7: sqrt(7) = 2.65 (7 trading days/week)
        - Crypto session-filtered: sqrt(5) = 2.24 (weekdays only)
        - Equity: sqrt(5) = 2.24 (5 trading days/week)
        """
        DAYS_PER_WEEK = {
            "crypto_24_7": 7,
            "crypto_session_filtered": 5,  # London-NY weekdays only
            "equity": 5,
            "forex": 5,
        }

        days = DAYS_PER_WEEK.get(self.market_type)
        if days is None:
            raise ValueError(f"Unknown market type: {self.market_type}")

        return np.sqrt(days)

IS_Sharpe Threshold: Signal-to-Noise Derivation

def compute_is_sharpe_threshold(n_samples: int | None = None) -> float:
    """Compute minimum IS_Sharpe threshold from signal-to-noise ratio.

    Principle: IS_Sharpe must be statistically distinguishable from zero.

    Under null hypothesis (no skill), Sharpe ~ N(0, 1/sqrt(n)).
    To reject null at alpha=0.05 (one-sided), need Sharpe > 1.645/sqrt(n).

    For practical use, we use 2 sigma threshold (~97.7% confidence):
        threshold = 2.0 / sqrt(n)

    This adapts to sample size:
    - n=100: threshold ~ 0.20
    - n=400: threshold ~ 0.10
    - n=1600: threshold ~ 0.05

    Fallback for unknown n: 0.1 (assumes n~400, typical fold size)

    Rationale for 0.1 fallback:
    - 2/sqrt(400) = 0.1, so 0.1 assumes ~400 samples per fold
    - This is conservative: 400 samples is typical for weekly folds
    - If actual n is smaller, threshold is looser (accepts more noise)
    - If actual n is larger, threshold is tighter (fine, we're conservative)
    - The 0.1 value also corresponds to "not statistically distinguishable
      from zero at reasonable sample sizes" - a natural floor for Sharpe SE
    """
    if n_samples is None or n_samples < 10:
        # Conservative fallback: 0.1 assumes ~400 samples (typical fold size)
        # Derivation: 2/sqrt(400) = 0.1; see rationale above
        return 0.1

    return 2.0 / np.sqrt(n_samples)

Skill: Adaptive WFO Epoch Selection

Efficient Frontier Algorithm

Pareto-Optimal Epoch Selection

def compute_efficient_frontier(
    epoch_metrics: list[dict],
    wfe_weight: float = 1.0,
    time_weight: float = 0.1,
) -> tuple[list[int], int]:
    """
    Find Pareto-optimal epochs and select best.

    An epoch is on the frontier if no other epoch dominates it
    (better WFE AND lower training time).

    Args:
        epoch_metrics: List of {epoch, wfe, training_time_sec}
        wfe_weight: Weight for WFE in selection (higher = prefer generalization)
        time_weight: Weight for training time (higher = prefer speed)

    Returns:
        (frontier_epochs, selected_epoch)
    """
    import numpy as np

    # Filter valid metrics
    valid = [(m["epoch"], m["wfe"], m.get("training_time_sec", m["epoch"]))
             for m in epoch_metrics
             if m["wfe"] is not None and np.isfinite(m["wfe"])]

    if not valid:
        # Fallback: return epoch with best OOS Sharpe
        best_oos = max(epoch_metrics, key=lambda m: m.get("oos_sharpe", 0))
        return ([best_oos["epoch"]], best_oos["epoch"])

    # Pareto dominance check
    frontier = []
    for i, (epoch_i, wfe_i, time_i) in enumerate(valid):
        dominated = False
        for j, (epoch_j, wfe_j, time_j) in enumerate(valid):
            if i == j:
                continue
            # j dominates i if: better/equal WFE AND lower/equal time (strict in at least one)
            if (wfe_j >= wfe_i and time_j <= time_i and
                (wfe_j > wfe_i or time_j < time_i)):
                dominated = True
                break
        if not dominated:
            frontier.append((epoch_i, wfe_i, time_i))

    frontier_epochs = [e for e, _, _ in frontier]

    if len(frontier) == 1:
        return (frontier_epochs, frontier[0][0])

    # Weighted score selection
    wfes = np.array([w for _, w, _ in frontier])
    times = np.array([t for _, _, t in frontier])

    wfe_norm = (wfes - wfes.min()) / (wfes.max() - wfes.min() + 1e-10)
    time_norm = (times.max() - times) / (times.max() - times.min() + 1e-10)

    scores = wfe_weight * wfe_norm + time_weight * time_norm
    best_idx = np.argmax(scores)

    return (frontier_epochs, frontier[best_idx][0])

Carry-Forward Mechanism

class AdaptiveEpochSelector:
    """Maintains epoch selection state across WFO folds with adaptive stability."""

    def __init__(self, epoch_configs: list[int]):
        self.epoch_configs = epoch_configs
        self.selection_history: list[dict] = []
        self.last_selected: int | None = None
        self.stability = AdaptiveStabilityPenalty()  # Use adaptive, not fixed

    def select_epoch(self, epoch_metrics: list[dict]) -> int:
        """Select epoch with adaptive stability penalty for changes."""
        frontier_epochs, candidate = compute_efficient_frontier(epoch_metrics)

        # Apply adaptive stability penalty if changing epochs
        if self.last_selected is not None and candidate != self.last_selected:
            candidate_wfe = next(
                m["wfe"] for m in epoch_metrics if m["epoch"] == candidate
            )
            last_wfe = next(
                (m["wfe"] for m in epoch_metrics if m["epoch"] == self.last_selected),
                0.0
            )

            # Use adaptive threshold derived from WFE variance
            if not self.stability.should_change_epoch(
                last_wfe, candidate_wfe, self.last_selected, candidate
            ):
                candidate = self.last_selected

        # Record and return
        self.selection_history.append({
            "epoch": candidate,
            "frontier": frontier_epochs,
            "changed": candidate != self.last_selected,
        })
        self.last_selected = candidate
        return candidate

Decision Tree: Adaptive Walk-Forward Epoch Selection

Table of Contents

• Master Decision Tree
• Detailed Decision Nodes
• Node 1: Validate Prerequisites
• Node 2: IS_Sharpe Validation
• Node 3: WFE Computation
• Node 4: WFE Threshold Check
• Node 5: Efficient Frontier
• Node 6: Stability Penalty
• Node 7: Record and Carry Forward
• Complete Pipeline Example
• Diagnostic Checks
• Check 1: Peak Picking
• Check 2: Selection Stability
• Check 3: WFE Distribution
• Summary Flowchart
• Quick Reference: Thresholds

Practitioner decision tree for implementing AWFES in production.

Master Decision Tree

                                    START
                                      │
                                      ▼
                    ┌─────────────────────────────────────┐
                    │  1. VALIDATE PREREQUISITES          │
                    │     - Data span ≥ 2 years?          │
                    │     - Folds ≥ 30?                   │
                    │     - Epoch range defined?          │
                    └─────────────────────────────────────┘
                                      │
                         ┌────────────┴────────────┐
                         │                         │
                        YES                        NO
                         │                         │
                         ▼                         ▼
              ┌──────────────────┐       ┌──────────────────┐
              │ Proceed to       │       │ STOP: Expand     │
              │ Step 2           │       │ data or reduce   │
              │                  │       │ fold complexity  │
              └──────────────────┘       └──────────────────┘
                         │
                         ▼
                    ┌─────────────────────────────────────┐
                    │  2. COMPUTE IS_SHARPE FOR FOLD      │
                    │     Train model, evaluate IS        │
                    └─────────────────────────────────────┘
                                      │
                         ┌────────────┴────────────┐
                         │                         │
                   IS_SR > 1.0?              IS_SR ≤ 1.0?
                         │                         │
                         ▼                         ▼
              ┌──────────────────┐       ┌──────────────────┐
              │ WFE is valid     │       │ WFE INVALID      │
              │ Continue         │       │ Use fallback:    │
              │                  │       │ - Previous epoch │
              │                  │       │ - Median epoch   │
              └──────────────────┘       └──────────────────┘
                         │
                         ▼
                    ┌─────────────────────────────────────┐
                    │  3. COMPUTE WFE FOR EACH EPOCH      │
                    │     WFE = OOS_SR / IS_SR            │
                    └─────────────────────────────────────┘
                                      │
                                      ▼
                    ┌─────────────────────────────────────┐
                    │  4. CHECK WFE THRESHOLD             │
                    │     Any WFE ≥ 0.30?                 │
                    └─────────────────────────────────────┘
                                      │
                         ┌────────────┴────────────┐
                         │                         │
                        YES                        NO
                         │                         │
                         ▼                         ▼
              ┌──────────────────┐       ┌──────────────────┐
              │ Continue to      │       │ REJECT ALL       │
              │ frontier         │       │ Severe overfit   │
              │ analysis         │       │ Investigate      │
              │                  │       │ model/features   │
              └──────────────────┘       └──────────────────┘
                         │
                         ▼
                    ┌─────────────────────────────────────┐
                    │  5. COMPUTE EFFICIENT FRONTIER      │
                    │     Find Pareto-optimal epochs      │
                    └─────────────────────────────────────┘
                                      │
                                      ▼
                    ┌─────────────────────────────────────┐
                    │  6. APPLY STABILITY PENALTY         │
                    │     Change only if >10% improvement │
                    └─────────────────────────────────────┘
                                      │
                                      ▼
                    ┌─────────────────────────────────────┐
                    │  7. SELECT & RECORD EPOCH           │
                    │     - Record selection history      │
                    │     - Carry forward to next fold    │
                    └─────────────────────────────────────┘
                                      │
                                      ▼
                                  NEXT FOLD

Detailed Decision Nodes

Node 1: Validate Prerequisites

def validate_prerequisites(
    data_span_years: float,
    n_folds: int,
    epoch_configs: list[int],
) -> tuple[bool, list[str]]:
    """Check if prerequisites are met."""
    issues = []

    if data_span_years < 2:
        issues.append(f"Data span {data_span_years:.1f} years < 2 years minimum")

    if n_folds < 30:
        issues.append(f"Folds {n_folds} < 30 minimum for statistical significance")

    if len(epoch_configs) < 2:
        issues.append("Need at least 2 epoch candidates")

    if len(epoch_configs) > 5:
        issues.append(f"Too many epochs ({len(epoch_configs)}) - limit to 3-5")

    # Check geometric spacing
    ratios = [epoch_configs[i+1] / epoch_configs[i]
              for i in range(len(epoch_configs) - 1)]
    if max(ratios) / min(ratios) > 2:
        issues.append("Epoch spacing should be roughly geometric")

    return len(issues) == 0, issues

Actions by Outcome:

Node 2: IS_Sharpe Validation

def check_is_sharpe(is_sharpe: float, min_threshold: float = 1.0) -> dict:
    """Validate in-sample Sharpe is sufficient for WFE computation."""
    return {
        "is_valid": is_sharpe >= min_threshold,
        "is_sharpe": is_sharpe,
        "action": "proceed" if is_sharpe >= min_threshold else "use_fallback",
        "reason": (
            None if is_sharpe >= min_threshold
            else f"IS_Sharpe {is_sharpe:.2f} < {min_threshold} threshold"
        ),
    }

Fallback Strategy:

def get_fallback_epoch(
    previous_epoch: int | None,
    epoch_configs: list[int],
    selection_history: list[dict],
) -> int:
    """Get fallback epoch when WFE is invalid."""
    # Priority 1: Use previous fold's selection
    if previous_epoch is not None:
        return previous_epoch

    # Priority 2: Use mode from selection history
    if selection_history:
        epochs = [s["epoch"] for s in selection_history]
        return max(set(epochs), key=epochs.count)

    # Priority 3: Use median of config range
    return sorted(epoch_configs)[len(epoch_configs) // 2]

Node 3: WFE Computation

def compute_all_wfes(
    epoch_results: list[dict],
    is_sharpe_min: float = 1.0,
) -> list[dict]:
    """Compute WFE for each epoch candidate."""
    wfe_results = []

    for result in epoch_results:
        epoch = result["epoch"]
        is_sharpe = result["is_sharpe"]
        oos_sharpe = result["oos_sharpe"]
        training_time = result.get("training_time_sec", epoch)

        if is_sharpe < is_sharpe_min:
            wfe = None
            status = "IS_TOO_LOW"
        elif oos_sharpe < 0:
            wfe = oos_sharpe / is_sharpe  # Negative WFE
            status = "NEGATIVE_OOS"
        else:
            wfe = oos_sharpe / is_sharpe
            status = "VALID"

        wfe_results.append({
            "epoch": epoch,
            "wfe": wfe,
            "status": status,
            "is_sharpe": is_sharpe,
            "oos_sharpe": oos_sharpe,
            "training_time_sec": training_time,
        })

    return wfe_results

Node 4: WFE Threshold Check

def check_wfe_threshold(
    wfe_results: list[dict],
    hard_reject: float = 0.30,
    warning: float = 0.50,
) -> dict:
    """Check if any epoch passes WFE threshold."""
    valid_wfes = [r["wfe"] for r in wfe_results if r["wfe"] is not None]

    if not valid_wfes:
        return {
            "decision": "REJECT_ALL",
            "reason": "No valid WFE values computed",
            "max_wfe": None,
        }

    max_wfe = max(valid_wfes)

    if max_wfe < hard_reject:
        return {
            "decision": "REJECT_ALL",
            "reason": f"Max WFE {max_wfe:.2f} < {hard_reject} (severe overfitting)",
            "max_wfe": max_wfe,
        }
    elif max_wfe < warning:
        return {
            "decision": "WARNING",
            "reason": f"Max WFE {max_wfe:.2f} < {warning} (moderate overfitting)",
            "max_wfe": max_wfe,
        }
    else:
        return {
            "decision": "PROCEED",
            "reason": None,
            "max_wfe": max_wfe,
        }

Actions by Decision:

Node 5: Efficient Frontier

def find_efficient_frontier(wfe_results: list[dict]) -> list[dict]:
    """Find Pareto-optimal epochs (maximize WFE, minimize time)."""
    valid = [r for r in wfe_results if r["wfe"] is not None]

    if not valid:
        return []

    frontier = []
    for candidate in valid:
        dominated = False
        for other in valid:
            if candidate["epoch"] == other["epoch"]:
                continue
            # Other dominates if: better/equal WFE AND lower/equal time
            # with at least one strict inequality
            if (other["wfe"] >= candidate["wfe"] and
                other["training_time_sec"] <= candidate["training_time_sec"] and
                (other["wfe"] > candidate["wfe"] or
                 other["training_time_sec"] < candidate["training_time_sec"])):
                dominated = True
                break

        if not dominated:
            frontier.append(candidate)

    return sorted(frontier, key=lambda x: x["wfe"], reverse=True)

Node 6: Stability Penalty

def apply_stability_penalty(
    frontier: list[dict],
    previous_epoch: int | None,
    min_improvement: float = 0.10,
) -> dict:
    """Select from frontier with stability penalty."""
    if not frontier:
        raise ValueError("Empty frontier")

    # Best by WFE
    best = frontier[0]

    if previous_epoch is None:
        return {
            "selected": best["epoch"],
            "changed": True,
            "reason": "Initial selection (no previous)",
        }

    # Find previous epoch in results
    prev_result = next(
        (r for r in frontier if r["epoch"] == previous_epoch),
        None
    )

    if prev_result is None:
        # Previous not on frontier - must change
        return {
            "selected": best["epoch"],
            "changed": True,
            "reason": f"Previous epoch {previous_epoch} not on frontier",
        }

    # Check if improvement exceeds threshold
    improvement = (best["wfe"] - prev_result["wfe"]) / prev_result["wfe"]

    if improvement > min_improvement:
        return {
            "selected": best["epoch"],
            "changed": True,
            "reason": f"Improvement {improvement:.1%} > {min_improvement:.0%} threshold",
        }
    else:
        return {
            "selected": previous_epoch,
            "changed": False,
            "reason": f"Improvement {improvement:.1%} < {min_improvement:.0%} threshold",
        }

Node 7: Record and Carry Forward

def record_selection(
    fold_idx: int,
    selected_epoch: int,
    wfe_results: list[dict],
    frontier: list[dict],
    selection_history: list[dict],
) -> dict:
    """Record selection for tracking and analysis."""
    selected_result = next(
        (r for r in wfe_results if r["epoch"] == selected_epoch),
        None
    )

    record = {
        "fold_idx": fold_idx,
        "epoch": selected_epoch,
        "wfe": selected_result["wfe"] if selected_result else None,
        "frontier_epochs": [r["epoch"] for r in frontier],
        "all_wfes": {r["epoch"]: r["wfe"] for r in wfe_results},
        "changed": (
            len(selection_history) == 0 or
            selection_history[-1]["epoch"] != selected_epoch
        ),
    }

    selection_history.append(record)
    return record

Complete Pipeline Example

def run_adaptive_epoch_selection(
    data: pd.DataFrame,
    epoch_configs: list[int] = [400, 800, 1000, 2000],
    n_folds: int = 50,
    min_wfe_improvement: float = 0.10,
) -> dict:
    """Complete AWFES pipeline."""

    # 1. Validate prerequisites
    data_span_years = (data.index[-1] - data.index[0]).days / 365
    is_valid, issues = validate_prerequisites(data_span_years, n_folds, epoch_configs)

    if not is_valid:
        raise ValueError(f"Prerequisites not met: {issues}")

    # Initialize
    selection_history = []
    previous_epoch = None
    fold_results = []

    # Generate folds
    folds = generate_wfo_folds(data, n_folds)

    for fold_idx, fold in enumerate(folds):
        # 2-3. Train all epochs, compute WFE
        epoch_results = []
        for epoch in epoch_configs:
            is_sharpe, oos_sharpe, time_sec = train_and_evaluate(fold, epoch)
            epoch_results.append({
                "epoch": epoch,
                "is_sharpe": is_sharpe,
                "oos_sharpe": oos_sharpe,
                "training_time_sec": time_sec,
            })

        wfe_results = compute_all_wfes(epoch_results)

        # 4. Check threshold
        threshold_check = check_wfe_threshold(wfe_results)

        if threshold_check["decision"] == "REJECT_ALL":
            # Use fallback
            selected = get_fallback_epoch(previous_epoch, epoch_configs, selection_history)
            record = {
                "fold_idx": fold_idx,
                "epoch": selected,
                "wfe": None,
                "status": "FALLBACK",
                "reason": threshold_check["reason"],
            }
        else:
            # 5. Compute frontier
            frontier = find_efficient_frontier(wfe_results)

            # 6. Apply stability penalty
            selection = apply_stability_penalty(frontier, previous_epoch)
            selected = selection["selected"]

            # 7. Record
            record = record_selection(
                fold_idx, selected, wfe_results, frontier, selection_history
            )

        fold_results.append(record)
        previous_epoch = selected

    # Aggregate results
    return {
        "fold_results": fold_results,
        "selection_history": selection_history,
        "summary": summarize_selection_history(selection_history),
    }

Diagnostic Checks

After completing the pipeline, run these diagnostics:

Check 1: Peak Picking

def diagnose_peak_picking(history: list[dict], epoch_configs: list[int]) -> dict:
    """Check if selections cluster at boundaries."""
    epochs = [h["epoch"] for h in history if h["epoch"] is not None]
    min_e, max_e = min(epoch_configs), max(epoch_configs)

    boundary_count = sum(1 for e in epochs if e in [min_e, max_e])
    boundary_rate = boundary_count / len(epochs) if epochs else 0

    return {
        "boundary_rate": boundary_rate,
        "is_problematic": boundary_rate > 0.5,
        "recommendation": (
            "Expand epoch range" if boundary_rate > 0.5
            else "Range appears adequate"
        ),
    }

Check 2: Selection Stability

def diagnose_stability(history: list[dict]) -> dict:
    """Check selection stability across folds."""
    changes = sum(1 for h in history if h.get("changed", False))
    change_rate = changes / len(history) if history else 0

    epochs = [h["epoch"] for h in history if h["epoch"] is not None]
    epoch_cv = np.std(epochs) / np.mean(epochs) if epochs else 0

    return {
        "change_rate": change_rate,
        "epoch_cv": epoch_cv,
        "is_stable": change_rate < 0.3 and epoch_cv < 0.5,
        "recommendation": (
            "Consider increasing stability penalty" if change_rate > 0.3
            else "Stability acceptable"
        ),
    }

Check 3: WFE Distribution

def diagnose_wfe_distribution(history: list[dict]) -> dict:
    """Analyze WFE distribution across folds."""
    wfes = [h["wfe"] for h in history if h.get("wfe") is not None]

    if not wfes:
        return {"status": "NO_VALID_WFE", "recommendation": "Investigate model"}

    return {
        "mean": np.mean(wfes),
        "median": np.median(wfes),
        "std": np.std(wfes),
        "ci_95": np.percentile(wfes, [2.5, 97.5]).tolist(),
        "below_threshold": sum(1 for w in wfes if w < 0.30) / len(wfes),
        "status": "HEALTHY" if np.median(wfes) >= 0.50 else "CONCERNING",
    }

Summary Flowchart

┌────────────────────────────────────────────────────────────────────┐
│                    AWFES DECISION SUMMARY                          │
├────────────────────────────────────────────────────────────────────┤
│                                                                    │
│  Prerequisites OK? ──NO──> Fix data/folds first                   │
│         │                                                          │
│        YES                                                         │
│         │                                                          │
│         ▼                                                          │
│  IS_Sharpe > 1.0? ──NO──> Use fallback epoch                      │
│         │                                                          │
│        YES                                                         │
│         │                                                          │
│         ▼                                                          │
│  Any WFE > 0.30? ──NO──> REJECT: Severe overfitting               │
│         │                                                          │
│        YES                                                         │
│         │                                                          │
│         ▼                                                          │
│  Compute Efficient Frontier                                        │
│         │                                                          │
│         ▼                                                          │
│  Improvement > 10%? ──NO──> Keep previous epoch                   │
│         │                                                          │
│        YES                                                         │
│         │                                                          │
│         ▼                                                          │
│  Select new epoch, record, carry forward                          │
│                                                                    │
└────────────────────────────────────────────────────────────────────┘

Quick Reference: Thresholds

Skill: Adaptive WFO Epoch Selection

Epoch Smoothing Methods

Why Smooth Epoch Selections?

Raw per-fold epoch selections are noisy due to:

• Limited validation data per fold
• Regime changes between folds
• Stochastic training dynamics

Smoothing reduces variance while preserving signal.

Method Comparison

Bayesian Updating (Primary Method)

def bayesian_epoch_update(
    prior_mean: float,
    prior_variance: float,
    observed_epoch: int,
    observation_variance: float,
    wfe_weight: float = 1.0,
) -> tuple[float, float]:
    """Single Bayesian update step.

    Mathematical formulation:
    - Prior: N(mu_0, sigma_0^2)
    - Observation: N(x, sigma_obs^2/wfe)  # WFE-weighted
    - Posterior: N(mu_1, sigma_1^2)

    Where:
    mu_1 = (mu_0/sigma_0^2 + x*wfe/sigma_obs^2) / (1/sigma_0^2 + wfe/sigma_obs^2)
    sigma_1^2 = 1 / (1/sigma_0^2 + wfe/sigma_obs^2)
    """
    # Effective observation variance (lower WFE = less reliable)
    eff_obs_var = observation_variance / max(wfe_weight, 0.1)

    prior_precision = 1.0 / prior_variance
    obs_precision = 1.0 / eff_obs_var

    posterior_precision = prior_precision + obs_precision
    posterior_mean = (
        prior_precision * prior_mean + obs_precision * observed_epoch
    ) / posterior_precision
    posterior_variance = 1.0 / posterior_precision

    return posterior_mean, posterior_variance

Exponential Moving Average (Alternative)

def ema_epoch_update(
    current_ema: float,
    observed_epoch: int,
    alpha: float = 0.3,
) -> float:
    """EMA update: more weight on recent observations.

    alpha = 0.3 means ~90% of signal from last 7 folds.
    alpha = 0.5 means ~90% of signal from last 4 folds.
    """
    return alpha * observed_epoch + (1 - alpha) * current_ema

Initialization Strategies

# RECOMMENDED: Use AWFESConfig for principled derivation
config = AWFESConfig.from_search_space(
    min_epoch=80,
    max_epoch=400,
    granularity=5,
)
# prior_variance = ((400-80)/3.92)^2 ~ 6,658 (derived automatically)
# observation_variance = prior_variance/4 ~ 1,665 (derived automatically)

# Alternative strategies (if manual configuration needed):

# Strategy 1: Search-space derived (same as AWFESConfig)
epoch_range = max(EPOCH_CONFIGS) - min(EPOCH_CONFIGS)
prior_mean = np.mean(EPOCH_CONFIGS)
prior_variance = (epoch_range / 3.92) ** 2  # 95% CI spans search space

# Strategy 2: Burn-in (use first 5 folds)
burn_in_optima = [run_fold_sweep(fold) for fold in folds[:5]]
prior_mean = np.mean(burn_in_optima)
base_variance = (epoch_range / 3.92) ** 2 / 4  # Reduced after burn-in
prior_variance = max(np.var(burn_in_optima), base_variance)

See epoch-smoothing.md for extended mathematical analysis.

Epoch Smoothing Methods Reference

Detailed mathematical formulation and implementation for epoch smoothing.

Mathematical Foundation

The Problem: Noisy Epoch Selection

Per-fold optimal epochs are noisy estimates of the true optimal:

observed_optimal_i = true_optimal + noise_i

Where noise_i arises from:

• Limited validation samples
• Stochastic training dynamics
• Market regime variation

Goal: Estimate true_optimal by combining noisy observations.

Bayesian Updating (Primary Method)

Conjugate Normal-Normal Model

Assuming:

• Prior: true_optimal ~ N(μ₀, σ₀²)
• Likelihood: observed | true_optimal ~ N(true_optimal, σ²/wfe)

The posterior is:

true_optimal | observed ~ N(μ₁, σ₁²)

where:
μ₁ = (μ₀/σ₀² + x·wfe/σ²) / (1/σ₀² + wfe/σ²)
σ₁² = 1 / (1/σ₀² + wfe/σ²)

WFE Weighting Rationale

WFE measures how reliable the epoch selection is:

• High WFE (0.7+) → validation closely tracks training → reliable selection
• Low WFE (0.3-0.5) → validation diverges → noisy selection

Weighting by WFE gives more influence to reliable observations.

Full Implementation

from dataclasses import dataclass
from typing import Optional
import numpy as np


@dataclass
class BayesianState:
    """State of Bayesian epoch estimator."""

    mean: float
    variance: float
    n_observations: int = 0


class BayesianEpochSmoother:
    """Bayesian smoothing for epoch selection.

    Also known as: BayesianEpochSelector (alias in SKILL.md)

    Uses conjugate Normal-Normal updating with WFE-weighted observations.
    """

    def __init__(
        self,
        epoch_configs: list[int],
        prior_mean: Optional[float] = None,
        prior_variance: Optional[float] = None,
        observation_variance: Optional[float] = None,
        min_wfe_weight: float = 0.1,
    ):
        """Initialize smoother.

        Args:
            epoch_configs: Valid epoch values
            prior_mean: Prior mean (default: midpoint of configs)
            prior_variance: Prior variance (default: derived from search space)
            observation_variance: Base observation noise variance (default: prior_var/4)
            min_wfe_weight: Minimum WFE weight to prevent division by zero

        Variance Derivation (if not provided):
            Prior should span search space with ~95% coverage.
            For Normal: 95% CI = mean ± 1.96σ → range = 3.92σ → σ² = (range/3.92)²
            Observation variance: prior_variance/4 for balanced learning rate.
        """
        self.epoch_configs = sorted(epoch_configs)
        self.min_wfe_weight = min_wfe_weight

        # Derive variances from search space if not provided
        epoch_range = max(self.epoch_configs) - min(self.epoch_configs)
        default_prior_var = (epoch_range / 3.92) ** 2  # 95% CI spans search space
        default_obs_var = default_prior_var / 4  # Balanced learning rate

        self.observation_variance = observation_variance or default_obs_var

        # Initialize state with derived or provided variance
        self.state = BayesianState(
            mean=prior_mean or np.mean(epoch_configs),
            variance=prior_variance or default_prior_var,
            n_observations=0,
        )

        # History for diagnostics
        self.history: list[dict] = []

    def update(self, observed_epoch: int, wfe: float) -> int:
        """Update posterior with new observation.

        Args:
            observed_epoch: Optimal epoch from current fold's validation
            wfe: Walk-Forward Efficiency (reliability weight)

        Returns:
            Smoothed epoch selection (snapped to valid config)
        """
        # Clamp WFE to [min_wfe_weight, 2.0] to prevent extreme weights:
        #   - Lower bound (min_wfe_weight=0.1): Prevents division issues
        #   - Upper bound (2.0): WFE > 2 indicates OOS >> IS, which suggests
        #     regime shift or data anomaly rather than genuine skill transfer.
        #     Capping at 2.0 treats such observations with appropriate skepticism.
        wfe_clamped = max(self.min_wfe_weight, min(wfe, 2.0))

        # Effective observation variance (lower WFE = higher variance)
        eff_obs_var = self.observation_variance / wfe_clamped

        # Bayesian update
        prior_precision = 1.0 / self.state.variance
        obs_precision = 1.0 / eff_obs_var

        posterior_precision = prior_precision + obs_precision
        posterior_mean = (
            prior_precision * self.state.mean +
            obs_precision * observed_epoch
        ) / posterior_precision
        posterior_variance = 1.0 / posterior_precision

        # Record history
        self.history.append({
            "observed_epoch": observed_epoch,
            "wfe": wfe,
            "wfe_clamped": wfe_clamped,
            "prior_mean": self.state.mean,
            "prior_variance": self.state.variance,
            "posterior_mean": posterior_mean,
            "posterior_variance": posterior_variance,
            "selected_epoch": self._snap_to_config(posterior_mean),
        })

        # Update state
        self.state = BayesianState(
            mean=posterior_mean,
            variance=posterior_variance,
            n_observations=self.state.n_observations + 1,
        )

        return self._snap_to_config(posterior_mean)

    def get_current_epoch(self) -> int:
        """Get current smoothed epoch without updating."""
        return self._snap_to_config(self.state.mean)

    def get_confidence_interval(self, level: float = 0.95) -> tuple[int, int]:
        """Get confidence interval for true optimal epoch.

        Args:
            level: Confidence level (default: 95%)

        Returns:
            (lower, upper) epoch bounds
        """
        from scipy.stats import norm

        z = norm.ppf((1 + level) / 2)
        std = np.sqrt(self.state.variance)

        lower = self.state.mean - z * std
        upper = self.state.mean + z * std

        return (
            self._snap_to_config(lower),
            self._snap_to_config(upper),
        )

    def _snap_to_config(self, continuous: float) -> int:
        """Snap continuous value to nearest valid config."""
        return min(self.epoch_configs, key=lambda e: abs(e - continuous))

    def reset(self, prior_mean: Optional[float] = None) -> None:
        """Reset to prior state with derived variance."""
        epoch_range = max(self.epoch_configs) - min(self.epoch_configs)
        default_prior_var = (epoch_range / 3.92) ** 2  # 95% CI spans search space

        self.state = BayesianState(
            mean=prior_mean or np.mean(self.epoch_configs),
            variance=default_prior_var,
            n_observations=0,
        )
        self.history.clear()

Alternative Methods

Exponential Moving Average (EMA)

Simpler than Bayesian, good for quick implementation.

class EMAEpochSmoother:
    """Exponential moving average epoch smoothing."""

    def __init__(
        self,
        epoch_configs: list[int],
        alpha: float = 0.3,
        initial: Optional[float] = None,
    ):
        """Initialize EMA smoother.

        Args:
            epoch_configs: Valid epoch values
            alpha: Smoothing factor (higher = more responsive)
                   α=0.3 → ~90% signal from last 7 observations
                   α=0.5 → ~90% signal from last 4 observations
            initial: Initial EMA value
        """
        self.epoch_configs = sorted(epoch_configs)
        self.alpha = alpha
        self.ema = initial or np.mean(epoch_configs)
        self.history: list[dict] = []

    def update(self, observed_epoch: int) -> int:
        """Update EMA with new observation."""
        new_ema = self.alpha * observed_epoch + (1 - self.alpha) * self.ema

        self.history.append({
            "observed": observed_epoch,
            "prior_ema": self.ema,
            "posterior_ema": new_ema,
            "selected": self._snap_to_config(new_ema),
        })

        self.ema = new_ema
        return self._snap_to_config(new_ema)

    def _snap_to_config(self, continuous: float) -> int:
        return min(self.epoch_configs, key=lambda e: abs(e - continuous))

Simple Moving Average (SMA)

Most stable but slowest to adapt.

class SMAEpochSmoother:
    """Simple moving average epoch smoothing."""

    def __init__(
        self,
        epoch_configs: list[int],
        window: int = 5,
    ):
        self.epoch_configs = sorted(epoch_configs)
        self.window = window
        self.observations: list[int] = []

    def update(self, observed_epoch: int) -> int:
        """Update SMA with new observation."""
        self.observations.append(observed_epoch)
        if len(self.observations) > self.window:
            self.observations.pop(0)

        sma = np.mean(self.observations)
        return self._snap_to_config(sma)

    def _snap_to_config(self, continuous: float) -> int:
        return min(self.epoch_configs, key=lambda e: abs(e - continuous))

Median Smoother

Robust to outliers from regime changes.

class MedianEpochSmoother:
    """Median-based epoch smoothing."""

    def __init__(
        self,
        epoch_configs: list[int],
        window: int = 5,
    ):
        self.epoch_configs = sorted(epoch_configs)
        self.window = window
        self.observations: list[int] = []

    def update(self, observed_epoch: int) -> int:
        """Update with new observation, return median."""
        self.observations.append(observed_epoch)
        if len(self.observations) > self.window:
            self.observations.pop(0)

        median_val = np.median(self.observations)
        return self._snap_to_config(median_val)

    def _snap_to_config(self, continuous: float) -> int:
        return min(self.epoch_configs, key=lambda e: abs(e - continuous))

Method Selection Guide

Recommendations

1. Default choice: Bayesian (principled, handles WFE weighting) 2. Quick prototype: EMA with α=0.3 3. Regime change prone: Median with window=5 4. Maximum stability: SMA with window=7

Initialization Strategies

Strategy 1: Search-Space Derived (RECOMMENDED)

# Principled: Derive from search bounds (no magic numbers)
# Prior spans search space with 95% coverage
epoch_range = max(EPOCH_CONFIGS) - min(EPOCH_CONFIGS)
prior_mean = np.mean(EPOCH_CONFIGS)  # Midpoint
prior_variance = (epoch_range / 3.92) ** 2  # 95% CI spans search space

# Example: EPOCH_CONFIGS = [100, 200, 400, 800, 1600]
# epoch_range = 1500, prior_variance = (1500/3.92)² ≈ 146,506

Strategy 2: Uninformative Prior

# No domain knowledge - very wide prior
prior_mean = np.mean(EPOCH_CONFIGS)  # Midpoint
prior_variance = np.var(EPOCH_CONFIGS) * 4  # Very wide

Strategy 3: Literature-Informed Prior

# BiLSTM literature suggests 100-300 optimal for financial data
# But prefer deriving from YOUR search space
prior_mean = 200
epoch_range = max(EPOCH_CONFIGS) - min(EPOCH_CONFIGS)
prior_variance = (epoch_range / 3.92) ** 2  # Still principled

Strategy 4: Burn-In Initialization

# Use first N folds to establish prior
BURN_IN_FOLDS = 5
burn_in_optima = [get_fold_optimal(fold) for fold in folds[:BURN_IN_FOLDS]]

prior_mean = np.mean(burn_in_optima)
# Combine observed variance with search-space derived base
epoch_range = max(EPOCH_CONFIGS) - min(EPOCH_CONFIGS)
base_variance = (epoch_range / 3.92) ** 2 / 4  # Reduced after burn-in
prior_variance = max(np.var(burn_in_optima), base_variance)

Strategy 5: Empirical Bayes

# Estimate prior from full sweep data (use with caution - slight look-ahead)
all_fold_optima = [r["optimal_epoch"] for r in full_sweep_results]

prior_mean = np.mean(all_fold_optima)
prior_variance = max(np.var(all_fold_optima), 100)  # Floor to prevent collapse

Convergence Analysis

Bayesian Posterior Convergence

After N observations, posterior variance:

σ_N² = 1 / (1/σ₀² + N·wfe_avg/σ_obs²)

Example with principled derivation (search space [100, 2000], granularity=5):

epoch_range = 2000 - 100 = 1900
prior_variance = (1900 / 3.92)² ≈ 235,000
observation_variance = prior_variance / 4 ≈ 58,750

With wfe_avg=0.5:

• After 5 folds: σ² ≈ 31,000 (±176 epochs)
• After 10 folds: σ² ≈ 17,600 (±133 epochs)
• After 20 folds: σ² ≈ 9,600 (±98 epochs)

Key insight: Larger search spaces need more folds to converge. This is principled - uncertainty should scale with search space size.

EMA Effective Memory

For EMA with α:

• Effective window = 2/α - 1
• 90% of signal from last log(0.1)/log(1-α) observations

Diagnostic Plots

Posterior Evolution

import matplotlib.pyplot as plt

def plot_bayesian_evolution(smoother: BayesianEpochSmoother):
    """Plot Bayesian posterior evolution."""
    fig, axes = plt.subplots(2, 1, figsize=(12, 8))

    folds = list(range(len(smoother.history)))
    observed = [h["observed_epoch"] for h in smoother.history]
    posterior_mean = [h["posterior_mean"] for h in smoother.history]
    posterior_std = [np.sqrt(h["posterior_variance"]) for h in smoother.history]

    # Mean evolution
    ax1 = axes[0]
    ax1.scatter(folds, observed, label="Observed", alpha=0.6)
    ax1.plot(folds, posterior_mean, label="Posterior Mean", color="red")
    ax1.fill_between(
        folds,
        [m - 2*s for m, s in zip(posterior_mean, posterior_std)],
        [m + 2*s for m, s in zip(posterior_mean, posterior_std)],
        alpha=0.2, color="red", label="95% CI"
    )
    ax1.set_xlabel("Fold")
    ax1.set_ylabel("Epoch")
    ax1.legend()
    ax1.set_title("Bayesian Epoch Posterior Evolution")

    # Variance evolution
    ax2 = axes[1]
    ax2.plot(folds, posterior_std, color="blue")
    ax2.set_xlabel("Fold")
    ax2.set_ylabel("Posterior Std")
    ax2.set_title("Posterior Uncertainty (decreasing = learning)")

    plt.tight_layout()
    return fig

Evolution Log

Convention: Reverse chronological order (newest on top, oldest at bottom). Prepend new entries.

---

2026-02-26: Initial Evolution Log

Status: Skill is in use and maintained. Track improvements here.

Purpose

This evolution log tracks updates to the skill. Each entry should note:

• What changed (content, structure, tooling)
• Why it changed (bug fix, feature request, best practice)
• Files affected

How to Use

1. When updating SKILL.md or references, add an entry here with the date 2. Keep entries reverse-chronological (newest first) 3. Link to ADRs or GitHub issues when relevant 4. Reference specific line changes when helpful

---

Feature Sets for BiLSTM Training

Reference for standardized feature sets used in AWFES experiments. Documents the evolution from A_baseline (v1) to A_baseline_v2 with stationary features.

Feature Set Evolution

A_baseline (v1) - Legacy 4 Features

Status: Deprecated - use A_baseline_v2 instead.

Known Issues:

• Non-stationary features create distribution shift across folds
• MixedScaler fit can leak information if not carefully applied
• Heavy tails cause gradient instability in LSTM training
• Scale-dependent features don't transfer across assets

A_baseline_v2 - Stationary Features (RECOMMENDED)

Status: Current standard for AWFES experiments.

Why v2 Features Are Better

1. Pre-transformed stationarity: All features bounded and normalized before training 2. No scaler lookahead: TemporalScaler is a no-op since features already normalized 3. Rolling z-score: 100-bar window prevents information leakage 4. Better gradient flow: Bounded ranges prevent exploding/vanishing gradients 5. Cross-asset transferability: Scale-invariant features work across different price levels

Computation Example

def compute_stationary_features(df: pd.DataFrame, zscore_window: int = 100) -> pd.DataFrame:
    """Compute A_baseline_v2 stationary features.

    All features are pre-transformed to be stationary with bounded ranges.
    Uses rolling z-score normalization to prevent lookahead bias.
    """
    # Helper for rolling z-score with clipping
    def rolling_zscore(series: pd.Series, window: int = zscore_window) -> pd.Series:
        mean = series.rolling(window, min_periods=20).mean()
        std = series.rolling(window, min_periods=20).std()
        z = (series - mean) / (std + 1e-8)
        return z.clip(-4, 4)  # Bound to [-4, 4]

    # Raw intermediate calculations
    returns = df["close"].pct_change()
    rolling_vol = returns.rolling(20).std()
    momentum_20 = df["close"].pct_change(20)
    atr_14 = compute_atr(df, 14)  # Your ATR implementation
    volume_ma = df["volume"].rolling(20).mean()
    rsi_14 = compute_rsi(df["close"], 14)  # Your RSI implementation

    # Bollinger Bands
    bb_mid = df["close"].rolling(20).mean()
    bb_std = df["close"].rolling(20).std()
    bb_upper = bb_mid + 2 * bb_std
    bb_lower = bb_mid - 2 * bb_std

    # Stationary features
    df["returns_vs"] = (returns / (rolling_vol + 1e-8)).clip(-4, 4)
    df["momentum_z"] = rolling_zscore(momentum_20)
    df["atr_pct"] = rolling_zscore(atr_14 / df["close"] * 100)
    df["volume_z"] = rolling_zscore(np.log(df["volume"] / (volume_ma + 1)))
    df["rsi_14"] = rsi_14 / 100  # Already bounded [0, 1]
    df["bb_pct_b"] = ((df["close"] - bb_lower) / (bb_upper - bb_lower + 1e-8)).clip(0, 1)
    df["vol_regime"] = (atr_14 > atr_14.rolling(100).median()).astype(float)
    df["return_accel"] = rolling_zscore(returns.rolling(5).mean() - returns.rolling(10).mean())
    df["pv_divergence"] = rolling_zscore(
        returns.rolling(20).corr(df["volume"].pct_change())
    )

    return df


A_BASELINE_V2_FEATURES = [
    "returns_vs",
    "momentum_z",
    "atr_pct",
    "volume_z",
    "rsi_14",
    "bb_pct_b",
    "vol_regime",
    "return_accel",
    "pv_divergence",
]

TemporalScaler (No-Op Scaler)

For A_baseline_v2 features, use TemporalScaler which is a no-op:

class TemporalScaler:
    """No-op scaler for pre-transformed stationary features.

    Features in A_baseline_v2 are already:
    - Z-score normalized with rolling windows
    - Clipped to bounded ranges
    - Stationary by construction

    This scaler exists to maintain API compatibility with pipelines
    that expect a scaler object.
    """

    def fit(self, X: np.ndarray) -> "TemporalScaler":
        return self  # No-op

    def transform(self, X: np.ndarray) -> np.ndarray:
        return X  # Pass-through

    def fit_transform(self, X: np.ndarray) -> np.ndarray:
        return X  # Pass-through

    def inverse_transform(self, X: np.ndarray) -> np.ndarray:
        return X  # Pass-through

Automatic Scaler Selection

def create_sequences_with_scaler(
    df: pd.DataFrame,
    features: list[str],
    target: str,
    seq_len: int,
) -> tuple[np.ndarray, np.ndarray, Any, np.ndarray]:
    """Create sequences with automatic scaler selection.

    If features are a subset of A_BASELINE_V2_FEATURES, uses TemporalScaler.
    Otherwise, uses MixedScaler (legacy behavior).
    """
    if set(features).issubset(set(A_BASELINE_V2_FEATURES)):
        scaler = TemporalScaler()
    else:
        scaler = MixedScaler(features)

    # ... rest of sequence creation
    return X, y, scaler, timestamps

Migration Guide: v1 to v2

Before (v1)

from features import compute_features, A_BASELINE_FEATURES
from scalers import MixedScaler

df = compute_features(df)
features = A_BASELINE_FEATURES  # ['returns', 'momentum_20', 'atr_14', 'volume_change']

scaler = MixedScaler(features)
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

After (v2)

from features import compute_stationary_features, A_BASELINE_V2_FEATURES
from scalers import TemporalScaler

df = compute_stationary_features(df)
features = A_BASELINE_V2_FEATURES  # 9 stationary features

scaler = TemporalScaler()  # No-op, features already normalized
X_train_scaled = scaler.fit_transform(X_train)  # Pass-through
X_test_scaled = scaler.transform(X_test)  # Pass-through

Validation Checklist

Before using a feature set in AWFES:

• [ ] All features bounded (no unbounded ranges)
• [ ] Stationarity test passes (ADF p < 0.05)
• [ ] No lookahead in feature computation (rolling windows only)
• [ ] Scaler fit on train only (or no scaler needed for v2)
• [ ] Feature correlation < 0.95 (no redundant features)
• [ ] Missing values handled (forward-fill or drop)

References

• rangebar-eval-metrics - Metric computation
• look-ahead-bias.md - Bias prevention
• anti-patterns.md - Common mistakes

Skill: Adaptive WFO Epoch Selection

Guardrails (Principled Guidelines)

G1: WFE Thresholds

The traditional thresholds (0.30, 0.50, 0.70) are guidelines based on practitioner consensus, not derived from first principles. They represent:

# These are GUIDELINES, not hard rules
# Adjust based on your domain and risk tolerance
WFE_THRESHOLDS = {
    "hard_reject": 0.30,  # Below this: almost certainly overfitting
    "warning": 0.50,      # Below this: significant signal loss
    "target": 0.70,       # Above this: good generalization
}

def classify_wfe(wfe: float | None) -> str:
    """Classify WFE with principled thresholds."""
    if wfe is None:
        return "INVALID"  # IS_Sharpe below noise floor
    if wfe < WFE_THRESHOLDS["hard_reject"]:
        return "REJECT"
    if wfe < WFE_THRESHOLDS["warning"]:
        return "INVESTIGATE"
    if wfe < WFE_THRESHOLDS["target"]:
        return "ACCEPTABLE"
    return "EXCELLENT"

G2: IS_Sharpe Minimum (Data-Driven)

OLD (magic number):

# WRONG: Fixed threshold regardless of sample size
if is_sharpe < 1.0:
    wfe = None

NEW (principled):

# CORRECT: Threshold adapts to sample size
min_is_sharpe = compute_is_sharpe_threshold(n_samples)
if is_sharpe < min_is_sharpe:
    wfe = None  # Below noise floor for this sample size

The threshold derives from the standard error of Sharpe ratio: SE(SR) ~ 1/sqrt(n).

Note on SE(Sharpe) approximation: The formula 1/sqrt(n) is a first-order approximation valid when SR is small (close to 0). The full Lo (2002) formula is:

SE(SR) = sqrt((1 + 0.5*SR^2) / n)

For high-Sharpe strategies (SR > 1.0), the simplified formula underestimates SE by ~25-50%. Use the full formula when evaluating strategies with SR > 1.0.

G3: Stability Penalty for Epoch Changes (Adaptive)

The stability penalty prevents hyperparameter churn. Instead of fixed thresholds, use relative improvement based on WFE variance:

def compute_stability_threshold(wfe_history: list[float]) -> float:
    """Compute stability threshold from observed WFE variance.

    Principle: Require improvement exceeding noise level.

    If WFE has std=0.15 across folds, random fluctuation could be +/-0.15.
    To distinguish signal from noise, require improvement > 1 sigma of WFE.

    Minimum: 5% (prevent switching on negligible improvements)
    Maximum: 20% (don't be overly conservative)
    """
    if len(wfe_history) < 3:
        return 0.10  # Default until enough history

    wfe_std = np.std(wfe_history)
    threshold = max(0.05, min(0.20, wfe_std))
    return threshold


class AdaptiveStabilityPenalty:
    """Stability penalty that adapts to observed WFE variance."""

    def __init__(self):
        self.wfe_history: list[float] = []
        self.epoch_changes: list[int] = []

    def should_change_epoch(
        self,
        current_wfe: float,
        candidate_wfe: float,
        current_epoch: int,
        candidate_epoch: int,
    ) -> bool:
        """Decide whether to change epochs based on adaptive threshold."""
        self.wfe_history.append(current_wfe)

        if current_epoch == candidate_epoch:
            return False  # Same epoch, no change needed

        threshold = compute_stability_threshold(self.wfe_history)
        improvement = (candidate_wfe - current_wfe) / max(abs(current_wfe), 0.01)

        if improvement > threshold:
            self.epoch_changes.append(len(self.wfe_history))
            return True

        return False  # Improvement not significant

G4: DSR Adjustment for Epoch Search (Principled)

def adjusted_dsr_for_epoch_search(
    sharpe: float,
    n_folds: int,
    n_epochs: int,
    sharpe_se: float | None = None,
    n_samples_per_fold: int | None = None,
) -> float:
    """Deflated Sharpe Ratio accounting for epoch selection multiplicity.

    When selecting from K epochs, the expected maximum Sharpe under null
    is inflated. This adjustment corrects for that selection bias.

    Principled SE estimation:
    - If n_samples provided: SE(Sharpe) ~ 1/sqrt(n)
    - Otherwise: estimate from typical fold size

    Reference: Bailey & Lopez de Prado (2014), Gumbel distribution
    """
    from math import sqrt, log, pi

    n_trials = n_folds * n_epochs  # Total selection events

    if n_trials < 2:
        return sharpe  # No multiple testing correction needed

    # Expected maximum under null (Gumbel distribution)
    # E[max(Z_1, ..., Z_n)] ~ sqrt(2*ln(n)) - (gamma + ln(pi/2)) / sqrt(2*ln(n))
    # where gamma ~ 0.5772 is Euler-Mascheroni constant
    euler_gamma = 0.5772156649
    sqrt_2_log_n = sqrt(2 * log(n_trials))
    e_max_z = sqrt_2_log_n - (euler_gamma + log(pi / 2)) / sqrt_2_log_n

    # Estimate Sharpe SE if not provided
    if sharpe_se is None:
        if n_samples_per_fold is not None:
            sharpe_se = 1.0 / sqrt(n_samples_per_fold)
        else:
            # Conservative default: assume ~300 samples per fold
            sharpe_se = 1.0 / sqrt(300)

    # Expected maximum Sharpe under null
    e_max_sharpe = e_max_z * sharpe_se

    # Deflated Sharpe
    return max(0, sharpe - e_max_sharpe)

Example: For 5 epochs x 50 folds = 250 trials with 300 samples/fold:

• sharpe_se ~ 0.058
• e_max_z ~ 2.88
• e_max_sharpe ~ 0.17
• A Sharpe of 1.0 deflates to 0.83 after adjustment.

Skill: Adaptive WFO Epoch Selection

Look-Ahead Bias Prevention

The Problem

Using the same data for epoch selection AND final evaluation creates look-ahead bias:

WRONG: Use fold's own optimal epoch for fold's OOS evaluation
   - Epoch selection "sees" validation returns
   - Then apply same epoch to OOS from same period
   - Result: Overly optimistic performance

The Solution: Nested WFO + Bayesian Lag

CORRECT: Bayesian-smoothed epoch from PRIOR folds for current TEST
   - Epoch selection on train/validation (inner loop)
   - Update Bayesian posterior with validation-optimal
   - Apply Bayesian-selected epoch to TEST (outer loop)
   - TEST data completely untouched during selection

v3 Temporal Ordering (CRITICAL - 2026 Fix)

The v3 implementation fixes a subtle but critical look-ahead bias bug in the original AWFES workflow. The key insight: TEST must use `prior_bayesian_epoch`, NOT `val_optimal_epoch`.

The Bug (v2 and earlier)

# v2 BUG: Bayesian update BEFORE test evaluation
for fold in folds:
    epoch_metrics = sweep_epochs(fold.train, fold.validation)
    val_optimal_epoch = select_optimal(epoch_metrics)

    # WRONG: Update Bayesian with current fold's val_optimal
    bayesian.update(val_optimal_epoch, wfe)
    selected_epoch = bayesian.get_current_epoch()  # CONTAMINATED!

    # This selected_epoch is influenced by val_optimal from SAME fold
    test_metrics = evaluate(selected_epoch, fold.test)  # LOOK-AHEAD BIAS

The Fix (v3)

# v3 CORRECT: Get prior epoch BEFORE any work on current fold
for fold in folds:
    # Step 1: FIRST - Get epoch from ONLY prior folds
    prior_bayesian_epoch = bayesian.get_current_epoch()  # BEFORE any fold work

    # Step 2: Train and sweep to find this fold's optimal
    epoch_metrics = sweep_epochs(fold.train, fold.validation)
    val_optimal_epoch = select_optimal(epoch_metrics)

    # Step 3: TEST uses prior_bayesian_epoch (NOT val_optimal!)
    test_metrics = evaluate(prior_bayesian_epoch, fold.test)  # UNBIASED

    # Step 4: AFTER test - update Bayesian for FUTURE folds only
    bayesian.update(val_optimal_epoch, wfe)  # For fold+1, fold+2, ...

Why This Matters

Validation Checkpoint

# MANDATORY: Log these values for audit trail
fold_log.info(
    f"Fold {fold_idx}: "
    f"prior_bayesian_epoch={prior_bayesian_epoch}, "
    f"val_optimal_epoch={val_optimal_epoch}, "
    f"test_uses={prior_bayesian_epoch}"  # MUST equal prior_bayesian_epoch
)

See look-ahead-bias.md for detailed examples.

Embargo Requirements

def compute_embargo_indices(
    n_total: int,
    train_pct: float = 0.60,
    val_pct: float = 0.20,
    test_pct: float = 0.20,
    embargo_pct: float = 0.06,
) -> dict[str, tuple[int, int]]:
    """Compute indices for nested split with embargoes.

    Returns dict with (start, end) tuples for each segment.
    """
    embargo_size = int(n_total * embargo_pct)

    train_end = int(n_total * train_pct)
    val_start = train_end + embargo_size
    val_end = val_start + int(n_total * val_pct)
    test_start = val_end + embargo_size
    test_end = n_total

    return {
        "train": (0, train_end),
        "embargo_1": (train_end, val_start),
        "validation": (val_start, val_end),
        "embargo_2": (val_end, test_start),
        "test": (test_start, test_end),
    }

Validation Checklist

Before running AWFES with OOS application:

• [ ] Three-way split: Train/Validation/Test clearly separated
• [ ] Embargoes: 6% gap at each boundary
• [ ] Bayesian lag: Current fold uses posterior from prior folds
• [ ] No peeking: Test data untouched until final evaluation
• [ ] Temporal order: No shuffling, strict time sequence
• [ ] Feature computation: Features computed BEFORE split, no recalculation

Anti-Patterns

Mathematical Formulation: Adaptive Walk-Forward Epoch Selection

1. Walk-Forward Efficiency (WFE)

Definition

WFE = SR_OOS / SR_IS

Where SR is the Sharpe Ratio:

SR = (μ - r_f) / σ

• μ = Mean return
• r_f = Risk-free rate (typically 0 for crypto)
• σ = Standard deviation of returns

Statistical Properties

Sharpe Ratio Distribution (Lo, 2002)

Under normality assumptions:

SR ~ N(SR*, √((1 + SR*²/2) / T))

Where:

• SR\* = True Sharpe ratio
• T = Number of observations

Standard Error of Sharpe Ratio

SE(SR) ≈ √((1 + 0.5 × SR²) / T)

WFE Variance (Delta Method)

For WFE as ratio of two correlated random variables:

Var(WFE) ≈ WFE² × [Var(SR_OOS)/SR_OOS² + Var(SR_IS)/SR_IS² - 2×Cov(SR_OOS, SR_IS)/(SR_OOS × SR_IS)]

Assuming independence between IS and OOS:

Var(WFE) ≈ (SR_OOS/SR_IS)² × [(1 + 0.5×SR_OOS²)/T_OOS + (1 + 0.5×SR_IS²)/T_IS]

Bias Characteristics

WFE is NOT unbiased.

1. Ratio Bias: E[X/Y] ≠ E[X]/E[Y] (Jensen's inequality) 2. Selection Bias: IS_Sharpe is inflated due to optimization 3. Net Direction: Typically downward bias (denominator inflated)

First-Order Bias Correction:

WFE_corrected ≈ WFE × (1 + Var(SR_IS) / SR_IS²)

2. WFE Aggregation Methods

Method 1: Pooled WFE

WFE_pooled = Σ(T_OOS_i × SR_OOS_i) / Σ(T_IS_i × SR_IS_i)

Properties:

• Weights by sample size (precision)
• More stable than arithmetic mean
• Handles varying fold sizes well

Method 2: Median WFE

WFE_median = median(WFE_1, WFE_2, ..., WFE_K)

Properties:

• Robust to outliers
• Breakdown point = 0.5
• Loses information from distribution tails

Method 3: Inverse-Variance Weighted Mean

WFE_weighted = Σ(w_i × WFE_i) / Σ(w_i)

where w_i = 1 / Var(WFE_i) ≈ T_OOS_i × T_IS_i / (T_OOS_i + T_IS_i)

Properties:

• Optimal efficiency under homoscedasticity
• Downweights noisy estimates

3. WFE Distribution Under Null (No Skill)

Under H₀: SR_true = 0, both SR_IS and SR_OOS are sampling noise:

SR_IS ~ N(0, 1/√T_IS)
SR_OOS ~ N(0, 1/√T_OOS)

WFE Distribution Under Null:

The ratio of two independent standard normals follows a Cauchy distribution:

WFE | H₀ ~ Cauchy(0, √(T_IS/T_OOS))

Critical Properties:

• No defined mean or variance
• Heavy tails (extreme values common)
• Makes arithmetic mean unreliable

4. Deflated Sharpe Ratio (DSR)

Formula

DSR = Φ[(SR - SR₀) × √(N-1) / √(1 + 0.5×SR² - γ₃×SR + (γ₄-3)/4×SR²)]

Where:

• Φ = Standard normal CDF
• SR₀ = Expected maximum Sharpe under null
• N = Sample size
• γ₃ = Skewness
• γ₄ = Kurtosis

Expected Maximum Under Null

For K independent trials (Bailey & López de Prado, 2014):

SR₀ = √(2 × ln(K)) × (1 - γ / √(2 × ln(K)) - ln(ln(K) + ln(4π)) / (2 × √(2 × ln(K))))

Where γ ≈ 0.5772 (Euler-Mascheroni constant).

Simplified approximation:

SR₀ ≈ √(2 × ln(K)) - (γ + ln(π/2)) / √(2 × ln(K))

Application to Epoch Selection

Total trials = K_epochs × F_folds

For 4 epochs × 31 folds = 124 trials:

import math

K = 124
gamma = 0.5772  # Euler-Mascheroni

sr0 = math.sqrt(2 * math.log(K))
sr0 -= (gamma + math.log(math.pi / 2)) / math.sqrt(2 * math.log(K))
sr0 *= 0.3  # Typical SE(SR)

# sr0 ≈ 0.75

5. Efficient Frontier Formulation

Pareto Dominance

Epoch A dominates Epoch B if:

WFE(A) ≥ WFE(B) AND Time(A) ≤ Time(B)

with at least one strict inequality.

Efficient Frontier Set

Frontier = {e ∈ Epochs : ∄ e' ∈ Epochs s.t. e' dominates e}

Selection from Frontier

Weighted Score Method:

Score(e) = w_wfe × norm(WFE(e)) + w_time × (1 - norm(Time(e)))

Where:

• norm(x) = (x - min) / (max - min) (min-max normalization)
• w_wfe = Weight for WFE (default: 1.0)
• w_time = Weight for time (default: 0.1)

Knee-Point Method:

Find epoch where marginal WFE gain per unit time decreases most sharply.

Knee = argmax_e |∂²WFE/∂Time²|

6. Stability Penalty Formulation

Penalty Function

AdjustedWFE(e_t) = WFE(e_t) - λ × I(e_t ≠ e_{t-1})

Where:

• λ = Stability penalty coefficient (default: 0.1 × WFE_mean)
• I(·) = Indicator function (1 if condition true, 0 otherwise)

Selection Rule

e_t* = argmax_e [WFE(e) - λ × I(e ≠ e_{t-1}*)]

Only change epochs if improvement exceeds penalty threshold.

Alternative: Bayesian Shrinkage

e_t* = α × argmax_e WFE(e) + (1-α) × e_{t-1}*

With α ∈ [0, 1] controlling adaptation speed.

7. Effective Sample Size (N_eff)

Reduction from Epoch Selection

N_eff = N_samples × selection_factor × correlation_factor

Where:

• selection_factor = 1 / √K_epochs
• correlation_factor = (1 - ρ) / (1 + ρ) (Kish's formula)
• ρ = Autocorrelation from carry-forward

Example Calculation

For 31 folds, 4 epochs, autocorrelation 0.3:

n_samples = 31
n_epochs = 4
autocorr = 0.3

selection_factor = 1 / math.sqrt(n_epochs)  # 0.5
correlation_factor = (1 - autocorr) / (1 + autocorr)  # 0.54

n_eff = n_samples * selection_factor * correlation_factor
# n_eff ≈ 8.4

31 folds provide ~8 effective independent observations.

8. Minimum Sample Size Requirements

For Reliable WFE

For SE(WFE) < target precision ε:

T_OOS ≥ (1 + 0.5×SR²) / (ε/WFE)² - T_IS×(1 + 0.5×SR²) / T_IS

Practical Minimums (20% precision)

Rule of Thumb

• Minimum: T_OOS ≥ 63 trading days (1 quarter)
• Recommended: T_OOS ≥ 126 trading days (6 months)
• Robust: T_OOS ≥ 252 trading days (1 year)

9. Confidence Intervals for WFE

Fieller's Method (Exact)

For WFE = SR_OOS / SR_IS:

CI = [WFE × (1 - z_α × CV_IS²) ± z_α × SE_ratio] / (1 - z_α² × CV_IS²)

Where:

• CV_IS = SE(SR_IS) / SR_IS
• SE_ratio = WFE × √(CV_OOS² + CV_IS² - 2×ρ×CV_OOS×CV_IS)

Bootstrap Method (Recommended)

def bootstrap_wfe_ci(
    returns_is,
    returns_oos,
    n_bootstrap=10000,
    alpha=0.05,
    annualization_factor=None,  # Use AWFESConfig.get_annualization_factor()
    is_threshold=None,          # Use compute_is_sharpe_threshold()
):
    """Bootstrap confidence interval for WFE.

    Args:
        annualization_factor: sqrt(periods_per_year). Use:
            - sqrt(365) for crypto_24_7 daily
            - sqrt(252) for equity/session-filtered daily
            - Or get from AWFESConfig.get_annualization_factor()
        is_threshold: Minimum IS Sharpe. Use compute_is_sharpe_threshold(n).
    """
    # Default to equity convention if not specified
    ann_factor = annualization_factor or np.sqrt(252)
    min_is = is_threshold or 0.1

    wfe_samples = []
    for _ in range(n_bootstrap):
        is_sample = np.random.choice(returns_is, size=len(returns_is), replace=True)
        oos_sample = np.random.choice(returns_oos, size=len(returns_oos), replace=True)

        sr_is = is_sample.mean() / is_sample.std() * ann_factor
        sr_oos = oos_sample.mean() / oos_sample.std() * ann_factor

        if sr_is > min_is:
            wfe_samples.append(sr_oos / sr_is)

    return np.percentile(wfe_samples, [100*alpha/2, 100*(1-alpha/2)])

10. Summary: Key Formulas

Skill: Adaptive WFO Epoch Selection

Troubleshooting