"""Signal store: manifest, coverage queries, and append-only Hub writes.

The store holds *raw model outputs and prices only* -- never trade decisions.
Trading rules, costs and sizing are applied live by engine.py per request.

Layout inside the dataset repo:

    manifest.json
    signals/{model_slug}/{asset}/{timeframe}/{year}.parquet
    prices/{asset}/{timeframe}/{year}.parquet
    comparisons/*.parquet | *.json
    runs/{run_id}.json

Writes are staged into a local mirror directory and pushed as a *single*
atomic commit per flush, with the manifest as the last operation in the
commit. A CommitScheduler can be attached to batch flushes in the running
Space (see `attach_scheduler`).
"""

from __future__ import annotations

import json
import logging
import os
import threading
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, Literal

import pandas as pd

from . import config

log = logging.getLogger("bit.store")

# --------------------------------------------------------------------------
# Errors
# --------------------------------------------------------------------------


class StoreError(RuntimeError):
    """Raised when the store is asked to do something inconsistent."""


class SchemaError(StoreError):
    """Raised when a manifest or parquet slice fails schema validation."""


# --------------------------------------------------------------------------
# Schema
# --------------------------------------------------------------------------

SIGNAL_COLUMNS_QUANTILE = ["ts", "q10", "q50", "q90", "context_len", "inference_version"]
SIGNAL_COLUMNS_CLASSIFIER = ["ts", "pred", "confidence", "context_len", "inference_version"]
PRICE_COLUMNS = ["ts", "open", "high", "low", "close", "volume", "source"]

SignalKind = Literal["quantile", "classifier"]


def _utc(ts) -> pd.Timestamp:
    """Coerce anything timestamp-ish to a UTC-aware pandas Timestamp."""
    t = pd.Timestamp(ts)
    return t.tz_localize("UTC") if t.tzinfo is None else t.tz_convert("UTC")


def _iso(ts) -> str:
    return _utc(ts).isoformat().replace("+00:00", "Z")


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")


def signal_key(model_slug: str, model_revision: str, asset: str, timeframe: str) -> str:
    """Canonical manifest key. Revision is part of the identity on purpose:
    a different model revision is a different signal series."""
    return f"{model_slug}@{model_revision}|{asset}|{timeframe}"


def price_key(asset: str, timeframe: str) -> str:
    return f"{asset}|{timeframe}"


@dataclass
class CoverageEntry:
    """One (model, revision, asset, timeframe) coverage record."""

    model_slug: str
    model_id: str
    model_revision: str
    asset: str
    timeframe: str
    start_ts: str
    end_ts: str
    rows: int
    inference_version: str
    last_updated: str
    contributed_by: str
    signal_kind: SignalKind = "quantile"

    @property
    def key(self) -> str:
        return signal_key(self.model_slug, self.model_revision, self.asset, self.timeframe)

    @property
    def is_placeholder(self) -> bool:
        return self.inference_version == config.PLACEHOLDER_VERSION

    def validate(self) -> None:
        for f in ("model_slug", "model_id", "model_revision", "asset",
                  "timeframe", "inference_version", "contributed_by"):
            if not getattr(self, f):
                raise SchemaError(f"CoverageEntry.{f} must be non-empty")
        if self.timeframe not in config.TIMEFRAMES:
            raise SchemaError(f"unknown timeframe {self.timeframe!r}")
        if self.rows < 0:
            raise SchemaError("CoverageEntry.rows must be >= 0")
        if _utc(self.start_ts) > _utc(self.end_ts):
            raise SchemaError(f"start_ts after end_ts for {self.key}")
        if self.signal_kind not in ("quantile", "classifier"):
            raise SchemaError(f"bad signal_kind {self.signal_kind!r}")

    @classmethod
    def from_dict(cls, d: dict) -> "CoverageEntry":
        known = {k: v for k, v in d.items() if k in cls.__dataclass_fields__}
        return cls(**known)


