"""Resistance-cluster breakout: a validated-band alternative to Turtle's
naive N-day-high/low channel (issue #36).
Ported from `../moirai`'s unmerged `n_hit_resistance_breakthrough`
(`sup_res.py`, branch `learn-graphing-prediction`) -- a research prototype
that was never wired into moirai's own framework, but whose construction is a
real refinement over `TurtleStrategy`'s Donchian channel:
1. Find swing highs: a candle whose High equals the rolling max over a
CENTERED `alpha`-bar window (a fractal detector -- `local_extrema` in the
source), never the current/last bar, which has no future bars to confirm
it with.
2. Take the `n` most recent swing highs and compute their mean High.
3. VALIDATE the level is real: if any of those `n` highs deviates from the
mean by more than half the configured band width, the highs are too
scattered to call a real resistance level -- no signal, regardless of
where the price sits. This is the piece Turtle's single N-day extreme
cannot express at all: Turtle's channel is real by construction (it is
just "the highest high"), so it has no way to say "there have been highs
near here, but they don't agree with each other."
4. Otherwise the resistance band is `[mean - half_width, mean + half_width]`.
5. BREAKOUT: the latest close is the first one past `band_high + buffer`
since the band's own most recent swing high -- never a re-break. A
candle that already closed past the threshold earlier (while the level
was already valid) means today's close above it is old news, not a new
signal; Turtle's channel has no equivalent guard and fires again on every
bar the price happens to stay above the old high.
Only the resistance (upside) side existed in the source. The downside/exit
half -- swing LOWS, a support band, and a breakDOWN threshold -- is built
here by mirroring the same construction onto `Bar.low`, so this strategy has
an entry/exit shape comparable to Turtle's entry_period/exit_period channel:
BUY on a fresh resistance breakout while flat, SELL on a fresh support
breakdown while holding.
**Design decision not specified by the source: `region_width` and
`breakout_buffer` are expressed here as PERCENTAGES OF THE CLUSTER'S OWN
MEAN (`region_width_pct`, `breakout_buffer_pct`), not the source's raw
absolute-price parameters.** The source was a single-symbol research
prototype; this app trades symbols from roughly $1 to $500 in the same
configured universe, and one fixed dollar band cannot be simultaneously
sane for both ends of that range. A mean-relative percentage is the same
scale-free convention this codebase already uses elsewhere for exactly this
reason (`trail_pct`/`floor_pct` on `TrailingStopFloorStrategy`,
`reuse_price_band_pct` on `LlmStrategy`).
**`n`, `alpha`, `region_width_pct`, and `breakout_buffer_pct` are
conservative starting values, not validated ones** -- the source algorithm
was never tuned against a real dataset. See `config/strategies.yaml` for the
reasoning behind each chosen default; a future backtest sweep (the same kind
`docs/ideas.md` already flags `rsi_revert`'s `oversold` needed) should
revisit all four before this strategy is ever considered for `mode: trading`.
Ships `mode: shadow`. `turtle_20_10` is untouched -- this is an additional,
comparable alternative, not a replacement, so the shadow-portfolio simulator
(issue #33) can measure which construction actually performs better.
"""
from collections.abc import Sequence
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from typing import ClassVar
from trader.domain import Bar, Position
from trader.errors import ConfigError
from trader.strategies.base import Action, Signal
__all__ = ["ResistanceBreakoutStrategy"]
_HUNDRED = Decimal(100)
#: A cluster of 1 swing point has no deviation to validate against.
_MIN_N = 2
#: `alpha // 2` bars of context on each side of a candidate swing point; below
#: this the centered window has no context bar at all (radius 0).
_MIN_ALPHA = 3
def _to_decimal_param(
strategy_id: str, name: str, value: object, *, minimum: Decimal | None = None
) -> Decimal:
"""A configured YAML number as `Decimal`, never via `float()`.
Same conversion discipline as `TrailingStopFloorStrategy._to_decimal_param`
and `LlmStrategy`'s own handling of `reuse_price_band_pct`: YAML parses a
bare decimal literal (`2.0`) as a Python `float` before this code ever
sees it, so the one thing this function guarantees is that *this* code
never calls `float()` on it -- `Decimal(str(value))` is exact where
`Decimal(value)` on a float would carry that float's binary-fraction
error into a number that gates money. Kept as its own small copy rather
than a shared import, matching the existing per-module convention (both
of the functions named above do the same).
"""
if isinstance(value, bool):
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a number, got {value!r} (bool)."
)
if isinstance(value, Decimal):
candidate = value
elif isinstance(value, (int, float, str)):
try:
candidate = Decimal(str(value))
except InvalidOperation:
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a number, got {value!r}."
) from None
else:
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a number, got {value!r}."
)
# `Decimal("NaN")`/`Decimal("Infinity")` both parse without error and then
# make every comparison downstream either always-False or always-True.
if not candidate.is_finite():
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be a finite number, got {value!r}."
)
if minimum is not None and candidate < minimum:
raise ConfigError(
f"Strategy {strategy_id!r}: {name} must be at least {minimum}, got {value!r}."
)
return candidate
@dataclass(frozen=True, slots=True)
class _ClusterLevel:
"""A validated resistance-or-support band over the `n` most recent
swing points, plus the index of the most recent one (needed by the
freshness guard)."""
mean: Decimal
band_low: Decimal
band_high: Decimal
last_swing_index: int
def _swing_points(
bars: Sequence[Bar], alpha: int, *, use_high: bool
) -> list[tuple[int, Decimal]]:
"""Indices of local extrema: a candle whose High (or Low) equals the
rolling max (or min) over a CENTERED `alpha`-bar window.
`radius = alpha // 2` bars on each side, so a candidate needs
`i - radius >= 0` and `i + radius <= len(bars) - 1`. In particular the
last bar in `bars` (the current decision bar) is never a candidate --
confirming it would require bars this strategy has not been shown yet,
which is exactly what `Strategy.evaluate`'s contract forbids.
One function serves both directions via `use_high`, rather than two
near-identical copies, since the shape is otherwise identical.
"""
radius = alpha // 2
n = len(bars)
points: list[tuple[int, Decimal]] = []
for i in range(radius, n - radius):
window = bars[i - radius : i + radius + 1]
if use_high:
candidate = bars[i].high
extreme = max(b.high for b in window)
else:
candidate = bars[i].low
extreme = min(b.low for b in window)
if candidate == extreme:
points.append((i, candidate))
return points
def _cluster_level(
swings: list[tuple[int, Decimal]], n: int, region_width_pct: Decimal
) -> _ClusterLevel | None:
"""The validated band over the `n` most recent swing points, or `None`.
`None` covers two different failure modes the caller does not need to
tell apart beyond its own HOLD-reason text: fewer than `n` swing points
exist yet, or the `n` most recent ones are real but too scattered --
at least one deviates from their mean by MORE than half the configured
band width -- to call a real level. Deliberately `>`, not `>=`: a swing
point sitting exactly on the band edge is still inside the band.
"""
if len(swings) < n:
return None
recent = swings[-n:]
prices = [p for _, p in recent]
mean = sum(prices, Decimal(0)) / n
half_width = mean * region_width_pct / _HUNDRED / 2
if any(abs(p - mean) > half_width for p in prices):
return None
last_index = recent[-1][0]
return _ClusterLevel(
mean=mean,
band_low=mean - half_width,
band_high=mean + half_width,
last_swing_index=last_index,
)
def _is_fresh_break(
bars: Sequence[Bar], last_swing_index: int, threshold: Decimal, *, above: bool
) -> bool:
"""Whether `bars[-1]`'s close is the FIRST close past `threshold` since
the level's most recent swing point -- never a re-break.
Scans every candle strictly between `last_swing_index` and the current
(last) bar; if any of them already closed past `threshold`, the current
close being past it too is a re-break, not a new one, and this returns
`False` regardless of how far past `threshold` the current close sits.
This is the guard a naive N-day-high check has no equivalent of, and why
it re-fires on every later bar the price happens to stay past the old
extreme.
"""
current_index = len(bars) - 1
for j in range(last_swing_index + 1, current_index):
already_past = bars[j].close > threshold if above else bars[j].close < threshold
if already_past:
return False
close = bars[current_index].close
return close > threshold if above else close < threshold
[docs]
@dataclass(frozen=True, slots=True)
class ResistanceBreakoutStrategy:
"""BUY on a fresh resistance-cluster breakout while flat; SELL on a
fresh support-cluster breakdown while holding. See the module docstring
for the full algorithm and the reasoning behind the default parameters.
"""
id: str
#: Swing points required before a resistance/support level is
#: considered real. 3 is the classic technical-analysis minimum -- 2
#: touches is easily coincidental.
n: int = 3
#: Width (in bars) of the centered window a candle must be the extreme
#: of to count as a swing point; `alpha // 2` bars of context on each
#: side. 5 needs 2 bars either side: enough to reject single-bar noise
#: without an unrealistically long confirmation lag.
alpha: int = 5
#: Total resistance/support band width, as a percent of the cluster's
#: own mean (see the module docstring for why this is relative, not the
#: source's absolute `region_width`). 2.0% is a conservative starting
#: guess.
region_width_pct: Decimal = Decimal("2.0")
#: How far past the band edge a close must sit before it counts as a
#: breakout/breakdown, as a percent of the cluster's own mean. 0.5% is a
#: conservative starting guess -- enough to filter a single noisy tick
#: at the band edge without requiring a large move.
breakout_buffer_pct: Decimal = Decimal("0.5")
# `Strategy.backtestable`'s Protocol default is not structurally
# inherited by a class that doesn't subclass `Strategy` -- see
# `TurtleStrategy`. `ClassVar` keeps this out of the generated
# `__init__`/`__repr__`/`__eq__`.
backtestable: ClassVar[bool] = True
def __post_init__(self) -> None:
if not isinstance(self.n, int) or isinstance(self.n, bool) or self.n < _MIN_N:
raise ConfigError(
f"Strategy {self.id!r}: n must be an integer >= {_MIN_N} (a "
"cluster of 1 swing point has no deviation to validate), got "
f"{self.n!r}."
)
if (
not isinstance(self.alpha, int)
or isinstance(self.alpha, bool)
or self.alpha < _MIN_ALPHA
):
raise ConfigError(
f"Strategy {self.id!r}: alpha must be an integer >= {_MIN_ALPHA} "
"(at least one bar of context on each side of a candidate "
f"swing point, via radius = alpha // 2), got {self.alpha!r}."
)
region_width_pct = _to_decimal_param(
self.id, "region_width_pct", self.region_width_pct, minimum=Decimal(0)
)
if region_width_pct <= 0:
raise ConfigError(
f"Strategy {self.id!r}: region_width_pct must be > 0, got "
f"{region_width_pct!r}."
)
breakout_buffer_pct = _to_decimal_param(
self.id, "breakout_buffer_pct", self.breakout_buffer_pct, minimum=Decimal(0)
)
object.__setattr__(self, "region_width_pct", region_width_pct)
object.__setattr__(self, "breakout_buffer_pct", breakout_buffer_pct)
[docs]
def warmup_bars(self) -> int:
# A heuristic for sizing lookback windows (backtests, ranking's
# `build_comparables`), not a guarantee: swing points recur roughly
# every `alpha` bars in practice, so `n` of them takes roughly
# `n * alpha` bars to accumulate, plus `radius` trailing bars of
# context to confirm the most recent one. Real markets may need more
# or fewer -- `evaluate()` itself is the actual authority, and holds
# until it has actually found `n` valid swing points, no matter what
# this returns.
radius = self.alpha // 2
return self.n * self.alpha + radius
[docs]
def evaluate(self, bars: Sequence[Bar], position: Position | None) -> Signal:
if not bars:
return Signal(Action.HOLD, "No bars available to evaluate.")
close = bars[-1].close
indicators: dict[str, Decimal | float] = {"close": close}
resistance = _cluster_level(
_swing_points(bars, self.alpha, use_high=True), self.n, self.region_width_pct
)
support = _cluster_level(
_swing_points(bars, self.alpha, use_high=False), self.n, self.region_width_pct
)
if resistance is not None:
indicators["resistance_mean"] = resistance.mean
indicators["resistance_band_low"] = resistance.band_low
indicators["resistance_band_high"] = resistance.band_high
if support is not None:
indicators["support_mean"] = support.mean
indicators["support_band_low"] = support.band_low
indicators["support_band_high"] = support.band_high
if position is None:
if resistance is None:
return Signal(
Action.HOLD,
"Flat, and no valid resistance cluster yet -- either fewer "
f"than {self.n} swing highs have formed, or the {self.n} "
"most recent ones are too scattered to call a real level.",
indicators,
)
buffer = resistance.mean * self.breakout_buffer_pct / _HUNDRED
threshold = resistance.band_high + buffer
indicators["breakout_threshold"] = threshold
if _is_fresh_break(bars, resistance.last_swing_index, threshold, above=True):
return Signal(
Action.BUY,
f"Close {close} is the first close above the resistance "
f"band [{resistance.band_low}, {resistance.band_high}] "
f"plus buffer -> {threshold} since the band's most recent "
"swing high.",
indicators,
)
return Signal(
Action.HOLD,
f"Flat, and close {close} has not made a fresh break above "
f"the resistance breakout threshold {threshold}.",
indicators,
)
if support is None:
return Signal(
Action.HOLD,
"Holding, and no valid support cluster yet -- either fewer "
f"than {self.n} swing lows have formed, or the {self.n} most "
"recent ones are too scattered to call a real level.",
indicators,
)
buffer = support.mean * self.breakout_buffer_pct / _HUNDRED
threshold = support.band_low - buffer
indicators["breakdown_threshold"] = threshold
if _is_fresh_break(bars, support.last_swing_index, threshold, above=False):
return Signal(
Action.SELL,
f"Close {close} is the first close below the support band "
f"[{support.band_low}, {support.band_high}] minus buffer -> "
f"{threshold} since the band's most recent swing low.",
indicators,
)
return Signal(
Action.HOLD,
f"Holding, and close {close} has not made a fresh break below "
f"the support breakdown threshold {threshold}.",
indicators,
)