""" 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**.")