90 lines
3.6 KiB
Python
90 lines
3.6 KiB
Python
"""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
|
|
|
|
|
|
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 and not (allow_future and lower == len(times) - 1):
|
|
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
|