Implement M2 session-aware aggregation

app/bars/aggregator.py

@@ -0,0 +1,57 @@
+from dataclasses import replace
+
+from app.bars.models import Bar, Timeframe
+from app.bars.session import bucket_start
+
+
+class Aggregator:
+    def __init__(self, timeframes: list[Timeframe] | None = None):
+        self.timeframes = timeframes or list(Timeframe)
+        self.forming: dict[Timeframe, Bar] = {}
+
+    @staticmethod
+    def _can_aggregate(source: Timeframe, target: Timeframe) -> bool:
+        if source is Timeframe.M1:
+            return True
+        if source is Timeframe.H1:
+            return target in (Timeframe.H1, Timeframe.H4, Timeframe.D1)
+        return source is target
+
+    def update(self, incoming: Bar) -> list[Bar]:
+        emitted: list[Bar] = []
+        for tf in self.timeframes:
+            if not self._can_aggregate(incoming.tf, tf):
+                continue
+            if tf is incoming.tf:
+                emitted.append(replace(incoming))
+                continue
+
+            start = bucket_start(incoming.t, tf)
+            current = self.forming.get(tf)
+            if current is not None and start < current.t:
+                continue
+            if current is None or start > current.t:
+                if current is not None:
+                    emitted.append(replace(current, closed=True))
+                current = Bar(
+                    tf=tf,
+                    t=start,
+                    o=incoming.o,
+                    h=incoming.h,
+                    l=incoming.l,
+                    c=incoming.c,
+                    v=incoming.v,
+                    closed=False,
+                    symbol=incoming.symbol,
+                    source=incoming.source,
+                )
+                self.forming[tf] = current
+            else:
+                current.h = max(current.h, incoming.h)
+                current.l = min(current.l, incoming.l)
+                current.c = incoming.c
+                current.v += incoming.v
+                current.symbol = incoming.symbol
+                current.source = incoming.source
+            emitted.append(replace(current))
+        return emitted

app/bars/session.py

@@ -0,0 +1,30 @@
+from datetime import datetime, time, timedelta
+from zoneinfo import ZoneInfo
+
+from app.bars.models import Timeframe
+
+UTC = ZoneInfo("UTC")
+EASTERN = ZoneInfo("America/New_York")
+SESSION_OPEN = time(18, 0)
+
+
+def _session_open_local(current: datetime) -> datetime:
+    session_date = current.date() if current.timetz().replace(tzinfo=None) >= SESSION_OPEN else current.date() - timedelta(days=1)
+    return datetime.combine(session_date, SESSION_OPEN, EASTERN)
+
+
+def bucket_start(t: int, tf: Timeframe) -> int:
+    if tf not in (Timeframe.H4, Timeframe.D1):
+        return (t // tf.seconds) * tf.seconds
+
+    current = datetime.fromtimestamp(t, UTC).astimezone(EASTERN)
+    session_open = _session_open_local(current)
+    if tf is Timeframe.D1:
+        return int(session_open.timestamp())
+
+    # CME's 4h anchors are wall-clock ET anchors. This intentionally makes the
+    # DST-transition bucket three or five elapsed hours instead of shifting it.
+    elapsed_wall = current.replace(tzinfo=None) - session_open.replace(tzinfo=None)
+    bucket_hours = int(elapsed_wall.total_seconds() // 14400) * 4
+    local_start = session_open.replace(tzinfo=None) + timedelta(hours=bucket_hours)
+    return int(local_start.replace(tzinfo=EASTERN).timestamp())

app/market/base.py

@@ -15,6 +15,8 @@ class MarketDataSource(Protocol):
         tf: Timeframe,
         start: int | None,
         end: int | None,
+        *,
+        range_: str | None = None,
     ) -> list[Bar]: ...
 
     def supports_stream(self) -> bool: ...

app/market/replay.py

@@ -31,6 +31,8 @@ class ReplaySource:
         tf: Timeframe,
         start: int | None = None,
         end: int | None = None,
+        *,
+        range_: str | None = None,
     ) -> list[Bar]:
         return [
             bar

app/market/stream.py

@@ -26,8 +26,7 @@ class StreamService:
     ) -> None:
         if source is None or not source.supports_history():
             return
