automation.py

"""
automation.py — daily auto-update pipeline.

Chosen update time (requirement #2): **18:10 America/New_York**, Mon–Fri.
Why: NYSE/Nasdaq close at 16:00 ET; the closing auction, after-hours prints
and Yahoo's consolidated daily bar settle over the following ~1–2 hours.
By 18:10 ET the official daily OHLCV is stable, so we always capture the
finished trading day exactly once (= 07:10 next morning Beijing time).

Pipeline per run:
    1. refresh data for the signal pool + holdings + sector ETFs (yfinance)
    2. re-run multi-level Chan signals  → cached for the Signals tab
    3. rebuild the sector rotation tables
    4. check today's news for every holding (push brief / ignore if quiet)

NOTE for Hugging Face free Spaces: free hardware sleeps after ~48h without
traffic, and a sleeping Space cannot fire its scheduler. Everything here also
runs on demand via the "Run now" button; on paid "always-on" hardware the
schedule fires unattended.
"""
from __future__ import annotations

import os
import datetime as dt
import threading
import traceback
from zoneinfo import ZoneInfo

import paths

NY = ZoneInfo("America/New_York")
RUN_HOUR, RUN_MINUTE = 18, 10

STATE = {
    "signals_df": None,
    "signals_details": {},
    "signals_summary": "Not run yet — press “Run now” or wait for the 18:10 ET schedule.",
    "rotation": (None, None, None, "—"),
    "rotation_narrative": "",
    "news_md": "",
    "last_run": None,
    "running": False,
    "log": [],
}
_lock = threading.Lock()
_SCHED = None  # holds the BackgroundScheduler so it isn't garbage-collected

_LAST_RUN_FILE = os.path.join(paths.OUTPUT_DIR, "last_run.txt")
_RESULTS_FILE = os.path.join(paths.OUTPUT_DIR, "last_results.json")


def _df_to_records(df):
    try:
        return df.to_dict(orient="records") if df is not None and hasattr(df, "to_dict") else []
    except Exception:
        return []


def save_results():
    """Persist the latest pipeline output (signals + rotation + news) to /data so
    the app shows the last run's results after a restart, with no recompute."""
    import json
    try:
        import rotation
        d1, d5, d20, asof = STATE.get("rotation", (None, None, None, "—"))
        payload = {
            "signals_rows": _df_to_records(STATE.get("signals_df")),
            "signals_summary": STATE.get("signals_summary", ""),
            "rotation": {
                "asof": asof,
                "d1": _df_to_records(rotation.fmt_table(d1)) if d1 is not None else [],
                "d5": _df_to_records(rotation.fmt_table(d5)) if d5 is not None else [],
                "d20": _df_to_records(rotation.fmt_table(d20)) if d20 is not None else [],
            },
            "rotation_narrative": STATE.get("rotation_narrative", ""),
            "news_md": STATE.get("news_md", ""),
            "saved_at": dt.datetime.now(NY).isoformat(),
        }
        with open(_RESULTS_FILE, "w", encoding="utf-8") as f:
            json.dump(payload, f, ensure_ascii=False)
    except Exception as e:
        _log(f"save_results failed: {e}")


def load_results() -> dict:
    """Read the persisted last-run results (for the frontend on page load)."""
    import json
    try:
        with open(_RESULTS_FILE, encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}


def _save_last_run(when: dt.datetime):
    try:
        with open(_LAST_RUN_FILE, "w", encoding="utf-8") as f:
            f.write(when.isoformat())
    except Exception:
        pass


def _load_last_run():
    try:
        with open(_LAST_RUN_FILE, encoding="utf-8") as f:
            STATE["last_run"] = dt.datetime.fromisoformat(f.read().strip())
    except Exception:
        STATE["last_run"] = None


_load_last_run()


def _log(msg: str):
    stamp = dt.datetime.now(NY).strftime("%m-%d %H:%M:%S ET")
    STATE["log"] = (STATE["log"] + [f"[{stamp}] {msg}"])[-60:]


