Upload 10 files

Dockerfile

@@ -0,0 +1,34 @@
+FROM ubuntu:22.04
+
+ENV DEBIAN_FRONTEND=noninteractive
+
+RUN apt-get update && apt-get install -y \
+    git curl python3 python3-pip \
+    && pip3 install --no-cache-dir pyyaml \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install ClickHouse static binary
+RUN curl https://clickhouse.com/ | sh && \
+    mv clickhouse /usr/local/bin/ && \
+    chmod +x /usr/local/bin/clickhouse
+
+# Install yq — YAML parser for bash (used by refresh_sources.sh)
+RUN curl -fsSL https://github.com/mikefarah/yq/releases/latest/download/yq_linux_amd64 \
+    -o /usr/local/bin/yq && \
+    chmod +x /usr/local/bin/yq
+
+WORKDIR /app
+COPY . .
+
+# Make scripts executable
+RUN chmod +x init_clickhouse.sh refresh_sources.sh
+
+# Create all directories ClickHouse needs — writable by uid 1000 (HF default user)
+RUN mkdir -p /app/ch/data /app/ch/tmp /app/ch/user_files /app/ch/format_schemas \
+             /app/ch/access /app/ch/log /app/data && \
+    chmod -R 777 /app
+
+# HF Spaces only exposes this port — gateway sits here
+EXPOSE 7860
+
+CMD ["bash", "init_clickhouse.sh"]

clickhouse-config.xml

@@ -0,0 +1,51 @@
+<clickhouse>
+    <!-- Redirect ALL storage paths to /app/ch (writable by HF uid 1000) -->
+    <path>/app/ch/data/</path>
+    <tmp_path>/app/ch/tmp/</tmp_path>
+    <user_files_path>/app/ch/user_files/</user_files_path>
+    <format_schema_path>/app/ch/format_schemas/</format_schema_path>
+    <access_control_path>/app/ch/access/</access_control_path>
+
+    <logger>
+        <log>/app/ch/log/clickhouse-server.log</log>
+        <errorlog>/app/ch/log/clickhouse-server.err.log</errorlog>
+        <level>warning</level>
+    </logger>
+
+    <!-- ClickHouse HTTP on 8123 (internal) — gateway proxies from 7860 -->
+    <listen_host>0.0.0.0</listen_host>
+    <http_port>8123</http_port>
+    <tcp_port>9000</tcp_port>
+
+    <!-- Default profiles -->
+    <profiles>
+        <default/>
+    </profiles>
+
+    <!-- Default user — open for HF Space (no external network access anyway) -->
+    <users>
+        <default>
+            <password></password>
+            <networks>
+                <ip>::/0</ip>
+            </networks>
+            <profile>default</profile>
+            <quota>default</quota>
+            <access_management>1</access_management>
+        </default>
+    </users>
+
+    <!-- Default quota (unlimited) -->
+    <quotas>
+        <default>
+            <interval>
+                <duration>3600</duration>
+                <queries>0</queries>
+                <errors>0</errors>
+                <result_rows>0</result_rows>
+                <read_rows>0</read_rows>
+                <execution_time>0</execution_time>
+            </interval>
+        </default>
+    </quotas>
+</clickhouse>

gateway.py