-        history = getattr(source, "history")
+        bars = await source.history(self.symbol, tf, None, None, range_=range_)
-        bars = await history(self.symbol, tf, None, None, range_=range_)
         for bar in bars:
             await self._emit(bar)
 

app/runtime.py

@@ -2,6 +2,7 @@ import asyncio
 from dataclasses import dataclass, field
 
 from app.bars.models import Bar, Timeframe
+from app.bars.aggregator import Aggregator
 from app.bars.store import InMemoryBarStore
 from app.config import Settings
 from app.market.factory import live_source, seed_source
@@ -14,24 +15,27 @@ class Runtime:
     store: InMemoryBarStore = field(init=False)
     stream: StreamService = field(init=False)
     subscribers: set[asyncio.Queue[Bar]] = field(default_factory=set)
+    aggregator: Aggregator = field(init=False)
 
     def __post_init__(self) -> None:
         self.store = InMemoryBarStore(self.settings.max_bars_per_tf)
+        self.aggregator = Aggregator(self.settings.enabled_timeframes)
         self.stream = StreamService(live_source(self.settings), self.settings.yahoo_symbol)
         self.stream.add_handler(self.on_bar)
 
     async def on_bar(self, bar: Bar) -> None:
-        self.store.put(bar)
+        for aggregated in self.aggregator.update(bar):
-        for queue in self.subscribers.copy():
+            self.store.put(aggregated)
-            if queue.full():
+            for queue in self.subscribers.copy():
-                queue.get_nowait()
+                if queue.full():
-            queue.put_nowait(bar)
+                    queue.get_nowait()
+                queue.put_nowait(aggregated)
 
     async def start(self) -> asyncio.Task:
         try:
-            await self.stream.seed(
+            source = seed_source(self.settings)
-                seed_source(self.settings), Timeframe.M1, self.settings.seed_1m_range
+            await self.stream.seed(source, Timeframe.H1, self.settings.seed_1h_range)
-            )
+            await self.stream.seed(source, Timeframe.M1, self.settings.seed_1m_range)
         except Exception:
             # A transient seed failure must not prevent the live stream or UI starting.
             pass

static/app.js

