Files
2026-07-09 05:08:16 +08:00

563 lines
20 KiB
Python

"""
Known-value oracle tests: permanent ground truth (Priority 2 - no optional deps).
Hand-computable ground truth that never depends on external libraries.
These tests encode fundamental mathematical properties and serve as a permanent
oracle for correctness.
All tests use NO optional dependencies - they run in every CI environment.
"""
from __future__ import annotations
import numpy as np
import ferro_ta
# ---------------------------------------------------------------------------
# SMA Known Values
# ---------------------------------------------------------------------------
class TestSMAKnownValues:
"""SMA is the simple average over a window."""
def test_sma_simple_sequence(self):
"""SMA([1,2,3,4,5], 3) == [nan, nan, 2.0, 3.0, 4.0]."""
data = np.array([1.0, 2.0, 3.0, 4.0, 5.0])
result = ferro_ta.SMA(data, timeperiod=3)
assert np.isnan(result[0])
assert np.isnan(result[1])
assert np.abs(result[2] - 2.0) < 1e-10 # (1+2+3)/3 = 2.0
assert np.abs(result[3] - 3.0) < 1e-10 # (2+3+4)/3 = 3.0
assert np.abs(result[4] - 4.0) < 1e-10 # (3+4+5)/3 = 4.0
def test_sma_period_one_is_identity(self):
"""SMA with period=1 should be the identity function."""
data = np.array([10.0, 12.0, 15.0, 11.0, 13.0])
result = ferro_ta.SMA(data, timeperiod=1)
assert np.allclose(result, data, atol=1e-10)
def test_sma_constant_series(self):
"""SMA of constant series should equal that constant."""
data = np.ones(10) * 42.0
result = ferro_ta.SMA(data, timeperiod=5)
# After warmup, all values should be 42.0
assert np.allclose(result[4:], 42.0, atol=1e-10)
# ---------------------------------------------------------------------------
# EMA Known Values
# ---------------------------------------------------------------------------
class TestEMAKnownValues:
"""EMA is an exponentially weighted moving average."""
def test_ema_period_one_is_identity(self):
"""EMA with period=1 should be the identity function (alpha=1)."""
data = np.array([10.0, 12.0, 15.0, 11.0, 13.0])
result = ferro_ta.EMA(data, timeperiod=1)
assert np.allclose(result, data, atol=1e-10)
def test_ema_constant_series_converges(self):
"""EMA of constant series should converge to that constant."""
data = np.ones(100) * 42.0
result = ferro_ta.EMA(data, timeperiod=10)
# After sufficient warmup, should converge to 42.0
assert np.allclose(result[-10:], 42.0, atol=1e-6)
def test_ema_monotone_rising_is_increasing(self):
"""EMA of monotone rising series should be strictly increasing after warmup."""
data = np.arange(1.0, 51.0) # 1, 2, 3, ..., 50
result = ferro_ta.EMA(data, timeperiod=10)
# After warmup, EMA should be strictly increasing
for i in range(20, len(result) - 1):
assert result[i + 1] > result[i], (
f"EMA not increasing at index {i}: {result[i]} >= {result[i + 1]}"
)
# ---------------------------------------------------------------------------
# WMA Known Values
# ---------------------------------------------------------------------------
class TestWMAKnownValues:
"""WMA is a linearly weighted moving average."""
def test_wma_manual_calculation(self):
"""WMA([3,5,7], 2) at index 2 = (1*5 + 2*7)/(1+2) = 6.333..."""
data = np.array([3.0, 5.0, 7.0])
result = ferro_ta.WMA(data, timeperiod=2)
# Index 0: warmup (NaN)
assert np.isnan(result[0])
# Index 1: (1*3 + 2*5)/(1+2) = 13/3 = 4.333...
expected_1 = (1 * 3.0 + 2 * 5.0) / (1 + 2)
assert np.abs(result[1] - expected_1) < 1e-10
# Index 2: (1*5 + 2*7)/(1+2) = 19/3 = 6.333...
expected_2 = (1 * 5.0 + 2 * 7.0) / (1 + 2)
assert np.abs(result[2] - expected_2) < 1e-10
def test_wma_period_one_is_identity(self):
"""WMA with period=1 should be the identity function."""
data = np.array([10.0, 12.0, 15.0, 11.0, 13.0])
result = ferro_ta.WMA(data, timeperiod=1)
assert np.allclose(result, data, atol=1e-10)
# ---------------------------------------------------------------------------
# BBANDS Known Values
# ---------------------------------------------------------------------------
class TestBBANDSKnownValues:
"""Bollinger Bands: middle = SMA, upper/lower = middle ± (nbdevup/nbdevdn * stddev)."""
def test_bbands_constant_series(self):
"""For constant series: upper == middle == lower (stddev=0)."""
data = np.ones(20) * 50.0
upper, middle, lower = ferro_ta.BBANDS(data, timeperiod=5)
# After warmup, all three bands should be 50.0
assert np.allclose(upper[4:], 50.0, atol=1e-10)
assert np.allclose(middle[4:], 50.0, atol=1e-10)
assert np.allclose(lower[4:], 50.0, atol=1e-10)
def test_bbands_middle_is_sma(self):
"""Middle band should equal SMA."""
data = np.array([10.0, 12.0, 15.0, 11.0, 13.0, 14.0, 16.0, 12.0])
upper, middle, lower = ferro_ta.BBANDS(data, timeperiod=5)
sma = ferro_ta.SMA(data, timeperiod=5)
assert np.allclose(middle, sma, atol=1e-10, equal_nan=True)
def test_bbands_symmetric(self):
"""Bands should be symmetric: upper-middle == middle-lower (with same nbdev)."""
data = np.array([10.0, 12.0, 15.0, 11.0, 13.0, 14.0, 16.0, 12.0, 18.0, 10.0])
upper, middle, lower = ferro_ta.BBANDS(
data, timeperiod=5, nbdevup=2.0, nbdevdn=2.0
)
# After warmup, bands should be symmetric
upper_dist = upper[4:] - middle[4:]
lower_dist = middle[4:] - lower[4:]
assert np.allclose(upper_dist, lower_dist, atol=1e-10)
# ---------------------------------------------------------------------------
# RSI Known Values
# ---------------------------------------------------------------------------
class TestRSIKnownValues:
"""RSI measures momentum: monotone rising → RSI > 50, monotone falling → RSI < 50."""
def test_rsi_monotone_rising(self):
"""Monotone rising series should produce RSI > 50 after warmup."""
data = np.arange(1.0, 51.0) # 1, 2, 3, ..., 50
result = ferro_ta.RSI(data, timeperiod=14)
# After warmup, RSI should be > 50 (strong uptrend)
assert np.all(result[20:] > 50.0), "RSI of rising series should be > 50"
def test_rsi_monotone_falling(self):
"""Monotone falling series should produce RSI < 50 after warmup."""
data = np.arange(50.0, 0.0, -1.0) # 50, 49, 48, ..., 1
result = ferro_ta.RSI(data, timeperiod=14)
# After warmup, RSI should be < 50 (strong downtrend)
assert np.all(result[20:] < 50.0), "RSI of falling series should be < 50"
def test_rsi_constant_series(self):
"""Constant series should produce RSI = 100 or NaN (no momentum).
Note: For constant series with no change, ferro_ta returns 100
(no downward movement), which is mathematically correct.
"""
data = np.ones(30) * 42.0
result = ferro_ta.RSI(data, timeperiod=14)
# Constant series has no momentum; RSI should be NaN or 100
# ferro_ta returns 100 (no down movement = 100% bullish)
valid_values = result[~np.isnan(result)]
if len(valid_values) > 0:
# Should be either NaN everywhere or 100 everywhere
assert np.all(np.abs(valid_values - 100.0) < 1e-10) or np.all(
np.abs(valid_values - 50.0) < 5.0
), "RSI of constant series should be 100 (no down movement) or close to 50"
# ---------------------------------------------------------------------------
# ATR Known Values
# ---------------------------------------------------------------------------
class TestATRKnownValues:
"""ATR measures volatility: H==L==C → ATR=0."""
def test_atr_zero_range(self):
"""When H==L==C, ATR should be 0 (no volatility)."""
n = 30
high = np.ones(n) * 50.0
low = np.ones(n) * 50.0
close = np.ones(n) * 50.0
result = ferro_ta.ATR(high, low, close, timeperiod=14)
# After warmup, ATR should be 0
assert np.allclose(result[14:], 0.0, atol=1e-10)
def test_atr_manual_tr_calculation(self):
"""Manually verify TR formula for 3-bar sequence.
Note: ATR requires warmup period. For period=14, first 13 bars are NaN.
We test with longer period to see TR values.
"""
# Bar 0: H=11, L=9, C=10
# Bar 1: H=13, L=10, C=12 → TR = max(13-10, |13-10|, |10-10|) = 3
# Bar 2: H=14, L=11, C=13 → TR = max(14-11, |14-12|, |11-12|) = 3
high = np.array(
[
11.0,
13.0,
14.0,
15.0,
16.0,
17.0,
18.0,
19.0,
20.0,
21.0,
22.0,
23.0,
24.0,
25.0,
26.0,
]
)
low = np.array(
[
9.0,
10.0,
11.0,
12.0,
13.0,
14.0,
15.0,
16.0,
17.0,
18.0,
19.0,
20.0,
21.0,
22.0,
23.0,
]
)
close = np.array(
[
10.0,
12.0,
13.0,
14.0,
15.0,
16.0,
17.0,
18.0,
19.0,
20.0,
21.0,
22.0,
23.0,
24.0,
25.0,
]
)
# For period=1, ATR still has warmup. Use TRANGE to check TR values directly
tr = ferro_ta.TRANGE(high, low, close)
# TR[0] = H-L = 11-9 = 2
# TR[1] = max(13-10, |13-10|, |10-10|) = max(3, 3, 0) = 3
# TR[2] = max(14-11, |14-12|, |11-12|) = max(3, 2, 1) = 3
assert np.abs(tr[0] - 2.0) < 1e-10
assert np.abs(tr[1] - 3.0) < 1e-10
assert np.abs(tr[2] - 3.0) < 1e-10
# ---------------------------------------------------------------------------
# MOM Known Values
# ---------------------------------------------------------------------------
class TestMOMKnownValues:
"""MOM is the difference: close[i] - close[i - period]."""
def test_mom_manual_calculation(self):
"""MOM([10,12,15,11], period=2) == [nan,nan,5,-1]."""
data = np.array([10.0, 12.0, 15.0, 11.0])
result = ferro_ta.MOM(data, timeperiod=2)
assert np.isnan(result[0])
assert np.isnan(result[1])
assert np.abs(result[2] - 5.0) < 1e-10 # 15 - 10 = 5
assert np.abs(result[3] - (-1.0)) < 1e-10 # 11 - 12 = -1
# ---------------------------------------------------------------------------
# ROC Known Values
# ---------------------------------------------------------------------------
class TestROCKnownValues:
"""ROC is the percentage change: 100 * (close[i] - close[i-period]) / close[i-period]."""
def test_roc_manual_calculation(self):
"""ROC([10,12], period=1)[1] == 20.0."""
data = np.array([10.0, 12.0])
result = ferro_ta.ROC(data, timeperiod=1)
# ROC[1] = 100 * (12 - 10) / 10 = 100 * 0.2 = 20.0
assert np.abs(result[1] - 20.0) < 1e-10
# ---------------------------------------------------------------------------
# MACD Known Values
# ---------------------------------------------------------------------------
class TestMACDKnownValues:
"""MACD: histogram == macd - signal always."""
def test_macd_histogram_identity(self):
"""histogram should always equal macd - signal."""
data = np.arange(1.0, 51.0)
macd, signal, histogram = ferro_ta.MACD(
data, fastperiod=12, slowperiod=26, signalperiod=9
)
# histogram = macd - signal (within floating-point tolerance)
expected_histogram = macd - signal
assert np.allclose(histogram, expected_histogram, atol=1e-10, equal_nan=True)
# ---------------------------------------------------------------------------
# VWAP Known Values
# ---------------------------------------------------------------------------
class TestVWAPKnownValues:
"""VWAP: period=1 VWAP == TYPPRICE."""
def test_vwap_period_one_equals_typprice(self):
"""For period=1, VWAP should equal typical price (H+L+C)/3."""
high = np.array([11.0, 13.0, 14.0])
low = np.array([9.0, 10.0, 11.0])
close = np.array([10.0, 12.0, 13.0])
volume = np.array([1000.0, 1000.0, 1000.0])
result = ferro_ta.VWAP(high, low, close, volume, timeperiod=1)
expected = ferro_ta.TYPPRICE(high, low, close)
assert np.allclose(result, expected, atol=1e-10)
def test_vwap_cumulative_manual(self):
"""Manually verify cumulative VWAP for simple 3-bar sequence."""
# Bar 0: TP=10, Vol=100 → VWAP = (10*100)/(100) = 10.0
# Bar 1: TP=12, Vol=200 → VWAP = (10*100 + 12*200)/(100+200) = 3400/300 = 11.333...
# Bar 2: TP=11, Vol=150 → VWAP = (10*100 + 12*200 + 11*150)/(100+200+150) = 5050/450 = 11.222...
high = np.array([11.0, 13.0, 12.0])
low = np.array([9.0, 11.0, 10.0])
close = np.array([10.0, 12.0, 11.0])
volume = np.array([100.0, 200.0, 150.0])
result = ferro_ta.VWAP(high, low, close, volume, timeperiod=0) # cumulative
typ = (high + low + close) / 3.0
expected_0 = typ[0]
expected_1 = (typ[0] * volume[0] + typ[1] * volume[1]) / (volume[0] + volume[1])
expected_2 = (typ[0] * volume[0] + typ[1] * volume[1] + typ[2] * volume[2]) / (
volume[0] + volume[1] + volume[2]
)
assert np.abs(result[0] - expected_0) < 1e-10
assert np.abs(result[1] - expected_1) < 1e-10
assert np.abs(result[2] - expected_2) < 1e-10
# ---------------------------------------------------------------------------
# DONCHIAN Known Values
# ---------------------------------------------------------------------------
class TestDONCHIANKnownValues:
"""DONCHIAN: upper = MAX(high), lower = MIN(low), middle = (upper+lower)/2."""
def test_donchian_structure(self):
"""upper == MAX(high), lower == MIN(low), middle == (upper+lower)/2."""
high = np.array([11.0, 13.0, 14.0, 12.0, 15.0])
low = np.array([9.0, 10.0, 11.0, 10.0, 12.0])
period = 3
upper, middle, lower = ferro_ta.DONCHIAN(high, low, timeperiod=period)
# upper should match rolling max of high
max_high = ferro_ta.MAX(high, timeperiod=period)
assert np.allclose(upper, max_high, atol=1e-10, equal_nan=True)
# lower should match rolling min of low
min_low = ferro_ta.MIN(low, timeperiod=period)
assert np.allclose(lower, min_low, atol=1e-10, equal_nan=True)
# middle should be (upper + lower) / 2
expected_middle = (upper + lower) / 2.0
assert np.allclose(middle, expected_middle, atol=1e-10, equal_nan=True)
# ---------------------------------------------------------------------------
# PIVOT_POINTS Known Values
# ---------------------------------------------------------------------------
class TestPIVOT_POINTSKnownValues:
"""PIVOT_POINTS classic formula: P=(H+L+C)/3, R1=2P-L, S1=2P-H, R2=P+(H-L), S2=P-(H-L)."""
def test_pivot_points_classic_formula(self):
"""Given H=110, L=90, C=100: P=100, R1=110, S1=90, R2=120, S2=80.
Note: PIVOT_POINTS operates on OHLC bars. Single bar produces valid pivots.
"""
high = np.array([110.0, 110.0]) # Need at least 2 bars
low = np.array([90.0, 90.0])
close = np.array([100.0, 100.0])
pivot, r1, s1, r2, s2 = ferro_ta.PIVOT_POINTS(
high, low, close, method="classic"
)
# Check last bar (index 1) which has full history
# P = (110 + 90 + 100) / 3 = 100
assert np.abs(pivot[1] - 100.0) < 1e-10
# R1 = 2*P - L = 2*100 - 90 = 110
assert np.abs(r1[1] - 110.0) < 1e-10
# S1 = 2*P - H = 2*100 - 110 = 90
assert np.abs(s1[1] - 90.0) < 1e-10
# R2 = P + (H - L) = 100 + 20 = 120
assert np.abs(r2[1] - 120.0) < 1e-10
# S2 = P - (H - L) = 100 - 20 = 80
assert np.abs(s2[1] - 80.0) < 1e-10
# ---------------------------------------------------------------------------
# Statistic Known Values
# ---------------------------------------------------------------------------
class TestStatisticKnownValues:
"""Statistical functions: correlation, linear regression."""
def test_linearreg_slope_of_linear_sequence(self):
"""LINEARREG_SLOPE([0,1,2,3,4], 5) == 1.0."""
data = np.array([0.0, 1.0, 2.0, 3.0, 4.0])
result = ferro_ta.LINEARREG_SLOPE(data, timeperiod=5)
# Last value should be slope = 1.0
assert np.abs(result[-1] - 1.0) < 1e-10
def test_correl_x_with_x_is_one(self):
"""CORREL(x, x) should be 1.0."""
x = np.array([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0])
result = ferro_ta.CORREL(x, x, timeperiod=5)
# After warmup, correlation should be 1.0
assert np.allclose(result[4:], 1.0, atol=1e-10)
def test_correl_x_with_negative_x_is_minus_one(self):
"""CORREL(x, -x) should be -1.0."""
x = np.array([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0])
neg_x = -x
result = ferro_ta.CORREL(x, neg_x, timeperiod=5)
# After warmup, correlation should be -1.0
assert np.allclose(result[4:], -1.0, atol=1e-10)
# ---------------------------------------------------------------------------
# Pattern Known Values
# ---------------------------------------------------------------------------
class TestPatternKnownValues:
"""Candlestick patterns: construct known-good OHLC sequences."""
def test_doji_known_sequence(self):
"""Construct a perfect doji: open == close, small body."""
# Doji: open == close (or very close), H and L have range
high = np.array([11.0, 11.0, 11.0, 11.0, 11.0])
low = np.array([9.0, 9.0, 9.0, 9.0, 9.0])
close = np.array([10.0, 10.0, 10.0, 10.0, 10.0])
open_ = np.array([10.0, 10.0, 10.0, 10.0, 10.0])
result = ferro_ta.CDLDOJI(open_, high, low, close)
# Should detect doji (non-zero pattern)
# At least some values should be non-zero
assert np.any(result != 0), "CDLDOJI should detect perfect doji pattern"
def test_engulfing_known_sequence(self):
"""Construct a bullish engulfing pattern."""
# Bullish engulfing: bar[i-1] is bearish (O > C), bar[i] is bullish (C > O) and engulfs bar[i-1]
# Bar 0: O=12, H=12, L=10, C=10 (bearish)
# Bar 1: O=9, H=13, L=9, C=13 (bullish, engulfs bar 0)
open_ = np.array([12.0, 9.0])
high = np.array([12.0, 13.0])
low = np.array([10.0, 9.0])
close = np.array([10.0, 13.0])
result = ferro_ta.CDLENGULFING(open_, high, low, close)
# Should detect engulfing at index 1
assert result[1] != 0, "CDLENGULFING should detect bullish engulfing pattern"
def test_hammer_known_sequence(self):
"""Construct a hammer pattern: small body at top, long lower shadow."""
# Hammer: small body, long lower shadow (>= 2x body), little/no upper shadow
# O=11, H=11.5, L=9, C=11 → body=0, lower_shadow=2, upper_shadow=0.5
open_ = np.array([11.0])
high = np.array([11.5])
low = np.array([9.0])
close = np.array([11.0])
result = ferro_ta.CDLHAMMER(open_, high, low, close)
# Should detect hammer (non-zero)
# Note: hammer detection depends on lookback, so we test multiple bars
open_ = np.array([10.0, 10.5, 11.0])
high = np.array([10.5, 11.0, 11.5])
low = np.array([9.5, 10.0, 9.0])
close = np.array([10.0, 10.5, 11.0])
result = ferro_ta.CDLHAMMER(open_, high, low, close)
# Last bar has hammer characteristics
# (actual detection may vary based on implementation)
assert result.shape == close.shape