chart/app/analysis/bar_space.py

"""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

# Same bound as ConfluenceChart.MAX_INTRADAY_GAP_SECONDS: short tape holes
# occupy empty slots on the chart, so source index must count them too.
MAX_INTRADAY_GAP_SECONDS = 30 * 60


def fill_short_gaps(times: list[int], tf: Timeframe) -> list[int]:
    """Insert missing bucket opens inside short intraday holes.

    A 9-minute 1m hole is eight empty columns on screen. Without these times,
    source index treats the two surrounding bars as adjacent and the line
    goes flat across the hole. Settlement and weekends stay compressed.
    """
    if tf is Timeframe.D1 or len(times) < 2:
        return times
    step = tf.seconds
    if step <= 0:
        return times
    filled = [times[0]]
    for time in times[1:]:
        prev = filled[-1]
        gap = time - prev
        if step < gap <= MAX_INTRADAY_GAP_SECONDS:
            filled.extend(range(prev + step, time, step))
        filled.append(time)
    return filled


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.
    times = fill_short_gaps(times, tf)
    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