61 lines
2 KiB
Python
61 lines
2 KiB
Python
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from app.analysis.moving_averages import build_ma_levels, project_step
|
|
from app.bars.models import Bar, Timeframe
|
|
|
|
ET = ZoneInfo("America/New_York")
|
|
|
|
|
|
def daily_bars(count: int, last_forming: bool = False) -> list[Bar]:
|
|
start = datetime(2025, 1, 5, 18, tzinfo=ET)
|
|
return [
|
|
Bar(
|
|
Timeframe.D1,
|
|
int((start + timedelta(days=index)).timestamp()),
|
|
index + 1,
|
|
index + 2,
|
|
index,
|
|
index + 1,
|
|
100,
|
|
not (last_forming and index == count - 1),
|
|
"ES=F",
|
|
"replay",
|
|
)
|
|
for index in range(count)
|
|
]
|
|
|
|
|
|
def test_daily_sma_set_has_known_values_and_stable_ids():
|
|
bars = daily_bars(210)
|
|
definitions = {Timeframe.D1: [("sma", p) for p in (10, 20, 50, 100, 200)]}
|
|
levels = build_ma_levels({Timeframe.D1: bars}, definitions)
|
|
|
|
assert [level.id for level in levels] == [f"ma:1d:sma:{p}" for p in (10, 20, 50, 100, 200)]
|
|
assert [level.anchor_p for level in levels] == [205.5, 200.5, 185.5, 160.5, 110.5]
|
|
|
|
|
|
def test_nothing_emitted_before_closed_bar_warmup():
|
|
bars = daily_bars(200, last_forming=True)
|
|
assert build_ma_levels({Timeframe.D1: bars}, {Timeframe.D1: [("sma", 200)]}) == []
|
|
|
|
|
|
def test_forming_daily_value_is_provisional():
|
|
bars = daily_bars(201, last_forming=True)
|
|
level = build_ma_levels({Timeframe.D1: bars}, {Timeframe.D1: [("sma", 200)]})[0]
|
|
assert level.provisional is True
|
|
|
|
|
|
def test_step_projection_changes_only_at_session_boundary():
|
|
session_one = daily_bars(1)[0].t
|
|
session_two = daily_bars(2)[1].t
|
|
minutes = [
|
|
Bar(Timeframe.M1, t, 1, 1, 1, 1, 0, True, "ES=F", "replay")
|
|
for t in (session_one, session_one + 60, session_two - 60, session_two)
|
|
]
|
|
assert project_step([(session_one, 100), (session_two, 101)], minutes) == [
|
|
(session_one, 100),
|
|
(session_one + 60, 100),
|
|
(session_two - 60, 100),
|
|
(session_two, 101),
|
|
]
|