Implement M1 live one-minute chart

.env.example

@@ -0,0 +1,25 @@
+# Data sources
+LIVE_SOURCE=yahoo
+SEED_SOURCE=yahoo
+YAHOO_SYMBOL=ES=F
+YAHOO_POLL_SECONDS=20
+SEED_1H_RANGE=730d
+SEED_1M_RANGE=8d
+
+# Chart and analysis
+TIMEFRAMES=1m,2m,5m,15m,30m,1h,4h,1d
+BASE_TIMEFRAMES=1m,30m,1d
+MAX_BARS_PER_TF=5000
+MA_SETS__1D=sma10,sma20,sma50,sma100,sma200
+MA_SETS__4H=
+MA_SETS__1H=
+DAILY_ANCHOR_ET=18:00
+MANUAL_LINES_PATH=./data/manual_lines.json
+CONFLUENCE_MIN_SCORE=28
+ALERT_COOLDOWN_SECONDS=900
+
+# Notifications and access
+NTFY_TOPIC=
+NTFY_SERVER=https://ntfy.sh
+CHART_AUTH_TOKEN=
+REPLAY_FILE=

app/api/__init__.py

@@ -0,0 +1 @@
+"""HTTP and WebSocket API."""

app/api/routes.py

@@ -0,0 +1,33 @@
+from fastapi import APIRouter, HTTPException, Query, Request
+
+from app.bars.models import Timeframe
+
+router = APIRouter(prefix="/api")
+
+
+@router.get("/health")
+def health():
+    return {"status": "ok", "service": "chart"}
+
+
+@router.get("/status")
+def status(request: Request):
+    runtime = request.app.state.runtime
+    return {
+        "stream": runtime.stream.status,
+        "source": runtime.stream.source.name,
+        "symbol": runtime.stream.symbol,
+        "last_bar_t": runtime.stream.last_bar_t,
+        "bars_held": runtime.store.counts(),
+        "warm": {tf.value: bool(runtime.store.get(tf)) for tf in Timeframe},
+    }
+
+
+@router.get("/bars")
+def bars(request: Request, tf: str = "1m", limit: int = Query(500, ge=1, le=5000)):
+    try:
+        timeframe = Timeframe(tf)
+    except ValueError as exc:
+        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]}

app/api/ws.py

@@ -0,0 +1,50 @@
+import asyncio
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+
+from app.bars.models import Timeframe
+
+router = APIRouter()
+
+
+def snapshot(runtime, tf: Timeframe) -> dict:
+    return {
+        "type": "snapshot",
+        "tf": tf.value,
+        "bars": [bar.to_dict() for bar in runtime.store.get(tf, 1000)],
+        "levels": [],
+        "clusters": [],
+        "price": runtime.store.get(Timeframe.M1, 1)[-1].c
+        if runtime.store.get(Timeframe.M1, 1)
+        else None,
+    }
+
+
+@router.websocket("/ws")
+async def websocket_endpoint(websocket: WebSocket):
+    await websocket.accept()
+    runtime = websocket.app.state.runtime
+    queue: asyncio.Queue = asyncio.Queue(maxsize=100)
+    runtime.subscribers.add(queue)
+    tf = Timeframe.M1
+    await websocket.send_json(snapshot(runtime, tf))
+
+    async def receive():
+        nonlocal tf
+        while True:
+            message = await websocket.receive_json()
+            if message.get("type") == "subscribe":
+                tf = Timeframe(message.get("tf", "1m"))
+                await websocket.send_json(snapshot(runtime, tf))
+
+    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()})
+    except (WebSocketDisconnect, asyncio.CancelledError):
+        pass
+    finally:
+        receiver.cancel()
+        runtime.subscribers.discard(queue)

app/bars/store.py

