diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..5b39b76 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,36 @@ +name: CI + +on: + push: + pull_request: + +jobs: + test-and-build: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.11", "3.12"] + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Upgrade pip + run: python -m pip install --upgrade pip + + - name: Install package + run: pip install -e . + + - name: Install test and build tools + run: pip install pytest build + + - name: Run tests + run: pytest -q + + - name: Build package + run: python -m build diff --git a/README.md b/README.md index fdba7b0..e580e7f 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,40 @@ print(outcome.multiple_testing["battery_summary"]) > Warning: this battery provides evidence against compatibility with weak-form efficiency **under this battery**. > It is **not** definitive proof of market inefficiency. > -> Note: `rolling = None` in this version. +Rolling mode is available by setting `BatteryConfig(rolling_window=..., rolling_step=...)`. +In this version, rolling results are provided only for: +- variance ratio multi-q +- Ljung-Box returns +- Ljung-Box squared returns + +Rolling mode is **not** implemented in this version for: +- Holm summary +- runs test +- ARCH LM + +```python +import numpy as np +from variance_test import BatteryConfig, run_weak_form_battery + +rng = np.random.default_rng(123) +returns = rng.normal(0.0, 0.01, size=1200) + +config = BatteryConfig( + input_kind="returns", + q_list=(2, 4), + ljung_box_lags=(5, 10), + rolling_window=120, + rolling_step=20, +) +outcome = run_weak_form_battery(returns, config=config) +print(outcome.rolling["n_windows"]) +``` + +Offline runnable examples are available under `examples/`: +- `examples/basic_battery.py` +- `examples/rolling_battery.py` +- `examples/variance_ratio_only.py` +- `examples/empirical_application.py` ### Running Simulations @@ -237,11 +270,18 @@ variance-test/ │ └── variance_test/ │ ├── __init__.py │ ├── core.py # Core VRT implementation (EMH class) +│ ├── battery.py # Weak-form battery v1 +│ ├── rolling.py # Internal rolling helpers +│ ├── data.py # Input normalization +│ ├── models.py # Dataclass contracts │ ├── price_paths.py # Stochastic process generators │ ├── simulation.py # Simulation framework and CLI │ └── visuals.py # Plotting utilities ├── examples/ -│ └── empirical_application.py # Optional external example (uses external deps) +│ ├── basic_battery.py # Offline battery example (full sample) +│ ├── rolling_battery.py # Offline battery example (rolling mode) +│ ├── variance_ratio_only.py # Offline EMH.vrt usage example +│ └── empirical_application.py # Optional external example ├── tests/ ├── pyproject.toml ├── README.md @@ -296,4 +336,4 @@ If you use this implementation in your research, please cite: url = {https://github.com/LautaroParada/variance-test}, note = {Implementation following Lo \& MacKinlay (1988)} } -``` \ No newline at end of file +``` diff --git a/examples/basic_battery.py b/examples/basic_battery.py new file mode 100755 index 0000000..5736e3b --- /dev/null +++ b/examples/basic_battery.py @@ -0,0 +1,17 @@ +#!/usr/bin/env python3 +"""Basic offline example for the weak-form battery without rolling windows.""" + +from __future__ import annotations + +import numpy as np + +from variance_test import BatteryConfig, run_weak_form_battery + + +rng = np.random.default_rng(42) +returns = rng.normal(0.0, 0.01, size=1500) +config = BatteryConfig(input_kind="returns", q_list=(2, 4, 8), ljung_box_lags=(5, 10, 20)) +outcome = run_weak_form_battery(returns, config=config) + +print("Battery summary:") +print(outcome.multiple_testing["battery_summary"]) diff --git a/examples/rolling_battery.py b/examples/rolling_battery.py new file mode 100755 index 0000000..3da2b4c --- /dev/null +++ b/examples/rolling_battery.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python3 +"""Offline rolling-window example for the weak-form battery.""" + +from __future__ import annotations + +import numpy as np + +from variance_test import BatteryConfig, run_weak_form_battery + + +rng = np.random.default_rng(123) +returns = rng.normal(0.0, 0.01, size=1200) + +config = BatteryConfig( + input_kind="returns", + q_list=(2, 4), + ljung_box_lags=(5, 10), + rolling_window=120, + rolling_step=20, +) +outcome = run_weak_form_battery(returns, config=config) + +rolling = outcome.rolling or {} +vr_q2 = rolling.get("tests", {}).get("variance_ratio_q2", []) + +print(f"n_windows={rolling.get('n_windows')}") +print("first_two_variance_ratio_q2_entries=") +print(vr_q2[:2]) diff --git a/examples/variance_ratio_only.py b/examples/variance_ratio_only.py new file mode 100755 index 0000000..42f8425 --- /dev/null +++ b/examples/variance_ratio_only.py @@ -0,0 +1,21 @@ +#!/usr/bin/env python3 +"""Minimal EMH.vrt usage example for log-prices and returns inputs.""" + +from __future__ import annotations + +import numpy as np + +from variance_test import EMH + + +rng = np.random.default_rng(7) +returns = rng.normal(0.0, 0.01, size=1000) +log_prices = np.concatenate(([0.0], np.cumsum(returns))) + +emh = EMH() + +z_log, p_log = emh.vrt(X=log_prices, q=4, input_kind="log_prices") +z_ret, p_ret = emh.vrt(X=returns, q=4, input_kind="returns") + +print(f"log_prices mode -> z={z_log:.6f}, p={p_log:.6f}") +print(f"returns mode -> z={z_ret:.6f}, p={p_ret:.6f}") diff --git a/src/variance_test/battery.py b/src/variance_test/battery.py index a0d42b8..5fd3786 100644 --- a/src/variance_test/battery.py +++ b/src/variance_test/battery.py @@ -9,6 +9,7 @@ from .core import EMH from .data import normalize_series from .models import BatteryConfig, BatteryOutcome, TestOutcome +from .rolling import _build_rolling_results def _validate_battery_compatibility(normalized, config: BatteryConfig) -> None: @@ -32,48 +33,77 @@ def _run_variance_ratio_family(normalized, config: BatteryConfig) -> dict[str, T outcomes: dict[str, TestOutcome] = {} for q in config.q_list: - z_score, p_value = emh.vrt( - X=normalized.log_prices, - q=q, - heteroskedastic=True, - centered=True, - unbiased=True, - annualize=False, - input_kind="log_prices", - alternative="two-sided", - ) name = f"variance_ratio_q{q}" - outcomes[name] = TestOutcome( - name=name, - null_hypothesis="Variance ratio equals 1 under the random walk null.", - statistic=float(z_score), - p_value=float(p_value), - alpha=config.alpha, - reject_null=bool(p_value < config.alpha), - metadata={ - "q": q, - "heteroskedastic": True, - "centered": True, - "unbiased": True, - "annualize": False, - "input_kind_used": "log_prices", - "alternative": "two-sided", - }, - warnings=[], - ) + + try: + z_score, p_value = emh.vrt( + X=normalized.log_prices, + q=q, + heteroskedastic=True, + centered=True, + unbiased=True, + annualize=False, + input_kind="log_prices", + alternative="two-sided", + ) + + outcomes[name] = TestOutcome( + name=name, + null_hypothesis="Variance ratio equals 1 under the random walk null.", + statistic=float(z_score), + p_value=float(p_value), + alpha=config.alpha, + reject_null=bool(p_value < config.alpha), + metadata={ + "q": q, + "heteroskedastic": True, + "centered": True, + "unbiased": True, + "annualize": False, + "input_kind_used": "log_prices", + "alternative": "two-sided", + }, + warnings=[], + ) + + except ValueError as exc: + outcomes[name] = TestOutcome( + name=name, + null_hypothesis="Variance ratio equals 1 under the random walk null.", + statistic=None, + p_value=None, + alpha=config.alpha, + reject_null=None, + metadata={ + "q": q, + "heteroskedastic": True, + "centered": True, + "unbiased": True, + "annualize": False, + "input_kind_used": "log_prices", + "alternative": "two-sided", + "reason": str(exc), + }, + warnings=[f"Variance ratio not computable for q={q}: {exc}"], + ) return outcomes def _apply_holm_bonferroni(vr_outcomes: list[TestOutcome], alpha: float) -> dict[str, object]: - """Apply Holm-Bonferroni correction to the variance-ratio family.""" + """Apply Holm-Bonferroni correction to the variance-ratio family. + + Non-computable tests (p_value is None) are treated as non-rejections and + are excluded from the step-down ordering, but still appear in the summary. + """ family = [outcome.name for outcome in vr_outcomes] - raw_p_values = {outcome.name: float(outcome.p_value) for outcome in vr_outcomes} + raw_p_values = {outcome.name: outcome.p_value for outcome in vr_outcomes} - sorted_outcomes = sorted(vr_outcomes, key=lambda item: (float(item.p_value), item.name)) + computable = [outcome for outcome in vr_outcomes if outcome.p_value is not None] + sorted_outcomes = sorted(computable, key=lambda item: (float(item.p_value), item.name)) m = len(sorted_outcomes) - thresholds: dict[str, float] = {} + thresholds: dict[str, float | None] = {name: None for name in family} rejections: dict[str, bool] = {name: False for name in family} stop = False @@ -102,46 +132,68 @@ def _apply_holm_bonferroni(vr_outcomes: list[TestOutcome], alpha: float) -> dict } -def _run_ljung_box(series: np.ndarray, lags: tuple[int, ...], alpha: float, name: str, null_hypothesis: str) -> TestOutcome: +def _run_ljung_box( + series: np.ndarray, + lags: tuple[int, ...], + alpha: float, + name: str, + null_hypothesis: str, +) -> TestOutcome: """Run Ljung-Box test and select the lag with minimum p-value.""" - lb_stat, lb_pvalue = diagnostic.acorr_ljungbox(series, lags=list(lags), return_df=False) + result = diagnostic.acorr_ljungbox(series, lags=list(lags), return_df=True) + + lb_stat = result["lb_stat"].to_numpy(dtype=float) + lb_pvalue = result["lb_pvalue"].to_numpy(dtype=float) per_lag: dict[int, dict[str, float | bool]] = {} - selected_lag = int(lags[0]) - selected_p = float(lb_pvalue[0]) - - for lag, stat_value, p_value in zip(lags, lb_stat, lb_pvalue): - lag_int = int(lag) - stat_float = float(stat_value) - p_float = float(p_value) - per_lag[lag_int] = { - "statistic": stat_float, - "p_value": p_float, - "reject_null": bool(p_float < alpha), - } - if p_float < selected_p or (np.isclose(p_float, selected_p) and lag_int < selected_lag): - selected_lag = lag_int - selected_p = p_float + finite_pairs = [ + (int(lag), float(stat_value), float(p_value)) + for lag, stat_value, p_value in zip(lags, lb_stat, lb_pvalue) + if np.isfinite(stat_value) and np.isfinite(p_value) + ] + + if not finite_pairs: + return TestOutcome( + name=name, + null_hypothesis=null_hypothesis, + statistic=None, + p_value=None, + alpha=alpha, + reject_null=None, + metadata={ + "tested_lags": list(lags), + "selected_lag": None, + "per_lag": {}, + "reason": "Ljung-Box statistics are not finite for this series.", + }, + warnings=[f"{name} not computable: Ljung-Box statistics are not finite."], + ) + + selected_lag, selected_stat, selected_p = min(finite_pairs, key=lambda x: (x[2], x[0])) - selected_stat = float(per_lag[selected_lag]["statistic"]) + for lag, stat_value, p_value in finite_pairs: + per_lag[int(lag)] = { + "statistic": float(stat_value), + "p_value": float(p_value), + "reject_null": bool(p_value < alpha), + } return TestOutcome( name=name, null_hypothesis=null_hypothesis, - statistic=selected_stat, - p_value=selected_p, + statistic=float(selected_stat), + p_value=float(selected_p), alpha=alpha, reject_null=bool(selected_p < alpha), metadata={ "tested_lags": list(lags), - "selected_lag": selected_lag, + "selected_lag": int(selected_lag), "per_lag": per_lag, }, warnings=[], ) - def _run_runs_test(normalized, alpha: float) -> TestOutcome: """Run bilateral Wald-Wolfowitz runs test over non-zero return signs.""" non_zero_returns = normalized.returns[normalized.returns != 0.0] @@ -235,22 +287,47 @@ def _run_runs_test(normalized, alpha: float) -> TestOutcome: def _run_arch_lm(returns: np.ndarray, nlags: int, alpha: float) -> TestOutcome: - """Run ARCH LM test and return a standardized TestOutcome.""" + """Run ARCH LM test on returns.""" lm_stat, lm_p_value, f_stat, f_p_value = diagnostic.het_arch(returns, nlags=nlags) + values = [lm_stat, lm_p_value, f_stat, f_p_value] + if not all(np.isfinite(value) for value in values): + return TestOutcome( + name="arch_lm", + null_hypothesis="No ARCH effects are present up to the tested lag order.", + statistic=None, + p_value=None, + alpha=alpha, + reject_null=None, + metadata={ + "nlags": nlags, + "lm_stat": None, + "lm_p_value": None, + "f_stat": None, + "f_p_value": None, + "reason": "ARCH LM statistics are not finite for this series.", + }, + warnings=["arch_lm not computable: ARCH LM statistics are not finite."], + ) + + lm_stat = float(lm_stat) + lm_p_value = float(lm_p_value) + f_stat = float(f_stat) + f_p_value = float(f_p_value) + return TestOutcome( name="arch_lm", null_hypothesis="No ARCH effects are present up to the tested lag order.", - statistic=float(lm_stat), - p_value=float(lm_p_value), + statistic=lm_stat, + p_value=lm_p_value, alpha=alpha, reject_null=bool(lm_p_value < alpha), metadata={ "nlags": nlags, - "lm_stat": float(lm_stat), - "lm_p_value": float(lm_p_value), - "f_stat": float(f_stat), - "f_p_value": float(f_p_value), + "lm_stat": lm_stat, + "lm_p_value": lm_p_value, + "f_stat": f_stat, + "f_p_value": f_p_value, }, warnings=[], ) @@ -284,7 +361,10 @@ def _build_battery_summary(tests: dict[str, TestOutcome]) -> dict[str, object]: } -def run_weak_form_battery(series, config: BatteryConfig | None = None) -> BatteryOutcome: +def run_weak_form_battery( + series, + config: BatteryConfig | None = None, +) -> BatteryOutcome: """Run the weak-form efficiency battery v1 and return structured outcomes.""" if config is None: config = BatteryConfig() @@ -292,6 +372,9 @@ def run_weak_form_battery(series, config: BatteryConfig | None = None) -> Batter normalized = normalize_series(series, input_kind=config.input_kind) _validate_battery_compatibility(normalized, config) + if config.rolling_window is not None and config.rolling_window > normalized.n_raw: + raise ValueError("config.rolling_window must satisfy <= normalized.n_raw.") + tests: dict[str, TestOutcome] = {} vr_family = _run_variance_ratio_family(normalized, config) @@ -344,13 +427,17 @@ def run_weak_form_battery(series, config: BatteryConfig | None = None) -> Batter for outcome in tests.values(): warnings.extend(outcome.warnings) + rolling = None + if config.rolling_window is not None: + rolling = _build_rolling_results(normalized=normalized, config=config) + return BatteryOutcome( input_kind=config.input_kind, n_obs=normalized.n_raw, returns_n_obs=normalized.n_returns, tests=tests, multiple_testing=multiple_testing, - rolling=None, + rolling=rolling, warnings=warnings, ) diff --git a/src/variance_test/core.py b/src/variance_test/core.py index 7bdbb14..8cc4aba 100644 --- a/src/variance_test/core.py +++ b/src/variance_test/core.py @@ -28,7 +28,7 @@ def __variance_estimators( unbiased: bool = True, annualize: bool = True, ) -> tuple[float, float]: - """Compute both volatility estimators in a single pass for efficiency.""" + """Compute q-period and 1-period variance estimators for the variance-ratio test.""" if q <= 0: raise ValueError("Aggregation horizon q must be a positive integer.") @@ -40,57 +40,33 @@ def __variance_estimators( mu_est = (series[-1] - series[0]) / n_obs - max_index = n_obs - 1 - num_increments = max_index // q - if num_increments < 1: - raise ValueError("Not enough data to compute aggregated first differences.") + # 1-period demeaned increments + one_period_diffs = series[1:] - series[:-1] - mu_est + n_one = one_period_diffs.size + if n_one < 1: + raise ValueError("Not enough one-period increments to estimate variance.") - upper_bound = int(np.floor(n_obs / q)) * q - if upper_bound <= q: - raise ValueError("At least two aggregated periods are required to estimate variance.") + denom_b = (n_one - 1) if unbiased else n_one + if denom_b <= 0: + raise ValueError("Degrees of freedom for one-period variance must be positive.") + + sigma_b = float(np.dot(one_period_diffs, one_period_diffs)) / denom_b - increments = series[q : (num_increments + 1) * q : q] - series[: num_increments * q : q] - increments = increments - (q * mu_est) - sigma_a = float(np.dot(increments, increments)) + # Overlapping q-period demeaned increments + q_period_diffs = series[q:] - series[:-q] - (q * mu_est) + n_q = q_period_diffs.size + if n_q < 1: + raise ValueError("Not enough q-period increments to estimate variance.") - denom_a = num_increments - 1 if unbiased else num_increments + denom_a = (n_q - 1) if unbiased else n_q if denom_a <= 0: - raise ValueError("Degrees of freedom must be positive.") + raise ValueError("Degrees of freedom for q-period variance must be positive.") - sigma_a /= denom_a + # Normalize by q so it is comparable to the 1-period variance under the null + sigma_a = float(np.dot(q_period_diffs, q_period_diffs)) / (denom_a * q) - # Calculate sigma_b using weighted autocovariances (Lo & MacKinlay 1988, eq. 6a) - # sigma_b(q) = sum_{j=0}^{q-1} [2(q-j)/q] * gamma(j) - # where gamma(j) is the j-th lag autocovariance of 1-period returns - # Use observations from 0 to upper_bound (inclusive), giving upper_bound one-period differences - # Edge case: when upper_bound = n_obs, we can only access up to n_obs-1 - if upper_bound < n_obs: - one_period_diffs = series[1:upper_bound + 1] - series[:upper_bound] - mu_est - else: - # When upper_bound = n_obs, we can only get (n_obs - 1) differences - one_period_diffs = series[1:n_obs] - series[:n_obs - 1] - mu_est - n_diffs = len(one_period_diffs) - - sigma_b = 0.0 - for j in range(q): - weight = (2 * (q - j) / q) if j > 0 else 1.0 - - if j == 0: - # Variance (lag-0 autocovariance) - if unbiased: - gamma_j = float(np.dot(one_period_diffs, one_period_diffs)) / (n_diffs - 1) - else: - gamma_j = float(np.dot(one_period_diffs, one_period_diffs)) / n_diffs - else: - # Autocovariance at lag j - lead = one_period_diffs[j:] - lagged = one_period_diffs[:-j] - if unbiased: - gamma_j = float(np.dot(lead, lagged)) / (n_diffs - j) - else: - gamma_j = float(np.dot(lead, lagged)) / n_diffs - - sigma_b += weight * gamma_j + if sigma_a <= 0 or sigma_b <= 0: + raise ValueError("Variance estimates must be positive.") if annualize: sigma_a = float(np.sqrt(sigma_a * 252)) @@ -121,6 +97,10 @@ def __mr(self, X, q: int, unbiased: bool = True, annualize: bool = True): vol_a, vol_b = self.__variance_estimators( X, q=q, unbiased=unbiased, annualize=annualize ) + + if vol_b <= 0: + raise ValueError("One-period variance estimate must be positive.") + return (vol_a / vol_b) - 1 def __h1(self, X, q: int, centered: bool = True, unbiased: bool = True, annualize: bool = True): @@ -192,8 +172,8 @@ def __delta_from_diffs( def __v_hat(self, X, q: int): """Asymptotic variance of the centered ratio under heteroskedasticity.""" - if q < 3: - raise ValueError("q must be at least 3 for the heteroskedastic variance estimator.") + if q < 2: + raise ValueError("q must be at least 2 for the heteroskedastic variance estimator.") prices = np.asarray(X, dtype=float) n = int(np.floor(prices.shape[0] / q)) @@ -208,7 +188,7 @@ def __v_hat(self, X, q: int): raise ValueError("Second moment of the differenced series is non-positive.") v_hat = 0.0 - for j in range(1, q - 1): + for j in range(1, q): delta = self.__delta_from_diffs(diffs, denominator, upper_bound, j) weight = (2 * (q - j) / q) ** 2 v_hat += weight * delta diff --git a/src/variance_test/rolling.py b/src/variance_test/rolling.py new file mode 100644 index 0000000..54400ea --- /dev/null +++ b/src/variance_test/rolling.py @@ -0,0 +1,208 @@ +"""Internal rolling-window helpers for weak-form battery tests.""" + +from __future__ import annotations + +from collections.abc import Iterator + +import numpy as np +from statsmodels.stats import diagnostic + +from .core import EMH +from .data import normalize_series + + +def _iter_rolling_windows(n_raw: int, window: int, step: int) -> Iterator[tuple[int, int]]: + """Yield ``(start, end)`` index pairs for rolling windows over raw input.""" + for start in range(0, n_raw - window + 1, step): + yield start, start + window + + +def _non_computable_result(start: int, end: int, warning: str) -> dict[str, object]: + """Build a standardized rolling record for non-computable windows.""" + return { + "start": start, + "end": end, + "statistic": None, + "p_value": None, + "reject_null": None, + "warning": warning, + } + + +def _compute_rolling_variance_ratio(normalized, config) -> dict[str, list[dict[str, object]]]: + """Compute rolling variance ratio outcomes for every q in the config list.""" + emh = EMH() + per_test: dict[str, list[dict[str, object]]] = { + f"variance_ratio_q{q}": [] for q in config.q_list + } + + for start, end in _iter_rolling_windows( + n_raw=normalized.n_raw, + window=config.rolling_window, + step=config.rolling_step, + ): + window_raw = normalized.raw[start:end] + window_normalized = normalize_series(window_raw, input_kind=config.input_kind) + + for q in config.q_list: + key = f"variance_ratio_q{q}" + if q >= window_normalized.log_prices.size: + per_test[key].append( + _non_computable_result( + start=start, + end=end, + warning="Variance ratio not computable for this window: q must be smaller than window log-price length.", + ) + ) + continue + + try: + z_score, p_value = emh.vrt( + X=window_normalized.log_prices, + q=q, + heteroskedastic=True, + centered=True, + unbiased=True, + annualize=False, + input_kind="log_prices", + alternative="two-sided", + ) + p_float = float(p_value) + per_test[key].append( + { + "start": start, + "end": end, + "statistic": float(z_score), + "p_value": p_float, + "reject_null": bool(p_float < config.alpha), + "warning": None, + } + ) + except Exception as exc: # pragma: no cover - defensive by contract + per_test[key].append( + _non_computable_result( + start=start, + end=end, + warning=f"Variance ratio failed for this window: {exc}", + ) + ) + + return per_test + + +def _compute_rolling_ljung_box( + normalized, + config, + *, + field: str, + name: str, +) -> list[dict[str, object]]: + """Compute rolling Ljung-Box outcomes for one normalized series field.""" + outcomes: list[dict[str, object]] = [] + max_lag = max(config.ljung_box_lags) + + for start, end in _iter_rolling_windows( + n_raw=normalized.n_raw, + window=config.rolling_window, + step=config.rolling_step, + ): + window_raw = normalized.raw[start:end] + window_normalized = normalize_series(window_raw, input_kind=config.input_kind) + series = getattr(window_normalized, field) + + if max_lag >= window_normalized.n_returns: + outcomes.append( + _non_computable_result( + start=start, + end=end, + warning=( + f"{name} not computable for this window: " + "max(config.ljung_box_lags) must be smaller than window n_returns." + ), + ) + ) + continue + + try: + result = diagnostic.acorr_ljungbox( + series, + lags=list(config.ljung_box_lags), + return_df=True, + ) + + lb_stat = result["lb_stat"].to_numpy(dtype=float) + lb_pvalue = result["lb_pvalue"].to_numpy(dtype=float) + + finite_pairs = [ + (int(lag), float(stat_value), float(p_value)) + for lag, stat_value, p_value in zip(config.ljung_box_lags, lb_stat, lb_pvalue) + if np.isfinite(stat_value) and np.isfinite(p_value) + ] + + if not finite_pairs: + outcomes.append( + _non_computable_result( + start=start, + end=end, + warning=f"{name} not computable for this window: Ljung-Box statistics are not finite.", + ) + ) + continue + + selected_lag, selected_stat, selected_p = min( + finite_pairs, + key=lambda x: (x[2], x[0]), + ) + + outcomes.append( + { + "start": start, + "end": end, + "statistic": float(selected_stat), + "p_value": float(selected_p), + "reject_null": bool(selected_p < config.alpha), + "warning": None, + } + ) + except Exception as exc: # pragma: no cover - defensive by contract + outcomes.append( + _non_computable_result( + start=start, + end=end, + warning=f"{name} failed for this window: {exc}", + ) + ) + + return outcomes + + +def _build_rolling_results(normalized, config) -> dict[str, object]: + """Build rolling payload for supported subtests in battery v1.""" + vr_results = _compute_rolling_variance_ratio(normalized=normalized, config=config) + lb_returns = _compute_rolling_ljung_box( + normalized=normalized, + config=config, + field="returns", + name="ljung_box_returns", + ) + lb_sq_returns = _compute_rolling_ljung_box( + normalized=normalized, + config=config, + field="squared_returns", + name="ljung_box_squared_returns", + ) + + tests: dict[str, list[dict[str, object]]] = { + **vr_results, + "ljung_box_returns": lb_returns, + "ljung_box_squared_returns": lb_sq_returns, + } + + n_windows = len(next(iter(tests.values()))) if tests else 0 + + return { + "window": config.rolling_window, + "step": config.rolling_step, + "n_windows": n_windows, + "tests": tests, + } diff --git a/tests/test_rolling.py b/tests/test_rolling.py new file mode 100644 index 0000000..5363706 --- /dev/null +++ b/tests/test_rolling.py @@ -0,0 +1,258 @@ +"""Rolling-window tests for weak-form battery (Sprint 4).""" + +from __future__ import annotations + +import numpy as np +import pytest + +from variance_test import BatteryConfig, run_weak_form_battery + + +def _log_prices(seed: int, n_obs: int) -> np.ndarray: + rng = np.random.default_rng(seed) + returns = rng.normal(0.0, 0.01, size=n_obs - 1) + return np.concatenate(([0.0], np.cumsum(returns))) + + +def _returns(seed: int, n_obs: int) -> np.ndarray: + rng = np.random.default_rng(seed) + return rng.normal(0.0, 0.01, size=n_obs) + + +def _rolling_config(**kwargs) -> BatteryConfig: + base = { + "input_kind": "log_prices", + "q_list": (2, 4), + "ljung_box_lags": (5, 10), + "arch_lm_lags": 5, + "runs_test": True, + "rolling_window": 120, + "rolling_step": 1, + } + base.update(kwargs) + return BatteryConfig(**base) + + +def test_01_rolling_none_when_disabled() -> None: + series = _log_prices(seed=1, n_obs=400) + config = _rolling_config(rolling_window=None) + outcome = run_weak_form_battery(series, config=config) + assert outcome.rolling is None + + +def test_02_rolling_present_when_enabled() -> None: + series = _log_prices(seed=2, n_obs=400) + outcome = run_weak_form_battery(series, config=_rolling_config(rolling_window=100, rolling_step=1)) + assert outcome.rolling is not None + + +def test_03_rolling_top_level_keys_exact() -> None: + series = _log_prices(seed=3, n_obs=400) + rolling = run_weak_form_battery(series, config=_rolling_config()).rolling + assert rolling is not None + assert set(rolling.keys()) == {"window", "step", "n_windows", "tests"} + + +def test_04_rolling_test_keys_exact() -> None: + series = _log_prices(seed=4, n_obs=400) + rolling = run_weak_form_battery(series, config=_rolling_config(q_list=(2, 4))).rolling + assert rolling is not None + assert set(rolling["tests"].keys()) == { + "variance_ratio_q2", + "variance_ratio_q4", + "ljung_box_returns", + "ljung_box_squared_returns", + } + + +def test_05_n_windows_formula_matches() -> None: + n_obs = 501 + rolling_window = 100 + rolling_step = 7 + series = _log_prices(seed=5, n_obs=n_obs) + rolling = run_weak_form_battery( + series, + config=_rolling_config(rolling_window=rolling_window, rolling_step=rolling_step), + ).rolling + assert rolling is not None + expected = ((n_obs - rolling_window) // rolling_step) + 1 + assert rolling["n_windows"] == expected + + +def test_06_each_rolling_list_matches_n_windows() -> None: + series = _log_prices(seed=6, n_obs=500) + rolling = run_weak_form_battery(series, config=_rolling_config(rolling_window=120, rolling_step=10)).rolling + assert rolling is not None + for values in rolling["tests"].values(): + assert len(values) == rolling["n_windows"] + + +def test_07_each_rolling_entry_keys_exact() -> None: + series = _log_prices(seed=7, n_obs=420) + rolling = run_weak_form_battery(series, config=_rolling_config(rolling_window=100, rolling_step=13)).rolling + assert rolling is not None + for values in rolling["tests"].values(): + for item in values: + assert set(item.keys()) == { + "start", + "end", + "statistic", + "p_value", + "reject_null", + "warning", + } + + +def test_08_window_indices_monotonic_and_correct() -> None: + n_obs = 501 + rolling_window = 100 + rolling_step = 50 + series = _log_prices(seed=8, n_obs=n_obs) + rolling = run_weak_form_battery( + series, + config=_rolling_config(rolling_window=rolling_window, rolling_step=rolling_step), + ).rolling + assert rolling is not None + starts = [entry["start"] for entry in rolling["tests"]["variance_ratio_q2"]] + ends = [entry["end"] for entry in rolling["tests"]["variance_ratio_q2"]] + assert starts == sorted(starts) + assert all(curr > prev for prev, curr in zip(starts, starts[1:])) + assert all(end - start == rolling_window for start, end in zip(starts, ends)) + assert all(curr - prev == rolling_step for prev, curr in zip(starts, starts[1:])) + + +def test_09_n_windows_501_100_1() -> None: + series = _log_prices(seed=9, n_obs=501) + rolling = run_weak_form_battery(series, config=_rolling_config(rolling_window=100, rolling_step=1)).rolling + assert rolling is not None + assert rolling["n_windows"] == 402 + + +def test_10_n_windows_501_100_50() -> None: + series = _log_prices(seed=10, n_obs=501) + rolling = run_weak_form_battery(series, config=_rolling_config(rolling_window=100, rolling_step=50)).rolling + assert rolling is not None + assert rolling["n_windows"] == 9 + + +def test_11_rolling_window_greater_than_n_obs_raises() -> None: + series = _log_prices(seed=11, n_obs=100) + with pytest.raises(ValueError): + run_weak_form_battery(series, config=_rolling_config(rolling_window=101)) + + +def test_12_rolling_window_equal_n_obs_has_one_window() -> None: + n_obs = 250 + series = _log_prices(seed=12, n_obs=n_obs) + rolling = run_weak_form_battery(series, config=_rolling_config(rolling_window=n_obs, rolling_step=1)).rolling + assert rolling is not None + assert rolling["n_windows"] == 1 + + +def test_13_small_window_for_q_is_non_computable_not_global_error() -> None: + series = _log_prices(seed=13, n_obs=400) + rolling = run_weak_form_battery( + series, + config=_rolling_config(q_list=(2, 50), rolling_window=20, rolling_step=5), + ).rolling + assert rolling is not None + entries = rolling["tests"]["variance_ratio_q50"] + assert entries + assert any(item["statistic"] is None for item in entries) + assert any(item["warning"] for item in entries) + + +def test_14_small_window_for_ljung_box_is_non_computable_not_global_error() -> None: + series = _log_prices(seed=14, n_obs=500) + rolling = run_weak_form_battery( + series, + config=_rolling_config(ljung_box_lags=(5, 10), rolling_window=8, rolling_step=2), + ).rolling + assert rolling is not None + entries = rolling["tests"]["ljung_box_returns"] + assert entries + assert all(item["statistic"] is None for item in entries) + assert all(item["warning"] for item in entries) + + +def test_15_full_sample_tests_still_present_when_rolling_active() -> None: + series = _log_prices(seed=15, n_obs=450) + outcome = run_weak_form_battery(series, config=_rolling_config(rolling_window=120)) + assert outcome.tests + assert "variance_ratio_holm" in outcome.tests + assert "arch_lm" in outcome.tests + + +def test_16_multiple_testing_keys_unchanged() -> None: + series = _log_prices(seed=16, n_obs=450) + outcome = run_weak_form_battery(series, config=_rolling_config(rolling_window=100)) + assert set(outcome.multiple_testing.keys()) == {"variance_ratio_holm", "battery_summary"} + + +def test_17_rolling_only_contains_allowed_entries() -> None: + series = _log_prices(seed=17, n_obs=450) + rolling = run_weak_form_battery(series, config=_rolling_config()).rolling + assert rolling is not None + assert "variance_ratio_holm" not in rolling["tests"] + assert "runs_test_signs" not in rolling["tests"] + assert "arch_lm" not in rolling["tests"] + + +def test_18_returns_and_log_prices_rolling_first_vr_window_matches() -> None: + returns = _returns(seed=18, n_obs=600) + log_prices = np.concatenate(([0.0], np.cumsum(returns))) + + out_returns = run_weak_form_battery( + returns, + config=_rolling_config(input_kind="returns", rolling_window=100, q_list=(2,)), + ) + out_log = run_weak_form_battery( + log_prices, + config=_rolling_config(input_kind="log_prices", rolling_window=101, q_list=(2,)), + ) + + assert out_returns.rolling is not None + assert out_log.rolling is not None + assert out_returns.rolling["n_windows"] == out_log.rolling["n_windows"] + + first_returns = out_returns.rolling["tests"]["variance_ratio_q2"][0] + first_log = out_log.rolling["tests"]["variance_ratio_q2"][0] + assert first_returns["statistic"] is not None + assert first_log["statistic"] is not None + assert np.isclose(first_returns["statistic"], first_log["statistic"], atol=1e-10, rtol=1e-8) + + +def test_19_full_sample_results_unchanged_with_rolling_active() -> None: + series = _log_prices(seed=19, n_obs=500) + base = run_weak_form_battery(series, config=_rolling_config(rolling_window=None)) + rolling = run_weak_form_battery(series, config=_rolling_config(rolling_window=120, rolling_step=3)) + + assert rolling.rolling is not None + assert set(base.tests.keys()) == set(rolling.tests.keys()) + for name in base.tests: + assert base.tests[name].statistic == rolling.tests[name].statistic + assert base.tests[name].p_value == rolling.tests[name].p_value + assert base.tests[name].reject_null == rolling.tests[name].reject_null + + assert base.multiple_testing == rolling.multiple_testing + + +def test_20_all_computable_p_values_are_in_unit_interval() -> None: + series = _log_prices(seed=20, n_obs=550) + rolling = run_weak_form_battery(series, config=_rolling_config(rolling_window=120, rolling_step=4)).rolling + assert rolling is not None + for values in rolling["tests"].values(): + for item in values: + if item["p_value"] is not None: + assert 0.0 <= item["p_value"] <= 1.0 + + +def test_21_all_computable_reject_null_are_consistent_with_alpha() -> None: + series = _log_prices(seed=21, n_obs=550) + config = _rolling_config(rolling_window=120, rolling_step=4, alpha=0.05) + rolling = run_weak_form_battery(series, config=config).rolling + assert rolling is not None + for values in rolling["tests"].values(): + for item in values: + if item["p_value"] is not None: + assert item["reject_null"] == (item["p_value"] < config.alpha)