@@ -0,0 +1,192 @@
+#!/usr/bin/env python3
+"""
+gateway.py — Lightweight HTTP gateway for the Observatory
+
+Sits on port 7860 (the only port HF Spaces exposes) and routes:
+
+  GET  /api/refresh   →  git pull all repos + recreate all views (on-demand)
+  GET  /api/sources   →  return current sources.yaml as JSON
+  *    /*             →  proxy everything else to ClickHouse on 8123
+
+This lets you push to GitHub, then immediately:
+  curl https://your-space.hf.space/api/refresh
+  curl https://your-space.hf.space/?query=SELECT * FROM ohlc LIMIT 5
+"""
+
+import json
+import os
+import subprocess
+import sys
+import time
+import urllib.request
+import urllib.error
+import yaml
+from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
+
+CLICKHOUSE_URL = "http://127.0.0.1:8123"
+CONFIG_PATH = "/app/sources.yaml"
+REFRESH_SCRIPT = "/app/refresh_sources.py"
+GATEWAY_PORT = 7860
+
+
+class GatewayHandler(BaseHTTPRequestHandler):
+
+    # ── /api/refresh ─────────────────────────────────────────────────────
+    def _handle_refresh(self):
+        """Run refresh_sources.sh and return JSON result."""
+        try:
+            start = time.time()
+            result = subprocess.run(
+                ["bash", REFRESH_SCRIPT, "full"],
+                capture_output=True, text=True, timeout=300,
+            )
+            elapsed = round(time.time() - start, 2)
+
+            body = {
+                "action": "refresh",
+                "elapsed_seconds": elapsed,
+                "exit_code": result.returncode,
+            }
+
+            # Try to parse script's JSON output
+            try:
+                body["result"] = json.loads(result.stdout)
+            except (json.JSONDecodeError, ValueError):
+                body["stdout"] = result.stdout
+
+            if result.stderr:
+                body["stderr"] = result.stderr
+
+            self._json_response(200, body)
+
+        except subprocess.TimeoutExpired:
+            self._json_response(504, {"error": "refresh timed out after 300s"})
+        except Exception as e:
+            self._json_response(500, {"error": str(e)})
+
+    # ── /api/sources ─────────────────────────────────────────────────────
+    def _handle_sources(self):
+        """Return current sources.yaml as JSON."""
+        try:
+            with open(CONFIG_PATH, "r") as f:
+                config = yaml.safe_load(f)
+            self._json_response(200, config)
+        except Exception as e:
+            self._json_response(500, {"error": str(e)})
+
+    # ── Proxy to ClickHouse ──────────────────────────────────────────────
+    def _proxy_to_clickhouse(self):
+        """Forward the request to ClickHouse HTTP interface."""
+        # Read request body if present
+        content_length = int(self.headers.get("Content-Length", 0))
+        body = self.rfile.read(content_length) if content_length else None
+
+        url = f"{CLICKHOUSE_URL}{self.path}"
+
+        req = urllib.request.Request(url, data=body, method=self.command)
+
+        # Forward relevant headers
+        for header in self.headers:
+            lower = header.lower()
+            if lower not in ("host", "content-length", "transfer-encoding"):
+                req.add_header(header, self.headers[header])
+
+        try:
+            resp = urllib.request.urlopen(req, timeout=600)
+            self.send_response(resp.status)
+            # Forward response headers
+            for key, val in resp.headers.items():
+                if key.lower() != "transfer-encoding":
+                    self.send_header(key, val)
+            # Add CORS headers for browser-based SQL playgrounds
+            self.send_header("Access-Control-Allow-Origin", "*")
+            self.end_headers()
+            # Stream response body
+            while True:
+                chunk = resp.read(65536)
+                if not chunk:
+                    break
+                self.wfile.write(chunk)
+
+        except urllib.error.HTTPError as e:
+            self.send_response(e.code)
+            for key, val in e.headers.items():
+                if key.lower() != "transfer-encoding":
+                    self.send_header(key, val)
+            self.send_header("Access-Control-Allow-Origin", "*")
+            self.end_headers()
+            self.wfile.write(e.read())
+
+        except Exception as e:
+            self._json_response(502, {"error": f"ClickHouse unreachable: {e}"})
+
+    # ── HTTP method handlers ─────────────────────────────────────────────
+    def do_GET(self):
+        if self.path == "/api/refresh":
+            self._handle_refresh()
+        elif self.path == "/api/sources":
+            self._handle_sources()
+        else:
+            self._proxy_to_clickhouse()
+
+    def do_POST(self):
+        self._proxy_to_clickhouse()
+
+    def do_OPTIONS(self):
+        """CORS preflight."""
+        self.send_response(204)
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
+        self.send_header("Access-Control-Allow-Headers", "*")
+        self.end_headers()
+
+    # ── Helpers ──────────────────────────────────────────────────────────
+    def _json_response(self, code, data):
+        body = json.dumps(data, indent=2).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("Access-Control-Allow-Origin", "*")
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, format, *args):
+        """Cleaner log format."""
+        sys.stderr.write(f"[gateway] {self.address_string()} {format % args}\n")
+
+
+def wait_for_clickhouse(timeout=60):
+    """Block until ClickHouse is responding on 8123."""
+    print(f"[gateway] Waiting for ClickHouse on {CLICKHOUSE_URL}...")
+    for i in range(timeout):
+        try:
+            urllib.request.urlopen(f"{CLICKHOUSE_URL}/ping", timeout=2)
+            print(f"[gateway] ClickHouse is ready (took {i+1}s)")
+            return True
+        except Exception:
+            time.sleep(1)
+    print(f"[gateway] WARNING: ClickHouse not ready after {timeout}s, starting anyway")
+    return False
+
+
+if __name__ == "__main__":
+    # Wait for ClickHouse to be ready before accepting traffic
+    wait_for_clickhouse()
+
+    # Run initial full refresh (clone repos + create views)
+    print("[gateway] Running initial data sync + view creation...")
+    result = subprocess.run(
+        ["python3", REFRESH_SCRIPT, "full"],
+        capture_output=True, text=True, timeout=300,
+    )
+    print(result.stdout)
+    if result.stderr:
+        print(f"[gateway] stderr: {result.stderr}", file=sys.stderr)
+
+    # Start gateway
+    server = ThreadingHTTPServer(("0.0.0.0", GATEWAY_PORT), GatewayHandler)
+    print(f"[gateway] Listening on port {GATEWAY_PORT}")
+    print(f"[gateway]   /api/refresh  → on-demand git pull + view recreation")
+    print(f"[gateway]   /api/sources  → current config as JSON")
+    print(f"[gateway]   /*            → proxy to ClickHouse")
+    server.serve_forever()

