472 lines
22 KiB
Python
472 lines
22 KiB
Python
"""App-internal background scheduler — refreshes real data on its own schedule.
|
|
|
|
This app runs on its own server (Docker/EasyPanel), independent of Hermes. So
|
|
the periodic data-fetching/analysis must live INSIDE the app. This module runs a
|
|
daemon thread inside the Flask process that:
|
|
|
|
- on an interval (default every 60 min), triggers each collector through the
|
|
shared daily cache (so fresh real data lands as the "current" state),
|
|
- records each successful refresh as a timestamped snapshot under the data dir
|
|
(persistent history the dashboard/sources table can report),
|
|
- logs progress; swallows transient failures (a source being down once must
|
|
not crash the app or the loop).
|
|
|
|
Thread-safety: the loop only calls collectors + writes cache/snapshots; the HTTP
|
|
layer reads the daily cache. No shared mutable in-process state beyond that.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable, List, Optional
|
|
|
|
log = logging.getLogger("set50.scheduler")
|
|
|
|
# collectors returning a .to_dict()/dict, keyed by cache key
|
|
# (imported lazily to avoid import cycles at module load)
|
|
# `fetch_module` = the FACTORS.fetch module name this job feeds (for history).
|
|
# `frequency` = natural refresh cadence of the source. A job is only run
|
|
# once its cooldown window has elapsed — the strategy rebalances
|
|
# a few times a year, so polling slow sources hourly wastes
|
|
# resources. Values: daily | weekly | monthly | quarterly.
|
|
_REFRESH_JOBS: List[dict] = [
|
|
{"key": "bot_tourism", "label": "ท่องเที่ยว (BOT)", "module": "bot_tourism", "fn": "BotTourismSource().fetch", "fetch_module": "macro_thai", "frequency": "monthly"},
|
|
{"key": "auto_credit/tourism", "label": "ยอดขายรถ (TradingEconomics)", "module": "auto_credit", "fn": "fetch_auto_credit", "fetch_module": "auto_credit", "frequency": "monthly"},
|
|
{"key": "auto_npl", "label": "NPL รถยนต์ (BOT)", "module": "auto_npl", "fn": "fetch_auto_npl", "fetch_module": "auto_npl", "frequency": "quarterly"},
|
|
{"key": "energy_thai", "label": "โรงกลั่น TOP", "module": "energy_thai", "fn": "fetch_energy_thai", "fetch_module": "energy_thai", "frequency": "quarterly"},
|
|
{"key": "macro_thai", "label": "ภาพรวมประเทศไทย (BOT)", "module": "macro_thai", "fn": "fetch_macro_thai", "fetch_module": "macro_thai", "frequency": "monthly"},
|
|
{"key": "bank_npl", "label": "NPL ภาคการเงิน (BOT)", "module": "bank_npl", "fn": "fetch_bank_npl", "fetch_module": "bank_npl", "frequency": "quarterly"},
|
|
]
|
|
|
|
# Frequencies -> minimum seconds between successful refreshes of a job.
|
|
# These implement the "adjust cadence to the source" requirement.
|
|
_FREQ_SECONDS: dict[str, int] = {
|
|
"daily": 24 * 3600,
|
|
"weekly": 7 * 24 * 3600,
|
|
"monthly": 30 * 24 * 3600,
|
|
"quarterly": 91 * 24 * 3600,
|
|
}
|
|
|
|
# Data that legitimately changes every day (or faster) is refreshed separately
|
|
# at a daily cadence rather than on the slow factor loop.
|
|
_DAILY_JOBS: List[dict] = [
|
|
{"key": "siamchart_vintages", "label": "Siamchart SET50 snapshot (vintage)", "fn": "_record_siamchart_vintages", "frequency": "daily"},
|
|
{"key": "price_snapshot", "label": "ราคาหุ้น SET50 (Yahoo)", "fn": "_refresh_price_snapshot", "frequency": "daily"},
|
|
]
|
|
|
|
|
|
class AppDataScheduler:
|
|
"""Runs periodic refresh of the real data collectors inside the app."""
|
|
|
|
def __init__(self, cache, data_root: Path, interval_seconds: int = 3600,
|
|
dividend_cooldown_seconds: int = 6 * 3600):
|
|
self.cache = cache
|
|
self.data_root = data_root
|
|
self.interval = max(30, int(interval_seconds)) # never faster than 30s
|
|
self.dividend_cooldown = max(300, int(dividend_cooldown_seconds)) # >=5min
|
|
self._stop = threading.Event()
|
|
self._thread: Optional[threading.Thread] = None
|
|
self._snap_dir = data_root / "scheduler"
|
|
self._snap_dir.mkdir(parents=True, exist_ok=True)
|
|
self._div_marker = self._snap_dir / "dividends_last_refresh.json"
|
|
self._lock = threading.Lock()
|
|
|
|
# -- public lifecycle ---------------------------------------------------
|
|
def start(self) -> None:
|
|
if self._thread and self._thread.is_alive():
|
|
return
|
|
self._stop.clear()
|
|
self._thread = threading.Thread(target=self._loop, name="set50-refresh", daemon=True)
|
|
self._thread.start()
|
|
log.info("set50 data scheduler started (interval=%ss)", self.interval)
|
|
|
|
def stop(self) -> None:
|
|
self._stop.set()
|
|
|
|
# -- internals ----------------------------------------------------------
|
|
def _loop(self) -> None:
|
|
# refresh once shortly after boot (so fresh data is live), then on interval
|
|
self.refresh_all()
|
|
while not self._stop.wait(self.interval):
|
|
try:
|
|
self.refresh_all()
|
|
except Exception: # noqa: BLE001 — loop must survive individual failures
|
|
log.exception("set50 scheduler refresh_all failed (will retry)")
|
|
|
|
def refresh_all(self) -> list[dict]:
|
|
"""Run each collector (respecting its natural cadence), warm the cache,
|
|
snapshot state, write vintages, and log per-source health.
|
|
|
|
Slow sources (monthly/quarterly factors, dividends, Siamchart) are only
|
|
re-fetched once their cooldown has elapsed; only daily-changing data
|
|
(prices) runs every eligible tick. This keeps resource usage aligned
|
|
with how often the data actually changes (the strategy rebalances a few
|
|
times a year) while still front-loading an initial refresh at boot.
|
|
"""
|
|
results: list[dict] = []
|
|
fetched_by_module: dict[str, dict] = {}
|
|
|
|
for job in _REFRESH_JOBS:
|
|
if not self._job_due(job):
|
|
continue
|
|
res = self._run_job(job)
|
|
results.append(res)
|
|
if res.get("ok"):
|
|
self._mark_job_run(job)
|
|
fetch_module = job.get("fetch_module")
|
|
if isinstance(res.get("value"), dict) and fetch_module:
|
|
fetched_by_module[fetch_module] = res["value"]
|
|
|
|
# daily cadence data (Siamchart vintages + price snapshots)
|
|
for djob in _DAILY_JOBS:
|
|
if not self._job_due(djob):
|
|
continue
|
|
try:
|
|
fn = getattr(self, djob["fn"])
|
|
fn()
|
|
self._mark_job_run(djob)
|
|
results.append({"key": djob["key"], "label": djob["label"],
|
|
"ok": True, "at": self._now()})
|
|
except Exception as exc: # noqa: BLE001 — non-fatal
|
|
results.append({"key": djob["key"], "label": djob["label"],
|
|
"ok": False, "error": str(exc), "at": self._now()})
|
|
|
|
self._record_history(fetched_by_module)
|
|
self._record_pit_factor_vintages(fetched_by_module)
|
|
self._maybe_refresh_dated_dividends()
|
|
self._write_marker(results)
|
|
self._append_source_log(results)
|
|
return results
|
|
|
|
# -- per-job cadence cooldown ------------------------------------------
|
|
def _job_marker_path(self, job: dict) -> Path:
|
|
safe = "".join(c if (c.isalnum() or c in "._-") else "_" for c in job["key"])
|
|
return self._snap_dir / f"job_{safe}.json"
|
|
|
|
def _job_due(self, job: dict) -> bool:
|
|
"""True if the job's cooldown (from its `frequency`) has elapsed."""
|
|
import json
|
|
import datetime as _dt
|
|
freq = job.get("frequency")
|
|
cooldown = _FREQ_SECONDS.get(freq if isinstance(freq, str) else "daily", _FREQ_SECONDS["daily"])
|
|
path = self._job_marker_path(job)
|
|
if not path.is_file():
|
|
return True # never run -> run at boot
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
last = _dt.datetime.fromisoformat(data["at"])
|
|
elapsed = (_dt.datetime.now().astimezone() - last).total_seconds()
|
|
return elapsed >= cooldown
|
|
except (OSError, ValueError, KeyError):
|
|
return True # corrupt/missing marker -> allow retry
|
|
|
|
def _mark_job_run(self, job: dict) -> None:
|
|
import json
|
|
try:
|
|
self._job_marker_path(job).write_text(
|
|
json.dumps({"at": self._now()}), encoding="utf-8")
|
|
except OSError:
|
|
log.exception("could not write job marker for %s", job["key"])
|
|
|
|
def _refresh_price_snapshot(self) -> None:
|
|
"""Refresh the SET50 Yahoo price snapshot (daily cadence).
|
|
|
|
Pulls live daily bars over a rolling ~3y window into the price snapshot
|
|
store so valuation and next-trading-day execution use fresh prices. This
|
|
is the one data type that legitimately changes every trading day, so it
|
|
runs on the daily cadence rather than the slow factor loop. Non-fatal:
|
|
on network failure the previous snapshot is retained.
|
|
"""
|
|
import datetime as _dt
|
|
from .prices import collect_price_snapshot, PriceSourceError
|
|
# rolling window: start ~3y back (enough history for momentum + next-day
|
|
# execution), end = yesterday (SET session close is the latest tradable).
|
|
today = _dt.date.today()
|
|
start = (today - _dt.timedelta(days=3 * 366)).isoformat()
|
|
end = (today - _dt.timedelta(days=1)).isoformat()
|
|
prices_dir = self.data_root / "prices"
|
|
try:
|
|
collect_price_snapshot(prices_dir, start=start, end=end)
|
|
except PriceSourceError as exc:
|
|
log.warning("set50 price snapshot refresh failed (non-fatal): %s", exc)
|
|
raise
|
|
|
|
def _record_history(self, fetched_by_module: dict[str, dict]) -> None:
|
|
"""Append current factor values to the historical store (append-only).
|
|
|
|
`fetched_by_module` maps FACTORS.fetch module name -> collector dict.
|
|
Runs after each refresh so vintages accumulate; macro/demographic
|
|
factors become learnable (P4) once they have enough history points.
|
|
"""
|
|
try:
|
|
from .factor_history import FactorHistory
|
|
fh = FactorHistory(self.data_root / "factor_history")
|
|
fh.record_all(fetched_by_module)
|
|
except Exception: # noqa: BLE001 — never let history break the refresh loop
|
|
log.exception("set50 factor-history record failed (non-fatal)")
|
|
|
|
def _record_pit_factor_vintages(self, fetched_by_module: dict[str, dict]) -> None:
|
|
"""Write each factor's current PIT vintage into the strict PIT store.
|
|
|
|
The strict backtest (`backtest_readiness`) reads *this* store, so the
|
|
scheduler must keep it populated or strict runs stay blocked forever.
|
|
We use the honest, evidence-safe timestamps available at collection
|
|
time: ``released_at == retrieved_at == now`` — we do NOT assume a
|
|
reporting lag we do not know, so we only claim the value was knowable
|
|
from the moment we actually collected it. A point is appended only when
|
|
the factor's value differs from its last stored value, so repeated
|
|
ticks do not spam rows. This runs on the deployed server too, so
|
|
vintages accumulate wherever the app runs.
|
|
"""
|
|
from . import factors as factors_mod
|
|
from .factor_vintages import FactorVintageStore, FactorVintageError
|
|
try:
|
|
store = FactorVintageStore(self.data_root)
|
|
except Exception: # noqa: BLE001
|
|
log.exception("set50 PIT-vintage store init failed (non-fatal)")
|
|
return
|
|
now = self._now() # tz-aware ISO (UTC seconds)
|
|
for fkey, fact in factors_mod.FACTORS.items():
|
|
if not fact.get("fetch"):
|
|
continue
|
|
val = factors_mod.factor_value(fact, fetched_by_module.get(fact.get("fetch")))
|
|
if val is None:
|
|
continue
|
|
try:
|
|
last = store.value_at(fkey, now)
|
|
except FactorVintageError:
|
|
last = None
|
|
if last is not None and abs(last - val) < 1e-9:
|
|
continue # unchanged since last stored -> skip
|
|
try:
|
|
store.record(
|
|
fkey, val,
|
|
observed_at=now, released_at=now,
|
|
retrieved_at=now, source=str(fact.get("source") or ""),
|
|
)
|
|
except FactorVintageError as exc:
|
|
log.warning("set50 PIT vintage skip %s: %s", fkey, exc)
|
|
|
|
def _record_siamchart_vintages(self) -> None:
|
|
"""Persist the current Siamchart SET50 snapshot as a vintage chain.
|
|
|
|
This makes the fundamental (40%) dimension reconstructible for strict
|
|
PIT backtests. Idempotent: ``SiamchartVintageStore.persist`` dedupes by
|
|
``retrieved_at`` + body hash, so re-inserting an unchanged snapshot is a
|
|
no-op. Non-fatal on network error. Reads the same latest snapshot file
|
|
the dashboard uses (``backend/data/siamchart/set50_master.json``).
|
|
"""
|
|
import json
|
|
from .siamchart_vintages import SiamchartVintageStore, SiamchartVintageError
|
|
path = self.data_root / "siamchart" / "set50_master.json"
|
|
try:
|
|
if not path.is_file():
|
|
return
|
|
snap = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError) as exc:
|
|
log.warning("set50 siamchart snapshot unreadable: %s", exc)
|
|
return
|
|
if not isinstance(snap, dict) or not snap:
|
|
return
|
|
try:
|
|
store = SiamchartVintageStore(self.data_root)
|
|
store.persist(snap)
|
|
except SiamchartVintageError as exc:
|
|
log.warning("set50 siamchart vintage skip: %s", exc)
|
|
except Exception: # noqa: BLE001
|
|
log.exception("set50 siamchart vintage persist failed (non-fatal)")
|
|
|
|
def _maybe_refresh_dated_dividends(self) -> None:
|
|
"""Cooldown-gated fetch of real dated dividend history into the ledger.
|
|
|
|
Dividend history changes slowly (a few times a year per name), so we
|
|
only re-fetch at most once per ``dividend_cooldown`` (default 6h), not
|
|
every refresh tick. Non-fatal: a network failure leaves the previous
|
|
ledger intact and just postpones the next update.
|
|
"""
|
|
with self._lock:
|
|
if self._dividends_fresh():
|
|
return
|
|
try:
|
|
from .dividend_ledger import DividendLedger, populate_dated_dividends
|
|
from .siamchart import fetch_dividend_history
|
|
from .siamchart_factors import build_factor_view
|
|
ledger = DividendLedger(self.data_root / "dividends" / "ledger.json")
|
|
fv = build_factor_view()
|
|
symbols = [f["symbol"] for f in fv.get("factors", [])]
|
|
populate_dated_dividends(ledger, symbols, fetch_dividend_history)
|
|
ledger.save()
|
|
self._mark_dividends_refreshed()
|
|
log.info("set50 dated-dividend ledger refreshed (%d symbols)", len(symbols))
|
|
except Exception: # noqa: BLE001 — non-fatal
|
|
log.exception("set50 dated-dividend refresh failed (will retry next cooldown)")
|
|
|
|
def _dividends_fresh(self) -> bool:
|
|
import json
|
|
import datetime as _dt
|
|
if not self._div_marker.is_file():
|
|
return False
|
|
try:
|
|
payload = json.loads(self._div_marker.read_text(encoding="utf-8"))
|
|
last = _dt.datetime.fromisoformat(payload["at"])
|
|
return (_dt.datetime.now().astimezone() - last).total_seconds() < self.dividend_cooldown
|
|
except (OSError, ValueError, KeyError):
|
|
return False
|
|
|
|
def _mark_dividends_refreshed(self) -> None:
|
|
import json
|
|
import datetime as _dt
|
|
try:
|
|
self._div_marker.write_text(
|
|
json.dumps({"at": _dt.datetime.now().astimezone().isoformat(timespec="seconds")}),
|
|
encoding="utf-8",
|
|
)
|
|
except OSError:
|
|
log.exception("could not write dividend refresh marker")
|
|
|
|
def _run_job(self, job: dict) -> dict:
|
|
key = job["key"]
|
|
label = job["label"]
|
|
try:
|
|
value = self.cache.fetch_or_stale(key, self._make_fetcher(job))
|
|
return {"key": key, "label": label, "ok": True, "at": self._now(),
|
|
"value": value}
|
|
except Exception as exc: # noqa: BLE001
|
|
log.warning("set50 refresh failed for %s: %s", key, exc)
|
|
return {"key": key, "label": label, "ok": False, "error": str(exc), "at": self._now()}
|
|
|
|
def _make_fetcher(self, job: dict) -> Callable[[], dict]:
|
|
module_name = job["module"]
|
|
fn = job["fn"]
|
|
|
|
def fetcher() -> dict:
|
|
mod = __import__(f"app.{module_name}", fromlist=["*"])
|
|
obj = mod
|
|
# Support "ClassName().method" (instantiate then call) and plain
|
|
# "func_name" and static object attribute chains.
|
|
parts = fn.split(".")
|
|
for i, part in enumerate(parts):
|
|
is_instantiate = part.endswith("()")
|
|
attr = part[:-2] if is_instantiate else part
|
|
obj = getattr(obj, attr)
|
|
if is_instantiate:
|
|
obj = obj() # construct instance
|
|
if callable(obj):
|
|
value = obj()
|
|
else:
|
|
value = obj
|
|
if hasattr(value, "to_dict"):
|
|
value = value.to_dict()
|
|
return value or {}
|
|
return fetcher
|
|
|
|
def _write_marker(self, results: list[dict]) -> None:
|
|
import json
|
|
import datetime as _dt
|
|
ok = [r for r in results if r["ok"]]
|
|
marker = {
|
|
"at": self._now(),
|
|
"ok": len(ok),
|
|
"total": len(results),
|
|
"sources": results,
|
|
}
|
|
path = self._snap_dir / "last_refresh.json"
|
|
try:
|
|
path.write_text(json.dumps(marker, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
except OSError:
|
|
log.exception("could not write scheduler marker")
|
|
|
|
def _append_source_log(self, results: list[dict]) -> None:
|
|
"""Append this refresh tick's per-source outcome to a durable health log
|
|
the frontend can render (with a one-click copy), and analyze failures.
|
|
|
|
Stored at ``data/scheduler/source_health.json`` (ring buffer, newest
|
|
first). Each entry categorizes the failure (network / http / parse /
|
|
structure / auth / other) so a source that "changed its page structure"
|
|
is distinguishable from a transient network blip.
|
|
"""
|
|
import json
|
|
if not results:
|
|
return
|
|
path = self._snap_dir / "source_health.json"
|
|
try:
|
|
existing = []
|
|
if path.is_file():
|
|
existing = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(existing, list):
|
|
existing = []
|
|
except (OSError, ValueError):
|
|
existing = []
|
|
entries = []
|
|
for r in results:
|
|
category = "ok"
|
|
if not r.get("ok"):
|
|
category = self._analyze_error(r.get("error", ""))
|
|
entries.append({
|
|
"key": r.get("key"), "label": r.get("label"),
|
|
"ok": bool(r.get("ok")), "category": category,
|
|
"at": r.get("at") or self._now(),
|
|
"detail": str(r.get("error") or "")[:500],
|
|
})
|
|
# newest-first ring buffer, cap at 500 entries
|
|
combined = entries + existing
|
|
combined = combined[:500]
|
|
try:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(combined, ensure_ascii=False), encoding="utf-8")
|
|
except OSError:
|
|
log.exception("could not write source-health log")
|
|
|
|
@staticmethod
|
|
def _analyze_error(message: str) -> str:
|
|
"""Classify a failure reason so the frontend can guide diagnosis.
|
|
|
|
Keys off the exception message/type text. Returns one of:
|
|
network, timeout, http, parse, structure, auth, other.
|
|
"""
|
|
m = (message or "").lower()
|
|
if not m:
|
|
return "other"
|
|
if any(t in m for t in ("timed out", "timeout", "timedout")):
|
|
return "timeout"
|
|
if any(t in m for t in ("no such host", "connection refused",
|
|
"name or service not known", "network is unreachable",
|
|
"connection reset", "getaddrinfo", "dns")):
|
|
return "network"
|
|
if any(t in m for t in ("http ", "status", "response code", "403", "404",
|
|
"429", "502", "503")):
|
|
return "http"
|
|
if any(t in m for t in ("json decode", "parse", "unable to find",
|
|
"regex", "no match", "value not found",
|
|
"expecting value", "jsondecodeerror")):
|
|
return "parse"
|
|
if any(t in m for t in ("structure", "schema", "changed", "column",
|
|
"field missing", "keyerror", "attributeerror")):
|
|
return "structure"
|
|
if any(t in m for t in ("auth", "login", "token", "credentials",
|
|
"unauthorized", "forbidden", "401")):
|
|
return "auth"
|
|
return "other"
|
|
|
|
def _load_source_health(self, limit: int = 200) -> list[dict]:
|
|
"""Return the most recent source-health entries (for the API/UI)."""
|
|
import json
|
|
path = self._snap_dir / "source_health.json"
|
|
if not path.is_file():
|
|
return []
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return []
|
|
if not isinstance(data, list):
|
|
return []
|
|
return data[:limit]
|
|
|
|
@staticmethod
|
|
def _now() -> str:
|
|
import datetime as _dt
|
|
return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds")
|