886 lines
28 KiB
Python
886 lines
28 KiB
Python
"""S/R Detector service.
|
||
|
||
Detects support/resistance levels from Volume Profile (POC/VA/HVN peaks)
|
||
and Pivot Points (prominent swing highs/lows), plus light psychological
|
||
round numbers. Scores by rejection-weighted recent touches, merges nearby
|
||
levels with ATR-adaptive tolerance, tags support/resistance, caps count,
|
||
and persists to DB.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import datetime
|
||
|
||
from sqlalchemy import delete, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.exceptions import NotFoundError, ValidationError
|
||
from app.models.sr_level import SRLevel
|
||
from app.models.ticker import Ticker
|
||
from app.services.indicator_service import (
|
||
_extract_ohlcv,
|
||
compute_atr,
|
||
compute_pivot_points,
|
||
compute_volume_profile,
|
||
)
|
||
from app.services.price_service import query_ohlcv
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tunable constants (keep detection pure / deterministic)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
DEFAULT_TOLERANCE = 0.005 # fallback when ATR unavailable; also API legacy default
|
||
|
||
VP_LOOKBACK = 252
|
||
TOUCH_LOOKBACK = 252
|
||
PIVOT_LOOKBACK = 504
|
||
PIVOT_PROMINENCE_ATR = 0.75
|
||
PIVOT_PROMINENCE_PCT = 0.006
|
||
|
||
MERGE_TOL_ATR_MULT = 0.35
|
||
MERGE_TOL_MIN = 0.004 # 0.4%
|
||
MERGE_TOL_MAX = 0.015 # 1.5%
|
||
|
||
MAX_LEVELS = 16
|
||
STRENGTH_HALF_LIFE = 60 # bars
|
||
# Raw respect score is soft-mapped to 0–100 (see _raw_to_strength).
|
||
STRENGTH_SCALE = 8.0
|
||
STRENGTH_SOFT_K = 35.0 # higher → slower approach to 100
|
||
|
||
ROUND_NUMBER_RANGE = 0.15 # ±15% of spot
|
||
ROUND_NUMBER_MAX = 8
|
||
|
||
# Base strength seed before touch scoring (method priors)
|
||
_METHOD_BASE_STRENGTH = {
|
||
"volume_profile": 12,
|
||
"pivot_point": 8,
|
||
"round_number": 4,
|
||
}
|
||
|
||
|
||
async def _get_ticker(db: AsyncSession, symbol: str) -> Ticker:
|
||
"""Look up a ticker by symbol."""
|
||
normalised = symbol.strip().upper()
|
||
result = await db.execute(select(Ticker).where(Ticker.symbol == normalised))
|
||
ticker = result.scalar_one_or_none()
|
||
if ticker is None:
|
||
raise NotFoundError(f"Ticker not found: {normalised}")
|
||
return ticker
|
||
|
||
|
||
def _slice_tail(
|
||
highs: list[float],
|
||
lows: list[float],
|
||
closes: list[float],
|
||
volumes: list[int],
|
||
lookback: int,
|
||
) -> tuple[list[float], list[float], list[float], list[int]]:
|
||
"""Return the last *lookback* bars (or all if shorter)."""
|
||
n = len(closes)
|
||
if lookback <= 0 or n <= lookback:
|
||
return highs, lows, closes, volumes
|
||
start = n - lookback
|
||
return highs[start:], lows[start:], closes[start:], volumes[start:]
|
||
|
||
|
||
def _atr_pct(
|
||
highs: list[float],
|
||
lows: list[float],
|
||
closes: list[float],
|
||
) -> float | None:
|
||
"""ATR as a fraction of last close, or None if insufficient data."""
|
||
try:
|
||
result = compute_atr(highs, lows, closes)
|
||
except ValidationError:
|
||
return None
|
||
atr = result["atr"]
|
||
last = closes[-1]
|
||
if last == 0:
|
||
return None
|
||
return atr / last
|
||
|
||
|
||
def _merge_tolerance(
|
||
highs: list[float],
|
||
lows: list[float],
|
||
closes: list[float],
|
||
tolerance: float | None,
|
||
) -> float:
|
||
"""Resolve merge tolerance: explicit value or ATR-adaptive clamp."""
|
||
if tolerance is not None:
|
||
return tolerance
|
||
atr_frac = _atr_pct(highs, lows, closes)
|
||
if atr_frac is None:
|
||
return DEFAULT_TOLERANCE
|
||
return max(MERGE_TOL_MIN, min(MERGE_TOL_MAX, MERGE_TOL_ATR_MULT * atr_frac))
|
||
|
||
|
||
def _bar_respect_weight(
|
||
price_level: float,
|
||
high: float,
|
||
low: float,
|
||
close: float,
|
||
prev_close: float | None,
|
||
tolerance: float,
|
||
) -> float:
|
||
"""Weight for how much a bar *respects* a level (not mere occupancy).
|
||
|
||
Only bars whose high/low **probes near the level** and closes away from
|
||
that extreme count as rejections. Full-range pass-throughs score near zero.
|
||
"""
|
||
tol = price_level * tolerance if price_level != 0 else tolerance
|
||
if tol <= 0:
|
||
tol = abs(price_level) * DEFAULT_TOLERANCE if price_level else DEFAULT_TOLERANCE
|
||
# Tight probe band: ~0.4% of price (capped), not 2× merge tolerance
|
||
band = min(max(abs(price_level) * 0.004, tol * 0.35), abs(price_level) * 0.008)
|
||
if band <= 0:
|
||
band = abs(price_level) * 0.004 if price_level else 0.01
|
||
|
||
if high + band < price_level or low - band > price_level:
|
||
return 0.0
|
||
|
||
bar_range = high - low
|
||
# Support test: low probes near level, close recovers above
|
||
support_test = abs(low - price_level) <= band and close > price_level
|
||
if support_test and bar_range > 0:
|
||
support_test = (close - low) >= 0.25 * bar_range
|
||
# Resistance test: high probes near level, close rejects below
|
||
resist_test = abs(high - price_level) <= band and close < price_level
|
||
if resist_test and bar_range > 0:
|
||
resist_test = (high - close) >= 0.25 * bar_range
|
||
|
||
if support_test or resist_test:
|
||
return 1.0
|
||
|
||
# Clear directional pass-through — barely counts
|
||
if (
|
||
prev_close is not None
|
||
and (prev_close - price_level) * (close - price_level) < 0
|
||
and low < price_level - tol
|
||
and high > price_level + tol
|
||
):
|
||
return 0.1
|
||
|
||
return 0.0
|
||
|
||
|
||
def _raw_to_strength(raw: float) -> int:
|
||
"""Map unbounded raw score to 0–100 with soft saturation (no hard pin)."""
|
||
if raw <= 0:
|
||
return 0
|
||
# 1 - e^(-raw/k): raw=k → ~63, 2k → ~86, 3k → ~95
|
||
return max(0, min(100, int(round(100.0 * (1.0 - math.exp(-raw / STRENGTH_SOFT_K))))))
|
||
|
||
|
||
def _respect_evidence(
|
||
price_level: float,
|
||
highs: list[float],
|
||
lows: list[float],
|
||
closes: list[float],
|
||
tolerance: float = DEFAULT_TOLERANCE,
|
||
base: int = 0,
|
||
half_life: float = STRENGTH_HALF_LIFE,
|
||
lookback: int = TOUCH_LOOKBACK,
|
||
cooldown: int = 3,
|
||
) -> dict[str, float | int | None]:
|
||
"""Return rejection evidence and its soft-mapped strength.
|
||
|
||
*cooldown* bars after a full rejection are ignored so multi-day chop at a
|
||
level counts as one test cluster, not N identical rejections.
|
||
"""
|
||
n = len(closes)
|
||
if n == 0:
|
||
return {
|
||
"strength": _raw_to_strength(float(base)),
|
||
"rejection_count": 0,
|
||
"last_rejection_age": None,
|
||
"weighted_respects": 0.0,
|
||
}
|
||
|
||
start = max(0, n - lookback) if lookback > 0 else 0
|
||
weighted = 0.0
|
||
rejection_count = 0
|
||
last_rejection_age: int | None = None
|
||
next_ok = start
|
||
for i in range(start, n):
|
||
age = n - 1 - i
|
||
decay = 0.5 ** (age / half_life) if half_life > 0 else 1.0
|
||
prev = closes[i - 1] if i > 0 else None
|
||
w = _bar_respect_weight(
|
||
price_level, highs[i], lows[i], closes[i], prev, tolerance
|
||
)
|
||
if w >= 0.9:
|
||
if i < next_ok:
|
||
continue
|
||
weighted += decay * w
|
||
rejection_count += 1
|
||
if last_rejection_age is None or age < last_rejection_age:
|
||
last_rejection_age = age
|
||
next_ok = i + max(cooldown, 1)
|
||
elif w > 0:
|
||
weighted += decay * w
|
||
|
||
raw = float(base) + weighted * STRENGTH_SCALE
|
||
return {
|
||
"strength": _raw_to_strength(raw),
|
||
"rejection_count": rejection_count,
|
||
"last_rejection_age": last_rejection_age,
|
||
"weighted_respects": round(weighted, 6),
|
||
}
|
||
|
||
|
||
def _strength_from_respects(
|
||
price_level: float,
|
||
highs: list[float],
|
||
lows: list[float],
|
||
closes: list[float],
|
||
tolerance: float = DEFAULT_TOLERANCE,
|
||
base: int = 0,
|
||
half_life: float = STRENGTH_HALF_LIFE,
|
||
lookback: int = TOUCH_LOOKBACK,
|
||
cooldown: int = 3,
|
||
) -> int:
|
||
"""Compatibility wrapper returning only the evidence-derived strength."""
|
||
return int(_respect_evidence(
|
||
price_level,
|
||
highs,
|
||
lows,
|
||
closes,
|
||
tolerance,
|
||
base,
|
||
half_life,
|
||
lookback,
|
||
cooldown,
|
||
)["strength"])
|
||
|
||
|
||
def _round_number_candidates(
|
||
current_price: float,
|
||
range_pct: float = ROUND_NUMBER_RANGE,
|
||
max_count: int = ROUND_NUMBER_MAX,
|
||
) -> list[float]:
|
||
"""Psychological round levels near spot (cheap order-magnet candidates)."""
|
||
if current_price <= 0:
|
||
return []
|
||
|
||
if current_price < 5:
|
||
steps = [0.5, 1.0]
|
||
elif current_price < 20:
|
||
steps = [1.0, 5.0]
|
||
elif current_price < 100:
|
||
steps = [5.0, 10.0, 25.0]
|
||
elif current_price < 500:
|
||
steps = [10.0, 25.0, 50.0, 100.0]
|
||
else:
|
||
steps = [25.0, 50.0, 100.0, 250.0]
|
||
|
||
lo = current_price * (1.0 - range_pct)
|
||
hi = current_price * (1.0 + range_pct)
|
||
found: set[float] = set()
|
||
|
||
for step in steps:
|
||
if step <= 0:
|
||
continue
|
||
# Start at first multiple at or below lo
|
||
k = math.floor(lo / step)
|
||
while True:
|
||
level = round(k * step, 4)
|
||
if level > hi + step:
|
||
break
|
||
if lo <= level <= hi and level > 0:
|
||
# Skip levels that are essentially current price
|
||
if abs(level - current_price) / current_price > 0.001:
|
||
found.add(level)
|
||
k += 1
|
||
if k > 1_000_000: # safety
|
||
break
|
||
|
||
ordered = sorted(found, key=lambda p: abs(p - current_price))
|
||
return ordered[:max_count]
|
||
|
||
|
||
def _extract_candidate_levels(
|
||
highs: list[float],
|
||
lows: list[float],
|
||
closes: list[float],
|
||
volumes: list[int],
|
||
) -> list[tuple[float, str]]:
|
||
"""Extract candidate S/R levels from VP nodes, prominent pivots, rounds.
|
||
|
||
Returns list of (price_level, detection_method) tuples.
|
||
"""
|
||
candidates: list[tuple[float, str]] = []
|
||
if not closes:
|
||
return candidates
|
||
|
||
current_price = closes[-1]
|
||
|
||
# --- Volume profile on recent window ---
|
||
vp_h, vp_l, vp_c, vp_v = _slice_tail(
|
||
highs, lows, closes, volumes, VP_LOOKBACK
|
||
)
|
||
try:
|
||
vp = compute_volume_profile(vp_h, vp_l, vp_c, vp_v)
|
||
# Structural VP levels: POC, value-area edges, local HVN peaks.
|
||
# LVN intentionally omitted (rejection voids ≠ support/resistance lines).
|
||
for key in ("poc", "value_area_low", "value_area_high"):
|
||
price = vp.get(key)
|
||
if price is not None and price > 0:
|
||
candidates.append((float(price), "volume_profile"))
|
||
for price in vp.get("hvn", []):
|
||
candidates.append((float(price), "volume_profile"))
|
||
except ValidationError:
|
||
pass
|
||
|
||
# --- Prominent pivots on pivot lookback ---
|
||
p_h, p_l, p_c, _ = _slice_tail(highs, lows, closes, volumes, PIVOT_LOOKBACK)
|
||
atr_frac = _atr_pct(p_h, p_l, p_c)
|
||
last = p_c[-1] if p_c else current_price
|
||
if atr_frac is not None and last > 0:
|
||
prominence = max(PIVOT_PROMINENCE_ATR * atr_frac * last, PIVOT_PROMINENCE_PCT * last)
|
||
else:
|
||
prominence = PIVOT_PROMINENCE_PCT * last if last > 0 else None
|
||
|
||
try:
|
||
pp = compute_pivot_points(p_h, p_l, p_c, min_prominence=prominence)
|
||
for price in pp.get("swing_highs", []):
|
||
candidates.append((float(price), "pivot_point"))
|
||
for price in pp.get("swing_lows", []):
|
||
candidates.append((float(price), "pivot_point"))
|
||
except ValidationError:
|
||
pass
|
||
|
||
# --- Psychological round numbers near spot ---
|
||
for price in _round_number_candidates(current_price):
|
||
candidates.append((price, "round_number"))
|
||
|
||
return candidates
|
||
|
||
|
||
def _legacy_volume_profile_nodes(
|
||
highs: list[float],
|
||
lows: list[float],
|
||
closes: list[float],
|
||
volumes: list[int],
|
||
num_bins: int = 20,
|
||
) -> list[float]:
|
||
"""Reproduce the deployed pre-rewrite HVN/LVN grid for a control arm.
|
||
|
||
This intentionally retains the old span-volume double counting. It exists
|
||
only so a local research report can prove that its control still reproduces
|
||
the frozen production baseline while the live detector is being rewritten.
|
||
"""
|
||
if len(closes) < 20:
|
||
raise ValidationError(
|
||
f"Volume Profile requires at least 20 bars, got {len(closes)}"
|
||
)
|
||
price_min = min(lows)
|
||
price_max = max(highs)
|
||
if price_max == price_min:
|
||
price_max = price_min + 1.0
|
||
bin_width = (price_max - price_min) / num_bins
|
||
bins: list[float] = [0.0] * num_bins
|
||
prices = [price_min + (i + 0.5) * bin_width for i in range(num_bins)]
|
||
for i in range(len(closes)):
|
||
for b in range(num_bins):
|
||
low_edge = price_min + b * bin_width
|
||
high_edge = low_edge + bin_width
|
||
if highs[i] >= low_edge and lows[i] <= high_edge:
|
||
bins[b] += volumes[i]
|
||
average = sum(bins) / num_bins
|
||
hvn = [round(prices[i], 4) for i in range(num_bins) if bins[i] > average]
|
||
lvn = [round(prices[i], 4) for i in range(num_bins) if bins[i] < average]
|
||
return hvn + lvn
|
||
|
||
|
||
def detect_sr_levels_legacy(
|
||
highs: list[float],
|
||
lows: list[float],
|
||
closes: list[float],
|
||
volumes: list[int],
|
||
tolerance: float = DEFAULT_TOLERANCE,
|
||
*,
|
||
include_volume_profile: bool = True,
|
||
include_pivots: bool = True,
|
||
neutral_strength: bool = False,
|
||
) -> list[dict]:
|
||
"""Deployed detector plus source-isolation controls for local research.
|
||
|
||
The defaults remain the exact production control. Keyword-only switches
|
||
allow a research arm to remove one source or neutralize strength without
|
||
copying or subtly changing the legacy implementation.
|
||
"""
|
||
if not closes:
|
||
return []
|
||
|
||
candidates: list[tuple[float, str]] = []
|
||
if include_volume_profile:
|
||
try:
|
||
for price in _legacy_volume_profile_nodes(highs, lows, closes, volumes):
|
||
candidates.append((float(price), "volume_profile"))
|
||
except ValidationError:
|
||
pass
|
||
if include_pivots:
|
||
try:
|
||
pivots = compute_pivot_points(highs, lows, closes)
|
||
candidates.extend(
|
||
(float(price), "pivot_point")
|
||
for price in pivots.get("swing_highs", []) + pivots.get("swing_lows", [])
|
||
)
|
||
except ValidationError:
|
||
pass
|
||
if not candidates:
|
||
return []
|
||
|
||
total_bars = len(closes)
|
||
raw: list[dict] = []
|
||
for price, method in candidates:
|
||
tol = price * tolerance if price != 0 else tolerance
|
||
touches = sum(
|
||
1 for low, high in zip(lows, highs, strict=False)
|
||
if low - tol <= price <= high + tol
|
||
)
|
||
strength = max(0, min(100, int(round((touches / total_bars) * 500.0))))
|
||
raw.append({
|
||
"price_level": price,
|
||
"strength": strength,
|
||
"detection_method": method,
|
||
"type": "",
|
||
"sources": [method],
|
||
"rejection_count": touches,
|
||
"last_rejection_age": None,
|
||
"weighted_respects": float(touches),
|
||
})
|
||
|
||
merged: list[dict] = []
|
||
for level in sorted(raw, key=lambda row: row["price_level"]):
|
||
if not merged:
|
||
merged.append(dict(level))
|
||
continue
|
||
last = merged[-1]
|
||
ref = last["price_level"]
|
||
tol = ref * tolerance if ref != 0 else tolerance
|
||
if abs(level["price_level"] - ref) > tol:
|
||
merged.append(dict(level))
|
||
continue
|
||
last["price_level"] = round(
|
||
(last["price_level"] + level["price_level"]) / 2.0, 4
|
||
)
|
||
last["strength"] = min(100, last["strength"] + level["strength"])
|
||
sources = set(last.get("sources") or [last["detection_method"]])
|
||
sources |= set(level.get("sources") or [level["detection_method"]])
|
||
last["sources"] = sorted(sources)
|
||
last["detection_method"] = (
|
||
next(iter(sources)) if len(sources) == 1 else "merged"
|
||
)
|
||
last["rejection_count"] = max(
|
||
int(last.get("rejection_count", 0)),
|
||
int(level.get("rejection_count", 0)),
|
||
)
|
||
|
||
_tag_levels(merged, closes[-1])
|
||
if neutral_strength:
|
||
for level in merged:
|
||
level["strength"] = 50
|
||
merged.sort(key=lambda row: row["strength"], reverse=True)
|
||
return merged
|
||
|
||
|
||
def _merge_levels(
|
||
levels: list[dict],
|
||
tolerance: float = DEFAULT_TOLERANCE,
|
||
) -> list[dict]:
|
||
"""Merge levels within tolerance into consolidated levels.
|
||
|
||
Strength combines via max + partial min (avoids instant saturation) with
|
||
a confluence bonus when detection methods differ. Price is strength-weighted.
|
||
"""
|
||
if not levels:
|
||
return []
|
||
|
||
sorted_levels = sorted(levels, key=lambda x: x["price_level"])
|
||
merged: list[dict] = []
|
||
|
||
for level in sorted_levels:
|
||
if not merged:
|
||
entry = dict(level)
|
||
sources = level.get("sources") or [level["detection_method"]]
|
||
entry["sources"] = sorted(set(sources))
|
||
merged.append(entry)
|
||
continue
|
||
|
||
last = merged[-1]
|
||
ref_price = last["price_level"]
|
||
tol = ref_price * tolerance if ref_price != 0 else tolerance
|
||
|
||
if abs(level["price_level"] - ref_price) <= tol:
|
||
s1 = last["strength"]
|
||
s2 = level["strength"]
|
||
# Soft combine — avoid merge math pinning everything at 100
|
||
combined = int(round(0.85 * max(s1, s2) + 0.15 * min(s1, s2)))
|
||
sources = set(last.get("sources") or [last["detection_method"]])
|
||
sources |= set(level.get("sources") or [level["detection_method"]])
|
||
if len(sources) > 1:
|
||
combined = min(100, combined + 5)
|
||
else:
|
||
combined = min(100, combined)
|
||
|
||
w1, w2 = max(s1, 1), max(s2, 1)
|
||
avg_price = (last["price_level"] * w1 + level["price_level"] * w2) / (w1 + w2)
|
||
|
||
if len(sources) == 1:
|
||
method = next(iter(sources))
|
||
else:
|
||
method = "merged"
|
||
|
||
last["price_level"] = round(avg_price, 4)
|
||
last["strength"] = combined
|
||
last["detection_method"] = method
|
||
last["sources"] = sorted(sources)
|
||
# Nearby candidates often describe the same price reaction, so do
|
||
# not add their rejection counts and double-count one market event.
|
||
last["rejection_count"] = max(
|
||
int(last.get("rejection_count", 0)),
|
||
int(level.get("rejection_count", 0)),
|
||
)
|
||
ages = [
|
||
age for age in (
|
||
last.get("last_rejection_age"),
|
||
level.get("last_rejection_age"),
|
||
)
|
||
if age is not None
|
||
]
|
||
last["last_rejection_age"] = min(ages) if ages else None
|
||
last["weighted_respects"] = max(
|
||
float(last.get("weighted_respects", 0.0)),
|
||
float(level.get("weighted_respects", 0.0)),
|
||
)
|
||
else:
|
||
entry = dict(level)
|
||
sources = level.get("sources") or [level["detection_method"]]
|
||
entry["sources"] = sorted(set(sources))
|
||
merged.append(entry)
|
||
|
||
return merged
|
||
|
||
|
||
def _tag_levels(
|
||
levels: list[dict],
|
||
current_price: float,
|
||
) -> list[dict]:
|
||
"""Tag each level as 'support' or 'resistance' relative to current price."""
|
||
for level in levels:
|
||
if level["price_level"] < current_price:
|
||
level["type"] = "support"
|
||
else:
|
||
level["type"] = "resistance"
|
||
return levels
|
||
|
||
|
||
def _cap_levels(
|
||
levels: list[dict],
|
||
max_levels: int = MAX_LEVELS,
|
||
) -> list[dict]:
|
||
"""Keep up to *max_levels* levels, interleaving support/resistance by strength."""
|
||
if max_levels <= 0 or len(levels) <= max_levels:
|
||
return levels
|
||
|
||
support = sorted(
|
||
[lvl for lvl in levels if lvl.get("type") == "support"],
|
||
key=lambda x: x["strength"],
|
||
reverse=True,
|
||
)
|
||
resistance = sorted(
|
||
[lvl for lvl in levels if lvl.get("type") != "support"],
|
||
key=lambda x: x["strength"],
|
||
reverse=True,
|
||
)
|
||
|
||
selected: list[dict] = []
|
||
si, ri = 0, 0
|
||
pick_support = True
|
||
while len(selected) < max_levels and (si < len(support) or ri < len(resistance)):
|
||
if pick_support:
|
||
if si < len(support):
|
||
selected.append(support[si])
|
||
si += 1
|
||
elif ri < len(resistance):
|
||
selected.append(resistance[ri])
|
||
ri += 1
|
||
else:
|
||
if ri < len(resistance):
|
||
selected.append(resistance[ri])
|
||
ri += 1
|
||
elif si < len(support):
|
||
selected.append(support[si])
|
||
si += 1
|
||
pick_support = not pick_support
|
||
|
||
selected.sort(key=lambda x: x["strength"], reverse=True)
|
||
return selected
|
||
|
||
|
||
def detect_sr_levels(
|
||
highs: list[float],
|
||
lows: list[float],
|
||
closes: list[float],
|
||
volumes: list[int],
|
||
tolerance: float | None = None,
|
||
max_levels: int = MAX_LEVELS,
|
||
) -> list[dict]:
|
||
"""Detect, score, merge, tag, and cap S/R levels from OHLCV data.
|
||
|
||
Parameters
|
||
----------
|
||
tolerance:
|
||
Relative merge tolerance. ``None`` (default) uses ATR-adaptive
|
||
tolerance clamped to [0.4%, 1.5%]. Pass an explicit fraction to override.
|
||
max_levels:
|
||
Hard cap after merge (balanced support/resistance). 0 = no cap.
|
||
|
||
Returns list of dicts with keys: price_level, type, strength,
|
||
detection_method — sorted by strength descending.
|
||
"""
|
||
if not closes:
|
||
return []
|
||
|
||
candidates = _extract_candidate_levels(highs, lows, closes, volumes)
|
||
if not candidates:
|
||
return []
|
||
|
||
current_price = closes[-1]
|
||
merge_tol = _merge_tolerance(highs, lows, closes, tolerance)
|
||
# Touch tolerance for strength: use merge tol (same price scale)
|
||
touch_tol = merge_tol
|
||
|
||
# Score each candidate on recent rejection-weighted touches
|
||
raw_levels: list[dict] = []
|
||
for price, method in candidates:
|
||
base = _METHOD_BASE_STRENGTH.get(method, 0)
|
||
evidence = _respect_evidence(
|
||
price, highs, lows, closes, touch_tol, base=base
|
||
)
|
||
raw_levels.append({
|
||
"price_level": price,
|
||
"strength": int(evidence["strength"]),
|
||
"detection_method": method,
|
||
"type": "",
|
||
"sources": [method],
|
||
"rejection_count": int(evidence["rejection_count"]),
|
||
"last_rejection_age": evidence["last_rejection_age"],
|
||
"weighted_respects": float(evidence["weighted_respects"]),
|
||
})
|
||
|
||
merged = _merge_levels(raw_levels, merge_tol)
|
||
tagged = _tag_levels(merged, current_price)
|
||
capped = _cap_levels(tagged, max_levels=max_levels)
|
||
capped.sort(key=lambda x: x["strength"], reverse=True)
|
||
return capped
|
||
|
||
|
||
def cluster_sr_zones(
|
||
levels: list[dict],
|
||
current_price: float,
|
||
tolerance: float = 0.02,
|
||
max_zones: int | None = None,
|
||
strength_mode: str = "sum",
|
||
) -> list[dict]:
|
||
"""Cluster nearby S/R levels into zones.
|
||
|
||
Returns list of zone dicts:
|
||
{
|
||
"low": float,
|
||
"high": float,
|
||
"midpoint": float,
|
||
"strength": int, # sum of constituent strengths, capped at 100
|
||
"type": "support" | "resistance",
|
||
"level_count": int,
|
||
}
|
||
"""
|
||
if not levels:
|
||
return []
|
||
|
||
if max_zones is not None and max_zones <= 0:
|
||
return []
|
||
|
||
# 1. Sort levels by price_level ascending
|
||
sorted_levels = sorted(levels, key=lambda x: x["price_level"])
|
||
|
||
# 2. Greedy merge into clusters
|
||
clusters: list[list[dict]] = []
|
||
current_cluster: list[dict] = [sorted_levels[0]]
|
||
|
||
for level in sorted_levels[1:]:
|
||
# Compute current cluster midpoint
|
||
prices = [lvl["price_level"] for lvl in current_cluster]
|
||
cluster_low = min(prices)
|
||
cluster_high = max(prices)
|
||
cluster_mid = (cluster_low + cluster_high) / 2.0
|
||
|
||
# Check if within tolerance of cluster midpoint
|
||
if cluster_mid != 0:
|
||
distance_pct = abs(level["price_level"] - cluster_mid) / cluster_mid
|
||
else:
|
||
distance_pct = abs(level["price_level"])
|
||
|
||
if distance_pct <= tolerance:
|
||
current_cluster.append(level)
|
||
else:
|
||
clusters.append(current_cluster)
|
||
current_cluster = [level]
|
||
|
||
clusters.append(current_cluster)
|
||
|
||
# 3. Compute zone for each cluster
|
||
zones: list[dict] = []
|
||
for cluster in clusters:
|
||
prices = [lvl["price_level"] for lvl in cluster]
|
||
low = min(prices)
|
||
high = max(prices)
|
||
midpoint = (low + high) / 2.0
|
||
if strength_mode == "soft":
|
||
strongest = max(int(lvl["strength"]) for lvl in cluster)
|
||
all_sources = {
|
||
source
|
||
for lvl in cluster
|
||
for source in (lvl.get("sources") or [lvl.get("detection_method", "unknown")])
|
||
}
|
||
strength = min(100, strongest + (5 if len(all_sources) > 1 else 0))
|
||
elif strength_mode == "sum":
|
||
strength = min(100, sum(int(lvl["strength"]) for lvl in cluster))
|
||
all_sources = {
|
||
source
|
||
for lvl in cluster
|
||
for source in (lvl.get("sources") or [lvl.get("detection_method", "unknown")])
|
||
}
|
||
else:
|
||
raise ValueError(f"Unsupported S/R zone strength mode: {strength_mode}")
|
||
level_count = len(cluster)
|
||
rejection_count = max(int(lvl.get("rejection_count", 0)) for lvl in cluster)
|
||
ages = [lvl.get("last_rejection_age") for lvl in cluster if lvl.get("last_rejection_age") is not None]
|
||
|
||
# 4. Tag zone type
|
||
zone_type = "support" if midpoint < current_price else "resistance"
|
||
|
||
zones.append({
|
||
"low": low,
|
||
"high": high,
|
||
"midpoint": midpoint,
|
||
"strength": strength,
|
||
"type": zone_type,
|
||
"level_count": level_count,
|
||
"sources": sorted(all_sources),
|
||
"rejection_count": rejection_count,
|
||
"last_rejection_age": min(ages) if ages else None,
|
||
})
|
||
|
||
# 5. Split into support and resistance pools, each sorted by strength desc
|
||
support_zones = sorted(
|
||
[z for z in zones if z["type"] == "support"],
|
||
key=lambda z: z["strength"],
|
||
reverse=True,
|
||
)
|
||
resistance_zones = sorted(
|
||
[z for z in zones if z["type"] == "resistance"],
|
||
key=lambda z: z["strength"],
|
||
reverse=True,
|
||
)
|
||
|
||
# 6. Interleave pick: alternate strongest from each pool
|
||
selected: list[dict] = []
|
||
limit = max_zones if max_zones is not None else len(zones)
|
||
si, ri = 0, 0
|
||
pick_support = True # start with support pool
|
||
|
||
while len(selected) < limit and (si < len(support_zones) or ri < len(resistance_zones)):
|
||
if pick_support:
|
||
if si < len(support_zones):
|
||
selected.append(support_zones[si])
|
||
si += 1
|
||
elif ri < len(resistance_zones):
|
||
selected.append(resistance_zones[ri])
|
||
ri += 1
|
||
else:
|
||
if ri < len(resistance_zones):
|
||
selected.append(resistance_zones[ri])
|
||
ri += 1
|
||
elif si < len(support_zones):
|
||
selected.append(support_zones[si])
|
||
si += 1
|
||
pick_support = not pick_support
|
||
|
||
# 7. Sort final selection by strength descending
|
||
selected.sort(key=lambda z: z["strength"], reverse=True)
|
||
|
||
return selected
|
||
|
||
|
||
async def recalculate_sr_levels(
|
||
db: AsyncSession,
|
||
symbol: str,
|
||
tolerance: float | None = None,
|
||
) -> list[SRLevel]:
|
||
"""Recalculate S/R levels for a ticker and persist to DB.
|
||
|
||
1. Fetch OHLCV data
|
||
2. Detect levels
|
||
3. Delete old levels for ticker
|
||
4. Insert new levels
|
||
5. Return new levels sorted by strength desc
|
||
"""
|
||
ticker = await _get_ticker(db, symbol)
|
||
|
||
records = await query_ohlcv(db, symbol)
|
||
if not records:
|
||
# No OHLCV data — clear any existing levels
|
||
await db.execute(
|
||
delete(SRLevel).where(SRLevel.ticker_id == ticker.id)
|
||
)
|
||
await db.commit()
|
||
return []
|
||
|
||
_, highs, lows, closes, volumes = _extract_ohlcv(records)
|
||
|
||
levels = detect_sr_levels(highs, lows, closes, volumes, tolerance)
|
||
|
||
# Delete old levels
|
||
await db.execute(
|
||
delete(SRLevel).where(SRLevel.ticker_id == ticker.id)
|
||
)
|
||
|
||
# Insert new levels
|
||
now = datetime.utcnow()
|
||
new_models: list[SRLevel] = []
|
||
for lvl in levels:
|
||
model = SRLevel(
|
||
ticker_id=ticker.id,
|
||
price_level=lvl["price_level"],
|
||
type=lvl["type"],
|
||
strength=lvl["strength"],
|
||
detection_method=lvl["detection_method"],
|
||
created_at=now,
|
||
)
|
||
db.add(model)
|
||
new_models.append(model)
|
||
|
||
await db.commit()
|
||
|
||
# Refresh to get IDs
|
||
for m in new_models:
|
||
await db.refresh(m)
|
||
|
||
return new_models
|
||
|
||
|
||
async def get_sr_levels(
|
||
db: AsyncSession,
|
||
symbol: str,
|
||
tolerance: float | None = None,
|
||
) -> list[SRLevel]:
|
||
"""Get S/R levels for a ticker, recalculating on every request (MVP).
|
||
|
||
Returns levels sorted by strength descending.
|
||
"""
|
||
return await recalculate_sr_levels(db, symbol, tolerance)
|