Opendeviation Eval Metrics

Open Deviation Bar Evaluation Metrics

Machine-readable reference + computation scripts for state-of-the-art metrics evaluating open deviation bar (ODB, brim-to-brim price-based sampling) data.

Cross-reference: Project-level experiment catalogue at signal-archaeology skill in opendeviationbar-patterns repo — contains 10 BHR-validated experiments with auditable SQL.

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:

• Evaluating ML model performance on open deviation bar data
• Computing Sharpe ratios with non-IID bar sequences
• Running Walk-Forward Optimization metric analysis
• Calculating PSR, DSR, or MinTRL statistical tests
• Generating evaluation reports from fold results

Quick Start

# Compute metrics from predictions + actuals
python scripts/compute_metrics.py --predictions preds.npy --actuals actuals.npy --timestamps ts.npy

# Generate full evaluation report
python scripts/generate_report.py --results folds.jsonl --output report.md

Metric Tiers

Why Open Deviation Bars Need Special Treatment

Open deviation bars violate standard IID assumptions:

1. Variable duration: Bars form based on price movement, not time 2. Autocorrelation: High-volatility periods cluster bars → temporal correlation 3. Non-constant information: More bars during volatility = more information per day

Canonical solution: Daily aggregation via _group_by_day() before Sharpe calculation.

References

Core Reference Files

Related Skills

Dependencies

pip install -r requirements.txt
# Or: pip install numpy>=1.24 pandas>=2.0 scipy>=1.10

Key Formulas

Daily-Aggregated Sharpe (Primary Metric)

def weekly_sharpe(pnl: np.ndarray, timestamps: np.ndarray) -> float:
    """Sharpe with daily aggregation for open deviation bars."""
    daily_pnl = _group_by_day(pnl, timestamps)  # Sum PnL per calendar day
    if len(daily_pnl) < 2 or np.std(daily_pnl) == 0:
        return 0.0
    daily_sharpe = np.mean(daily_pnl) / np.std(daily_pnl)
    # For crypto (7-day week): sqrt(7). For equities: sqrt(5)
    return daily_sharpe * np.sqrt(7)  # Crypto default

Information Coefficient (Prediction Quality)

from scipy.stats import spearmanr

def information_coefficient(predictions: np.ndarray, actuals: np.ndarray) -> float:
    """Spearman rank IC - captures magnitude alignment."""
    ic, _ = spearmanr(predictions, actuals)
    return ic  # Range: [-1, 1]. >0.02 acceptable, >0.05 good, >0.10 excellent

Probabilistic Sharpe Ratio (Statistical Validation)

from scipy.stats import norm

def psr(sharpe: float, se: float, benchmark: float = 0.0) -> float:
    """P(true Sharpe > benchmark)."""
    return norm.cdf((sharpe - benchmark) / se)

Annualization Factors

NEVER use sqrt(252) for crypto markets.

CRITICAL: Session Filter Changes Annualization

Using sqrt(7) for session-filtered data overstates Sharpe by ~18%!

See crypto-markets.md for detailed rationale.

Dual-View Metrics

For comprehensive analysis, compute metrics with BOTH views:

1. Session-filtered (London 08:00 to NY 16:00): Primary strategy evaluation 2. All-bars: Regime detection, data quality diagnostics

Academic References

Beyond Hit Rate (BHR) Framework

Hit rate is a necessary but insufficient metric. Always supplement with outcome predictability metrics. See beyond-hit-rate.md for the full framework.

Minimum Viable Signal Evaluation

Every signal MUST be evaluated with at least:

1. One sequence structure test: entropy, LZC, or runs test on the W/L sequence 2. One temporal decay test: CUSUM on equity curve or rolling hit rate 3. One regime awareness test: per-session hit rate or HMM decomposition

A signal that passes all three is robust. A signal with only high hit rate is noise.

Outcome Predictability Index (OPI)

OPI = 0.25 * (1 - LZC_norm) + 0.25 * |z_runs| + 0.25 * Var(HR_per_regime) + 0.25 * AUC_meta

Higher OPI = more predictable win/loss timing. A 45% HR signal with OPI=0.8 is more valuable than a 65% HR signal with OPI=0.1.

Decision Framework

Go Criteria (Research)

go_criteria:
  - positive_sharpe_rate > 0.55
  - mean_weekly_sharpe > 0
  - cv_fold_returns < 1.5
  - bhr_sequence_test_passes: true # At least one of: entropy, LZC, runs test significant
  - bhr_cusum_verdict: "ALIVE" # No recent regime break

Publication Criteria

publication_criteria:
  - binomial_pvalue < 0.05
  - psr > 0.85
  - dsr > 0.50 # If n_trials > 1
  - bhr_lzc_shuffle_z < -2.0 # W/L sequence has genuine structure
  - bhr_alpha_halflife > 200 # Edge persists for 200+ trades

Scripts

Remediations (2026-01-19 Multi-Agent Audit)

The following fixes were applied based on a 12-subagent adversarial audit:

Model Collapse Detection

# ALWAYS check for model collapse after prediction
pred_std = np.std(predictions)
if pred_std < 1e-6:
    logger.warning(
        f"Constant predictions detected (std={pred_std:.2e}). "
        "Model collapsed to mean - check architecture."
    )

Recommended BiLSTM Architecture

# BEFORE (causes collapse on open deviation bars)
HIDDEN_SIZE = 16
DROPOUT = 0.5

# AFTER (prevents collapse)
HIDDEN_SIZE = 48  # Triple capacity
DROPOUT = 0.3     # Less aggressive regularization

See reference docs for complete implementation details.

---

Troubleshooting

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.

Anti-Patterns in Range Bar Metrics

Common pitfalls and their remediation when computing metrics on range bar data.

Transaction Costs (CRITICAL)

The Problem

Zero-cost assumption in Sharpe calculation overstates performance by 15-30% for typical high-frequency strategies.

# WRONG: Zero-cost assumption (common in backtesting)
pnl = predictions * actuals  # No transaction costs
sharpe = mean(pnl) / std(pnl)  # Overstated by 15-30%

Impact by Strategy Type

Cost Model

def compute_transaction_costs(
    predictions: np.ndarray,
    prices: np.ndarray,
    cost_bps: float = 5.0,  # 5 bps = 0.05%
) -> np.ndarray:
    """Compute transaction costs from position changes.

    Args:
        predictions: Signed position sizes
        prices: Asset prices at each bar
        cost_bps: Round-trip cost in basis points

    Returns:
        Array of transaction costs (negative values)
    """
    position_changes = np.abs(np.diff(predictions, prepend=0))
    notional_traded = position_changes * prices
    return -notional_traded * (cost_bps / 10000)


def net_sharpe_with_costs(
    predictions: np.ndarray,
    actuals: np.ndarray,
    timestamps: np.ndarray,
    prices: np.ndarray,
    cost_bps: float = 5.0,
    days_per_week: int = 7,
) -> float:
    """Weekly Sharpe net of transaction costs."""
    gross_pnl = predictions * actuals
    tx_costs = compute_transaction_costs(predictions, prices, cost_bps)
    net_pnl = gross_pnl + tx_costs  # tx_costs are negative

    daily_pnl = _group_by_day(net_pnl, timestamps)
    if len(daily_pnl) < 2 or np.std(daily_pnl) < 1e-10:
        return 0.0

    return float(np.mean(daily_pnl) / np.std(daily_pnl) * np.sqrt(days_per_week))

Typical Cost Assumptions

Remediation

1. Always report both gross and net Sharpe 2. Use conservative cost assumptions (upper bound of range) 3. Log turnover alongside Sharpe for cost sensitivity analysis

def evaluate_fold_with_costs(
    predictions: np.ndarray,
    actuals: np.ndarray,
    timestamps: np.ndarray,
    prices: np.ndarray,
    cost_bps: float = 5.0,
) -> dict:
    """Full evaluation with cost transparency."""
    gross_sharpe = compute_weekly_sharpe(predictions * actuals, timestamps)
    net_sharpe = net_sharpe_with_costs(
        predictions, actuals, timestamps, prices, cost_bps
    )

    # Turnover metrics
    position_changes = np.abs(np.diff(predictions, prepend=0))
    daily_turnover = _group_by_day(position_changes, timestamps)
    monthly_turnover = np.mean(daily_turnover) * 30

    return {
        "gross_weekly_sharpe": gross_sharpe,
        "net_weekly_sharpe": net_sharpe,
        "sharpe_cost_haircut": (gross_sharpe - net_sharpe) / gross_sharpe if gross_sharpe > 0 else 0,
        "monthly_turnover_pct": float(monthly_turnover * 100),
        "cost_bps_assumed": cost_bps,
    }

Survivorship Bias

The Problem

Backtesting on currently traded assets excludes delisted/failed assets, biasing results upward.

Crypto-specific: Many altcoins have been delisted, depegged, or gone to zero (e.g., LUNA, FTT).

Impact

• Equity studies show 1-2% annual return bias
• Crypto markets: 3-5% bias due to higher failure rates

Remediation

1. Use point-in-time datasets (e.g., Kaiko, CryptoDataDownload historical) 2. Include delisted assets in backtests 3. Report "survivorship-adjusted" returns

Look-Ahead Bias

The Problem

Using future information in feature computation or model training.

Common sources:

• Normalizing features with full-sample statistics
• Using daily OHLC when predicting intraday
• Leaking test data into training via feature engineering

Detection

def detect_lookahead_bias(
    predictions: np.ndarray,
    timestamps: np.ndarray,
    model_training_end: pd.Timestamp,
) -> dict:
    """Check for temporal consistency."""
    pred_times = pd.to_datetime(timestamps, utc=True)

    # Predictions before training end are suspicious
    early_predictions = pred_times < model_training_end
    n_early = np.sum(early_predictions)

    return {
        "n_predictions_before_training_end": int(n_early),
        "potential_lookahead_bias": n_early > 0,
    }

Remediation

See adaptive-wfo-epoch/references/look-ahead-bias.md for detailed prevention strategies.

sqrt(252) for Crypto

The Problem

Using equity annualization (sqrt(252)) for 24/7 crypto markets understates volatility.

# WRONG
crypto_annual_sharpe = daily_sharpe * np.sqrt(252)  # Wrong!

# CORRECT
crypto_annual_sharpe = daily_sharpe * np.sqrt(365)  # 24/7 markets
crypto_weekly_sharpe = daily_sharpe * np.sqrt(7)    # Weekly

Impact

Using sqrt(252) instead of sqrt(365) underestimates true Sharpe by:

• Daily → Annual: 17% understatement
• Daily → Weekly: 15% understatement (sqrt(5) vs sqrt(7))

See crypto-markets.md for full annualization guide.

Model Collapse

The Problem

BiLSTM and other RNN models can collapse to constant predictions (output = mean of targets).

Symptoms:

• prediction_autocorr ≈ 1.0
• std(predictions) < 1e-6
• weekly_sharpe = 0 despite positive historical returns

Causes

1. Undersized hidden layer: 16 units too small for range bar complexity 2. Excessive dropout: 0.5 dropout removes too much signal 3. Learning rate too high: Model overshoots and converges to mean 4. Insufficient data: Model can't learn meaningful patterns

Remediation

# BEFORE (causes collapse on range bars)
HIDDEN_SIZE = 16
DROPOUT = 0.5

# AFTER (prevents collapse)
HIDDEN_SIZE = 48  # Triple capacity
DROPOUT = 0.3     # Less aggressive regularization

See ml-prediction-quality.md for detection code.

Insufficient Sample Size

The Problem

Computing ratios (Omega, Profit Factor, Sharpe SE) with too few observations produces unreliable estimates.

Minimum Sample Sizes

Remediation

All functions in compute_metrics.py now include min_days guards:

def compute_omega(pnl, timestamps, threshold=0.0, min_days=5):
    daily_pnl = _group_by_day(pnl, timestamps)
    if len(daily_pnl) < min_days:
        return float("nan")  # Unreliable
    # ... rest of computation

Summary Checklist

Before reporting range bar metrics:

• [ ] Transaction costs modeled (gross AND net Sharpe)
• [ ] Survivorship bias considered (point-in-time data)
• [ ] Look-ahead bias prevented (temporal validation)
• [ ] Correct annualization (sqrt(7) or sqrt(365) for crypto)
• [ ] Model collapse checked (prediction_autocorr, is_collapsed)
• [ ] Sufficient sample size (n_days > min thresholds)
• [ ] Session filter matches annualization (sqrt(5) if filtered)

References

Beyond Hit Rate: Outcome Predictability Framework

Hit rate is among the worst evaluation metrics. A strategy with 80% historical win rate but all wins concentrated in 2022 and losses in 2025 is worthless. The right question is not "how often does it win?" but "can we predict WHEN it wins?"

