storage/database.py

"""
SQLite-backed cache for articles and narrative snapshots.
Prevents re-processing the same URLs and stores historical trend data.
"""

import sqlite3
import json
import logging
from datetime import datetime, timezone, timedelta
from pathlib import Path

from config import DB_PATH, MAX_ARTICLE_AGE_HOURS

logger = logging.getLogger(__name__)

_SCHEMA = """
CREATE TABLE IF NOT EXISTS articles (
    url         TEXT PRIMARY KEY,
    source      TEXT,
    title       TEXT,
    summary     TEXT,
    published   TEXT,
    type        TEXT,
    score       REAL,
    text        TEXT,
    sentiment   TEXT,     -- JSON
    narratives  TEXT,     -- JSON list
    entities    TEXT,     -- JSON
    fetched_at  TEXT
);

CREATE TABLE IF NOT EXISTS narrative_snapshots (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    snapshot_at TEXT,
    data        TEXT      -- JSON list of scored narratives
);

CREATE TABLE IF NOT EXISTS fear_greed (
    timestamp   TEXT PRIMARY KEY,
    value       INTEGER,
    label       TEXT
);

CREATE TABLE IF NOT EXISTS coin_prices (
    id          TEXT,
    snapshot_at TEXT,
    data        TEXT,     -- JSON
    PRIMARY KEY (id, snapshot_at)
);

CREATE TABLE IF NOT EXISTS sentiment_history (
    snapshot_at TEXT PRIMARY KEY,
    net_score   REAL,
    index_val   INTEGER,
    label       TEXT,
    bullish     INTEGER,
    bearish     INTEGER,
    neutral     INTEGER,
    per_narrative TEXT,   -- JSON: {narrative: net_score}
    bull_prob   REAL,     -- #7 probabilistic outputs
    bear_prob   REAL,
    vol_prob    REAL,
    confidence  REAL
);

-- #1 Market reaction labeling: forward BTC/ETH moves after each story
CREATE TABLE IF NOT EXISTS market_reactions (
    url         TEXT PRIMARY KEY,
    published   TEXT,
    sentiment   TEXT,      -- label at publish time (for later analysis)
    btc_at_pub  REAL,
    eth_at_pub  REAL,
    btc_1h REAL, btc_4h REAL, btc_24h REAL,
    eth_1h REAL, eth_4h REAL, eth_24h REAL,
    label_24h   TEXT,      -- 'bullish'|'bearish'|'flat' from BTC 24h move
    updated_at  TEXT
);

-- #5 On-chain + derivatives signals
CREATE TABLE IF NOT EXISTS onchain_signals (
    snapshot_at      TEXT PRIMARY KEY,
    funding_btc      REAL,
    funding_eth      REAL,
    oi_btc           REAL,
    oi_btc_change    REAL,
    liq_proxy        REAL,
    stablecoin_flow  REAL,
    data             TEXT   -- JSON raw
);

-- #3 Per-story consensus across duplicate sources
CREATE TABLE IF NOT EXISTS story_consensus (
    url               TEXT PRIMARY KEY,
    cluster_size      INTEGER,
    sentiment_variance REAL,
    consensus_strength REAL,
    snapshot_at       TEXT
);

-- #6 Social attention velocity: mention counts over time
CREATE TABLE IF NOT EXISTS mention_history (
    snapshot_at TEXT,
    entity      TEXT,
    kind        TEXT,   -- 'coin' | 'narrative'
    count       INTEGER,
    PRIMARY KEY (snapshot_at, entity, kind)
);

-- Magnet track record: snapshot the liquidation magnets, evaluate how price
-- actually behaved toward them after a horizon.
CREATE TABLE IF NOT EXISTS magnet_snapshots (
    snapshot_at   TEXT PRIMARY KEY,
    price         REAL,
    near_price    REAL, near_strength INTEGER, near_side TEXT, near_dist REAL,
    strong_price  REAL, strong_strength INTEGER, strong_side TEXT, strong_dist REAL,
    reached_near  INTEGER, reached_strong INTEGER,
    end_price     REAL, evaluated INTEGER DEFAULT 0, updated_at TEXT
);

-- Magnet target registry: each distinct magnet level is registered once and
-- tracked for 24h; we record if/when price hit it (never re-added while active).
CREATE TABLE IF NOT EXISTS magnet_targets (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    first_seen    TEXT,
    ref_price     REAL,
    price_level   REAL,
    side          TEXT,        -- above | below
    strength      INTEGER,     -- 0-100 at registration
    dist_pct      REAL,        -- distance from price at registration
    is_strongest  INTEGER DEFAULT 0,
    is_nearest_strong INTEGER DEFAULT 0,
    hit           INTEGER DEFAULT 0,
    hit_at        TEXT,
    time_to_hit   REAL,        -- hours from first_seen to first touch
    expired       INTEGER DEFAULT 0
);

-- #8 System health dataset
CREATE TABLE IF NOT EXISTS system_health (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    ts         TEXT,
    component  TEXT,    -- 'source' | 'model' | 'pipeline'
    event_type TEXT,    -- 'source_failure'|'model_fallback'|'disagreement'|'component_failure'|'ok'
    name       TEXT,    -- source/model/component name
    detail     TEXT,
    value      REAL
);

CREATE TABLE IF NOT EXISTS clients (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    email     TEXT UNIQUE,
    name      TEXT,
    active    INTEGER DEFAULT 1,
    added_at  TEXT
);

CREATE TABLE IF NOT EXISTS email_settings (
    id        INTEGER PRIMARY KEY CHECK (id = 1),
    enabled   INTEGER DEFAULT 0,
    send_time_1 TEXT DEFAULT '09:00',   -- UTC HH:MM
    send_time_2 TEXT DEFAULT '21:00',
    subject_prefix TEXT DEFAULT 'Crypto Narrative Brief',
    last_sent_slot TEXT DEFAULT ''       -- 'YYYY-MM-DD#1' to avoid double-send
);

CREATE TABLE IF NOT EXISTS email_log (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    sent_at    TEXT,
    slot       TEXT,
    recipients INTEGER,
    status     TEXT,
    detail     TEXT
);
"""