@@ -0,0 +1,31 @@
+from collections import defaultdict, deque
+from typing import Protocol
+
+from app.bars.models import Bar, Timeframe
+
+
+class BarStore(Protocol):
+    def put(self, bar: Bar) -> None: ...
+
+    def get(self, tf: Timeframe, limit: int | None = None) -> list[Bar]: ...
+
+
+class InMemoryBarStore:
+    def __init__(self, max_bars_per_tf: int = 5000):
+        self._bars: dict[Timeframe, deque[Bar]] = defaultdict(
+            lambda: deque(maxlen=max_bars_per_tf)
+        )
+
+    def put(self, bar: Bar) -> None:
+        bars = self._bars[bar.tf]
+        if bars and bars[-1].t == bar.t:
+            bars[-1] = bar
+        elif not bars or bar.t > bars[-1].t:
+            bars.append(bar)
+
+    def get(self, tf: Timeframe, limit: int | None = None) -> list[Bar]:
+        bars = list(self._bars[tf])
+        return bars[-limit:] if limit is not None else bars
+
+    def counts(self) -> dict[str, int]:
+        return {tf.value: len(self._bars[tf]) for tf in Timeframe}

app/config.py

@@ -0,0 +1,47 @@
+from pathlib import Path
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from app.bars.models import Timeframe
+
+
+TIMEFRAME_WEIGHT = {
+    Timeframe.M1: 1,
+    Timeframe.M2: 1,
+    Timeframe.M5: 1,
+    Timeframe.M15: 2,
+    Timeframe.M30: 3,
+    Timeframe.H1: 4,
+    Timeframe.H4: 8,
+    Timeframe.D1: 16,
+}
+MA_WEIGHT_FACTOR = 0.75
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+    live_source: str = "yahoo"
+    seed_source: str = "yahoo"
+    yahoo_symbol: str = "ES=F"
+    yahoo_poll_seconds: float = 20
+    seed_1h_range: str = "730d"
+    seed_1m_range: str = "8d"
+    timeframes: str = "1m,2m,5m,15m,30m,1h,4h,1d"
+    base_timeframes: str = "1m,30m,1d"
+    max_bars_per_tf: int = 5000
+    ma_sets__1d: str = "sma10,sma20,sma50,sma100,sma200"
+    ma_sets__4h: str = ""
+    ma_sets__1h: str = ""
+    daily_anchor_et: str = "18:00"
+    manual_lines_path: Path = Path("./data/manual_lines.json")
+    confluence_min_score: float = 28
+    alert_cooldown_seconds: int = 900
+    ntfy_topic: str = ""
+    ntfy_server: str = "https://ntfy.sh"
+    chart_auth_token: str = ""
+    replay_file: Path | None = None
+
+    @property
+    def enabled_timeframes(self) -> list[Timeframe]:
+        return [Timeframe(value.strip()) for value in self.timeframes.split(",") if value.strip()]

app/market/factory.py

@@ -0,0 +1,22 @@
+from app.config import Settings
+from app.market.base import MarketDataSource
+from app.market.replay import ReplaySource
+from app.market.yahoo import YahooSource
+
+
+def live_source(settings: Settings) -> MarketDataSource:
+    if settings.live_source == "yahoo":
+        return YahooSource(settings.yahoo_poll_seconds)
+    if settings.live_source == "replay" and settings.replay_file:
+        return ReplaySource(settings.replay_file)
+    raise ValueError(f"Unsupported LIVE_SOURCE: {settings.live_source}")
+
+
+def seed_source(settings: Settings) -> MarketDataSource | None:
+    if settings.seed_source == "none":
+        return None
+    if settings.seed_source == "yahoo":
+        return YahooSource(settings.yahoo_poll_seconds)
+    if settings.seed_source == "replay" and settings.replay_file:
+        return ReplaySource(settings.replay_file)
+    raise ValueError(f"Unsupported SEED_SOURCE: {settings.seed_source}")

app/market/stream.py