Empirically validated on 866K ODB bars (BTCUSDT 250dbps, 2018-2026): hit-rate ranking was inversely correlated with signal robustness. The highest hit-rate signal (64% mean-reversion) was dead by 2025. The lowest (38% raw displacement) was the most robust and strengthening.

The Three Failure Modes of Hit Rate

Metric Stack (Replaces Hit Rate)

Implementation: src/opendeviationbar_patterns/eval/bhr.py (BHR metrics) + eval/orthogonal.py (orthogonal metrics). Call bhr_report(outcomes) for full diagnostic with auto-verdict, orthogonal_report(returns, timestamps) for complementary dimensions.

Tier 1: CRITICAL — sequence structure tests

CRITICAL: Always compare entropy/LZC against a shuffle test (permute the W/L sequence 1000 times, compute z-score). Without base-rate correction, low-frequency events (11% hit rate) falsely appear structured because they mechanically produce low entropy.

Tier 2: HIGH — temporal decay detection

| Metric | Formula | Detects | Library | | ------------------------------------ | -------------------------------------------------- | ---------------------------------------------- | -------------------------------- | ------------------------ | ------------- | | CUSUM on equity curve | S_t = max(0, S_{t-1} + (x_t - μ₀ - k)) | Exact changepoint where strategy stops working | ruptures (PELT) | | HHI across time buckets | Σ(w_i/W_total)² per quarter/year | Temporal concentration of edge | numpy | | KL Divergence period-over-period | D_KL(P‖Q) between quarterly return distributions | Distributional shift / regime change | scipy.stats.entropy | | Alpha half-life | t_half = ln(2)/ | φ | from AR(1) on rolling hit rate | How fast the edge erodes | statsmodels |

Tier 3: META-LABELING — outcome prediction

Composite: Outcome Predictability Index (OPI)

OPI = 0.25 * (1 - LZC_norm)           # Sequence compressibility
    + 0.25 * |z_runs|                   # Departure from randomness
    + 0.25 * Var(HR_per_regime)         # Regime-conditional spread
    + 0.25 * AUC_meta                   # Meta-model discriminative power

OPI = 0 means outcomes are completely random. Higher = more predictable timing = more valuable even at lower hit rate.

Anti-Pattern: Hit Rate as Primary Metric

NEVER use hit rate alone. Always pair with at least:

1. One sequence structure metric (entropy, LZC, or runs test) 2. One temporal decay metric (CUSUM, HHI, or half-life) 3. One regime awareness metric (per-session HR, HMM, or KL divergence)

Empirical Validation (ODB 866K bars)

The highest hit-rate signal is dead. The lowest is strongest. This is the canonical example of why hit rate fails.

Orthogonal Metrics (complement BHR)

BHR operates on binary W/L only. These metrics bring in dimensions BHR cannot see:

Key empirical finding: CECP confirms returns are NOISE (H≈1.0) — the edge is purely in W/L direction. Burstiness (0.35-0.68) and Memory (0.25-0.50) reveal all signals fire in REGIME BURSTS, not uniformly.

Key References

Crypto Market Considerations

Annualization: sqrt(7) and sqrt(365)

CRITICAL: Crypto trades 24/7, 365 days/year.

# CORRECT for crypto
DAYS_PER_WEEK_CRYPTO = 7
DAYS_PER_YEAR_CRYPTO = 365

weekly_sharpe = daily_sharpe * np.sqrt(DAYS_PER_WEEK_CRYPTO)  # sqrt(7) = 2.65
annual_sharpe = daily_sharpe * np.sqrt(DAYS_PER_YEAR_CRYPTO)  # sqrt(365) = 19.1

# WRONG for crypto (equity assumptions)
# weekly_sharpe = daily_sharpe * np.sqrt(5)  # DON'T DO THIS
# annual_sharpe = daily_sharpe * np.sqrt(252)  # DON'T DO THIS

Session Definitions (UTC)

sessions:
  asia:
    start: "00:00"
    end: "08:00"
    characteristics: "Lower volume, mean-reversion tendency"

  europe:
    start: "08:00"
    end: "16:00"
    characteristics: "Increasing volume, momentum starts"

  americas:
    start: "14:00"
    end: "22:00"
    characteristics: "Highest volume, trend continuation"

  london_ny_overlap:
    start: "14:00"
    end: "16:00"
    characteristics: "Peak liquidity, best for execution"

  overnight:
    start: "22:00"
    end: "08:00"
    characteristics: "Lowest liquidity, higher spreads"

Session Filter Implementation

CRITICAL: Use `zoneinfo.ZoneInfo` for DST-aware timezone handling.

Python's zoneinfo module (stdlib since 3.9) uses the IANA timezone database and automatically handles Daylight Saving Time transitions for both London (GMT/BST) and New York (EST/EDT).

DST Transition Handling

London and New York have different DST transition dates:

• UK: Last Sunday of March (forward), Last Sunday of October (back)
• US: 2nd Sunday of March (forward), 1st Sunday of November (back)

This creates 2-3 week gaps where one region is in DST and the other isn't. ZoneInfo handles this automatically:

from datetime import datetime, time
from zoneinfo import ZoneInfo
import pandas as pd

# CORRECT: Use ZoneInfo for DST-aware handling
LONDON_TZ = ZoneInfo("Europe/London")   # GMT (winter) / BST (summer)
NY_TZ = ZoneInfo("America/New_York")    # EST (winter) / EDT (summer)

LONDON_OPEN = time(8, 0)   # 8:00 AM London local
NY_CLOSE = time(16, 0)     # 4:00 PM NY local


def get_session_bounds_utc(date) -> tuple[pd.Timestamp, pd.Timestamp]:
    """Get London open to NY close in UTC for a given date.

    DST is handled automatically by ZoneInfo.
    """
    london_open = pd.Timestamp(
        datetime.combine(date, LONDON_OPEN), tz=LONDON_TZ
    ).tz_convert("UTC")

    ny_close = pd.Timestamp(
        datetime.combine(date, NY_CLOSE), tz=NY_TZ
    ).tz_convert("UTC")

    return london_open, ny_close


def is_tradeable_bar(bar_close_ts: pd.Timestamp) -> bool:
    """Check if bar falls within London-NY session.

    Uses bar close timestamp for session membership.
    """
    if bar_close_ts.tzinfo is None:
        bar_close_ts = bar_close_ts.tz_localize("UTC")

    ts_london = bar_close_ts.tz_convert(LONDON_TZ)
    weekday = ts_london.weekday()

    # Skip weekends
    if weekday >= 5:
        return False

    session_open, session_close = get_session_bounds_utc(ts_london.date())
    return session_open <= bar_close_ts <= session_close


def compute_tradeable_mask(timestamps: np.ndarray) -> np.ndarray:
    """Boolean mask for tradeable bars."""
    return np.array([
        is_tradeable_bar(pd.Timestamp(ts))
        for ts in timestamps
    ])

Anti-Patterns (DO NOT USE)

# WRONG: Using fixed UTC offsets (ignores DST)
# london_open_utc = datetime(..., hour=8) - timedelta(hours=0)  # WRONG!

# WRONG: Using pytz without localize() (deprecated)
# import pytz
# tz = pytz.timezone("Europe/London")
# ts = datetime(..., tzinfo=tz)  # WRONG! Use tz.localize() instead

# WRONG: Hardcoded session hours in UTC
# SESSION_START_UTC = 8  # WRONG! Varies with DST
# SESSION_END_UTC = 21   # WRONG! Varies with DST

Weekend/Weekday Split

Research shows distinct characteristics:

def compute_weekend_weekday_split(
    pnl: np.ndarray,
    timestamps: np.ndarray
) -> dict:
    """Separate metrics for weekends vs weekdays.

    Empirical findings (Bitcoin 2014-2024):
    - Weekend volume: 60-70% of weekday
    - Weekend volatility: Lower
    - Weekend momentum: Higher returns (Monday effect)
    """
    df = pd.DataFrame({
        "pnl": pnl,
        "ts": pd.to_datetime(timestamps, utc=True)
    })
    df["is_weekend"] = df["ts"].dt.dayofweek >= 5

    weekday_pnl = df[~df["is_weekend"]]["pnl"].values
    weekend_pnl = df[df["is_weekend"]]["pnl"].values

    def safe_sharpe(arr):
        if len(arr) < 2 or np.std(arr) < 1e-10:
            return 0.0
        return float(np.mean(arr) / np.std(arr))

    return {
        "sharpe_weekday": safe_sharpe(weekday_pnl),
        "sharpe_weekend": safe_sharpe(weekend_pnl),
        "n_weekday_bars": len(weekday_pnl),
        "n_weekend_bars": len(weekend_pnl),
        "pnl_frac_weekend": (
            weekend_pnl.sum() / (weekday_pnl.sum() + weekend_pnl.sum())
            if (weekday_pnl.sum() + weekend_pnl.sum()) != 0 else 0.0
        )
    }

Funding Rate Exposure

For perpetual futures strategies:

def estimate_funding_impact(
    positions: np.ndarray,
    timestamps: np.ndarray,
    avg_funding_rate_8h: float = 0.0001  # 1 bp per 8h
) -> float:
    """Estimate funding cost/income for perpetuals.

    Funding settles every 8 hours (00:00, 08:00, 16:00 UTC).
    Long pays short when rate > 0.

    Args:
        positions: Position sizes (positive = long)
        timestamps: Position timestamps
        avg_funding_rate_8h: Average 8h funding rate (positive = longs pay)

    Returns:
        Total funding PnL (negative = cost)
    """
    df = pd.DataFrame({
        "position": positions,
        "ts": pd.to_datetime(timestamps, utc=True)
    })

    # Funding times
    df["hour"] = df["ts"].dt.hour
    df["is_funding"] = df["hour"].isin([0, 8, 16])

    # Funding impact: -position * rate (longs pay when rate > 0)
    funding_events = df[df["is_funding"]]
    total_funding = -(funding_events["position"] * avg_funding_rate_8h).sum()

    return float(total_funding)

UTC Day Boundaries

def group_by_utc_day(
    pnl: np.ndarray,
    timestamps: np.ndarray
) -> pd.DataFrame:
    """Group by UTC calendar day.

    CRITICAL: Always use UTC for crypto aggregation.
    """
    df = pd.DataFrame({
        "pnl": pnl,
        "ts": pd.to_datetime(timestamps, utc=True)  # Explicit UTC
    })
    df["date"] = df["ts"].dt.date

    return df.groupby("date").agg({
        "pnl": "sum",
        "ts": "count"  # Bar count per day
    }).rename(columns={"ts": "n_bars"})

Dual-View Evaluation

For comprehensive analysis:

dual_view:
  session_filtered:
    purpose: "Strategy performance evaluation"
    filter: "London 08:00 to NY 16:00, weekdays only"
    annualization: "sqrt(5) - 5 trading days per week"
    use_for:
      - "Primary Sharpe calculation"
      - "Risk metrics"
      - "Go/no-go decisions"

  all_bars:
    purpose: "Regime detection and data quality"
    filter: "None (all bars)"
    annualization: "sqrt(7) - crypto trades 24/7"
    use_for:
      - "Bar count stability diagnostic"
      - "Weekend/weekday comparison"
      - "Volatility regime detection"

CRITICAL: Session-Specific Annualization

THIS IS THE MOST IMPORTANT DISTINCTION FOR CRYPTO RANGE BARS.

# CORRECT dual-view implementation
def compute_dual_view_metrics(
    predictions: np.ndarray,
    actuals: np.ndarray,
    timestamps: np.ndarray
) -> dict:
    """Compute metrics with CORRECT annualization for each view."""

    # All bars view - sqrt(7) because crypto is 24/7
    all_bars = evaluate_fold(
        predictions, actuals, None, timestamps,
        days_per_week=7  # CRITICAL: 7 for all-bars
    )

    # Session-filtered view - sqrt(5) because we filter to 5 trading days
    mask = compute_tradeable_mask(timestamps)
    filtered = evaluate_fold(
        predictions, actuals, mask, timestamps,
        days_per_week=5  # CRITICAL: 5 for session-filtered
    )

    return {
        "oos_metrics": filtered,      # Primary (sqrt(5))
        "oos_metrics_all": all_bars   # Diagnostic (sqrt(7))
    }


# WRONG - using same annualization for both views
# filtered = evaluate_fold(..., days_per_week=7)  # INCORRECT!

Why This Matters

1. Session-filtered uses sqrt(5): When you filter to London-NY weekday hours, you're effectively trading a 5-day week like equities. The variance scaling must match.

2. All-bars uses sqrt(7): The full 24/7 crypto dataset has 7 trading days worth of data per week, so annualization must use sqrt(7).

3. Mixing them is a methodological error: Using sqrt(7) for session-filtered overstates the Sharpe ratio by ~18% (sqrt(7)/sqrt(5) = 1.183).