class Database:
    def __init__(self, path: str = DB_PATH):
        self.path = path
        try:
            # Make sure the parent directory exists and is writable
            parent = Path(path).resolve().parent
            parent.mkdir(parents=True, exist_ok=True)
            self._conn = sqlite3.connect(path, check_same_thread=False)
        except (sqlite3.OperationalError, OSError) as exc:
            # Fall back to a guaranteed-writable temp location (e.g. read-only
            # working dir on some cloud hosts). History won't persist across
            # restarts, but the app stays up.
            import tempfile
            fallback = str(Path(tempfile.gettempdir()) / "narrative_analysis.db")
            logger.warning(f"[DB] Could not open {path} ({exc}); using {fallback}")
            self.path = fallback
            self._conn = sqlite3.connect(fallback, check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        self._conn.executescript(_SCHEMA)
        self._migrate()
        self._conn.commit()

    def _migrate(self) -> None:
        """Add columns introduced after a table was first created.
        CREATE TABLE IF NOT EXISTS won't alter an existing table, so we add
        any missing columns here (safe, idempotent)."""
        wanted = {
            "sentiment_history": [
                ("bull_prob", "REAL"), ("bear_prob", "REAL"),
                ("vol_prob", "REAL"), ("confidence", "REAL"),
            ],
            "clients": [
                ("confirmed", "INTEGER DEFAULT 0"),   # double opt-in state
                ("token", "TEXT"),                    # confirm / unsubscribe token
                ("source", "TEXT"),                   # where they signed up
                ("ref_code", "TEXT"),                 # this subscriber's own referral code
                ("referred_by", "TEXT"),              # ref_code of whoever referred them
                ("referrals", "INTEGER DEFAULT 0"),   # how many they've referred (confirmed)
            ],
        }
        for table, cols in wanted.items():
            try:
                existing = {r["name"] for r in self._conn.execute(f"PRAGMA table_info({table})")}
            except Exception:
                continue
            for name, typ in cols:
                if name not in existing:
                    try:
                        self._conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {typ}")
                    except Exception as exc:
                        logger.debug(f"[DB] migrate {table}.{name}: {exc}")

    # ------------------------------------------------------------------
    # Articles
    # ------------------------------------------------------------------

    def save_articles(self, articles: list[dict]) -> int:
        """Upserts articles. Returns number inserted."""
        inserted = 0
        now = datetime.now(timezone.utc).isoformat()
        for a in articles:
            url = a.get("url", "")
            if not url:
                continue
            try:
                self._conn.execute(
                    """
                    INSERT OR REPLACE INTO articles
                      (url, source, title, summary, published, type, score,
                       text, sentiment, narratives, entities, fetched_at)
                    VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
                    """,
                    (
                        url,
                        a.get("source", ""),
                        a.get("title", ""),
                        a.get("summary", ""),
                        a.get("published", ""),
                        a.get("type", "news"),
                        a.get("score", 0),
                        a.get("text", ""),
                        json.dumps(a.get("sentiment", {})),
                        json.dumps(a.get("narratives", [])),
                        json.dumps(a.get("entities", {})),
                        now,
                    ),
                )
                inserted += 1
            except Exception as exc:
                logger.debug(f"DB insert failed for {url}: {exc}")
        self._conn.commit()
        return inserted

    def get_recent_articles(self, hours: int = MAX_ARTICLE_AGE_HOURS) -> list[dict]:
        cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
        rows = self._conn.execute(
            "SELECT * FROM articles WHERE published >= ? ORDER BY published DESC",
            (cutoff,),
        ).fetchall()
        return [self._row_to_article(r) for r in rows]

    def article_count(self) -> int:
        return self._conn.execute("SELECT COUNT(*) FROM articles").fetchone()[0]

    def get_article(self, url: str) -> dict | None:
        row = self._conn.execute("SELECT * FROM articles WHERE url=?", (url,)).fetchone()
        return self._row_to_article(row) if row else None

    def snapshot_to(self, path: str) -> bool:
        """Write a clean, consistent copy of the DB to `path` (for backup).
        Uses SQLite's online backup API — safe on a live connection and, unlike
        VACUUM, does NOT touch the main connection's transaction state."""
        import os
        try:
            if os.path.exists(path):
                os.remove(path)
            dest = sqlite3.connect(path)
            try:
                self._conn.backup(dest)   # read-only on source; no COMMIT issued
            finally:
                dest.close()
            return True
        except Exception as exc:
            logger.warning(f"[DB] snapshot failed: {exc}")
            return False

    @staticmethod
    def _row_to_article(row: sqlite3.Row) -> dict:
        d = dict(row)
        for key in ("sentiment", "narratives", "entities"):
            if d.get(key):
                try:
                    d[key] = json.loads(d[key])
                except Exception:
                    pass
        # Re-derive the macro flag from the source (not stored as a column)
        d["is_macro"] = str(d.get("source", "")).startswith("Macro")
        return d

    # ------------------------------------------------------------------
    # #1 Market reactions
    # ------------------------------------------------------------------

    def upsert_reaction_seed(self, url: str, published: str, sentiment: str,
                             btc: float | None, eth: float | None) -> None:
        """Record the publish-time price reference for a story (once)."""
        if not url:
            return
        now = datetime.now(timezone.utc).isoformat()
        self._conn.execute(
            """INSERT OR IGNORE INTO market_reactions
               (url, published, sentiment, btc_at_pub, eth_at_pub, updated_at)
               VALUES (?,?,?,?,?,?)""",
            (url, published, sentiment, btc, eth, now),
        )
        self._conn.commit()

    def reactions_needing_fill(self) -> list[dict]:
        rows = self._conn.execute(
            "SELECT * FROM market_reactions WHERE btc_24h IS NULL"
        ).fetchall()
        return [dict(r) for r in rows]

    def update_reaction(self, url: str, fields: dict) -> None:
        if not fields:
            return
        fields["updated_at"] = datetime.now(timezone.utc).isoformat()
        cols = ", ".join(f"{k}=?" for k in fields)
        self._conn.execute(f"UPDATE market_reactions SET {cols} WHERE url=?",
                           (*fields.values(), url))
        self._conn.commit()

    def labeled_reactions(self, limit: int = 5000) -> list[dict]:
        """Stories that now have a 24h BTC label — training data for #2."""
        rows = self._conn.execute(
            "SELECT * FROM market_reactions WHERE label_24h IS NOT NULL ORDER BY published DESC LIMIT ?",
            (limit,),
        ).fetchall()
        return [dict(r) for r in rows]

    # ------------------------------------------------------------------
    # #5 On-chain / derivatives
    # ------------------------------------------------------------------

    def save_onchain(self, sig: dict) -> None:
        now = datetime.now(timezone.utc).isoformat()
        self._conn.execute(
            """INSERT OR REPLACE INTO onchain_signals
               (snapshot_at, funding_btc, funding_eth, oi_btc, oi_btc_change,
                liq_proxy, stablecoin_flow, data) VALUES (?,?,?,?,?,?,?,?)""",
            (now, sig.get("funding_btc"), sig.get("funding_eth"), sig.get("oi_btc"),
             sig.get("oi_btc_change"), sig.get("liq_proxy"), sig.get("stablecoin_flow"),
             json.dumps(sig)),
        )
        self._conn.commit()

    def last_onchain(self) -> dict | None:
        row = self._conn.execute(
            "SELECT * FROM onchain_signals ORDER BY snapshot_at DESC LIMIT 1"
        ).fetchone()
        return dict(row) if row else None

    def prev_oi_btc(self) -> float | None:
        row = self._conn.execute(
            "SELECT oi_btc FROM onchain_signals WHERE oi_btc IS NOT NULL ORDER BY snapshot_at DESC LIMIT 1"
        ).fetchone()
        return row["oi_btc"] if row else None

    # ------------------------------------------------------------------
    # #3 Story consensus
    # ------------------------------------------------------------------

    def save_consensus(self, url: str, cluster_size: int, variance: float, strength: float) -> None:
        if not url:
            return
        self._conn.execute(
            """INSERT OR REPLACE INTO story_consensus
               (url, cluster_size, sentiment_variance, consensus_strength, snapshot_at)
               VALUES (?,?,?,?,?)""",
            (url, cluster_size, variance, strength, datetime.now(timezone.utc).isoformat()),
        )
        self._conn.commit()

    # ------------------------------------------------------------------
    # #6 Mention history (velocity)
    # ------------------------------------------------------------------

    def save_mentions(self, mentions: dict[str, int], kind: str = "coin") -> None:
        now = datetime.now(timezone.utc).isoformat()
        for entity, count in mentions.items():
            self._conn.execute(
                "INSERT OR REPLACE INTO mention_history (snapshot_at, entity, kind, count) VALUES (?,?,?,?)",
                (now, entity, kind, count),
            )
        self._conn.commit()

    def mention_series(self, kind: str = "coin", limit: int = 400) -> list[dict]:
        rows = self._conn.execute(
            "SELECT snapshot_at, entity, count FROM mention_history WHERE kind=? ORDER BY snapshot_at DESC LIMIT ?",
            (kind, limit),
        ).fetchall()
        return [dict(r) for r in rows]

    # ------------------------------------------------------------------
    # Magnet track record
    # ------------------------------------------------------------------

    def save_magnet_seed(self, liq: dict) -> None:
        near = liq.get("nearest_magnet")
        strong = liq.get("strongest_magnet")
        price = liq.get("current_price")
        if not price or not near or not strong:
            return
        now = datetime.now(timezone.utc).isoformat()
        self._conn.execute(
            """INSERT OR IGNORE INTO magnet_snapshots
               (snapshot_at, price, near_price, near_strength, near_side, near_dist,
                strong_price, strong_strength, strong_side, strong_dist, updated_at)
               VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
            (now, price, near["price"], near["strength"], near["side"], near["dist_pct"],
             strong["price"], strong["strength"], strong["side"], strong["dist_pct"], now),
        )
        self._conn.commit()

    def get_magnet_strength_history(self, limit: int = 2000) -> list[dict]:
        """[{snapshot_at, strength}] — strongest liquidation magnet (0-100) per
        cycle, chronological. Used to mark high-liquidation points on the chart."""
        rows = self._conn.execute(
            "SELECT snapshot_at, strong_strength FROM magnet_snapshots "
            "ORDER BY snapshot_at DESC LIMIT ?", (limit,)
        ).fetchall()
        return [{"snapshot_at": r["snapshot_at"], "strength": r["strong_strength"]}
                for r in reversed(rows) if r["strong_strength"] is not None]

    def magnets_needing_eval(self) -> list[dict]:
        return [dict(r) for r in self._conn.execute(
            "SELECT * FROM magnet_snapshots WHERE evaluated=0"
        ).fetchall()]

    def update_magnet_eval(self, snapshot_at: str, fields: dict) -> None:
        fields["evaluated"] = 1
        fields["updated_at"] = datetime.now(timezone.utc).isoformat()
        cols = ", ".join(f"{k}=?" for k in fields)
        self._conn.execute(f"UPDATE magnet_snapshots SET {cols} WHERE snapshot_at=?",
                           (*fields.values(), snapshot_at))
        self._conn.commit()

    def evaluated_magnets(self, limit: int = 5000) -> list[dict]:
        return [dict(r) for r in self._conn.execute(
            "SELECT * FROM magnet_snapshots WHERE evaluated=1 ORDER BY snapshot_at ASC LIMIT ?",
            (limit,),
        ).fetchall()]

    # -- magnet target registry (24h tracking) -------------------------
    def add_magnet_target(self, t: dict) -> None:
        self._conn.execute(
            """INSERT INTO magnet_targets
               (first_seen, ref_price, price_level, side, strength, dist_pct,
                is_strongest, is_nearest_strong)
               VALUES (?,?,?,?,?,?,?,?)""",
            (t["first_seen"], t["ref_price"], t["price_level"], t["side"],
             t["strength"], t["dist_pct"], t.get("is_strongest", 0),
             t.get("is_nearest_strong", 0)),
        )
        self._conn.commit()

    def open_magnet_targets(self) -> list[dict]:
        """Targets still within their 24h window (not expired)."""
        return [dict(r) for r in self._conn.execute(
            "SELECT * FROM magnet_targets WHERE expired=0"
        ).fetchall()]

    def mark_magnet_hit(self, tid: int, hit_at: str, tth: float) -> None:
        self._conn.execute(
            "UPDATE magnet_targets SET hit=1, hit_at=?, time_to_hit=? WHERE id=?",
            (hit_at, tth, tid))
        self._conn.commit()

    def expire_magnet_target(self, tid: int) -> None:
        self._conn.execute("UPDATE magnet_targets SET expired=1 WHERE id=?", (tid,))
        self._conn.commit()

    def set_magnet_flags(self, tid: int, strongest: int, nearest_strong: int) -> None:
        # only ever set to 1 (a level that was ever strongest stays flagged)
        self._conn.execute(
            """UPDATE magnet_targets
               SET is_strongest = MAX(is_strongest, ?),
                   is_nearest_strong = MAX(is_nearest_strong, ?)
               WHERE id=?""", (strongest, nearest_strong, tid))
        self._conn.commit()

    def resolved_magnet_targets(self, limit: int = 8000) -> list[dict]:
        """Targets whose outcome is known: hit, or expired without a hit."""
        return [dict(r) for r in self._conn.execute(
            "SELECT * FROM magnet_targets WHERE hit=1 OR expired=1 ORDER BY id DESC LIMIT ?",
            (limit,)).fetchall()]

    def magnet_target_active_count(self) -> int:
        return self._conn.execute(
            "SELECT COUNT(*) FROM magnet_targets WHERE expired=0 AND hit=0").fetchone()[0]

    # ------------------------------------------------------------------
    # #8 System health
    # ------------------------------------------------------------------

    def log_health(self, component: str, event_type: str, name: str = "",
                   detail: str = "", value: float | None = None) -> None:
        try:
            self._conn.execute(
                "INSERT INTO system_health (ts, component, event_type, name, detail, value) VALUES (?,?,?,?,?,?)",
                (datetime.now(timezone.utc).isoformat(), component, event_type, name, detail, value),
            )
            self._conn.commit()
        except Exception:
            pass

    def health_summary(self, hours: int = 24) -> dict:
        cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
        rows = self._conn.execute(
            "SELECT event_type, COUNT(*) c FROM system_health WHERE ts >= ? GROUP BY event_type",
            (cutoff,),
        ).fetchall()
        by_type = {r["event_type"]: r["c"] for r in rows}
        recent = self._conn.execute(
            "SELECT * FROM system_health ORDER BY id DESC LIMIT 30"
        ).fetchall()
        return {"by_type": by_type, "recent": [dict(r) for r in recent]}

    # ------------------------------------------------------------------
    # Narrative snapshots
    # ------------------------------------------------------------------

    def save_snapshot(self, scored_narratives: list[dict]) -> None:
        now = datetime.now(timezone.utc).isoformat()
        self._conn.execute(
            "INSERT INTO narrative_snapshots (snapshot_at, data) VALUES (?, ?)",
            (now, json.dumps(scored_narratives)),
        )
        self._conn.commit()

    def get_snapshots(self, limit: int = 48) -> list[dict]:
        rows = self._conn.execute(
            "SELECT snapshot_at, data FROM narrative_snapshots ORDER BY id DESC LIMIT ?",
            (limit,),
        ).fetchall()
        return [{"snapshot_at": r["snapshot_at"], "data": json.loads(r["data"])} for r in rows]

    # ------------------------------------------------------------------
    # Sentiment history (for charting + spike detection)
    # ------------------------------------------------------------------

    def save_sentiment_history(self, sentiment_24h: dict) -> None:
        o = sentiment_24h.get("overall", {})
        per = {n["name"]: n["net_score"] for n in sentiment_24h.get("per_narrative", [])}
        now = datetime.now(timezone.utc).isoformat()
        self._conn.execute(
            """INSERT OR REPLACE INTO sentiment_history
               (snapshot_at, net_score, index_val, label, bullish, bearish, neutral,
                per_narrative, bull_prob, bear_prob, vol_prob, confidence)
               VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
            (now, o.get("net_score", 0), o.get("index", 50), o.get("label", "NEUTRAL"),
             o.get("bullish", 0), o.get("bearish", 0), o.get("neutral", 0), json.dumps(per),
             o.get("bull_prob"), o.get("bear_prob"), o.get("vol_prob"), o.get("confidence")),
        )
        self._conn.commit()

    def get_sentiment_history(self, limit: int = 96) -> list[dict]:
        rows = self._conn.execute(
            "SELECT * FROM sentiment_history ORDER BY snapshot_at DESC LIMIT ?", (limit,)
        ).fetchall()
        out = []
        for r in reversed(rows):   # chronological order for charting
            d = dict(r)
            try:
                d["per_narrative"] = json.loads(d["per_narrative"])
            except Exception:
                d["per_narrative"] = {}
            out.append(d)
        return out

    def sentiment_span(self) -> dict:
        """How much sentiment history we hold: count + first/last + days span.
        Used to show how many days the relative-index baseline is built on."""
        row = self._conn.execute(
            "SELECT COUNT(*) AS n, MIN(snapshot_at) AS first, MAX(snapshot_at) AS last "
            "FROM sentiment_history"
        ).fetchone()
        n = int(row["n"]) if row and row["n"] else 0
        first, last = (row["first"], row["last"]) if row else (None, None)
        days = None
        if first and last:
            try:
                f = datetime.fromisoformat(first)
                l = datetime.fromisoformat(last)
                days = round((l - f).total_seconds() / 86400, 1)
            except Exception:
                days = None
        return {"count": n, "first": first, "last": last, "days": days}

    def get_index_near(self, hours_ago: float, tolerance_hours: float = 8.0) -> dict | None:
        """
        Return the recorded sentiment snapshot closest to `hours_ago` hours in
        the past, if one exists within `tolerance_hours`. Used for a reliable
        'vs ~24h ago' comparison instead of recomputing from stale articles.
        """
        target = datetime.now(timezone.utc) - timedelta(hours=hours_ago)
        target_iso = target.isoformat()
        row = self._conn.execute(
            """SELECT *,
                      ABS(strftime('%s', snapshot_at) - strftime('%s', ?)) AS diff
               FROM sentiment_history
               ORDER BY diff ASC LIMIT 1""",
            (target_iso,),
        ).fetchone()
        if not row:
            return None
        d = dict(row)
        if d.get("diff") is not None and d["diff"] > tolerance_hours * 3600:
            return None
        return d

    # ------------------------------------------------------------------
    # Fear & Greed
    # ------------------------------------------------------------------

    def save_fear_greed(self, entries: list[dict]) -> None:
        for e in entries:
            self._conn.execute(
                "INSERT OR REPLACE INTO fear_greed (timestamp, value, label) VALUES (?,?,?)",
                (e["timestamp"], e["value"], e["label"]),
            )
        self._conn.commit()

    def get_fear_greed(self, limit: int = 7) -> list[dict]:
        rows = self._conn.execute(
            "SELECT * FROM fear_greed ORDER BY timestamp DESC LIMIT ?", (limit,)
        ).fetchall()
        return [dict(r) for r in rows]

    # ------------------------------------------------------------------
    # Coin prices
    # ------------------------------------------------------------------

    def save_prices(self, prices: list[dict]) -> None:
        now = datetime.now(timezone.utc).isoformat()
        for p in prices:
            self._conn.execute(
                "INSERT OR REPLACE INTO coin_prices (id, snapshot_at, data) VALUES (?,?,?)",
                (p.get("id", ""), now, json.dumps(p)),
            )
        self._conn.commit()

    def get_price_path(self, coin_id: str, t0_iso: str, t1_iso: str) -> list[float]:
        """All recorded prices for a coin between two timestamps (for extremes)."""
        rows = self._conn.execute(
            "SELECT data FROM coin_prices WHERE id=? AND snapshot_at>=? AND snapshot_at<=?",
            (coin_id, t0_iso, t1_iso),
        ).fetchall()
        out = []
        for r in rows:
            try:
                p = json.loads(r["data"]).get("price_usd")
                if p:
                    out.append(p)
            except Exception:
                pass
        return out

    def get_price_history(self, coin_id: str = "bitcoin", limit: int = 200) -> list[dict]:
        """Returns [{snapshot_at, price}] for a coin, chronological order."""
        rows = self._conn.execute(
            "SELECT snapshot_at, data FROM coin_prices WHERE id = ? ORDER BY snapshot_at DESC LIMIT ?",
            (coin_id, limit),
        ).fetchall()
        out = []
        for r in reversed(rows):
            try:
                price = json.loads(r["data"]).get("price_usd")
            except Exception:
                price = None
            if price is not None:
                out.append({"snapshot_at": r["snapshot_at"], "price": price})
        return out

    # ------------------------------------------------------------------
    # Clients & email settings (admin panel)
    # ------------------------------------------------------------------

    def _ensure_email_settings(self) -> None:
        self._conn.execute("INSERT OR IGNORE INTO email_settings (id) VALUES (1)")
        self._conn.commit()

    def get_email_settings(self) -> dict:
        self._ensure_email_settings()
        row = self._conn.execute("SELECT * FROM email_settings WHERE id = 1").fetchone()
        return dict(row) if row else {}

    def update_email_settings(self, **fields) -> None:
        self._ensure_email_settings()
        allowed = {"enabled", "send_time_1", "send_time_2", "subject_prefix", "last_sent_slot"}
        sets = {k: v for k, v in fields.items() if k in allowed}
        if not sets:
            return
        cols = ", ".join(f"{k} = ?" for k in sets)
        self._conn.execute(f"UPDATE email_settings SET {cols} WHERE id = 1", tuple(sets.values()))
        self._conn.commit()

    def add_client(self, email: str, name: str = "") -> bool:
        """Admin-added client: trusted, so active + confirmed immediately."""
        try:
            self._conn.execute(
                "INSERT INTO clients (email, name, active, confirmed, added_at, source) "
                "VALUES (?,?,1,1,?,?)",
                (email.strip().lower(), name.strip(),
                 datetime.now(timezone.utc).isoformat(), "admin"),
            )
            self._conn.commit()
            return True
        except sqlite3.IntegrityError:
            return False

    def add_pending_subscriber(self, email: str, token: str, source: str = "web",
                               ref_code: str = "", referred_by: str = "") -> str:
        """Self-serve signup (double opt-in). Returns:
        'pending'   -> new pending row (send confirmation)
        'resend'    -> existed but unconfirmed; token refreshed (resend confirmation)
        'exists'    -> already confirmed & active (no email needed)
        """
        email = email.strip().lower()
        now = datetime.now(timezone.utc).isoformat()
        row = self._conn.execute(
            "SELECT id, confirmed, active FROM clients WHERE email = ?", (email,)
        ).fetchone()
        if row is None:
            self._conn.execute(
                "INSERT INTO clients (email, name, active, confirmed, added_at, token, "
                "source, ref_code, referred_by, referrals) VALUES (?,?,0,0,?,?,?,?,?,0)",
                (email, "", now, token, source, ref_code, referred_by or None),
            )
            self._conn.commit()
            return "pending"
        if row["confirmed"] and row["active"]:
            return "exists"
        # existed but not confirmed (or unsubscribed) -> refresh token, resend
        self._conn.execute(
            "UPDATE clients SET token = ?, confirmed = 0, active = 0 WHERE id = ?",
            (token, row["id"]),
        )
        self._conn.commit()
        return "resend"

    def confirm_subscriber(self, token: str) -> dict | None:
        """Activate a subscriber by their confirm token. Credits the referrer on
        the FIRST confirmation only. Returns {email, ref_code, referrals} or None."""
        if not token:
            return None
        row = self._conn.execute(
            "SELECT id, email, confirmed, ref_code, referred_by FROM clients WHERE token = ?",
            (token,),
        ).fetchone()
        if row is None:
            return None
        first_time = not row["confirmed"]
        self._conn.execute(
            "UPDATE clients SET active = 1, confirmed = 1 WHERE id = ?", (row["id"],)
        )
        if first_time and row["referred_by"]:
            self._conn.execute(
                "UPDATE clients SET referrals = referrals + 1 WHERE ref_code = ?",
                (row["referred_by"],),
            )
        self._conn.commit()
        refs = self._conn.execute(
            "SELECT referrals FROM clients WHERE id = ?", (row["id"],)
        ).fetchone()
        return {"email": row["email"], "ref_code": row["ref_code"] or "",
                "referrals": int(refs["referrals"]) if refs else 0}

    def top_referrers(self, limit: int = 3) -> list[dict]:
        """Confirmed subscribers with the most referrals (referrals > 0)."""
        rows = self._conn.execute(
            "SELECT email, referrals FROM clients "
            "WHERE confirmed = 1 AND referrals > 0 "
            "ORDER BY referrals DESC, id ASC LIMIT ?",
            (limit,),
        ).fetchall()
        return [{"email": r["email"], "referrals": int(r["referrals"])} for r in rows]

    def referral_count(self, ref_code: str) -> int | None:
        """Live referral count for a ref_code, or None if the code is unknown."""
        if not ref_code:
            return None
        row = self._conn.execute(
            "SELECT referrals FROM clients WHERE ref_code = ?", (ref_code,)
        ).fetchone()
        return int(row["referrals"]) if row else None

    def unsubscribe_email(self, email: str) -> bool:
        """Deactivate a subscriber by email (keeps the row). Returns True if found."""
        cur = self._conn.execute(
            "UPDATE clients SET active = 0 WHERE email = ?", (email.strip().lower(),)
        )
        self._conn.commit()
        return cur.rowcount > 0

    def remove_client(self, client_id: int) -> None:
        self._conn.execute("DELETE FROM clients WHERE id = ?", (client_id,))
        self._conn.commit()

    def set_client_active(self, client_id: int, active: bool) -> None:
        self._conn.execute("UPDATE clients SET active = ? WHERE id = ?", (1 if active else 0, client_id))
        self._conn.commit()

    def subscriber_count(self, confirmed_only: bool = True) -> int:
        q = "SELECT COUNT(*) AS n FROM clients"
        if confirmed_only:
            q += " WHERE confirmed = 1"
        row = self._conn.execute(q).fetchone()
        return int(row["n"]) if row else 0

    def get_clients(self, active_only: bool = False) -> list[dict]:
        q = "SELECT * FROM clients"
        if active_only:
            q += " WHERE active = 1"
        q += " ORDER BY added_at DESC"
        return [dict(r) for r in self._conn.execute(q).fetchall()]

    def log_email(self, slot: str, recipients: int, status: str, detail: str = "") -> None:
        self._conn.execute(
            "INSERT INTO email_log (sent_at, slot, recipients, status, detail) VALUES (?,?,?,?,?)",
            (datetime.now(timezone.utc).isoformat(), slot, recipients, status, detail),
        )
        self._conn.commit()

    def get_email_log(self, limit: int = 30) -> list[dict]:
        return [dict(r) for r in self._conn.execute(
            "SELECT * FROM email_log ORDER BY id DESC LIMIT ?", (limit,)
        ).fetchall()]

    # ------------------------------------------------------------------
    # Maintenance
    # ------------------------------------------------------------------

    def purge_old(self) -> None:
        """Remove articles older than MAX_ARTICLE_AGE_HOURS, and cap the growth
        of the high-volume time-series tables (they were never purged, so the
        DB grew unbounded and eventually strained the small Space)."""
        now = datetime.now(timezone.utc)
        art_cutoff = (now - timedelta(hours=MAX_ARTICLE_AGE_HOURS)).isoformat()
        self._conn.execute("DELETE FROM articles WHERE published < ?", (art_cutoff,))

        # Keep ~60 days of time-series so the track records stay intact.
        ts_cutoff = (now - timedelta(days=60)).isoformat()
        for tbl, col in (("coin_prices", "snapshot_at"),
                         ("mention_history", "snapshot_at"),
                         ("sentiment_history", "snapshot_at"),
                         ("onchain_signals", "snapshot_at")):
            try:
                self._conn.execute(f"DELETE FROM {tbl} WHERE {col} < ?", (ts_cutoff,))
            except Exception:
                pass
        # Drop resolved magnet targets older than 60 days (stats already counted)
        try:
            self._conn.execute(
                "DELETE FROM magnet_targets WHERE (hit=1 OR expired=1) AND first_seen < ?",
                (ts_cutoff,))
        except Exception:
            pass
        self._conn.commit()

    def close(self) -> None:
        self._conn.close()