@@ -0,0 +1,61 @@
+import asyncio
+import logging
+from collections.abc import Awaitable, Callable
+
+from app.bars.models import Bar, Timeframe
+from app.market.base import MarketDataSource
+
+logger = logging.getLogger(__name__)
+BarHandler = Callable[[Bar], Awaitable[None]]
+
+
+class StreamService:
+    def __init__(self, source: MarketDataSource, symbol: str):
+        self.source = source
+        self.symbol = symbol
+        self.status = "disconnected"
+        self.last_bar_t: int | None = None
+        self._handlers: list[BarHandler] = []
+        self._stop = asyncio.Event()
+
+    def add_handler(self, handler: BarHandler) -> None:
+        self._handlers.append(handler)
+
+    async def seed(
+        self, source: MarketDataSource | None, tf: Timeframe, range_: str
+    ) -> 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_)
+        for bar in bars:
+            await self._emit(bar)
+
+    async def _emit(self, bar: Bar) -> None:
+        self.last_bar_t = max(self.last_bar_t or bar.t, bar.t)
+        for handler in self._handlers:
+            await handler(bar)
+
+    async def run(self) -> None:
+        while not self._stop.is_set():
+            try:
+                self.status = "replay" if self.source.name == "replay" else "connected"
+                async for bar in self.source.stream(self.symbol):
+                    await self._emit(bar)
+                    if self._stop.is_set():
+                        break
+                if self.source.name == "replay":
+                    return
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.exception("Market stream failed; reconnecting")
+            self.status = "disconnected"
+            try:
+                await asyncio.wait_for(self._stop.wait(), timeout=5)
+            except TimeoutError:
+                pass
+
+    def stop(self) -> None:
+        self._stop.set()
+        self.status = "disconnected"

app/runtime.py

@@ -0,0 +1,38 @@
+import asyncio
+from dataclasses import dataclass, field
+
+from app.bars.models import Bar, Timeframe
+from app.bars.store import InMemoryBarStore
+from app.config import Settings
+from app.market.factory import live_source, seed_source
+from app.market.stream import StreamService
+
+
+@dataclass
+class Runtime:
+    settings: Settings
+    store: InMemoryBarStore = field(init=False)
+    stream: StreamService = field(init=False)
+    subscribers: set[asyncio.Queue[Bar]] = field(default_factory=set)
+
+    def __post_init__(self) -> None:
+        self.store = InMemoryBarStore(self.settings.max_bars_per_tf)
+        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)
+
+    async def start(self) -> asyncio.Task:
+        try:
+            await self.stream.seed(
+                seed_source(self.settings), Timeframe.M1, self.settings.seed_1m_range
+            )
+        except Exception:
+            # A transient seed failure must not prevent the live stream or UI starting.
+            pass
+        return asyncio.create_task(self.stream.run(), name="market-stream")

main.py

@@ -1,31 +1,39 @@
-"""chart.amow.com — FastAPI backend.
-
-Placeholder app: serves the Vue 3 single-page frontend from static/ and a
-couple of JSON endpoints under /api. Replace the endpoints as the real app
-takes shape; the serving/deploy wiring below does not need to change.
-"""
+"""chart.amow.com FastAPI backend."""
+import asyncio
+from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import FastAPI
 from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
 
+from app.api.routes import router as api_router
+from app.api.ws import router as ws_router
+from app.config import Settings
+from app.runtime import Runtime
+
 BASE_DIR = Path(__file__).parent
 STATIC_DIR = BASE_DIR / "static"
 
-app = FastAPI(title="chart")
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    runtime = Runtime(Settings())
+    app.state.runtime = runtime
+    task = await runtime.start()
+    yield
+    runtime.stream.stop()
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        pass
+
+
+app = FastAPI(title="chart", lifespan=lifespan)
 
 app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
-
-
-@app.get("/api/health")
-def health():
-    return {"status": "ok", "service": "chart"}
-
-
-@app.get("/api/hello")
-def hello():
-    return {"msg": "this is fing awesome"}
+app.include_router(api_router)
+app.include_router(ws_router)
 
 
 @app.get("/")

static/app.js

