From 3b1bc9a2bf06e7f8e1d73e271d57d4028f90be42 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 26 Jul 2026 22:56:07 +0000 Subject: [PATCH] fix: preserve metrics with atomic persistence --- dashboard/persistence.py | 221 ++++++++++++++++++++++++++ dashboard/server.py | 119 ++++++++------ tests/test_reliability_persistence.py | 99 ++++++++++++ tests/test_server_reliability.py | 119 ++++++++++++++ 4 files changed, 512 insertions(+), 46 deletions(-) create mode 100644 dashboard/persistence.py create mode 100644 tests/test_reliability_persistence.py create mode 100644 tests/test_server_reliability.py diff --git a/dashboard/persistence.py b/dashboard/persistence.py new file mode 100644 index 0000000..58b2972 --- /dev/null +++ b/dashboard/persistence.py @@ -0,0 +1,221 @@ +"""Small, dependency-free persistence primitives for dashboard state.""" + +from __future__ import annotations + +import json +import os +import tempfile +import threading +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterator + +try: + import fcntl +except ImportError: # pragma: no cover - Windows fallback uses the process lock + fcntl = None + + +_LOCKS: dict[str, threading.RLock] = {} +_LOCKS_GUARD = threading.Lock() +_METADATA_KEYS = {"observed_at", "source", "stale", "last_error"} + + +def _thread_lock(path: Path) -> threading.RLock: + key = str(path.resolve()) + with _LOCKS_GUARD: + return _LOCKS.setdefault(key, threading.RLock()) + + +@contextmanager +def file_lock(path: str | os.PathLike[str]) -> Iterator[None]: + """Serialize readers/writers across threads and, on POSIX, processes.""" + target = Path(path) + target.parent.mkdir(parents=True, exist_ok=True) + lock_path = target.with_name(f".{target.name}.lock") + with _thread_lock(target): + with lock_path.open("a+b") as lock_file: + if fcntl is not None: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + if fcntl is not None: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +def atomic_write_json(path: str | os.PathLike[str], data: Any, *, indent: int = 2) -> None: + """Durably replace a JSON file without exposing a partial document.""" + target = Path(path) + target.parent.mkdir(parents=True, exist_ok=True) + with file_lock(target): + fd, temporary = tempfile.mkstemp( + prefix=f".{target.name}.", suffix=".tmp", dir=target.parent + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(data, handle, indent=indent, default=str) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, target) + try: + directory_fd = os.open(target.parent, os.O_DIRECTORY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + except (AttributeError, OSError): + pass + finally: + try: + os.unlink(temporary) + except FileNotFoundError: + pass + + +def load_json(path: str | os.PathLike[str], default: Any = None) -> Any: + target = Path(path) + if not target.exists(): + return default + with file_lock(target): + try: + with target.open(encoding="utf-8") as handle: + return json.load(handle) + except (OSError, ValueError): + return default + + +def _tail_bytes(target: Path, *, line_hint: int, chunk_size: int) -> bytes: + with target.open("rb") as handle: + handle.seek(0, os.SEEK_END) + position = handle.tell() + blocks: list[bytes] = [] + newlines = 0 + while position > 0 and newlines <= line_hint: + size = min(chunk_size, position) + position -= size + handle.seek(position) + block = handle.read(size) + blocks.append(block) + newlines += block.count(b"\n") + return b"".join(reversed(blocks)) + + +def load_jsonl_tail( + path: str | os.PathLike[str], *, limit: int = 90, chunk_size: int = 8192 +) -> list[dict[str, Any]]: + """Read only enough of a JSONL file to return its last valid entries.""" + if limit <= 0: + return [] + target = Path(path) + if not target.exists(): + return [] + with file_lock(target): + raw = _tail_bytes(target, line_hint=limit + 8, chunk_size=max(chunk_size, 32)) + entries: list[dict[str, Any]] = [] + for line in raw.splitlines(): + try: + value = json.loads(line) + except (UnicodeDecodeError, ValueError): + continue + if isinstance(value, dict): + entries.append(value) + return entries[-limit:] + + +def _utc_day(timestamp: Any) -> str | None: + if not isinstance(timestamp, str): + return None + try: + parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc).date().isoformat() + except ValueError: + return None + + +def append_daily_jsonl(path: str | os.PathLike[str], entry: dict[str, Any]) -> bool: + """Append at most one record per UTC day, inspecting only the bounded tail.""" + target = Path(path) + target.parent.mkdir(parents=True, exist_ok=True) + entry_day = _utc_day(entry.get("timestamp")) + if entry_day is None: + raise ValueError("entry timestamp must be an ISO-8601 datetime") + with file_lock(target): + if target.exists(): + raw = _tail_bytes(target, line_hint=8, chunk_size=4096) + for line in reversed(raw.splitlines()): + try: + previous = json.loads(line) + except (UnicodeDecodeError, ValueError): + continue + if _utc_day(previous.get("timestamp")) == entry_day: + return False + break + payload = (json.dumps(entry, default=str) + "\n").encode("utf-8") + fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644) + try: + os.write(fd, payload) + os.fsync(fd) + finally: + os.close(fd) + return True + + +def _has_observation(payload: Any) -> bool: + if not isinstance(payload, dict): + return payload is not None + return any(value is not None for key, value in payload.items() if key not in _METADATA_KEYS) + + +def merge_observation( + previous: Any, + observed: Any, + *, + source: str, + observed_at: str | None = None, + error: str | None = None, +) -> dict[str, Any]: + """Annotate a fresh observation or retain the last-known-good value as stale.""" + if _has_observation(observed): + merged = dict(observed) if isinstance(observed, dict) else {"value": observed} + merged.update( + observed_at=observed_at or datetime.now(timezone.utc).isoformat(), + source=source, + stale=False, + last_error=None, + ) + return merged + + merged = dict(previous) if isinstance(previous, dict) else {} + merged.update( + source=merged.get("source") or source, + stale=True, + last_error=error or "metric was not observed", + ) + merged.setdefault("observed_at", None) + return merged + + +def onchain_refresh_due( + timestamp: Any, + *, + now: datetime | None = None, + ttl_seconds: int = 6 * 60 * 60, +) -> bool: + """Return whether the last successful on-chain observation exceeded its TTL.""" + if not isinstance(timestamp, str) or not timestamp: + return True + try: + observed = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) + if observed.tzinfo is None: + observed = observed.replace(tzinfo=timezone.utc) + except ValueError: + return True + current = now or datetime.now(timezone.utc) + if current.tzinfo is None: + current = current.replace(tzinfo=timezone.utc) + return (current.astimezone(timezone.utc) - observed.astimezone(timezone.utc)).total_seconds() >= ttl_seconds diff --git a/dashboard/server.py b/dashboard/server.py index ffc076f..84c5ad5 100644 --- a/dashboard/server.py +++ b/dashboard/server.py @@ -13,6 +13,7 @@ import sys import threading import time import traceback +from contextlib import asynccontextmanager from datetime import datetime, timezone import requests @@ -28,8 +29,36 @@ sys.path.insert(0, BASE_DIR) from scrapers import fear_greed, price from scoring import engine +from dashboard.persistence import ( + append_daily_jsonl, + atomic_write_json, + load_json, + load_jsonl_tail, + merge_observation, + onchain_refresh_due, +) -app = FastAPI(title="Bitcoin Accumulation Zone Monitor") +_shutdown_event = threading.Event() +_background_threads = [] + + +@asynccontextmanager +async def lifespan(_app): + """Own background worker startup and graceful shutdown.""" + _shutdown_event.clear() + scraper_thread = threading.Thread(target=scraper_loop, name="scraper-scheduler") + _background_threads.append(scraper_thread) + scraper_thread.start() + try: + yield + finally: + _shutdown_event.set() + for thread in list(_background_threads): + thread.join(timeout=30) + _background_threads.clear() + + +app = FastAPI(title="Bitcoin Accumulation Zone Monitor", lifespan=lifespan) CONFIG_DIR = os.path.join(BASE_DIR, "config") DATA_DIR = os.path.join(BASE_DIR, "data") @@ -48,18 +77,11 @@ _last_error = None # ── Cache management ────────────────────────────────────────────────────── def load_cache(): - if os.path.exists(CACHE_PATH): - try: - with open(CACHE_PATH) as f: - return json.load(f) - except Exception: - pass - return {} + return load_json(CACHE_PATH, {}) def save_cache(data): - with open(CACHE_PATH, "w") as f: - json.dump(data, f, indent=2, default=str) + atomic_write_json(CACHE_PATH, data) def append_history(score_data): @@ -73,23 +95,11 @@ def append_history(score_data): for m in score_data.get("metrics", []) }, } - with open(HISTORY_PATH, "a") as f: - f.write(json.dumps(entry) + "\n") + append_daily_jsonl(HISTORY_PATH, entry) def load_history(): - if not os.path.exists(HISTORY_PATH): - return [] - entries = [] - with open(HISTORY_PATH) as f: - for line in f: - line = line.strip() - if line: - try: - entries.append(json.loads(line)) - except Exception: - pass - return entries + return load_jsonl_tail(HISTORY_PATH, limit=90) # ── Background scraper ──────────────────────────────────────────────────── @@ -111,18 +121,35 @@ def run_scrape(force_full=False): _scraper_running = True try: - # Load existing cache to preserve on-chain data existing_cache = load_cache() metrics = {} + cycle_errors = [] - # 1. Fear & Greed (fast API call) + # Fast metrics fail independently so partial outages retain last-known-good data. log.info("Fetching Fear & Greed...") - metrics["fear_greed"] = fear_greed.fetch() + try: + metrics["fear_greed"] = merge_observation( + existing_cache.get("fear_greed"), fear_greed.fetch(), source="alternative.me" + ) + except Exception as e: + cycle_errors.append(f"Fear & Greed: {e}") + metrics["fear_greed"] = merge_observation( + existing_cache.get("fear_greed"), None, + source="alternative.me", error=str(e), + ) - # 2. BTC Price data (fast API calls) log.info("Fetching BTC price...") - price_current = price.fetch_current() - metrics["price"] = price_current + try: + price_current = price.fetch_current() + metrics["price"] = merge_observation( + existing_cache.get("price"), price_current, source="coingecko" + ) + except Exception as e: + cycle_errors.append(f"Price: {e}") + metrics["price"] = merge_observation( + existing_cache.get("price"), None, source="coingecko", error=str(e) + ) + price_current = metrics["price"] log.info("Fetching BTC ATH...") ath_data = price.fetch_ath() @@ -169,21 +196,27 @@ def run_scrape(force_full=False): "active_address_momentum", "txcount_momentum", "nvt_price", "vdd_multiple"] - has_cached_onchain = any(existing_cache.get(k, {}).get("value") is not None for k in onchain_keys) - - if force_full or not has_cached_onchain: - # Only do a full Playwright scrape if explicitly requested or no data exists - log.info("Scraping on-chain metrics from LookIntoBitcoin (full refresh requested)...") + refresh_onchain = force_full or onchain_refresh_due(existing_cache.get("_onchain_timestamp")) + + if refresh_onchain: + log.info("Refreshing on-chain metrics (forced, missing, or TTL expired)...") try: from scrapers import lookintobitcoin onchain = lookintobitcoin.scrape_all() - metrics.update(onchain) try: from scrapers import checkonchain - metrics.update(checkonchain.scrape_all()) + onchain.update(checkonchain.scrape_all()) except Exception as e: log.error("CheckOnChain scraping failed: %s\n%s", e, traceback.format_exc()) - _last_error = f"CheckOnChain scraping failed: {e}" + cycle_errors.append(f"CheckOnChain: {e}") + checkonchain_keys = {"sopr", "sellside_risk", "active_address_momentum", + "txcount_momentum", "nvt_price", "vdd_multiple"} + for key in onchain_keys: + source = "checkonchain" if key in checkonchain_keys else "lookintobitcoin" + metrics[key] = merge_observation( + existing_cache.get(key), onchain.get(key), source=source, + error="metric missing from scrape", + ) metrics["_onchain_timestamp"] = datetime.now(timezone.utc).isoformat() except Exception as e: log.error("LookIntoBitcoin scraping failed: %s\n%s", e, traceback.format_exc()) @@ -224,7 +257,7 @@ def run_scrape(force_full=False): log.warning("History update failed (non-critical): %s", e) _last_update = datetime.now(timezone.utc).isoformat() - _last_error = None + _last_error = "; ".join(cycle_errors) if cycle_errors else None log.info("Scrape cycle complete. Composite score: %s", scored["composite_score"]) except Exception as e: @@ -241,16 +274,10 @@ def scraper_loop(): has_data = any(cache.get(k, {}).get("value") is not None for k in ["puell_multiple", "mvrv_zscore", "nupl"]) run_scrape(force_full=not has_data) # Full only if no cached on-chain data - while True: - time.sleep(900) # 15 minutes + while not _shutdown_event.wait(900): run_scrape() # Quick refresh only -# Start background scraper on import -_scraper_thread = threading.Thread(target=scraper_loop, daemon=True) -_scraper_thread.start() - - # ── LLM Settings (preserved from original) ─────────────────────────────── class LLMSettingsUpdate(BaseModel): diff --git a/tests/test_reliability_persistence.py b/tests/test_reliability_persistence.py new file mode 100644 index 0000000..24b334a --- /dev/null +++ b/tests/test_reliability_persistence.py @@ -0,0 +1,99 @@ +import json +import threading +from datetime import datetime, timedelta, timezone + +import pytest + +from dashboard.persistence import ( + append_daily_jsonl, + atomic_write_json, + load_jsonl_tail, + merge_observation, + onchain_refresh_due, +) + + +def test_atomic_write_json_remains_readable_under_concurrent_writers(tmp_path): + path = tmp_path / "cache.json" + + threads = [ + threading.Thread(target=atomic_write_json, args=(path, {"writer": i, "values": list(range(100))})) + for i in range(12) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + saved = json.loads(path.read_text()) + assert saved["writer"] in range(12) + assert saved["values"] == list(range(100)) + assert not list(tmp_path.glob(".cache.json.*.tmp")) + + +def test_append_daily_jsonl_writes_at_most_one_entry_per_utc_day(tmp_path): + path = tmp_path / "scores.jsonl" + first = {"timestamp": "2026-07-26T01:00:00+00:00", "score": 10} + duplicate_day = {"timestamp": "2026-07-26T23:59:00+00:00", "score": 20} + next_day = {"timestamp": "2026-07-27T00:01:00+00:00", "score": 30} + + assert append_daily_jsonl(path, first) is True + assert append_daily_jsonl(path, duplicate_day) is False + assert append_daily_jsonl(path, next_day) is True + + assert load_jsonl_tail(path, limit=90) == [first, next_day] + + +def test_load_jsonl_tail_is_bounded_and_ignores_malformed_lines(tmp_path): + path = tmp_path / "scores.jsonl" + path.write_text("".join(json.dumps({"n": i}) + "\n" for i in range(200)) + "partial{") + + assert load_jsonl_tail(path, limit=3, chunk_size=64) == [{"n": 197}, {"n": 198}, {"n": 199}] + + +def test_merge_observation_preserves_last_known_good_with_stale_metadata(): + old = { + "value": 1.25, + "observed_at": "2026-07-25T12:00:00+00:00", + "source": "lookintobitcoin", + "stale": False, + "last_error": None, + } + + merged = merge_observation(old, None, source="lookintobitcoin", error="timeout") + + assert merged == { + "value": 1.25, + "observed_at": "2026-07-25T12:00:00+00:00", + "source": "lookintobitcoin", + "stale": True, + "last_error": "timeout", + } + + +def test_merge_observation_records_metadata_for_fresh_value(): + observed_at = "2026-07-26T12:00:00+00:00" + + merged = merge_observation( + {"value": 1.0}, {"value": 2.0, "trend": "up"}, + source="checkonchain", observed_at=observed_at, + ) + + assert merged["value"] == 2.0 + assert merged["trend"] == "up" + assert merged["observed_at"] == observed_at + assert merged["source"] == "checkonchain" + assert merged["stale"] is False + assert merged["last_error"] is None + + +@pytest.mark.parametrize("timestamp", [None, "", "not-a-time"]) +def test_onchain_refresh_due_when_timestamp_is_missing_or_invalid(timestamp): + assert onchain_refresh_due(timestamp, now=datetime(2026, 7, 26, tzinfo=timezone.utc)) is True + + +def test_onchain_refresh_due_after_ttl(): + now = datetime(2026, 7, 26, 12, tzinfo=timezone.utc) + + assert onchain_refresh_due((now - timedelta(hours=5)).isoformat(), now=now, ttl_seconds=21600) is False + assert onchain_refresh_due((now - timedelta(hours=7)).isoformat(), now=now, ttl_seconds=21600) is True diff --git a/tests/test_server_reliability.py b/tests/test_server_reliability.py new file mode 100644 index 0000000..2d910a9 --- /dev/null +++ b/tests/test_server_reliability.py @@ -0,0 +1,119 @@ +import importlib +import json +import sys +import types +from datetime import datetime, timedelta, timezone + +import pytest + + +@pytest.fixture +def server(monkeypatch): + started = [] + monkeypatch.setattr("threading.Thread.start", lambda self: started.append(self)) + sys.modules.pop("dashboard.server", None) + module = importlib.import_module("dashboard.server") + module._threads_started_during_import = started + return module + + +def test_server_import_does_not_start_scheduler_threads(server): + assert server._threads_started_during_import == [] + + +def test_server_cache_and_history_use_reliable_persistence(server, monkeypatch, tmp_path): + monkeypatch.setattr(server, "CACHE_PATH", str(tmp_path / "cache.json")) + monkeypatch.setattr(server, "HISTORY_PATH", str(tmp_path / "scores.jsonl")) + + server.save_cache({"metric": {"value": 1}}) + server.append_history({ + "composite_score": 50, + "scored_count": 1, + "metrics": [{"key": "metric", "score": 5, "value": 1}], + }) + server.append_history({ + "composite_score": 60, + "scored_count": 1, + "metrics": [{"key": "metric", "score": 6, "value": 2}], + }) + + assert server.load_cache() == {"metric": {"value": 1}} + assert len(server.load_history()) == 1 + + +def test_partial_fast_scrape_preserves_last_known_good_metric(server, monkeypatch, tmp_path): + cache_path = tmp_path / "cache.json" + history_path = tmp_path / "scores.jsonl" + now = datetime.now(timezone.utc) + cache_path.write_text(json.dumps({ + "price": { + "price": 65000, + "observed_at": (now - timedelta(minutes=15)).isoformat(), + "source": "coingecko", + "stale": False, + "last_error": None, + }, + "puell_multiple": {"value": 1.2}, + "_onchain_timestamp": now.isoformat(), + })) + monkeypatch.setattr(server, "CACHE_PATH", str(cache_path)) + monkeypatch.setattr(server, "HISTORY_PATH", str(history_path)) + monkeypatch.setattr(server.fear_greed, "fetch", lambda: {"value": 25}) + monkeypatch.setattr(server.price, "fetch_current", lambda: (_ for _ in ()).throw(RuntimeError("price timeout"))) + monkeypatch.setattr(server.price, "fetch_ath", lambda: {"ath": 70000}) + monkeypatch.setattr(server.price, "fetch_historical", lambda: []) + monkeypatch.setattr(server.engine, "score_all", lambda metrics: {"composite_score": 50, "scored_count": 1, "metrics": []}) + monkeypatch.setattr(server.engine, "score_all_ml", lambda metrics: {"composite_score": 50, "scored_count": 1, "metrics": []}) + fake_updater = types.ModuleType("scrapers.history_updater") + fake_updater.update_history = lambda: None + monkeypatch.setitem(sys.modules, "scrapers.history_updater", fake_updater) + + server.run_scrape() + + saved = json.loads(cache_path.read_text()) + assert saved["price"]["price"] == 65000 + assert saved["price"]["stale"] is True + assert "price timeout" in saved["price"]["last_error"] + assert saved["fear_greed"]["value"] == 25 + assert saved["fear_greed"]["stale"] is False + assert saved["fear_greed"]["source"] == "alternative.me" + + +def test_expired_onchain_timestamp_triggers_real_refresh(server, monkeypatch, tmp_path): + old = datetime.now(timezone.utc) - timedelta(hours=7) + cache_path = tmp_path / "cache.json" + cache_path.write_text(json.dumps({ + "puell_multiple": {"value": 1.2}, + "_onchain_timestamp": old.isoformat(), + })) + monkeypatch.setattr(server, "CACHE_PATH", str(cache_path)) + monkeypatch.setattr(server, "HISTORY_PATH", str(tmp_path / "scores.jsonl")) + monkeypatch.setattr(server.fear_greed, "fetch", lambda: {"value": 25}) + monkeypatch.setattr(server.price, "fetch_current", lambda: {"price": 65000}) + monkeypatch.setattr(server.price, "fetch_ath", lambda: {"ath": 70000}) + monkeypatch.setattr(server.price, "fetch_historical", lambda: []) + monkeypatch.setattr(server.engine, "score_all", lambda metrics: {"composite_score": 50, "scored_count": 1, "metrics": []}) + monkeypatch.setattr(server.engine, "score_all_ml", lambda metrics: {"composite_score": 50, "scored_count": 1, "metrics": []}) + + calls = [] + fake_lib = types.ModuleType("scrapers.lookintobitcoin") + fake_lib.scrape_all = lambda: calls.append("lib") or {"puell_multiple": {"value": 0.8}} + fake_coc = types.ModuleType("scrapers.checkonchain") + fake_coc.scrape_all = lambda: calls.append("coc") or {"sopr": {"value": 0.99}} + fake_updater = types.ModuleType("scrapers.history_updater") + fake_updater.update_history = lambda: None + import scrapers + monkeypatch.setattr(scrapers, "lookintobitcoin", fake_lib, raising=False) + monkeypatch.setattr(scrapers, "checkonchain", fake_coc, raising=False) + monkeypatch.setitem(sys.modules, "scrapers.lookintobitcoin", fake_lib) + monkeypatch.setitem(sys.modules, "scrapers.checkonchain", fake_coc) + monkeypatch.setitem(sys.modules, "scrapers.history_updater", fake_updater) + + server.run_scrape() + + assert calls == ["lib", "coc"] + saved = json.loads(cache_path.read_text()) + assert saved["puell_multiple"]["value"] == 0.8 + assert saved["puell_multiple"]["source"] == "lookintobitcoin" + assert saved["sopr"]["source"] == "checkonchain" + assert saved["_onchain_timestamp"] != old.isoformat()