Web dashboard v2: static publisher + Apache, with calendar & news panels
- dashboard_publisher.py: standalone candle-close daemon that atomically writes snapshot.json/news.json/dashboard.html to an Apache-served dir (DASHBOARD_OUT_DIR). Fully independent of telegram_bot/oanda_daemon — local confirmation map + boundary math, no imports either way, no new deps, no open port. Due-TF classification from the RAW wake time. - static/dashboard.html: self-contained dark page — hero rail, four TF cards (score bar, 3x3 indicator grid, MTF, setup, reasons), DeepSeek market context, week economic calendar, scored newswire. Clock offset from the HTTP Date header (publish-time epoch is stale between closes). - news_engine.get_week_events(): additive read-only accessor over the cached FF week; no signal path calls it. - deploy/: systemd unit + live-server (LAMP/SELinux) checklist. - test_all.py: Module 20 (9 tests) + get_week_events test; 296 green. Claude-Session: https://claude.ai/code/session_01Wc9ZoqKLYBT2QJcpRWKoGm
This commit is contained in:
@@ -0,0 +1,420 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
XAU/USD static dashboard publisher.
|
||||
|
||||
Standalone daemon that scans XAU/USD at candle closes and atomically
|
||||
writes `snapshot.json` + `news.json` (plus a copy of
|
||||
`static/dashboard.html`) into an Apache-served directory. Apache does
|
||||
all web serving — this process never opens a port, so a failure here
|
||||
can only ever mean a stale page.
|
||||
|
||||
Independence rules (owner requirement — the telegram bot must be
|
||||
immune to anything on the web side):
|
||||
* Runs as its own process; never imported by, and never imports,
|
||||
`telegram_bot` / `oanda_daemon` (telegram_bot resolves credentials
|
||||
at import). The confirmation map and boundary math are therefore
|
||||
duplicated here as local constants.
|
||||
* The only shared-code dependency beyond the standard signal path is
|
||||
the additive `NewsEngine.get_week_events` accessor, which no
|
||||
production signal path calls.
|
||||
|
||||
Scheduling: every M15/H1/H4 boundary is also an M5 boundary (all
|
||||
intervals are multiples of 5, anchored to UTC midnight), so one
|
||||
sequential loop sleeps to the next M5 boundary (+10 s buffer) and scans
|
||||
whichever timeframes just closed. Due-classification uses the RAW wake
|
||||
time — flooring 12:00:10 to the 5-minute grid yields the 12:00 boundary
|
||||
that just closed. Do NOT subtract the buffer first: 12:00:10 − 15 s
|
||||
floors to 11:55 and silently skips every H1/H4 scan (v1 bug, 2026-08-02).
|
||||
|
||||
Scans run sequentially M5→M15→H1→H4 through one shared data provider so
|
||||
later timeframes hit the warm boundary cache — that ordering IS the
|
||||
duplicate-fetch dedup (providers have no in-flight dedup; Oanda has no
|
||||
throttle). Keep it sequential.
|
||||
|
||||
Run: python dashboard_publisher.py [--once] [--out DIR]
|
||||
Out dir precedence: --out → DASHBOARD_OUT_DIR env → /var/www/html/gold-dashboard
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import math
|
||||
import numbers
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from logging.handlers import TimedRotatingFileHandler
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from config import PlatformConfig, Timeframe, XAUUSD
|
||||
from main import report_to_dict
|
||||
from news_engine import NewsEngine
|
||||
from pair_signals import produce_signal
|
||||
from providers import make_data_provider
|
||||
import providers.base as _providers_base
|
||||
|
||||
logger = logging.getLogger("dashboard")
|
||||
|
||||
DASHBOARD_TFS = [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4]
|
||||
|
||||
# Local copy of the bot's confirmation map (see independence rules above).
|
||||
CONFIRMATIONS = {
|
||||
Timeframe.M5: [Timeframe.M15, Timeframe.H1],
|
||||
Timeframe.M15: [Timeframe.M5, Timeframe.H1],
|
||||
Timeframe.H1: [Timeframe.M15, Timeframe.H4],
|
||||
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
|
||||
}
|
||||
|
||||
TF_INTERVAL_MINUTES = {
|
||||
Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.H1: 60, Timeframe.H4: 240,
|
||||
}
|
||||
|
||||
DEFAULT_OUT_DIR = "/var/www/html/gold-dashboard"
|
||||
SCAN_BUFFER_SECONDS = 10
|
||||
SNAPSHOT_FILE = "snapshot.json"
|
||||
NEWS_FILE = "news.json"
|
||||
PAGE_FILE = "dashboard.html"
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════
|
||||
# Pure layer (no network, fully covered by test_all.py)
|
||||
# ══════════════════════════════════════════════════════════════════
|
||||
|
||||
def _safe_num(v):
|
||||
"""Coerce to a JSON-safe float; NaN/±inf become None."""
|
||||
try:
|
||||
f = float(v)
|
||||
except (TypeError, ValueError):
|
||||
return v
|
||||
if math.isnan(f) or math.isinf(f):
|
||||
return None
|
||||
return f
|
||||
|
||||
|
||||
def _sanitize(obj):
|
||||
"""Recursively make a payload safe for json.dumps(allow_nan=False):
|
||||
NaN/inf → None, numpy scalars → native ints/floats."""
|
||||
if isinstance(obj, bool):
|
||||
return obj
|
||||
if isinstance(obj, dict):
|
||||
return {k: _sanitize(v) for k, v in obj.items()}
|
||||
if isinstance(obj, (list, tuple)):
|
||||
return [_sanitize(v) for v in obj]
|
||||
if isinstance(obj, numbers.Integral):
|
||||
return int(obj)
|
||||
if isinstance(obj, numbers.Real):
|
||||
return _safe_num(obj)
|
||||
return obj
|
||||
|
||||
|
||||
def analysis_to_dict(analysis) -> Optional[dict]:
|
||||
"""Serialize a TechnicalAnalysis. Iterates analysis.indicators only —
|
||||
never vars()/asdict(), which would drag dynamic pandas attrs
|
||||
(_rsi_series, _macd_histogram, …) into the payload. ATR (weight 0)
|
||||
is surfaced separately as context, not in the directional grid."""
|
||||
if analysis is None:
|
||||
return None
|
||||
indicators = {}
|
||||
atr_context = None
|
||||
for name, ind in analysis.indicators.items():
|
||||
if name.startswith("_"):
|
||||
continue
|
||||
entry = {
|
||||
"signal": ind.signal.name,
|
||||
"value": _safe_num(ind.value),
|
||||
"confidence": _safe_num(ind.confidence),
|
||||
"weight": _safe_num(ind.weight),
|
||||
"description": ind.description,
|
||||
}
|
||||
if name == "atr":
|
||||
atr_context = entry
|
||||
elif ind.weight != 0:
|
||||
indicators[name] = entry
|
||||
return {
|
||||
"timeframe": analysis.timeframe,
|
||||
"timestamp": str(analysis.timestamp),
|
||||
"composite_signal": analysis.composite_signal.name,
|
||||
"composite_score": _safe_num(analysis.composite_score),
|
||||
"confidence": _safe_num(analysis.confidence),
|
||||
"market_regime": analysis.market_regime,
|
||||
"volatility_regime": analysis.volatility_regime,
|
||||
"pattern_signals": list(analysis.pattern_signals),
|
||||
"key_levels": {k: _safe_num(v) for k, v in analysis.key_levels.items()},
|
||||
"indicators": indicators,
|
||||
"atr_context": atr_context,
|
||||
}
|
||||
|
||||
|
||||
def report_to_webapp_dict(report) -> dict:
|
||||
"""main.report_to_dict + the per-TF detail the dashboard needs."""
|
||||
d = report_to_dict(report)
|
||||
d["primary_signal"] = report.primary_signal.name
|
||||
d["primary_score"] = _safe_num(report.primary_score)
|
||||
d["mtf_scores"] = {k: _safe_num(v) for k, v in report.mtf_scores.items()}
|
||||
d["exhaustion_count"] = report.exhaustion_count
|
||||
d["primary_analysis"] = analysis_to_dict(report.primary_analysis)
|
||||
d["confirmation_analyses"] = [
|
||||
analysis_to_dict(a) for a in report.confirmation_analyses
|
||||
]
|
||||
return _sanitize(d)
|
||||
|
||||
|
||||
def event_to_dict(ev) -> dict:
|
||||
"""EconomicEvent → dict. surprise/minutes_until/is_high_impact are
|
||||
properties, so asdict() would skip them — added explicitly.
|
||||
actual/forecast/previous stay strings as published ('228K', '0.3%')."""
|
||||
return {
|
||||
"time": ev.time.astimezone(timezone.utc).isoformat(),
|
||||
"currency": ev.currency,
|
||||
"impact": ev.impact,
|
||||
"event": ev.event,
|
||||
"forecast": ev.forecast,
|
||||
"previous": ev.previous,
|
||||
"actual": ev.actual,
|
||||
"surprise": ev.surprise,
|
||||
"minutes_until": _safe_num(ev.minutes_until),
|
||||
"is_high_impact": ev.is_high_impact,
|
||||
}
|
||||
|
||||
|
||||
def headline_to_dict(h) -> dict:
|
||||
return {
|
||||
"headline": h.headline,
|
||||
"source": h.source,
|
||||
"timestamp": h.timestamp.astimezone(timezone.utc).isoformat()
|
||||
if h.timestamp else None,
|
||||
"currencies": list(h.currencies_mentioned),
|
||||
"score": _safe_num(h.sentiment_score),
|
||||
"keywords": list(h.keywords),
|
||||
"score_by_currency": {
|
||||
k: _safe_num(v) for k, v in h.sentiment_by_currency.items()
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def build_news_payload(context, week_events, now: datetime) -> dict:
|
||||
"""news.json: full calendar week + scored headlines + LLM summary."""
|
||||
headlines = sorted(
|
||||
context.news_headlines,
|
||||
key=lambda h: h.timestamp or now, reverse=True,
|
||||
)
|
||||
return _sanitize({
|
||||
"generated_at": now.isoformat(),
|
||||
"llm_summary": context.llm_summary,
|
||||
"sentiment": dict(context.news_sentiment),
|
||||
"high_impact_warning": context.high_impact_warning,
|
||||
"warning_message": context.warning_message,
|
||||
"calendar_available": context.calendar_available,
|
||||
"events": [event_to_dict(e) for e in week_events],
|
||||
"headlines": [headline_to_dict(h) for h in headlines],
|
||||
})
|
||||
|
||||
|
||||
def due_timeframes(wake_dt: datetime) -> List[Timeframe]:
|
||||
"""Timeframes whose candle just closed at the boundary this wake
|
||||
belongs to. Takes the RAW wake time (boundary + buffer): flooring to
|
||||
the 5-minute grid yields the boundary that just passed. Never
|
||||
subtract the buffer first — see module docstring (v1 bug)."""
|
||||
minutes = wake_dt.hour * 60 + wake_dt.minute
|
||||
floored = minutes - (minutes % 5)
|
||||
return [tf for tf in DASHBOARD_TFS if floored % TF_INTERVAL_MINUTES[tf] == 0]
|
||||
|
||||
|
||||
def seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
|
||||
# Same math as telegram_bot.py / oanda_daemon.py — kept local so the
|
||||
# dashboard never imports bot modules (independence rules above).
|
||||
now = datetime.now(timezone.utc)
|
||||
total_minutes = now.hour * 60 + now.minute
|
||||
elapsed_minutes = total_minutes % interval_minutes
|
||||
elapsed_seconds = elapsed_minutes * 60 + now.second + now.microsecond / 1e6
|
||||
interval_seconds = interval_minutes * 60
|
||||
remaining = interval_seconds - elapsed_seconds + buffer
|
||||
if remaining <= buffer:
|
||||
remaining += interval_seconds
|
||||
return remaining
|
||||
|
||||
|
||||
def _empty_tf_entry() -> dict:
|
||||
return {"status": "warming", "scanned_at": None, "scan_seconds": None,
|
||||
"error": None, "error_at": None, "report": None}
|
||||
|
||||
|
||||
def build_snapshot_payload(snapshots: Dict[str, dict], now: datetime) -> dict:
|
||||
"""snapshot.json: one entry per dashboard TF. Timeframes not yet
|
||||
scanned appear as status="warming" so the page can render a
|
||||
placeholder instead of a hole. server_time_epoch lets the client
|
||||
compute its clock offset once so countdowns don't drift."""
|
||||
timeframes = {}
|
||||
for tf in DASHBOARD_TFS:
|
||||
interval = TF_INTERVAL_MINUTES[tf]
|
||||
minutes = now.hour * 60 + now.minute
|
||||
secs_into = (minutes % interval) * 60 + now.second + now.microsecond / 1e6
|
||||
next_close = now.timestamp() + (interval * 60 - secs_into)
|
||||
entry = _empty_tf_entry()
|
||||
entry.update(snapshots.get(tf.value) or {})
|
||||
entry["next_close_epoch"] = next_close
|
||||
entry["next_scan_epoch"] = next_close + SCAN_BUFFER_SECONDS
|
||||
timeframes[tf.value] = entry
|
||||
return _sanitize({
|
||||
"symbol": XAUUSD,
|
||||
"server_time_epoch": now.timestamp(),
|
||||
"market_closed": _providers_base._is_forex_weekend(now),
|
||||
"timeframes": timeframes,
|
||||
})
|
||||
|
||||
|
||||
def publish(obj: dict, path: str) -> None:
|
||||
"""Atomic JSON write: temp file + os.replace, so Apache never serves
|
||||
a torn file. allow_nan=False — a NaN slipping through would render
|
||||
the page blank (JSON.parse rejects bare NaN)."""
|
||||
data = json.dumps(obj, allow_nan=False, separators=(",", ":"))
|
||||
tmp = path + ".tmp"
|
||||
with open(tmp, "w", encoding="utf-8") as fh:
|
||||
fh.write(data)
|
||||
os.replace(tmp, path)
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════
|
||||
# Runtime layer (constructed only from main(); never at import)
|
||||
# ══════════════════════════════════════════════════════════════════
|
||||
|
||||
def _setup_logging(level: str = "INFO") -> None:
|
||||
log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs")
|
||||
os.makedirs(log_dir, exist_ok=True)
|
||||
log_file = os.path.join(log_dir, "dashboard.log")
|
||||
|
||||
fmt = logging.Formatter(
|
||||
"%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||||
datefmt="%Y-%m-%d %H:%M:%S")
|
||||
|
||||
file_handler = TimedRotatingFileHandler(
|
||||
log_file, when="h", interval=3, backupCount=240, utc=True)
|
||||
file_handler.setFormatter(fmt)
|
||||
|
||||
console_handler = logging.StreamHandler(sys.stdout)
|
||||
console_handler.setFormatter(fmt)
|
||||
|
||||
logging.basicConfig(
|
||||
level=getattr(logging, level.upper(), logging.INFO),
|
||||
handlers=[file_handler, console_handler],
|
||||
force=True)
|
||||
|
||||
|
||||
class DashboardPublisher:
|
||||
def __init__(self, out_dir: str):
|
||||
self.out_dir = out_dir
|
||||
self.snapshots: Dict[str, dict] = {}
|
||||
self.news_payload: Optional[dict] = None
|
||||
self._dp = None
|
||||
self._news_engine: Optional[NewsEngine] = None
|
||||
|
||||
def _provider(self):
|
||||
if self._dp is None:
|
||||
self._dp = make_data_provider(PlatformConfig())
|
||||
return self._dp
|
||||
|
||||
def _news(self) -> NewsEngine:
|
||||
if self._news_engine is None:
|
||||
self._news_engine = NewsEngine(PlatformConfig().news)
|
||||
return self._news_engine
|
||||
|
||||
def install_static(self) -> None:
|
||||
"""Copy the repo's dashboard.html into the out dir (atomic), so
|
||||
git pull + restart also updates the page."""
|
||||
src = os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||||
"static", PAGE_FILE)
|
||||
if not os.path.exists(src):
|
||||
logger.warning(f"{src} missing — page not installed")
|
||||
return
|
||||
dst = os.path.join(self.out_dir, PAGE_FILE)
|
||||
shutil.copyfile(src, dst + ".tmp")
|
||||
os.replace(dst + ".tmp", dst)
|
||||
|
||||
def scan_tf(self, tf: Timeframe) -> None:
|
||||
started = time.time()
|
||||
try:
|
||||
report = produce_signal(XAUUSD, tf, self._provider(),
|
||||
CONFIRMATIONS[tf])
|
||||
self.snapshots[tf.value] = {
|
||||
"status": "ok",
|
||||
"scanned_at": datetime.now(timezone.utc).isoformat(),
|
||||
"scan_seconds": round(time.time() - started, 2),
|
||||
"error": None,
|
||||
"error_at": None,
|
||||
"report": report_to_webapp_dict(report),
|
||||
}
|
||||
logger.info(f"{tf.value}: {report.final_signal.name} "
|
||||
f"q={report.signal_quality or '-'} "
|
||||
f"({time.time() - started:.1f}s)")
|
||||
except Exception as e:
|
||||
# Keep the last good report; the page shows it with an
|
||||
# error badge instead of going blank.
|
||||
logger.exception(f"{tf.value} scan failed")
|
||||
entry = self.snapshots.get(tf.value) or _empty_tf_entry()
|
||||
entry["status"] = "error"
|
||||
entry["error"] = f"{type(e).__name__}: {e}"
|
||||
entry["error_at"] = datetime.now(timezone.utc).isoformat()
|
||||
self.snapshots[tf.value] = entry
|
||||
|
||||
def refresh_news(self) -> None:
|
||||
try:
|
||||
ne = self._news()
|
||||
context = ne.get_fundamental_context()
|
||||
week = ne.get_week_events()
|
||||
self.news_payload = build_news_payload(
|
||||
context, week, datetime.now(timezone.utc))
|
||||
except Exception:
|
||||
logger.exception("news refresh failed — keeping last payload")
|
||||
|
||||
def publish_all(self) -> None:
|
||||
now = datetime.now(timezone.utc)
|
||||
publish(build_snapshot_payload(self.snapshots, now),
|
||||
os.path.join(self.out_dir, SNAPSHOT_FILE))
|
||||
if self.news_payload is not None:
|
||||
publish(self.news_payload, os.path.join(self.out_dir, NEWS_FILE))
|
||||
|
||||
def run_cycle(self, tfs: List[Timeframe]) -> None:
|
||||
for tf in tfs: # sequential — see module docstring (cache dedup)
|
||||
self.scan_tf(tf)
|
||||
self.refresh_news()
|
||||
self.publish_all()
|
||||
|
||||
def run_forever(self) -> None:
|
||||
self.run_cycle(DASHBOARD_TFS) # populate the page at boot
|
||||
while True:
|
||||
delay = seconds_until_boundary(5, buffer=SCAN_BUFFER_SECONDS)
|
||||
logger.info(f"next M5 boundary in {delay:.0f}s")
|
||||
time.sleep(delay)
|
||||
try:
|
||||
due = due_timeframes(datetime.now(timezone.utc))
|
||||
logger.info(f"boundary wake — scanning {[t.value for t in due]}")
|
||||
self.run_cycle(due)
|
||||
except Exception:
|
||||
logger.exception("cycle failed")
|
||||
|
||||
|
||||
def main(argv=None) -> None:
|
||||
ap = argparse.ArgumentParser(description="XAU/USD static dashboard publisher")
|
||||
ap.add_argument("--once", action="store_true",
|
||||
help="one scan+publish cycle, then exit")
|
||||
ap.add_argument("--out",
|
||||
default=os.environ.get("DASHBOARD_OUT_DIR", DEFAULT_OUT_DIR),
|
||||
help="output directory served by Apache "
|
||||
f"(default: $DASHBOARD_OUT_DIR or {DEFAULT_OUT_DIR})")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
_setup_logging()
|
||||
os.makedirs(args.out, exist_ok=True)
|
||||
pub = DashboardPublisher(args.out)
|
||||
pub.install_static()
|
||||
if args.once:
|
||||
pub.run_cycle(DASHBOARD_TFS)
|
||||
print(f"Published to {args.out}")
|
||||
return
|
||||
pub.run_forever()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user