Harwinder Sekhon
f7c244c6b5
- providers: _to_oanda_instrument / _to_tiingo_ticker / _to_td_symbol no longer guess a wire symbol for pairs that are not registered — the registries are the only source, so an unknown symbol raises the provider's DataIntegrityError instead of silently fetching something else. Every scanned symbol is validated against PAIR_INFO at import, so nothing reachable changes. - main / telegram_bot _fmt_price: the sub-$10 five-decimal branch could never fire for gold. - news_engine._parse_pair: BASE/QUOTE only; the Yahoo "=X" fallback was unreachable (its test goes with it). - dashboard_publisher.event_to_dict: drop minutes_until (frozen at publish time, misleading) and is_high_impact — dashboard.js never read either. - static/dashboard.html: unused id="cal-table". Tried and reverted: trimming the LogRow fields nothing in analyze_signal_log reads — the tests read them to verify the loader round-trips those DB columns, so they stay. Suite: 460 passed, 0 failed. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0156YhogLytzj9P2iH1skGbL
482 lines
19 KiB
Python
482 lines
19 KiB
Python
#!/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 copies of the static
|
||
`dashboard.html` / `.css` / `.js` and `news_summary.php`) 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` (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/trade
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import hashlib
|
||
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 agent_panel import attach_panel
|
||
from app_config import get_app_config, violent_watch_enabled
|
||
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
|
||
from violent_watch import attach_watch
|
||
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/trade"
|
||
SCAN_BUFFER_SECONDS = 10
|
||
SNAPSHOT_FILE = "snapshot.json"
|
||
NEWS_FILE = "news.json"
|
||
PAGE_FILE = "dashboard.html"
|
||
SUMMARY_PHP_FILE = "news_summary.php"
|
||
# Repo files copied into the out dir at daemon start (git pull + restart
|
||
# updates them). The PHP endpoint reads news.json in place, so it lives
|
||
# next to the page.
|
||
STATIC_FILES = [PAGE_FILE, "dashboard.css", "dashboard.js", SUMMARY_PHP_FILE]
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# 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 is a property, so asdict() would
|
||
skip it — 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,
|
||
}
|
||
|
||
|
||
def headline_to_dict(h) -> dict:
|
||
# id uses the same sha1(headline) scheme as NewsEngine._llm_sent_cache;
|
||
# it is the join key the PHP summary endpoint whitelists against.
|
||
return {
|
||
"id": hashlib.sha1(h.headline.encode("utf-8")).hexdigest(),
|
||
"headline": h.headline,
|
||
"source": h.source,
|
||
"link": h.link,
|
||
"description": h.description,
|
||
"publisher": h.publisher,
|
||
"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 — 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
|
||
# NOTE: remaining > buffer always (elapsed < interval by construction);
|
||
# a call at the exact boundary instant targets the NEXT boundary —
|
||
# steady-state loops wake at boundary+buffer, so only a process
|
||
# started in the boundary microsecond skips one scan (2026-08-04
|
||
# audit I42; the old `remaining <= buffer` guard was unreachable).
|
||
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._reports = {}
|
||
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 static files (page + PHP summary endpoint)
|
||
into the out dir (atomic), so git pull + restart also updates
|
||
them. The ai/ cache dir is NOT created here — the daemon runs as
|
||
a different user than Apache; that dir is a one-time deploy step
|
||
(deploy/README-dashboard.md)."""
|
||
static_dir = os.path.join(
|
||
os.path.dirname(os.path.abspath(__file__)), "static")
|
||
for name in STATIC_FILES:
|
||
src = os.path.join(static_dir, name)
|
||
if not os.path.exists(src):
|
||
logger.warning(f"{src} missing — not installed")
|
||
continue
|
||
dst = os.path.join(self.out_dir, name)
|
||
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])
|
||
# Display-only violent-move watch (dark by default). Never
|
||
# raises and never touches the signal — only watch_line.
|
||
if violent_watch_enabled():
|
||
attach_watch(report, tf,
|
||
get_app_config().violent_watch_settings)
|
||
self._reports[tf.value] = report
|
||
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:
|
||
# Keep the last good report; the page shows it with an
|
||
# error badge instead of going blank.
|
||
logger.exception(f"{tf.value} scan failed")
|
||
self._reports.pop(tf.value, None)
|
||
entry = self.snapshots.get(tf.value) or _empty_tf_entry()
|
||
entry["status"] = "error"
|
||
# The public snapshot carries a stable message; exception
|
||
# detail remains in dashboard.log for the operator.
|
||
entry["error"] = "Scan unavailable"
|
||
entry["error_at"] = datetime.now(timezone.utc).isoformat()
|
||
self.snapshots[tf.value] = entry
|
||
|
||
def attach_panels(self, tfs: List[Timeframe]) -> bool:
|
||
"""Add advisory panels after publishing the technical snapshot.
|
||
|
||
A panel is display-only and can take tens of seconds on a cold LLM
|
||
request. Keeping its report object locally lets the page receive the
|
||
candle-close signal first, then a second atomic snapshot only when a
|
||
panel produced a result.
|
||
"""
|
||
updated = False
|
||
for tf in tfs:
|
||
report = self._reports.get(tf.value)
|
||
if report is None:
|
||
continue
|
||
if not attach_panel(report, surface="dashboard"):
|
||
continue
|
||
entry = self.snapshots.get(tf.value)
|
||
if entry is None or entry.get("status") != "ok":
|
||
continue
|
||
entry["report"] = report_to_webapp_dict(report)
|
||
updated = True
|
||
return updated
|
||
|
||
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_snapshot(self) -> None:
|
||
publish(build_snapshot_payload(
|
||
self.snapshots, datetime.now(timezone.utc)),
|
||
os.path.join(self.out_dir, SNAPSHOT_FILE))
|
||
|
||
def publish_news(self) -> None:
|
||
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)
|
||
# Technical signals land before any optional slow work. News and the
|
||
# advisory panel are display-only and may take 20-30 seconds cold.
|
||
self.publish_snapshot()
|
||
if self.attach_panels(tfs):
|
||
self.publish_snapshot()
|
||
self.refresh_news()
|
||
self.publish_news()
|
||
|
||
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()
|