init_clickhouse.sh

@@ -0,0 +1,74 @@
+#!/bin/bash
+# ─────────────────────────────────────────────────────────────────────────────
+#  Market-Data Observatory — Init Script
+#
+#  Boot sequence:
+#    1. Start ClickHouse server (port 8123, internal)
+#    2. Start background refresh loop (git pull every N seconds)
+#    3. Start Python gateway (port 7860, public-facing)
+#       → gateway waits for ClickHouse, then does initial clone + view creation
+#
+#  The gateway handles:
+#    /api/refresh  →  on-demand git pull + view recreation
+#    /api/sources  →  current config as JSON
+#    /*            →  proxy to ClickHouse
+# ─────────────────────────────────────────────────────────────────────────────
+set -e
+
+CONFIG="/app/sources.yaml"
+
+echo "╔══════════════════════════════════════════════════════════╗"
+echo "║   Market-Data Observatory — Boot Sequence               ║"
+echo "╚══════════════════════════════════════════════════════════╝"
+echo ""
+
+# ── Validate config ──────────────────────────────────────────────────────────
+if [ ! -f "$CONFIG" ]; then
+  echo "[FATAL] Config file $CONFIG not found!"
+  exit 1
+fi
+
+REFRESH_INTERVAL=$(yq '.refresh_interval_seconds // 300' "$CONFIG")
+SOURCE_COUNT=$(yq '.sources | length' "$CONFIG")
+echo "[config] $SOURCE_COUNT source(s), background refresh every ${REFRESH_INTERVAL}s"
+echo ""
+
+# ── 1. Start ClickHouse (background, port 8123) ─────────────────────────────
+echo "[1/3] Starting ClickHouse on port 8123 (internal)..."
+clickhouse server --config-file=/app/clickhouse-config.xml &
+CH_PID=$!
+echo "       PID: $CH_PID"
+echo ""
+
+# ── 2. Background auto-refresh loop ─────────────────────────────────────────
+if [ "$REFRESH_INTERVAL" -gt 0 ] 2>/dev/null; then
+  echo "[2/3] Starting background refresh loop (every ${REFRESH_INTERVAL}s)..."
+  (
+    # Wait for initial setup to complete before starting loop
+    sleep "$REFRESH_INTERVAL"
+    while true; do
+      echo "[refresh-loop] Syncing all sources..."
+      python3 /app/refresh_sources.py sync_only > /dev/null 2>&1 || true
+      sleep "$REFRESH_INTERVAL"
+    done
+  ) &
+else
+  echo "[2/3] Background refresh loop DISABLED (refresh_interval_seconds is 0)."
+  echo "      You must manually call /api/refresh to update data."
+fi
+echo ""
+
+# ── 3. Start gateway (foreground, port 7860) ────────────────────────────────
+echo "[3/3] Starting gateway on port 7860 (public)..."
+echo ""
+echo "╔══════════════════════════════════════════════════════════╗"
+echo "║   Endpoints:                                             ║"
+echo "║     /api/refresh  → instant git pull + view recreation  ║"
+echo "║     /api/sources  → current config as JSON              ║"
+echo "║     /?query=...   → ClickHouse SQL (proxied)            ║"
+echo "╚══════════════════════════════════════════════════════════╝"
+echo ""
+
+# Gateway runs in foreground — keeps the container alive
+# If it dies, the container restarts
+exec python3 /app/gateway.py