@dataclass
class PriceCoverage:
    """Cached OHLCV coverage for one (asset, timeframe).

    `provider_max_days` records an honest provider depth boundary (e.g. Yahoo
    only serves ~730d of 1h bars). It is a fact about coverage, not an error.
    """

    asset: str
    timeframe: str
    start_ts: str
    end_ts: str
    rows: int
    sources: list[str]
    last_updated: str
    provider_max_days: int | None = None
    gaps: int = 0

    @property
    def key(self) -> str:
        return price_key(self.asset, self.timeframe)

    def validate(self) -> None:
        if self.timeframe not in config.TIMEFRAMES:
            raise SchemaError(f"unknown timeframe {self.timeframe!r}")
        if self.rows < 0:
            raise SchemaError("PriceCoverage.rows must be >= 0")
        if _utc(self.start_ts) > _utc(self.end_ts):
            raise SchemaError(f"start_ts after end_ts for {self.key}")

    @classmethod
    def from_dict(cls, d: dict) -> "PriceCoverage":
        known = {k: v for k, v in d.items() if k in cls.__dataclass_fields__}
        return cls(**known)


@dataclass
class Manifest:
    schema_version: int = config.MANIFEST_SCHEMA_VERSION
    updated_at: str = field(default_factory=_now_iso)
    signals: dict[str, CoverageEntry] = field(default_factory=dict)
    prices: dict[str, PriceCoverage] = field(default_factory=dict)

    # -- serialisation ----------------------------------------------------

    def to_dict(self) -> dict:
        return {
            "schema_version": self.schema_version,
            "updated_at": self.updated_at,
            "signals": {k: asdict(v) for k, v in sorted(self.signals.items())},
            "prices": {k: asdict(v) for k, v in sorted(self.prices.items())},
        }

    @classmethod
    def from_dict(cls, d: dict) -> "Manifest":
        if not isinstance(d, dict):
            raise SchemaError("manifest must be a JSON object")
        ver = d.get("schema_version")
        if ver is None:
            raise SchemaError("manifest missing schema_version")
        if int(ver) > config.MANIFEST_SCHEMA_VERSION:
            raise SchemaError(
                f"manifest schema_version {ver} is newer than this app supports "
                f"({config.MANIFEST_SCHEMA_VERSION}); upgrade the Space"
            )
        m = cls(
            schema_version=int(ver),
            updated_at=d.get("updated_at") or _now_iso(),
            signals={k: CoverageEntry.from_dict(v) for k, v in (d.get("signals") or {}).items()},
            prices={k: PriceCoverage.from_dict(v) for k, v in (d.get("prices") or {}).items()},
        )
        m.validate()
        return m

    def to_json(self) -> str:
        return json.dumps(self.to_dict(), indent=2, sort_keys=False) + "\n"

    @classmethod
    def from_json(cls, text: str) -> "Manifest":
        try:
            return cls.from_dict(json.loads(text))
        except json.JSONDecodeError as e:
            raise SchemaError(f"manifest is not valid JSON: {e}") from e

    def validate(self) -> None:
        for k, e in self.signals.items():
            e.validate()
            if e.key != k:
                raise SchemaError(f"manifest signal key {k!r} != entry key {e.key!r}")
        for k, p in self.prices.items():
            p.validate()
            if p.key != k:
                raise SchemaError(f"manifest price key {k!r} != entry key {p.key!r}")

    # -- queries ----------------------------------------------------------

    def get_signal(self, model_slug, model_revision, asset, timeframe) -> CoverageEntry | None:
        return self.signals.get(signal_key(model_slug, model_revision, asset, timeframe))

    def find_signals(self, model_slug=None, asset=None, timeframe=None) -> list[CoverageEntry]:
        out = []
        for e in self.signals.values():
            if model_slug and e.model_slug != model_slug:
                continue
            if asset and e.asset != asset:
                continue
            if timeframe and e.timeframe != timeframe:
                continue
            out.append(e)
        return sorted(out, key=lambda e: (e.model_slug, e.asset, e.timeframe))

    def upsert_signal(self, entry: CoverageEntry) -> None:
        """Merge a new slice into coverage, widening the range and summing rows.

        Coverage is a *union*; re-writing an overlapping slice must not double
        count, so callers pass the post-merge row count via `rows`.
        """
        entry.validate()
        prev = self.signals.get(entry.key)
        if prev is not None:
            entry.start_ts = _iso(min(_utc(prev.start_ts), _utc(entry.start_ts)))
            entry.end_ts = _iso(max(_utc(prev.end_ts), _utc(entry.end_ts)))
        self.signals[entry.key] = entry
        self.updated_at = _now_iso()

    def upsert_price(self, cov: PriceCoverage) -> None:
        cov.validate()
        prev = self.prices.get(cov.key)
        if prev is not None:
            cov.start_ts = _iso(min(_utc(prev.start_ts), _utc(cov.start_ts)))
            cov.end_ts = _iso(max(_utc(prev.end_ts), _utc(cov.end_ts)))
            cov.sources = sorted(set(prev.sources) | set(cov.sources))
        self.prices[cov.key] = cov
        self.updated_at = _now_iso()