# Example of the overstatement
daily_sharpe = 0.1

correct_filtered = daily_sharpe * np.sqrt(5)   # 0.224
incorrect_filtered = daily_sharpe * np.sqrt(7)  # 0.265

overstatement = incorrect_filtered / correct_filtered  # 1.183 = 18.3% overstatement!

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

---

Metrics JSON Schema

JSON Schema for validating range bar evaluation metrics output.

Schema Version

schema_version: "1.0.0"
compatible_with: "rangebar-eval-metrics@9.37+"

Full Schema

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://github.com/terrylica/cc-skills/rangebar-eval-metrics/v1",
  "title": "Range Bar Evaluation Metrics",
  "description": "Output schema for compute_metrics.py",
  "type": "object",
  "properties": {
    "weekly_sharpe": {
      "type": "number",
      "description": "Daily-aggregated Sharpe scaled to weekly (sqrt(7) for crypto, sqrt(5) for equity)"
    },
    "hit_rate": {
      "type": "number",
      "minimum": 0,
      "maximum": 1,
      "description": "Directional accuracy (proportion of correct sign predictions)"
    },
    "cumulative_pnl": {
      "type": "number",
      "description": "Total PnL over evaluation period"
    },
    "n_bars": {
      "type": "integer",
      "minimum": 0,
      "description": "Number of range bars in evaluation"
    },
    "positive_sharpe_rate": {
      "type": "number",
      "minimum": 0,
      "maximum": 1,
      "description": "Proportion of folds with positive Sharpe (aggregate only)"
    },
    "max_drawdown": {
      "type": "number",
      "maximum": 0,
      "description": "Maximum drawdown (negative value)"
    },
    "bar_sharpe": {
      "type": "number",
      "description": "Raw bar-level Sharpe (NOT daily-aggregated, for comparison only)"
    },
    "return_per_bar": {
      "type": "number",
      "description": "Average return per bar"
    },
    "profit_factor": {
      "type": "number",
      "minimum": 0,
      "description": "Gross profit / gross loss (Inf if no losses)"
    },
    "cv_fold_returns": {
      "type": "number",
      "minimum": 0,
      "description": "Coefficient of variation across fold returns"
    },
    "ic": {
      "type": ["number", "null"],
      "minimum": -1,
      "maximum": 1,
      "description": "Information Coefficient (Spearman rank correlation)"
    },
    "prediction_autocorr": {
      "type": ["number", "null"],
      "minimum": -1,
      "maximum": 1,
      "description": "Lag-1 autocorrelation of predictions (detects sticky LSTM)"
    },
    "sharpe_se": {
      "type": ["number", "null"],
      "minimum": 0,
      "description": "Standard error of Sharpe (Mertens 2002)"
    },
    "psr": {
      "type": ["number", "null"],
      "minimum": 0,
      "maximum": 1,
      "description": "Probabilistic Sharpe Ratio (Bailey & López de Prado 2012)"
    },
    "dsr": {
      "type": ["number", "null"],
      "minimum": 0,
      "maximum": 1,
      "description": "Deflated Sharpe Ratio (Bailey & López de Prado 2014)"
    },
    "skewness": {
      "type": "number",
      "description": "Skewness of daily returns"
    },
    "kurtosis": {
      "type": "number",
      "description": "Kurtosis of daily returns (Pearson form, normal = 3)"
    },
    "binomial_pvalue": {
      "type": "number",
      "minimum": 0,
      "maximum": 1,
      "description": "P-value for sign test (n_positive vs n_total)"
    },
    "autocorr_lag1": {
      "type": "number",
      "minimum": -1,
      "maximum": 1,
      "description": "Lag-1 autocorrelation of fold Sharpes (aggregate only)"
    },
    "effective_n": {
      "type": "number",
      "minimum": 0,
      "description": "Autocorrelation-adjusted sample size (aggregate only)"
    },
    "var_95": {
      "type": "number",
      "description": "Value at Risk at 95% confidence (daily, negative value)"
    },
    "cvar_95": {
      "type": "number",
      "description": "Conditional VaR (Expected Shortfall) at 95%"
    },
    "omega_ratio": {
      "type": ["number", "null"],
      "minimum": 0,
      "description": "Omega ratio (gains/losses above threshold)"
    },
    "sortino_ratio": {
      "type": ["number", "null"],
      "description": "Sortino ratio (downside deviation only)"
    },
    "ulcer_index": {
      "type": "number",
      "minimum": 0,
      "description": "Ulcer Index (RMS of percentage drawdowns)"
    },
    "calmar_ratio": {
      "type": ["number", "null"],
      "description": "Calmar ratio (annual return / max drawdown)"
    },
    "error": {
      "type": "string",
      "description": "Error message if computation failed"
    }
  },
  "oneOf": [
    {
      "required": ["weekly_sharpe", "hit_rate", "cumulative_pnl", "n_bars"],
      "not": { "required": ["error"] }
    },
    {
      "required": ["error"]
    }
  ]
}

Tier Mappings

Primary Metrics (Tier 1)

Secondary Metrics (Tier 2)

Diagnostic Metrics (Tier 3)

Risk Metrics (Extended)

Validation Script

#!/usr/bin/env python3
"""Validate metrics JSON against schema."""

import json
import sys
from pathlib import Path

try:
    from jsonschema import validate, ValidationError
except ImportError:
    print("Install jsonschema: pip install jsonschema")
    sys.exit(1)

SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "properties": {
        "weekly_sharpe": {"type": "number"},
        "hit_rate": {"type": "number", "minimum": 0, "maximum": 1},
        "cumulative_pnl": {"type": "number"},
        "n_bars": {"type": "integer", "minimum": 0},
    },
    "oneOf": [
        {
            "required": ["weekly_sharpe", "hit_rate", "cumulative_pnl", "n_bars"],
            "not": {"required": ["error"]}
        },
        {"required": ["error"]}
    ]
}


def validate_metrics(metrics_path: Path) -> bool:
    """Validate metrics file against schema."""
    with open(metrics_path) as f:
        metrics = json.load(f)

    try:
        validate(instance=metrics, schema=SCHEMA)
        print(f"✓ {metrics_path}: Valid")
        return True
    except ValidationError as e:
        print(f"✗ {metrics_path}: {e.message}")
        return False


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python validate_schema.py <metrics.json>")
        sys.exit(1)

    success = validate_metrics(Path(sys.argv[1]))
    sys.exit(0 if success else 1)

Example Valid Output

Single Fold

{
  "weekly_sharpe": 1.23,
  "hit_rate": 0.54,
  "cumulative_pnl": 0.0456,
  "n_bars": 1247,
  "max_drawdown": -0.0234,
  "bar_sharpe": 0.89,
  "profit_factor": 1.34,
  "ic": 0.067,
  "prediction_autocorr": 0.45,
  "sharpe_se": 0.31,
  "psr": 0.91,
  "dsr": 0.62,
  "skewness": -0.23,
  "kurtosis": 4.12
}

Aggregate

{
  "mean_weekly_sharpe": 0.87,
  "std_weekly_sharpe": 0.45,
  "median_weekly_sharpe": 0.92,
  "positive_sharpe_rate": 0.68,
  "n_folds": 31,
  "binomial_pvalue": 0.023,
  "autocorr_lag1": 0.12,
  "effective_n": 27.4
}

Error Case

{
  "error": "no_data"
}

Usage with compute_metrics.py

# Compute and validate
python scripts/compute_metrics.py \
  --predictions preds.npy \
  --actuals actuals.npy \
  --timestamps ts.npy \
  --output metrics.json

python scripts/validate_schema.py metrics.json

Semantic Versioning

References

• JSON Schema Specification
• rangebar-eval-metrics SKILL.md

ML Prediction Quality Metrics

Information Coefficient (IC)

The standard metric for alpha model quality:

from scipy.stats import spearmanr, pearsonr

def compute_ic(
    predictions: np.ndarray,
    actuals: np.ndarray,
    method: str = "spearman"
) -> float:
    """Information Coefficient - prediction/return correlation.

    Args:
        predictions: Model predictions
        actuals: Actual returns
        method: "spearman" (rank, preferred) or "pearson" (linear)

    Returns:
        IC in [-1, 1]. Thresholds:
        - >0.02: Acceptable
        - >0.05: Good
        - >0.10: Excellent
    """
    if len(predictions) < 10:
        return float("nan")

    if method == "spearman":
        ic, _ = spearmanr(predictions, actuals)
    else:
        ic, _ = pearsonr(predictions, actuals)

    return float(ic)


def compute_icir(
    predictions: np.ndarray,
    actuals: np.ndarray,
    timestamps: np.ndarray,
    window: int = 20  # Rolling window
) -> float:
    """IC Information Ratio - IC stability over time.

    ICIR = mean(IC) / std(IC) across rolling windows

    Higher ICIR = more consistent signal quality.
    """
    df = pd.DataFrame({
        "pred": predictions,
        "actual": actuals,
        "ts": pd.to_datetime(timestamps, utc=True)
    }).sort_values("ts")

    # Rolling IC
    ics = []
    for i in range(window, len(df)):
        window_data = df.iloc[i-window:i]
        ic, _ = spearmanr(window_data["pred"], window_data["actual"])
        if not np.isnan(ic):
            ics.append(ic)

    if len(ics) < 2:
        return float("nan")

    ic_mean = np.mean(ics)
    ic_std = np.std(ics)

    if ic_std < 1e-10:
        return float("nan")

    return float(ic_mean / ic_std)

Hit Rate (Directional Accuracy)

def compute_hit_rate(
    predictions: np.ndarray,
    actuals: np.ndarray
) -> float:
    """Directional accuracy.

    LIMITATIONS for regression models:
    - Ignores prediction magnitude
    - pred=0.001 and pred=1.0 treated equally
    - Use alongside IC for full picture
    """
    correct = np.sign(predictions) == np.sign(actuals)
    return float(np.mean(correct))

Prediction Autocorrelation (Sticky Detection)

Detects common LSTM pathology of "sticky" predictions:

def compute_prediction_autocorr(
    predictions: np.ndarray,
    lag: int = 1
) -> float:
    """Lag-1 autocorrelation of predictions.

    Healthy range: 0.3 - 0.7

    WARNING signs:
    - >0.9: Predictions barely change ("sticky LSTM")
    - =1.0: Constant predictions (model collapsed to mean)
    - <0.1: Predictions are noise (no memory)

    REMEDIATION (2026-01-19 audit):
    - Return 1.0 for constant predictions (std < 1e-10)
    - NaN from corrcoef division-by-zero is incorrect semantically

    Source: Multi-agent audit finding (model-expert subagent)
    """
    if len(predictions) < lag + 2:
        return float("nan")

    # REMEDIATION: Check for constant predictions (std ≈ 0)
    if np.std(predictions) < 1e-10:
        return 1.0  # Constant predictions have perfect autocorrelation

    return float(np.corrcoef(
        predictions[:-lag],
        predictions[lag:]
    )[0, 1])


def detect_sticky_predictions(
    predictions: np.ndarray,
    threshold: float = 0.9
) -> dict:
    """Full sticky prediction diagnostic.

    Returns:
        - autocorr_lag1: Lag-1 autocorrelation
        - variance_ratio: var(diff) / var(pred)
        - is_sticky: Boolean flag
    """
    autocorr = compute_prediction_autocorr(predictions, lag=1)

    # Variance ratio: should be >0.2 for healthy models
    var_pred = np.var(predictions)
    var_diff = np.var(np.diff(predictions))
    var_ratio = var_diff / var_pred if var_pred > 1e-10 else 0.0

    return {
        "autocorr_lag1": autocorr,
        "variance_ratio": float(var_ratio),
        "is_sticky": autocorr > threshold or var_ratio < 0.1
    }

Residual Diagnostics (Ljung-Box)

from statsmodels.stats.diagnostic import acorr_ljungbox

def compute_residual_diagnostics(
    predictions: np.ndarray,
    actuals: np.ndarray,
    lags: list[int] = [1, 5, 10]
) -> dict:
    """Ljung-Box test for residual autocorrelation.

    If residuals are autocorrelated, model is missing structure.

    Returns:
        - lb_stats: Test statistics per lag
        - lb_pvalues: P-values (>0.05 = white noise residuals)
        - has_structure: True if p < 0.05 for any lag
    """
    residuals = actuals - predictions

    result = acorr_ljungbox(residuals, lags=lags, return_df=True)

    return {
        "lb_stats": result["lb_stat"].tolist(),
        "lb_pvalues": result["lb_pvalue"].tolist(),
        "has_structure": any(result["lb_pvalue"] < 0.05)
    }

Combined Quality Score

