ForexBot2/dashboard_publisher.py

#!/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/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:
    # 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()