def empty_manifest() -> Manifest:
    return Manifest()


# --------------------------------------------------------------------------
# Frame validation
# --------------------------------------------------------------------------


def validate_signal_frame(df: pd.DataFrame, kind: SignalKind = "quantile") -> pd.DataFrame:
    """Check and normalise a signal frame. Returns a sorted, UTC-indexed copy."""
    cols = SIGNAL_COLUMNS_QUANTILE if kind == "quantile" else SIGNAL_COLUMNS_CLASSIFIER
    missing = [c for c in cols if c not in df.columns]
    if missing:
        raise SchemaError(f"signal frame missing columns: {missing}")
    out = df.loc[:, cols].copy()
    out["ts"] = out["ts"].map(_utc)
    if out["ts"].duplicated().any():
        dupes = out.loc[out["ts"].duplicated(), "ts"].head(3).tolist()
        raise SchemaError(f"signal frame has duplicate timestamps, e.g. {dupes}")
    if kind == "quantile":
        for c in ("q10", "q50", "q90"):
            if out[c].isna().any():
                raise SchemaError(f"signal frame column {c} contains NaN")
        # Quantiles must be monotone; a crossed quantile means a broken adapter.
        bad = (out["q10"] > out["q50"]) | (out["q50"] > out["q90"])
        if bad.any():
            raise SchemaError(
                f"{int(bad.sum())} rows have crossed quantiles (q10>q50 or q50>q90)"
            )
    return out.sort_values("ts").reset_index(drop=True)


@dataclass
class PriceValidation:
    """Outcome of validating an OHLCV frame -- the gap report lives here."""

    rows: int
    gaps: int = 0
    gap_ranges: list[tuple[str, str]] = field(default_factory=list)
    problems: list[str] = field(default_factory=list)

    @property
    def ok(self) -> bool:
        return not self.problems


def validate_price_frame(
    df: pd.DataFrame, timeframe: str, *, strict: bool = True
) -> tuple[pd.DataFrame, PriceValidation]:
    """Validate an OHLCV frame and report gaps.

    Rejects: missing columns, duplicate timestamps, negative/zero prices,
    non-finite values, and OHLC bars that are internally inconsistent
    (high < low, or a high below open/close).
    """
    missing = [c for c in PRICE_COLUMNS if c not in df.columns]
    if missing:
        raise SchemaError(f"price frame missing columns: {missing}")

    out = df.loc[:, PRICE_COLUMNS].copy()
    out["ts"] = out["ts"].map(_utc)
    out = out.sort_values("ts").reset_index(drop=True)
    report = PriceValidation(rows=len(out))

    if out["ts"].duplicated().any():
        dupes = out.loc[out["ts"].duplicated(), "ts"].head(3).tolist()
        report.problems.append(f"duplicate timestamps: {dupes}")

    ohlc = ["open", "high", "low", "close"]
    for c in ohlc:
        vals = pd.to_numeric(out[c], errors="coerce")
        if vals.isna().any():
            report.problems.append(f"{c} has non-numeric or NaN values")
        if (vals <= 0).any():
            n = int((vals <= 0).sum())
            report.problems.append(f"{c} has {n} non-positive values")
        out[c] = vals

    vol = pd.to_numeric(out["volume"], errors="coerce")
    if (vol < 0).any():
        report.problems.append(f"volume has {int((vol < 0).sum())} negative values")
    out["volume"] = vol

    inconsistent = (
        (out["high"] < out["low"])
        | (out["high"] < out["open"])
        | (out["high"] < out["close"])
        | (out["low"] > out["open"])
        | (out["low"] > out["close"])
    )
    if inconsistent.any():
        report.problems.append(f"{int(inconsistent.sum())} bars have inconsistent OHLC")

    # Gap report: how many expected bars are absent from the series.
    if len(out) > 2:
        step = pd.Timedelta(minutes=config.TIMEFRAMES[timeframe].minutes)
        deltas = out["ts"].diff().dropna()
        gap_mask = deltas > step * 1.5
        report.gaps = int(gap_mask.sum())
        idx = list(deltas.index[gap_mask])[:20]
        report.gap_ranges = [
            (_iso(out["ts"].iloc[i - 1]), _iso(out["ts"].iloc[i])) for i in idx
        ]

    if strict and report.problems:
        raise SchemaError("price validation failed: " + "; ".join(report.problems))
    return out, report