def compute_prediction_quality_score(
    predictions: np.ndarray,
    actuals: np.ndarray,
    timestamps: np.ndarray
) -> dict:
    """Combined quality assessment.

    Returns score 0-100 with letter grade.
    """
    ic = compute_ic(predictions, actuals, method="spearman")
    hit_rate = compute_hit_rate(predictions, actuals)
    sticky = detect_sticky_predictions(predictions)

    # Scoring
    score = 0.0

    # IC contribution (0-40 points)
    if not np.isnan(ic):
        if ic > 0.10:
            score += 40
        elif ic > 0.05:
            score += 30
        elif ic > 0.02:
            score += 20
        elif ic > 0:
            score += 10

    # Hit rate contribution (0-30 points)
    if hit_rate > 0.55:
        score += 30
    elif hit_rate > 0.52:
        score += 20
    elif hit_rate > 0.50:
        score += 10

    # Non-sticky contribution (0-30 points)
    if not sticky["is_sticky"]:
        score += 30
    elif sticky["autocorr_lag1"] < 0.95:
        score += 15

    # Letter grade
    if score >= 90:
        grade = "A"
    elif score >= 80:
        grade = "B"
    elif score >= 70:
        grade = "C"
    elif score >= 60:
        grade = "D"
    else:
        grade = "F"

    return {
        "score": score,
        "grade": grade,
        "ic": ic,
        "hit_rate": hit_rate,
        "is_sticky": sticky["is_sticky"],
        "autocorr_lag1": sticky["autocorr_lag1"]
    }

Model Collapse Detection (2026-01-19 Audit Addition)

CRITICAL: BiLSTM models can collapse to mean prediction when:

• hidden_size is too small (e.g., 16)
• dropout is too aggressive (e.g., 0.5)
• Signal-to-noise ratio is too low

def detect_model_collapse(
    predictions: np.ndarray,
    threshold: float = 1e-6
) -> dict:
    """Detect if model has collapsed to constant predictions.

    REMEDIATION (2026-01-19 audit):
    - Check prediction standard deviation
    - Log warning when detected
    - Continue recording for diagnostics

    Source: Multi-agent audit finding (model-expert subagent)

    Args:
        predictions: Model output predictions
        threshold: Std threshold below which collapse is detected

    Returns:
        Dictionary with collapse detection results
    """
    pred_std = np.std(predictions)
    is_collapsed = pred_std < threshold

    if is_collapsed:
        import logging
        logging.warning(
            f"Model collapse detected: std(predictions)={pred_std:.2e}. "
            "Check architecture/hyperparameters."
        )

    return {
        "is_collapsed": is_collapsed,
        "prediction_std": float(pred_std),
        "prediction_mean": float(np.mean(predictions)),
        "prediction_range": float(np.ptp(predictions)),  # max - min
    }


# RECOMMENDED ARCHITECTURE FIXES for BiLSTM mean prediction collapse:
# 1. Increase hidden_size: 16 → 48 (triple capacity)
# 2. Reduce dropout: 0.5 → 0.3 (less aggressive regularization)
# 3. Check learning rate: may need adjustment
# 4. Verify input feature variance: constant inputs → constant outputs

When to Use Each Metric

Risk Metrics for Range Bars

Key Principle: Daily Aggregation First

For all risk metrics on range bars, aggregate to daily before computation:

def _group_by_day(pnl: np.ndarray, timestamps: np.ndarray) -> np.ndarray:
    """Standard daily aggregation."""
    df = pd.DataFrame({"pnl": pnl, "ts": pd.to_datetime(timestamps, utc=True)})
    df["date"] = df["ts"].dt.date
    return df.groupby("date")["pnl"].sum().values

VaR and CVaR (Expected Shortfall)

def compute_var_cvar(
    pnl: np.ndarray,
    timestamps: np.ndarray,
    confidence: float = 0.95
) -> tuple[float, float]:
    """Value at Risk and Conditional VaR on daily-aggregated PnL.

    Args:
        pnl: Bar-level PnL
        timestamps: Bar timestamps
        confidence: Confidence level (0.95 = 95%)

    Returns:
        (VaR, CVaR) tuple
    """
    daily_pnl = _group_by_day(pnl, timestamps)

    alpha = 1 - confidence
    var = float(np.percentile(daily_pnl, alpha * 100))

    tail = daily_pnl[daily_pnl <= var]
    cvar = float(np.mean(tail)) if len(tail) > 0 else var

    return var, cvar

Omega Ratio

def compute_omega(
    pnl: np.ndarray,
    timestamps: np.ndarray,
    threshold: float = 0.0,
    min_days: int = 5
) -> float:
    """Omega ratio with daily aggregation.

    Omega = sum(gains above threshold) / sum(losses below threshold)

    Reference: Keating & Shadwick (2002)

    REMEDIATION (2026-01-19 audit):
    - Added min_days parameter to avoid unreliable values with too few samples.
    - Return NaN when n_days < min_days.

    Source: Multi-agent audit finding (robustness-analyst subagent)
    """
    daily_pnl = _group_by_day(pnl, timestamps)

    # REMEDIATION: Minimum sample size check
    if len(daily_pnl) < min_days:
        return float("nan")  # Unreliable with too few days

    excess = daily_pnl - threshold
    gains = excess[excess > 0].sum()
    losses = (-excess[excess < 0]).sum()

    if losses < 1e-10:
        return float("nan")  # No losses

    return float(gains / losses)

Ulcer Index

CRITICAL: Uses equity curve, NOT cumsum of returns.

def compute_ulcer_index(
    pnl: np.ndarray,
    timestamps: np.ndarray,
    initial_equity: float = 10000.0
) -> float:
    """Ulcer Index from equity curve.

    Ulcer = sqrt(mean(drawdown_pct^2))

    Reference: Peter Martin (1987)

    REMEDIATION (2026-01-19 audit):
    - Guard against division by zero when peak equity = 0.
    - Can happen if initial_equity + early losses < 0.

    Source: Multi-agent audit finding (risk-analyst subagent)
    """
    daily_pnl = _group_by_day(pnl, timestamps)

    # Build equity curve (NOT just cumsum)
    equity = initial_equity + np.cumsum(daily_pnl)

    # Percentage drawdowns from peak
    peak = np.maximum.accumulate(equity)

    # REMEDIATION: Guard against division by zero when peak = 0
    with np.errstate(divide='ignore', invalid='ignore'):
        drawdown_pct = np.where(peak > 1e-10, (equity - peak) / peak, 0.0)

    return float(np.sqrt((drawdown_pct ** 2).mean()))

Sortino Ratio

Preferred over Sharpe for crypto (asymmetric returns):

def compute_sortino(
    pnl: np.ndarray,
    timestamps: np.ndarray,
    mar: float = 0.0,  # Minimum Acceptable Return
    annualization: int = 365  # Crypto default
) -> float:
    """Sortino ratio using downside deviation only.

    Sortino = (Mean - MAR) / Downside Deviation

    Reference: Sortino & Price (1994)
    """
    daily_pnl = _group_by_day(pnl, timestamps)

    # Downside returns only
    downside = daily_pnl[daily_pnl < mar]
    if len(downside) == 0:
        return float("inf")  # No downside

    downside_std = np.std(downside, ddof=1)
    if downside_std < 1e-10:
        return float("nan")

    excess_return = np.mean(daily_pnl) - mar
    sortino = (excess_return / downside_std) * np.sqrt(annualization)

    return float(sortino)

Max Drawdown and Recovery Factor

def compute_max_drawdown(
    pnl: np.ndarray,
    timestamps: np.ndarray,
    initial_equity: float = 10000.0
) -> tuple[float, int]:
    """Max drawdown and duration.

    Returns:
        (max_dd_pct, duration_days)
    """
    daily_pnl = _group_by_day(pnl, timestamps)
    equity = initial_equity + np.cumsum(daily_pnl)

    peak = np.maximum.accumulate(equity)
    drawdown = (equity - peak) / peak

    max_dd = float(drawdown.min())

    # Duration: days from peak to recovery
    in_drawdown = drawdown < 0
    if not in_drawdown.any():
        return max_dd, 0

    # Find longest drawdown period
    changes = np.diff(in_drawdown.astype(int))
    starts = np.where(changes == 1)[0] + 1
    ends = np.where(changes == -1)[0] + 1

    if len(starts) == 0:
        starts = np.array([0]) if in_drawdown[0] else np.array([])
    if len(ends) == 0 or (len(starts) > 0 and ends[-1] < starts[-1]):
        ends = np.append(ends, len(drawdown))

    durations = ends[:len(starts)] - starts[:len(starts)]
    max_duration = int(durations.max()) if len(durations) > 0 else 0

    return max_dd, max_duration


def compute_recovery_factor(total_return: float, max_drawdown: float) -> float:
    """Recovery Factor = Total Return / |Max Drawdown|."""
    if abs(max_drawdown) < 1e-10:
        return float("nan")
    return float(total_return / abs(max_drawdown))

Profit Factor

def compute_profit_factor(
    pnl: np.ndarray,
    timestamps: np.ndarray,
    min_days: int = 5
) -> float:
    """Profit Factor with daily aggregation.

    PF = sum(winning days) / |sum(losing days)|

    REMEDIATION (2026-01-19 audit):
    - Added min_days parameter to avoid unreliable values with too few samples.
    - Return NaN when n_days < min_days.

    Source: Multi-agent audit finding (robustness-analyst subagent)
    """
    daily_pnl = _group_by_day(pnl, timestamps)

    # REMEDIATION: Minimum sample size check
    if len(daily_pnl) < min_days:
        return float("nan")  # Unreliable with too few days

    gains = daily_pnl[daily_pnl > 0].sum()
    losses = abs(daily_pnl[daily_pnl < 0].sum())

    if losses < 1e-10:
        return float("inf") if gains > 0 else 1.0

    return float(gains / losses)

Academic References

@article{keating2002omega,
  title={An Introduction to Omega},
  author={Keating, Con and Shadwick, William F},
  journal={AIMA Newsletter},
  year={2002}
}

@article{martin1987ulcer,
  title={The Ulcer Index},
  author={Martin, Peter},
  journal={Technical Analysis of Stocks \& Commodities},
  year={1987}
}

@article{sortino1994performance,
  title={Performance Measurement in a Downside Risk Framework},
  author={Sortino, Frank A and Price, Lee N},
  journal={The Journal of Investing},
  year={1994}
}

Sharpe Ratio Formulas for Range Bars

Note: This file contains quick-reference formulas for the ρ=0 (IID) special case used in range bar evaluation.

For the full non-IID framework (ρ-corrected variance, PSR, DSR, MinTRL, pFDR, oFDR, SFDR with 82 verified equations

and Numba JIT implementation), see the dedicated skill:

sharpe-ratio-non-iid-corrections.

Why Standard Sharpe Fails for Range Bars

Standard Sharpe annualization assumes:

1. IID returns: Each observation is independent, identically distributed 2. Fixed intervals: Observations occur at regular time intervals 3. sqrt(N) scaling: Volatility scales with sqrt(time)

Range bars violate ALL three assumptions.

CRITICAL: Rangebar v9 Duration Unit Mismatch

WARNING (2026-01-22): The rangebar crate v9 returns duration_us in MILLISECONDS

despite the column name suggesting microseconds. This causes ~1000x Sharpe inflation.

Detection heuristic: If 95th percentile of duration_us < 1e5, values are likely milliseconds.

import numpy as np

def validate_duration_units(duration_us: np.ndarray) -> str:
    """Detect if duration_us is actually in milliseconds (rangebar v9 bug)."""
    p95 = np.percentile(duration_us, 95)

    if p95 < 1e5:  # < 0.1 seconds if microseconds → likely milliseconds
        return "milliseconds"  # Need MS_TO_US = 1000 conversion
    elif p95 < 1e8:  # < 100 seconds if microseconds → plausible
        return "microseconds"  # No conversion needed
    else:
        return "unknown"  # Investigate further

# Fix: Convert milliseconds to microseconds before compute_time_weighted_sharpe
MS_TO_US = 1000
duration_us_fixed = duration_us * MS_TO_US  # If validate returns "milliseconds"

Impact without fix: Sharpe inflated by ~170x (total_days 1000x smaller → annualization factor ~31x larger).

Canonical Approach 1: Time-Weighted Sharpe Ratio (TWSR)

Preferred for bar-level evaluation. Directly handles variable-duration bars without aggregation.

import numpy as np

