Harwinder Sekhon
ad14ea4fed
Agent panel v2: personas return p_win anchored to the book base rates with a code-computed decision context; verdict = equal-weight mean (verdict_source "mean"); only the macro persona runs live (agent_panel.personas ["macro"]); prompt version and exact payloads logged; offline replay harness scripts/panel_replay.py; base-rate table agent_panel_base_rates.json (scripts/build_panel_base_rates.py). Backtester `strategy` column now carries the generator's own stamp (label-only; five reference books re-pinned, all 62 other columns identical). News corpus (after evaluating the owner's ML-NEWS scikit model against DeepSeek: golden set 39/78 vs 78/78, out-of-sample agreement 64% against a 90% self-agreement ceiling; verdict: fallback only, not a replacement): - NewsSentiment.xau_score + archive column xau_score keep DeepSeek's gold score for every headline (930 of the first 1,546 DeepSeek rows had none); an existing newsdata.db gains the column in place on first open - headlines naming no tracked currency are scored and archived (currencies_json []) but never returned to the platform - _apply_llm_sentiment sends every new headline in its cycle, 25 per call; a failed call ends the cycle's sending Dashboard publisher never writes a database: main() calls app_config.set_db_writes_for_process(False), honoured by signal_log_enabled() and news_archive_enabled() (owner decision: the bot is the single writer). Test suite: the hermetic app config now points both writers at a suite-private temp dir — the RSS fixture tests were writing fixture headlines into the real logs/newsdata.db — with a Module 32 test pinning it. 529 tests.
581 lines
24 KiB
Python
581 lines
24 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.
|
||
* Never writes a database. It is not always running, so the Telegram
|
||
bot is the single writer of `signals.db` and `newsdata.db`; main()
|
||
switches every SQLite writer off for this process
|
||
(`app_config.set_db_writes_for_process`, owner decision 2026-09-17).
|
||
|
||
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).
|
||
A cycle can overrun its 5-minute slot (or the process can be down), so
|
||
each wake also catches up the timeframes whose latest closed boundary is
|
||
newer than their last scanned one (`catch_up_timeframes`) — without it the
|
||
page served an hours-old H1/H4 report behind a fresh countdown
|
||
(2026-09-13 audit).
|
||
|
||
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, timedelta, timezone
|
||
from logging.handlers import TimedRotatingFileHandler
|
||
from typing import Dict, List, Optional
|
||
|
||
import pandas as pd
|
||
|
||
from agent_panel import attach_panel
|
||
from app_config import (get_app_config, set_db_writes_for_process,
|
||
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).
|
||
|
||
A timeframe whose just-closed bar would carry a label inside gold's
|
||
daily settlement pause is skipped: no such bar exists, so the scan
|
||
would only re-read the previous one (2026-09-13 audit)."""
|
||
minutes = wake_dt.hour * 60 + wake_dt.minute
|
||
floored = minutes - (minutes % 5)
|
||
boundary = wake_dt.replace(
|
||
hour=floored // 60, minute=floored % 60, second=0, microsecond=0)
|
||
return [
|
||
tf for tf in DASHBOARD_TFS
|
||
if floored % TF_INTERVAL_MINUTES[tf] == 0
|
||
and not _providers_base._is_daily_maintenance(
|
||
boundary - timedelta(minutes=TF_INTERVAL_MINUTES[tf]))
|
||
]
|
||
|
||
|
||
def catch_up_timeframes(wake_dt: datetime,
|
||
last_boundary: Dict[str, float]) -> List[Timeframe]:
|
||
"""Timeframes whose most recent closed boundary (<= wake) is newer than
|
||
the boundary of their last successful scan — i.e. boundaries that were
|
||
missed while a cycle overran or the process was down. A TF never scanned
|
||
is due. Excludes bars whose label sits in the daily settlement pause
|
||
(same rule as due_timeframes).
|
||
|
||
`due_timeframes` only ever names the timeframes whose boundary coincides
|
||
with THIS wake, so an overrunning cycle (scan + advisory panel + news can
|
||
take minutes) or a restart left the page serving the previous H1/H4
|
||
report for up to 1 h / 4 h while advertising a fresh next_close_epoch
|
||
(2026-09-13 audit)."""
|
||
minutes = wake_dt.hour * 60 + wake_dt.minute
|
||
out = []
|
||
for tf in DASHBOARD_TFS:
|
||
interval = TF_INTERVAL_MINUTES[tf]
|
||
floored = minutes - (minutes % interval)
|
||
last_closed = wake_dt.replace(
|
||
hour=floored // 60, minute=floored % 60, second=0, microsecond=0)
|
||
# No bar carries a label inside a scheduled closure (settlement
|
||
# pause or weekend), so nothing can have been "missed" there — and a
|
||
# weekend-blocked scan never advances the ledger, so without the
|
||
# weekend half of this test every weekend wake would re-scan
|
||
# M15/H1/H4 for nothing (~1,800 blocked rows per weekend).
|
||
# (Whole-span rule, as in providers.base._has_latest_closed_bar: an
|
||
# H4 bar labelled Sunday 20:00 UTC starts in the weekend but runs
|
||
# through the reopen, so it exists.)
|
||
label = last_closed - timedelta(minutes=interval)
|
||
if (_providers_base._is_scheduled_closure(label)
|
||
and _providers_base._is_scheduled_closure(
|
||
last_closed - timedelta(seconds=1))):
|
||
continue
|
||
scanned = last_boundary.get(tf.value)
|
||
if scanned is None or scanned < last_closed.timestamp():
|
||
out.append(tf)
|
||
return out
|
||
|
||
|
||
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 = {}
|
||
# tf.value → epoch seconds of the boundary the last successful scan
|
||
# belonged to; feeds catch_up_timeframes (2026-09-13 audit).
|
||
self._last_boundary: Dict[str, float] = {}
|
||
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,
|
||
boundary_epoch: Optional[float] = None) -> None:
|
||
"""Scan one timeframe and store its snapshot entry.
|
||
|
||
`boundary_epoch` is the fallback boundary recorded when the report
|
||
carries no analysis (weekend/blocked or failed scan) or its bar
|
||
label will not parse; normally the boundary comes from the report
|
||
itself (bar label + interval)."""
|
||
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)
|
||
# Which boundary this report belongs to: the bar LABEL it closed
|
||
# on plus one interval. Recorded so a boundary missed by an
|
||
# overrunning cycle or a restart is caught up on the next wake
|
||
# instead of sitting stale (2026-09-13 audit).
|
||
boundary = boundary_epoch
|
||
if report.primary_analysis is not None:
|
||
try:
|
||
boundary = (pd.Timestamp(report.timestamp).timestamp()
|
||
+ TF_INTERVAL_MINUTES[tf] * 60)
|
||
except Exception:
|
||
boundary = boundary_epoch
|
||
if boundary is not None:
|
||
self._last_boundary[tf.value] = boundary
|
||
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:
|
||
now = datetime.now(timezone.utc)
|
||
due = due_timeframes(now)
|
||
logger.info(f"boundary wake — scanning {[t.value for t in due]}")
|
||
missed = catch_up_timeframes(now, self._last_boundary)
|
||
extra = [tf for tf in missed if tf not in due]
|
||
if extra:
|
||
logger.info(f"catch-up — scanning "
|
||
f"{[t.value for t in extra]} (missed boundary)")
|
||
self.run_cycle(due + extra)
|
||
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()
|
||
# This process never writes a database. It is not always running, so
|
||
# the Telegram bot is the single writer of signals.db and newsdata.db
|
||
# (owner decision 2026-09-17). The switch is process-wide and beats
|
||
# config.json / env; set here in main(), not at import, so importing
|
||
# the module (the test suite does) leaves every other writer on.
|
||
set_db_writes_for_process(False)
|
||
logger.info("database writes disabled for this process "
|
||
"(the Telegram bot is the only writer)")
|
||
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()
|