@@ -1,25 +1,59 @@
-const { createApp, ref } = Vue;
+const { createApp, ref, computed, onMounted, onUnmounted } = Vue;
 
 createApp({
   setup() {
-    const result = ref('');
-    const error = ref('');
-    const loading = ref(false);
+    const status = ref({ stream: 'disconnected', bars_held: {} });
+    const price = ref(null);
+    const now = ref(Date.now());
+    let chartApi = null;
+    let socket = null;
+    let timer = null;
 
-    async function ping() {
-      loading.value = true;
-      error.value = '';
-      try {
-        const res = await fetch('/api/hello');
-        if (!res.ok) throw new Error(`HTTP ${res.status}`);
-        result.value = JSON.stringify(await res.json(), null, 2);
-      } catch (e) {
-        error.value = String(e);
-      } finally {
-        loading.value = false;
-      }
+    const barAge = computed(() => {
+      if (!status.value.last_bar_t) return '—';
+      const seconds = Math.max(0, Math.floor(now.value / 1000 - status.value.last_bar_t));
+      return seconds < 60 ? `${seconds}s` : `${Math.floor(seconds / 60)}m`;
+    });
+
+    async function refreshStatus() {
+      const response = await fetch('/api/status');
+      if (response.ok) status.value = await response.json();
     }
 
-    return { result, error, loading, ping };
+    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.onmessage = ({ data }) => {
+        const message = JSON.parse(data);
+        if (message.type === 'snapshot') {
+          chartApi.setBars(message.bars);
+          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;
+        }
+      };
+      socket.onclose = () => {
+        status.value.stream = 'disconnected';
+        setTimeout(connect, 2000);
+      };
+    }
+
+    onMounted(() => {
+      chartApi = new ConfluenceChart();
+      chartApi.create(document.getElementById('chart'));
+      refreshStatus();
+      connect();
+      timer = setInterval(() => { now.value = Date.now(); refreshStatus(); }, 5000);
+    });
+    onUnmounted(() => {
+      clearInterval(timer);
+      if (socket) socket.close();
+      if (chartApi) chartApi.destroy();
+    });
+
+    return { status, price, barAge };
   },
 }).mount('#app');

static/chart.js

@@ -0,0 +1,49 @@
+class ConfluenceChart {
+  constructor() {
+    this.chart = null;
+    this.candles = null;
+    this.resizeObserver = null;
+  }
+
+  create(el) {
+    this.chart = LightweightCharts.createChart(el, {
+      autoSize: true,
+      layout: {
+        background: { color: getComputedStyle(document.documentElement).getPropertyValue('--chart-bg').trim() },
+        textColor: getComputedStyle(document.documentElement).getPropertyValue('--muted').trim(),
+      },
+      grid: {
+        vertLines: { color: 'rgba(128,128,128,.10)' },
+        horzLines: { color: 'rgba(128,128,128,.10)' },
+      },
+      timeScale: { timeVisible: true, secondsVisible: false },
+      rightPriceScale: { borderVisible: false },
+    });
+    this.candles = this.chart.addSeries(LightweightCharts.CandlestickSeries, {
+      upColor: '#3fb984', downColor: '#e45757', borderVisible: false,
+      wickUpColor: '#3fb984', wickDownColor: '#e45757',
+    });
+    this.resizeObserver = new ResizeObserver(() => {
+      this.chart.applyOptions({ width: el.clientWidth, height: el.clientHeight });
+    });
+    this.resizeObserver.observe(el);
+  }
+
+  setBars(bars) {
+    this.candles.setData(bars.map(this.toCandle));
+    this.chart.timeScale().fitContent();
+  }
+
+  updateBar(bar) { this.candles.update(this.toCandle(bar)); }
+
+  toCandle(bar) {
+    return { time: bar.t, open: bar.o, high: bar.h, low: bar.l, close: bar.c };
+  }
+
+  destroy() {
+    if (this.resizeObserver) this.resizeObserver.disconnect();
+    if (this.chart) this.chart.remove();
+  }
+}
+
+window.ConfluenceChart = ConfluenceChart;

static/index.html

@@ -3,24 +3,39 @@
 <head>
   <meta charset="utf-8">
   <meta name="viewport" content="width=device-width, initial-scale=1">
-  <title>chart</title>
+  <title>/ES Confluence</title>
   <link rel="stylesheet" href="/static/style.css">
   <script src="https://unpkg.com/vue@3/dist/vue.global.prod.js"></script>
+  <script src="https://unpkg.com/lightweight-charts@5.2.0/dist/lightweight-charts.standalone.production.js"></script>
 </head>
 <body>
   <div id="app">
-    <h1>chart</h1>
-    <p class="sub">FastAPI + Vue 3 &mdash; placeholder</p>
-
-    <div class="card">
-      <button @click="ping" :disabled="loading">
-        {{ loading ? 'calling…' : 'call /api/hello' }}
-      </button>
-      <pre v-if="result">{{ result }}</pre>
-      <p v-if="error" class="error">{{ error }}</p>
+    <header>
+      <div><span class="eyebrow">CME FUTURES</span><h1>/ES <strong>CONFLUENCE</strong></h1></div>
+      <div class="status" :class="status.stream"><i></i>{{ status.stream }} · {{ status.source || 'source' }}</div>
+    </header>
+    <main>
+      <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>
+        <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>
         </div>