def compute_time_weighted_sharpe(
    bar_pnl: np.ndarray,
    duration_us: np.ndarray,
) -> tuple[float, float, float]:
    """
    Time-Weighted Sharpe Ratio (TWSR) for variable-duration bars.

    FORMULA:
      TWSR = (simple_mean(r) / time_weighted_std(r)) × √(365.25 / T)

    Where:
      - simple_mean(r) = sum(r_i) / N (preserves total P&L sign)
      - time_weighted_std(r) = √(sum(w_i × (r_i - simple_mean)²))
      - w_i = d_i / T (time weights, sum to 1)
      - T = total observation time in days
      - 365.25 = days per year (24/7 crypto markets)

    The annualization factor √(365.25 / T) projects the observed Sharpe to
    annual units based on actual observation time T. Derived from Wiener
    process property where variance scales linearly with time.

    Returns:
        (sharpe, weighted_std, total_days)
    """
    if len(bar_pnl) == 0 or len(duration_us) == 0:
        return 0.0, 0.0, 0.0

    bar_pnl = np.asarray(bar_pnl, dtype=np.float64)
    duration_us = np.asarray(duration_us, dtype=np.float64)

    MICROSECONDS_PER_DAY = 86400.0 * 1e6
    duration_days = duration_us / MICROSECONDS_PER_DAY
    total_days = float(np.sum(duration_days))

    if total_days < 1e-15:
        return 0.0, 0.0, 0.0

    weights = duration_days / total_days
    simple_mean = float(np.mean(bar_pnl))
    weighted_var = float(np.sum(weights * (bar_pnl - simple_mean) ** 2))

    if weighted_var < 1e-20:
        return 0.0, 0.0, total_days

    weighted_std = np.sqrt(weighted_var)
    raw_sharpe = simple_mean / weighted_std

    # Time-scaled annualization: project to 365.25 days (24/7 crypto)
    annualization_factor = np.sqrt(365.25 / total_days)
    sharpe = raw_sharpe * annualization_factor

    return float(sharpe), float(weighted_std), total_days

Key Properties:

• Positive total P&L always produces positive Sharpe (sign preservation)
• Long losing bars get penalized via higher weighted volatility
• Microsecond precision for duration calculations
• Proper time-based annualization (no arbitrary 252)

Reference: ~/.claude/docs/GLOSSARY.md (TWSR canonical definition)

Canonical Approach 2: Daily Aggregation

import pandas as pd
import numpy as np

def _group_by_day(pnl: np.ndarray, timestamps: np.ndarray) -> np.ndarray:
    """Aggregate bar-level PnL to daily PnL.

    This restores IID-like properties by:
    1. Normalizing variable bar counts per day
    2. Creating fixed-interval (daily) observations
    3. Reducing autocorrelation from bar clustering
    """
    df = pd.DataFrame({
        "pnl": pnl,
        "ts": pd.to_datetime(timestamps, utc=True)
    })
    df["date"] = df["ts"].dt.date
    return df.groupby("date")["pnl"].sum().values


def weekly_sharpe(pnl: np.ndarray, timestamps: np.ndarray, days_per_week: int = 7) -> float:
    """Daily-aggregated Sharpe scaled to weekly.

    Args:
        pnl: Bar-level PnL array
        timestamps: Bar close timestamps (UTC)
        days_per_week: 7 for crypto, 5 for equities

    Returns:
        Weekly Sharpe ratio
    """
    daily_pnl = _group_by_day(pnl, timestamps)

    if len(daily_pnl) < 2:
        return 0.0

    std = np.std(daily_pnl, ddof=1)
    if std < 1e-10:
        return 0.0

    daily_sharpe = np.mean(daily_pnl) / std
    return daily_sharpe * np.sqrt(days_per_week)

Annualization Factors

annualization:
  # TWSR (Time-Weighted Sharpe Ratio) - for bar-level range bar data
  twsr_crypto:
    formula: "sqrt(365.25 / T)" # T = observation period in days
    rationale: "Projects to annual based on ACTUAL observation time"
    use_when: "Bar-level evaluation with duration_us available"

  # Daily Aggregation - for daily-aggregated data
  crypto_24_7:
    daily_to_weekly: 2.6458 # sqrt(7)
    daily_to_annual: 19.1049 # sqrt(365)
    rationale: "Crypto markets trade 24/7, 365 days/year"

  equity:
    daily_to_weekly: 2.2361 # sqrt(5)
    daily_to_annual: 15.8745 # sqrt(252)
    rationale: "Equity markets trade ~252 days/year"

  # CRITICAL: Never mix these!
  anti_patterns:
    - "Using sqrt(252) for crypto TWSR"
    - "Using sqrt(365) for equities"
    - "Using sqrt(N) on bar-level data directly (use TWSR instead)"
    - "Using sqrt(T * 252) - WRONG formula (should be sqrt(365.25 / T))"

Mertens (2002) Standard Error

Non-normality adjustment for Sharpe SE:

from scipy import stats

def sharpe_standard_error(
    sharpe: float,
    n_observations: int,
    skewness: float,
    kurtosis: float  # Pearson form: normal = 3
) -> float:
    """Sharpe SE with Mertens (2002) non-normality adjustment.

    SE(SR) = sqrt((1 + 0.5×SR² - γ₃×SR + (γ₄-3)/4×SR²) / (n-1))

    Where:
        γ₃ = skewness
        γ₄ = kurtosis (Pearson, normal = 3)
    """
    if n_observations < 2:
        return float("nan")

    # Excess kurtosis term
    excess_kurt = kurtosis - 3.0

    variance_term = (
        1.0
        + 0.5 * sharpe**2
        - skewness * sharpe
        + (excess_kurt / 4.0) * sharpe**2
    )

    if variance_term < 0:
        return float("nan")

    return float(np.sqrt(variance_term / (n_observations - 1)))

PSR (Probabilistic Sharpe Ratio)

from scipy.stats import norm

def probabilistic_sharpe_ratio(
    sharpe: float,
    standard_error: float,
    benchmark: float = 0.0
) -> float:
    """P(true Sharpe > benchmark).

    Bailey & López de Prado (2012):
    PSR = Φ((SR - benchmark) / SE(SR))

    Returns:
        Probability in [0, 1]. >0.95 indicates significance.
    """
    if standard_error <= 1e-10:
        return float("nan")

    z_score = (sharpe - benchmark) / standard_error
    return float(norm.cdf(z_score))

DSR (Deflated Sharpe Ratio)

Corrects for multiple testing:

def deflated_sharpe_ratio(
    sharpe: float,
    standard_error: float,
    n_trials: int
) -> float:
    """Sharpe corrected for multiple testing.

    Bailey & López de Prado (2014):
    Uses Gumbel approximation for expected maximum Sharpe.

    Args:
        sharpe: Observed Sharpe (or max across trials)
        standard_error: SE of Sharpe
        n_trials: Number of independent strategies tested

    Returns:
        Probability in [0, 1]. >0.50 indicates robust performance.
    """
    if n_trials < 1 or standard_error <= 1e-10:
        return float("nan")

    gamma = 0.5772156649  # Euler-Mascheroni constant

    if n_trials == 1:
        sr_expected = 0.0
    else:
        q1 = norm.ppf(1.0 - 1.0 / n_trials)
        q2 = norm.ppf(1.0 - 1.0 / (n_trials * np.e))
        sr_expected = standard_error * ((1 - gamma) * q1 + gamma * q2)

    z_score = (sharpe - sr_expected) / standard_error
    return float(norm.cdf(z_score))

MinTRL (Minimum Track Record Length)

def minimum_track_record_length(
    sharpe: float,
    benchmark: float,
    skewness: float,
    kurtosis: float,  # Pearson form
    alpha: float = 0.05
) -> float:
    """Observations needed for statistical significance.

    Uses Mertens (2002) variance formula.

    CRITICAL: Use (kurtosis - 3) for excess kurtosis!
    """
    if sharpe <= benchmark:
        return float("nan")

    z_crit = norm.ppf(1 - alpha)
    excess_kurt = kurtosis - 3.0

    # Mertens variance term
    variance_term = (
        1.0
        + 0.5 * sharpe**2
        - skewness * sharpe
        + (excess_kurt / 4.0) * sharpe**2
    )

    sr_diff = sharpe - benchmark
    mintrl = variance_term * (z_crit / sr_diff) ** 2

    return float(max(1.0, mintrl))

Academic References

@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},
  year={2014}
}

@article{mertens2002sharpe,
  title={The Sharpe Ratio and the Information Ratio},
  author={Mertens, Elmar},
  journal={Financial Analysts Journal},
  year={2002}
}

@article{lo2002statistics,
  title={The Statistics of Sharpe Ratios},
  author={Lo, Andrew W},
  journal={Financial Analysts Journal},
  volume={58},
  number={4},
  pages={36--52},
  year={2002}
}

State-of-the-Art Methods (2025-2026)

Table of Contents

• Regime Detection
• Why It Matters for Range Bars
• BOCPD (Bayesian Online Changepoint Detection)
• PELT (Pruned Exact Linear Time)
• Integration with Range Bar Metrics
• References
• Probabilistic Forecasting
• Why Point Estimates Are Insufficient
• Conformal Prediction (Distribution-Free)
• Quantile Regression
• References
• Explainability (SHAP/LIME)
• EU AI Act 2025 Compliance
• SHAP (SHapley Additive exPlanations)
• LIME (Local Interpretable Model-agnostic Explanations)
• Integration with Range Bar Metrics
• References
• Transformers for Time Series (2025 SOTA)
• Why Transformers?
• Recommended: PatchTST for Range Bars
• References
• Summary: SOTA Integration Checklist
• Installation

Advanced techniques for range bar analysis beyond baseline metrics.

Regime Detection

Why It Matters for Range Bars

Range bars exhibit regime-dependent behavior:

• High volatility: More bars per day, shorter durations
• Low volatility: Fewer bars, longer durations
• Trend regimes: Directional clustering
• Mean-reversion regimes: Oscillatory patterns

A single model trained across all regimes underperforms regime-specific models.

BOCPD (Bayesian Online Changepoint Detection)

Adams & MacKay (2007) - Gold standard for online detection.

import numpy as np
from scipy import stats

def bocpd_online(
    data: np.ndarray,
    hazard_rate: float = 0.01,
    observation_model: str = "gaussian",
) -> np.ndarray:
    """Bayesian Online Changepoint Detection.

    Args:
        data: Time series observations (e.g., daily returns)
        hazard_rate: Prior probability of changepoint (1/expected_run_length)
        observation_model: "gaussian" or "student_t"

    Returns:
        Array of changepoint probabilities per timestep
    """
    n = len(data)
    R = np.zeros((n + 1, n + 1))  # Run length distribution
    R[0, 0] = 1.0

    # Sufficient statistics for Gaussian
    sum_x = np.zeros(n + 1)
    sum_x2 = np.zeros(n + 1)

    changepoint_probs = np.zeros(n)

    for t in range(n):
        x = data[t]

        # Update sufficient statistics
        sum_x[1:t+2] = sum_x[:t+1] + x
        sum_x2[1:t+2] = sum_x2[:t+1] + x**2

        # Predictive probability under each run length
        predprobs = np.zeros(t + 1)
        for r in range(t + 1):
            n_obs = r + 1
            mean_r = sum_x[r+1] / n_obs if n_obs > 0 else 0
            var_r = max(1e-10, sum_x2[r+1] / n_obs - mean_r**2 if n_obs > 1 else 1.0)
            predprobs[r] = stats.norm.pdf(x, mean_r, np.sqrt(var_r))

        # Growth probability
        R[1:t+2, t+1] = R[:t+1, t] * predprobs * (1 - hazard_rate)

        # Changepoint probability
        R[0, t+1] = np.sum(R[:t+1, t] * predprobs) * hazard_rate

        # Normalize
        R[:t+2, t+1] /= R[:t+2, t+1].sum()

        # Changepoint probability at t
        changepoint_probs[t] = R[0, t+1]

    return changepoint_probs


def detect_regimes(data: np.ndarray, threshold: float = 0.5) -> list[int]:
    """Get changepoint indices where probability > threshold."""
    cp_probs = bocpd_online(data)
    return list(np.where(cp_probs > threshold)[0])

PELT (Pruned Exact Linear Time)

Killick et al. (2012) - Offline detection with optimal segmentation.

# Using ruptures library (pip install ruptures)
import ruptures as rpt

def pelt_offline(
    data: np.ndarray,
    model: str = "rbf",  # "l1", "l2", "rbf", "normal"
    min_size: int = 10,
    penalty: float | None = None,
) -> list[int]:
    """PELT changepoint detection (offline).

    Args:
        data: Time series (1D or 2D)
        model: Cost function model
        min_size: Minimum segment length
        penalty: Penalty value (None = auto BIC)

    Returns:
        List of changepoint indices
    """
    if penalty is None:
        # BIC penalty
        penalty = np.log(len(data)) * data.ndim

    algo = rpt.Pelt(model=model, min_size=min_size).fit(data)
    changepoints = algo.predict(pen=penalty)

    # Remove last point (always included by ruptures)
    return changepoints[:-1] if changepoints else []

Integration with Range Bar Metrics

