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()