@@ -4,6 +4,8 @@ createApp({
   setup() {
     const status = ref({ stream: 'disconnected', bars_held: {} });
     const price = ref(null);
+    const timeframe = ref('1m');
+    const timeframes = ['1m', '2m', '5m', '15m', '30m', '1h', '4h', '1d'];
     const now = ref(Date.now());
     let chartApi = null;
     let socket = null;
@@ -23,7 +25,7 @@ createApp({
     function connect() {
       const protocol = location.protocol === 'https:' ? 'wss' : 'ws';
       socket = new WebSocket(`${protocol}://${location.host}/ws`);
-      socket.onopen = () => socket.send(JSON.stringify({ type: 'subscribe', tf: '1m' }));
+      socket.onopen = () => socket.send(JSON.stringify({ type: 'subscribe', tf: timeframe.value }));
       socket.onmessage = ({ data }) => {
         const message = JSON.parse(data);
         if (message.type === 'snapshot') {
@@ -41,6 +43,13 @@ createApp({
       };
     }
 
+    function selectTimeframe(tf) {
+      timeframe.value = tf;
+      if (socket?.readyState === WebSocket.OPEN) {
+        socket.send(JSON.stringify({ type: 'subscribe', tf }));
+      }
+    }
+
     onMounted(() => {
       chartApi = new ConfluenceChart();
       chartApi.create(document.getElementById('chart'));
@@ -54,6 +63,6 @@ createApp({
       if (chartApi) chartApi.destroy();
     });
 
-    return { status, price, barAge };
+    return { status, price, barAge, timeframe, timeframes, selectTimeframe };
   },
 }).mount('#app');

static/index.html

@@ -18,13 +18,13 @@
       <section class="chart-shell">
         <div class="chart-head">
           <div><span class="symbol">{{ status.symbol || 'ES=F' }}</span><span class="price">{{ price == null ? '—' : price.toFixed(2) }}</span></div>
-          <div class="timeframes"><button class="active">1m</button></div>
+          <div class="timeframes"><button v-for="tf in timeframes" :key="tf" :class="{active: timeframe === tf}" @click="selectTimeframe(tf)">{{ tf }}</button></div>
         </div>
         <div id="chart"></div>
         <div class="statusbar">
           <span>FEED <b>{{ status.stream }}</b></span>
           <span>LAST BAR <b>{{ barAge }}</b></span>
-          <span>HELD <b>{{ status.bars_held?.['1m'] || 0 }}</b></span>
+          <span>HELD <b>{{ status.bars_held?.[timeframe] || 0 }} {{ timeframe }}</b></span>
         </div>
       </section>
       <aside>

static/style.css

@@ -9,10 +9,11 @@ h1 { margin:0; font-size:22px; letter-spacing:-1px; } h1 strong { color:var(--ac
 .status { text-transform:uppercase; color:var(--muted); font-size:11px; }.status i { display:inline-block; width:7px; height:7px; border-radius:50%; background:var(--red); margin-right:8px; }.status.connected i,.status.replay i { background:var(--green); box-shadow:0 0 9px var(--green); }
 main { display:grid; grid-template-columns:minmax(0, 1fr) 300px; gap:16px; }
 .chart-shell,aside { background:var(--panel); border:1px solid var(--line); }
-.chart-head { height:56px; padding:10px 14px; display:flex; align-items:center; justify-content:space-between; border-bottom:1px solid var(--line); }
+.chart-head { min-height:56px; padding:10px 14px; display:flex; align-items:center; justify-content:space-between; gap:12px; border-bottom:1px solid var(--line); }
 .symbol { font-weight:700; margin-right:14px; }.price { color:var(--accent); font-size:19px; }
 button { border:1px solid var(--line); background:transparent; color:var(--muted); padding:6px 11px; font:inherit; cursor:pointer; }button.active { color:var(--bg); background:var(--accent); border-color:var(--accent); }
+.timeframes { display:flex; flex-wrap:wrap; justify-content:flex-end; }.timeframes button+button { border-left:0; }
 #chart { height:calc(100vh - 190px); min-height:420px; }
 .statusbar { min-height:34px; display:flex; align-items:center; gap:24px; padding:6px 13px; border-top:1px solid var(--line); color:var(--muted); font-size:10px; }.statusbar b { color:var(--fg); text-transform:uppercase; }
 aside { padding:16px; }h2 { margin:0 0 12px; color:var(--muted); font-size:11px; text-transform:uppercase; letter-spacing:1.3px; }h2:not(:first-child) { margin-top:30px; }.empty { border-left:2px solid var(--line); padding:10px 12px; color:var(--muted); font-size:11px; }
-@media (max-width:850px) { #app { padding:10px; }main { grid-template-columns:1fr; }#chart { height:55vh; min-height:360px; }aside { min-height:180px; }header { height:54px; } }
+@media (max-width:850px) { #app { padding:10px; }main { grid-template-columns:1fr; }#chart { height:55vh; min-height:360px; }aside { min-height:180px; }header { height:54px; }.chart-head { align-items:flex-start; flex-direction:column; }.timeframes { justify-content:flex-start; }.timeframes button { padding:5px 8px; } }

tests/test_aggregator.py

@@ -0,0 +1,43 @@
+from app.bars.aggregator import Aggregator
+from app.bars.models import Bar, Timeframe
+
+
+def minute(t: int, price: float = 100, volume: int = 1) -> Bar:
+    return Bar(Timeframe.M1, t, price, price + 1, price - 1, price + 0.5, volume, True, "ES=F", "replay")
+
+
+def test_aggregates_all_timeframes_and_closes_on_later_bucket():
+    aggregator = Aggregator()
+    first = aggregator.update(minute(0, 100, 2))
+    second = aggregator.update(minute(60, 102, 3))
+    boundary = aggregator.update(minute(900, 105, 4))
+
+    assert {bar.tf for bar in first} == set(Timeframe)
+    forming_15m = [bar for bar in second if bar.tf is Timeframe.M15][-1]
+    assert (forming_15m.o, forming_15m.h, forming_15m.l, forming_15m.c, forming_15m.v) == (
+        100,
+        103,
+        99,
+        102.5,
+        5,
+    )
+    bars_15m = [bar for bar in boundary if bar.tf is Timeframe.M15]
+    assert [(bar.t, bar.closed) for bar in bars_15m] == [(0, True), (900, False)]
+
+
+def test_gap_closes_previous_bucket_without_synthesizing_empty_bars():
+    aggregator = Aggregator([Timeframe.M5])
+    aggregator.update(minute(0))
+    output = aggregator.update(minute(1800))
+
+    assert [(bar.t, bar.closed) for bar in output] == [(0, True), (1800, False)]
+
+
+def test_replay_is_deterministic():
+    tape = [minute(t, 100 + index) for index, t in enumerate(range(0, 3600, 60))]
+
+    def run():
+        aggregator = Aggregator()
+        return [bar.to_dict() for source in tape for bar in aggregator.update(source)]
+
+    assert run() == run()

tests/test_session.py

@@ -0,0 +1,63 @@
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from app.bars.models import Timeframe
+from app.bars.session import bucket_start
+
+UTC = ZoneInfo("UTC")
+ET = ZoneInfo("America/New_York")
+
+
+def epoch(value: str, zone=UTC) -> int:
+    return int(datetime.fromisoformat(value).replace(tzinfo=zone).timestamp())
+
+
+@pytest.mark.parametrize(
+    ("value", "tf", "expected"),
+    [
+        ("2026-08-09T22:00:00", Timeframe.D1, "2026-08-09T22:00:00"),
+        ("2026-08-10T16:37:00", Timeframe.D1, "2026-08-09T22:00:00"),
+        ("2026-08-14T20:59:00", Timeframe.D1, "2026-08-13T22:00:00"),
+        ("2026-08-10T21:30:00", Timeframe.D1, "2026-08-09T22:00:00"),
+        ("2026-08-10T22:00:00", Timeframe.D1, "2026-08-10T22:00:00"),
+        ("2026-08-10T01:59:00", Timeframe.H4, "2026-08-09T22:00:00"),
+        ("2026-08-10T02:00:00", Timeframe.H4, "2026-08-10T02:00:00"),
+        ("2026-08-10T17:59:00", Timeframe.H4, "2026-08-10T14:00:00"),
+    ],
+)
+def test_session_boundaries(value, tf, expected):
+    assert bucket_start(epoch(value), tf) == epoch(expected)
+
+
+def test_intraday_buckets_use_utc_boundaries():
+    assert bucket_start(epoch("2026-08-10T12:37:45"), Timeframe.M15) == epoch(
+        "2026-08-10T12:30:00"
+    )
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        # Spring forward: the 22:00 ET bucket ends at 02:00 EDT after three real hours.
+        ("2026-03-08T06:59:00", "2026-03-08T03:00:00"),
+        ("2026-03-08T07:00:00", "2026-03-08T07:00:00"),
+        # Fall back: the 22:00 ET bucket lasts five real hours and ends at 02:00 EST.
+        ("2026-11-01T06:59:00", "2026-11-01T02:00:00"),
+        ("2026-11-01T07:00:00", "2026-11-01T07:00:00"),
+    ],
+)
+def test_four_hour_wall_clock_anchor_across_dst(value, expected):
+    assert bucket_start(epoch(value), Timeframe.H4) == epoch(expected)
+
+
+@pytest.mark.parametrize(
+    ("local_value", "expected_local"),
+    [
+        ("2026-03-08T18:00:00", "2026-03-08T18:00:00"),
+        ("2026-11-01T18:00:00", "2026-11-01T18:00:00"),
+    ],
+)
+def test_sunday_open_across_dst(local_value, expected_local):
+    assert bucket_start(epoch(local_value, ET), Timeframe.D1) == epoch(expected_local, ET)