Chris Amow
f5ba154c98
Prod already owned the empty 1m slots; source index still treated the bars on either side as adjacent. Also post a geometry SNAPDBG on ?diag=1.
137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
from app.analysis.bar_space import (
|
|
fill_short_gaps, index_at, price_in_bar_space, price_in_timeframe_space,
|
|
timeframe_index_at,
|
|
)
|
|
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
|
|
|
|
|
|
def test_a_short_1m_hole_still_advances_one_index_per_minute():
|
|
times = list(range(0, 10 * MINUTE, MINUTE)) + list(range(19 * MINUTE, 30 * MINUTE, MINUTE))
|
|
filled = fill_short_gaps(times, Timeframe.M1)
|
|
assert 10 * MINUTE in filled
|
|
assert timeframe_index_at(filled, 19 * MINUTE, Timeframe.M1) == 19
|
|
line = sloped(0, 100.0, 30 * MINUTE, 130.0)
|
|
assert price_in_timeframe_space(line, times, Timeframe.M1, 19 * MINUTE) == 119.0
|
|
|
|
|
|
def test_a_weekend_is_not_filled_as_short_gaps():
|
|
weekend = 49 * 3600
|
|
times = [0, MINUTE, MINUTE + weekend, MINUTE + weekend + MINUTE]
|
|
assert fill_short_gaps(times, Timeframe.M1) == times
|
|
|
|
|
|
def test_a_missing_5m_bucket_inside_a_ten_minute_hole_is_filled():
|
|
five = 5 * MINUTE
|
|
times = [0, five, 3 * five]
|
|
filled = fill_short_gaps(times, Timeframe.M5)
|
|
assert filled == [0, five, 2 * five, 3 * five]
|