def regime_aware_metrics(
    predictions: np.ndarray,
    actuals: np.ndarray,
    timestamps: np.ndarray,
    regime_labels: np.ndarray,
) -> dict:
    """Compute metrics per regime for regime-aware evaluation."""
    results = {"overall": evaluate_fold(predictions, actuals, timestamps)}

    unique_regimes = np.unique(regime_labels)
    for regime in unique_regimes:
        mask = regime_labels == regime
        if mask.sum() < 10:
            continue

        results[f"regime_{regime}"] = evaluate_fold(
            predictions[mask],
            actuals[mask],
            timestamps[mask],
        )

    return results

References

@article{adams2007bocpd,
  title={Bayesian Online Changepoint Detection},
  author={Adams, Ryan Prescott and MacKay, David JC},
  journal={arXiv preprint arXiv:0710.3742},
  year={2007}
}

@article{killick2012pelt,
  title={Optimal Detection of Changepoints with a Linear Computational Cost},
  author={Killick, Rebecca and Fearnhead, Paul and Eckley, Idris A},
  journal={Journal of the American Statistical Association},
  year={2012}
}

---

Probabilistic Forecasting

Why Point Estimates Are Insufficient

Point predictions (e.g., prediction = 0.003) provide no uncertainty quantification.

For risk management, you need:

• Prediction intervals: "95% CI: [-0.01, 0.02]"
• Distribution forecasts: Full predictive distribution
• Calibration: Stated coverage matches empirical coverage

Conformal Prediction (Distribution-Free)

Vovk et al. (2005), Angelopoulos et al. (2023) - Finite-sample guarantees.

def conformal_prediction_interval(
    residuals_cal: np.ndarray,  # Calibration set residuals
    predictions_test: np.ndarray,
    alpha: float = 0.05,
) -> tuple[np.ndarray, np.ndarray]:
    """Split conformal prediction interval.

    Guarantees: P(Y_test in interval) >= 1 - alpha (finite sample)

    Args:
        residuals_cal: |y_cal - pred_cal| from calibration set
        predictions_test: Point predictions for test set
        alpha: Miscoverage rate (0.05 = 95% interval)

    Returns:
        (lower_bounds, upper_bounds) for test predictions
    """
    n_cal = len(residuals_cal)

    # Quantile with finite-sample correction
    q_level = np.ceil((n_cal + 1) * (1 - alpha)) / n_cal
    q_hat = np.quantile(residuals_cal, min(q_level, 1.0))

    lower = predictions_test - q_hat
    upper = predictions_test + q_hat

    return lower, upper


def conformal_prediction_with_wfo(
    predictions: np.ndarray,
    actuals: np.ndarray,
    fold_indices: list[tuple[int, int]],  # (cal_start, test_start, test_end)
    alpha: float = 0.05,
) -> dict:
    """Conformal prediction integrated with WFO.

    Each fold uses previous OOS data as calibration set.
    """
    all_intervals = []
    coverages = []

    for i, (cal_start, test_start, test_end) in enumerate(fold_indices):
        if i == 0:
            # First fold: no calibration data, skip
            continue

        # Calibration: previous fold's OOS residuals
        cal_residuals = np.abs(
            actuals[cal_start:test_start] - predictions[cal_start:test_start]
        )

        # Test predictions
        test_preds = predictions[test_start:test_end]
        test_actuals = actuals[test_start:test_end]

        lower, upper = conformal_prediction_interval(
            cal_residuals, test_preds, alpha
        )

        # Check coverage
        covered = (test_actuals >= lower) & (test_actuals <= upper)
        coverage = np.mean(covered)

        all_intervals.append({
            "fold": i,
            "lower": lower,
            "upper": upper,
            "coverage": coverage,
            "target_coverage": 1 - alpha,
        })
        coverages.append(coverage)

    return {
        "intervals": all_intervals,
        "mean_coverage": np.mean(coverages),
        "coverage_std": np.std(coverages),
        "is_calibrated": np.mean(coverages) >= (1 - alpha - 0.05),  # Within 5%
    }

Quantile Regression

Direct prediction of quantiles (e.g., 5th, 50th, 95th percentiles).

import torch
import torch.nn as nn

class QuantileLoss(nn.Module):
    """Pinball loss for quantile regression."""

    def __init__(self, quantiles: list[float]):
        super().__init__()
        self.quantiles = torch.tensor(quantiles)

    def forward(self, predictions: torch.Tensor, targets: torch.Tensor) -> torch.Tensor:
        # predictions: (batch, n_quantiles)
        # targets: (batch, 1)

        errors = targets - predictions
        losses = torch.max(
            (self.quantiles - 1) * errors,
            self.quantiles * errors,
        )
        return losses.mean()


class QuantileBiLSTM(nn.Module):
    """BiLSTM with multiple quantile outputs."""

    def __init__(
        self,
        input_size: int,
        hidden_size: int = 48,
        quantiles: list[float] = [0.05, 0.25, 0.50, 0.75, 0.95],
    ):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size, hidden_size, bidirectional=True, batch_first=True
        )
        self.fc = nn.Linear(hidden_size * 2, len(quantiles))
        self.quantiles = quantiles

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        lstm_out, _ = self.lstm(x)
        last_hidden = lstm_out[:, -1, :]
        return self.fc(last_hidden)

    def predict_interval(self, x: torch.Tensor, alpha: float = 0.05) -> tuple:
        """Get prediction interval from quantile outputs."""
        quantiles = self.forward(x)
        # Assuming quantiles = [0.05, 0.25, 0.50, 0.75, 0.95]
        lower = quantiles[:, 0]  # 5th percentile
        median = quantiles[:, 2]  # 50th percentile
        upper = quantiles[:, 4]  # 95th percentile
        return lower, median, upper

References

@book{vovk2005conformal,
  title={Algorithmic Learning in a Random World},
  author={Vovk, Vladimir and Gammerman, Alex and Shafer, Glenn},
  year={2005},
  publisher={Springer}
}

@article{angelopoulos2023conformal,
  title={Conformal Prediction: A Gentle Introduction},
  author={Angelopoulos, Anastasios N and Bates, Stephen},
  journal={Foundations and Trends in Machine Learning},
  year={2023}
}

---

Explainability (SHAP/LIME)

EU AI Act 2025 Compliance

The EU AI Act (effective 2025) requires explainability for "high-risk" AI systems, including financial trading algorithms with significant economic impact.

Key requirements:

• Transparency: Users must understand how predictions are made
• Documentation: Model behavior must be documented
• Auditability: Explanations must be reproducible

SHAP (SHapley Additive exPlanations)

Lundberg & Lee (2017) - Game-theoretic feature attribution.

import shap
import numpy as np

def explain_bilstm_predictions(
    model,  # Trained BiLSTM
    X_train: np.ndarray,
    X_explain: np.ndarray,
    feature_names: list[str],
    n_background: int = 100,
) -> dict:
    """SHAP explanations for BiLSTM predictions.

    Args:
        model: Trained model with predict() method
        X_train: Training data for background distribution
        X_explain: Samples to explain
        feature_names: Names of input features
        n_background: Number of background samples

    Returns:
        SHAP values and summary statistics
    """
    # Sample background data
    if len(X_train) > n_background:
        idx = np.random.choice(len(X_train), n_background, replace=False)
        background = X_train[idx]
    else:
        background = X_train

    # DeepExplainer for neural networks
    explainer = shap.DeepExplainer(model, background)
    shap_values = explainer.shap_values(X_explain)

    # Aggregate across time steps (for sequence data)
    # Shape: (n_samples, seq_len, n_features) -> (n_samples, n_features)
    if len(shap_values.shape) == 3:
        shap_aggregated = np.abs(shap_values).mean(axis=1)
    else:
        shap_aggregated = np.abs(shap_values)

    # Feature importance ranking
    feature_importance = shap_aggregated.mean(axis=0)
    ranked_features = sorted(
        zip(feature_names, feature_importance),
        key=lambda x: x[1],
        reverse=True,
    )

    return {
        "shap_values": shap_values,
        "shap_aggregated": shap_aggregated,
        "feature_importance": dict(ranked_features),
        "top_5_features": [f[0] for f in ranked_features[:5]],
    }


def generate_explanation_report(
    shap_results: dict,
    predictions: np.ndarray,
    actuals: np.ndarray,
    output_path: str,
) -> None:
    """Generate EU AI Act compliant explanation report."""
    report = f"""
# Model Explanation Report

## Feature Importance (SHAP)

Top contributing features to model predictions:

| Rank | Feature | Mean |SHAP| |
|------|---------|------------|
"""
    for i, (feat, importance) in enumerate(shap_results["feature_importance"].items()):
        if i >= 10:
            break
        report += f"| {i+1} | {feat} | {importance:.4f} |\n"

    report += f"""
## Model Performance Summary

- Total predictions: {len(predictions)}
- Mean prediction: {np.mean(predictions):.4f}
- Mean actual: {np.mean(actuals):.4f}
- Correlation: {np.corrcoef(predictions, actuals)[0,1]:.4f}

## Interpretation

The model primarily relies on the following features:
{', '.join(shap_results['top_5_features'])}

This report is generated for EU AI Act compliance documentation.
"""
    with open(output_path, "w") as f:
        f.write(report)

LIME (Local Interpretable Model-agnostic Explanations)

Ribeiro et al. (2016) - Local surrogate explanations.

from lime import lime_tabular

def lime_explain_prediction(
    model,
    X_train: np.ndarray,
    x_instance: np.ndarray,
    feature_names: list[str],
    num_features: int = 10,
) -> dict:
    """LIME explanation for a single prediction.

    Better for individual "why this prediction?" questions.
    """
    explainer = lime_tabular.LimeTabularExplainer(
        X_train,
        feature_names=feature_names,
        mode="regression",
    )

    explanation = explainer.explain_instance(
        x_instance,
        model.predict,
        num_features=num_features,
    )

    return {
        "feature_weights": dict(explanation.as_list()),
        "local_prediction": explanation.local_pred[0],
        "intercept": explanation.intercept[0],
        "r2_score": explanation.score,
    }

Integration with Range Bar Metrics

def evaluate_fold_with_explanations(
    predictions: np.ndarray,
    actuals: np.ndarray,
    timestamps: np.ndarray,
    model,
    X_train: np.ndarray,
    X_test: np.ndarray,
    feature_names: list[str],
) -> dict:
    """Full evaluation including explainability metrics."""
    # Standard metrics
    metrics = evaluate_fold(predictions, actuals, timestamps)

    # Explainability
    shap_results = explain_bilstm_predictions(
        model, X_train, X_test[:100], feature_names
    )

    metrics["explainability"] = {
        "top_5_features": shap_results["top_5_features"],
        "feature_importance": shap_results["feature_importance"],
        "explanation_method": "SHAP DeepExplainer",
    }

    return metrics

References

@inproceedings{lundberg2017shap,
  title={A Unified Approach to Interpreting Model Predictions},
  author={Lundberg, Scott M and Lee, Su-In},
  booktitle={NeurIPS},
  year={2017}
}

@inproceedings{ribeiro2016lime,
  title={Why Should I Trust You?: Explaining the Predictions of Any Classifier},
  author={Ribeiro, Marco Tulio and Singh, Sameer and Guestrin, Carlos},
  booktitle={KDD},
  year={2016}
}

---

Transformers for Time Series (2025 SOTA)

Why Transformers?

BiLSTMs are being replaced by Transformer architectures for time series:

Recommended: PatchTST for Range Bars

# Using pytorch-forecasting or huggingface
# pip install pytorch-forecasting

from pytorch_forecasting import TemporalFusionTransformer
from pytorch_forecasting.data import TimeSeriesDataSet

# Example setup (conceptual)
training = TimeSeriesDataSet(
    data,
    time_idx="time_idx",
    target="return",
    group_ids=["asset"],
    max_encoder_length=60,  # Look-back
    max_prediction_length=1,  # 1-step ahead
    # ... additional config
)

model = TemporalFusionTransformer.from_dataset(
    training,
    hidden_size=32,
    attention_head_size=4,
    dropout=0.1,
    hidden_continuous_size=16,
)

References

@article{nie2023patchtst,
  title={A Time Series is Worth 64 Words: Long-term Forecasting with Transformers},
  author={Nie, Yuqi and others},
  journal={ICLR},
  year={2023}
}

@article{lim2021tft,
  title={Temporal Fusion Transformers for Interpretable Multi-horizon Time Series Forecasting},
  author={Lim, Bryan and others},
  journal={International Journal of Forecasting},
  year={2021}
}

---

Summary: SOTA Integration Checklist

When upgrading range bar analysis to 2025 SOTA:

