87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
from app.analysis.levels import Side
|
|
from app.analysis.manual_lines import ManualLine
|
|
from app.bars.models import Bar, Timeframe
|
|
from app.config import Settings
|
|
from app.runtime import Runtime
|
|
|
|
|
|
MINUTE = 60
|
|
|
|
|
|
def bar(tf: Timeframe, t: int) -> Bar:
|
|
return Bar(tf, t, 100, 101, 99, 100, 1, True, "/ES", "test")
|
|
|
|
|
|
def add_weekend_line(runtime: Runtime) -> tuple[ManualLine, list[int], int]:
|
|
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 = runtime.manual_lines.add(ManualLine(
|
|
id="ml_30m", tf=Timeframe.M30, side=Side.SUPPORT,
|
|
anchor_t=source_times[0], anchor_p=100,
|
|
slope=3 / (source_times[3] - source_times[0]),
|
|
last_t=source_times[3], created_at=source_times[3],
|
|
))
|
|
for t in source_times:
|
|
runtime.store.put(bar(Timeframe.M30, t))
|
|
target = source_times[-1]
|
|
for t in range(source_times[-2], target + MINUTE, MINUTE):
|
|
runtime.store.put(bar(Timeframe.M1, t))
|
|
return line, source_times, target
|
|
|
|
|
|
def test_runtime_prices_a_30m_line_in_30m_space_when_1m_anchors_are_off_window(tmp_path):
|
|
runtime = Runtime(Settings(
|
|
manual_lines_path=tmp_path / "lines.json",
|
|
trendline_source_geometry=True,
|
|
))
|
|
add_weekend_line(runtime)
|
|
|
|
runtime.rebuild_levels()
|
|
|
|
level = next(value for value in runtime.levels if value.id == "ml_30m")
|
|
assert level.current_p == 105.0
|
|
assert level.geometry_resolved is True
|
|
|
|
|
|
def test_rollback_gate_restores_legacy_1m_pricing(tmp_path):
|
|
runtime = Runtime(Settings(
|
|
manual_lines_path=tmp_path / "lines.json",
|
|
trendline_source_geometry=False,
|
|
))
|
|
add_weekend_line(runtime)
|
|
|
|
runtime.rebuild_levels()
|
|
|
|
level = next(value for value in runtime.levels if value.id == "ml_30m")
|
|
assert level.current_p != 105.0
|
|
assert level.geometry_resolved is True
|
|
|
|
|
|
def test_a_line_without_its_source_anchor_history_cannot_cluster_or_alert(tmp_path):
|
|
runtime = Runtime(Settings(
|
|
manual_lines_path=tmp_path / "lines.json",
|
|
trendline_source_geometry=True,
|
|
))
|
|
line = runtime.manual_lines.add(ManualLine(
|
|
id="ml_old", tf=Timeframe.M30, side=Side.SUPPORT,
|
|
anchor_t=0, anchor_p=100, slope=1 / 1800,
|
|
last_t=1800, created_at=1800,
|
|
))
|
|
runtime.store.put(bar(Timeframe.M30, 3600))
|
|
runtime.store.put(bar(Timeframe.M30, 5400))
|
|
runtime.store.put(bar(Timeframe.M1, 5400))
|
|
runtime.price = 102
|
|
runtime.atr15 = 10
|
|
|
|
runtime.rebuild_levels()
|
|
|
|
level = next(value for value in runtime.levels if value.id == line.id)
|
|
assert level.current_p is None
|
|
assert level.geometry_resolved is False
|
|
assert all(level.id not in {member.id for member in cluster.members}
|
|
for cluster in runtime.clusters)
|