# --------------------------------------------------------------------------
# The store
# --------------------------------------------------------------------------


def _years(start, end) -> list[int]:
    return list(range(_utc(start).year, _utc(end).year + 1))


class SignalStore:
    """Read/write access to the signal store.

    `local_root` is a working mirror of the repo. Reads fall back to the Hub
    when a file is absent locally; writes stage into the mirror and are pushed
    by `flush()` as one atomic commit.

    Set `offline=True` (or leave `repo_id=None`) for a purely local store --
    used by tests and by the seed script before it has a token.
    """

    def __init__(
        self,
        repo_id: str | None = config.STORE_REPO,
        local_root: str | os.PathLike | None = None,
        token: str | None = None,
        offline: bool = False,
        revision: str = "main",
    ) -> None:
        self.repo_id = repo_id
        self.revision = revision
        self.token = token or os.environ.get("HF_WRITE_TOKEN") or os.environ.get("HF_TOKEN")
        self.offline = offline or repo_id is None
        self.local_root = Path(local_root or Path(os.environ.get("BIT_STORE_CACHE", ".cache/store")))
        self.local_root.mkdir(parents=True, exist_ok=True)
        self._manifest: Manifest | None = None
        self._pending: set[str] = set()  # repo-relative paths staged for commit
        self._lock = threading.RLock()
        self._scheduler = None

    # -- paths ------------------------------------------------------------

    @staticmethod
    def signal_path(model_slug: str, asset: str, timeframe: str, year: int) -> str:
        return f"signals/{model_slug}/{asset}/{timeframe}/{year}.parquet"

    @staticmethod
    def price_path(asset: str, timeframe: str, year: int) -> str:
        return f"prices/{asset}/{timeframe}/{year}.parquet"

    @staticmethod
    def comparison_path(name: str) -> str:
        return f"comparisons/{name}"

    def _local(self, repo_path: str) -> Path:
        return self.local_root / repo_path

    # -- low-level file access -------------------------------------------

    def _fetch(self, repo_path: str) -> Path | None:
        """Return a local path for `repo_path`, pulling from the Hub if needed."""
        local = self._local(repo_path)
        if local.exists():
            return local
        if self.offline:
            return None
        try:
            from huggingface_hub import hf_hub_download

            got = hf_hub_download(
                repo_id=self.repo_id,
                repo_type=config.STORE_REPO_TYPE,
                filename=repo_path,
                revision=self.revision,
                token=self.token,
            )
            return Path(got)
        except Exception:
            # Absent from the Hub is a normal "no coverage" answer, not a fault.
            return None

    def read_parquet(self, repo_path: str) -> pd.DataFrame | None:
        p = self._fetch(repo_path)
        if p is None:
            return None
        return pd.read_parquet(p)

    def _stage(self, repo_path: str, write) -> None:
        """Write a file into the local mirror and mark it for the next commit."""
        local = self._local(repo_path)
        local.parent.mkdir(parents=True, exist_ok=True)
        write(local)
        with self._lock:
            self._pending.add(repo_path)

    # -- manifest ---------------------------------------------------------

    def load_manifest(self, force: bool = False) -> Manifest:
        with self._lock:
            if self._manifest is not None and not force:
                return self._manifest
            local = self._local(config.MANIFEST_PATH)
            text = None
            if local.exists():
                text = local.read_text()
            elif not self.offline:
                p = self._fetch(config.MANIFEST_PATH)
                if p is not None:
                    text = Path(p).read_text()
            self._manifest = Manifest.from_json(text) if text else empty_manifest()
            return self._manifest

    def save_manifest(self, manifest: Manifest | None = None) -> Manifest:
        with self._lock:
            m = manifest or self.load_manifest()
            m.validate()
            m.updated_at = _now_iso()
            self._manifest = m
            self._stage(config.MANIFEST_PATH, lambda p: p.write_text(m.to_json()))
            return m

    # -- coverage queries -------------------------------------------------

    def has_coverage(
        self, model_slug: str, model_revision: str, asset: str, timeframe: str,
        start=None, end=None, *, allow_placeholder: bool = True,
    ) -> bool:
        e = self.load_manifest().get_signal(model_slug, model_revision, asset, timeframe)
        if e is None or e.rows == 0:
            return False
        if not allow_placeholder and e.is_placeholder:
            return False
        if start is not None and _utc(start) < _utc(e.start_ts):
            return False
        if end is not None and _utc(end) > _utc(e.end_ts):
            return False
        return True

    def missing_ranges(
        self, model_slug: str, model_revision: str, asset: str, timeframe: str, start, end
    ) -> list[tuple[pd.Timestamp, pd.Timestamp]]:
        """Sub-ranges of [start, end] not yet covered. Drives extension dedup:
        an already-covered request returns [] and must never be recomputed."""
        s, e = _utc(start), _utc(end)
        if s > e:
            return []
        entry = self.load_manifest().get_signal(model_slug, model_revision, asset, timeframe)
        if entry is None or entry.rows == 0:
            return [(s, e)]
        cs, ce = _utc(entry.start_ts), _utc(entry.end_ts)
        out = []
        if s < cs:
            out.append((s, min(e, cs - pd.Timedelta(seconds=1))))
        if e > ce:
            out.append((max(s, ce + pd.Timedelta(seconds=1)), e))
        return [(a, b) for a, b in out if a <= b]

    # -- reads ------------------------------------------------------------

    def _read_years(self, path_fn, years: Iterable[int]) -> pd.DataFrame | None:
        frames = []
        for y in years:
            df = self.read_parquet(path_fn(y))
            if df is not None and len(df):
                frames.append(df)
        if not frames:
            return None
        out = pd.concat(frames, ignore_index=True)
        out["ts"] = out["ts"].map(_utc)
        return out.sort_values("ts").reset_index(drop=True)

    def get_signals(
        self, model_slug: str, asset: str, timeframe: str, start=None, end=None
    ) -> pd.DataFrame:
        """Signal slice, ts-indexed. Empty frame when there is no coverage."""
        m = self.load_manifest()
        entries = m.find_signals(model_slug=model_slug, asset=asset, timeframe=timeframe)
        if start is None or end is None:
            if not entries:
                return pd.DataFrame(columns=SIGNAL_COLUMNS_QUANTILE).set_index(
                    pd.DatetimeIndex([], tz="UTC", name="ts")
                )
            start = start or min(_utc(e.start_ts) for e in entries)
            end = end or max(_utc(e.end_ts) for e in entries)
        s, e = _utc(start), _utc(end)
        df = self._read_years(
            lambda y: self.signal_path(model_slug, asset, timeframe, y), _years(s, e)
        )
        if df is None:
            return pd.DataFrame(columns=SIGNAL_COLUMNS_QUANTILE).set_index(
                pd.DatetimeIndex([], tz="UTC", name="ts")
            )
        df = df[(df["ts"] >= s) & (df["ts"] <= e)]
        return df.set_index("ts").sort_index()

    def get_prices(self, asset: str, timeframe: str, start=None, end=None) -> pd.DataFrame:
        """OHLCV slice, ts-indexed. Empty frame when there is no coverage."""
        m = self.load_manifest()
        cov = m.prices.get(price_key(asset, timeframe))
        if start is None:
            start = cov.start_ts if cov else "1970-01-01"
        if end is None:
            end = cov.end_ts if cov else _now_iso()
        s, e = _utc(start), _utc(end)
        df = self._read_years(lambda y: self.price_path(asset, timeframe, y), _years(s, e))
        if df is None:
            return pd.DataFrame(columns=PRICE_COLUMNS).set_index(
                pd.DatetimeIndex([], tz="UTC", name="ts")
            )
        df = df[(df["ts"] >= s) & (df["ts"] <= e)]
        return df.set_index("ts").sort_index()

    # -- writes -----------------------------------------------------------

    def _merge_year(self, repo_path: str, new: pd.DataFrame,
                    supersede_on: str | None = None) -> pd.DataFrame:
        """Merge a slice into a year file.

        Re-writing the *same* thing is idempotent: existing rows win on a
        timestamp collision, so a repeated seed or extension changes nothing.

        `supersede_on` names a column that identifies which computation produced
        a row -- `inference_version` for signals. When the incoming rows carry a
        different value there, they are a *different* computation and must
        replace what is stored.

        This matters because the parquet path is keyed on model slug, not model
        revision, while the manifest is keyed on both. Without superseding, a
        second revision would be recorded in the manifest while the file still
        held the first revision's numbers -- and a PLACEHOLDER slice would
        shadow real output permanently.
        """
        existing = self.read_parquet(repo_path)
        if existing is None or not len(existing):
            return new.sort_values("ts").reset_index(drop=True)

        existing = existing.copy()
        existing["ts"] = existing["ts"].map(_utc)

        if supersede_on and supersede_on in existing.columns \
                and supersede_on in new.columns and len(new):
            incoming_version = new[supersede_on].iloc[0]
            # Drop stored rows that this write supersedes at the same instant.
            superseded = (existing["ts"].isin(set(new["ts"]))
                          & (existing[supersede_on] != incoming_version))
            if superseded.any():
                log.info("superseding %d row(s) in %s (%s -> %s)",
                         int(superseded.sum()), repo_path,
                         existing.loc[superseded, supersede_on].iloc[0],
                         incoming_version)
                existing = existing[~superseded]

        merged = pd.concat([existing, new], ignore_index=True)
        merged = merged.drop_duplicates(subset="ts", keep="first")
        return merged.sort_values("ts").reset_index(drop=True)

    def write_signals(
        self,
        model_slug: str,
        model_id: str,
        model_revision: str,
        asset: str,
        timeframe: str,
        df: pd.DataFrame,
        *,
        inference_version: str = config.INFERENCE_VERSION,
        contributed_by: str = "seed",
        signal_kind: SignalKind = "quantile",
    ) -> CoverageEntry:
        """Stage a signal slice and update the manifest. Idempotent per ts."""
        frame = validate_signal_frame(df, kind=signal_kind)
        if frame.empty:
            raise StoreError("refusing to write an empty signal frame")

        # The caller passes `inference_version` *and* the frame carries a column
        # of the same name. If those disagree, the manifest records one version
        # while the rows claim another -- and supersede compares the wrong
        # value, silently keeping stale numbers. One of them has to win, and it
        # is the argument, because that is what the manifest entry records.
        stamped = set(frame["inference_version"].unique())
        if stamped != {inference_version}:
            log.debug("stamping inference_version %s over %s",
                      inference_version, sorted(stamped))
            frame = frame.copy()
            frame["inference_version"] = inference_version

        total_rows = 0
        for year, part in frame.groupby(frame["ts"].dt.year):
            path = self.signal_path(model_slug, asset, timeframe, int(year))
            merged = self._merge_year(path, part, supersede_on="inference_version")
            self._stage(path, lambda p, m=merged: m.to_parquet(p, index=False))
            total_rows += len(merged)

        # Row count reflects the union across every year touched, so an
        # overlapping re-write does not inflate the manifest.
        m = self.load_manifest()
        prev = m.get_signal(model_slug, model_revision, asset, timeframe)
        untouched = 0
        if prev is not None:
            touched_years = set(frame["ts"].dt.year.unique())
            for y in _years(prev.start_ts, prev.end_ts):
                if y not in touched_years:
                    old = self.read_parquet(self.signal_path(model_slug, asset, timeframe, y))
                    untouched += 0 if old is None else len(old)

        entry = CoverageEntry(
            model_slug=model_slug,
            model_id=model_id,
            model_revision=model_revision,
            asset=asset,
            timeframe=timeframe,
            start_ts=_iso(frame["ts"].iloc[0]),
            end_ts=_iso(frame["ts"].iloc[-1]),
            rows=total_rows + untouched,
            inference_version=inference_version,
            last_updated=_now_iso(),
            contributed_by=contributed_by,
            signal_kind=signal_kind,
        )
        m.upsert_signal(entry)
        self.save_manifest(m)
        return entry

    def write_prices(
        self, asset: str, timeframe: str, df: pd.DataFrame, *, strict: bool = True
    ) -> PriceCoverage:
        """Stage an OHLCV slice and update price coverage."""
        frame, report = validate_price_frame(df, timeframe, strict=strict)
        if frame.empty:
            raise StoreError("refusing to write an empty price frame")

        total_rows = 0
        for year, part in frame.groupby(frame["ts"].dt.year):
            path = self.price_path(asset, timeframe, int(year))
            merged = self._merge_year(path, part)
            self._stage(path, lambda p, m=merged: m.to_parquet(p, index=False))
            total_rows += len(merged)

        m = self.load_manifest()
        prev = m.prices.get(price_key(asset, timeframe))
        untouched = 0
        if prev is not None:
            touched_years = set(frame["ts"].dt.year.unique())
            for y in _years(prev.start_ts, prev.end_ts):
                if y not in touched_years:
                    old = self.read_parquet(self.price_path(asset, timeframe, y))
                    untouched += 0 if old is None else len(old)

        cov = PriceCoverage(
            asset=asset,
            timeframe=timeframe,
            start_ts=_iso(frame["ts"].iloc[0]),
            end_ts=_iso(frame["ts"].iloc[-1]),
            rows=total_rows + untouched,
            sources=sorted(set(frame["source"].dropna().astype(str))),
            last_updated=_now_iso(),
            provider_max_days=config.TIMEFRAMES[timeframe].yahoo_max_days,
            gaps=report.gaps,
        )
        m.upsert_price(cov)
        self.save_manifest(m)
        return cov

    def write_json(self, repo_path: str, payload: dict) -> None:
        self._stage(repo_path, lambda p: p.write_text(json.dumps(payload, indent=2) + "\n"))

    def write_table(self, repo_path: str, df: pd.DataFrame) -> None:
        self._stage(repo_path, lambda p: df.to_parquet(p, index=False))

    # -- commit -----------------------------------------------------------

    @property
    def pending(self) -> list[str]:
        with self._lock:
            return sorted(self._pending)

    def flush(self, message: str = "Update signal store") -> str | None:
        """Push every staged file as ONE atomic commit, manifest operation last.

        A single commit is strictly stronger than writing the manifest after
        the data files: readers never observe a manifest that references a
        parquet slice that is not yet present.
        """
        with self._lock:
            paths = sorted(self._pending)
            if not paths:
                return None
            if self.offline:
                self._pending.clear()
                return None

            from huggingface_hub import CommitOperationAdd, HfApi

            # Manifest last so it is the final operation in the commit.
            ordered = [p for p in paths if p != config.MANIFEST_PATH]
            if config.MANIFEST_PATH in paths:
                ordered.append(config.MANIFEST_PATH)

            ops = [
                CommitOperationAdd(path_in_repo=p, path_or_fileobj=str(self._local(p)))
                for p in ordered
            ]
            api = HfApi(token=self.token)
            info = api.create_commit(
                repo_id=self.repo_id,
                repo_type=config.STORE_REPO_TYPE,
                revision=self.revision,
                operations=ops,
                commit_message=message,
            )
            self._pending.clear()
            return getattr(info, "oid", None) or str(info)

    def attach_scheduler(self, every_minutes: float = 5.0):
        """Batch commits in the background while the Space runs.

        The scheduler watches the same local mirror `flush()` stages into, so
        the two paths never disagree about what is on disk.
        """
        if self.offline or self._scheduler is not None:
            return self._scheduler
        from huggingface_hub import CommitScheduler

        self._scheduler = CommitScheduler(
            repo_id=self.repo_id,
            repo_type=config.STORE_REPO_TYPE,
            folder_path=str(self.local_root),
            every=every_minutes,
            token=self.token,
            squash_history=False,
        )
        return self._scheduler