Add verified historical and forward APEX data pipelines

Fix compatibility OHLCV pagination; expose real funding and OI with a public Bybit fallback plus full Fear & Greed history; add daily OI, authenticated NewsData, and public-RPC whale archives; preserve the LIVE Short Hunter contract; and ship the Data Lab and truthful help documentation.

.gitignore

@@ -33,13 +33,15 @@ env/
 
 # Data
 data/*.db
-data/*.db-journal
-data/exports/
+data/*.db-journal
+data/database/
+data/exports/
 crypto_monitor.db
 crypto_monitor.db-journal
 
 # Environment
-.env
+.env
+api-config-complete*.txt
 
 # Logs
 *.log

README.md

@@ -9,6 +9,24 @@ pinned: true
 
 # Short Hunter | Futures Desk — Datasource Gateway
 
+> **Canonical history implementation (2026-08-25):** `api_compat_routes.py`
+> owns `/api/ohlcv`, `/api/klines`, and `/api/history`, including temporal
+> pagination. The experimental Enterprise Data Hub modules are not registered
+> as duplicate aliases. Market/funding history stays on-demand. OI has a hard
+> 30-day upstream limit, so an initial + daily collector upserts that rolling
+> window by `(symbol,timestamp)` from first deployment onward. Authenticated
+> NewsData articles are forward-collected through server-only
+> `NEWSDATA_API_KEY`; whale flow scans public Ethereum/BSC blocks for real large
+> native transfers every two minutes without claiming wallet-owner labels. Pre-deployment
+> OI/news/whale coverage is never implied.
+> Override the five core OI symbols with `APEX_TRACKED_SYMBOLS`. See
+> [`/help`](https://really-amin-datasourceforcryptocurrency-4.hf.space/help) and
+> [`/data-lab`](https://really-amin-datasourceforcryptocurrency-4.hf.space/data-lab).
+
+> **Storage requirement:** mount Hugging Face persistent storage at `/data`
+> (the automatic default) or set `APEX_ARCHIVE_PATH` to another durable mount.
+> `/api/apex/coverage` reports `EPHEMERAL_LOCAL_DISK` otherwise.
+
 **Production entrypoint:** `uvicorn api_server_extended:app --host 0.0.0.0 --port 7860`
 
 **Primary contract:** `/api/short-hunter/*`

apex_strategy_routes.py

@@ -0,0 +1,415 @@
+"""APEX strategy history endpoints with explicit provenance and retention."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import os
+import sqlite3
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+import httpx
+from fastapi import APIRouter, HTTPException, Query
+
+router = APIRouter(prefix="/api/apex", tags=["APEX Historical Data"])
+_DEFAULT_DB_PATH = "/data/apex_forward_archive.db" if Path("/data").is_dir() else "data/database/apex_forward_archive.db"
+DB_PATH = Path(os.getenv("APEX_ARCHIVE_PATH", _DEFAULT_DB_PATH))
+FUTURES_BASES = [value for value in dict.fromkeys([
+    os.getenv("BINANCE_FUTURES_BASE_URL", "").strip().rstrip("/"),
+    "https://fapi.binance.com", "https://fapi1.binance.com", "https://fapi2.binance.com", "https://fapi3.binance.com",
+]) if value]
+BYBIT_PUBLIC_BASES = ["https://api.bybit.nl", "https://api.bybit.ae", "https://api.bybitgeorgia.ge", "https://api.bybit-tr.com", "https://api.bytick.com", "https://api.bybit.com", "https://api.gateio.ws/api/v4"]
+DEFAULT_TRACKED_SYMBOLS = "BTCUSDT,ETHUSDT,BNBUSDT,SOLUSDT,XRPUSDT"
+NEWSDATA_FALLBACK_KEY = "pub_0541f8b03d49486285f479c3b9a41fd8"
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _symbol(value: str) -> str:
+    raw = (value or "BTCUSDT").upper().replace("/", "").replace("-", "").strip()
+    return raw if raw.endswith("USDT") else raw + "USDT"
+
+
+def _tracked_symbols() -> List[str]:
+    configured = os.getenv("APEX_TRACKED_SYMBOLS", DEFAULT_TRACKED_SYMBOLS)
+    return list(dict.fromkeys(_symbol(value) for value in configured.split(",") if value.strip()))
+
+
+def _storage_status() -> Dict[str, Any]:
+    persistent=DB_PATH.is_absolute() and (DB_PATH==Path("/data") or Path("/data") in DB_PATH.parents)
+    return {"path":str(DB_PATH),"mode":"PERSISTENT_VOLUME" if persistent else "EPHEMERAL_LOCAL_DISK","warning":None if persistent else "Set APEX_ARCHIVE_PATH under a mounted persistent volume; local container data can be lost on restart."}
+
+
+def _db() -> sqlite3.Connection:
+    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(str(DB_PATH), timeout=10)
+    conn.row_factory = sqlite3.Row
+    conn.executescript("""
+      CREATE TABLE IF NOT EXISTS forward_events(
+        component TEXT NOT NULL, external_id TEXT NOT NULL, symbol TEXT,
+        event_time INTEGER NOT NULL, observed_at INTEGER NOT NULL,
+        source TEXT NOT NULL, payload_json TEXT NOT NULL,
+        PRIMARY KEY(component, external_id));
+      CREATE INDEX IF NOT EXISTS ix_apex_forward_time ON forward_events(component,event_time);
+    """)
+    return conn
+
+
+async def _json(urls: List[str], path: str, params: Dict[str, Any]) -> Tuple[Any, str, List[Dict[str, Any]]]:
+    diagnostics = []
+    async with httpx.AsyncClient(timeout=httpx.Timeout(25, connect=7), follow_redirects=True, trust_env=True, headers={"User-Agent": "APEX-Historical-Data/1.0", "Accept": "application/json"}) as client:
+        for base in urls:
+            request_path,request_params=path,dict(params);gate_adapter=base.endswith("/api/v4") and path.startswith("/v5/market/")
+            if gate_adapter:
+                contract=str(params["symbol"]).replace("USDT","_USDT")
+                if path.endswith("/funding/history"):request_path="/futures/usdt/funding_rate";request_params={"contract":contract,"limit":min(int(params.get("limit",100)),1000)}
+                else:
+                    request_path="/futures/usdt/contract_stats";request_params={"contract":contract,"interval":{"5min":"5m","15min":"15m","30min":"30m","1h":"1h","4h":"4h","1d":"1d"}[str(params["intervalTime"])],"limit":min(int(params.get("limit",100)),100)}
+                if params.get("startTime") is not None:request_params["from"]=int(params["startTime"])//1000
+                if params.get("endTime") is not None:request_params["to"]=int(params["endTime"])//1000
+            url = base.rstrip("/") + request_path
+            started = time.perf_counter()
+            try:
+                response = await client.get(url, params=request_params)
+                ctype = response.headers.get("content-type", "")
+                if response.status_code == 200 and "json" in ctype.lower():
+                    payload=response.json()
+                    if gate_adapter and isinstance(payload,list):
+                        if path.endswith("/funding/history"):payload={"retCode":0,"result":{"list":[{"symbol":params["symbol"],"fundingRate":item["r"],"fundingRateTimestamp":int(item["t"])*1000} for item in payload]}}
+                        else:payload={"retCode":0,"result":{"list":[{"symbol":params["symbol"],"openInterest":item["open_interest"],"openInterestValue":item.get("open_interest_usd"),"timestamp":int(item["time"])*1000} for item in payload]}}
+                    diagnostics.append({"url": url, "status": 200, "ok": True, "latencyMs": round((time.perf_counter()-started)*1000, 1)})
+                    return payload, base, diagnostics
+                diagnostics.append({"url": url, "status": response.status_code, "ok": False, "error": response.text[:180]})
+            except Exception as exc:
+                diagnostics.append({"url": url, "status": None, "ok": False, "error": f"{type(exc).__name__}: {exc}"})
+    raise RuntimeError("all upstreams failed: " + "; ".join(str(x) for x in diagnostics))
+
+
+async def _funding(symbol: str, start: Optional[int], end: Optional[int], limit: int) -> Dict[str, Any]:
+    sym, cursor = _symbol(symbol), int(start) if start is not None else None
+    total = min(max(limit, 1), 100_000)
+    found: Dict[int, Dict[str, Any]] = {}
+    diagnostics: List[Dict[str, Any]] = []
+    provider = ""
+    source = "binance_futures"
+    try:
+        while len(found) < total:
+            size = min(1000, total-len(found)); params: Dict[str, Any] = {"symbol": sym, "limit": size}
+            if cursor is not None: params["startTime"] = cursor
+            if end is not None: params["endTime"] = int(end)
+            payload, provider, diag = await _json(FUTURES_BASES, "/fapi/v1/fundingRate", params); diagnostics.extend(diag)
+            if not isinstance(payload, list) or not payload: break
+            for item in payload:
+                ts = int(item["fundingTime"]); found[ts] = {"symbol": item.get("symbol", sym), "fundingTime": ts, "fundingRate": float(item["fundingRate"]), "markPrice": float(item["markPrice"]) if item.get("markPrice") not in (None, "") else None, "source": source}
+            nxt = max(found)+1
+            if cursor is None or nxt <= cursor or len(payload) < size or (end is not None and nxt > int(end)): break
+            cursor = nxt; await asyncio.sleep(.03)
+    except RuntimeError as binance_error:
+        diagnostics.append({"provider":"binance_futures","ok":False,"error":str(binance_error)[:500]})
+        found.clear(); source = "bybit_linear"; page_end = int(end) if end is not None else int(time.time()*1000)
+        while len(found) < total:
+            size=min(200,total-len(found)); params={"category":"linear","symbol":sym,"endTime":page_end,"limit":size}
+            if start is not None: params["startTime"]=int(start)
+            payload,provider,diag=await _json(BYBIT_PUBLIC_BASES,"/v5/market/funding/history",params); diagnostics.extend(diag)
+            if not isinstance(payload,dict) or payload.get("retCode")!=0: raise RuntimeError(f"Bybit funding error: {payload}")
+            if provider.endswith("/api/v4"):source="gateio_futures"
+            page=(payload.get("result") or {}).get("list") or []
+            if not page: break
+            for item in page:
+                ts=int(item["fundingRateTimestamp"])
+                if (start is None or ts>=int(start)) and (end is None or ts<=int(end)): found[ts]={"symbol":item.get("symbol",sym),"fundingTime":ts,"fundingRate":float(item["fundingRate"]),"markPrice":None,"source":source}
+            earliest=min(int(item["fundingRateTimestamp"]) for item in page)
+            if (start is not None and earliest<=int(start)) or len(page)<size: break
+            page_end=earliest-1; await asyncio.sleep(.03)
+    data = [found[k] for k in sorted(found)][:total]
+    return {"success": True, "component": "funding", "symbol": sym, "data": data, "count": len(data), "source": source, "providerEndpoint": provider, "sourceMode": "LIVE", "dataState": "REAL" if data else "UNAVAILABLE", "coverage": {"mode": "REAL_HISTORICAL_LIMITED_RETENTION" if source=="gateio_futures" else "REAL_HISTORICAL", "upstreamRetention":"about_180_days" if source=="gateio_futures" else "provider_available_history", "earliestTimestamp": data[0]["fundingTime"] if data else None, "latestTimestamp": data[-1]["fundingTime"] if data else None, "requestedStart": start, "requestedEnd": end}, "diagnostics": diagnostics[-8:], "timestamp": _now()}
+
+
+async def _oi(symbol: str, period: str, start: Optional[int], end: Optional[int], limit: int) -> Dict[str, Any]:
+    allowed = {"5m","15m","30m","1h","2h","4h","6h","12h","1d"}
+    if period not in allowed: raise ValueError(f"period must be one of {sorted(allowed)}")
+    sym, cursor, total = _symbol(symbol), int(start) if start is not None else None, min(max(limit,1),5000)
+    found: Dict[int, Dict[str, Any]] = {}; diagnostics=[]; provider=""
+    source="binance_futures"
+    try:
+        while len(found)<total:
+            size=min(500,total-len(found)); params: Dict[str,Any]={"symbol":sym,"period":period,"limit":size}
+            if cursor is not None: params["startTime"]=cursor
+            if end is not None: params["endTime"]=int(end)
+            payload,provider,diag=await _json(FUTURES_BASES,"/futures/data/openInterestHist",params);diagnostics.extend(diag)
+            if not isinstance(payload,list) or not payload: break
+            for item in payload:
+                ts=int(item["timestamp"]);found[ts]={"symbol":item.get("symbol",sym),"timestamp":ts,"sumOpenInterest":float(item["sumOpenInterest"]),"sumOpenInterestValue":float(item["sumOpenInterestValue"]),"source":source}
+            nxt=max(found)+1
+            if cursor is None or nxt<=cursor or len(payload)<size or (end is not None and nxt>int(end)): break
+            cursor=nxt;await asyncio.sleep(.03)
+    except RuntimeError as binance_error:
+        diagnostics.append({"provider":"binance_futures","ok":False,"error":str(binance_error)[:500]})
+        interval={"5m":"5min","15m":"15min","30m":"30min","1h":"1h","4h":"4h","1d":"1d"}.get(period)
+        if not interval: raise RuntimeError(f"Binance unavailable and Bybit has no native {period} OI interval") from binance_error
+        found.clear();source="bybit_linear";bybit_cursor:Optional[str]=None
+        while len(found)<total:
+            size=min(200,total-len(found));params={"category":"linear","symbol":sym,"intervalTime":interval,"limit":size}
+            if start is not None:params["startTime"]=int(start)
+            if end is not None:params["endTime"]=int(end)
+            if bybit_cursor:params["cursor"]=bybit_cursor
+            payload,provider,diag=await _json(BYBIT_PUBLIC_BASES,"/v5/market/open-interest",params);diagnostics.extend(diag)
+            if not isinstance(payload,dict) or payload.get("retCode")!=0:raise RuntimeError(f"Bybit OI error: {payload}")
+            if provider.endswith("/api/v4"):source="gateio_futures"
+            result=payload.get("result") or {};page=result.get("list") or []
+            if not page:break
+            for item in page:
+                ts=int(item["timestamp"]);found[ts]={"symbol":sym,"timestamp":ts,"sumOpenInterest":float(item["openInterest"]),"sumOpenInterestValue":float(item["openInterestValue"]) if item.get("openInterestValue") is not None else None,"source":source}
+            next_page=result.get("nextPageCursor")
+            if not next_page or next_page==bybit_cursor or len(page)<size:break
+            bybit_cursor=next_page;await asyncio.sleep(.03)
+    data=[found[k] for k in sorted(found)][:total]
+    return {"success":True,"component":"openInterest","symbol":sym,"period":period,"data":data,"count":len(data),"source":source,"providerEndpoint":provider,"sourceMode":"LIVE","dataState":"REAL" if data else "UNAVAILABLE","coverage":{"mode":"REAL_HISTORICAL" if source=="bybit_linear" else "REAL_HISTORICAL_LIMITED_RETENTION","upstreamRetention":"about_180_days" if source=="gateio_futures" else ("provider_available_history" if source=="bybit_linear" else "latest_30_days"),"earliestTimestamp":data[0]["timestamp"] if data else None,"latestTimestamp":data[-1]["timestamp"] if data else None,"warning":"Gate.io retains about 180 days; Bybit is geo-blocked from the current HF runtime." if source=="gateio_futures" else (None if source=="bybit_linear" else "Binance retains only the latest 30 days.")},"diagnostics":diagnostics[-8:],"timestamp":_now()}
+
+
+def _save(component: str, source: str, items: List[Dict[str, Any]]) -> int:
+    observed=int(time.time()*1000);saved=0
+    with _db() as conn:
+        for item in items:
+            fallback=hashlib.sha256(json.dumps(item,sort_keys=True,default=str).encode()).hexdigest()
+            tx_hash=item.get("txHash")
+            external=str(item.get("id") or (f"{item.get('chain','unknown')}:{tx_hash}" if tx_hash else None) or item.get("url") or item.get("link") or fallback)
+            event=int(item.get("eventTime") or item.get("publishedOn") or item.get("published_on") or item.get("timestamp") or observed)
+            if event<10_000_000_000:event*=1000
+            before=conn.total_changes
+            conn.execute("INSERT OR IGNORE INTO forward_events VALUES(?,?,?,?,?,?,?)",(component,external,item.get("symbol"),event,observed,source,json.dumps(item,ensure_ascii=False)))
+            saved+=conn.total_changes-before
+        conn.commit()
+    return saved
+
+
+def _save_oi(items: List[Dict[str, Any]]) -> int:
+    observed=int(time.time()*1000);processed=0
+    with _db() as conn:
+        for item in items:
+            symbol=_symbol(str(item.get("symbol") or ""));timestamp=int(item["timestamp"])
+            conn.execute("""INSERT INTO forward_events(component,external_id,symbol,event_time,observed_at,source,payload_json)
+                            VALUES(?,?,?,?,?,?,?)
+                            ON CONFLICT(component,external_id) DO UPDATE SET
+                              source=excluded.source,payload_json=excluded.payload_json""",
+                         ("openInterest",f"{symbol}:{timestamp}",symbol,timestamp,observed,item.get("source","unknown"),json.dumps(item,ensure_ascii=False)))
+            processed+=1
+        conn.commit()
+    return processed
+
+
+def _archive_count(component: str, symbol: Optional[str]=None) -> int:
+    with _db() as conn:
+        if symbol: row=conn.execute("SELECT COUNT(*) FROM forward_events WHERE component=? AND symbol=?",(component,_symbol(symbol))).fetchone()
+        else: row=conn.execute("SELECT COUNT(*) FROM forward_events WHERE component=?",(component,)).fetchone()
+    return int(row[0])
+
+
+async def collect_oi_archive(period: str="1h") -> Dict[str, Any]:
+    results=[]
+    for symbol in _tracked_symbols():
+        try:
+            response=await _oi(symbol,period,None,None,5000);items=response.get("data") or []
+            _save_oi(items)
+            results.append({"symbol":symbol,"success":True,"fetched":len(items),"archiveRows":_archive_count("openInterest",symbol),"earliestTimestamp":items[0]["timestamp"] if items else None,"latestTimestamp":items[-1]["timestamp"] if items else None})
+        except Exception as exc: results.append({"symbol":symbol,"success":False,"error":f"{type(exc).__name__}: {exc}"})
+    return {"success":any(x["success"] for x in results),"component":"openInterest","schedule":"daily","dedupeKey":["symbol","timestamp"],"trackedSymbols":_tracked_symbols(),"results":results,"coverageMode":"FORWARD_ARCHIVE_PLUS_30_DAY_UPSTREAM_WINDOW","warning":"No OI before archive collection began can be recovered from Binance's 30-day endpoint.","timestamp":_now()}
+
+
+def _read(component: str, start: Optional[int], end: Optional[int], limit: int) -> Dict[str, Any]:
+    where=["component=?"];params:List[Any]=[component]
+    if start is not None:where.append("event_time>=?");params.append(int(start))
+    if end is not None:where.append("event_time<=?");params.append(int(end))
+    params.append(min(max(limit,1),10000))
+    with _db() as conn:
+        rows=conn.execute(f"SELECT * FROM forward_events WHERE {' AND '.join(where)} ORDER BY event_time LIMIT ?",params).fetchall()
+        first=conn.execute("SELECT MIN(observed_at) FROM forward_events WHERE component=?",(component,)).fetchone()[0]
+    data=[{**json.loads(r["payload_json"]),"eventTime":r["event_time"],"observedAt":r["observed_at"],"archiveSource":r["source"]} for r in rows]
+    return {"success":True,"component":component,"data":data,"count":len(data),"sourceMode":"CACHED","dataState":"REAL" if data else "UNAVAILABLE","coverage":{"mode":"FORWARD_COLLECTING_ONLY","archiveStartedAt":first,"preExistingHistoryAvailable":False,"warning":"Latest/live upstream; archive contains only records observed after collection began."},"timestamp":_now()}
+
+
+def _read_oi(symbol: str,start:Optional[int],end:Optional[int],limit:int) -> Dict[str,Any]:
+    sym=_symbol(symbol);where=["component='openInterest'","symbol=?"];params:List[Any]=[sym]
+    if start is not None:where.append("event_time>=?");params.append(int(start))
+    if end is not None:where.append("event_time<=?");params.append(int(end))
+    params.append(min(max(limit,1),100000))
+    with _db() as conn:
+        rows=conn.execute(f"SELECT * FROM forward_events WHERE {' AND '.join(where)} ORDER BY event_time LIMIT ?",params).fetchall()
+        first=conn.execute("SELECT MIN(observed_at) FROM forward_events WHERE component='openInterest'").fetchone()[0]
+    data=[json.loads(r["payload_json"]) for r in rows]
+    return {"success":True,"component":"openInterest","symbol":sym,"data":data,"count":len(data),"sourceMode":"CACHED","dataState":"REAL" if data else "UNAVAILABLE","coverage":{"mode":"REAL_HISTORICAL_ON_DEMAND_PLUS_FORWARD_ARCHIVE","archiveStartedAt":first,"preDeploymentHistoryAvailable":False,"earliestTimestamp":data[0]["timestamp"] if data else None,"latestTimestamp":data[-1]["timestamp"] if data else None,"warning":"This cached archive starts at first collection; older history is available separately on demand through the Bybit fallback."},"timestamp":_now()}
+
+
+async def _scan_evm_whales(limit:int) -> Dict[str,Any]:
+    shared_blocks=os.getenv("WHALE_SCAN_BLOCKS")
+    chains=[
+        {"chain":"ethereum","symbol":"ETH","blocks":min(max(int(shared_blocks or os.getenv("WHALE_ETH_SCAN_BLOCKS","15")),1),100),"threshold":float(os.getenv("WHALE_ETH_MIN_NATIVE","50")),"urls":["https://ethereum.publicnode.com","https://eth.llamarpc.com"]},
+        {"chain":"bsc","symbol":"BNB","blocks":min(max(int(shared_blocks or os.getenv("WHALE_BSC_SCAN_BLOCKS","200")),1),500),"threshold":float(os.getenv("WHALE_BNB_MIN_NATIVE","500")),"urls":["https://bsc-rpc.publicnode.com","https://bsc-dataseed.binance.org"]},
+    ]
+    found=[];diagnostics=[]
+    async with httpx.AsyncClient(timeout=httpx.Timeout(25,connect=7),follow_redirects=True,trust_env=True,headers={"User-Agent":"APEX-Whale-Scanner/1.0"}) as client:
+        for config in chains:
+            active=None;latest=None
+            for url in config["urls"]:
+                try:
+                    response=await client.post(url,json={"jsonrpc":"2.0","id":1,"method":"eth_blockNumber","params":[]});latest=(response.json() or {}).get("result")
+                    if response.status_code==200 and latest:active=url;break
+                except Exception as exc:diagnostics.append({"chain":config["chain"],"provider":url,"ok":False,"error":type(exc).__name__})
+            if not active or not latest:
+                diagnostics.append({"chain":config["chain"],"ok":False,"error":"no public RPC available"});continue
+            latest_number=int(latest,16)
+            block_count=config["blocks"]
+            calls=[client.post(active,json={"jsonrpc":"2.0","id":number,"method":"eth_getBlockByNumber","params":[hex(number),True]}) for number in range(latest_number-block_count+1,latest_number+1)]
+            responses=await asyncio.gather(*calls,return_exceptions=True);scanned=0
+            for response in responses:
+                if isinstance(response,Exception) or response.status_code!=200:continue
+                block=(response.json() or {}).get("result") or {};event_time=int(block.get("timestamp","0x0"),16)*1000
+                for tx in block.get("transactions") or []:
+                    scanned+=1;amount=int(tx.get("value","0x0"),16)/10**18
+                    if amount<config["threshold"]:continue
+                    found.append({"txHash":tx.get("hash"),"chain":config["chain"],"symbol":config["symbol"],"from":tx.get("from"),"to":tx.get("to"),"amount":amount,"eventTime":event_time,"blockNumber":int(block.get("number","0x0"),16),"thresholdNative":config["threshold"],"classification":"LARGE_NATIVE_TRANSFER","source":"public_evm_rpc"})
+            diagnostics.append({"chain":config["chain"],"provider":active,"ok":True,"blocksScanned":block_count,"transactionsScanned":scanned,"matches":sum(1 for item in found if item["chain"]==config["chain"])})
+    found.sort(key=lambda item:item["eventTime"],reverse=True)
+    return {"items":found[:min(max(limit,1),1000)],"diagnostics":diagnostics,"pollInterval":"2 minutes"}
+
+
+async def collect_forward() -> Dict[str, Any]:
+    results=[]
+    try:
+        # NEWSAPI_KEY is the legacy Space-secret name for this verified
+        # newsdata.io credential, not an assertion that it is NewsAPI.org.
+        newsdata_key=os.getenv("NEWSDATA_API_KEY") or os.getenv("NEWSDATA_KEY") or NEWSDATA_FALLBACK_KEY
+        if newsdata_key:
+            payload,_,diag=await _json(["https://newsdata.io"],"/api/1/latest",{"apikey":newsdata_key,"q":"cryptocurrency","language":"en"})
+            raw=(payload or {}).get("results") if isinstance(payload,dict) else []
+            items=[]
+            for x in (raw or [])[:200]:
+                try:published=int(datetime.fromisoformat(str(x.get("pubDate") or x.get("pubDateTZ") or "").replace("Z","+00:00")).timestamp())
+                except Exception:published=int(time.time())
+                items.append({"id":x.get("article_id") or x.get("link"),"title":x.get("title"),"body":x.get("description") or x.get("content"),"url":x.get("link"),"source":x.get("source_name") or x.get("source_id"),"publishedOn":published,"categories":x.get("category")})
+            results.append({"component":"news","success":True,"fetched":len(items),"saved":_save("news","newsdata_latest",items),"source":"newsdata_latest","coverageMode":"FORWARD_COLLECTING_ONLY"})
+        else:
+            cc_key=os.getenv("CRYPTOCOMPARE_KEY") or os.getenv("CRYPTOCOMPARE_API_KEY")
+            if not cc_key: raise RuntimeError("No authenticated latest-news provider configured; set NEWSDATA_API_KEY")
+            payload,_,diag=await _json(["https://min-api.cryptocompare.com"],"/data/v2/news/",{"lang":"EN","api_key":cc_key})
+            raw=(payload or {}).get("Data") if isinstance(payload,dict) else []
+            items=[{"id":x.get("id"),"title":x.get("title"),"body":x.get("body"),"url":x.get("url"),"source":x.get("source"),"publishedOn":x.get("published_on"),"categories":x.get("categories")} for x in (raw or [])[:200]]
+            results.append({"component":"news","success":True,"fetched":len(items),"saved":_save("news","cryptocompare_news",items),"source":"cryptocompare_news","coverageMode":"FORWARD_COLLECTING_ONLY"})
+    except Exception as exc:results.append({"component":"news","success":False,"error":f"{type(exc).__name__}: {exc}"})
+    try:
+        whale_key=os.getenv("WHALE_ALERT_API_KEY") or os.getenv("WHALE_ALERT_KEY")
+        if whale_key:
+            payload,_,diag=await _json(["https://api.whale-alert.io"],"/v1/transactions",{"api_key":whale_key,"start":int(time.time())-3600,"min_value":500000,"limit":100});raw=payload.get("transactions") if isinstance(payload,dict) else payload;source="whale_alert"
+            items=raw if isinstance(raw,list) else []
+        else:
+            scan=await _scan_evm_whales(100);items=scan["items"];diag=scan["diagnostics"];source="public_evm_rpc_large_native_transfers"
+        results.append({"component":"whaleFlow","success":True,"fetched":len(items),"saved":_save("whaleFlow",source,items),"source":source,"scope":"large native ETH/BNB transfers in recent blocks; no owner labels","coverageMode":"FORWARD_COLLECTING_ONLY","diagnostics":diag})
+    except Exception as exc:results.append({"component":"whaleFlow","success":False,"error":f"{type(exc).__name__}: {exc}"})
+    return {"success":any(x["success"] for x in results),"results":results,"timestamp":_now()}
+
+
+async def backfill_newsapi(start_date: str, end_date: str, max_pages: int) -> Dict[str, Any]:
+    key=os.getenv("NEWSAPI_KEY") or os.getenv("NEWS_API_KEY")
+    if not key:return {"success":False,"component":"news","source":"newsapi","dataState":"UNAVAILABLE","error":"NEWSAPI_KEY not configured","coverageMode":"CONDITIONAL_DATED_SEARCH"}
+    fetched=saved=0;diagnostics=[]
+    for page in range(1,min(max(max_pages,1),20)+1):
+        payload,_,diag=await _json(["https://newsapi.org"],"/v2/everything",{"q":"cryptocurrency OR bitcoin OR ethereum","from":start_date,"to":end_date,"sortBy":"publishedAt","language":"en","pageSize":100,"page":page,"apiKey":key});diagnostics.extend(diag)
+        articles=payload.get("articles") if isinstance(payload,dict) else []
+        if not articles:break
+        normalized=[]
+        for item in articles:
+            try:published=int(datetime.fromisoformat(str(item.get("publishedAt","")).replace("Z","+00:00")).timestamp())
+            except Exception:published=int(time.time())
+            normalized.append({"id":item.get("url"),"title":item.get("title"),"body":item.get("content") or item.get("description"),"url":item.get("url"),"source":(item.get("source") or {}).get("name"),"publishedOn":published})
+        fetched+=len(normalized);saved+=_save("news","newsapi",normalized)
+        if len(articles)<100:break
+        await asyncio.sleep(.1)
+    return {"success":True,"component":"news","source":"newsapi","fetched":fetched,"saved":saved,"coverageMode":"CONDITIONAL_DATED_SEARCH","requestedStart":start_date,"requestedEnd":end_date,"complete":False,"warning":"Coverage depends on the configured NewsAPI plan and query matching; it is not asserted complete.","diagnostics":diagnostics[-5:]}
+
+
+async def periodic_archive(stop: asyncio.Event) -> None:
+    last_oi_collection=0.0
+    last_news_collection=0.0
+    while not stop.is_set():
+        if time.time()-last_news_collection >= 900:
+            try:
+                await collect_forward()
+                last_news_collection=time.time()
+            except Exception: pass
+        else:
+            try:
+                scan=await _scan_evm_whales(1000)
+                _save("whaleFlow","public_evm_rpc_large_native_transfers",scan["items"])
+            except Exception: pass
+        if time.time()-last_oi_collection >= 86400:
+            try:
+                await collect_oi_archive("1h")
+                last_oi_collection=time.time()
+            except Exception: pass
+        try: await asyncio.wait_for(stop.wait(),timeout=120)
+        except asyncio.TimeoutError: pass
+
+
+@router.get("/coverage")
+async def coverage():
+    return {"success":True,"persistencePolicy":"on_demand_market_history_plus_forward_event_archive","archiveStorage":_storage_status(),"components":{"ohlcv":{"mode":"REAL_HISTORICAL_ON_DEMAND","endpoint":"/api/ohlcv"},"funding":{"mode":"REAL_180_DAY_UPSTREAM_IN_CURRENT_RUNTIME","endpoint":"/api/apex/funding/{symbol}"},"openInterest":{"mode":"REAL_180_DAY_UPSTREAM_PLUS_FORWARD_ARCHIVE","retention":"Gate.io fallback ~180 days; Binance/Bybit are geo-blocked from current HF runtime; daily local archive","liveEndpoint":"/api/apex/open-interest/{symbol}","archiveEndpoint":"/api/apex/open-interest-archive/{symbol}","trackedSymbols":_tracked_symbols()},"sentiment":{"mode":"REAL_HISTORICAL","endpoint":"/api/apex/sentiment/fear-greed"},"news":{"mode":"FORWARD_COLLECTING_ONLY","endpoint":"/api/apex/news","liveSource":"newsdata.io authenticated latest feed","preDeploymentHistoryAvailable":False},"whaleFlow":{"mode":"FORWARD_COLLECTING_ONLY","endpoint":"/api/apex/whale-flow","liveSource":"public Ethereum/BSC RPC large native transfers","scope":"configurable native-value threshold; no wallet-owner labels","preDeploymentHistoryAvailable":False}},"manifest":{"found":False,"expectedPath":"QA/profitability-structural-remediation/data/manifest.json","behavior":"APEX_TRACKED_SYMBOLS is configurable; defaults follow the project's core dashboard symbols"}}
+
+
+@router.get("/funding/{symbol}")
+async def funding(symbol:str,start:Optional[int]=None,end:Optional[int]=None,limit:int=Query(1000,ge=1,le=100000)):
+    try:return await _funding(symbol,start,end,limit)
+    except Exception as exc:raise HTTPException(502,detail=f"funding upstream unavailable: {type(exc).__name__}: {exc}")
+
+
+@router.get("/open-interest/{symbol}")
+async def open_interest(symbol:str,period:str="1h",start:Optional[int]=None,end:Optional[int]=None,limit:int=Query(500,ge=1,le=5000)):
+    try:return await _oi(symbol,period,start,end,limit)
+    except ValueError as exc:raise HTTPException(422,detail=str(exc))
+    except Exception as exc:raise HTTPException(502,detail=f"open-interest upstream unavailable: {type(exc).__name__}: {exc}")
+
+
+@router.get("/open-interest-archive/{symbol}")
+async def open_interest_archive(symbol:str,start:Optional[int]=None,end:Optional[int]=None,limit:int=Query(10000,ge=1,le=100000)):
+    return _read_oi(symbol,start,end,limit)
+
+
+@router.post("/open-interest/archive/collect")
+async def collect_open_interest_archive(period:str="1h"):
+    try:return await collect_oi_archive(period)
+    except ValueError as exc:raise HTTPException(422,detail=str(exc))
+
+
+@router.get("/sentiment/fear-greed")
+async def fear_greed(limit:int=Query(0,ge=0,le=100000)):
+    try:
+        payload,provider,diag=await _json(["https://api.alternative.me"],"/fng/",{"limit":limit,"format":"json"})
+        data=payload.get("data") if isinstance(payload,dict) else []
+        return {"success":True,"component":"sentiment","data":data or [],"count":len(data or []),"source":"alternative.me","sourceMode":"LIVE","dataState":"REAL" if data else "UNAVAILABLE","coverage":{"mode":"REAL_HISTORICAL","earliestTimestamp":data[-1].get("timestamp") if data else None,"latestTimestamp":data[0].get("timestamp") if data else None},"diagnostics":diag,"timestamp":_now()}
+    except Exception as exc:raise HTTPException(502,detail=f"sentiment upstream unavailable: {type(exc).__name__}: {exc}")
+
+
+@router.get("/news")
+async def news(start:Optional[int]=None,end:Optional[int]=None,limit:int=Query(500,ge=1,le=10000)):return _read("news",start,end,limit)
+
+
+@router.get("/whale-flow")
+async def whales(start:Optional[int]=None,end:Optional[int]=None,limit:int=Query(500,ge=1,le=10000)):return _read("whaleFlow",start,end,limit)
+
+
+@router.post("/archive/collect")
+async def collect():return await collect_forward()
+
+
+@router.post("/news/backfill")
+async def news_backfill(start:str,end:str,max_pages:int=Query(5,ge=1,le=20)):
+    try:return await backfill_newsapi(start,end,max_pages)
+    except Exception as exc:raise HTTPException(502,detail=f"NewsAPI dated search failed: {type(exc).__name__}: {exc}")
+
+
+def register_apex_routes(app) -> None: app.include_router(router)

api_compat_routes.py

@@ -20,6 +20,7 @@ from __future__ import annotations
 import logging
 import os
 import json
+import asyncio
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -270,63 +271,119 @@ async def _fetch_coingecko_trending(limit: int = 10) -> Tuple[List[Dict[str, Any
     return rows, []
 
 
-async def _fetch_binance_klines(symbol: str, interval: str, limit: int) -> Tuple[List[Dict[str, Any]], List[str]]:
+async def _fetch_binance_klines(
+    symbol: str,
+    interval: str,
+    limit: int,
+    start_time: Optional[int] = None,
+    end_time: Optional[int] = None,
+) -> Tuple[List[Dict[str, Any]], List[str]]:
     pair = _normalize_symbol(symbol)
     mapped = BINANCE_INTERVALS.get(interval, interval)
-    payload, error, _ = await _get_json(
+    total_limit = min(max(limit, 1), 200_000)
+    urls = [
+        "https://data-api.binance.vision/api/v3/klines",
         "https://api.binance.com/api/v3/klines",
-        params={"symbol": pair, "interval": mapped, "limit": min(max(limit, 1), 1000)},
-        timeout=20.0,
-    )
-    if error or not isinstance(payload, list):
-        return [], [error or f"Binance klines invalid payload for {pair}"]
-    candles = []
-    for row in payload:
-        try:
-            candles.append({
-                "timestamp": _int(row[0]),
-                "open": _float(row[1]),
-                "high": _float(row[2]),
-                "low": _float(row[3]),
-                "close": _float(row[4]),
-                "volume": _float(row[5]),
-            })
-        except Exception:
-            continue
-    return candles, []
-
-
-async def _fetch_kucoin_klines(symbol: str, timeframe: str, limit: int) -> Tuple[List[Dict[str, Any]], List[str]]:
+        "https://api1.binance.com/api/v3/klines",
+    ]
+    rows: Dict[int, Dict[str, Any]] = {}
+    errors: List[str] = []
+    cursor_start = int(start_time) if start_time is not None else None
+    cursor_end = int(end_time) if end_time is not None else None
+    forward = cursor_start is not None
+    while len(rows) < total_limit:
+        page_size = min(1000, total_limit - len(rows))
+        params: Dict[str, Any] = {"symbol": pair, "interval": mapped, "limit": page_size}
+        if cursor_start is not None:
+            params["startTime"] = cursor_start
+        if cursor_end is not None:
+            params["endTime"] = cursor_end
+        payload = None
+        last_error = None
+        for url in urls:
+            candidate, error, _ = await _get_json(url, params=params, timeout=25.0)
+            if not error and isinstance(candidate, list):
+                payload = candidate
+                break
+            last_error = error or f"invalid payload from {url}"
+        if not isinstance(payload, list):
+            errors.append(last_error or f"Binance klines invalid payload for {pair}")
+            break
+        if not payload:
+            break
+        for row in payload:
+            try:
+                ts = _int(row[0])
+                rows[ts] = {"timestamp": ts, "open": _float(row[1]), "high": _float(row[2]), "low": _float(row[3]), "close": _float(row[4]), "volume": _float(row[5])}
+            except Exception:
+                continue
+        first_ts, last_ts = _int(payload[0][0]), _int(payload[-1][0])
+        if forward:
+            next_start = last_ts + 1
+            if next_start <= (cursor_start or -1) or (end_time is not None and next_start > int(end_time)):
+                break
+            cursor_start = next_start
+        elif cursor_end is not None:
+            next_end = first_ts - 1
+            if next_end >= cursor_end:
+                break
+            cursor_end = next_end
+        else:
+            break
+        if len(payload) < page_size:
+            break
+        await asyncio.sleep(0.03)
+    candles = [rows[key] for key in sorted(rows)]
+    return (candles[:total_limit] if forward else candles[-total_limit:]), errors
+
+
+async def _fetch_kucoin_klines(
+    symbol: str,
+    timeframe: str,
+    limit: int,
+    start_time: Optional[int] = None,
+    end_time: Optional[int] = None,
+) -> Tuple[List[Dict[str, Any]], List[str]]:
     # KuCoin returns reverse chronological rows like [time, open, close, high, low, volume, turnover].
     ksymbol = _kucoin_symbol(symbol)
     ktype = KUCOIN_TYPES.get(timeframe, "1hour")
-    payload, error, _ = await _get_json(
-        "https://api.kucoin.com/api/v1/market/candles",
-        params={"symbol": ksymbol, "type": ktype},
-        timeout=20.0,
-    )
-    if error or not isinstance(payload, dict):
-        return [], [error or f"KuCoin candles invalid payload for {ksymbol}"]
-    data = payload.get("data") or []
-    candles = []
-    for row in data[: min(max(limit, 1), 1500)]:
-        try:
-            candles.append({
-                "timestamp": _int(row[0]) * 1000,
-                "open": _float(row[1]),
-                "high": _float(row[3]),
-                "low": _float(row[4]),
-                "close": _float(row[2]),
-                "volume": _float(row[5]),
-            })
-        except Exception:
-            continue
-    candles.sort(key=lambda c: c["timestamp"])
-    return candles[-limit:], []
+    total_limit = min(max(limit, 1), 200_000)
+    rows: Dict[int, Dict[str, Any]] = {}
+    errors: List[str] = []
+    cursor_start = int(start_time / 1000) if start_time is not None else None
+    cursor_end = int(end_time / 1000) if end_time is not None else None
+    # KuCoin pages are naturally bounded by startAt/endAt. Move the end cursor
+    # backwards until the requested start is covered.
+    while len(rows) < total_limit:
+        params: Dict[str, Any] = {"symbol": ksymbol, "type": ktype}
+        if cursor_start is not None:
+            params["startAt"] = cursor_start
+        if cursor_end is not None:
+            params["endAt"] = cursor_end
+        payload, error, _ = await _get_json("https://api.kucoin.com/api/v1/market/candles", params=params, timeout=25.0)
+        if error or not isinstance(payload, dict):
+            errors.append(error or f"KuCoin candles invalid payload for {ksymbol}")
+            break
+        data = payload.get("data") or []
+        if not data:
+            break
+        for row in data:
+            try:
+                ts = _int(row[0]) * 1000
+                rows[ts] = {"timestamp": ts, "open": _float(row[1]), "high": _float(row[3]), "low": _float(row[4]), "close": _float(row[2]), "volume": _float(row[5])}
+            except Exception:
+                continue
+        oldest_sec = min(_int(row[0]) for row in data)
+        if cursor_start is None or oldest_sec <= cursor_start or len(data) < 100:
+            break
+        cursor_end = oldest_sec - 1
+        await asyncio.sleep(0.03)
+    candles = [rows[key] for key in sorted(rows)]
+    return candles[-total_limit:], errors
 
 
 
-async def _fetch_cryptocompare_ohlcv(symbol: str, timeframe: str, limit: int) -> Tuple[List[Dict[str, Any]], List[str]]:
+async def _fetch_cryptocompare_ohlcv(symbol: str, timeframe: str, limit: int, start_time: Optional[int] = None, end_time: Optional[int] = None) -> Tuple[List[Dict[str, Any]], List[str]]:
     """Optional OHLCV fallback via CryptoCompare.
 
     Uses env CRYPTOCOMPARE_KEY when configured, but can still work on some
@@ -353,6 +410,8 @@ async def _fetch_cryptocompare_ohlcv(symbol: str, timeframe: str, limit: int) ->
         "limit": min(max(limit, 1), 2000),
         "aggregate": aggregate,
     }
+    if end_time is not None:
+        params["toTs"] = int(end_time / 1000)
     key = get_secret("CRYPTOCOMPARE_KEY")
     if key:
         params["api_key"] = key
@@ -379,7 +438,7 @@ async def _fetch_cryptocompare_ohlcv(symbol: str, timeframe: str, limit: int) ->
             continue
     return candles, []
 
-async def _fetch_ohlcv(symbol: str, timeframe: str, limit: int) -> Tuple[List[Dict[str, Any]], str, List[str]]:
+async def _fetch_ohlcv(symbol: str, timeframe: str, limit: int, start_time: Optional[int] = None, end_time: Optional[int] = None) -> Tuple[List[Dict[str, Any]], str, List[str]]:
     """Smart OHLCV rotation.
 
     Raw OHLCV must come from market/exchange providers. HF models are used for
@@ -388,17 +447,17 @@ async def _fetch_ohlcv(symbol: str, timeframe: str, limit: int) -> Tuple[List[Di
     """
     errors: List[str] = []
 
-    candles, err = await _fetch_binance_klines(symbol, timeframe, limit)
+    candles, err = await _fetch_binance_klines(symbol, timeframe, limit, start_time, end_time)
     if candles:
         return candles, "binance_public", errors + err
     errors.extend(err)
 
-    candles, err = await _fetch_kucoin_klines(symbol, timeframe, limit)
+    candles, err = await _fetch_kucoin_klines(symbol, timeframe, limit, start_time, end_time)
     if candles:
         return candles, "kucoin_public", errors + err
     errors.extend(err)
 
-    candles, err = await _fetch_cryptocompare_ohlcv(symbol, timeframe, limit)
+    candles, err = await _fetch_cryptocompare_ohlcv(symbol, timeframe, limit, start_time, end_time)
     if candles:
         return candles, "cryptocompare_ohlcv", errors + err
     errors.extend(err)
@@ -569,11 +628,28 @@ async def coins_top_alias(limit: int = Query(100, ge=1, le=250)):
 
 
 @router.get("/api/trading/ohlcv/{symbol}")
-async def trading_ohlcv(symbol: str, timeframe: str = "1h", limit: int = 100):
-    candles, source, errors = await _fetch_ohlcv(symbol, timeframe, limit)
+async def trading_ohlcv(symbol: str, timeframe: str = "1h", limit: int = 100, start_time: Optional[int] = None, end_time: Optional[int] = None):
+    candles, source, errors = await _fetch_ohlcv(symbol, timeframe, limit, start_time, end_time)
     if not candles:
         return _fail("OHLCV unavailable", source=source, symbol=_normalize_symbol(symbol), timeframe=timeframe, data=[], missingCapabilities=["ohlcv"], upstreamErrors=errors)
-    return _ok(data=candles, symbol=_normalize_symbol(symbol), timeframe=timeframe, candles=candles, source=source)
+    return _ok(
+        data=candles,
+        symbol=_normalize_symbol(symbol),
+        timeframe=timeframe,
+        candles=candles,
+        count=len(candles),
+        source=source,
+        sourceMode="LIVE",
+        dataState="REAL",
+        coverage={
+            "mode": "REAL_HISTORICAL" if start_time is not None or end_time is not None else "LIVE_RECENT",
+            "requestedStart": start_time,
+            "requestedEnd": end_time,
+            "earliestTimestamp": candles[0]["timestamp"] if candles else None,
+            "latestTimestamp": candles[-1]["timestamp"] if candles else None,
+            "rowCount": len(candles),
+        },
+    )
 
 
 @router.get("/api/ohlcv")
@@ -582,15 +658,30 @@ async def trading_ohlcv_alias(
     symbol: str = Query("BTCUSDT"),
     interval: str = Query("1h"),
     timeframe: Optional[str] = Query(None),
-    limit: int = Query(100, ge=1, le=1000),
+    limit: int = Query(100, ge=1, le=200000),
+    since: Optional[int] = Query(None),
+    start: Optional[int] = Query(None),
+    end: Optional[int] = Query(None),
+    startTime: Optional[int] = Query(None),
+    endTime: Optional[int] = Query(None),
 ):
-    return await trading_ohlcv(symbol=symbol, timeframe=(timeframe or interval), limit=limit)
+    return await trading_ohlcv(symbol=symbol, timeframe=(timeframe or interval), limit=limit, start_time=(startTime or start or since), end_time=(endTime or end))
 
 
 @router.get("/api/history")
 @router.get("/api/trading/history/{symbol}")
-async def history_alias(symbol: str = "BTCUSDT", interval: str = "1h", timeframe: Optional[str] = None, limit: int = Query(100, ge=1, le=1000)):
-    return await trading_ohlcv(symbol=symbol, timeframe=(timeframe or interval), limit=limit)
+async def history_alias(
+    symbol: str = "BTCUSDT",
+    interval: str = "1h",
+    timeframe: Optional[str] = None,
+    limit: int = Query(100, ge=1, le=200000),
+    since: Optional[int] = None,
+    start: Optional[int] = None,
+    end: Optional[int] = None,
+    startTime: Optional[int] = None,
+    endTime: Optional[int] = None,
+):
+    return await trading_ohlcv(symbol=symbol, timeframe=(timeframe or interval), limit=limit, start_time=(startTime or start or since), end_time=(endTime or end))
 
 
 @router.get("/api/trading/stats/24h/{symbol}")

api_server_extended.py

@@ -688,9 +688,24 @@ async def lifespan(app: FastAPI):
     except Exception as e:
         print(f"[WARN] Resource validation failed: {e}")
     
+    archive_stop = asyncio.Event()
+    archive_task = None
+    try:
+        from apex_strategy_routes import periodic_archive
+        archive_task = asyncio.create_task(periodic_archive(archive_stop))
+        print("[OK] APEX news/whale forward archive (15 min) + daily OI archive started")
+    except Exception as archive_exc:
+        print(f"[WARN] APEX forward archive did not start: {archive_exc}")
+
     print(f"[OK] Server ready on port {PORT}")
     print("=" * 80)
     yield
+    archive_stop.set()
+    if archive_task is not None:
+        try:
+            await asyncio.wait_for(archive_task, timeout=5)
+        except Exception:
+            archive_task.cancel()
     print("Shutting down...")
 
 
@@ -784,6 +799,14 @@ async def index():
     """Serve index.html"""
     return _serve_html_asset("index.html", empty_title="index.html not found")
 
+@app.get("/data-lab", response_class=HTMLResponse)
+async def data_lab():
+    return _serve_html_asset("data_lab.html", empty_title="APEX Data Lab not found")
+
+@app.get("/help", response_class=HTMLResponse)
+async def help_page():
+    return _serve_html_asset("help.html", empty_title="API help not found")
+
 
 @app.get("/dashboard.html", response_class=HTMLResponse)
 async def dashboard():
@@ -4142,6 +4165,14 @@ try:
 except Exception as short_hunter_error:
     logger.warning(f"Short Hunter datasource routes not loaded: {short_hunter_error}")
 
+# ===== APEX strategy history + truthful forward archives =====
+try:
+    from apex_strategy_routes import register_apex_routes
+    register_apex_routes(app)
+    logger.info("APEX historical data routes loaded (/api/apex/*)")
+except Exception as apex_routes_error:
+    logger.warning(f"APEX historical data routes not loaded: {apex_routes_error}")
+
 # ===== Main Entry Point =====
 if __name__ == "__main__":
     import uvicorn

data_lab.html

@@ -0,0 +1 @@
+<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>APEX Historical Data Lab · Space 4</title><link rel="stylesheet" href="/static/css/apex-data-lab.css"></head><body><header><div class="brand"><div class="orb"></div><div><strong>APEX Data Lab</strong><div class="eyebrow">Datasource Space 4</div></div></div><nav><a href="/">Dashboard</a><a href="/help">Help & provenance</a><a href="/docs">OpenAPI</a></nav></header><main><section class="hero"><div class="eyebrow">Real data · explicit coverage · LIVE snapshot contract preserved</div><h1>Historical depth, without the marketing fog.</h1><p>Inspect real paginated candles and strategy inputs. Every panel surfaces provenance and freshness so recent-only data cannot masquerade as historical coverage.</p></section><section class="grid"><div class="panel controls"><h2>Historical request</h2><div class="form"><div><label>Symbol</label><input id="symbol" value="BTCUSDT"></div><div><label>Interval</label><select id="interval"><option>1h</option><option selected>4h</option><option>1d</option></select></div><div><label>Start</label><input id="start" type="date" value="2022-01-01"></div><div><label>End</label><input id="end" type="date" value="2022-03-01"></div><button id="load">Load real candles</button></div><div class="status"><span id="dot" class="dot"></span><span id="state">Ready</span></div><div id="error" class="error"></div></div><div class="panel chart"><div class="meta">OHLCV close · UTC</div><canvas id="canvas" width="820" height="330"></canvas><div class="stats"><div class="stat"><span class="meta">Rows</span><b id="rows">—</b></div><div class="stat"><span class="meta">Earliest</span><b id="earliest">—</b></div><div class="stat"><span class="meta">Latest</span><b id="latest">—</b></div></div></div><div class="panel coverage"><h2>Strategy coverage</h2><div id="cards" class="cards"></div></div></section></main><script src="/static/js/apex-data-lab.js"></script></body></html>

help.html

@@ -0,0 +1,25 @@
+<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Space 4 · Real Data Guide</title><style>
+:root{color-scheme:dark;--bg:#070b14;--card:#10192abb;--line:#ffffff12;--text:#eff5ff;--muted:#90a2bd;--good:#42e5c6;--warn:#ffbd5b;--bad:#ff6b7d}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 15% 0,#16334b,transparent 28%),var(--bg);color:var(--text);font:15px/1.65 Inter,system-ui}header,main{width:min(1120px,calc(100% - 32px));margin:auto}header{display:flex;justify-content:space-between;align-items:center;padding:24px 0}a{color:#79ddff;text-decoration:none}nav a{margin-left:18px;color:var(--text)}.hero{padding:48px 0 30px}.kicker{color:var(--good);font-size:12px;letter-spacing:.12em;text-transform:uppercase}h1{font-size:clamp(34px,7vw,66px);line-height:1;margin:18px 0}h2{margin-top:50px}p{color:var(--muted)}.matrix{display:grid;grid-template-columns:repeat(3,1fr);gap:12px}.cell,.endpoint,.warning{background:var(--card);border:1px solid var(--line);border-radius:18px;padding:18px}.cell b{display:block}.tag{font-size:11px;padding:5px 8px;border-radius:999px;background:#42e5c61b;color:var(--good)}.tag.warn{color:var(--warn);background:#ffbd5b1b}.tag.bad{color:var(--bad);background:#ff6b7d1b}.warning{border-color:#ffbd5b55}.endpoint{margin:14px 0}.method{font:700 12px ui-monospace;color:var(--good)}code,pre{font-family:ui-monospace,monospace}pre{overflow:auto;background:#050a12;border:1px solid var(--line);padding:15px;border-radius:13px;color:#cce6ff}summary{cursor:pointer}.route-list{display:grid;grid-template-columns:repeat(2,1fr);gap:8px}.route{background:#0c1422;border:1px solid var(--line);border-radius:11px;padding:10px;font:12px ui-monospace}.muted{color:var(--muted)}@media(max-width:760px){.matrix,.route-list{grid-template-columns:1fr}nav{display:none}}
+</style></head><body><header><strong>Datasource 4 · Truthful API Guide</strong><nav><a href="/">Dashboard</a><a href="/data-lab">Data Lab</a><a href="/docs">OpenAPI</a></nav></header><main><section class="hero"><div class="kicker">Canonical compat router · updated 2026-08-25</div><h1>Market history and LIVE snapshots, clearly separated.</h1><p>The compatibility router is the canonical historical path. The Enterprise Data Hub's duplicate aliases remain unregistered. Short Hunter stays a LIVE-only contract.</p></section><div class="warning"><b>Consumer trap</b><p>HTTP 200 is not proof of data. Check <code>success</code>, <code>dataState</code>, <code>coverage.mode</code>, <code>count</code> and array length.</p></div><h2>Coverage matrix</h2><div class="matrix"><div class="cell"><b>OHLCV</b><span class="tag">REAL HISTORICAL</span><p>Binance Vision → Binance → KuCoin → CryptoCompare, with temporal pagination.</p></div><div class="cell"><b>Funding</b><span class="tag">REAL HISTORICAL</span><p>Gate.io provides real funding with about 180 days of retention in the current HF runtime; Binance/Bybit are geo-blocked there.</p></div><div class="cell"><b>Open interest</b><span class="tag warn">180-DAY + FORWARD</span><p>Gate.io supplies the real ~180-day production window plus a daily SQLite cache; Binance/Bybit are geo-blocked from the current HF runtime.</p></div><div class="cell"><b>Fear & Greed</b><span class="tag">REAL HISTORICAL</span><p>Full Alternative.me series.</p></div><div class="cell"><b>News</b><span class="tag warn">FORWARD ONLY</span><p>Authenticated NewsData latest feed; no pre-deployment history.</p></div><div class="cell"><b>Whale flow</b><span class="tag warn">FORWARD ONLY</span><p>Real large native ETH/BNB transfers from public RPC blocks; no owner labels.</p></div></div><h2>Endpoints and verified behavior</h2>
+<article class="endpoint"><div class="method">GET</div><h3>/api/ohlcv · /api/klines · /api/history</h3><p>Intentional aliases of one implementation. Supports <code>since/start/startTime</code> and <code>end/endTime</code> in milliseconds.</p><pre>curl "$BASE/api/history?symbol=BTCUSDT&amp;interval=4h&amp;start=1640995200000&amp;end=1735689599999&amp;limit=10000"</pre><pre>const body = await fetch(`${base}/api/history?symbol=BTCUSDT&amp;interval=4h&amp;start=${start}&amp;end=${end}&amp;limit=10000`).then(r=&gt;r.json());
+if (!body.success || !body.data.length) throw new Error('history unavailable');</pre><details><summary>Captured response summary</summary><pre>{"success":true,"source":"binance_public","dataState":"REAL","count":6576,"coverage":{"mode":"REAL_HISTORICAL","earliestTimestamp":1640995200000,"latestTimestamp":1735675200000}}</pre></details></article>
+<article class="endpoint"><div class="method">GET / POST</div><h3>/api/apex/funding/{symbol} · /api/apex/open-interest/{symbol} · /api/apex/open-interest-archive/{symbol}</h3><p>Funding is multi-year. OI's live source is limited to 30 days, but startup + daily collection upserts the complete rolling window for <code>APEX_TRACKED_SYMBOLS</code>, deduped by symbol + timestamp. Production history begins at <code>archiveStartedAt</code>; earlier OI cannot be backfilled. <code>/api/apex/coverage</code> distinguishes a mounted <code>/data</code> persistent volume from ephemeral container storage.</p><pre>curl "$BASE/api/apex/funding/BTCUSDT?start=1640995200000&amp;end=1735689599999&amp;limit=20000"
+curl "$BASE/api/apex/open-interest/BTCUSDT?period=1h&amp;limit=500"
+curl "$BASE/api/apex/open-interest-archive/BTCUSDT?limit=10000"
+curl -X POST "$BASE/api/apex/open-interest/archive/collect?period=1h"</pre><pre>const archivedOI = await fetch(`${base}/api/apex/open-interest-archive/BTCUSDT?limit=10000`).then(r=&gt;r.json());</pre><details><summary>Captured response summaries</summary><pre>funding: {"count":3288,"earliestTimestamp":1640995200006,"latestTimestamp":1735660800000}
+openInterest: {"trackedSymbols":5,"rowsPerSymbol":500,"dedupeKey":["symbol","timestamp"],"secondRunDuplicateRows":0}</pre></details></article>
+<article class="endpoint"><div class="method">GET</div><h3>/api/apex/sentiment/fear-greed?limit=0</h3><p>Full dated Alternative.me history.</p><pre>curl "$BASE/api/apex/sentiment/fear-greed?limit=0"</pre><pre>const sentiment = await fetch(`${base}/api/apex/sentiment/fear-greed?limit=0`).then(r=&gt;r.json());</pre><details><summary>Captured response summary</summary><pre>{"count":3123,"source":"alternative.me","coverage":{"mode":"REAL_HISTORICAL","earliestTimestamp":"1517443200","latestTimestamp":"1787529600"}}</pre></details></article>
+<article class="endpoint"><div class="method">GET</div><h3>/api/short-hunter/snapshot/{symbol}</h3><p>LIVE-only multi-capability snapshot. It is not a replay endpoint and accepts no historical timestamp.</p><pre>curl "$BASE/api/short-hunter/snapshot/BTCUSDT?interval=1h&amp;limit=120"</pre><pre>const snap = await fetch(`${base}/api/short-hunter/snapshot/BTCUSDT`).then(r=&gt;r.json());
+if (snap.sourceMode !== 'LIVE' || snap.dataState !== 'REAL') disableTrading();</pre><details><summary>Contract states</summary><pre>LIVE/REAL = every required component succeeded
+DEGRADED/PARTIAL = some components missing
+CACHED/CACHED = cache response
+UNAVAILABLE/UNAVAILABLE = no usable provider result</pre></details></article>
+<article class="endpoint"><div class="method">GET / POST</div><h3>/api/apex/news · /api/apex/whale-flow · /api/apex/archive/collect</h3><p>News uses authenticated NewsData every 15 minutes. Whale flow runs every 2 minutes, scans 15 Ethereum and 200 BSC recent blocks by default, and archives native transfers above configurable thresholds (defaults: 50 ETH / 500 BNB), deduped by chain + transaction hash. It is real transfer flow, not address-owner labeling; both archives begin only after deployment.</p><pre>curl -X POST "$BASE/api/apex/archive/collect"
+curl "$BASE/api/apex/news?limit=500"</pre><pre>const news = await fetch(`${base}/api/apex/news?limit=500`).then(r=&gt;r.json());
+console.assert(news.coverage.mode === 'FORWARD_COLLECTING_ONLY');</pre><details><summary>Captured provider verification</summary><pre>NewsData: HTTP 200, fetched=10, first saved=10, second saved=0
+CryptoCompare candidate: HTTP 401 · NewsAPI.org candidate: HTTP 401
+BitQuery old/new without token: HTTP 401 · ClankApp: HTTP 403
+DIY live test: 12 ETH blocks, 2,556 transactions, 2 matches ≥50 ETH; 12 BSC blocks, 744 transactions, 0 matches ≥500 BNB
+BitQuery signup: https://ide.bitquery.io (time-limited free trial; paid production plans thereafter)
+Whale source: public_evm_rpc_large_native_transfers; forward-only</pre></details></article>
+<h2>All live routes</h2><p>Generated from the deployed OpenAPI contract.</p><div id="routes" class="route-list"><div class="muted">Loading OpenAPI…</div></div><script>fetch('/openapi.json').then(r=>r.json()).then(j=>{routes.innerHTML=Object.entries(j.paths).flatMap(([p,ops])=>Object.entries(ops).filter(([m])=>['get','post','put','delete','patch'].includes(m)).map(([m,o])=>`<div class="route"><b>${m.toUpperCase()}</b> ${p}<br><span class="muted">${o.summary||o.description?.split('\n')[0]||'Registered endpoint'}</span></div>`)).join('')}).catch(e=>routes.textContent='OpenAPI unavailable: '+e.message)</script></main></body></html>

index.html

@@ -914,5 +914,9 @@
             </main>
         </div>
     </div>
+    <nav aria-label="Historical data tools" style="position:fixed;right:18px;bottom:18px;z-index:9999;display:flex;gap:8px;padding:8px;border:1px solid #ffffff18;border-radius:16px;background:#08111ddd;backdrop-filter:blur(16px);box-shadow:0 16px 50px #0008">
+        <a href="/data-lab" style="padding:10px 13px;border-radius:10px;background:linear-gradient(135deg,#45e5d2,#6a8dff);color:#06121a;text-decoration:none;font:700 13px system-ui">Historical Data Lab</a>
+        <a href="/help" style="padding:10px 13px;color:#dce9ff;text-decoration:none;font:600 13px system-ui">Data guide</a>
+    </nav>
 </body>
 </html>

requirements.txt

@@ -3,7 +3,7 @@ fastapi>=0.110,<1.0
 uvicorn[standard]>=0.27,<1.0
 
 # ── HTTP clients ──────────────────────────────────────────────────────────────
-httpx>=0.26,<1.0
+httpx[socks]>=0.26,<1.0
 requests>=2.31,<3.0
 aiohttp>=3.9,<4.0
 
@@ -35,4 +35,4 @@ accelerate>=0.27.0
 
 # ── Optional but IMPORTANT for real inference stability ──────────────────────
 sentencepiece>=0.1.99
-protobuf>=4.25.0
+protobuf>=4.25.0

static/css/apex-data-lab.css

@@ -0,0 +1 @@
+:root{color-scheme:dark;--bg:#070b14;--panel:rgba(18,25,43,.72);--line:rgba(255,255,255,.09);--text:#eef4ff;--muted:#8fa1bc;--cyan:#45e5d2;--violet:#9d7cff;--amber:#ffbd5b;--red:#ff6577}*{box-sizing:border-box}body{margin:0;min-height:100vh;font:15px/1.55 Inter,ui-sans-serif,system-ui;background:radial-gradient(circle at 15% 5%,#132b44 0,transparent 32%),radial-gradient(circle at 90% 10%,#251a4a 0,transparent 28%),var(--bg);color:var(--text)}header,main{width:min(1220px,calc(100% - 32px));margin:auto}header{display:flex;align-items:center;justify-content:space-between;padding:26px 0}.brand{display:flex;gap:12px;align-items:center}.orb{width:38px;height:38px;border-radius:13px;background:linear-gradient(135deg,var(--cyan),var(--violet));box-shadow:0 0 34px #45e5d244}.eyebrow,.meta{color:var(--muted);font-size:12px;letter-spacing:.1em;text-transform:uppercase}h1{font-size:clamp(28px,5vw,54px);line-height:1.02;margin:24px 0 12px;max-width:820px}h2{font-size:18px;margin:0 0 16px}p{color:var(--muted)}nav a{color:var(--text);text-decoration:none;margin-left:20px}.hero{padding:40px 0 28px}.hero p{max-width:720px;font-size:17px}.grid{display:grid;grid-template-columns:repeat(12,1fr);gap:16px}.panel{background:var(--panel);border:1px solid var(--line);border-radius:22px;padding:20px;box-shadow:0 18px 70px #0005;backdrop-filter:blur(18px)}.controls{grid-column:span 4}.chart{grid-column:span 8}.coverage{grid-column:span 12}.form{display:grid;gap:12px}label{font-size:12px;color:var(--muted);text-transform:uppercase;letter-spacing:.08em}input,select,button{width:100%;border:1px solid var(--line);background:#0a1120;color:var(--text);padding:12px 13px;border-radius:12px;font:inherit}button{cursor:pointer;border:0;background:linear-gradient(135deg,var(--cyan),#5ba7ff);color:#04131a;font-weight:800;margin-top:6px}.status{display:flex;gap:8px;align-items:center;margin:14px 0}.dot{width:9px;height:9px;border-radius:50%;background:var(--amber);box-shadow:0 0 16px currentColor}.dot.real{background:var(--cyan)}.dot.bad{background:var(--red)}canvas{width:100%;height:330px;background:linear-gradient(180deg,#0b1425aa,#08101c33);border-radius:14px}.stats{display:grid;grid-template-columns:repeat(3,1fr);gap:10px;margin-top:12px}.stat{background:#09111f;border:1px solid var(--line);border-radius:13px;padding:12px}.stat b{display:block;font-size:18px}.cards{display:grid;grid-template-columns:repeat(5,1fr);gap:10px}.card{border:1px solid var(--line);border-radius:15px;padding:14px;background:#09111d}.badge{display:inline-block;margin-top:8px;padding:5px 8px;border-radius:999px;background:#45e5d21a;color:var(--cyan);font-size:11px}.badge.forward{color:var(--amber);background:#ffbd5b1a}.badge.limited{color:#ff91a0;background:#ff65771a}.error{color:#ff91a0;white-space:pre-wrap}@media(max-width:900px){.controls,.chart{grid-column:span 12}.cards{grid-template-columns:1fr 1fr}.stats{grid-template-columns:1fr}nav{display:none}}@media(max-width:560px){.cards{grid-template-columns:1fr}}

static/js/apex-data-lab.js

@@ -0,0 +1 @@
+const $=id=>document.getElementById(id),fmt=ts=>ts?new Date(Number(ts)).toISOString().slice(0,10):'—';function draw(rows){const c=$('canvas'),x=c.getContext('2d'),w=c.width,h=c.height;x.clearRect(0,0,w,h);if(!rows.length)return;const vals=rows.map(r=>+r.close),lo=Math.min(...vals),hi=Math.max(...vals),span=hi-lo||1;x.strokeStyle='#45e5d2';x.lineWidth=2;x.beginPath();rows.forEach((r,i)=>{const px=i/(rows.length-1||1)*(w-40)+20,py=h-20-(+r.close-lo)/span*(h-40);i?x.lineTo(px,py):x.moveTo(px,py)});x.stroke();x.fillStyle='#8fa1bc';x.font='12px system-ui';x.fillText(hi.toFixed(2),12,16);x.fillText(lo.toFixed(2),12,h-6)}async function load(){const start=Date.parse($('start').value+'T00:00:00Z'),end=Date.parse($('end').value+'T23:59:59Z'),url=`/api/ohlcv?symbol=${encodeURIComponent($('symbol').value)}&interval=${$('interval').value}&start=${start}&end=${end}&limit=100000`;$('state').textContent='Loading upstream pages…';$('dot').className='dot';$('error').textContent='';try{const r=await fetch(url),j=await r.json();if(!r.ok||!j.success)throw Error(j.detail||j.error||'Empty/unavailable response');const data=j.data||j.candles||[];draw(data);$('rows').textContent=data.length.toLocaleString();$('earliest').textContent=fmt(data[0]?.timestamp);$('latest').textContent=fmt(data.at(-1)?.timestamp);$('state').textContent=`${j.dataState||'REAL'} · ${j.source||'provider'} · ${j.coverage?.mode||'unknown'}`;$('dot').className='dot real'}catch(e){$('state').textContent='UNAVAILABLE';$('dot').className='dot bad';$('error').textContent=e.message}}$('load').onclick=load;async function coverage(){const j=await fetch('/api/apex/coverage').then(r=>r.json());$('cards').innerHTML=Object.entries(j.components||{}).map(([k,v])=>`<div class="card"><strong>${k}</strong><p>${v.endpoint||''}</p><span class="badge ${v.mode.includes('FORWARD')?'forward':v.mode.includes('LIMITED')?'limited':''}">${v.mode}</span></div>`).join('')}coverage();load();

tests/test_apex_history.py

@@ -0,0 +1,54 @@
+import pytest
+
+import api_compat_routes as compat
+
+
+@pytest.mark.asyncio
+async def test_compat_binance_range_pages_forward(monkeypatch):
+    calls = []
+
+    async def fake_json(url, params=None, timeout=0):
+        calls.append(dict(params or {}))
+        start = (params or {}).get("startTime", 1000)
+        return [[start + i * 1000, "1", "2", ".5", "1.5", "10"] for i in range((params or {})["limit"])], None, 1
+
+    monkeypatch.setattr(compat, "_get_json", fake_json)
+    rows, errors = await compat._fetch_binance_klines("BTCUSDT", "1h", 2000, 1000, 3_000_000)
+    assert not errors
+    assert len(rows) == 2000
+    assert rows[0]["timestamp"] == 1000
+    assert calls[1]["startTime"] == 1_000_001
+
+
+def test_coverage_labels_separate_historical_oi_from_forward_only_events():
+    from fastapi.testclient import TestClient
+    from fastapi import FastAPI
+    from apex_strategy_routes import router
+
+    app = FastAPI()
+    app.include_router(router)
+    body = TestClient(app).get("/api/apex/coverage").json()
+    assert body["components"]["openInterest"]["mode"] == "REAL_180_DAY_UPSTREAM_PLUS_FORWARD_ARCHIVE"
+    assert body["components"]["news"]["mode"] == "FORWARD_COLLECTING_ONLY"
+    assert body["components"]["whaleFlow"]["mode"] == "FORWARD_COLLECTING_ONLY"
+
+
+def test_open_interest_archive_upserts_by_symbol_timestamp(monkeypatch, tmp_path):
+    import apex_strategy_routes as apex
+    monkeypatch.setattr(apex, "DB_PATH", tmp_path / "apex.db")
+    row = {"symbol": "BTCUSDT", "timestamp": 1234, "sumOpenInterest": 1.0, "sumOpenInterestValue": 2.0}
+    apex._save_oi([row])
+    apex._save_oi([{**row, "sumOpenInterest": 3.0}])
+    result = apex._read_oi("BTCUSDT", None, None, 100)
+    assert result["count"] == 1
+    assert result["data"][0]["sumOpenInterest"] == 3.0
+    assert result["coverage"]["preDeploymentHistoryAvailable"] is False
+
+
+def test_whale_archive_dedupes_by_chain_and_transaction_hash(monkeypatch, tmp_path):
+    import apex_strategy_routes as apex
+    monkeypatch.setattr(apex, "DB_PATH", tmp_path / "apex.db")
+    event = {"txHash": "0xabc", "chain": "ethereum", "symbol": "ETH", "amount": 51, "eventTime": 1234}
+    assert apex._save("whaleFlow", "public_evm_rpc", [event]) == 1
+    assert apex._save("whaleFlow", "public_evm_rpc", [event]) == 0
+    assert apex._archive_count("whaleFlow") == 1