mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-11-15 21:53:58 +08:00
c6d097b271
* fix(technical): use open-to-close in Yang-Zhang The Yang-Zhang estimator combines overnight (close-to-open) variance, open-to-close variance and Rogers-Satchell variance. The close term used close-to-close log returns, which already contain the overnight move, so overnight variance was counted twice and the estimate was inflated by up to sqrt(1 + k). Use log(close / open) as the docstring and the original paper describe. * test(technical): check Yang-Zhang against a hand-computed window --------- Co-authored-by: Ryan Duguid <ryanduguid@users.noreply.github.com> Co-authored-by: Danglewood <85772166+deeleeramone@users.noreply.github.com> Co-authored-by: dlee-bq <dlee@brightquery.com>
246 lines
8.7 KiB
Python
246 lines
8.7 KiB
Python
"""Tests for openbb_technical.helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import warnings
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from openbb_technical.helpers import (
|
|
calculate_cones,
|
|
calculate_fib_levels,
|
|
clenow_momentum,
|
|
garman_klass,
|
|
hodges_tompkins,
|
|
parkinson,
|
|
rogers_satchell,
|
|
standard_deviation,
|
|
validate_data,
|
|
yang_zhang,
|
|
)
|
|
|
|
|
|
class TestValidateData:
|
|
def test_accepts_int_length(self, ohlcv_df):
|
|
validate_data(ohlcv_df["close"].tolist(), 20)
|
|
|
|
def test_accepts_list_of_lengths(self, ohlcv_df):
|
|
validate_data(ohlcv_df["close"].tolist(), [10, 20, 30])
|
|
|
|
def test_raises_when_too_short(self):
|
|
with pytest.raises(ValueError, match="less than required"):
|
|
validate_data([1.0, 2.0], 5)
|
|
|
|
def test_raises_when_any_length_exceeds(self):
|
|
with pytest.raises(ValueError, match="less than required"):
|
|
validate_data([1.0, 2.0, 3.0], [2, 5])
|
|
|
|
|
|
class _VolatilityShared:
|
|
"""Reusable branch sweep for the six volatility helpers."""
|
|
|
|
func: staticmethod = None # set in subclasses
|
|
|
|
def test_default_returns_non_empty(self, ohlcv_df):
|
|
assert not self.__class__.func(ohlcv_df).empty
|
|
|
|
def test_is_crypto_picks_365_periods(self, ohlcv_df):
|
|
assert not self.__class__.func(ohlcv_df, is_crypto=True).empty
|
|
|
|
def test_trading_periods_override_warns_when_crypto(self, ohlcv_df):
|
|
with pytest.warns(UserWarning, match="is_crypto is overridden"):
|
|
self.__class__.func(ohlcv_df, trading_periods=252, is_crypto=True)
|
|
|
|
def test_clean_false_keeps_leading_nans(self, ohlcv_df):
|
|
result = self.__class__.func(ohlcv_df, clean=False)
|
|
assert result.isna().any()
|
|
|
|
def test_window_too_small_warns_and_defaults(self, ohlcv_df):
|
|
with pytest.warns(UserWarning, match="Window must be at least"):
|
|
self.__class__.func(ohlcv_df, window=0)
|
|
|
|
|
|
class TestParkinson(_VolatilityShared):
|
|
func = staticmethod(parkinson)
|
|
|
|
|
|
class TestStandardDeviation(_VolatilityShared):
|
|
func = staticmethod(standard_deviation)
|
|
|
|
|
|
class TestGarmanKlass(_VolatilityShared):
|
|
func = staticmethod(garman_klass)
|
|
|
|
|
|
class TestHodgesTompkins(_VolatilityShared):
|
|
func = staticmethod(hodges_tompkins)
|
|
|
|
def test_window_too_small_warns_and_defaults(self, ohlcv_df):
|
|
with pytest.warns(UserWarning, match="Window must be at least 2"):
|
|
self.__class__.func(ohlcv_df, window=1)
|
|
|
|
|
|
class TestRogersSatchell(_VolatilityShared):
|
|
func = staticmethod(rogers_satchell)
|
|
|
|
|
|
class TestYangZhang(_VolatilityShared):
|
|
func = staticmethod(yang_zhang)
|
|
|
|
def test_window_too_small_warns_and_defaults(self, ohlcv_df):
|
|
with pytest.warns(UserWarning, match="Window must be at least 2"):
|
|
self.__class__.func(ohlcv_df, window=1)
|
|
|
|
def test_overnight_moves_only_count_once(self):
|
|
# Prices move only between sessions (no intraday range), so the
|
|
# open-to-close and Rogers-Satchell terms are zero and the estimate
|
|
# reduces to the overnight (close-to-open) volatility.
|
|
gaps = np.array([0.0, 0.01, -0.02, 0.015, -0.005, 0.02, -0.01, 0.005])
|
|
price = 100 * np.exp(np.cumsum(gaps))
|
|
df = pd.DataFrame(
|
|
{"open": price, "high": price, "low": price, "close": price},
|
|
index=pd.date_range("2021-01-01", periods=len(price), freq="D"),
|
|
)
|
|
window = 5
|
|
log_oc = np.log(df["open"] / df["close"].shift(1))
|
|
expected = np.sqrt(
|
|
(log_oc**2).rolling(window).sum() / (window - 1) * 252
|
|
).dropna()
|
|
|
|
result = yang_zhang(df, window=window)
|
|
|
|
pd.testing.assert_series_equal(result, expected, check_names=False)
|
|
|
|
def test_matches_hand_computed_window(self):
|
|
# A flat first bar, then one window of 3 sessions. Per-session log
|
|
# moves: gap g = ln(O/C_prev), body c = ln(C/O), h = ln(H/O), lo = ln(L/O).
|
|
g = np.array([0.0, 0.01, -0.02, 0.03])
|
|
c = np.array([0.0, 0.02, -0.01, 0.0])
|
|
h = np.array([0.0, 0.03, 0.01, 0.02])
|
|
lo = np.array([0.0, -0.01, -0.02, -0.01])
|
|
log_close = np.log(100) + np.cumsum(g + c)
|
|
log_open = log_close - c
|
|
df = pd.DataFrame(
|
|
{
|
|
"open": np.exp(log_open),
|
|
"high": np.exp(log_open + h),
|
|
"low": np.exp(log_open + lo),
|
|
"close": np.exp(log_close),
|
|
},
|
|
index=pd.date_range("2021-01-01", periods=len(g), freq="D"),
|
|
)
|
|
|
|
result = yang_zhang(df, window=3)
|
|
|
|
# sum g^2 = 14e-4, sum c^2 = 5e-4, sum RS = 6e-4 + 4e-4 + 5e-4 = 15e-4,
|
|
# k = 0.34 / (1.34 + 4 / 2) = 17 / 167, divisor window - 1 = 2.
|
|
# var = (14 + 17/167 * 5 + 150/167 * 15) / 2 * 1e-4 = 4673 / 3340000.
|
|
# vol = sqrt(4673 / 3340000 * 252). The old close-to-close term,
|
|
# sum (g + c)^2 = 27e-4, gave sqrt(5047 / 3340000 * 252) = 0.61708.
|
|
assert len(result) == 1
|
|
assert result.iloc[0] == pytest.approx(0.5937791278704656, rel=1e-9)
|
|
|
|
|
|
class TestClenowMomentum:
|
|
def test_returns_r2_coef_predictions(self, ohlcv_df):
|
|
r2, coef, preds = clenow_momentum(ohlcv_df["close"])
|
|
assert isinstance(r2, float)
|
|
assert isinstance(coef, float)
|
|
assert len(preds) == 90
|
|
|
|
def test_flat_window_has_zero_r2(self):
|
|
r2, coef, preds = clenow_momentum(pd.Series([100.0] * 120))
|
|
assert r2 == 0.0
|
|
assert coef == pytest.approx(0.0, abs=1e-10)
|
|
assert preds.to_numpy() == pytest.approx(np.log(100.0))
|
|
|
|
def test_raises_when_window_exceeds_data(self):
|
|
with pytest.raises(ValueError, match="at least last"):
|
|
clenow_momentum(pd.Series([1.0, 2.0]), window=90)
|
|
|
|
|
|
class TestCalculateCones:
|
|
def test_default_model_std(self, ohlcv_df):
|
|
result = calculate_cones(
|
|
ohlcv_df, lower_q=0.1, upper_q=0.9, is_crypto=False, model="std"
|
|
)
|
|
assert not result.empty
|
|
|
|
@pytest.mark.parametrize(
|
|
"model",
|
|
[
|
|
"std",
|
|
"parkinson",
|
|
"garman_klass",
|
|
"hodges_tompkins",
|
|
"rogers_satchell",
|
|
"yang_zhang",
|
|
],
|
|
)
|
|
def test_every_model(self, ohlcv_df, model):
|
|
result = calculate_cones(
|
|
ohlcv_df, lower_q=0.2, upper_q=0.8, is_crypto=False, model=model
|
|
)
|
|
assert not result.empty
|
|
|
|
def test_is_crypto_branch(self, ohlcv_df):
|
|
result = calculate_cones(
|
|
ohlcv_df, lower_q=0.1, upper_q=0.9, is_crypto=True, model="std"
|
|
)
|
|
assert not result.empty
|
|
|
|
def test_swaps_quantiles_when_lower_above_upper(self, ohlcv_df):
|
|
"""When lower_q > upper_q the function swaps them silently."""
|
|
normal = calculate_cones(
|
|
ohlcv_df, lower_q=0.1, upper_q=0.9, is_crypto=False, model="std"
|
|
)
|
|
swapped = calculate_cones(
|
|
ohlcv_df, lower_q=0.9, upper_q=0.1, is_crypto=False, model="std"
|
|
)
|
|
assert list(normal.columns) == list(swapped.columns)
|
|
|
|
def test_raises_on_quantile_out_of_range(self, ohlcv_df):
|
|
with pytest.raises(ValueError, match="between 0 and 1"):
|
|
calculate_cones(
|
|
ohlcv_df, lower_q=1.5, upper_q=2.0, is_crypto=False, model="std"
|
|
)
|
|
|
|
|
|
class TestCalculateFibLevels:
|
|
def test_default_branch(self, ohlcv_df):
|
|
result = calculate_fib_levels(ohlcv_df, "close")
|
|
assert result is not None
|
|
assert len(result) >= 6
|
|
|
|
def test_raises_on_missing_close_col(self, ohlcv_df):
|
|
with pytest.raises(ValueError, match="not in data"):
|
|
calculate_fib_levels(ohlcv_df, "missing_col")
|
|
|
|
def test_with_explicit_start_end_in_index(self, ohlcv_df):
|
|
start = ohlcv_df.index[10]
|
|
end = ohlcv_df.index[60]
|
|
result = calculate_fib_levels(ohlcv_df, "close", start_date=start, end_date=end)
|
|
assert result is not None
|
|
|
|
def test_with_explicit_start_end_not_in_index_warns(self, ohlcv_df):
|
|
"""When user supplies dates outside the index, nearest neighbours are used and a warning is emitted."""
|
|
start = ohlcv_df.index[10] + pd.Timedelta(hours=12)
|
|
end = ohlcv_df.index[60] + pd.Timedelta(hours=12)
|
|
with warnings.catch_warnings(record=True) as caught:
|
|
warnings.simplefilter("always")
|
|
calculate_fib_levels(ohlcv_df, "close", start_date=start, end_date=end)
|
|
assert any("not in data" in str(w.message) for w in caught)
|
|
|
|
def test_swaps_when_min_date_after_max(self, ohlcv_df):
|
|
"""When start_date is after end_date the function swaps the pair."""
|
|
df = ohlcv_df.copy()
|
|
df["close"] = np.concatenate(
|
|
[np.arange(50, 0, -1, dtype=float), np.arange(1, 50, dtype=float)]
|
|
)
|
|
start = df.index[5]
|
|
end = df.index[80]
|
|
result = calculate_fib_levels(df, "close", start_date=start, end_date=end)
|
|
assert result is not None
|