diff --git a/app/analysis/__init__.py b/app/analysis/__init__.py new file mode 100644 index 0000000..412767f --- /dev/null +++ b/app/analysis/__init__.py @@ -0,0 +1 @@ +"""Pure analysis engines.""" diff --git a/app/analysis/indicators.py b/app/analysis/indicators.py new file mode 100644 index 0000000..5dc267c --- /dev/null +++ b/app/analysis/indicators.py @@ -0,0 +1,40 @@ +from app.bars.models import Bar + + +def sma(values: list[float], period: int) -> list[float | None]: + if period <= 0: + raise ValueError("period must be positive") + output: list[float | None] = [None] * len(values) + total = 0.0 + for index, value in enumerate(values): + total += value + if index >= period: + total -= values[index - period] + if index >= period - 1: + output[index] = total / period + return output + + +def ema(values: list[float], period: int) -> list[float | None]: + if period <= 0: + raise ValueError("period must be positive") + output: list[float | None] = [None] * len(values) + if len(values) < period: + return output + value = sum(values[:period]) / period + output[period - 1] = value + multiplier = 2 / (period + 1) + for index in range(period, len(values)): + value = (values[index] - value) * multiplier + value + output[index] = value + return output + + +def atr(bars: list[Bar], period: int = 14) -> list[float | None]: + if period <= 0: + raise ValueError("period must be positive") + ranges: list[float] = [] + for index, bar in enumerate(bars): + previous_close = bars[index - 1].c if index else bar.c + ranges.append(max(bar.h - bar.l, abs(bar.h - previous_close), abs(bar.l - previous_close))) + return sma(ranges, period) diff --git a/app/analysis/levels.py b/app/analysis/levels.py new file mode 100644 index 0000000..228dfaf --- /dev/null +++ b/app/analysis/levels.py @@ -0,0 +1,48 @@ +from dataclasses import asdict, dataclass +from enum import Enum +from typing import Any + +from app.bars.models import Timeframe + + +class LevelKind(str, Enum): + MANUAL = "manual" + MA = "ma" + TRENDLINE = "trendline" + HORIZONTAL = "horizontal" + + +class Side(str, Enum): + SUPPORT = "support" + RESISTANCE = "resistance" + + +@dataclass(slots=True) +class Level: + id: str + kind: LevelKind + tf: Timeframe + side: Side + weight: float + score: float + label: str + anchor_t: int + anchor_p: float + slope: float + points: list[tuple[int, float]] | None + touches: int + first_t: int + last_t: int + provisional: bool + hidden: bool + period: int | None = None + + def price_at(self, t: int) -> float: + return self.anchor_p + self.slope * (t - self.anchor_t) + + def to_dict(self) -> dict[str, Any]: + value = asdict(self) + value["kind"] = self.kind.value + value["tf"] = self.tf.value + value["side"] = self.side.value + return value diff --git a/app/analysis/moving_averages.py b/app/analysis/moving_averages.py new file mode 100644 index 0000000..d9de1a9 --- /dev/null +++ b/app/analysis/moving_averages.py @@ -0,0 +1,62 @@ +from app.analysis.indicators import ema, sma +from app.analysis.levels import Level, LevelKind, Side +from app.bars.models import Bar, Timeframe +from app.config import MA_WEIGHT_FACTOR, TIMEFRAME_WEIGHT + + +MA_FUNCTIONS = {"sma": sma, "ema": ema} + + +def build_ma_levels( + bars_by_tf: dict[Timeframe, list[Bar]], + ma_sets: dict[Timeframe, list[tuple[str, int]]], +) -> list[Level]: + levels: list[Level] = [] + for tf, definitions in ma_sets.items(): + bars = bars_by_tf.get(tf, []) + closed_count = sum(bar.closed for bar in bars) + closes = [bar.c for bar in bars] + for kind, period in definitions: + if closed_count < period or kind not in MA_FUNCTIONS: + continue + values = MA_FUNCTIONS[kind](closes, period) + points = [(bar.t, value) for bar, value in zip(bars, values) if value is not None] + if not points: + continue + current = points[-1][1] + provisional = not bars[-1].closed + levels.append( + Level( + id=f"ma:{tf.value}:{kind}:{period}", + kind=LevelKind.MA, + tf=tf, + side=Side.SUPPORT if current <= bars[-1].c else Side.RESISTANCE, + weight=TIMEFRAME_WEIGHT[tf] * MA_WEIGHT_FACTOR, + score=1.0, + label=f"{tf.value} {kind.upper()}{period}", + anchor_t=points[-1][0], + anchor_p=current, + slope=0.0, + points=points, + touches=0, + first_t=points[0][0], + last_t=points[-1][0], + provisional=provisional, + hidden=False, + period=period, + ) + ) + return levels + + +def project_step(points: list[tuple[int, float]], bars: list[Bar]) -> list[tuple[int, float]]: + projected: list[tuple[int, float]] = [] + point_index = 0 + current: float | None = None + for bar in bars: + while point_index < len(points) and points[point_index][0] <= bar.t: + current = points[point_index][1] + point_index += 1 + if current is not None: + projected.append((bar.t, current)) + return projected diff --git a/app/api/routes.py b/app/api/routes.py index 00066db..bfb4bc2 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -31,3 +31,15 @@ def bars(request: Request, tf: str = "1m", limit: int = Query(500, ge=1, le=5000 raise HTTPException(400, "Unknown timeframe") from exc values = request.app.state.runtime.store.get(timeframe, limit) return {"tf": timeframe.value, "bars": [bar.to_dict() for bar in values]} + + +@router.get("/levels") +def levels(request: Request, tf: str = "all"): + values = request.app.state.runtime.levels + if tf != "all": + try: + timeframe = Timeframe(tf) + except ValueError as exc: + raise HTTPException(400, "Unknown timeframe") from exc + values = [level for level in values if level.tf is timeframe] + return {"levels": [level.to_dict() for level in values]} diff --git a/app/api/ws.py b/app/api/ws.py index e0ebf31..575fee3 100644 --- a/app/api/ws.py +++ b/app/api/ws.py @@ -12,7 +12,7 @@ def snapshot(runtime, tf: Timeframe) -> dict: "type": "snapshot", "tf": tf.value, "bars": [bar.to_dict() for bar in runtime.store.get(tf, 1000)], - "levels": [], + "levels": [level.to_dict() for level in runtime.levels], "clusters": [], "price": runtime.store.get(Timeframe.M1, 1)[-1].c if runtime.store.get(Timeframe.M1, 1) @@ -40,9 +40,15 @@ async def websocket_endpoint(websocket: WebSocket): receiver = asyncio.create_task(receive()) try: while True: - bar = await queue.get() - if bar.tf is tf: - await websocket.send_json({"type": "bar", "tf": tf.value, "bar": bar.to_dict()}) + event = await queue.get() + if event["type"] == "bar" and event["bar"].tf is tf: + await websocket.send_json( + {"type": "bar", "tf": tf.value, "bar": event["bar"].to_dict()} + ) + elif event["type"] == "levels": + await websocket.send_json( + {"type": "levels", "levels": [level.to_dict() for level in event["levels"]]} + ) except (WebSocketDisconnect, asyncio.CancelledError): pass finally: diff --git a/app/config.py b/app/config.py index 13fa4d6..fd9487e 100644 --- a/app/config.py +++ b/app/config.py @@ -45,3 +45,21 @@ class Settings(BaseSettings): @property def enabled_timeframes(self) -> list[Timeframe]: return [Timeframe(value.strip()) for value in self.timeframes.split(",") if value.strip()] + + @property + def ma_sets(self) -> dict[Timeframe, list[tuple[str, int]]]: + configured = { + Timeframe.D1: self.ma_sets__1d, + Timeframe.H4: self.ma_sets__4h, + Timeframe.H1: self.ma_sets__1h, + } + result: dict[Timeframe, list[tuple[str, int]]] = {} + for tf, value in configured.items(): + definitions = [] + for item in filter(None, (part.strip().lower() for part in value.split(","))): + kind = "sma" if item.startswith("sma") else "ema" if item.startswith("ema") else "" + if not kind or not item[len(kind) :].isdigit(): + raise ValueError(f"Invalid MA definition: {item}") + definitions.append((kind, int(item[len(kind) :]))) + result[tf] = definitions + return result diff --git a/app/runtime.py b/app/runtime.py index 6f8d0ca..eeb94ad 100644 --- a/app/runtime.py +++ b/app/runtime.py @@ -3,6 +3,8 @@ from dataclasses import dataclass, field from app.bars.models import Bar, Timeframe from app.bars.aggregator import Aggregator +from app.analysis.levels import Level +from app.analysis.moving_averages import build_ma_levels from app.bars.store import InMemoryBarStore from app.config import Settings from app.market.factory import live_source, seed_source @@ -14,8 +16,9 @@ class Runtime: settings: Settings store: InMemoryBarStore = field(init=False) stream: StreamService = field(init=False) - subscribers: set[asyncio.Queue[Bar]] = field(default_factory=set) + subscribers: set[asyncio.Queue[dict]] = field(default_factory=set) aggregator: Aggregator = field(init=False) + levels: list[Level] = field(default_factory=list) def __post_init__(self) -> None: self.store = InMemoryBarStore(self.settings.max_bars_per_tf) @@ -26,10 +29,22 @@ class Runtime: async def on_bar(self, bar: Bar) -> None: 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) + self.broadcast({"type": "bar", "bar": aggregated}) + if self.settings.ma_sets.get(aggregated.tf): + self.rebuild_levels() + + def broadcast(self, event: dict) -> None: + for queue in self.subscribers.copy(): + if queue.full(): + queue.get_nowait() + queue.put_nowait(event) + + def rebuild_levels(self) -> None: + self.levels = build_ma_levels( + {tf: self.store.get(tf) for tf in self.settings.ma_sets}, + self.settings.ma_sets, + ) + self.broadcast({"type": "levels", "levels": self.levels}) async def start(self) -> asyncio.Task: try: diff --git a/static/app.js b/static/app.js index 94ed2eb..4691923 100644 --- a/static/app.js +++ b/static/app.js @@ -30,11 +30,14 @@ createApp({ const message = JSON.parse(data); if (message.type === 'snapshot') { chartApi.setBars(message.bars); + chartApi.syncLevels(message.levels || []); price.value = message.price; } else if (message.type === 'bar') { chartApi.updateBar(message.bar); price.value = message.bar.c; status.value.last_bar_t = message.bar.t; + } else if (message.type === 'levels') { + chartApi.syncLevels(message.levels); } }; socket.onclose = () => { diff --git a/static/chart.js b/static/chart.js index 3f6ebf5..50ced03 100644 --- a/static/chart.js +++ b/static/chart.js @@ -3,6 +3,7 @@ class ConfluenceChart { this.chart = null; this.candles = null; this.resizeObserver = null; + this.levelSeries = new Map(); } create(el) { @@ -36,6 +37,36 @@ class ConfluenceChart { updateBar(bar) { this.candles.update(this.toCandle(bar)); } + syncLevels(levels) { + const wanted = new Set(levels.filter(level => !level.hidden).map(level => level.id)); + for (const [id, entry] of this.levelSeries) { + if (!wanted.has(id)) { + this.chart.removeSeries(entry.series); + this.levelSeries.delete(id); + } + } + for (const level of levels) { + if (level.hidden) continue; + let entry = this.levelSeries.get(level.id); + const options = { + color: ConfluenceChart.tfColors[level.tf], + lineWidth: level.tf === '1d' ? 2 : 1, + lineType: LightweightCharts.LineType.WithSteps, + lineStyle: level.provisional ? LightweightCharts.LineStyle.Dashed : LightweightCharts.LineStyle.Solid, + priceLineVisible: false, + lastValueVisible: true, + title: level.label, + }; + if (!entry) { + entry = { series: this.chart.addSeries(LightweightCharts.LineSeries, options) }; + this.levelSeries.set(level.id, entry); + } else { + entry.series.applyOptions(options); + } + entry.series.setData((level.points || []).map(([time, value]) => ({ time, value }))); + } + } + toCandle(bar) { return { time: bar.t, open: bar.o, high: bar.h, low: bar.l, close: bar.c }; } @@ -46,4 +77,9 @@ class ConfluenceChart { } } +ConfluenceChart.tfColors = { + '1m':'#82909f', '2m':'#8a92df', '5m':'#65b7cf', '15m':'#45c39b', + '30m':'#a8c85d', '1h':'#efb643', '4h':'#ec7b42', '1d':'#d96073', +}; + window.ConfluenceChart = ConfluenceChart; diff --git a/tests/test_moving_averages.py b/tests/test_moving_averages.py new file mode 100644 index 0000000..e0ace76 --- /dev/null +++ b/tests/test_moving_averages.py @@ -0,0 +1,61 @@ +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), + ]