• [ ] Regime detection: Implement BOCPD or PELT for volatility regimes
• [ ] Uncertainty quantification: Add conformal prediction intervals
• [ ] Explainability: Generate SHAP reports for EU AI Act compliance
• [ ] Architecture: Consider PatchTST or TFT over BiLSTM for new projects
• [ ] Calibration: Verify prediction interval coverage empirically

Installation

# Regime detection
pip install ruptures

# Explainability
pip install shap lime

# Transformers (optional)
pip install pytorch-forecasting

Structured Logging Contract for AWFES Experiments

Machine-readable NDJSON logging standard for ML experiments. Enables observability, debugging, and post-hoc analysis.

Why Structured Logging?

1. Machine-analyzable: NDJSON format for programmatic analysis 2. Audit trail: Complete record of experiment decisions 3. Debugging: Correlate events across distributed components 4. Metrics extraction: Automated pipeline monitoring

NDJSON Schema

Every log line is a JSON object with stable schema:

{
  "timestamp": "2026-01-20T15:28:45.123456+00:00",
  "level": "INFO",
  "message": "Fold 1 complete: TestSharpe=0.0576, WFE=0.183",
  "component": "awfes_v3",
  "environment": "research",
  "pid": 12345,
  "tid": 67890,
  "trace_id": "exp066_20260120_152845_abc123",
  "file": "exp066_awfes_v3.py",
  "line": 456,
  "function": "run_nested_fold",
  "context": {
    "fold_idx": 1,
    "test_sharpe": 0.0576,
    "wfe": 0.183,
    "prior_bayesian_epoch": 450,
    "val_optimal_epoch": 500
  }
}

Required Fields

Optional Context Field

The context field contains structured data specific to the log event:

Fold Events

{
  "context": {
    "fold_idx": 1,
    "n_train": 3900,
    "n_val": 780,
    "n_test": 300,
    "embargo_bars": 70
  }
}

Epoch Selection Events

{
  "context": {
    "fold_idx": 1,
    "prior_bayesian_epoch": 450,
    "val_optimal_epoch": 500,
    "test_epoch_used": 450,
    "wfe": 0.183,
    "frontier_epochs": [250, 450, 650]
  }
}

Test Results Events

{
  "context": {
    "fold_idx": 1,
    "test_sharpe": 0.0576,
    "test_hit_rate": 0.512,
    "test_n_bars": 300,
    "dsr": {
      "dsr": 0.0685,
      "sharpe_se": 0.059,
      "expected_max_null": 0.107,
      "z_score": -1.14,
      "significant": false
    }
  }
}

Bayesian Update Events

{
  "context": {
    "fold_idx": 1,
    "prior_mean": 425.0,
    "posterior_mean": 437.5,
    "prior_variance": 6658.0,
    "posterior_variance": 5326.4,
    "observed_epoch": 500,
    "wfe_weight": 0.183
  }
}

Implementation with loguru

import json
from datetime import timezone
from uuid import uuid4
from loguru import logger

COMPONENT_NAME = "awfes_v3"
ENVIRONMENT = "research"
EXPERIMENT_TRACE_ID = f"exp066_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid4().hex[:6]}"


def ndjson_serializer(record: dict) -> str:
    """Serialize loguru record to NDJSON with stable schema."""
    log_entry = {
        "timestamp": record["time"].astimezone(timezone.utc).isoformat(),
        "level": record["level"].name,
        "message": record["message"],
        "component": COMPONENT_NAME,
        "environment": ENVIRONMENT,
        "pid": record["process"].id,
        "tid": record["thread"].id,
        "trace_id": EXPERIMENT_TRACE_ID,
        "file": record["file"].name,
        "line": record["line"],
        "function": record["function"],
    }

    # Add context from extra fields
    if record["extra"]:
        context = {k: v for k, v in record["extra"].items() if not k.startswith("_")}
        if context:
            log_entry["context"] = context

    return json.dumps(log_entry, default=str, ensure_ascii=False)


def configure_structured_logging(log_dir: str = "logs") -> None:
    """Configure loguru for structured NDJSON logging."""
    from pathlib import Path

    Path(log_dir).mkdir(parents=True, exist_ok=True)

    # Remove default handler
    logger.remove()

    # Console: human-readable format
    logger.add(
        lambda msg: print(msg, end=""),
        format="{time:HH:mm:ss} | {level: <8} | {message}",
        level="INFO",
    )

    # File: NDJSON format with rotation
    logger.add(
        f"{log_dir}/{COMPONENT_NAME}_{{time:YYYY-MM-DD}}.ndjson",
        format="{message}",
        serialize=False,
        filter=lambda record: True,
        level="DEBUG",
        rotation="100 MB",
        retention="30 days",
    )

    # Patch the sink to use our serializer
    # (In practice, use a custom sink or serialize manually)

Log Event Types

Experiment Lifecycle

Fold Lifecycle

Critical Checkpoints (MANDATORY)

These log events MUST be present for audit compliance:

# MANDATORY: v3 temporal ordering checkpoint
fold_log.info(
    "v3_temporal_checkpoint",
    fold_idx=fold_idx,
    prior_bayesian_epoch=prior_bayesian_epoch,
    val_optimal_epoch=val_optimal_epoch,
    test_epoch_used=prior_bayesian_epoch,  # MUST equal prior_bayesian_epoch
    temporal_order_valid=True,
)

# MANDATORY: DSR computation
fold_log.info(
    "dsr_computed",
    fold_idx=fold_idx,
    sharpe=test_sharpe,
    dsr=dsr_result["dsr"],
    significant=dsr_result["significant"],
    n_trials=n_trials,
)

Analysis Queries

Extract All Fold Results

# Using jq to extract fold results
cat logs/awfes_v3_*.ndjson | \
  jq -c 'select(.message | contains("fold_complete"))' | \
  jq -s '[.[] | .context]'

Find Look-Ahead Violations

# v3 temporal order must have test_epoch_used == prior_bayesian_epoch
cat logs/awfes_v3_*.ndjson | \
  jq -c 'select(.message == "v3_temporal_checkpoint")' | \
  jq 'select(.context.test_epoch_used != .context.prior_bayesian_epoch)'

Aggregate DSR Statistics

# Extract all DSR values
cat logs/awfes_v3_*.ndjson | \
  jq -c 'select(.message == "dsr_computed")' | \
  jq -s '{
    mean_dsr: ([.[].context.dsr] | add / length),
    n_significant: ([.[].context.significant] | map(select(. == true)) | length),
    n_total: length
  }'

File Naming Convention

{component}_{YYYY-MM-DD}.ndjson

Examples:

• awfes_v3_2026-01-20.ndjson
• bilstm_trainer_2026-01-20.ndjson

Retention Policy

References

• loguru documentation
• NDJSON specification
• devops-tools:python-logging-best-practices

Temporal Aggregation for Range Bars

Core Principle: UTC Day Boundaries

def _group_by_day(pnl: np.ndarray, timestamps: np.ndarray) -> np.ndarray:
    """Canonical daily aggregation function.

    ALWAYS use this before computing Sharpe or risk metrics.
    """
    df = pd.DataFrame({
        "pnl": pnl,
        "ts": pd.to_datetime(timestamps, utc=True)  # EXPLICIT UTC
    })
    df["date"] = df["ts"].dt.date
    return df.groupby("date")["pnl"].sum().values

Why UTC?

1. Crypto standard: Binance, CoinGecko, all major exchanges use UTC 2. No DST issues: UTC never changes 3. Consistent aggregation: Same day boundary globally

Session Filter Interaction

CRITICAL ORDER: Filter THEN aggregate.

def evaluate_with_filter(predictions, actuals, timestamps, session_filter=True):
    """Correct order: filter → aggregate → compute."""

    # Step 1: Apply session filter (if enabled)
    if session_filter:
        mask = compute_tradeable_mask(timestamps)
        predictions = predictions[mask]
        actuals = actuals[mask]
        timestamps = timestamps[mask]

    # Step 2: Now aggregate to daily
    pnl = predictions * actuals
    daily_pnl = _group_by_day(pnl, timestamps)

    # Step 3: Compute Sharpe on daily
    sharpe = np.mean(daily_pnl) / np.std(daily_pnl) * np.sqrt(7)

    return sharpe

Dual-View Computation

Compute BOTH views independently:

def compute_dual_view(predictions, actuals, timestamps):
    """Independent computation for each view."""

    pnl = predictions * actuals

    # View 1: All bars (no filter)
    all_bars_metrics = evaluate_fold(predictions, actuals, None, timestamps)

    # View 2: Session filtered (London-NY)
    mask = compute_tradeable_mask(timestamps)
    filtered_metrics = evaluate_fold(predictions, actuals, mask, timestamps)

    return {
        "oos_metrics": filtered_metrics,      # Primary (strategy evaluation)
        "oos_metrics_all": all_bars_metrics   # Diagnostic (regime detection)
    }

Timezone Reference

timezone_mapping:
  crypto_standard: "UTC"

  market_sessions:
    tokyo: "Asia/Tokyo" # UTC+9
    london: "Europe/London" # UTC+0 (winter) / UTC+1 (summer)
    new_york: "America/New_York" # UTC-5 (winter) / UTC-4 (summer)

  aggregation_rule:
    always_use: "UTC"
    never_use: "Local market timezone for aggregation"

Multi-Day Bar Handling (Edge Case)

Very rare for range bars, but handle gracefully:

def detect_multi_day_bars(timestamps: np.ndarray) -> dict:
    """Check for bars spanning multiple days.

    Only possible with very wide thresholds + low volatility.
    """
    df = pd.DataFrame({
        "ts": pd.to_datetime(timestamps, utc=True)
    })
    df["date"] = df["ts"].dt.date
    df["prev_date"] = df["date"].shift(1)

    # Bar duration in days
    df["duration_days"] = (df["ts"] - df["ts"].shift(1)).dt.total_seconds() / 86400

    multi_day = df[df["duration_days"] > 1]

    return {
        "n_multi_day": len(multi_day),
        "max_duration_days": float(df["duration_days"].max()),
        "has_multi_day": len(multi_day) > 0
    }

Validation Checklist

validation_checklist:
  - id: UTC_EXPLICIT
    check: "All pd.to_datetime() calls include utc=True"
    example: "pd.to_datetime(timestamps, utc=True)"

  - id: DAILY_BEFORE_SHARPE
    check: "Daily aggregation happens before Sharpe computation"
    rationale: "Raw bar-level Sharpe violates IID"

  - id: FILTER_THEN_AGGREGATE
    check: "Session filter applied before aggregation"
    order: "filter → aggregate → compute"

  - id: DUAL_VIEW_INDEPENDENT
    check: "Both views computed independently"
    not: "Filtering the all-bars view"

  - id: NO_LOCAL_TIMEZONE
    check: "No hardcoded local timezone assumptions"
    anti_pattern: "tz.localize(ts)"

Worked Examples

End-to-end examples for BTC/Binance and EUR/USD/EXNESS range bar data.

Example 1: BTC/USDT Range Bars (Binance)

Data Specification

Step 1: Load and Prepare Data

import numpy as np
import pandas as pd
from pathlib import Path

# Load range bar data (exported from trading platform)
df = pd.read_parquet("btc_usdt_range_50_2024.parquet")

# Required columns: open, high, low, close, volume, timestamp
print(df.columns)
# ['open', 'high', 'low', 'close', 'volume', 'timestamp']

# Ensure UTC timestamps
df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True)

# Compute returns (close-to-close)
df['return'] = df['close'].pct_change()

# Feature engineering (example: simple momentum features)
df['momentum_5'] = df['close'].rolling(5).mean() / df['close'] - 1
df['momentum_20'] = df['close'].rolling(20).mean() / df['close'] - 1
df['volatility_20'] = df['return'].rolling(20).std()

# Drop NaN rows
df = df.dropna()

print(f"Total bars: {len(df):,}")
print(f"Date range: {df['timestamp'].min()} to {df['timestamp'].max()}")
print(f"Bars per day: {len(df) / 365:.1f}")

Step 2: Train BiLSTM Model

import torch
import torch.nn as nn
from sklearn.preprocessing import StandardScaler
from torch.utils.data import DataLoader, TensorDataset

# Prepare sequences
SEQUENCE_LENGTH = 60
FEATURES = ['momentum_5', 'momentum_20', 'volatility_20']

scaler = StandardScaler()
features_scaled = scaler.fit_transform(df[FEATURES].values)

def create_sequences(data, target, seq_len):
    X, y = [], []
    for i in range(seq_len, len(data)):
        X.append(data[i-seq_len:i])
        y.append(target[i])
    return np.array(X), np.array(y)

X, y = create_sequences(features_scaled, df['return'].values, SEQUENCE_LENGTH)
timestamps = df['timestamp'].values[SEQUENCE_LENGTH:]

print(f"Sequences: {X.shape}")  # (n_samples, 60, 3)

