Add Start/End of Day controls and price history chart

- shared/config.py: add CONTROL_TOPIC for start/stop signals
- md_feeder: listen on control topic; pause/resume simulation on stop/start signals, reload securities on start
- dashboard: add SQLite OHLCV history DB (1-min buckets), record every Kafka trade
- dashboard: POST /session/start (reset prices → start), POST /session/end (save closing prices → stop)
- dashboard: GET /history/<symbol>?period=1h|8h|1d|1w|1m returns aggregated OHLCV candles (≤150 bars)
- dashboard UI: Start/End of Day buttons + session status badge in header
- dashboard UI: full-width Price History panel with candlestick + volume chart (canvas)
- docker-compose: mount shared_data volume into dashboard so it can read/write securities.txt

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

dashboard/dashboard.py

@@ -2,7 +2,7 @@ import sys
 sys.path.insert(0, "/app")
 
 from flask import Flask, render_template, jsonify, Response, request
-import threading, json, os, time, requests
+import threading, json, os, time, requests, sqlite3
 from queue import Queue, Empty
 
 from shared.config import Config
@@ -19,48 +19,136 @@ lock = threading.Lock()
 sse_clients = []
 sse_clients_lock = threading.Lock()
 
+# Session state
+session_state = {"active": False, "start_time": None}
+
+# ── OHLCV History ──────────────────────────────────────────────────────────────
+HISTORY_DB = os.getenv("HISTORY_DB", "/app/data/dashboard_history.db")
+BUCKET_SIZE = 60  # 1-minute candles
+
+PERIOD_SECONDS = {
+    "1h": 3600,
+    "8h": 28800,
+    "1d": 86400,
+    "1w": 604800,
+    "1m": 2592000,
+}
+
+
+def init_history_db():
+    os.makedirs(os.path.dirname(HISTORY_DB), exist_ok=True)
+    conn = sqlite3.connect(HISTORY_DB)
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS ohlcv (
+            symbol  TEXT,
+            bucket  INTEGER,
+            open    REAL,
+            high    REAL,
+            low     REAL,
+            close   REAL,
+            volume  INTEGER,
+            PRIMARY KEY (symbol, bucket)
+        )
+    """)
+    conn.commit()
+    conn.close()
+
+
+def record_trade(symbol, price, qty, ts):
+    if not symbol or price <= 0:
+        return
+    bucket = int(ts // BUCKET_SIZE) * BUCKET_SIZE
+    try:
+        conn = sqlite3.connect(HISTORY_DB)
+        existing = conn.execute(
+            "SELECT open FROM ohlcv WHERE symbol=? AND bucket=?", (symbol, bucket)
+        ).fetchone()
+        if existing:
+            conn.execute(
+                "UPDATE ohlcv SET high=MAX(high,?), low=MIN(low,?), close=?, volume=volume+? "
+                "WHERE symbol=? AND bucket=?",
+                (price, price, price, qty, symbol, bucket),
+            )
+        else:
+            conn.execute(
+                "INSERT INTO ohlcv VALUES (?,?,?,?,?,?,?)",
+                (symbol, bucket, price, price, price, price, qty),
+            )
+        conn.commit()
+        conn.close()
+    except Exception as e:
+        print(f"[History] Error recording trade: {e}")
+
+
+def load_securities_file():
+    securities = {}
+    try:
+        with open(Config.SECURITIES_FILE) as f:
+            for line in f:
+                line = line.strip()
+                if not line or line.startswith("#"):
+                    continue
+                parts = line.split()
+                if len(parts) >= 3:
+                    securities[parts[0]] = {
+                        "start": float(parts[1]),
+                        "current": float(parts[2]),
+                    }
+    except Exception as e:
+        print(f"[Dashboard] Cannot read securities: {e}")
+    return securities
+
+
+def save_securities_file(securities):
+    with open(Config.SECURITIES_FILE, "w") as f:
+        f.write("#SYMBOL\t<start_price>\t<current_price>\n")
+        for sym, vals in securities.items():
+            f.write(f"{sym}\t{vals['start']:.2f}\t{vals['current']:.2f}\n")
+
+
+# ── SSE broadcast ──────────────────────────────────────────────────────────────
+
 def broadcast_event(event_type, data):
-    """Send an event to all connected SSE clients."""
     message = f"event: {event_type}\ndata: {json.dumps(data)}\n\n"
     with sse_clients_lock:
-        dead_clients = []
+        dead = []
         for q in sse_clients:
             try:
                 q.put_nowait(message)
-            except:
-                dead_clients.append(q)
-        for q in dead_clients:
+            except Exception:
+                dead.append(q)
+        for q in dead:
             sse_clients.remove(q)
 
-# Kafka consumer thread
+
+# ── Kafka consumer thread ──────────────────────────────────────────────────────
+
 def consume_kafka():
     consumer = create_consumer(
         topics=[Config.ORDERS_TOPIC, Config.SNAPSHOTS_TOPIC, Config.TRADES_TOPIC],
         group_id="dashboard",
-        component_name="Dashboard"
+        component_name="Dashboard",
     )
     for msg in consumer:
         with lock:
             if msg.topic == "orders":
                 order = msg.value
                 orders.insert(0, order)
-                orders[:] = orders[:50]  # keep last 50
+                orders[:] = orders[:50]
                 broadcast_event("order", order)
 
             elif msg.topic == "snapshots":
                 snap = msg.value
-                # Handle both simple snapshots and MDF snapshots
                 symbol = snap.get("symbol")
                 if not symbol:
                     continue
-
                 bbo_data = {
                     "best_bid": snap.get("best_bid"),
                     "best_ask": snap.get("best_ask"),
                     "bid_size": snap.get("bid_size"),
                     "ask_size": snap.get("ask_size"),
                     "timestamp": snap.get("timestamp", time.time()),
-                    "source": snap.get("source", "unknown")
+                    "source": snap.get("source", "unknown"),
                 }
                 bbos[symbol] = bbo_data
                 broadcast_event("snapshot", {"symbol": symbol, **bbo_data})
@@ -68,18 +156,30 @@ def consume_kafka():
             elif msg.topic == "trades":
                 trade = msg.value
                 trades_cache.insert(0, trade)
-                trades_cache[:] = trades_cache[:200]  # keep last 200
+                trades_cache[:] = trades_cache[:200]
                 broadcast_event("trade", trade)
+                # Record for OHLCV history
+                sym = trade.get("symbol", "")
+                price = float(trade.get("price") or 0)
+                qty = int(trade.get("quantity") or trade.get("qty") or 0)
+                ts = float(trade.get("timestamp") or time.time())
+                record_trade(sym, price, qty, ts)
+
 
+# Initialise DB then start consumer thread
+init_history_db()
 threading.Thread(target=consume_kafka, daemon=True).start()
 
+
+# ── Flask routes ───────────────────────────────────────────────────────────────
+
 @app.route("/")
 def index():
     return render_template("index.html")
 
+
 @app.route("/health")
 def health():
-    """Health check endpoint."""
     status = {
         "status": "healthy",
         "service": "dashboard",
@@ -88,147 +188,228 @@ def health():
             "orders_cached": len(orders),
             "trades_cached": len(trades_cache),
             "symbols_tracked": len(bbos),
-            "sse_clients": len(sse_clients)
-        }
+            "sse_clients": len(sse_clients),
+        },
     }
-    # Check matcher connectivity
     try:
         r = requests.get(f"{Config.MATCHER_URL}/health", timeout=2)
-        if r.status_code == 200:
-            status["matcher"] = "connected"
-        else:
-            status["matcher"] = f"error: status {r.status_code}"
-            status["status"] = "degraded"
+        status["matcher"] = "connected" if r.status_code == 200 else f"error: {r.status_code}"
     except Exception as e:
         status["matcher"] = f"error: {e}"
         status["status"] = "degraded"
-
     return jsonify(status)
 
+
 @app.route("/data")
 def data():
-    """Fallback polling endpoint (still useful for initial load or SSE fallback)."""
-    # Always fetch trades from matcher (more reliable)
     try:
         r = requests.get(f"{Config.MATCHER_URL}/trades", timeout=2)
         resp = r.json()
-        trades = resp.get('trades', []) if isinstance(resp, dict) else resp
+        trades = resp.get("trades", []) if isinstance(resp, dict) else resp
     except Exception as e:
         print("Cannot fetch trades:", e)
         with lock:
             trades = list(trades_cache)
 
-    # Order book for dropdown (fetch latest for all symbols)
     try:
         r = requests.get(f"{Config.MATCHER_URL}/orderbook/ALPHA", timeout=2)
         book = r.json()
-    except Exception as e:
-        print("Cannot fetch orderbook:", e)
+    except Exception:
         book = {"bids": [], "asks": []}
 
     with lock:
-        return jsonify({
-            "orders": list(orders),
-            "bbos": dict(bbos),
-            "trades": trades,
-            "book": book
-        })
+        return jsonify({"orders": list(orders), "bbos": dict(bbos), "trades": trades, "book": book})
+
 
 @app.route("/orderbook/<symbol>")
 def orderbook(symbol):
-    """Proxy to matcher orderbook API."""
     try:
         r = requests.get(f"{Config.MATCHER_URL}/orderbook/{symbol}", timeout=2)
         return (r.text, r.status_code, {"Content-Type": "application/json"})
     except Exception as e:
         return jsonify({"error": str(e), "bids": [], "asks": []}), 500
 
-# Kafka producer for order management (lazy init)
+
+# Kafka producer (lazy init)
 _producer = None
+
+
 def get_producer():
     global _producer
     if _producer is None:
         _producer = create_producer(component_name="Dashboard")
     return _producer
 
+
 @app.route("/order/cancel", methods=["POST"])
 def cancel_order():
-    """Send cancel request to matcher via Kafka."""
     try:
         data = request.get_json()
         orig_cl_ord_id = data.get("orig_cl_ord_id")
         symbol = data.get("symbol")
-
         if not orig_cl_ord_id:
             return jsonify({"status": "error", "error": "Missing orig_cl_ord_id"}), 400
-
         cancel_msg = {
             "type": "cancel",
             "orig_cl_ord_id": orig_cl_ord_id,
             "symbol": symbol,
-            "timestamp": time.time()
+            "timestamp": time.time(),
         }
-
-        producer = get_producer()
-        producer.send(Config.ORDERS_TOPIC, cancel_msg)
-        producer.flush()
-
+        p = get_producer()
+        p.send(Config.ORDERS_TOPIC, cancel_msg)
+        p.flush()
         return jsonify({"status": "ok", "message": f"Cancel request sent for {orig_cl_ord_id}"})
     except Exception as e:
         return jsonify({"status": "error", "error": str(e)}), 500
 
+
 @app.route("/order/amend", methods=["POST"])
 def amend_order():
-    """Send amend request to matcher via Kafka."""
     try:
         data = request.get_json()
         orig_cl_ord_id = data.get("orig_cl_ord_id")
-        symbol = data.get("symbol")
-        quantity = data.get("quantity")
-        price = data.get("price")
-
         if not orig_cl_ord_id:
             return jsonify({"status": "error", "error": "Missing orig_cl_ord_id"}), 400
-
         amend_msg = {
             "type": "amend",
             "orig_cl_ord_id": orig_cl_ord_id,
             "cl_ord_id": f"amend-{int(time.time()*1000)}",
-            "symbol": symbol,
-            "quantity": quantity,
-            "price": price,
-            "timestamp": time.time()
+            "symbol": data.get("symbol"),
+            "quantity": data.get("quantity"),
+            "price": data.get("price"),
+            "timestamp": time.time(),
         }
+        p = get_producer()
+        p.send(Config.ORDERS_TOPIC, amend_msg)
+        p.flush()
+        return jsonify({"status": "ok", "message": f"Amend request sent for {orig_cl_ord_id}"})
+    except Exception as e:
+        return jsonify({"status": "error", "error": str(e)}), 500
 
-        producer = get_producer()
-        producer.send(Config.ORDERS_TOPIC, amend_msg)
-        producer.flush()
 
-        return jsonify({"status": "ok", "message": f"Amend request sent for {orig_cl_ord_id}"})
+# ── Session endpoints ──────────────────────────────────────────────────────────
+
+@app.route("/session/start", methods=["POST"])
+def session_start():
+    try:
+        # Reset current prices to start prices
+        securities = load_securities_file()
+        if securities:
+            for sym in securities:
+                securities[sym]["current"] = securities[sym]["start"]
+            save_securities_file(securities)
+
+        # Signal md_feeder to start
+        p = get_producer()
+        p.send(Config.CONTROL_TOPIC, {"action": "start"})
+        p.flush()
+
+        session_state["active"] = True
+        session_state["start_time"] = time.time()
+        broadcast_event("session", {"status": "started", "time": session_state["start_time"]})
+        return jsonify({"status": "ok", "message": "Day started"})
+    except Exception as e:
+        return jsonify({"status": "error", "error": str(e)}), 500
+
+
+@app.route("/session/end", methods=["POST"])
+def session_end():
+    try:
+        securities = load_securities_file()
+        # Update current prices from BBO mid where available
+        for sym in list(securities.keys()):
+            try:
+                r = requests.get(f"{Config.MATCHER_URL}/orderbook/{sym}", timeout=2)
+                book = r.json()
+                bids = book.get("bids", [])
+                asks = book.get("asks", [])
+                if bids and asks:
+                    best_bid = max(b["price"] for b in bids)
+                    best_ask = min(a["price"] for a in asks)
+                    securities[sym]["current"] = round((best_bid + best_ask) / 2, 2)
+            except Exception:
+                pass
+        if securities:
+            save_securities_file(securities)
+
+        # Signal md_feeder to stop
+        p = get_producer()
+        p.send(Config.CONTROL_TOPIC, {"action": "stop"})
+        p.flush()
+
+        session_state["active"] = False
+        broadcast_event("session", {"status": "ended", "time": time.time()})
+        return jsonify({"status": "ok", "message": "Day ended, closing prices saved"})
     except Exception as e:
         return jsonify({"status": "error", "error": str(e)}), 500
 
+
+@app.route("/session/status")
+def session_status():
+    return jsonify(session_state)
+
+
+# ── History endpoint ───────────────────────────────────────────────────────────
+
+@app.route("/history/<symbol>")
+def history(symbol):
+    period = request.args.get("period", "1d")
+    seconds = PERIOD_SECONDS.get(period, 86400)
+    since = int(time.time()) - seconds
+    try:
+        conn = sqlite3.connect(HISTORY_DB)
+        rows = conn.execute(
+            "SELECT bucket, open, high, low, close, volume FROM ohlcv "
+            "WHERE symbol=? AND bucket>=? ORDER BY bucket ASC",
+            (symbol, since),
+        ).fetchall()
+        conn.close()
+    except Exception as e:
+        return jsonify({"error": str(e), "candles": []}), 500
+
+    candles = [{"t": r[0], "o": r[1], "h": r[2], "l": r[3], "c": r[4], "v": r[5]} for r in rows]
+
+    # Aggregate to at most 150 bars for display
+    if len(candles) > 150:
+        step = len(candles) // 150 + 1
+        agg = []
+        for i in range(0, len(candles), step):
+            chunk = candles[i : i + step]
+            agg.append({
+                "t": chunk[0]["t"],
+                "o": chunk[0]["o"],
+                "h": max(c["h"] for c in chunk),
+                "l": min(c["l"] for c in chunk),
+                "c": chunk[-1]["c"],
+                "v": sum(c["v"] for c in chunk),
+            })
+        candles = agg
+
+    return jsonify({"symbol": symbol, "period": period, "candles": candles})
+
+
+# ── SSE stream ─────────────────────────────────────────────────────────────────
+
 @app.route("/stream")
 def stream():
-    """SSE endpoint for real-time updates."""
     def event_stream():
         q = Queue(maxsize=100)
         with sse_clients_lock:
             sse_clients.append(q)
         try:
-            # Send initial connection event
             yield f"event: connected\ndata: {json.dumps({'status': 'connected'})}\n\n"
-
-            # Send current state on connect (including trades)
             with lock:
-                yield f"event: init\ndata: {json.dumps({'orders': list(orders), 'bbos': dict(bbos), 'trades': list(trades_cache)})}\n\n"
-
+                yield (
+                    f"event: init\ndata: "
+                    f"{json.dumps({'orders': list(orders), 'bbos': dict(bbos), 'trades': list(trades_cache)})}\n\n"
+                )
+            # Also send current session state
+            yield f"event: session\ndata: {json.dumps({'status': 'started' if session_state['active'] else 'ended'})}\n\n"
             while True:
                 try:
-                    message = q.get(timeout=30)  # 30s timeout for keepalive
+                    message = q.get(timeout=30)
                     yield message
                 except Empty:
-                    # Send keepalive comment
                     yield ": keepalive\n\n"
         finally:
             with sse_clients_lock:
@@ -240,10 +421,11 @@ def stream():
         mimetype="text/event-stream",
         headers={
             "Cache-Control": "no-cache",
-            "X-Accel-Buffering": "no",  # Disable nginx buffering
-            "Connection": "keep-alive"
-        }
+            "X-Accel-Buffering": "no",
+            "Connection": "keep-alive",
+        },
     )
 
+
 if __name__ == "__main__":
     app.run(host="0.0.0.0", port=5000, debug=True, use_reloader=False)

dashboard/templates/index.html

@@ -127,10 +127,43 @@
     .ticker-change.up { color: #4caf50; }
     .ticker-change.down { color: #f44336; }
     .ticker-arrow { margin-right: 3px; }
+
+    /* Start/End of Day buttons */
+    .btn-day {
+      padding: 4px 14px;
+      border: none;
+      border-radius: 20px;
+      cursor: pointer;
+      font-size: 12px;
+      font-weight: bold;
+    }
+    .btn-start { background: #28a745; color: #fff; }
+    .btn-start:hover { background: #218838; }
+    .btn-end { background: #ff9800; color: #fff; }
+    .btn-end:hover { background: #e68900; }
+    .status.active { background: #d4edda; color: #155724; }
+    .status.active .dot { background: #28a745; }
+    .status.idle { background: #e8e8e8; color: #666; }
+    .status.idle .dot { background: #9e9e9e; }
+
+    /* Price History panel */
+    .panel-full {
+      background: #fff;
+      border-radius: 8px;
+      padding: 10px;
+      box-shadow: 0 2px 4px rgba(0,0,0,0.1);
+      margin-top: 20px;
+    }
   </style>
 </head>
 <body>
-  <h1>Trading Dashboard <span id="status" class="status connecting"><span class="dot"></span><span id="status-text">Connecting...</span></span></h1>
+  <h1>
+    Trading Dashboard
+    <span id="status" class="status connecting"><span class="dot"></span><span id="status-text">Connecting...</span></span>
+    <span id="session-badge" class="status idle"><span class="dot"></span><span id="session-text">IDLE</span></span>
+    <button onclick="startDay()" class="btn-day btn-start">Start of Day</button>
+    <button onclick="endDay()" class="btn-day btn-end">End of Day</button>
+  </h1>
   <div class="container">
 
     <div class="panel">
@@ -266,6 +299,24 @@
 
   </div>
 
+  <!-- Price History panel (full width) -->
+  <div class="panel-full">
+    <h2>Price History
+      <select id="history-symbol" onchange="loadHistory()" style="margin-left:10px; padding:2px 5px;">
+        <option value="">Select symbol...</option>
+      </select>
+      <select id="history-period" onchange="loadHistory()" style="margin-left:5px; padding:2px 5px;">
+        <option value="1h">1 Hour</option>
+        <option value="8h">8 Hours</option>
+        <option value="1d" selected>1 Day</option>
+        <option value="1w">1 Week</option>
+        <option value="1m">1 Month</option>
+      </select>
+      <span id="history-status" style="font-size:12px; color:#999; margin-left:12px;"></span>
+    </h2>
+    <canvas id="history-chart" style="width:100%; height:350px; display:block;"></canvas>
+  </div>
+
   <script>
     // State
     const state = {
@@ -598,6 +649,25 @@
         }
       });
       if (chartCurrent) chartSel.value = chartCurrent;
+
+      // Update history symbol select
+      const histSel = document.getElementById("history-symbol");
+      const histCurrent = histSel.value;
+      const histOpts = Array.from(histSel.options).map(o => o.value).filter(v => v !== "");
+      symbols.forEach(sym => {
+        if (!histOpts.includes(sym)) {
+          const opt = document.createElement("option");
+          opt.value = sym;
+          opt.textContent = sym;
+          histSel.appendChild(opt);
+        }
+      });
+      if (!histCurrent && histSel.options.length > 1) {
+        histSel.value = histSel.options[1].value;
+        loadHistory();
+      } else if (histCurrent) {
+        histSel.value = histCurrent;
+      }
     }
 
     async function renderOrderBookPanel() {
@@ -916,6 +986,11 @@
         updateBBO(snap.symbol, snap);
       });
 
+      eventSource.addEventListener("session", (e) => {
+        const data = JSON.parse(e.data);
+        updateSessionBadge(data.status === "started");
+      });
+
       eventSource.onerror = () => {
         setStatus("disconnected", "Disconnected");
         state.connected = false;
@@ -943,6 +1018,158 @@
       }
     }
 
+    // ── Session control ────────────────────────────────────────────────────────
+
+    function updateSessionBadge(active) {
+      const badge = document.getElementById("session-badge");
+      const text  = document.getElementById("session-text");
+      badge.className = "status " + (active ? "active" : "idle");
+      text.textContent = active ? "ACTIVE" : "IDLE";
+    }
+
+    async function startDay() {
+      if (!confirm("Start of Day: reset all security prices to their opening values and begin simulation?")) return;
+      try {
+        const r = await fetch("/session/start", { method: "POST" });
+        const result = await r.json();
+        if (result.status === "ok") {
+          updateSessionBadge(true);
+        } else {
+          alert("Error: " + (result.error || "Unknown error"));
+        }
+      } catch (e) {
+        alert("Error: " + e.message);
+      }
+    }
+
+    async function endDay() {
+      if (!confirm("End of Day: stop simulation and save closing prices?")) return;
+      try {
+        const r = await fetch("/session/end", { method: "POST" });
+        const result = await r.json();
+        if (result.status === "ok") {
+          updateSessionBadge(false);
+          alert("Day ended. Closing prices saved.");
+        } else {
+          alert("Error: " + (result.error || "Unknown error"));
+        }
+      } catch (e) {
+        alert("Error: " + e.message);
+      }
+    }
+
+    // ── Price History chart ────────────────────────────────────────────────────
+
+    async function loadHistory() {
+      const sym    = document.getElementById("history-symbol").value;
+      const period = document.getElementById("history-period").value;
+      const statusEl = document.getElementById("history-status");
+      if (!sym) return;
+      statusEl.textContent = "Loading...";
+      try {
+        const r    = await fetch(`/history/${sym}?period=${period}`);
+        const data = await r.json();
+        drawHistoryChart(data.candles || []);
+        statusEl.textContent = (data.candles && data.candles.length)
+          ? `${data.candles.length} candles`
+          : "No data yet";
+      } catch (e) {
+        statusEl.textContent = "Error: " + e.message;
+      }
+    }
+
+    function drawHistoryChart(candles) {
+      const canvas = document.getElementById("history-chart");
+      const ctx    = canvas.getContext("2d");
+      canvas.width  = canvas.offsetWidth || 900;
+      canvas.height = 350;
+
+      ctx.clearRect(0, 0, canvas.width, canvas.height);
+
+      if (!candles || candles.length === 0) {
+        ctx.fillStyle = "#999";
+        ctx.font = "14px Arial";
+        ctx.textAlign = "center";
+        ctx.fillText("No history data for this period", canvas.width / 2, canvas.height / 2);
+        return;
+      }
+
+      const pad    = { top: 25, right: 20, bottom: 40, left: 65 };
+      const W      = canvas.width  - pad.left - pad.right;
+      const H      = canvas.height - pad.top  - pad.bottom;
+      const priceH = H * 0.70;
+      const gapH   = H * 0.05;
+      const volH   = H * 0.25;
+      const priceY = pad.top;
+      const volY   = pad.top + priceH + gapH;
+
+      const maxP = Math.max(...candles.map(c => c.h)) * 1.002;
+      const minP = Math.min(...candles.map(c => c.l)) * 0.998;
+      const maxV = Math.max(...candles.map(c => c.v), 1);
+
+      // Price grid lines
+      ctx.strokeStyle = "#eee"; ctx.lineWidth = 1;
+      for (let i = 0; i <= 5; i++) {
+        const y = priceY + (priceH * i / 5);
+        ctx.beginPath(); ctx.moveTo(pad.left, y); ctx.lineTo(canvas.width - pad.right, y); ctx.stroke();
+        const p = maxP - (maxP - minP) * i / 5;
+        ctx.fillStyle = "#666"; ctx.font = "10px Arial"; ctx.textAlign = "right";
+        ctx.fillText(p.toFixed(2), pad.left - 5, y + 4);
+      }
+
+      // Divider between price and volume areas
+      ctx.strokeStyle = "#ddd"; ctx.lineWidth = 1;
+      ctx.beginPath(); ctx.moveTo(pad.left, volY); ctx.lineTo(canvas.width - pad.right, volY); ctx.stroke();
+
+      const n     = candles.length;
+      const slotW = W / n;
+      const barW  = Math.max(1, Math.floor(slotW * 0.7));
+
+      candles.forEach((c, i) => {
+        const x     = pad.left + i * slotW + (slotW - barW) / 2;
+        const isUp  = c.c >= c.o;
+        const color = isUp ? "#26a69a" : "#ef5350";
+        const toY   = p => priceY + priceH - ((p - minP) / (maxP - minP)) * priceH;
+
+        // Wick
+        const midX = x + barW / 2;
+        ctx.strokeStyle = color; ctx.lineWidth = 1;
+        ctx.beginPath(); ctx.moveTo(midX, toY(c.h)); ctx.lineTo(midX, toY(c.l)); ctx.stroke();
+
+        // Candle body
+        const bodyTop = Math.min(toY(c.o), toY(c.c));
+        const bodyH   = Math.max(1, Math.abs(toY(c.c) - toY(c.o)));
+        ctx.fillStyle = color;
+        ctx.fillRect(x, bodyTop, barW, bodyH);
+
+        // Volume bar
+        const vH = (c.v / maxV) * volH;
+        ctx.fillStyle = isUp ? "rgba(38,166,154,0.5)" : "rgba(239,83,80,0.5)";
+        ctx.fillRect(x, volY + volH - vH, barW, vH);
+      });
+
+      // Time axis labels (up to 6 evenly spaced)
+      ctx.fillStyle = "#666"; ctx.font = "10px Arial"; ctx.textAlign = "center";
+      const step = Math.max(1, Math.floor(n / 6));
+      for (let i = 0; i < n; i += step) {
+        const dt  = new Date(candles[i].t * 1000);
+        const lbl = (dt.getMonth()+1) + "/" + dt.getDate() + " "
+                  + dt.getHours().toString().padStart(2,"0") + ":"
+                  + dt.getMinutes().toString().padStart(2,"0");
+        const x = pad.left + i * slotW + slotW / 2;
+        ctx.fillText(lbl, x, canvas.height - pad.bottom + 15);
+      }
+
+      // VOL label rotated
+      ctx.save();
+      ctx.translate(10, volY + volH / 2);
+      ctx.rotate(-Math.PI / 2);
+      ctx.textAlign = "center";
+      ctx.fillStyle = "#aaa"; ctx.font = "9px Arial";
+      ctx.fillText("VOL", 0, 0);
+      ctx.restore();
+    }
+
     // Initial load: fetch data via REST, then connect SSE
     async function init() {
       await fetchData();

docker-compose.yml

@@ -155,6 +155,7 @@ services:
     volumes:
       - ./dashboard:/app
       - ./shared:/app/shared
+      - ./shared_data:/app/data
     depends_on:
       - kafka
       - matcher

md_feeder/mdf_simulator.py

@@ -2,19 +2,23 @@
 import sys
 sys.path.insert(0, "/app")
 
-import json, time, random, os
+import json, time, random, os, threading
 
 from shared.config import Config
-from shared.kafka_utils import create_producer
+from shared.kafka_utils import create_producer, create_consumer
 
 ORDER_INTERVAL = 60.0 / Config.ORDERS_PER_MIN
 
+# Module-level state (shared with control listener thread)
+_securities = {}
+_running = True
+
+
 def load_securities():
     """Load securities from file: SYMBOL start_price current_price"""
     securities = {}
     if not os.path.exists(Config.SECURITIES_FILE):
         raise FileNotFoundError(f"{Config.SECURITIES_FILE} not found")
-
     with open(Config.SECURITIES_FILE) as f:
         for line in f:
             if not line.strip() or line.startswith("#"):
@@ -25,6 +29,7 @@ def load_securities():
                 securities[symbol] = {"start": start, "current": current}
     return securities
 
+
 def save_securities(securities):
     """Persist securities with header line"""
     with open(Config.SECURITIES_FILE, "w") as f:
@@ -32,8 +37,10 @@ def save_securities(securities):
         for sym, vals in securities.items():
             f.write(f"{sym}\t{vals['start']:.2f}\t{vals['current']:.2f}\n")
 
+
 _order_counter = 0
 
+
 def make_order(symbol, side, price, qty):
     global _order_counter
     _order_counter += 1
@@ -47,6 +54,7 @@ def make_order(symbol, side, price, qty):
         "source": "MDF"
     }
 
+
 def make_snapshot(symbol, best_bid, best_ask, bid_size, ask_size):
     return {
         "symbol": symbol,
@@ -58,65 +66,96 @@ def make_snapshot(symbol, best_bid, best_ask, bid_size, ask_size):
         "source": "MDF"
     }
 
+
+def listen_control(ctrl_consumer):
+    """Background thread: listen for start/stop control messages."""
+    global _running, _securities
+    print("[MDF] Control listener started")
+    for msg in ctrl_consumer:
+        action = (msg.value or {}).get("action")
+        if action == "stop":
+            _running = False
+            print("[MDF] STOP signal received – pausing simulation")
+        elif action == "start":
+            try:
+                new_secs = load_securities()
+                _securities.clear()
+                _securities.update(new_secs)
+                print(f"[MDF] START signal – reloaded securities: {list(_securities.keys())}")
+            except Exception as e:
+                print(f"[MDF] Error reloading securities on start: {e}")
+            _running = True
+            print("[MDF] Simulation resumed")
+
+
 if __name__ == "__main__":
     producer = create_producer(component_name="MDF")
 
-    # Load securities and reset start=current
-    securities = load_securities()
-    for sym in securities:
-        securities[sym]["start"] = securities[sym]["current"]
-    save_securities(securities)
-    print(f"[MDF] Loaded securities: {list(securities.keys())}")
+    # Load securities and snapshot start prices
+    _securities = load_securities()
+    for sym in _securities:
+        _securities[sym]["start"] = _securities[sym]["current"]
+    save_securities(_securities)
+    print(f"[MDF] Loaded securities: {list(_securities.keys())}")
+
+    # Start control consumer in background thread
+    try:
+        ctrl_consumer = create_consumer(
+            topics=[Config.CONTROL_TOPIC],
+            group_id="md-feeder-control",
+            component_name="MDF-Control",
+            auto_offset_reset="latest",
+        )
+        threading.Thread(target=listen_control, args=(ctrl_consumer,), daemon=True).start()
+    except Exception as e:
+        print(f"[MDF] Warning: could not start control consumer: {e}")
 
     try:
         while True:
-            for sym, vals in securities.items():
+            if not _running:
+                time.sleep(0.5)
+                continue
+
+            for sym, vals in list(_securities.items()):
+                if not _running:
+                    break
+
                 mid = vals["current"]
-                # Define spread: bids below (mid - spread), asks above (mid + spread)
-                half_spread = 0.10  # 10 cents spread
+                half_spread = 0.10
 
-                # --- generate order ---
-                # 90% passive orders (build book), 10% aggressive (create trades)
                 side = random.choice(["BUY", "SELL"])
                 rand = random.random()
 
                 if rand < 0.90:
                     # Passive: place orders away from mid to rest on book
                     if side == "BUY":
-                        # Bids: from (mid - spread) down to (mid - spread - 0.50)
                         base = mid - half_spread
                         offset = random.randint(1, 50) * Config.TICK_SIZE
                         price = round(base - offset, 2)
                     else:
-                        # Asks: from (mid + spread) up to (mid + spread + 0.50)
                         base = mid + half_spread
                         offset = random.randint(1, 50) * Config.TICK_SIZE
                         price = round(base + offset, 2)
                 else:
-                    # Aggressive: cross the spread to create trades (rare)
+                    # Aggressive: cross the spread to create trades
                     if side == "BUY":
-                        # Buy into asks
                         price = round(mid + half_spread + random.randint(1, 5) * Config.TICK_SIZE, 2)
                     else:
-                        # Sell into bids
                         price = round(mid - half_spread - random.randint(1, 5) * Config.TICK_SIZE, 2)
 
                 qty = random.choice([50, 100, 150, 200, 250])
-
                 order = make_order(sym, side, price, qty)
                 producer.send(Config.ORDERS_TOPIC, order)
                 print(f"[MDF] Order: {order}")
 
-                # --- simulate small price drift (10% chance, max 2 ticks) ---
+                # Simulate small price drift (10% chance, max 2 ticks)
                 if random.random() < 0.10:
                     drift = random.choice([-2, -1, 1, 2]) * Config.TICK_SIZE
                     new_price = vals["current"] + drift
-                    # Keep price within reasonable bounds (min 1.00)
                     if new_price >= 1.00:
                         vals["current"] = round(new_price, 2)
-                        save_securities(securities)
+                        save_securities(_securities)
 
-                # --- snapshot around mid price ---
                 best_bid = mid - half_spread
                 best_ask = mid + half_spread
                 bid_size = random.choice([50, 100, 200])

shared/config.py

@@ -26,6 +26,9 @@ class Config:
     SECURITIES_FILE: str = os.getenv("SECURITIES_FILE", "/app/data/securities.txt")
     ORDER_ID_FILE: str = os.getenv("ORDER_ID_FILE", "/app/data/order_id.txt")
 
+    # Control topic for start/end of day signals
+    CONTROL_TOPIC: str = os.getenv("CONTROL_TOPIC", "control")
+
     # Trading simulation
     TICK_SIZE: float = float(os.getenv("TICK_SIZE", "0.05"))
     ORDERS_PER_MIN: int = int(os.getenv("ORDERS_PER_MIN", "8"))

shared_data/securities.txt

@@ -1,6 +1,6 @@
 #SYMBOL	<start_price>	<current_price>
-ALPHA	24.90	24.95
-PEIR	18.50	18.05
-EXAE	42.00	42.05
-QUEST	12.70	12.60
-NBG	18.50	18.05
+ALPHA	24.95	25.15
+PEIR	18.05	18.05
+EXAE	42.05	42.05
+QUEST	12.60	12.65
+NBG	18.05	18.05