From e0d178ae32a742625f734722d1a812d5a4db7ac1 Mon Sep 17 00:00:00 2001 From: Lautaro Parada <52385097+LautaroParada@users.noreply.github.com> Date: Wed, 8 Apr 2026 20:03:36 -0400 Subject: [PATCH 1/6] Add Sprint 4 rolling battery support and CI/examples --- .github/workflows/ci.yml | 36 +++++ README.md | 46 +++++- examples/basic_battery.py | 17 +++ examples/rolling_battery.py | 28 ++++ examples/variance_ratio_only.py | 21 +++ src/variance_test/battery.py | 23 ++- src/variance_test/rolling.py | 199 ++++++++++++++++++++++++ tests/test_rolling.py | 258 ++++++++++++++++++++++++++++++++ 8 files changed, 622 insertions(+), 6 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100755 examples/basic_battery.py create mode 100755 examples/rolling_battery.py create mode 100755 examples/variance_ratio_only.py create mode 100644 src/variance_test/rolling.py create mode 100644 tests/test_rolling.py 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..9f0f42a 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: @@ -102,7 +103,13 @@ 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) @@ -284,7 +291,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 +302,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 +357,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/rolling.py b/src/variance_test/rolling.py new file mode 100644 index 0000000..6258c2f --- /dev/null +++ b/src/variance_test/rolling.py @@ -0,0 +1,199 @@ +"""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: + lb_stat, lb_pvalue = diagnostic.acorr_ljungbox( + series, + lags=list(config.ljung_box_lags), + return_df=False, + ) + + selected_lag = int(config.ljung_box_lags[0]) + selected_p = float(lb_pvalue[0]) + selected_stat = float(lb_stat[0]) + + for lag, stat_value, p_value in zip(config.ljung_box_lags, lb_stat, lb_pvalue): + lag_int = int(lag) + stat_float = float(stat_value) + p_float = float(p_value) + 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 + selected_stat = stat_float + + outcomes.append( + { + "start": start, + "end": end, + "statistic": selected_stat, + "p_value": 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) From 8824d940d1c92bd35596b6834811f84729a47454 Mon Sep 17 00:00:00 2001 From: Lautaro Parada Date: Thu, 9 Apr 2026 09:10:28 -0400 Subject: [PATCH 2/6] Handle non-computable VR tests and fix v_hat checks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wrap variance-ratio computation in battery to catch ValueError from emh.vrt and record a TestOutcome with None statistic/p_value, a reason in metadata, and a warning. Update Holm–Bonferroni helper to exclude non-computable (None) p-values from the step-down ordering while keeping them in the family summary and initialize thresholds to None. In core EMH: raise on non-positive one-period variance to avoid invalid ratios, lower the q minimum check for the heteroskedastic variance estimator from 3 to 2, and correct the loop bounds so j runs up to q-1 when accumulating v_hat. These changes improve robustness for edge cases and small q values. --- src/variance_test/battery.py | 93 +++++++++++++++++++++++------------- src/variance_test/core.py | 10 ++-- 2 files changed, 68 insertions(+), 35 deletions(-) diff --git a/src/variance_test/battery.py b/src/variance_test/battery.py index 9f0f42a..464ac3b 100644 --- a/src/variance_test/battery.py +++ b/src/variance_test/battery.py @@ -33,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 diff --git a/src/variance_test/core.py b/src/variance_test/core.py index 7bdbb14..6ac7a48 100644 --- a/src/variance_test/core.py +++ b/src/variance_test/core.py @@ -121,6 +121,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 +196,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 +212,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 From 073c2ab7e3108d63c4b4aa2592cde36a094a0dca Mon Sep 17 00:00:00 2001 From: Lautaro Parada Date: Thu, 9 Apr 2026 14:20:06 -0400 Subject: [PATCH 3/6] Make VRT homoskedastic; handle non-finite stats Change variance ratio tests to use heteroskedastic=False in both battery and rolling codepaths. Refactor Ljung-Box handling to call acorr_ljungbox(return_df=True), convert results to numeric arrays, ignore non-finite lag results, and return a clear non-computable outcome or window warning when no finite statistics exist. Select the best lag by minimum p-value (tie-breaker: smaller lag) and populate per-lag metadata only for finite entries. Add non-finite checks to ARCH LM (het_arch) and return a non-computable TestOutcome with warnings when statistics are not finite; ensure numeric values are cast to floats in returned metadata. --- src/variance_test/battery.py | 107 ++++++++++++++++++++++++----------- src/variance_test/rolling.py | 45 +++++++++------ 2 files changed, 101 insertions(+), 51 deletions(-) diff --git a/src/variance_test/battery.py b/src/variance_test/battery.py index 464ac3b..ace152e 100644 --- a/src/variance_test/battery.py +++ b/src/variance_test/battery.py @@ -39,7 +39,7 @@ def _run_variance_ratio_family(normalized, config: BatteryConfig) -> dict[str, T z_score, p_value = emh.vrt( X=normalized.log_prices, q=q, - heteroskedastic=True, + heteroskedastic=False, centered=True, unbiased=True, annualize=False, @@ -56,7 +56,7 @@ def _run_variance_ratio_family(normalized, config: BatteryConfig) -> dict[str, T reject_null=bool(p_value < config.alpha), metadata={ "q": q, - "heteroskedastic": True, + "heteroskedastic": False, "centered": True, "unbiased": True, "annualize": False, @@ -76,7 +76,7 @@ def _run_variance_ratio_family(normalized, config: BatteryConfig) -> dict[str, T reject_null=None, metadata={ "q": q, - "heteroskedastic": True, + "heteroskedastic": False, "centered": True, "unbiased": True, "annualize": False, @@ -140,44 +140,60 @@ def _run_ljung_box( 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] @@ -270,23 +286,48 @@ 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.""" - lm_stat, lm_p_value, f_stat, f_p_value = diagnostic.het_arch(returns, nlags=nlags) +def _run_arch_lm(series: np.ndarray, nlags: int, alpha: float) -> TestOutcome: + """Run ARCH LM test on returns.""" + lm_stat, lm_p_value, f_stat, f_p_value = diagnostic.het_arch(series, 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=[], ) diff --git a/src/variance_test/rolling.py b/src/variance_test/rolling.py index 6258c2f..a172ebb 100644 --- a/src/variance_test/rolling.py +++ b/src/variance_test/rolling.py @@ -60,7 +60,7 @@ def _compute_rolling_variance_ratio(normalized, config) -> dict[str, list[dict[s z_score, p_value = emh.vrt( X=window_normalized.log_prices, q=q, - heteroskedastic=True, + heteroskedastic=False, centered=True, unbiased=True, annualize=False, @@ -124,33 +124,42 @@ def _compute_rolling_ljung_box( continue try: - lb_stat, lb_pvalue = diagnostic.acorr_ljungbox( + result = diagnostic.acorr_ljungbox( series, lags=list(config.ljung_box_lags), - return_df=False, + return_df=True, ) - selected_lag = int(config.ljung_box_lags[0]) - selected_p = float(lb_pvalue[0]) - selected_stat = float(lb_stat[0]) + lb_stat = result["lb_stat"].to_numpy(dtype=float) + lb_pvalue = result["lb_pvalue"].to_numpy(dtype=float) - for lag, stat_value, p_value in zip(config.ljung_box_lags, lb_stat, lb_pvalue): - lag_int = int(lag) - stat_float = float(stat_value) - p_float = float(p_value) - 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 - selected_stat = stat_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": selected_stat, - "p_value": selected_p, + "statistic": float(selected_stat), + "p_value": float(selected_p), "reject_null": bool(selected_p < config.alpha), "warning": None, } From 6e8e29befde4d49e378bb1c5d61f2066e8a57541 Mon Sep 17 00:00:00 2001 From: Lautaro Parada Date: Thu, 9 Apr 2026 14:29:58 -0400 Subject: [PATCH 4/6] Scale q-step variance; fix ARCH LM arg name MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Apply Lo–MacKinlay variance-ratio scaling by normalizing the q-step variance estimator with q (sigma_a /= denom_a * q) so the q-step variance is comparable to the 1-step variance under the random-walk null. Also rename the parameter in _run_arch_lm from series to returns, pass returns into diagnostic.het_arch, and add a brief docstring — improving clarity and correctness of the variance tests. --- src/variance_test/battery.py | 4 ++-- src/variance_test/core.py | 5 ++++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/variance_test/battery.py b/src/variance_test/battery.py index ace152e..207acd9 100644 --- a/src/variance_test/battery.py +++ b/src/variance_test/battery.py @@ -286,9 +286,9 @@ def _run_runs_test(normalized, alpha: float) -> TestOutcome: ) -def _run_arch_lm(series: np.ndarray, nlags: int, alpha: float) -> TestOutcome: +def _run_arch_lm(returns: np.ndarray, nlags: int, alpha: float) -> TestOutcome: """Run ARCH LM test on returns.""" - lm_stat, lm_p_value, f_stat, f_p_value = diagnostic.het_arch(series, nlags=nlags) + 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): diff --git a/src/variance_test/core.py b/src/variance_test/core.py index 6ac7a48..dcf0770 100644 --- a/src/variance_test/core.py +++ b/src/variance_test/core.py @@ -57,7 +57,10 @@ def __variance_estimators( if denom_a <= 0: raise ValueError("Degrees of freedom must be positive.") - sigma_a /= denom_a + # Lo-MacKinlay variance-ratio scaling: + # the q-step variance estimator must be normalized by q so it is + # comparable to the 1-step variance estimator under the random walk null. + sigma_a /= (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) From b896d548dde515dd7708bc6d93d04b91e072b49b Mon Sep 17 00:00:00 2001 From: Lautaro Parada Date: Thu, 9 Apr 2026 14:44:44 -0400 Subject: [PATCH 5/6] Enable heteroskedastic VRT in variance tests Use heteroskedastic=True for emh.vrt calls in variance tests to account for conditional heteroskedasticity when computing z-scores and p-values. Updated calls in battery._run_variance_ratio_family and rolling._compute_rolling_variance_ratio, and adjusted the metadata in battery to reflect the heteroskedastic flag. Other test parameters (centered, unbiased, annualize) remain unchanged. --- src/variance_test/battery.py | 4 ++-- src/variance_test/rolling.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/variance_test/battery.py b/src/variance_test/battery.py index 207acd9..fb69b2e 100644 --- a/src/variance_test/battery.py +++ b/src/variance_test/battery.py @@ -39,7 +39,7 @@ def _run_variance_ratio_family(normalized, config: BatteryConfig) -> dict[str, T z_score, p_value = emh.vrt( X=normalized.log_prices, q=q, - heteroskedastic=False, + heteroskedastic=True, centered=True, unbiased=True, annualize=False, @@ -56,7 +56,7 @@ def _run_variance_ratio_family(normalized, config: BatteryConfig) -> dict[str, T reject_null=bool(p_value < config.alpha), metadata={ "q": q, - "heteroskedastic": False, + "heteroskedastic": True, "centered": True, "unbiased": True, "annualize": False, diff --git a/src/variance_test/rolling.py b/src/variance_test/rolling.py index a172ebb..54400ea 100644 --- a/src/variance_test/rolling.py +++ b/src/variance_test/rolling.py @@ -60,7 +60,7 @@ def _compute_rolling_variance_ratio(normalized, config) -> dict[str, list[dict[s z_score, p_value = emh.vrt( X=window_normalized.log_prices, q=q, - heteroskedastic=False, + heteroskedastic=True, centered=True, unbiased=True, annualize=False, From cd4a0ade61844256aefae7cae43794268a4297cb Mon Sep 17 00:00:00 2001 From: Lautaro Parada Date: Thu, 9 Apr 2026 15:22:28 -0400 Subject: [PATCH 6/6] Simplify variance estimators and checks Set battery metadata to heteroskedastic and refactor the EMH variance computation. The battery now marks generated series as heteroskedastic. The EMH variance routine was rewritten to: compute 1-period demeaned increments and overlapping q-period demeaned increments, use dot-product variance estimators with explicit degrees-of-freedom handling (unbiased vs. biased), normalize the q-period variance by q, and add stricter input/DF/positive-variance validation and clearer error messages. Removed the previous weighted-autocovariance loop and other index/upper-bound logic to simplify and fix edge cases when sample sizes are small. --- src/variance_test/battery.py | 2 +- src/variance_test/core.py | 75 ++++++++++++------------------------ 2 files changed, 25 insertions(+), 52 deletions(-) diff --git a/src/variance_test/battery.py b/src/variance_test/battery.py index fb69b2e..5fd3786 100644 --- a/src/variance_test/battery.py +++ b/src/variance_test/battery.py @@ -76,7 +76,7 @@ def _run_variance_ratio_family(normalized, config: BatteryConfig) -> dict[str, T reject_null=None, metadata={ "q": q, - "heteroskedastic": False, + "heteroskedastic": True, "centered": True, "unbiased": True, "annualize": False, diff --git a/src/variance_test/core.py b/src/variance_test/core.py index dcf0770..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,60 +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.") - - # Lo-MacKinlay variance-ratio scaling: - # the q-step variance estimator must be normalized by q so it is - # comparable to the 1-step variance estimator under the random walk null. - sigma_a /= (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 + raise ValueError("Degrees of freedom for q-period variance must be positive.") + + # 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) + + 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))