feat: timezone cleanup + readiness check + caching + concurrent reads

Author: lucas-guerin-44

Dockerfile

@@ -27,4 +27,7 @@ EXPOSE 8000
 HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
     CMD curl -fsS http://localhost:8000/healthcheck || exit 1
 
+# Single worker on purpose: only one OS process can open the DuckDB file
+# read-write. Read concurrency comes from per-cursor reads + the threadpool
+# (see read_connection in src/core/datalake.py). Do NOT add --workers.
 CMD ["uvicorn", "src.api:app", "--host", "0.0.0.0", "--port", "8000"]

scripts/migrate_tick_tz_xauusd.py

@@ -0,0 +1,178 @@
+"""One-shot migration: correct broker-local-as-UTC timestamps in tick_data
+for XAUUSD.
+
+Background — see quant-strategies-research/docs/RESEARCH_NOTES.md lesson #80
+and docs/TZ_FIX_PHASE2_DATALAKE_RECONCILIATION.md.
+
+The XAUUSD ticks in tick_data were originally ingested from an MT5
+<DATE>/<TIME> CSV via _read_raw_tick(); that path treated broker-local
+clock values as UTC. This script applies a DST-aware Athens -> UTC
+correction in place via DuckDB's `timezone()` function.
+
+Run on the VPS where the DuckDB file lives. Read-write transaction;
+backs up the affected rows to `tick_data_xauusd_backup_<run-id>` before
+UPDATE. Idempotent guard: refuses to run if a `tick_data_migrations`
+ledger already contains a matching entry.
+
+Usage:
+    # dry-run (default) — prints sample before/after, no writes:
+    python scripts/migrate_tick_tz_xauusd.py
+
+    # apply:
+    python scripts/migrate_tick_tz_xauusd.py --confirm
+
+    # ad-hoc broker tz override (default = Europe/Athens):
+    MT5_BROKER_TZ=Europe/Bucharest python scripts/migrate_tick_tz_xauusd.py --confirm
+"""
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+# Make sibling 'src/' importable so we share the lake's DB path config.
+_HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(_HERE.parent))
+
+import duckdb  # type: ignore
+
+# Reuse the lake's own DB path config (avoid drift).
+try:
+    from src.config import DUCKDB_PATH  # type: ignore
+except Exception:
+    DUCKDB_PATH = Path(os.getenv("DUCKDB_PATH", str(_HERE.parent / "datalake" / "ohlc.duckdb")))
+
+
+MIGRATION_ID = "tick_tz_xauusd_athens_2026_05_28"
+
+
+def ensure_ledger(con):
+    con.execute("""
+        CREATE TABLE IF NOT EXISTS tick_data_migrations (
+            migration_id VARCHAR PRIMARY KEY,
+            instrument VARCHAR,
+            broker_tz VARCHAR,
+            rows_affected BIGINT,
+            backup_table VARCHAR,
+            applied_at TIMESTAMP
+        )
+    """)
+
+
+def already_applied(con) -> bool:
+    r = con.execute(
+        "SELECT COUNT(*) FROM tick_data_migrations WHERE migration_id = ?",
+        [MIGRATION_ID],
+    ).fetchone()
+    return bool(r and r[0])
+
+
+def sample_rows(con, n=8) -> list:
+    return con.execute(f"""
+        SELECT timestamp, price, bid, ask
+        FROM tick_data
+        WHERE instrument = 'XAUUSD'
+        ORDER BY timestamp
+        LIMIT {n}
+    """).fetchall()
+
+
+def sample_post(con, broker_tz: str, n=8) -> list:
+    return con.execute(f"""
+        SELECT
+            timestamp                                              AS before,
+            timezone('UTC', timezone(?, timestamp))                 AS after,
+            price, bid, ask
+        FROM tick_data
+        WHERE instrument = 'XAUUSD'
+        ORDER BY timestamp
+        LIMIT {n}
+    """, [broker_tz]).fetchall()
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--confirm", action="store_true", help="Apply the UPDATE (default: dry-run)")
+    ap.add_argument("--broker-tz", default=os.getenv("MT5_BROKER_TZ", "Europe/Athens"))
+    ap.add_argument("--db", default=str(DUCKDB_PATH))
+    args = ap.parse_args()
+
+    db_path = Path(args.db)
+    if not db_path.exists():
+        print(f"DuckDB file not found: {db_path}")
+        return 2
+
+    print(f"DB: {db_path}")
+    print(f"broker_tz: {args.broker_tz}")
+    print(f"migration_id: {MIGRATION_ID}")
+
+    con = duckdb.connect(str(db_path))
+
+    ensure_ledger(con)
+    if already_applied(con):
+        print("This migration is already recorded in tick_data_migrations. Refusing to re-run.")
+        return 0
+
+    total = con.execute("SELECT COUNT(*) FROM tick_data WHERE instrument='XAUUSD'").fetchone()[0]
+    print(f"\nXAUUSD tick rows to migrate: {total:,}")
+
+    print("\n--- BEFORE (first 8 rows) ---")
+    for r in sample_rows(con):
+        print(" ", r)
+
+    print(f"\n--- COMPUTED AFTER (via timezone('UTC', timezone('{args.broker_tz}', ts))) ---")
+    for r in sample_post(con, args.broker_tz):
+        print(" ", r)
+
+    if not args.confirm:
+        print("\nDry-run. Re-run with --confirm to apply.")
+        return 0
+
+    backup_table = f"tick_data_xauusd_backup_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}"
+    print(f"\nBacking up to {backup_table} ...")
+    con.execute(f"CREATE TABLE {backup_table} AS SELECT * FROM tick_data WHERE instrument='XAUUSD'")
+    backup_count = con.execute(f"SELECT COUNT(*) FROM {backup_table}").fetchone()[0]
+    print(f"  backed up {backup_count:,} rows")
+    assert backup_count == total, "Backup row count mismatch — aborting before UPDATE"
+
+    print(f"\nApplying UPDATE ...")
+    # DuckDB UPDATE semantics: timezone('UTC', timezone(tz, ts)) returns the
+    # naive TIMESTAMP at UTC corresponding to interpreting `ts` as wall-clock
+    # in `tz`. DST-aware via the IANA zone tables.
+    con.execute("""
+        UPDATE tick_data
+        SET timestamp = timezone('UTC', timezone(?, timestamp))
+        WHERE instrument = 'XAUUSD'
+    """, [args.broker_tz])
+    # DuckDB doesn't expose row-count for UPDATE separately; the backup table
+    # holds the authoritative pre-state count, which matches.
+    rows = total
+
+    print(f"\n--- AFTER UPDATE (first 8 rows) ---")
+    for r in sample_rows(con):
+        print(" ", r)
+
+    new_min, new_max = con.execute(
+        "SELECT MIN(timestamp), MAX(timestamp) FROM tick_data WHERE instrument='XAUUSD'"
+    ).fetchone()
+    print(f"\nNew XAUUSD tick range: {new_min} -> {new_max}")
+
+    print(f"\nRecording migration in tick_data_migrations ledger ...")
+    con.execute("""
+        INSERT INTO tick_data_migrations
+            (migration_id, instrument, broker_tz, rows_affected, backup_table, applied_at)
+        VALUES (?, ?, ?, ?, ?, ?)
+    """, [MIGRATION_ID, "XAUUSD", args.broker_tz, rows, backup_table, datetime.now(timezone.utc)])
+
+    print(f"Done. backup={backup_table} rows_affected={rows:,}")
+    print(f"Rollback: UPDATE tick_data SET timestamp = b.timestamp FROM {backup_table} b "
+          f"WHERE tick_data.instrument='XAUUSD' AND tick_data.<other PK columns> = b.<same>; "
+          f"DELETE FROM tick_data_migrations WHERE migration_id = '{MIGRATION_ID}';")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/mt5_bridge.py