-
+      </section>
+      <aside>
+        <h2>Confluence zones</h2>
+        <div class="empty">Analysis layers arrive after aggregation.</div>
+        <h2>Alert log</h2>
+        <div class="empty">No alerts fired.</div>
+      </aside>
+    </main>
+  </div>
+  <script src="/static/chart.js"></script>
   <script src="/static/app.js"></script>
 </body>
 </html>

static/style.css

@@ -1,62 +1,18 @@
-:root {
-  color-scheme: light dark;
-  --bg: #ffffff;
-  --fg: #16181d;
-  --muted: #6b7280;
-  --line: #e4e6eb;
-  --accent: #2f6feb;
-}
-
-@media (prefers-color-scheme: dark) {
-  :root {
-    --bg: #14161a;
-    --fg: #e8eaed;
-    --muted: #9aa1ab;
-    --line: #2a2e35;
-    --accent: #6d9bf5;
-  }
-}
-
-* { box-sizing: border-box; }
-
-body {
-  margin: 0;
-  padding: 3rem 1.5rem;
-  background: var(--bg);
-  color: var(--fg);
-  font: 16px/1.6 ui-sans-serif, system-ui, -apple-system, "Segoe UI", sans-serif;
-}
-
-#app { max-width: 40rem; margin: 0 auto; }
-
-h1 { margin: 0; font-size: 1.75rem; letter-spacing: -0.01em; }
-
-.sub { margin: 0.25rem 0 2rem; color: var(--muted); }
-
-.card {
-  border: 1px solid var(--line);
-  border-radius: 10px;
-  padding: 1.25rem;
-}
-
-button {
-  font: inherit;
-  padding: 0.5rem 1rem;
-  border: 0;
-  border-radius: 6px;
-  background: var(--accent);
-  color: #fff;
-  cursor: pointer;
-}
-
-button:disabled { opacity: 0.6; cursor: default; }
-
-pre {
-  margin: 1rem 0 0;
-  padding: 0.75rem;
-  overflow-x: auto;
-  border-radius: 6px;
-  background: color-mix(in srgb, var(--fg) 6%, transparent);
-}
-
-.error { color: #d24b4b; }
+:root { color-scheme: dark; --bg:#090d12; --panel:#10161d; --chart-bg:#0c1117; --fg:#e8edf3; --muted:#82909f; --line:#202b36; --accent:#efb643; --green:#3fb984; --red:#e45757; }
+@media (prefers-color-scheme: light) { :root { color-scheme:light; --bg:#eef1f3; --panel:#fff; --chart-bg:#fff; --fg:#17202a; --muted:#66717d; --line:#dce2e7; } }
+* { box-sizing:border-box; }
+body { margin:0; background:var(--bg); color:var(--fg); font:14px/1.45 "IBM Plex Mono", "SFMono-Regular", Consolas, monospace; }
+#app { min-height:100vh; padding:18px; }
+header { height:64px; display:flex; align-items:center; justify-content:space-between; border-bottom:1px solid var(--line); margin-bottom:16px; }
+h1 { margin:0; font-size:22px; letter-spacing:-1px; } h1 strong { color:var(--accent); font-weight:600; }
+.eyebrow { color:var(--muted); font-size:9px; letter-spacing:2px; }
+.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); }
+.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); }
+#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; } }

tests/test_store.py

@@ -0,0 +1,17 @@
+from app.bars.models import Bar, Timeframe
+from app.bars.store import InMemoryBarStore
+
+
+def bar(t, close=1):
+    return Bar(Timeframe.M1, t, 1, 2, 0, close, 10, False, "ES=F", "replay")
+
+
+def test_store_replaces_forming_bar_and_bounds_history():
+    store = InMemoryBarStore(2)
+    store.put(bar(60))
+    store.put(bar(60, 2))
+    store.put(bar(120))
+    store.put(bar(180))
+
+    assert [value.t for value in store.get(Timeframe.M1)] == [120, 180]
+    assert store.get(Timeframe.M1, 1)[0].t == 180