Spaces:
Running
Running
| 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() |