"""Positions within a bar series, rather than on a clock. A chart spaces bars evenly no matter how much time separates them: a weekend is forty-nine hours but one bar wide. So a line that looks straight is straight in *index* space, and a trendline advances per bar, not per second. Evaluating trendlines any other way makes the drawn line and the alerted price disagree — measured at 147 points across a weekend on a real /ES chart. """ from bisect import bisect_right from app.bars.models import Timeframe from app.bars.session import bucket_duration, next_bucket_start def index_at(times: list[int], t: int) -> float: """Fractional index of a timestamp within an ascending bar-time series.""" if not times: return 0.0 if len(times) == 1: return 0.0 # Outside the series there are no bars to measure against, so fall back to # the spacing at the nearest edge. if t <= times[0]: step = times[1] - times[0] return (t - times[0]) / step if step else 0.0 if t >= times[-1]: step = times[-1] - times[-2] return (len(times) - 1) + ((t - times[-1]) / step if step else 0.0) lower = bisect_right(times, t) - 1 span = times[lower + 1] - times[lower] return lower + ((t - times[lower]) / span if span else 0.0) def price_in_bar_space(level, times: list[int], t: int) -> float: """A level's price at `t`, interpolated across bars rather than seconds.""" start_index = index_at(times, level.anchor_t) end_index = index_at(times, level.last_t) if end_index == start_index: return level.anchor_p end_price = level.anchor_p + level.slope * (level.last_t - level.anchor_t) ratio = (index_at(times, t) - start_index) / (end_index - start_index) return level.anchor_p + (end_price - level.anchor_p) * ratio def timeframe_index_at( times: list[int], t: int, tf: Timeframe, *, allow_future: bool = False, ) -> float | None: """Position `t` in a timeframe's own logical bar space. Unlike ``index_at``, this never extrapolates backward from a truncated window. Within a real source bucket it advances by that bucket's normal duration, so the final minutes before a weekend do not get divided by the entire weekend gap. """ if not times: return None upper = bisect_right(times, t) if upper and times[upper - 1] == t: return float(upper - 1) lower = upper - 1 if lower < 0: return None duration = bucket_duration(times[lower], tf) elapsed = t - times[lower] if duration <= 0 or elapsed < 0: return None if elapsed > duration: if not (allow_future and lower == len(times) - 1): return None current = times[lower] index = float(lower) for _ in range(10000): following = next_bucket_start(current, tf) if t < following: active = bucket_duration(current, tf) return index + (t - current) / active if t <= current + active else None index += 1 current = following if t == current: return index return None return lower + elapsed / duration def price_in_timeframe_space( level, times: list[int], tf: Timeframe, t: int, ) -> float | None: """Price a line in the bar space of the timeframe it belongs to.""" # Endpoints may deliberately sit in the projection area. They use the same # repeated-source-bucket approximation as the browser; the live evaluation # instant itself must still belong to held source history. start_index = timeframe_index_at(times, level.anchor_t, tf, allow_future=True) end_index = timeframe_index_at(times, level.last_t, tf, allow_future=True) target_index = timeframe_index_at(times, t, tf) if start_index is None or end_index is None or target_index is None: return None if end_index == start_index: return level.anchor_p end_price = level.anchor_p + level.slope * (level.last_t - level.anchor_t) ratio = (target_index - start_index) / (end_index - start_index) return level.anchor_p + (end_price - level.anchor_p) * ratio