refresh_sources.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""
+refresh_sources.py — Standalone refresh script in Python
+
+Actions:
+  • git clone (first run) or git pull (subsequent) for each source
+  • Inject auth token if `auth_env_var` is defined in config
+  • CREATE OR REPLACE VIEW for each defined view
+"""
+import os
+import sys
+import json
+import yaml
+import urllib.request
+import urllib.error
+import subprocess
+
+# Allow local testing gracefully
+CONFIG = "/app/sources.yaml" if os.path.exists("/app/sources.yaml") else "sources.yaml"
+USER_FILES = "/app/ch/user_files" if os.path.exists("/app") else "user_files"
+CH_URL = "http://127.0.0.1:8123"
+
+def run_subprocess(cmd, cwd=None, env=None):
+    try:
+        result = subprocess.run(
+            cmd, cwd=cwd, env=env,
+            capture_output=True, text=True, check=True
+        )
+        return result.stdout.strip(), True
+    except subprocess.CalledProcessError as e:
+        return e.output.strip() + "\n" + e.stderr.strip(), False
+
+
+def execute_clickhouse_query(sql):
+    req = urllib.request.Request(CH_URL, data=sql.encode('utf-8'), method='POST')
+    try:
+        resp = urllib.request.urlopen(req, timeout=10)
+        return resp.read().decode('utf-8'), True
+    except urllib.error.URLError as e:
+        return str(e), False
+
+def main():
+    mode = "full"
+    if len(sys.argv) > 1:
+        mode = sys.argv[1]
+
+    if not os.path.exists(CONFIG):
+        print(json.dumps({"error": f"Config not found at {CONFIG}"}))
+        sys.exit(1)
+
+    with open(CONFIG, "r") as f:
+        config_data = yaml.safe_load(f)
+    
+    sources = config_data.get("sources", [])
+    
+    result = {
+        "mode": mode,
+        "source_count": len(sources),
+        "sources": []
+    }
+
+    env_vars = os.environ.copy()
+
+    for source in sources:
+        name = source.get("name", "unknown")
+        repo_url = source.get("repo_url")
+        local_dir = source.get("local_dir")
+        clone_depth = str(source.get("clone_depth", 1))
+        branch = source.get("branch", "")
+        auth_env_var = source.get("auth_env_var", "")
+        
+        target_dir = os.path.join(USER_FILES, local_dir)
+
+        # Build git clone/pull options
+        auth_flag = []
+        if auth_env_var and auth_env_var in env_vars:
+            token = env_vars[auth_env_var]
+            auth_flag = ["-c", f"http.extraHeader=Authorization: Bearer {token}"]
+        
+        branch_flag = []
+        if branch:
+            branch_flag = ["--branch", branch]
+
+        git_status = "unknown"
+        
+        # Git Sync
+        if not os.path.isdir(target_dir):
+            cmd = ["git", "clone", "--depth", clone_depth] + branch_flag + auth_flag + [repo_url, target_dir]
+            output, success = run_subprocess(cmd)
+            git_status = "cloned" if success else f"error_clone: {output}"
+        else:
+            # We want to fetch and hard reset
+            cmd_fetch = ["git"] + auth_flag + ["fetch", "--depth", "1", "origin"]
+            output_f, success_f = run_subprocess(cmd_fetch, cwd=target_dir)
+            if success_f:
+                # get current branch
+                rev_cmd = ["git", "rev-parse", "--abbrev-ref", "HEAD"]
+                curr_branch, _ = run_subprocess(rev_cmd, cwd=target_dir)
+                reset_cmd = ["git", "reset", "--hard", f"origin/{curr_branch}"]
+                output_r, success_r = run_subprocess(reset_cmd, cwd=target_dir)
+                git_status = "pulled" if success_r else f"error_reset: {output_r}"
+            else:
+                git_status = f"error_fetch: {output_f}"
+        
+        latest_commit, _ = run_subprocess(["git", "log", "-1", "--format=%h %s"], cwd=target_dir)
+        if not latest_commit:
+            latest_commit = "unknown"
+
+        source_info = {
+            "name": name,
+            "git_status": git_status,
+            "latest_commit": latest_commit
+        }
+
+        # View Creation
+        if mode == "full":
+            views_info = []
+            views = source.get("views", [])
+            for view in views:
+                view_name = view.get("view_name")
+                file_glob = view.get("file_glob")
+                format_ = view.get("format", "Parquet")
+                columns = view.get("columns", [])
+
+                full_glob = f"{local_dir}/{file_glob}"
+
+                if columns:
+                    cols_str = ", ".join(columns)
+                    select_clause = f"SELECT {cols_str}"
+                else:
+                    select_clause = "SELECT *"
+
+                sql = f"CREATE OR REPLACE VIEW {view_name} AS {select_clause} FROM file('{full_glob}', {format_})"
+
+                _, success = execute_clickhouse_query(sql)
+                view_status = "ok" if success else "error"
+                
+                views_info.append({
+                    "name": view_name,
+                    "glob": full_glob,
+                    "select": select_clause,
+                    "status": view_status
+                })
+            
+            source_info["views"] = views_info
+        
+        result["sources"].append(source_info)
+    
+    print(json.dumps(result, indent=2))
+
+if __name__ == "__main__":
+    main()

refresh_sources.sh

@@ -0,0 +1,126 @@
+#!/bin/bash
+# ─────────────────────────────────────────────────────────────────────────────
+#  refresh_sources.sh — Standalone refresh script
+#
+#  Can be called:
+#    1. At boot (from init_clickhouse.sh)
+#    2. On-demand (from gateway.py when you hit /api/refresh)
+#    3. Periodically (from background loop)
+#
+#  Actions:
+#    • git clone (first run) or git pull (subsequent) for each source
+#    • CREATE OR REPLACE VIEW for each defined view
+# ─────────────────────────────────────────────────────────────────────────────
+set -e
+
+CONFIG="/app/sources.yaml"
+USER_FILES="/app/ch/user_files"
+CH_URL="http://127.0.0.1:8123"
+
+# Optional argument: "sync_only" = just git pull, no view creation
+# (used by background loop to avoid hammering ClickHouse with DDL)
+MODE="${1:-full}"
+
+SOURCE_COUNT=$(yq '.sources | length' "$CONFIG")
+
+echo "{"
+echo "  \"mode\": \"$MODE\","
+echo "  \"source_count\": $SOURCE_COUNT,"
+echo "  \"sources\": ["
+
+for i in $(seq 0 $((SOURCE_COUNT - 1))); do
+  NAME=$(yq ".sources[$i].name" "$CONFIG")
+  REPO_URL=$(yq ".sources[$i].repo_url" "$CONFIG")
+  LOCAL_DIR=$(yq ".sources[$i].local_dir" "$CONFIG")
+  CLONE_DEPTH=$(yq ".sources[$i].clone_depth // 1" "$CONFIG")
+  BRANCH=$(yq ".sources[$i].branch // \"\"" "$CONFIG")
+
+  TARGET="$USER_FILES/$LOCAL_DIR"
+
+  BRANCH_FLAG=""
+  if [ -n "$BRANCH" ] && [ "$BRANCH" != "null" ] && [ "$BRANCH" != "" ]; then
+    BRANCH_FLAG="--branch $BRANCH"
+  fi
+
+  # ── Git sync ──
+  GIT_STATUS="unknown"
+  GIT_OUTPUT=""
+  if [ ! -d "$TARGET" ]; then
+    GIT_OUTPUT=$(git clone --depth "$CLONE_DEPTH" $BRANCH_FLAG "$REPO_URL" "$TARGET" 2>&1) || true
+    GIT_STATUS="cloned"
+  else
+    GIT_OUTPUT=$(cd "$TARGET" && git fetch --depth 1 origin && git reset --hard origin/$(git rev-parse --abbrev-ref HEAD) 2>&1) || true
+    GIT_STATUS="pulled"
+  fi
+
+  LATEST_COMMIT=$(cd "$TARGET" && git log -1 --format='%h %s' 2>/dev/null || echo "unknown")
+
+  # ── Trailing comma handling ──
+  COMMA=","
+  if [ $i -eq $((SOURCE_COUNT - 1)) ] && [ "$MODE" = "sync_only" ]; then
+    COMMA=""
+  fi
+
+  # ── View creation (only in "full" mode) ──
+  VIEWS_JSON=""
+  if [ "$MODE" = "full" ]; then
+    NUM_VIEWS=$(yq ".sources[$i].views | length" "$CONFIG")
+    VIEWS_JSON="\"views\": ["
+
+    for v in $(seq 0 $((NUM_VIEWS - 1))); do
+      VIEW_NAME=$(yq ".sources[$i].views[$v].view_name" "$CONFIG")
+      FILE_GLOB=$(yq ".sources[$i].views[$v].file_glob" "$CONFIG")
+      FORMAT=$(yq ".sources[$i].views[$v].format // \"Parquet\"" "$CONFIG")
+      NUM_COLS=$(yq ".sources[$i].views[$v].columns | length // 0" "$CONFIG")
+
+      FULL_GLOB="${LOCAL_DIR}/${FILE_GLOB}"
+
+      # Build SELECT clause — explicit columns if defined, else SELECT *
+      if [ "$NUM_COLS" -gt 0 ] 2>/dev/null; then
+        COLS=""
+        for c in $(seq 0 $((NUM_COLS - 1))); do
+          COL=$(yq ".sources[$i].views[$v].columns[$c]" "$CONFIG")
+          if [ -z "$COLS" ]; then
+            COLS="$COL"
+          else
+            COLS="$COLS, $COL"
+          fi
+        done
+        SELECT_CLAUSE="SELECT ${COLS}"
+      else
+        SELECT_CLAUSE="SELECT *"
+      fi
+
+      SQL="CREATE OR REPLACE VIEW ${VIEW_NAME} AS ${SELECT_CLAUSE} FROM file('${FULL_GLOB}', ${FORMAT})"
+
+      VIEW_STATUS="ok"
+      if ! curl -sf "$CH_URL" --data "$SQL" 2>/dev/null; then
+        VIEW_STATUS="error"
+      fi
+
+      VIEW_COMMA=","
+      if [ $v -eq $((NUM_VIEWS - 1)) ]; then
+        VIEW_COMMA=""
+      fi
+
+      VIEWS_JSON="${VIEWS_JSON}{\"name\":\"${VIEW_NAME}\",\"glob\":\"${FULL_GLOB}\",\"select\":\"${SELECT_CLAUSE}\",\"status\":\"${VIEW_STATUS}\"}${VIEW_COMMA}"
+    done
+
+    VIEWS_JSON="${VIEWS_JSON}]"
+  fi
+
+  # Determine trailing comma for this source
+  COMMA=","
+  if [ $i -eq $((SOURCE_COUNT - 1)) ]; then
+    COMMA=""
+  fi
+
+  if [ "$MODE" = "full" ]; then
+    echo "    {\"name\":\"$NAME\",\"git_status\":\"$GIT_STATUS\",\"latest_commit\":\"$LATEST_COMMIT\",$VIEWS_JSON}$COMMA"
+  else
+    echo "    {\"name\":\"$NAME\",\"git_status\":\"$GIT_STATUS\",\"latest_commit\":\"$LATEST_COMMIT\"}$COMMA"
+  fi
+done
+
+echo "  ]"
+echo "}"

requirements.txt

@@ -0,0 +1,3 @@
+fastapi
+uvicorn
+httpx

server.py

@@ -0,0 +1,68 @@
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import StreamingResponse, HTMLResponse
+import subprocess
+import httpx
+
+app = FastAPI(title="Market-Data Observatory")
+
+# Async HTTP client for proxying to ClickHouse's internal HTTP interface
+ch_client = httpx.AsyncClient(base_url="http://127.0.0.1:8123", timeout=60.0)
+
+
+# ── Refresh endpoint (git pull) ──
+@app.post("/refresh")
+def refresh():
+    try:
+        result = subprocess.check_output(
+            "cd /app/data/ohlc_data && git pull",
+            shell=True, stderr=subprocess.STDOUT
+        )
+        return {"status": "updated", "details": result.decode().strip()}
+    except subprocess.CalledProcessError as e:
+        msg = e.output.decode() if e.output else str(e)
+        raise HTTPException(status_code=500, detail={"error": "Refresh failed", "message": msg})
+
+
+# ── Health check ──
+@app.get("/health")
+async def health():
+    try:
+        r = await ch_client.get("/ping")
+        return {"clickhouse": r.text.strip(), "status": "ok"}
+    except Exception as e:
+        raise HTTPException(status_code=503, detail={"status": "unhealthy", "error": str(e)})
+
+
+# ── Catch-all reverse proxy to ClickHouse ──
+# This gives you /play, /dashboard, native HTTP API, everything.
+@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "HEAD", "PATCH"])
+async def proxy_to_clickhouse(request: Request, path: str):
+    url = httpx.URL(path=f"/{path}", query=request.url.query.encode("utf-8"))
+
+    # Forward headers, strip hop-by-hop
+    headers = dict(request.headers)
+    for h in ("host", "content-length", "transfer-encoding"):
+        headers.pop(h, None)
+
+    try:
+        body = await request.body()
+        req = ch_client.build_request(
+            method=request.method,
+            url=url,
+            headers=headers,
+            content=body,
+        )
+        r = await ch_client.send(req, stream=True)
+
+        # Pass through ClickHouse response headers
+        resp_headers = dict(r.headers)
+        for h in ("content-length", "content-encoding", "transfer-encoding"):
+            resp_headers.pop(h, None)
+
+        return StreamingResponse(
+            r.aiter_raw(),
+            status_code=r.status_code,
+            headers=resp_headers,
+        )
+    except httpx.RequestError as exc:
+        raise HTTPException(status_code=502, detail=f"ClickHouse proxy error: {exc}")

sources.yaml

@@ -0,0 +1,85 @@
+# ┌──────────────────────────────────────────────────────────────────────┐
+# │  Market-Data Observatory — Data Source Registry                     │
+# │                                                                     │
+# │  To add a new data source, simply append a new entry to the list.  │
+# │  The init script reads this file at boot and handles everything:   │
+# │    • git clone / pull                                               │
+# │    • ClickHouse VIEW creation (CREATE OR REPLACE)                  │
+# │    • background auto-refresh                                        │
+# │                                                                     │
+# │  On-demand refresh: curl https://<space>/api/refresh               │
+# │  No code changes needed — just edit this YAML file.                │
+# └──────────────────────────────────────────────────────────────────────┘
+
+# Global settings
+refresh_interval_seconds: 300   # background git-pull interval (safety net)
+
+# Data sources — each entry = git clone + ClickHouse view(s)
+sources:
+
+  - name: hf_ohlc_data
+    description: "HuggingFace OHLCV Dataset: altool/ohlcdata"
+    repo_url: "https://huggingface.co/datasets/altool/ohlcdata"
+    branch: "main"
+    auth_env_var: "HF_TOKEN" # Allows pulling from private HF dataset natively
+    clone_depth: 1
+    local_dir: "hf_ohlc_data"                   # → /app/ch/user_files/hf_ohlc_data/
+    views:
+      - view_name: hf_ohlc
+        file_glob: "**/*.parquet"
+        format: Parquet
+        description: "Full OHLCV data from HF dataset"
+        # columns:                             # explicit column list for the VIEW
+        #   - timestamp
+        #   - symbol
+        #   - isin
+        #   - series
+        #   - open
+        #   - high
+        #   - low
+        #   - close
+        #   - volume
+        #   - interval_minutes
+        #   - segment
+        #   - exchange
+
+  # ── EXAMPLE: Adding a second data source ──────────────────────────────
+  # Just uncomment and edit:
+  #
+  # - name: options_chain
+  #   description: "NSE options chain snapshots"
+  #   repo_url: "https://github.com/youruser/options_chain_data.git"
+  #   branch: "main"
+  #   clone_depth: 1
+  #   local_dir: "options_chain"
+  #   views:
+  #     - view_name: options
+  #       file_glob: "**/*.parquet"
+  #       format: Parquet
+  #       description: "Options chain snapshots"
+  #       columns:                           # omit 'columns' to use SELECT *
+  #         - strike
+  #         - expiry
+  #         - option_type
+  #         - open_interest
+  #         - volume
+  #         - ltp
+  #         - iv
+  #
+  # ── EXAMPLE: Multiple views from the same repo ───────────────────────
+  #
+  # - name: market_indices
+  #   description: "Broad market index data"
+  #   repo_url: "https://github.com/youruser/index_data.git"
+  #   branch: "main"
+  #   clone_depth: 1
+  #   local_dir: "indices"
+  #   views:
+  #     - view_name: nifty50
+  #       file_glob: "nifty50/**/*.parquet"
+  #       format: Parquet
+  #       description: "Nifty 50 index"
+  #     - view_name: banknifty
+  #       file_glob: "banknifty/**/*.parquet"
+  #       format: Parquet
+  #       description: "Bank Nifty index"