Gold Lab
金价 × 实际利率 × 央行购金 · 研究终端
访问码由部署者设置（GOLDLAB_ACCESS_CODE）。
验证通过后本机 30 天内免重复输入。
"""
# ---- 伦敦金现（XAU/USD 现货，v3）：内存缓存 + 休市降频 ----
SPOT_URL = "https://api.gold-api.com/price/XAU"
_spot = {"ts": 0.0, "data": None} # {price, updatedAt, fetchedAt}
def _market_open() -> bool:
"""现货金交易时段（北京时间，新浪交易日口径的近似）：周一 06:00 开盘 → 周六 04:59 收盘，
每日 05:00-05:59 结算休息；周日全天休市。
冬令时实际收盘后移约 1h——此处宁可「开市多轮询」也不「休市漏采」，故只卡明确休市窗口。"""
t = time.gmtime(time.time() + 8 * 3600) # 北京时间
wd, hm = t.tm_wday, t.tm_hour * 60 + t.tm_min
if wd == 6: # 周日
return False
if wd == 5 and hm >= 300: # 周六 05:00 起
return False
if wd == 0 and hm < 360: # 周一 06:00 前
return False
if 300 <= hm < 360: # 每日 05:00-06:00 结算休息
return False
return True
def _spot_ttl() -> int:
"""现货/分时/报价缓存 TTL 与守护线程节拍：开市 55-60s，休市放宽到 10 分钟（数据就是静止的）。"""
return 60 if _market_open() else 600
def _fetch_spot() -> dict:
"""TTL（现货时段 60s/休市 10m）缓存 → 最新抓取；失败回退最近缓存并标 stale
（三级降级的前两级；第三级=前端对 price null 隐藏行情位、只留 LBMA PM）。报价头由调用方叠加。"""
with _lock:
now = time.time()
if _spot["data"] and now - _spot["ts"] < _spot_ttl():
return {**_spot["data"], "cached": True}
try:
req = urllib.request.Request(SPOT_URL, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=10) as r:
d = json.loads(r.read().decode("utf-8"))
data = {"price": float(d["price"]), "updatedAt": d.get("updatedAt", ""),
"fetchedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
_spot.update(ts=now, data=data)
return {**data, "cached": False}
except Exception: # noqa: BLE001
if _spot["data"]:
return {**_spot["data"], "cached": True, "stale": True}
return {"price": None, "error": "spot unavailable"}
_TRENDS = {"ts": 0.0, "data": None}
def _fetch_trends() -> dict:
"""新浪 1 分钟分时（伦敦金现 XAU）→ 固定交易日轴的当日分时。
交易日（新浪口径）：北京时间 06:00 开盘 → 次日 04:59 收市结算（05:00-05:59 休息）。
新浪接口返回最近 ~1023 分钟的滑动窗口，可能混入上一交易日的尾巴——此处按交易日
锚点过滤，只保留当前交易日的分钟。返回：
{times:[HH:mm...], prices:[...], avg:[...], # 当日实际有数据的分钟
axisStart:"YYYY-MM-DD 06:00", slots:1380} # 前端固定全日轴（06:00→次日04:59）
失败回退旧缓存，再退空。"""
with _lock:
now = time.time()
if _TRENDS["data"] and now - _TRENDS["ts"] < _spot_ttl():
return _TRENDS["data"]
url = ("https://gu.sina.cn/ft/api/jsonp.php/var%20t=/GlobalService.getMinK"
"?symbol=XAU&type=1")
try:
req = urllib.request.Request(url, headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/128.0",
"Referer": "https://finance.sina.com.cn"})
with urllib.request.urlopen(req, timeout=20) as r:
txt = r.read().decode("gbk", "ignore")
m = re.search(r"var t=\((\[.*\])\)", txt, re.S)
if not m:
return _TRENDS["data"] or {"times": [], "prices": []}
rows = json.loads(m.group(1))
pts = []
for r0 in rows:
ds = r0.get("d") or ""
if len(ds) >= 16 and r0.get("c"):
pts.append((ds[:10], ds[11:16], float(r0["c"]))) # (日期, HH:MM, 收)
if not pts:
return _TRENDS["data"] or {"times": [], "prices": []}
# 交易日锚点：最新分钟 >= 06:00 属当日开盘的交易日；否则（00:00-04:59）属前一自然日
last_d, last_t, _ = pts[-1]
if last_t >= "06:00":
start_date = last_d
else:
start_date = (datetime.strptime(last_d, "%Y-%m-%d") - timedelta(days=1)).strftime("%Y-%m-%d")
next_date = (datetime.strptime(start_date, "%Y-%m-%d") + timedelta(days=1)).strftime("%Y-%m-%d")
day_pts = [p for p in pts
if (p[0] == start_date and p[1] >= "06:00") or (p[0] == next_date and p[1] < "05:00")]
if not day_pts:
return _TRENDS["data"] or {"times": [], "prices": []}
times = [p[1] for p in day_pts]
prices = [round(p[2], 2) for p in day_pts]
# 分时均价（滚动 30 分钟近似，从交易日开盘起算）
avg = []
WIN = 30
s3 = 0.0
for k in range(len(prices)):
s3 += prices[k]
if k >= WIN:
s3 -= prices[k - WIN]
avg.append(round(s3 / min(k + 1, WIN), 2))
data = {"times": times, "prices": prices, "avg": avg,
"axisStart": start_date + " 06:00", "slots": 1380}
with _lock:
_TRENDS.update(ts=now, data=data)
return data
except Exception: # noqa: BLE001
return _TRENDS["data"] or {"times": [], "prices": []}
_SINA = {"ts": 0.0, "data": None}
def _fetch_sina_quote() -> dict:
"""新浪 hf_XAU 实时报价（当前/昨收/今开/最高/最低 + 交易日 + 北京时间），55s 内存缓存。
报价头的今开/高低/昨收必须来自「当前交易日」的盘中值——此前取自现货日K文件
（只在管线运行时刷新），新交易日开盘后报价头会整页停在上一交易日（v4.6 修复）。"""
with _lock:
now = time.time()
if _SINA["data"] and now - _SINA["ts"] < _spot_ttl():
return _SINA["data"]
try:
req = urllib.request.Request("https://hq.sinajs.cn/list=hf_XAU,DINIW",
headers={"User-Agent": "Mozilla/5.0",
"Referer": "https://finance.sina.com.cn"})
with urllib.request.urlopen(req, timeout=10) as r:
txt = r.read().decode("gbk", "ignore")
m = re.search(r'hq_str_hf_XAU="([^"]+)"', txt)
if m:
p = m.group(1).split(",")
if len(p) >= 13:
# hf_ 实测格式：0现价 1昨收 2买价 3卖价 4最高 5最低 6时间 7昨收 8今开 9-11量 12日期 13名称
# （v4.1 起管线曾把 2/3/4 当作 开/高/低 → 高低互换、开盘=买价，v4.6 修正）
data = {"price": float(p[0]), "prevClose": float(p[1]), "open": float(p[8]),
"high": float(p[4]), "low": float(p[5]), "time": p[6], "date": p[12]}
if data["prevClose"] > 0 and data["open"] > 0 and data["high"] >= data["low"]:
data["amplitude"] = round((data["high"] - data["low"]) / data["prevClose"] * 100, 2)
data["chg"] = round(data["price"] - data["prevClose"], 2)
data["chgPct"] = round((data["price"] / data["prevClose"] - 1) * 100, 2)
# DINIW 美元指数（同一次请求捎带，数据页 DXY 实时读数）：
# 0时间 1现价 3今开 5低 6高 7昨收 9名称 10日期
md = re.search(r'hq_str_DINIW="([^"]+)"', txt)
if md:
q = md.group(1).split(",")
try:
if len(q) >= 11 and float(q[1] or 0) > 0:
data["dxy"] = {"price": float(q[1]), "open": float(q[3] or 0),
"low": float(q[5] or 0), "high": float(q[6] or 0),
"prevClose": float(q[7] or 0), "time": q[0], "date": q[10]}
except (ValueError, IndexError):
pass
with _lock:
_SINA.update(ts=now, data=data)
try: # 实时值回写文件，供管线间隔内的兜底读取（与管线同格式同字段）
tmp = DATA / "spot-sina.json.tmp"
tmp.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
tmp.replace(DATA / "spot-sina.json")
except Exception: # noqa: BLE001
pass
return data
except Exception: # noqa: BLE001
pass
return dict(_SINA["data"] or {})
def _quote_snapshot() -> dict:
"""报价头快照（v4.6 实时优先）：
1) 新浪 hf_XAU 实时盘中（今开/最高/最低/昨收 = 当前交易日，refresher 55s 刷新）；
2) 文件兜底：spot-sina.json（管线落盘，管线运行间隔内可能滞后一个交易日）；
3) 最后兜底：现货日K最后一根（与 K 线同源）。
现价主字段由 gold-api 单独提供，此快照只补 OHLC/昨收/数据日期。"""
live = _SINA.get("data") or {}
if live.get("prevClose"):
return {"open": live["open"], "high": live["high"], "low": live["low"],
"close": live["price"], "date": live["date"],
"prevClose": live["prevClose"],
"amplitude": round((live["high"] - live["low"]) / live["prevClose"] * 100, 2) if live["prevClose"] else None,
"chg": round(live["price"] - live["prevClose"], 2),
"chgPct": round((live["price"] / live["prevClose"] - 1) * 100, 2),
"src": "新浪财经 · 伦敦金现盘中"}
try:
s = json.loads((DATA / "spot-sina.json").read_text(encoding="utf-8"))
if s.get("prevClose"):
return {"open": s["open"], "high": s["high"], "low": s["low"], "close": s["price"],
"date": s.get("date"), "prevClose": s["prevClose"],
"amplitude": round((s["high"] - s["low"]) / s["prevClose"] * 100, 2) if s["prevClose"] else None,
"chg": round(s["price"] - s["prevClose"], 2),
"chgPct": round((s["price"] / s["prevClose"] - 1) * 100, 2),
"src": "新浪财经 · 伦敦金现（缓存）"}
except Exception: # noqa: BLE001
pass
try:
c = json.loads((DATA / "spot-candles.json").read_text(encoding="utf-8"))
last = c[-1]
prev = c[-2] if len(c) >= 2 else None
prev_close = (prev["c"] if prev else last["c"])
return {"open": last["o"], "high": last["h"], "low": last["l"], "close": last["c"],
"date": last["d"],
"prevClose": prev_close,
"amplitude": round((last["h"] - last["l"]) / prev_close * 100, 2) if prev_close else None,
"chg": round(last["c"] - prev_close, 2),
"chgPct": round((last["c"] / prev_close - 1) * 100, 2),
"src": "新浪财经 · 现货日K（管线刷新）"}
except Exception: # noqa: BLE001
return {}
def _spot_refresher() -> None:
"""后台守护线程：每 55s 主动刷新现货 + 当日分时（把外网延迟移出请求路径，请求永远读缓存，毫秒级）。"""
while True:
try:
_fetch_spot()
except Exception: # noqa: BLE001
pass
try:
_fetch_trends()
except Exception: # noqa: BLE001
pass
try:
_fetch_sina_quote() # v4.6：报价头实时盘中 OHLC/昨收（当前交易日）
except Exception: # noqa: BLE001
pass
time.sleep(55 if _market_open() else 600) # 休市时钟面静止：节拍放宽到 10 分钟（日志见 _spot_ttl 同步口径）
def _auto_update_loop() -> None:
"""自动更新守护线程（GOLDLAB_AUTO_UPDATE_HOURS，默认 12h，0=关闭）：
周期触发与「更新数据」按钮完全相同的全管线（复用互斥锁，手动更新进行中则本轮跳过）。"""
if AUTO_UPDATE_HOURS <= 0:
return
while True:
with _lock:
_auto["next"] = time.time() + AUTO_UPDATE_HOURS * 3600
time.sleep(AUTO_UPDATE_HOURS * 3600)
try:
if _start_update():
with _lock:
_state["logs"].append(f"[自动] 定时更新已触发（每 {AUTO_UPDATE_HOURS:g} 小时）")
except Exception: # noqa: BLE001
pass
def _start_update(step: str | None = None) -> bool:
with _lock:
if _state["running"]:
return False
_state.update(running=True, logs=[], done=False, error=None)
def progress(msg):
with _lock:
_state["logs"].append(msg)
def worker():
try:
# 每次更新前重载管线模块：pipeline.py 改动后无需重启服务即生效
# （2026-08-31 加固：旧进程曾用旧代码把新格式数据覆盖回旧格式）
importlib.reload(pipeline)
if step:
pipeline.run_one(step, progress) # 单源强制更新（数据目录 ↻）
else:
pipeline.run_all(progress)
except Exception as e: # noqa: BLE001
with _lock:
_state["error"] = str(e)
_state["logs"].append(f"[错误] {e}")
finally:
# 现货/分时/报价缓存作废并原地预热：更新完成的下一次 /api/spot 立即返回全新盘中值
# （不作废的话，报价头最长要等一个 TTL 才能摆脱更新前的读数）
with _lock:
_spot["ts"] = _TRENDS["ts"] = _SINA["ts"] = 0.0
for fn in (_fetch_spot, _fetch_trends, _fetch_sina_quote):
try:
fn()
except Exception: # noqa: BLE001
pass
with _lock:
_state["running"] = False
_state["done"] = True
threading.Thread(target=worker, daemon=True).start()
return True
class Handler(BaseHTTPRequestHandler):
def log_message(self, *a): # 静默访问日志
pass
def _session_ok(self) -> bool:
if not ACCESS_CODE:
return True
for part in (self.headers.get("Cookie") or "").split(";"):
k, _, v = part.strip().partition("=")
if k == COOKIE_NAME and _check_session(v):
return True
return False
def _gate(self) -> bool:
"""整站访问门（v4.4）：未设访问码=本地模式直接放行；未通过时页面回登录页、API 回 401。"""
if self._session_ok():
return True
if urlparse(self.path).path.startswith("/api/"):
self._json({"ok": False, "message": "未验证访问码"}, 401)
else:
self._send(200, LOGIN_HTML.encode("utf-8"), "text/html; charset=utf-8")
return False
def _send(self, code: int, body: bytes, ctype: str, cache: str = "no-store"):
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", cache)
self.end_headers()
self.wfile.write(body)
def _json(self, obj, code=200):
self._send(code, json.dumps(obj, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8")
def do_GET(self): # noqa: N802
path = urlparse(self.path).path
if path == "/api/login":
self._json({"ok": False, "message": "请用 POST 提交访问码"}, 405)
return
if not self._gate():
return
if path == "/" or path == "/index.html":
self._send(200, (UI / "index.html").read_bytes(), MIME[".html"])
elif path == "/api/data":
fp = DATA / "lab-data.json"
if not fp.exists():
pipeline.prepare()
self._send(200, fp.read_bytes(), MIME[".json"])
elif path == "/api/spot":
out = _fetch_spot()
out["quote"] = _quote_snapshot()
out["trends"] = _fetch_trends()
sina = _fetch_sina_quote() or {}
if isinstance(sina, dict) and sina.get("dxy"):
out["dxy"] = sina["dxy"] # 数据页 DXY 实时读数（与 hf_XAU 同一次请求，无额外成本）
with _lock:
nxt = _auto.get("next")
out["auto"] = {"everyHours": AUTO_UPDATE_HOURS,
"next": datetime.fromtimestamp(nxt).strftime("%m-%d %H:%M") if nxt else None}
self._json(out, 200 if out.get("price") is not None else 503)
elif path == "/api/candles":
# v4.1：伦敦金现现货日K（新浪 2006 起）优先；缺则回退 LBMA 定盘 OHLC
fp = DATA / "spot-candles.json"
src = "sina"
if not fp.exists():
fp = DATA / "candles.json"
src = "lbma"
self._send(200, fp.read_bytes(), MIME[".json"])
elif path == "/api/strategy-presets":
try:
self._json(strategy.history_presets())
except Exception as exc: # noqa: BLE001
self._json({"ok": False, "error": f"{type(exc).__name__}: {exc}"}, 500)
elif path == "/api/strategies":
# D18/D20：三轨策略跟踪（缓存重放，数据更新后自动重算）
try:
self._json(strategy.payload())
except Exception as exc: # noqa: BLE001
self._json({"ok": False, "error": f"{type(exc).__name__}: {exc}"}, 500)
elif path == "/api/update/status":
with _lock:
self._json({k: v for k, v in _state.items()})
else:
fp = (UI / path.lstrip("/")).resolve()
if not str(fp).startswith(str(UI.resolve())) or not fp.is_file():
self._send(404, b"not found", "text/plain")
return
ext = fp.suffix
if ext not in MIME:
self._send(404, b"not found", "text/plain")
return
# vendor 大文件（ECharts ~1MB）内容不变，可长缓存；UI 源码保持 no-store 便于热更
cache = "public, max-age=604800" if "/vendor/" in path else "no-store"
self._send(200, fp.read_bytes(), MIME[ext], cache)
def do_POST(self): # noqa: N802
path = urlparse(self.path).path
if path == "/api/login":
length = int(self.headers.get("Content-Length") or 0)
try:
body = json.loads(self.rfile.read(length).decode("utf-8")) if length else {}
except Exception: # noqa: BLE001
body = {}
code = str(body.get("code", "")) if isinstance(body, dict) else ""
if not ACCESS_CODE:
self._json({"ok": False, "message": "未启用访问控制（未设置 GOLDLAB_ACCESS_CODE）"})
elif hmac.compare_digest(code, ACCESS_CODE):
tok = _make_session()
body_b = json.dumps({"ok": True}).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Set-Cookie",
f"{COOKIE_NAME}={tok}; Path=/; HttpOnly; Max-Age={SESSION_TTL}; SameSite=Lax")
self.send_header("Content-Length", str(len(body_b)))
self.end_headers()
self.wfile.write(body_b)
else:
time.sleep(0.6) # 轻微防爆破
self._json({"ok": False, "message": "访问码不正确"}, 401)
return
if not self._gate():
return
if path == "/api/strategy-history":
length = int(self.headers.get("Content-Length") or 0)
try:
body = json.loads(self.rfile.read(length).decode("utf-8")) if length else {}
self._json(strategy.history(body))
except Exception as exc: # noqa: BLE001
self._json({"ok": False, "error": f"{type(exc).__name__}: {exc}"}, 500)
elif path == "/api/update":
# 访问码模式下会话即凭据；仅本地无访问码模式才走脚本 token（向后兼容）
token = os.environ.get("GOLDLAB_UPDATE_TOKEN", "")
if not ACCESS_CODE and token and self.headers.get("X-Update-Token") != token:
self._json({"ok": False, "message": "需要有效的 X-Update-Token"}, 401)
return
if _start_update():
self._json({"ok": True, "message": "更新已启动"})
else:
self._json({"ok": False, "message": "已有更新在进行中"}, 409)
elif path == "/api/update/one":
token = os.environ.get("GOLDLAB_UPDATE_TOKEN", "")
if not ACCESS_CODE and token and self.headers.get("X-Update-Token") != token:
self._json({"ok": False, "message": "需要有效的 X-Update-Token"}, 401)
return
qs = parse_qs(urlparse(self.path).query)
step = (qs.get("step") or [""])[0]
if step not in {s[0] for s in pipeline.STEPS}:
self._json({"ok": False, "message": f"未知数据源 step: {step}"}, 400)
return
if _start_update(step):
self._json({"ok": True, "message": "单源更新已启动"})
else:
self._json({"ok": False, "message": "已有更新在进行中"}, 409)
else:
self._send(404, b"not found", "text/plain")
def do_HEAD(self): # noqa: N802 — 反向代理/监控的健康检查
path = urlparse(self.path).path
self.send_response(200 if path in ("/", "/index.html", "/api/data", "/api/spot") else 404)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", "0")
self.end_headers()
def serve(port: int | None = None, host: str | None = None, open_browser: bool = True) -> None:
# 环境变量优先（云端），其次显式参数，最后本地默认
host = os.environ.get("HOST", host or "127.0.0.1")
port = int(os.environ.get("PORT", port if port else 8765))
if not (DATA / "lab-data.json").exists():
print("首次运行：准备数据 ...")
pipeline.prepare()
# 伦敦金现后台刷新线程（v4.1：请求路径零外网等待）
threading.Thread(target=_spot_refresher, daemon=True).start()
# 自动更新线程（v4.4：默认每 12h 全管线，GOLDLAB_AUTO_UPDATE_HOURS 可调/0 关闭）
if AUTO_UPDATE_HOURS > 0:
threading.Thread(target=_auto_update_loop, daemon=True).start()
print(f"自动更新：每 {AUTO_UPDATE_HOURS:g} 小时")
if ACCESS_CODE:
print("访问控制：已启用（GOLDLAB_ACCESS_CODE）")
httpd = ThreadingHTTPServer((host, port), Handler)
url = f"http://{host}:{port}"
print(f"Gold Lab 已启动: {url} (Ctrl+C 退出)")
if open_browser:
threading.Timer(0.6, lambda: webbrowser.open(url)).start()
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\n已退出")
if __name__ == "__main__":
serve()