# BiLSTM model (with AWFES-recommended architecture)
class BiLSTM(nn.Module):
    def __init__(self, input_size, hidden_size=48, dropout=0.3):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size, hidden_size,
            bidirectional=True, batch_first=True, dropout=dropout
        )
        self.fc = nn.Linear(hidden_size * 2, 1)

    def forward(self, x):
        out, _ = self.lstm(x)
        return self.fc(out[:, -1, :]).squeeze()

model = BiLSTM(input_size=len(FEATURES), hidden_size=48, dropout=0.3)

Step 3: Walk-Forward Optimization with AWFES

from datetime import timedelta

# WFO parameters
N_FOLDS = 12  # Monthly folds
TRAIN_MONTHS = 6
VAL_MONTHS = 1
TEST_MONTHS = 1
EPOCH_CONFIGS = [400, 800, 1000, 2000]

# Generate fold boundaries
fold_results = []
start_date = df['timestamp'].min()

for fold in range(N_FOLDS):
    # Define boundaries
    train_start = start_date + timedelta(days=30 * fold)
    train_end = train_start + timedelta(days=30 * TRAIN_MONTHS)
    val_end = train_end + timedelta(days=30 * VAL_MONTHS)
    test_end = val_end + timedelta(days=30 * TEST_MONTHS)

    # Split data
    train_mask = (df['timestamp'] >= train_start) & (df['timestamp'] < train_end)
    val_mask = (df['timestamp'] >= train_end) & (df['timestamp'] < val_end)
    test_mask = (df['timestamp'] >= val_end) & (df['timestamp'] < test_end)

    if test_mask.sum() < 100:
        continue

    # Epoch sweep on train → validate
    epoch_metrics = []
    for epochs in EPOCH_CONFIGS:
        model = BiLSTM(input_size=len(FEATURES))
        # ... train model for `epochs` epochs ...
        # ... compute IS and OOS Sharpe ...

        is_sharpe = 1.5  # Placeholder
        val_sharpe = 0.8  # Placeholder
        wfe = val_sharpe / is_sharpe if is_sharpe > 0.1 else None

        epoch_metrics.append({
            "epoch": epochs,
            "is_sharpe": is_sharpe,
            "val_sharpe": val_sharpe,
            "wfe": wfe,
        })

    # Select best epoch (efficient frontier)
    best_epoch = max(
        [m for m in epoch_metrics if m["wfe"]],
        key=lambda m: m["wfe"],
        default={"epoch": 800}
    )["epoch"]

    # Evaluate on test with selected epoch
    # ... retrain with train+val data ...
    # ... predict on test ...

    fold_results.append({
        "fold": fold,
        "selected_epoch": best_epoch,
        "test_predictions": np.random.randn(100),  # Placeholder
        "test_actuals": np.random.randn(100),       # Placeholder
        "test_timestamps": timestamps[test_mask][:100],
    })

print(f"Completed {len(fold_results)} folds")

Step 4: Compute Metrics

import sys
sys.path.insert(0, 'scripts')
from compute_metrics import evaluate_fold, compute_aggregate_metrics

# Per-fold metrics
fold_metrics = []
for result in fold_results:
    metrics = evaluate_fold(
        predictions=result["test_predictions"],
        actuals=result["test_actuals"],
        timestamps=result["test_timestamps"],
        days_per_week=7,        # Crypto: 24/7
        annualization=365,      # Crypto annual
        include_extended=True,  # VaR, Sortino, etc.
    )
    fold_metrics.append(metrics)
    print(f"Fold {result['fold']}: Sharpe={metrics['weekly_sharpe']:.2f}, "
          f"Hit Rate={metrics['hit_rate']:.2%}")

# Aggregate metrics
agg_metrics = compute_aggregate_metrics(fold_metrics)
print("\n=== Aggregate Results ===")
print(f"Mean Weekly Sharpe: {agg_metrics['mean_weekly_sharpe']:.2f}")
print(f"Positive Sharpe Rate: {agg_metrics['positive_sharpe_rate']:.2%}")
print(f"CV Fold Returns: {agg_metrics['cv_fold_returns']:.2f}")
print(f"Binomial p-value: {agg_metrics['binomial_pvalue']:.4f}")
print(f"Effective N: {agg_metrics['effective_n']:.1f}")

Step 5: Apply Decision Criteria

# Go criteria (research)
go_criteria = {
    "positive_sharpe_rate > 0.55": agg_metrics['positive_sharpe_rate'] > 0.55,
    "mean_weekly_sharpe > 0": agg_metrics['mean_weekly_sharpe'] > 0,
    "cv_fold_returns < 1.5": (agg_metrics['cv_fold_returns'] or 0) < 1.5,
    "mean_hit_rate > 0.50": (agg_metrics['mean_hit_rate'] or 0) > 0.50,
}

print("\n=== Go Criteria ===")
for criterion, passed in go_criteria.items():
    status = "PASS" if passed else "FAIL"
    print(f"  {criterion}: {status}")

all_pass = all(go_criteria.values())
print(f"\nOverall: {'GO' if all_pass else 'NO-GO'}")

Expected Output

Total bars: 182,500
Date range: 2024-01-01 00:00:00+00:00 to 2024-12-31 23:59:59+00:00
Bars per day: 500.0
Sequences: (182440, 60, 3)
Completed 12 folds

Fold 0: Sharpe=0.87, Hit Rate=52.30%
Fold 1: Sharpe=1.23, Hit Rate=54.10%
...

=== Aggregate Results ===
Mean Weekly Sharpe: 0.92
Positive Sharpe Rate: 75.00%
CV Fold Returns: 0.89
Binomial p-value: 0.0193
Effective N: 10.8

=== Go Criteria ===
  positive_sharpe_rate > 0.55: PASS
  mean_weekly_sharpe > 0: PASS
  cv_fold_returns < 1.5: PASS
  mean_hit_rate > 0.50: PASS

Overall: GO

---

Example 2: EUR/USD Range Bars (EXNESS)

Data Specification

Key Differences from Crypto

Step 1: Load Data with Session Filter

import pandas as pd
import pytz

# Load EXNESS range bar data
df = pd.read_csv("eurusd_range_10pip_2024.csv", parse_dates=['timestamp'])
df['timestamp'] = df['timestamp'].dt.tz_localize('UTC')

# Session filter: London 08:00 to NY 17:00 UTC
def is_in_session(ts):
    """Check if timestamp is within London-NY session."""
    hour = ts.hour
    weekday = ts.weekday()

    # Skip weekends
    if weekday >= 5:
        return False

    # London open (08:00) to NY close (17:00 UTC)
    # Note: Adjust for DST if needed
    return 8 <= hour < 17

df['in_session'] = df['timestamp'].apply(is_in_session)
df_session = df[df['in_session']].copy()

print(f"Total bars: {len(df):,}")
print(f"Session bars: {len(df_session):,} ({len(df_session)/len(df)*100:.1f}%)")

Step 2: Compute Metrics with Equity Annualization

from compute_metrics import evaluate_fold, compute_aggregate_metrics

# CRITICAL: Use sqrt(5) for session-filtered FX data
metrics = evaluate_fold(
    predictions=predictions,
    actuals=actuals,
    timestamps=timestamps,
    days_per_week=5,         # EQUITY: sqrt(5) scaling
    annualization=252,       # EQUITY: 252 trading days
    include_extended=True,
)

print(f"Weekly Sharpe (sqrt(5)): {metrics['weekly_sharpe']:.2f}")
print(f"Sortino Ratio (252 days): {metrics['sortino_ratio']:.2f}")

Step 3: Compare Filtered vs Unfiltered

# Compute metrics on BOTH views
metrics_session = evaluate_fold(
    predictions_session, actuals_session, timestamps_session,
    days_per_week=5, annualization=252  # Session-filtered
)

metrics_all = evaluate_fold(
    predictions_all, actuals_all, timestamps_all,
    days_per_week=7, annualization=365  # All bars (for comparison)
)

print("=== Dual-View Comparison ===")
print(f"Session-filtered (sqrt(5)): Sharpe = {metrics_session['weekly_sharpe']:.2f}")
print(f"All-bars (sqrt(7)):         Sharpe = {metrics_all['weekly_sharpe']:.2f}")
print(f"Ratio: {metrics_all['weekly_sharpe'] / metrics_session['weekly_sharpe']:.2f}x")
# Note: Using sqrt(7) on session-filtered data overstates Sharpe by ~18%

Step 4: Transaction Cost Analysis

# EXNESS typical costs
COST_BPS = 3.0  # 3 bps round-trip (maker + spread)

# Compute with costs
from compute_metrics import _group_by_day

def compute_sharpe_with_costs(pnl, timestamps, costs, days_per_week=5):
    net_pnl = pnl + costs  # costs are negative
    daily_pnl = _group_by_day(net_pnl, timestamps)

    if len(daily_pnl) < 2 or np.std(daily_pnl) < 1e-10:
        return 0.0

    return float(np.mean(daily_pnl) / np.std(daily_pnl) * np.sqrt(days_per_week))

# Example: 20% monthly turnover
position_changes = np.abs(np.diff(predictions, prepend=0))
costs = -position_changes * (COST_BPS / 10000)

gross_sharpe = metrics_session['weekly_sharpe']
net_sharpe = compute_sharpe_with_costs(pnl, timestamps, costs, days_per_week=5)

print(f"Gross Sharpe: {gross_sharpe:.2f}")
print(f"Net Sharpe (3 bps): {net_sharpe:.2f}")
print(f"Cost haircut: {(gross_sharpe - net_sharpe) / gross_sharpe * 100:.1f}%")

---

Example 3: Edge Cases and Validation

Edge Case 1: Model Collapse

# Simulate collapsed model (constant predictions)
collapsed_predictions = np.full(1000, 0.001)
actuals = np.random.randn(1000) * 0.01
timestamps = pd.date_range('2024-01-01', periods=1000, freq='h', tz='UTC')

metrics = evaluate_fold(collapsed_predictions, actuals, timestamps)

print("=== Model Collapse Detection ===")
print(f"is_collapsed: {metrics['is_collapsed']}")
print(f"prediction_autocorr: {metrics['prediction_autocorr']:.4f}")
print(f"weekly_sharpe: {metrics['weekly_sharpe']:.4f}")
# Expected: is_collapsed=True, autocorr=1.0, sharpe≈0

Edge Case 2: Insufficient Data

# Only 3 days of data
short_pnl = np.random.randn(10) * 0.01
short_timestamps = pd.date_range('2024-01-01', periods=10, freq='h', tz='UTC')

metrics = evaluate_fold(
    np.random.randn(10), np.random.randn(10), short_timestamps
)

print("=== Insufficient Data ===")
print(f"omega_ratio: {metrics.get('omega_ratio')}")  # Should be None
print(f"profit_factor: {metrics.get('profit_factor')}")  # Should be None
print(f"n_days: {metrics['n_days']}")  # Should be < 5

Edge Case 3: All Positive Returns

# Unrealistic: all trades profitable
all_positive_pnl = np.abs(np.random.randn(500) * 0.01)
timestamps = pd.date_range('2024-01-01', periods=500, freq='h', tz='UTC')

metrics = evaluate_fold(all_positive_pnl, all_positive_pnl, timestamps)

print("=== All Positive Returns (Suspicious) ===")
print(f"hit_rate: {metrics['hit_rate']:.2%}")  # 100%
print(f"profit_factor: {metrics['profit_factor']}")  # inf
print(f"max_drawdown: {metrics['max_drawdown']:.4f}")  # 0 or near-0
# Note: This should trigger review - likely data error or look-ahead bias

Edge Case 4: Session Boundary Handling

# Data spanning DST transition
df_dst = pd.DataFrame({
    'timestamp': pd.date_range('2024-03-09', '2024-03-12', freq='h', tz='UTC'),
    'pnl': np.random.randn(72) * 0.01,
})

# Check session detection around DST
for ts in df_dst['timestamp']:
    in_session = is_in_session(ts)
    print(f"{ts}: {'IN' if in_session else 'OUT'}")

---

Quick Reference: Metric Thresholds

Go Criteria (Research Phase)

go_criteria:
  positive_sharpe_rate: "> 0.55"
  mean_weekly_sharpe: "> 0"
  cv_fold_returns: "< 1.5"
  mean_hit_rate: "> 0.50"

Publication Criteria

publication_criteria:
  binomial_pvalue: "< 0.05"
  psr: "> 0.85"
  dsr: "> 0.50" # If n_trials > 1
  effective_n: ">= 30"

Red Flags

red_flags:
  is_collapsed: true
  prediction_autocorr: "> 0.95"
  hit_rate: "= 1.00" # Suspicious
  profit_factor: "= inf" # Suspicious
  max_drawdown: "= 0" # Suspicious

---

Summary