Spaces:
Running
Running
File size: 6,927 Bytes
9831ced | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | import unittest
import numpy as np
import pandas as pd
from indicators.indicators import _wilder_rma, add_all_indicators, bollinger_bands, ema, latest_signals, macd, rsi, sma, stochastic_oscillator
def _ohlcv(n: int, seed: int=5) -> pd.DataFrame:
rng = np.random.default_rng(seed)
close = 100 + np.cumsum(rng.normal(0, 0.5, n))
open_ = close + rng.normal(0, 0.1, n)
high = np.maximum(open_, close) + np.abs(rng.normal(0, 0.3, n))
low = np.minimum(open_, close) - np.abs(rng.normal(0, 0.3, n))
volume = rng.integers(100, 5000, n).astype(float)
idx = pd.date_range('2024-01-01', periods=n, freq='5min')
return pd.DataFrame({'Open': open_, 'High': high, 'Low': low, 'Close': close, 'Volume': volume}, index=idx)
class TestWilderRma(unittest.TestCase):
def test_seeds_from_a_window_length_average_then_recurses(self):
values = pd.Series([1.0, 2.0, 3.0, 4.0, 5.0])
out = _wilder_rma(values, 3)
self.assertTrue(out.iloc[:2].isna().all())
self.assertAlmostEqual(out.iloc[2], 2.0)
self.assertAlmostEqual(out.iloc[3], (2.0 * 2 + 4.0) / 3)
self.assertAlmostEqual(out.iloc[4], (out.iloc[3] * 2 + 5.0) / 3)
def test_skips_a_leading_nan_before_seeding(self):
values = pd.Series([np.nan, 2.0, 4.0, 6.0])
out = _wilder_rma(values, 3)
self.assertTrue(out.iloc[:3].isna().all())
self.assertAlmostEqual(out.iloc[3], 4.0)
def test_series_shorter_than_the_window_is_all_nan(self):
out = _wilder_rma(pd.Series([1.0, 2.0]), 14)
self.assertTrue(out.isna().all())
class TestRsi(unittest.TestCase):
def test_warmup_rows_are_nan_not_fifty(self):
out = rsi(_ohlcv(60)['Close'], 14)
self.assertTrue(out.iloc[:14].isna().all())
self.assertTrue(out.iloc[14:].notna().all())
self.assertEqual(int((out.iloc[:14] == 50.0).sum()), 0)
def test_only_gains_reads_one_hundred(self):
out = rsi(pd.Series(np.arange(40, dtype=float) + 100), 14)
self.assertTrue(out.iloc[:14].isna().all())
self.assertTrue((out.iloc[14:] == 100.0).all())
def test_flat_market_is_neutral_only_after_warmup(self):
out = rsi(pd.Series([100.0] * 40), 14)
self.assertTrue(out.iloc[:14].isna().all())
self.assertTrue((out.iloc[14:] == 50.0).all())
def test_stays_within_bounds(self):
out = rsi(_ohlcv(200, seed=9)['Close'], 14).dropna()
self.assertGreaterEqual(out.min(), 0.0)
self.assertLessEqual(out.max(), 100.0)
class TestMacdAndBollinger(unittest.TestCase):
def setUp(self):
self.close = _ohlcv(80, seed=3)['Close']
def test_macd_is_the_fast_minus_slow_ema_with_its_own_signal_ema(self):
line, signal, hist = macd(self.close)
expected_line = ema(self.close, 12) - ema(self.close, 26)
pd.testing.assert_series_equal(line, expected_line, check_names=False)
pd.testing.assert_series_equal(signal, ema(expected_line, 9), check_names=False)
pd.testing.assert_series_equal(hist, expected_line - signal, check_names=False)
def test_bollinger_uses_population_std_around_the_sma(self):
upper, mid, lower = bollinger_bands(self.close, window=20, num_std=2.0)
pd.testing.assert_series_equal(mid, sma(self.close, 20), check_names=False)
self.assertTrue(upper.iloc[:19].isna().all())
window = self.close.iloc[-20:].to_numpy()
expected_std = float(np.std(window))
self.assertAlmostEqual(float(upper.iloc[-1]), float(window.mean()) + 2 * expected_std, places=9)
self.assertAlmostEqual(float(lower.iloc[-1]), float(window.mean()) - 2 * expected_std, places=9)
class TestStochastic(unittest.TestCase):
def test_warmup_rows_are_nan_not_fifty(self):
df = _ohlcv(60, seed=4)
k, d = stochastic_oscillator(df['High'], df['Low'], df['Close'])
self.assertTrue(k.iloc[:13].isna().all())
self.assertTrue(k.iloc[13:].notna().all())
self.assertTrue(d.iloc[:15].isna().all())
self.assertEqual(int((k.iloc[:13] == 50.0).sum()), 0)
def test_percent_k_matches_the_definition(self):
df = _ohlcv(40, seed=6)
k, _ = stochastic_oscillator(df['High'], df['Low'], df['Close'])
lowest = float(df['Low'].iloc[-14:].min())
highest = float(df['High'].iloc[-14:].max())
expected = 100 * (float(df['Close'].iloc[-1]) - lowest) / (highest - lowest)
self.assertAlmostEqual(float(k.iloc[-1]), expected, places=9)
def test_zero_range_is_neutral_only_after_warmup(self):
flat = pd.Series([100.0] * 30)
k, d = stochastic_oscillator(flat, flat, flat)
self.assertTrue(k.iloc[:13].isna().all())
self.assertTrue((k.iloc[13:] == 50.0).all())
self.assertTrue((d.iloc[15:] == 50.0).all())
class TestAddAllIndicators(unittest.TestCase):
def setUp(self):
self.df = _ohlcv(60, seed=8)
self.out = add_all_indicators(self.df)
def test_adds_every_indicator_column_without_mutating_the_input(self):
expected = ['SMA_20', 'EMA_20', 'RSI_14', 'MACD', 'MACD_Signal', 'MACD_Hist', 'BB_Upper', 'BB_Mid', 'BB_Lower', 'Stoch_%K', 'Stoch_%D']
for column in expected:
self.assertIn(column, self.out.columns)
self.assertNotIn(column, self.df.columns)
def test_warmup_nan_counts_are_the_real_indicator_periods(self):
self.assertEqual(int(self.out['RSI_14'].isna().sum()), 14)
self.assertEqual(int(self.out['Stoch_%K'].isna().sum()), 13)
self.assertEqual(int(self.out['Stoch_%D'].isna().sum()), 15)
self.assertEqual(int(self.out['SMA_20'].isna().sum()), 19)
self.assertEqual(int(self.out['BB_Upper'].isna().sum()), 19)
class TestLatestSignals(unittest.TestCase):
def test_reports_insufficient_data_instead_of_a_fabricated_reading(self):
out = add_all_indicators(_ohlcv(3, seed=2))
signals = latest_signals(out)
self.assertEqual(signals['RSI (14)']['value'], 'n/a')
self.assertEqual(signals['RSI (14)']['signal'], 'Insufficient data')
self.assertEqual(signals['Stochastic (14,3)']['value'], 'n/a')
self.assertEqual(signals['Stochastic (14,3)']['signal'], 'Insufficient data')
self.assertEqual(signals['Bollinger Bands']['signal'], 'Insufficient data')
self.assertEqual(signals['Moving Averages']['signal'], 'Insufficient data for SMA')
self.assertIn('SMA20=n/a', signals['Moving Averages']['value'])
def test_reports_real_readings_once_warmed_up(self):
signals = latest_signals(add_all_indicators(_ohlcv(80, seed=1)))
for name, entry in signals.items():
self.assertNotIn('Insufficient', entry['signal'], name)
self.assertIsInstance(signals['RSI (14)']['value'], float)
self.assertIsInstance(signals['Stochastic (14,3)']['value'], float)
if __name__ == '__main__':
unittest.main() |