hanlap/test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: ATR-adaptive BOS buffer, FVG gap-size filter, configurable Ichimoku periods

Browse Source 
- indicator_service.py: _detect_bos()/market_structure() now scale the
  break-confirmation buffer by the symbol's own current ATR% instead of a
  fixed 0.3% for every symbol; falls back to the fixed value when ATR%
  isn't supplied.
- indicator_service.py: detect_fvg() rejects gaps smaller than 10% of
  current ATR% when atr_pct is given, filtering noise-sized gaps that
  carried no real "unfilled order" significance on low timeframes.
- candle_service.py: computes ATR% earlier so it can feed both
  market_structure() and detect_fvg(), not just detect_market_regime();
  backtest_engine.py reuses the same per-candle ATR% for BOS instead of
  computing it twice.
- indicator_service.py: ichimoku() takes tenkan/kijun/senkou_b_period and
  displacement as parameters (defaults unchanged at 9/26/52/26) so a
  future walk-forward comparison against crypto-scaled periods doesn't
  require editing the function — the classic Japanese-calendar defaults
  aren't changed here since that needs empirical validation, not a guess.

232 backend tests pass (+13).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Le
2026-07-04 21:02:59 +07:00
parent 2399d0cb0c
commit 842631744c
5 changed files with 407 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/backtest_engine.py
+11 -4
View File
@@ -428,11 +428,18 @@ def _compute_scores_series(
confirmed_div_low_idx.append(div_ptr)
div_ptr += 1
# ATR% as of this candle — feeds both BOS's break-confirmation
# buffer (fix uu) and detect_market_regime's adaptive "volatile"
# threshold (fix kk) below, computed once and reused for both.
atr_tail = _tail(atr_full, i, _SHORT_WINDOW)
last_atr = atr_tail[-1] if atr_tail else None
atr_pct = (last_atr / close_prices_full[i] * 100.0) if last_atr and close_prices_full[i] > 0 else None
# SMC (BOS/CHoCH/trend/order-blocks) from the causally-confirmed
# swings above, instead of rescanning raw candles for pivots.
recent_highs = confirmed_swing_highs[-3:]
recent_lows = confirmed_swing_lows[-3:]
bos = _detect_bos(recent_highs, recent_lows, [close_prices_full[i]])
bos = _detect_bos(recent_highs, recent_lows, [close_prices_full[i]], atr_pct=atr_pct)
choch = _detect_choch(
confirmed_swing_highs[-5:], confirmed_swing_lows[-5:],
close_prices_full[max(0, i - 19):i + 1],
@@ -475,6 +482,9 @@ def _compute_scores_series(
mtf_recent_highs = state["highs"][-3:]
mtf_recent_lows = state["lows"][-3:]
mtf_close = mtf["close_prices"][mtf_idx]
# No per-MTF-timeframe ATR series is precomputed — falls back
# to _detect_bos's fixed 0.3% buffer (fix uu only covers the
# main timeframe's own BOS check above).
mtf_bos = _detect_bos(mtf_recent_highs, mtf_recent_lows, [mtf_close])
mtf_choch = _detect_choch(
state["highs"][-5:], state["lows"][-5:],
@@ -517,9 +527,6 @@ def _compute_scores_series(
# in signal_scoring.py, applied in _simulate_from_scores once the
# threshold combo decides a concrete signal_type).
adx_data = _tail_dict(adx_full, i, _SHORT_WINDOW)
atr_tail = _tail(atr_full, i, _SHORT_WINDOW)
last_atr = atr_tail[-1] if atr_tail else None
atr_pct = (last_atr / close_prices_full[i] * 100.0) if last_atr and close_prices_full[i] > 0 else None
regime_start = max(0, i + 1 - _REGIME_WINDOW)
atr_history_start = max(0, i + 1 - _ATR_HISTORY_WINDOW)
market_regime = detect_market_regime(
backend/app/services/candle_service.py
+21 -18
View File
@@ -484,9 +484,26 @@ async def get_indicators(
else:
computed["rsi_divergence"] = (None, None)
# Compute ATR% ahead of Market Structure/ADX below — both need it:
# market_structure()'s BOS confirmation buffer adapts to it (fix
# uu), and detect_market_regime's "volatile" cutoff adapts to its
# full history (fix kk). Also keep the full history (not just the
# latest value) for that second use.
atr_pct_history: list[float | None] = []
try:
from app.services.indicator_service import atr as _calc_atr
raw_atr = _calc_atr(candle_dicts, period=14)
atr_pct_history = [
(a / c * 100.0) if a and c and c > 0 else None
for a, c in zip(raw_atr, close_prices)
]
atr_pct_val = atr_pct_history[-1] if atr_pct_history else None
except Exception:
atr_pct_val = None
# Add Market Structure (SMC)
from app.services.indicator_service import market_structure
computed["market_structure"] = market_structure(candle_dicts, pivot_lookback=3)
computed["market_structure"] = market_structure(candle_dicts, pivot_lookback=3, atr_pct=atr_pct_val)
# Add MACD divergence detection
macd_data = computed.get("macd", {})
@@ -500,21 +517,6 @@ async def get_indicators(
# Add ADX (Average Directional Index) + Market Regime
computed["adx_data"] = adx(candle_dicts, period=14)
atr_vals = computed.get("supertrend", {}).get("trend", None)
# Compute ATR% for regime detection — also keep the full history
# (not just the latest value) so detect_market_regime can judge
# "volatile" against THIS symbol's own recent ATR% distribution
# instead of one fixed cutoff shared by every symbol (fix kk).
atr_pct_history: list[float | None] = []
try:
from app.services.indicator_service import atr as _calc_atr
raw_atr = _calc_atr(candle_dicts, period=14)
atr_pct_history = [
(a / c * 100.0) if a and c and c > 0 else None
for a, c in zip(raw_atr, close_prices)
]
atr_pct_val = atr_pct_history[-1] if atr_pct_history else None
except Exception:
atr_pct_val = None
# Extract high/low prices for regime detection
high_prices = [c["high"] for c in candle_dicts]
@@ -535,8 +537,9 @@ async def get_indicators(
# Add MFI (Money Flow Index)
computed["mfi_14"] = mfi(candle_dicts, period=14)
# Add FVG (Fair Value Gap)
fvg_type, fvg_high, fvg_low = detect_fvg(candle_dicts, lookback=30)
# Add FVG (Fair Value Gap) — atr_pct filters out gaps too small to
# be meaningful relative to this symbol's own volatility (fix qq).
fvg_type, fvg_high, fvg_low = detect_fvg(candle_dicts, lookback=30, atr_pct=atr_pct_val)
computed["fvg"] = {"type": fvg_type, "gap_high": fvg_high, "gap_low": fvg_low}
# Add Candlestick Pattern Recognition
backend/app/services/indicator_service.py
+89 -21
View File
@@ -552,7 +552,13 @@ def volume_breakout(
# Ichimoku Cloud
# ======================================================================
def ichimoku(candles: list[dict]) -> dict[str, list]:
def ichimoku(
candles: list[dict],
tenkan_period: int = 9,
kijun_period: int = 26,
senkou_b_period: int = 52,
displacement: int = 26,
) -> dict[str, list]:
"""Ichimoku Cloud — comprehensive trend indicator.
Returns dict with keys:
@@ -562,10 +568,19 @@ def ichimoku(candles: list[dict]) -> dict[str, list]:
- ``senkou_b``: Leading Span B (cloud top/bottom)
- ``chikou``: Lagging Span
All values aligned to candle index. First 51 entries are None.
All values aligned to candle index. First `senkou_b_period - 1`
entries are None.
(fix pp) The 9/26/52/26 defaults are Goichi Hosoda's original values,
designed around Japan's historical 6-day trading week — there's no
inherent reason they suit crypto's 24/7 markets, but changing them
without empirical validation would just swap one unverified guess for
another. Parameterized here (defaults unchanged) so a future walk-
forward comparison against crypto-scaled values can be run without
touching this function.
"""
n = len(candles)
if n < 52:
if n < senkou_b_period:
return {"tenkan": [None] * n, "kijun": [None] * n,
"senkou_a": [None] * n, "senkou_b": [None] * n,
"chikou": [None] * n}
@@ -580,28 +595,28 @@ def ichimoku(candles: list[dict]) -> dict[str, list]:
senkou_b: list[Optional[float]] = [None] * n
chikou: list[Optional[float]] = [None] * n
for i in range(8, n):
tenkan[i] = (max(highs[i - 8 : i + 1]) + min(lows[i - 8 : i + 1])) / 2.0
for i in range(tenkan_period - 1, n):
tenkan[i] = (max(highs[i - tenkan_period + 1 : i + 1]) + min(lows[i - tenkan_period + 1 : i + 1])) / 2.0
for i in range(25, n):
kijun[i] = (max(highs[i - 25 : i + 1]) + min(lows[i - 25 : i + 1])) / 2.0
for i in range(kijun_period - 1, n):
kijun[i] = (max(highs[i - kijun_period + 1 : i + 1]) + min(lows[i - kijun_period + 1 : i + 1])) / 2.0
# Senkou spans: shift forward by 26
for i in range(25, n):
# Senkou spans: shift forward by `displacement`
for i in range(kijun_period - 1, n):
if tenkan[i] is not None and kijun[i] is not None:
sa = (tenkan[i] + kijun[i]) / 2.0
if i + 26 < n:
senkou_a[i + 26] = sa
if i + displacement < n:
senkou_a[i + displacement] = sa
for i in range(51, n):
sb = (max(highs[i - 51 : i + 1]) + min(lows[i - 51 : i + 1])) / 2.0
if i + 26 < n:
senkou_b[i + 26] = sb
for i in range(senkou_b_period - 1, n):
sb = (max(highs[i - senkou_b_period + 1 : i + 1]) + min(lows[i - senkou_b_period + 1 : i + 1])) / 2.0
if i + displacement < n:
senkou_b[i + displacement] = sb
# Chikou: current close plotted 26 periods back
# Chikou: current close plotted `displacement` periods back
for i in range(n):
if i + 26 < n:
chikou[i] = closes[i + 26]
if i + displacement < n:
chikou[i] = closes[i + displacement]
return {
"tenkan": tenkan,
@@ -742,24 +757,45 @@ def _find_pivot_lows_levels(prices: list[float], left: int = 3, right: int = 3)
return pivots
# (fix uu) BOS used a fixed 0.3% break-confirmation buffer regardless of
# the symbol's actual volatility — too wide for a calm major (BTC/ETH
# 15m ATR% often well under 0.3%, missing real breaks) and too narrow for
# a volatile altcoin (routinely whipsawing past 0.3% on noise). Scaling
# the buffer by the symbol's own current ATR% adapts it the same way
# fix (kk) adapted the "volatile" regime threshold.
_BOS_ATR_BUFFER_MULT = 0.15
_BOS_DEFAULT_BUFFER_PCT = 0.003 # fallback (0.3%) when ATR% isn't available
def _detect_bos(
swing_highs: list[Optional[float]],
swing_lows: list[Optional[float]],
prices: list[float],
atr_pct: Optional[float] = None,
) -> Optional[str]:
"""Detect Break of Structure (BOS).
Bullish BOS: price breaks above the most recent swing high.
Bearish BOS: price breaks below the most recent swing low.
Returns \"BULLISH\", \"BEARISH\", or None.
`atr_pct` (current ATR as % of price), if given, scales the break-
confirmation buffer to the symbol's own volatility instead of a fixed
0.3% for every symbol — see _BOS_ATR_BUFFER_MULT above.
"""
recent_highs = [sh for sh in swing_highs if sh is not None][-3:]
recent_lows = [sl for sl in swing_lows if sl is not None][-3:]
current_price = prices[-1]
if recent_highs and current_price > max(recent_highs) * 1.003:
buffer_pct = (
(atr_pct / 100.0) * _BOS_ATR_BUFFER_MULT
if atr_pct is not None and atr_pct > 0
else _BOS_DEFAULT_BUFFER_PCT
)
if recent_highs and current_price > max(recent_highs) * (1 + buffer_pct):
return "BULLISH"
if recent_lows and current_price < min(recent_lows) * 0.997:
if recent_lows and current_price < min(recent_lows) * (1 - buffer_pct):
return "BEARISH"
return None
@@ -854,6 +890,7 @@ def _detect_order_blocks(
def market_structure(
candles: list[dict],
pivot_lookback: int = 3,
atr_pct: Optional[float] = None,
) -> dict:
"""Comprehensive Market Structure analysis (SMC).
@@ -866,6 +903,10 @@ def market_structure(
- ``trend``: \"BULLISH\", \"BEARISH\", or \"NEUTRAL\"
- ``last_swing_high``: most recent swing high price
- ``last_swing_low``: most recent swing low price
`atr_pct` (current ATR as % of price), if given, adapts BOS's break-
confirmation buffer to this symbol's own volatility — see
`_detect_bos`/_BOS_ATR_BUFFER_MULT.
"""
n = len(candles)
if n < 20:
@@ -882,7 +923,7 @@ def market_structure(
swing_highs = _find_pivot_highs_levels(prices, pivot_lookback, pivot_lookback)
swing_lows = _find_pivot_lows_levels(prices, pivot_lookback, pivot_lookback)
bos = _detect_bos(swing_highs, swing_lows, prices)
bos = _detect_bos(swing_highs, swing_lows, prices, atr_pct=atr_pct)
choch = _detect_choch(swing_highs, swing_lows, prices)
obs = _detect_order_blocks(candles, lookback=min(40, n))
@@ -1371,9 +1412,20 @@ def mfi(candles: list[dict], period: int = 14) -> list[Optional[float]]:
# FVG — Fair Value Gap (SMC inefficiency)
# ======================================================================
# (fix qq) Any gap, however tiny, used to count as a valid FVG — on a 15m
# chart that's noise: a gap worth a few ticks carries none of the
# "unfilled institutional order" significance the concept is meant to
# capture, and votes in signal_scoring.py just as strongly as a
# meaningful gap. Requiring the gap to be at least this fraction of the
# symbol's own current ATR filters that out; falls back to no size filter
# (previous behavior) when ATR% isn't supplied.
_FVG_MIN_GAP_ATR_MULT = 0.1
def detect_fvg(
candles: list[dict],
lookback: int = 30,
atr_pct: Optional[float] = None,
) -> tuple[Optional[str], Optional[float], Optional[float]]:
"""Detect the most recent Fair Value Gap (imbalance / inefficiency).
@@ -1382,6 +1434,11 @@ def detect_fvg(
A Bearish FVG occurs when the high of candle i is lower than the
low of candle i+2 (gap down — unfilled sell orders).
`atr_pct` (current ATR as % of price), if given, rejects gaps smaller
than `_FVG_MIN_GAP_ATR_MULT` of it — see the module-level comment
above (fix qq). Without it, any nonzero gap still counts, same as
before this parameter existed.
Returns (fvg_type, gap_high, gap_low):
- fvg_type: "BULLISH", "BEARISH", or None
- gap_high: upper bound of the gap
@@ -1406,17 +1463,28 @@ def detect_fvg(
if c0_low > c2_high:
gap_high = c0_low
gap_low = c2_high
if _fvg_gap_too_small(gap_high, gap_low, atr_pct):
continue
return "BULLISH", gap_high, gap_low
# Bearish FVG: C0 high < C2 low → gap down
if c0_high < c2_low:
gap_high = c2_low
gap_low = c0_high
if _fvg_gap_too_small(gap_high, gap_low, atr_pct):
continue
return "BEARISH", gap_high, gap_low
return None, None, None
def _fvg_gap_too_small(gap_high: float, gap_low: float, atr_pct: Optional[float]) -> bool:
if atr_pct is None or atr_pct <= 0 or gap_low <= 0:
return False
gap_pct = (gap_high - gap_low) / gap_low * 100.0
return gap_pct < atr_pct * _FVG_MIN_GAP_ATR_MULT
# ======================================================================
# Candlestick Pattern Recognition (single vote from 30+ patterns)
# ======================================================================
backend/tests/test_indicator_service.py
+198
View File
@@ -29,6 +29,11 @@ from app.services.indicator_service import (
_DEFAULT_VOLATILE_THRESHOLD_PCT,
_MIN_VOLATILE_THRESHOLD_PCT,
_MAX_VOLATILE_THRESHOLD_PCT,
_detect_bos,
_BOS_DEFAULT_BUFFER_PCT,
_BOS_ATR_BUFFER_MULT,
detect_fvg,
ichimoku,
)
@@ -402,3 +407,196 @@ class TestPivotCausalConsistency:
confirmable = len(candles) - 3
assert ms_after["swing_highs"][:confirmable] == ms_before["swing_highs"][:confirmable]
assert ms_after["swing_lows"][:confirmable] == ms_before["swing_lows"][:confirmable]
class TestBosAtrBuffer:
"""Regression tests for fix (uu): BOS used a fixed 0.3% break-
confirmation buffer for every symbol — too wide for a calm major
(missing real breaks) and too narrow for a volatile altcoin (false
breaks on noise). The buffer now scales with the symbol's own current
ATR%, falling back to the original fixed 0.3% when ATR% isn't given.
"""
def test_no_atr_pct_falls_back_to_fixed_buffer(self):
swing_highs = [None, None, 100.0]
swing_lows = [None, None, 90.0]
# Just barely above the old fixed 0.3% buffer -> BULLISH.
just_over = 100.0 * (1 + _BOS_DEFAULT_BUFFER_PCT) + 0.01
just_under = 100.0 * (1 + _BOS_DEFAULT_BUFFER_PCT) - 0.01
assert _detect_bos(swing_highs, swing_lows, [just_over]) == "BULLISH"
assert _detect_bos(swing_highs, swing_lows, [just_under]) is None
def test_high_atr_pct_widens_the_buffer(self):
"""A break that would confirm under the fixed 0.3% buffer must NOT
confirm yet when the symbol's own ATR% implies a much wider
'normal' move (avoiding a false break on a volatile symbol)."""
swing_highs = [None, None, 100.0]
swing_lows = [None, None, 90.0]
price_just_over_fixed_buffer = 100.0 * (1 + _BOS_DEFAULT_BUFFER_PCT) + 0.01
assert _detect_bos(swing_highs, swing_lows, [price_just_over_fixed_buffer]) == "BULLISH"
# atr_pct=10 -> buffer = 10/100 * _BOS_ATR_BUFFER_MULT = 1.5%, far
# wider than the fixed 0.3% — the same price no longer confirms.
assert _detect_bos(swing_highs, swing_lows, [price_just_over_fixed_buffer], atr_pct=10.0) is None
def test_low_atr_pct_narrows_the_buffer(self):
"""A calm symbol's own ATR% implies a narrower 'normal' move than
the fixed 0.3% — a break should confirm sooner (closer to the
actual swing level) than the fixed buffer would allow."""
swing_highs = [None, None, 100.0]
swing_lows = [None, None, 90.0]
# atr_pct=0.2 -> buffer = 0.2/100 * 0.15 = 0.03%, much tighter
# than the fixed 0.3%.
price_within_fixed_buffer_but_past_atr_buffer = 100.0 * (1 + _BOS_DEFAULT_BUFFER_PCT / 2)
assert _detect_bos(swing_highs, swing_lows, [price_within_fixed_buffer_but_past_atr_buffer]) is None
assert _detect_bos(
swing_highs, swing_lows, [price_within_fixed_buffer_but_past_atr_buffer], atr_pct=0.2,
) == "BULLISH"
def test_bearish_break_also_scales_with_atr(self):
swing_highs = [None, None, 100.0]
swing_lows = [None, None, 90.0]
price_just_under_fixed_buffer = 90.0 * (1 - _BOS_DEFAULT_BUFFER_PCT) - 0.01
assert _detect_bos(swing_highs, swing_lows, [price_just_under_fixed_buffer]) == "BEARISH"
assert _detect_bos(swing_highs, swing_lows, [price_just_under_fixed_buffer], atr_pct=10.0) is None
def test_market_structure_threads_atr_pct_into_bos(self, monkeypatch):
"""market_structure() must actually pass its atr_pct argument
through to _detect_bos rather than silently ignoring it."""
import app.services.indicator_service as indicator_service_module
received = {}
def spy_detect_bos(swing_highs, swing_lows, prices, atr_pct=None):
received["atr_pct"] = atr_pct
return None
monkeypatch.setattr(indicator_service_module, "_detect_bos", spy_detect_bos)
prices = [100.0 + math.sin(i / 3.0) * 10 + (i % 5) for i in range(30)]
candles = [candle(p + 0.5, p - 0.5, p) for p in prices]
market_structure(candles, pivot_lookback=3, atr_pct=7.5)
assert received["atr_pct"] == 7.5
class TestDetectFvgAtrFilter:
"""Regression tests for fix (qq): any nonzero Fair Value Gap used to
count as valid regardless of size — noisy on a 15m chart, where a
gap worth a few ticks carries none of the "unfilled institutional
order" significance the concept is meant to capture. Gaps smaller
than `_FVG_MIN_GAP_ATR_MULT` x ATR% are now rejected when `atr_pct`
is supplied; without it, behavior is unchanged from before this fix.
"""
def _bullish_gap_candles(self, c0_low: float, c2_high: float):
# detect_fvg only reads high/low of c0 and c2 — the middle candle's
# values are never inspected, only its presence in the sequence.
filler = candle(1000, 1, 500) # engulfs everything -> can never itself form a gap
c0 = candle(c0_low + 1, c0_low, c0_low + 0.5)
c1 = candle(c0_low - 0.1, c2_high + 0.1, c0_low - 0.5)
c2 = candle(c2_high, c2_high - 1, c2_high - 0.5)
return [filler, filler, c0, c1, c2]
def test_gap_detected_without_atr_pct_regardless_of_size(self):
candles = self._bullish_gap_candles(c0_low=100.1, c2_high=100.0)
fvg_type, gap_high, gap_low = detect_fvg(candles)
assert fvg_type == "BULLISH"
assert gap_high == pytest.approx(100.1)
assert gap_low == pytest.approx(100.0)
def test_tiny_gap_rejected_when_atr_pct_given(self):
# gap = (100.1-100.0)/100.0 = 0.1% of price; atr_pct=5 -> minimum
# required = 5 * _FVG_MIN_GAP_ATR_MULT(0.1) = 0.5% -> too small.
candles = self._bullish_gap_candles(c0_low=100.1, c2_high=100.0)
fvg_type, gap_high, gap_low = detect_fvg(candles, atr_pct=5.0)
assert fvg_type is None
assert gap_high is None and gap_low is None
def test_large_enough_gap_accepted_with_atr_pct(self):
# gap = (102.0-100.0)/100.0 = 2% of price; atr_pct=5 -> minimum
# required 0.5% -> comfortably passes.
candles = self._bullish_gap_candles(c0_low=102.0, c2_high=100.0)
fvg_type, gap_high, gap_low = detect_fvg(candles, atr_pct=5.0)
assert fvg_type == "BULLISH"
assert gap_high == pytest.approx(102.0)
assert gap_low == pytest.approx(100.0)
def test_bearish_gap_also_filtered_by_atr(self):
# Bearish: C0 high < C2 low -> gap down.
filler = candle(1000, 1, 500) # engulfs everything -> can never itself form a gap
c0 = candle(100.0, 99.0, 99.5) # c0_high = 100.0
c1 = candle(101.0, 100.5, 100.8)
c2 = candle(101.0, 100.1, 100.5) # c2_low = 100.1 -> gap 0.1% of 100.1
tiny_gap_candles = [filler, filler, c0, c1, c2]
fvg_type, _, _ = detect_fvg(tiny_gap_candles, atr_pct=5.0)
assert fvg_type is None
fvg_type_no_atr, gap_high, gap_low = detect_fvg(tiny_gap_candles)
assert fvg_type_no_atr == "BEARISH"
assert gap_high == pytest.approx(100.1)
assert gap_low == pytest.approx(100.0)
class TestIchimokuConfigurablePeriods:
"""Regression tests for fix (pp): Ichimoku's tenkan/kijun/senkou_b
periods and displacement are now parameters (defaulting to the
original 9/26/52/26) instead of hardcoded, so a future walk-forward
comparison against crypto-scaled periods can be run without editing
this function. The defaults themselves are intentionally unchanged —
picking new ones needs empirical validation, not a guess.
"""
def _flat_range_candles(self, n: int):
# high == low == close for every candle -> for a strictly
# increasing close series, max(highs) over any trailing window is
# just the window's last close, and min(lows) is the window's
# first close, making the expected tenkan/kijun/senkou_b values
# trivial to hand-verify.
closes = [float(i) for i in range(1, n + 1)]
return closes, [candle(c, c, c) for c in closes]
def test_default_periods_match_original_9_26_52_26(self):
closes, candles = self._flat_range_candles(100)
result = ichimoku(candles)
i = 60 # well past every warmup period
assert result["tenkan"][i] == pytest.approx((closes[i] + closes[i - 8]) / 2.0)
assert result["kijun"][i] == pytest.approx((closes[i] + closes[i - 25]) / 2.0)
# senkou_a/senkou_b are shifted forward by displacement=26.
expected_senkou_a = (result["tenkan"][i] + result["kijun"][i]) / 2.0
assert result["senkou_a"][i + 26] == pytest.approx(expected_senkou_a)
assert result["senkou_b"][i + 26] == pytest.approx((closes[i] + closes[i - 51]) / 2.0)
assert result["chikou"][i] == pytest.approx(closes[i + 26])
def test_insufficient_data_for_default_periods_returns_all_none(self):
_closes, candles = self._flat_range_candles(51) # senkou_b_period=52, one short
result = ichimoku(candles)
assert all(v is None for v in result["tenkan"])
assert all(v is None for v in result["senkou_b"])
def test_custom_shorter_periods_need_less_warmup_and_change_values(self):
closes, candles = self._flat_range_candles(30)
default_needs_more_data = ichimoku(candles) # n=30 < senkou_b_period=52
assert all(v is None for v in default_needs_more_data["tenkan"])
custom = ichimoku(candles, tenkan_period=3, kijun_period=5, senkou_b_period=10, displacement=4)
i = 15
assert custom["tenkan"][i] == pytest.approx((closes[i] + closes[i - 2]) / 2.0)
assert custom["kijun"][i] == pytest.approx((closes[i] + closes[i - 4]) / 2.0)
assert custom["senkou_b"][i + 4] == pytest.approx((closes[i] + closes[i - 9]) / 2.0)
assert custom["chikou"][i] == pytest.approx(closes[i + 4])
def test_custom_periods_produce_different_cloud_than_defaults(self):
_closes, candles = self._flat_range_candles(100)
default_result = ichimoku(candles)
custom_result = ichimoku(candles, tenkan_period=5, kijun_period=13, senkou_b_period=26, displacement=13)
i = 70
assert custom_result["tenkan"][i] != pytest.approx(default_result["tenkan"][i])
assert custom_result["kijun"][i] != pytest.approx(default_result["kijun"][i])