def run_pipeline(tickers=None, force: bool = True) -> str:
    """Full daily refresh. Safe to call from the UI or the scheduler."""
    import news_watch
    import rotation
    import signal_runner

    with _lock:
        if STATE["running"]:
            return "A pipeline run is already in progress."
        STATE["running"] = True
    try:
        _log("Pipeline start: refreshing data + signals…")
        df, details, summary, errors = signal_runner.run_signals(tickers, force=force)
        STATE["signals_df"] = df
        STATE["signals_details"] = details
        STATE["signals_summary"] = summary
        for e in errors:
            _log(f"signal skip: {e}")
        _log(f"Signals done: {summary}")

        d1, d5, d20, asof = rotation.build_rotation(force=force)
        STATE["rotation"] = (d1, d5, d20, asof)
        # LLM narrative is generated ON DEMAND from the Rotation tab button —
        # the pipeline itself never waits on the model.
        STATE["rotation_narrative"] = rotation.rotation_brief(d1, d5, d20)
        _log(f"Sector rotation rebuilt (as of {asof}).")

        STATE["news_md"] = news_watch.check_holdings_news()
        _log("Holdings news checked.")

        # ── Feature 4: auto research report for every NEW ticker in the pool ──
        try:
            import json
            import os
            import paths
            import research_agent
            known_path = os.path.join(paths.OUTPUT_DIR, "known_tickers.json")
            try:
                with open(known_path, encoding="utf-8") as f:
                    known = set(json.load(f))
            except (OSError, ValueError):
                known = set()
            current = set(df["Ticker"].tolist()) if df is not None and len(df) else set()
            new_tickers = sorted(current - known)
            generated = set()
            for t in new_tickers[:5]:  # safety cap per run
                _log(f"New ticker {t} → auto-generating research report…")
                report, trace = research_agent.run_research(t, auto=True)
                if report:
                    generated.add(t)
                    _log(f"Report for {t} done{' (+trace)' if trace else ''}.")
                else:
                    _log(f"Report for {t} postponed (sub-agents still loading) — "
                         f"will retry on the next run.")
            done_set = known | (current - (set(new_tickers) - generated))
            if current:
                with open(known_path, "w", encoding="utf-8") as f:
                    json.dump(sorted(done_set), f)
        except Exception as e:
            _log(f"Auto-research skipped: {e}")

        STATE["last_run"] = dt.datetime.now(NY)
        _save_last_run(STATE["last_run"])
        save_results()
        _log("Pipeline finished — Last run updated, results saved.")
        return f"Done. {summary}"
    except Exception as e:
        traceback.print_exc()
        _log(f"Pipeline error (Last run not updated): {e}")
        return f"Pipeline error: {e}"
    finally:
        STATE["running"] = False


def start_scheduler():
    """Cron: Mon–Fri 18:10 America/New_York."""
    try:
        from apscheduler.schedulers.background import BackgroundScheduler
        from apscheduler.triggers.cron import CronTrigger
    except Exception as e:
        _log(f"APScheduler unavailable: {e}")
        return None
    sched = BackgroundScheduler(timezone=NY)
    sched.add_job(run_pipeline, CronTrigger(day_of_week="mon-fri",
                                            hour=RUN_HOUR, minute=RUN_MINUTE),
                  id="daily_pipeline", max_instances=1, coalesce=True)
    sched.start()
    _log(f"Scheduler armed: Mon–Fri {RUN_HOUR:02d}:{RUN_MINUTE:02d} America/New_York.")
    return sched


def schedule_info() -> str:
    now = dt.datetime.now(NY)
    last = STATE["last_run"].strftime("%Y-%m-%d %H:%M ET") if STATE["last_run"] else "never"
    armed = "✅ Scheduler armed" if _SCHED is not None else "⏳ Scheduler starting…"
    return (f"**Schedule:** Mon–Fri **{RUN_HOUR:02d}:{RUN_MINUTE:02d} America/New_York** "
            f"(after the 16:00 ET close).\n\n"
            f"**Now (ET):** {now.strftime('%Y-%m-%d %H:%M')} · **Last run:** {last}\n\n"
            f"{armed} — it runs automatically every weekday at "
            f"{RUN_HOUR:02d}:{RUN_MINUTE:02d} ET. You can also trigger it any time with **Run now**.")