from app.analysis.bar_space import index_at, price_in_bar_space, price_in_timeframe_space from app.analysis.levels import Level, LevelKind, Side from app.bars.models import Timeframe MINUTE = 60 def sloped(anchor_t: int, anchor_p: float, last_t: int, last_p: float) -> Level: slope = (last_p - anchor_p) / (last_t - anchor_t) return Level( "ml", LevelKind.MANUAL, Timeframe.M1, Side.SUPPORT, 1, 1, "line", anchor_t, anchor_p, slope, None, 0, anchor_t, last_t, False, False, ) def test_index_is_linear_when_bars_are_evenly_spaced(): times = [i * MINUTE for i in range(10)] assert index_at(times, 0) == 0 assert index_at(times, 5 * MINUTE) == 5 assert index_at(times, int(2.5 * MINUTE)) == 2.5 def test_a_gap_costs_one_index_however_long_it_is(): # Two bars either side of a weekend are adjacent on screen. times = [0, MINUTE, MINUTE + 49 * 3600, MINUTE + 49 * 3600 + MINUTE] assert index_at(times, MINUTE) == 1 assert index_at(times, MINUTE + 49 * 3600) == 2 def test_line_stays_straight_in_bar_space_across_a_gap(): # Bars: ten minutes, a 49-hour weekend, then ten more. weekend = 49 * 3600 times = [i * MINUTE for i in range(10)] + [10 * MINUTE + weekend + i * MINUTE for i in range(10)] level = sloped(times[0], 100.0, times[9], 109.0) # 1 point per bar # Extending nine further bars must add nine more points, gap notwithstanding. assert price_in_bar_space(level, times, times[18]) == 118.0 # Clock-based pricing would have run away during the halt. assert level.price_at(times[18]) > 3000 def test_flat_anchors_return_the_anchor_price(): times = [i * MINUTE for i in range(5)] level = sloped(0, 100.0, MINUTE, 100.0) assert price_in_bar_space(level, times, times[4]) == 100.0 def test_a_30m_line_does_not_change_slope_when_1m_history_starts_after_its_anchors(): half_hour = 30 * MINUTE weekend = 49 * 3600 source_times = [ 0, half_hour, half_hour + weekend, 2 * half_hour + weekend, 3 * half_hour + weekend, 4 * half_hour + weekend, ] line = sloped(source_times[0], 100.0, source_times[3], 103.0) line.tf = Timeframe.M30 # This is the browser's failure mode: its 1m window starts after both # anchors, so legacy edge extrapolation invents a different coordinate # system. Source geometry still advances one point per 30m bar. target = source_times[5] truncated_minutes = list(range(source_times[4], target + MINUTE, MINUTE)) assert price_in_timeframe_space(line, source_times, Timeframe.M30, target) == 105.0 assert price_in_bar_space(line, truncated_minutes, target) != 105.0 def test_complete_nested_30m_and_1m_grids_agree(): half_hour = 30 * MINUTE source_times = [i * half_hour for i in range(12)] minute_times = list(range(0, source_times[-1] + MINUTE, MINUTE)) line = sloped(source_times[1], 100.0, source_times[5], 104.0) line.tf = Timeframe.M30 target = source_times[9] assert price_in_timeframe_space(line, source_times, Timeframe.M30, target) == 108.0 assert price_in_bar_space(line, minute_times, target) == 108.0 def test_a_partial_30m_bar_advances_normally_before_a_weekend(): half_hour = 30 * MINUTE weekend = 49 * 3600 source_times = [0, half_hour, half_hour + weekend] line = sloped(0, 100.0, half_hour, 101.0) line.tf = Timeframe.M30 assert price_in_timeframe_space( line, source_times, Timeframe.M30, half_hour - MINUTE, ) == 100.0 + 29 / 30 assert price_in_timeframe_space( line, source_times, Timeframe.M30, half_hour + weekend, ) == 102.0 def test_source_geometry_refuses_to_invent_history_before_its_first_bar(): half_hour = 30 * MINUTE source_times = [half_hour, 2 * half_hour, 3 * half_hour] line = sloped(0, 100.0, half_hour, 101.0) line.tf = Timeframe.M30 assert price_in_timeframe_space(line, source_times, Timeframe.M30, 3 * half_hour) is None def test_a_future_endpoint_uses_the_same_source_bucket_projection_as_the_browser(): half_hour = 30 * MINUTE source_times = [0, half_hour, 2 * half_hour] line = sloped(0, 100.0, 10 * half_hour, 110.0) line.tf = Timeframe.M30 assert price_in_timeframe_space( line, source_times, Timeframe.M30, 2 * half_hour, ) == 102.0