@@ -22,6 +22,7 @@
 import os
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 try:
     import MetaTrader5 as mt5
@@ -35,6 +36,20 @@
 # EnvironmentFile so it matches MT5_BRIDGE_KEY on the API container side.
 BRIDGE_KEY = os.getenv("MT5_BRIDGE_KEY", "")
 
+# MT5's MqlRates.time is broker-server-local-time-as-int (NOT real UTC Unix
+# epoch). Localize as the broker tz then convert to real UTC before emitting.
+# Match scripts/mt5_fetch.py in the quant-strategies-research repo. Override
+# the broker tz via env (Eightcap = Europe/Athens / EET/EEST).
+# See quant-strategies-research/docs/RESEARCH_NOTES.md lesson #80 for context.
+BROKER_TZ_NAME = os.getenv("MT5_BROKER_TZ", "Europe/Athens")
+try:
+    BROKER_TZ = ZoneInfo(BROKER_TZ_NAME)
+except ZoneInfoNotFoundError as _e:
+    raise RuntimeError(
+        f"MT5_BROKER_TZ={BROKER_TZ_NAME!r} not resolvable. On Windows/Wine you "
+        f"may need `pip install tzdata`."
+    ) from _e
+
 TIMEFRAME_MAP = {
     "M1": mt5.TIMEFRAME_M1,
     "M5": mt5.TIMEFRAME_M5,
@@ -75,7 +90,15 @@ def fetch_bars(symbol: str, timeframe: str, start: datetime, end: datetime) -> l
 
     out = []
     for r in rates:
-        ts = datetime.fromtimestamp(int(r["time"]), tz=timezone.utc)
+        # MT5 `time` is broker-local-time-as-int. utcfromtimestamp gives a
+        # naive datetime whose wall-clock value visually equals the broker
+        # server clock. Attach the broker tz, then convert to real UTC.
+        # fold=0 handles DST fall-back ambiguity by picking the first
+        # (pre-transition) occurrence, which matches MT5's chronological
+        # emission order; spring-forward gaps don't appear in MT5 output.
+        naive = datetime.utcfromtimestamp(int(r["time"]))
+        local = naive.replace(tzinfo=BROKER_TZ)
+        ts = local.astimezone(timezone.utc)
         out.append({
             "timestamp": ts.isoformat(),
             "open": float(r["open"]),

src/api.py

@@ -1,5 +1,6 @@
 """Main FastAPI application - wires up all route modules."""
-from fastapi import FastAPI, Depends, Response
+from fastapi import FastAPI, Depends, Request, Response
+from fastapi.responses import JSONResponse
 from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
 from prometheus_fastapi_instrumentator import Instrumentator
 from slowapi.errors import RateLimitExceeded
@@ -36,6 +37,21 @@
 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 app.add_middleware(RequestLoggingMiddleware)
 
+
+@app.exception_handler(Exception)
+async def unhandled_exception_handler(request: Request, exc: Exception):
+    """
+    Last-resort handler so an unexpected error returns a JSON `{error, detail}`
+    envelope with a 500, never a bare gateway HTML body. Clients can always parse
+    the response as JSON. HTTPException keeps FastAPI's own handler (correct status
+    + `{detail}`); this only catches what would otherwise be an uncaught 500.
+    """
+    logger.error("Unhandled exception", exc_info=exc, extra={"path": request.url.path})
+    return JSONResponse(
+        status_code=500,
+        content={"error": "internal_error", "detail": str(exc)},
+    )
+
 # Instrument every route with request counters + latency histograms.
 # /metrics is exposed manually below so we can gate it behind admin scope.
 Instrumentator(excluded_handlers=["/metrics", "/healthcheck", "/healthcheck/ready"]).instrument(app)

src/core/cache.py

@@ -0,0 +1,43 @@
+"""
+In-memory, version-keyed cache for read endpoints whose answers only change on
+ingest/delete (e.g. /catalog, /instruments). Each cached value is tagged with the
+data-version token from src.core.datalake; a write bumps that token, so the next
+read recomputes and everything in between is served from memory without touching
+DuckDB. This keeps catalog/instrument listings from competing with data queries.
+See datalake-api-w9a.
+"""
+import threading
+from typing import Callable
+
+from src.core.datalake import get_data_version
+
+_lock = threading.Lock()
+_store: dict = {}  # key -> (data_version, value)
+
+
+def get_or_compute(key: str, producer: Callable[[], object]) -> object:
+    """
+    Return the cached value for `key` if it was computed at the current data
+    version, else call `producer()`, cache, and return it. `producer` runs
+    outside the lock so a slow build doesn't block other cache readers.
+    """
+    version = get_data_version()
+    with _lock:
+        hit = _store.get(key)
+        if hit is not None and hit[0] == version:
+            return hit[1]
+
+    value = producer()
+
+    with _lock:
+        # Re-tag with the version captured before producing. If a write landed
+        # mid-build, the entry is already stale-by-version and the next read
+        # recomputes — never serving data older than its tag claims.
+        _store[key] = (version, value)
+    return value
+
+
+def clear() -> None:
+    """Drop all cached entries. Mainly for tests; production self-invalidates by version."""
+    with _lock:
+        _store.clear()

src/core/concurrency.py

@@ -0,0 +1,38 @@
+"""
+Backpressure for read endpoints.
+
+DuckDB already parallelizes each query across cores, so a flood of concurrent
+heavy scans competes for the same CPU/memory (prod caps the container at 2 CPUs
+and DuckDB at a 2GB memory_limit). Rather than let that pile up until the gateway
+times out and emits a bare 502, we shed load honestly: once MAX_CONCURRENT_QUERIES
+are in flight, further read requests get a 503 + `Retry-After` so clients can back
+off and retry. A 503 means "busy, try again"; a 502 reads as "down". See
+datalake-api-c45.
+"""
+import os
+import threading
+
+from fastapi import HTTPException
+
+MAX_CONCURRENT_QUERIES = int(os.getenv("MAX_CONCURRENT_QUERIES", "8"))
+RETRY_AFTER_SECONDS = int(os.getenv("QUERY_RETRY_AFTER_SECONDS", "1"))
+
+_query_semaphore = threading.BoundedSemaphore(MAX_CONCURRENT_QUERIES)
+
+
+def query_slot():
+    """
+    FastAPI dependency: reserve one concurrent-read slot for the request's
+    lifetime (including streamed response bodies). Returns 503 + Retry-After
+    immediately when all slots are taken instead of queuing into a timeout.
+    """
+    if not _query_semaphore.acquire(blocking=False):
+        raise HTTPException(
+            status_code=503,
+            detail="Server busy — too many concurrent queries. Retry shortly.",
+            headers={"Retry-After": str(RETRY_AFTER_SECONDS)},
+        )
+    try:
+        yield
+    finally:
+        _query_semaphore.release()