diff --git a/app/bars/aggregator.py b/app/bars/aggregator.py new file mode 100644 index 0000000..048e62f --- /dev/null +++ b/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 diff --git a/app/bars/session.py b/app/bars/session.py new file mode 100644 index 0000000..e8fab41 --- /dev/null +++ b/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()) diff --git a/app/market/base.py b/app/market/base.py index 4c17552..93102f2 100644 --- a/app/market/base.py +++ b/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: ... diff --git a/app/market/replay.py b/app/market/replay.py index b619930..51fbdf4 100644 --- a/app/market/replay.py +++ b/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 diff --git a/app/market/stream.py b/app/market/stream.py index 3d5d2d8..9c0ea11 100644 --- a/app/market/stream.py +++ b/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 history(self.symbol, tf, None, None, range_=range_) + bars = await source.history(self.symbol, tf, None, None, range_=range_) for bar in bars: await self._emit(bar) diff --git a/app/runtime.py b/app/runtime.py index e720544..6f8d0ca 100644 --- a/app/runtime.py +++ b/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 queue in self.subscribers.copy(): - if queue.full(): - queue.get_nowait() - queue.put_nowait(bar) + for aggregated in self.aggregator.update(bar): + self.store.put(aggregated) + for queue in self.subscribers.copy(): + if queue.full(): + queue.get_nowait() + queue.put_nowait(aggregated) async def start(self) -> asyncio.Task: try: - await self.stream.seed( - seed_source(self.settings), Timeframe.M1, self.settings.seed_1m_range - ) + source = seed_source(self.settings) + 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 diff --git a/static/app.js b/static/app.js index e2f428b..94ed2eb 100644 --- a/static/app.js +++ b/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'); diff --git a/static/index.html b/static/index.html index 274960c..552d89f 100644 --- a/static/index.html +++ b/static/index.html @@ -18,13 +18,13 @@
{{ status.symbol || 'ES=F' }}{{ price == null ? '—' : price.toFixed(2) }}
-
+
FEED {{ status.stream }} LAST BAR {{ barAge }} - HELD {{ status.bars_held?.['1m'] || 0 }} + HELD {{ status.bars_held?.[timeframe] || 0 }} {{ timeframe }}