xiaobai-review/xiaobai-datahub/datahub/pipeline.py

from __future__ import annotations

import json
import math
import time
from collections.abc import Callable
from datetime import timedelta
from typing import Any

from datahub.adapters.base import AdapterError
from datahub.adapters.tushare import DEFAULT_INDEX_CODES, WEBSITE_INDEX_CODES, TushareAdapter
from datahub.db import DATASET_TABLES, HubDB
from datahub.governance.circuit import CircuitBreaker
from datahub.governance.ratelimit import TokenBucket
from datahub.governance.retry import RetryError, retry_call
from datahub.logutil import get_logger
from datahub.normalize import finite_number, normalize_daily
from datahub.settings import Settings
from datahub.timeutil import add_days, isoformat, now_shanghai, yyyymmdd

LOGGER = get_logger()

HARD_DATASETS = {"daily", "valuation", "index_daily"}
SOFT_DATASETS = {"moneyflow", "auction"}
OFFICIAL_DATASETS = HARD_DATASETS | SOFT_DATASETS
STOCKS_DATASET = "stocks"
STOCK_SNAPSHOT_FIELDS = ("ts_code", "symbol", "name", "area", "industry", "market", "list_status", "list_date")
EOD_A_DATASETS = ("daily", "valuation", "moneyflow", "auction")
EOD_B_DATASETS = ("index_daily",)
EMPTY_BATCH_ERROR = "empty official batch: 0 valid rows"

STAGING_INSERT = {
    "stocks": (
        "INSERT INTO staging_stocks(ts_code,trade_date,batch_id,symbol,name,area,industry,market,list_status,list_date) "
        "VALUES (?,?,?,?,?,?,?,?,?,?)",
        lambda r, b: (
            r["ts_code"], r["trade_date"], b, r.get("symbol"), r.get("name"), r.get("area"),
            r.get("industry"), r.get("market"), r.get("list_status"), r.get("list_date"),
        ),
    ),
    "daily": (
        "INSERT INTO staging_bars(ts_code,trade_date,batch_id,open,high,low,close,pct_chg,volume,amount,adj_factor) "
        "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
        lambda r, b: (
            r["ts_code"], r["trade_date"], b, r.get("open"), r.get("high"), r.get("low"),
            r.get("close"), r.get("pct_chg"), r.get("volume"), r.get("amount"), r.get("adj_factor"),
        ),
    ),
    "valuation": (
        "INSERT INTO staging_valuation(ts_code,trade_date,batch_id,turnover_rate,volume_ratio,total_mv,circ_mv,pe_ttm,pb,ps_ttm,dv_ttm) "
        "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
        lambda r, b: (
            r["ts_code"], r["trade_date"], b, r.get("turnover_rate"), r.get("volume_ratio"),
            r.get("total_mv"), r.get("circ_mv"), r.get("pe_ttm"), r.get("pb"), r.get("ps_ttm"), r.get("dv_ttm"),
        ),
    ),
    "moneyflow": (
        "INSERT INTO staging_moneyflow(ts_code,trade_date,batch_id,buy_sm_amount,sell_sm_amount,buy_md_amount,sell_md_amount,buy_lg_amount,sell_lg_amount,buy_elg_amount,sell_elg_amount,net_mf_amount) "
        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
        lambda r, b: (
            r["ts_code"], r["trade_date"], b,
            r.get("buy_sm_amount"), r.get("sell_sm_amount"), r.get("buy_md_amount"), r.get("sell_md_amount"),
            r.get("buy_lg_amount"), r.get("sell_lg_amount"), r.get("buy_elg_amount"), r.get("sell_elg_amount"),
            r.get("net_mf_amount"),
        ),
    ),
    "auction": (
        "INSERT INTO staging_auction(ts_code,trade_date,batch_id,volume,price,amount,pre_close,turnover_rate,volume_ratio,float_share) "
        "VALUES (?,?,?,?,?,?,?,?,?,?)",
        lambda r, b: (
            r["ts_code"], r["trade_date"], b, r.get("volume"), r.get("price"), r.get("amount"),
            r.get("pre_close"), r.get("turnover_rate"), r.get("volume_ratio"), r.get("float_share"),
        ),
    ),
    "index_daily": (
        "INSERT INTO staging_index_bars(ts_code,trade_date,batch_id,open,high,low,close,pct_chg,volume,amount) "
        "VALUES (?,?,?,?,?,?,?,?,?,?)",
        lambda r, b: (
            r["ts_code"], r["trade_date"], b, r.get("open"), r.get("high"), r.get("low"),
            r.get("close"), r.get("pct_chg"), r.get("volume"), r.get("amount"),
        ),
    ),
}

EOD_COPY = {
    "stocks": (
        "INSERT OR REPLACE INTO eod_stocks "
        "SELECT ts_code,trade_date,symbol,name,area,industry,market,list_status,list_date,batch_id "
        "FROM staging_stocks WHERE batch_id = ?"
    ),
    "daily": (
        "INSERT OR REPLACE INTO eod_bars "
        "SELECT ts_code,trade_date,open,high,low,close,pct_chg,volume,amount,adj_factor,batch_id "
        "FROM staging_bars WHERE batch_id = ?"
    ),
    "valuation": (
        "INSERT OR REPLACE INTO eod_valuation "
        "SELECT ts_code,trade_date,turnover_rate,volume_ratio,total_mv,circ_mv,pe_ttm,pb,ps_ttm,dv_ttm,batch_id "
        "FROM staging_valuation WHERE batch_id = ?"
    ),
    "moneyflow": (
        "INSERT OR REPLACE INTO eod_moneyflow "
        "SELECT ts_code,trade_date,buy_sm_amount,sell_sm_amount,buy_md_amount,sell_md_amount,"
        "buy_lg_amount,sell_lg_amount,buy_elg_amount,sell_elg_amount,net_mf_amount,batch_id "
        "FROM staging_moneyflow WHERE batch_id = ?"
    ),
    "auction": (
        "INSERT OR REPLACE INTO eod_auction "
        "SELECT ts_code,trade_date,volume,price,amount,pre_close,turnover_rate,volume_ratio,float_share,batch_id "
        "FROM staging_auction WHERE batch_id = ?"
    ),
    "index_daily": (
        "INSERT OR REPLACE INTO eod_index_bars "
        "SELECT ts_code,trade_date,open,high,low,close,pct_chg,volume,amount,batch_id "
        "FROM staging_index_bars WHERE batch_id = ?"
    ),
}


def _finite(value: Any) -> bool:
    try:
        return math.isfinite(float(value))
    except (TypeError, ValueError):
        return False


def _staging_row_count(connection: Any, dataset: str, batch_id: str) -> int:
    table = DATASET_TABLES[dataset][1]
    row = connection.execute(
        f"SELECT COUNT(*) AS n FROM {table} WHERE batch_id = ?",
        (batch_id,),
    ).fetchone()
    return int(row["n"] if row is not None else 0)


def _staging_count_or_raise(connection: Any, dataset: str, trade_date: str, batch_id: str) -> int:
    rows_out = _staging_row_count(connection, dataset, batch_id)
    if rows_out <= 0:
        report = {
            "rows": 0,
            "errors": [EMPTY_BATCH_ERROR],
            "warnings": [],
            "hard_fail": True,
            "soft_fail": False,
            "batch_id": batch_id,
            "dataset": dataset,
            "trade_date": trade_date,
        }
        LOGGER.warning(
            "skip official publish for empty batch",
            extra={
                "hub": {
                    "dataset": dataset,
                    "trade_date": trade_date,
                    "batch_id": batch_id,
                    "rows_out": rows_out,
                    "reason": "upstream_empty",
                }
            },
        )
        raise QualityError("empty batch cannot be officially published", report)
    return rows_out


def _upsert_publication(
    connection: Any,
    dataset: str,
    trade_date: str,
    batch_id: str,
    state: str,
    published_at: str,
) -> None:
    current = connection.execute(
        "SELECT active_batch FROM publications WHERE dataset = ? AND trade_date = ?",
        (dataset, trade_date),
    ).fetchone()
    prev = str(current["active_batch"]) if current else None
    connection.execute(
        """
        INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)
        VALUES (?, ?, ?, ?, ?, ?)
        ON CONFLICT(dataset, trade_date) DO UPDATE SET
          prev_batch=excluded.prev_batch,
          active_batch=excluded.active_batch,
          state=excluded.state,
          published_at=excluded.published_at
        """,
        (dataset, trade_date, batch_id, prev, state, published_at),
    )


def _record_publication_history(
    connection: Any,
    dataset: str,
    trade_date: str,
    batch_id: str,
    published_at: str,
    quality: dict[str, Any],
) -> None:
    max_gen = connection.execute(
        "SELECT COALESCE(MAX(generation), 0) AS g FROM publication_history WHERE dataset = ? AND trade_date = ?",
        (dataset, trade_date),
    ).fetchone()
    generation = int(max_gen["g"]) + 1
    connection.execute(
        "INSERT OR REPLACE INTO publication_history(dataset, trade_date, batch_id, published_at, generation) VALUES (?,?,?,?,?)",
        (dataset, trade_date, batch_id, published_at, generation),
    )
    keep = int(quality.get("publication_generations") or 3)
    stale = connection.execute(
        """
        SELECT batch_id FROM publication_history
        WHERE dataset = ? AND trade_date = ?
        ORDER BY generation DESC
        """,
        (dataset, trade_date),
    ).fetchall()
    for row in stale[keep:]:
        connection.execute(
            "DELETE FROM publication_history WHERE dataset = ? AND trade_date = ? AND batch_id = ?",
            (dataset, trade_date, row["batch_id"]),
        )


class QualityError(RuntimeError):
    def __init__(self, message: str, report: dict[str, Any]) -> None:
        super().__init__(message)
        self.report = report


class Pipeline:
    def __init__(
        self,
        db: HubDB,
        adapter: TushareAdapter,
        settings: Settings,
        bucket: TokenBucket | None = None,
        breaker: CircuitBreaker | None = None,
        before_commit: Callable[[], None] | None = None,
        clock=None,
    ) -> None:
        self.db = db
        self.adapter = adapter
        self.settings = settings
        self.bucket = bucket or TokenBucket(settings.tushare_rate_per_minute)
        self.breaker = breaker or CircuitBreaker()
        self.before_commit = before_commit
        self.clock = clock or now_shanghai

    def next_batch_id(self, dataset: str, trade_date: str) -> str:
        row = self.db.fetchone(
            "SELECT COUNT(*) AS n FROM batches WHERE dataset = ? AND trade_date = ?",
            (dataset, trade_date),
        )
        seq = int((row or {}).get("n") or 0) + 1
        return f"{trade_date}-{dataset}-{seq:03d}"

    def ingest_reference(
        self,
        trade_date: str | None = None,
        start: str | None = None,
        end: str | None = None,
    ) -> dict[str, Any]:
        """Refresh trade calendar and stock master. Not versioned by batch.

        Calendar defaults to 2016-01-01 through today+30 so a 5-year website
        query is not silently truncated. UPSERT makes repeats safe.
        """
        day = yyyymmdd(trade_date or self.clock())
        start = yyyymmdd(start or self.settings.calendar_start)
        end = yyyymmdd(end or add_days(day, 30))
        if start > end:
            start, end = end, start
        calendar = self.adapter.normalize(
            "calendar",
            self._guarded_fetch("calendar", {"exchange": "SSE", "start_date": start, "end_date": end}),
        )
        stocks = self.adapter.normalize("stocks", self._guarded_fetch("stocks", {"list_status": "L"}))
        fetched_at = isoformat(self.clock())
        with self.db.write() as connection:
            for row in calendar:
                connection.execute(
                    """
                    INSERT INTO trade_calendar(exchange, cal_date, is_open, pretrade_date, fetched_at)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(exchange, cal_date) DO UPDATE SET
                      is_open=excluded.is_open, pretrade_date=excluded.pretrade_date, fetched_at=excluded.fetched_at
                    """,
                    (row["exchange"], row["cal_date"], row["is_open"], row.get("pretrade_date"), fetched_at),
                )
            self._upsert_stock_master(connection, stocks, fetched_at)
        return {
            "calendar": len(calendar),
            "stocks": len(stocks),
            "trade_date": day,
            "calendar_from": start,
            "calendar_to": end,
        }

    @staticmethod
    def _upsert_stock_master(connection: Any, stocks: list[dict[str, Any]], fetched_at: str) -> None:
        for row in stocks:
            connection.execute(
                """
                INSERT INTO stock_master(ts_code,symbol,name,area,industry,market,list_status,list_date,updated_at)
                VALUES (?,?,?,?,?,?,?,?,?)
                ON CONFLICT(ts_code) DO UPDATE SET
                  symbol=excluded.symbol, name=excluded.name, area=excluded.area,
                  industry=excluded.industry, market=excluded.market,
                  list_status=excluded.list_status, list_date=excluded.list_date,
                  updated_at=excluded.updated_at
                """,
                (
                    row["ts_code"], row.get("symbol"), row.get("name"), row.get("area"),
                    row.get("industry"), row.get("market"), row.get("list_status"),
                    row.get("list_date"), fetched_at,
                ),
            )

    def latest_stocks_publication(self, on_or_before: str | None = None) -> dict[str, Any] | None:
        if on_or_before:
            row = self.db.fetchone(
                "SELECT * FROM publications WHERE dataset = ? AND trade_date <= ? ORDER BY trade_date DESC LIMIT 1",
                (STOCKS_DATASET, yyyymmdd(on_or_before)),
            )
        else:
            row = self.db.fetchone(
                "SELECT * FROM publications WHERE dataset = ? ORDER BY trade_date DESC LIMIT 1",
                (STOCKS_DATASET,),
            )
        return row

    def published_stock_snapshot(self, on_or_before: str | None = None) -> tuple[str | None, list[dict[str, Any]]]:
        pub = self.latest_stocks_publication(on_or_before)
        if not pub:
            return None, []
        rows = self.db.fetchall(
            f"SELECT {','.join(STOCK_SNAPSHOT_FIELDS)} FROM eod_stocks WHERE batch_id = ? ORDER BY ts_code",
            (pub["active_batch"],),
        )
        return str(pub["active_batch"]), rows

    def refresh_stocks(self, trade_date: str | None = None, force: bool = False) -> dict[str, Any]:
        """Refresh stock master from upstream and publish a versioned snapshot.

        Runs on trading days (scheduler) and via CLI. Idempotent: when the
        latest published snapshot already matches the upstream list, nothing
        is published; ``force`` re-publishes unconditionally. New listings,
        renames (incl. N/C prefix removal) and status changes all flow into
        the snapshot, which carries batch_id/published_at metadata.

        The ``stock_master`` UPSERT happens inside the same publish
        transaction as the snapshot switch — fetch / quality-gate / switch
        failures leave the master on the previous complete values.
        """
        day = yyyymmdd(trade_date or self.clock())
        try:
            rows = self._fetch_dataset(STOCKS_DATASET, day)
        except Exception as exc:
            self.audit(
                "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
                json.dumps({"state": "failed", "error": str(exc)}, ensure_ascii=False),
            )
            raise
        if not force:
            active, snapshot = self.published_stock_snapshot(day)
            if active is not None:
                upstream = sorted(
                    tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in rows
                )
                published = sorted(tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in snapshot)
                if upstream == published:
                    return {
                        "dataset": STOCKS_DATASET,
                        "trade_date": day,
                        "state": "skipped",
                        "reason": "unchanged",
                        "batch_id": active,
                        "rows": len(snapshot),
                    }
        try:
            result = self.run_dataset(STOCKS_DATASET, day, prepared_rows=rows)
        except Exception as exc:
            self.audit(
                "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
                json.dumps({"state": "failed", "error": str(exc)}, ensure_ascii=False),
            )
            raise
        self.audit(
            "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}",
            json.dumps({"batch_id": result["batch_id"], "rows": result["rows"]}, ensure_ascii=False),
        )
        return result

    def backfill_moneyflow_history(
        self,
        end_date: str | None = None,
        trading_days: int | None = None,
        force: bool = False,
    ) -> dict[str, Any]:
        """Incrementally publish official moneyflow history for the website window.

        The website queries moneyflow for any navigable trade date (stock
        detail point queries) and for the screener's last-5-days window, so
        the hub must cover a trailing window of trading days instead of only
        days published since go-live. Already published dates are skipped
        unless ``force``; per-date failures are recorded without aborting.
        """
        end = yyyymmdd(end_date or self.clock())
        limit = int(trading_days or self.settings.moneyflow_history_trading_days)
        open_dates = self.open_trade_dates(end, limit)
        if not open_dates:
            return {
                "start": None,
                "end": end,
                "requested_days": 0,
                "published": [],
                "skipped": [],
                "failed": [{"error": "calendar has no open dates on or before end"}],
                "ok": False,
            }
        start = open_dates[0]
        published_dates: set[str] = set()
        if not force:
            pubs = self.db.fetchall(
                "SELECT trade_date FROM publications WHERE dataset = 'moneyflow' AND trade_date >= ? AND trade_date <= ?",
                (start, end),
            )
            published_dates = {str(row["trade_date"]) for row in pubs}
        targets = [day for day in open_dates if day not in published_dates]
        skipped = [day for day in open_dates if day in published_dates]
        published: list[dict[str, Any]] = []
        failed: list[dict[str, Any]] = []
        for day in targets:
            try:
                raw = retry_call(
                    lambda day=day: self._guarded_fetch("moneyflow", {"trade_date": day}),
                    attempts=self.settings.max_publish_attempts,
                    base_delay=0.05,
                    sleeper=lambda _d: time.sleep(_d),
                )
                rows = self.adapter.normalize("moneyflow", raw)
                result = self.run_dataset("moneyflow", day, prepared_rows=rows)
                published.append(
                    {
                        "trade_date": day,
                        "batch_id": result["batch_id"],
                        "rows": result["rows"],
                        "state": result["state"],
                    }
                )
            except Exception as exc:
                failed.append({"trade_date": day, "error": str(exc)})
        return {
            "start": start,
            "end": end,
            "requested_days": len(open_dates),
            "published": published,
            "skipped": skipped,
            "failed": failed,
            "ok": not failed,
        }

    def open_trade_dates(self, end: str, limit: int) -> list[str]:
        end = yyyymmdd(end)
        rows = self.db.fetchall(
            """
            SELECT cal_date FROM trade_calendar
            WHERE exchange = 'SSE' AND is_open = 1 AND cal_date <= ?
            ORDER BY cal_date DESC
            LIMIT ?
            """,
            (end, max(1, int(limit))),
        )
        return sorted(str(row["cal_date"]) for row in rows)

    def backfill_history(
        self,
        trade_date: str | None = None,
        calendar_start: str | None = None,
        index_days: int | None = None,
        codes: tuple[str, ...] | None = None,
        force: bool = False,
    ) -> dict[str, Any]:
        """Idempotent calendar + website-index history backfill."""
        day = yyyymmdd(trade_date or self.clock())
        calendar = self.ingest_reference(day, start=calendar_start)
        index = self.backfill_index_history(
            end_date=day,
            trading_days=index_days,
            codes=codes,
            force=force,
        )
        return {"calendar": calendar, "index_daily": index, "ok": bool(index.get("ok"))}

    def backfill_index_history(
        self,
        end_date: str | None = None,
        trading_days: int | None = None,
        codes: tuple[str, ...] | None = None,
        force: bool = False,
    ) -> dict[str, Any]:
        """Incrementally publish official index bars for website index codes.

        One range fetch per code, then per-day publish. Already published dates
        are skipped unless ``force``. Failures are recorded and do not roll back
        successful days.
        """
        end = yyyymmdd(end_date or self.clock())
        limit = int(trading_days or self.settings.index_history_trading_days)
        codes = tuple(codes or WEBSITE_INDEX_CODES)
        open_dates = self.open_trade_dates(end, limit)
        if not open_dates:
            return {
                "start": None,
                "end": end,
                "codes": list(codes),
                "requested_days": 0,
                "published": [],
                "skipped": [],
                "failed": [{"error": "calendar has no open dates on or before end"}],
                "ok": False,
            }
        start = open_dates[0]
        complete_dates = set() if force else self._index_dates_with_all_codes(start, end, codes)
        targets = [day for day in open_dates if day not in complete_dates]
        skipped = [day for day in open_dates if day in complete_dates]
        by_date: dict[str, list[dict[str, Any]]] = {day: [] for day in targets}
        failed: list[dict[str, Any]] = []
        for ts_code in codes:
            try:
                raw = retry_call(
                    lambda code=ts_code: self._guarded_fetch(
                        "index_daily",
                        {"ts_code": code, "start_date": start, "end_date": end},
                    ),
                    attempts=self.settings.max_publish_attempts,
                    base_delay=0.05,
                    sleeper=lambda _d: time.sleep(_d),
                )
                for row in self.adapter.normalize("index_daily", raw):
                    day = str(row.get("trade_date") or "")
                    if day in by_date:
                        by_date[day].append(row)
            except Exception as exc:
                failed.append({"ts_code": ts_code, "error": str(exc)})
        published: list[dict[str, Any]] = []
        for day in targets:
            rows = by_date.get(day) or []
            try:
                result = self.run_dataset("index_daily", day, prepared_rows=rows)
                published.append(
                    {
                        "trade_date": day,
                        "batch_id": result["batch_id"],
                        "rows": result["rows"],
                        "state": result["state"],
                    }
                )
            except Exception as exc:
                failed.append({"trade_date": day, "error": str(exc), "rows": len(rows)})
        return {
            "start": start,
            "end": end,
            "codes": list(codes),
            "requested_days": len(open_dates),
            "published": published,
            "skipped": skipped,
            "failed": failed,
            "ok": not failed,
        }

    def _index_dates_with_all_codes(self, start: str, end: str, codes: tuple[str, ...]) -> set[str]:
        pubs = self.db.fetchall(
            """
            SELECT trade_date, active_batch FROM publications
            WHERE dataset = 'index_daily' AND trade_date >= ? AND trade_date <= ?
            """,
            (start, end),
        )
        needed = set(codes)
        complete: set[str] = set()
        for pub in pubs:
            rows = self.db.fetchall(
                "SELECT DISTINCT ts_code FROM eod_index_bars WHERE trade_date = ? AND batch_id = ?",
                (pub["trade_date"], pub["active_batch"]),
            )
            have = {str(row["ts_code"]) for row in rows}
            if needed <= have:
                complete.add(str(pub["trade_date"]))
        return complete

    def run_dataset(
        self,
        dataset: str,
        trade_date: str,
        attempts: int | None = None,
        prepared_rows: list[dict[str, Any]] | None = None,
    ) -> dict[str, Any]:
        trade_date = yyyymmdd(trade_date)
        batch_id = self.next_batch_id(dataset, trade_date)
        max_attempts = attempts or self.settings.max_publish_attempts
        self._set_batch(batch_id, dataset, trade_date, "scheduled", 0)
        rows: list[dict[str, Any]] = []
        try:
            self._set_batch(batch_id, dataset, trade_date, "fetching", 1)
            if prepared_rows is None:
                rows = retry_call(
                    lambda: self._fetch_dataset(dataset, trade_date),
                    attempts=max_attempts,
                    base_delay=0.05,
                    sleeper=lambda _d: None if attempts == 1 else time.sleep(_d),
                )
            else:
                rows = list(prepared_rows)
            self._stage(dataset, batch_id, rows)
            self._set_batch(batch_id, dataset, trade_date, "staged", 1, rows_in=len(rows), rows_out=len(rows))
            self._set_batch(batch_id, dataset, trade_date, "validating", 1)
            report = self.validate(dataset, batch_id, trade_date, rows)
            if report["hard_fail"]:
                self._reject_batch(batch_id, dataset, trade_date, rows, report)
                raise QualityError("integrity gate failed", report)
            self._set_batch(batch_id, dataset, trade_date, "deriving", 1, rows_in=len(rows), rows_out=len(rows), quality=report)
            self._set_batch(batch_id, dataset, trade_date, "publishing", 1, rows_in=len(rows), rows_out=len(rows), quality=report)
            state = "degraded" if report["soft_fail"] else "published"
            self.publish(dataset, trade_date, batch_id, state=state)
            self._set_batch(
                batch_id, dataset, trade_date, "published", 1,
                rows_in=len(rows), rows_out=len(rows), quality=report, finished=True,
            )
            return {"batch_id": batch_id, "dataset": dataset, "trade_date": trade_date, "rows": len(rows), "state": state, "quality": report}
        except RetryError as exc:
            self._set_batch(batch_id, dataset, trade_date, "failed", max_attempts, error=str(exc), finished=True)
            raise
        except QualityError as exc:
            current = self.db.fetchone("SELECT state FROM batches WHERE batch_id = ?", (batch_id,))
            if current and current["state"] not in {"staged", "failed"}:
                self._reject_batch(batch_id, dataset, trade_date, rows, exc.report)
            raise
        except Exception as exc:
            self._set_batch(batch_id, dataset, trade_date, "failed", 1, error=str(exc), finished=True)
            raise

    def missing_official_datasets(self, trade_date: str) -> list[str]:
        """Official datasets without an active publication for the date."""
        day = yyyymmdd(trade_date)
        placeholders = ",".join("?" for _ in OFFICIAL_DATASETS)
        rows = self.db.fetchall(
            f"SELECT dataset FROM publications WHERE trade_date = ? AND dataset IN ({placeholders})",
            (day, *sorted(OFFICIAL_DATASETS)),
        )
        published = {str(row["dataset"]) for row in rows}
        return [dataset for dataset in sorted(OFFICIAL_DATASETS) if dataset not in published]

    def run_eod_missing(self, trade_date: str) -> dict[str, Any]:
        """Republish every incomplete EOD consistency group for the date.

        A-group (daily/valuation/moneyflow/auction + stocks) and B-group
        (index_daily) are separate boundaries. Within a group, either the
        whole boundary is already published (idempotent skip) or every
        member is re-staged and switched together — never fill only the
        missing members on top of older batches from an earlier partial run.
        """
        results: dict[str, Any] = {}
        results.update(self.run_eod_batch_a(trade_date))
        results.update(self.run_eod_batch_b(trade_date))
        return results

    def run_eod_batch_a(self, trade_date: str, force: bool = False) -> dict[str, Any]:
        return self.run_release_group(EOD_A_DATASETS, trade_date, include_stocks=True, force=force)

    def run_eod_batch_b(self, trade_date: str, force: bool = False) -> dict[str, Any]:
        return self.run_release_group(EOD_B_DATASETS, trade_date, force=force)

    def force_republish_boundary(self, dataset: str, trade_date: str) -> dict[str, Any]:
        """Force-republish the full A/B consistency boundary that owns ``dataset``.

        CLI ``eod-refresh --force`` and admin manual backfill must not publish a
        single official member alone — that would mix old and new batches inside
        the same trade date. Naming any A-group member (or stocks) rebuilds the
        whole A group; naming ``index_daily`` rebuilds B.
        """
        name = str(dataset or "").strip()
        if name in EOD_A_DATASETS or name == STOCKS_DATASET:
            return self.run_eod_batch_a(trade_date, force=True)
        if name in EOD_B_DATASETS:
            return self.run_eod_batch_b(trade_date, force=True)
        raise ValueError(f"dataset is not part of an EOD release boundary: {dataset}")

    def run_release_group(
        self,
        datasets: tuple[str, ...],
        trade_date: str,
        include_stocks: bool = False,
        force: bool = False,
    ) -> dict[str, Any]:
        """One post-market publish/republish becomes one atomic visibility flip.

        Consistency boundary: every member (official datasets, plus the daily
        stocks snapshot when ``include_stocks``) is fetched, staged,
        field-gated and cross-validated BEFORE any reader can see it. Only
        when the whole group passes does a single SQLite transaction copy
        all staging batches to the official tables and flip every
        ``publications`` row at once. Any member failure aborts the group:
        the previous complete official version keeps serving and the reason
        is recorded on the batches and in the audit log.

        Skip is all-or-nothing for the boundary unless ``force``: if every
        official member (and stocks when required) is already published, the
        group is skipped. If any official member is still missing — or
        ``force`` is set — every official member is re-staged, so a retry or
        manual republish never mixes old and new batches in one release.
        """
        day = yyyymmdd(trade_date)
        results: dict[str, Any] = {}
        staged: dict[str, dict[str, Any]] = {}
        failure: str | None = None
        missing_official = [dataset for dataset in datasets if self.active_batch(dataset, day) is None]
        stocks_missing = include_stocks and self.active_batch(STOCKS_DATASET, day) is None

        if not force and not missing_official and not stocks_missing:
            for dataset in datasets:
                results[dataset] = {
                    "dataset": dataset,
                    "trade_date": day,
                    "state": "skipped",
                    "reason": "already_published",
                }
            if include_stocks:
                results[STOCKS_DATASET] = {
                    "dataset": STOCKS_DATASET,
                    "trade_date": day,
                    "state": "skipped",
                    "reason": "already_published",
                }
            return results

        # Incomplete or forced boundary → restage every official member together.
        pending = list(datasets)

        for dataset in pending:
            if failure is not None:
                results[dataset] = {
                    "dataset": dataset,
                    "trade_date": day,
                    "state": "aborted",
                    "reason": f"release group aborted: {failure}",
                }
                continue
            try:
                staged[dataset] = self._stage_and_validate(dataset, day)
            except Exception as exc:
                failure = f"{dataset}: {exc}"
                results[dataset] = {
                    "dataset": dataset,
                    "trade_date": day,
                    "state": "failed",
                    "error": str(exc),
                }

        # Stocks join the same switch when the official boundary is being
        # rebuilt (missing or forced), or when only the stocks snapshot is
        # still missing.
        rebuild_official = bool(force or missing_official)
        if include_stocks and failure is None and (rebuild_official or stocks_missing):
            try:
                stocks_plan = self._stage_stocks_snapshot(day, force=rebuild_official)
            except Exception as exc:
                failure = f"{STOCKS_DATASET}: {exc}"
                results[STOCKS_DATASET] = {
                    "dataset": STOCKS_DATASET,
                    "trade_date": day,
                    "state": "failed",
                    "error": str(exc),
                }
            else:
                if stocks_plan is not None:
                    staged[STOCKS_DATASET] = stocks_plan

        if failure is None and staged:
            cross_errors = self._cross_gate_errors(staged)
            if cross_errors:
                failure = "; ".join(cross_errors)

        if failure is not None:
            for dataset, item in staged.items():
                self._abandon_batch(item["batch_id"], f"release group not switched: {failure}")
                results[dataset] = {
                    "dataset": dataset,
                    "trade_date": day,
                    "state": "failed",
                    "error": f"release group not switched: {failure}",
                    "batch_id": item["batch_id"],
                }
            LOGGER.warning(
                "release group blocked, previous official version keeps serving",
                extra={
                    "hub": {
                        "trade_date": day,
                        "datasets": sorted(staged),
                        "reason": failure,
                        "event": "release_group_blocked",
                    }
                },
            )
            self.audit(
                "pipeline", "release-group", f"eod:{day}",
                json.dumps({"state": "failed", "reason": failure}, ensure_ascii=False),
            )
            return results

        if staged:
            try:
                self._switch_release_group(day, staged)
            except Exception as exc:
                reason = f"release group switch failed: {exc}"
                for item in staged.values():
                    self._abandon_batch(item["batch_id"], reason)
                LOGGER.warning(
                    "release group switch failed, previous official version keeps serving",
                    extra={
                        "hub": {
                            "trade_date": day,
                            "datasets": sorted(staged),
                            "reason": reason,
                            "event": "release_group_switch_failed",
                        }
                    },
                )
                self.audit(
                    "pipeline", "release-group", f"eod:{day}",
                    json.dumps(
                        {
                            "state": "failed",
                            "reason": reason,
                            "switched": [],
                            "force": bool(force),
                        },
                        ensure_ascii=False,
                    ),
                )
                raise
            for dataset, item in staged.items():
                results[dataset] = {
                    "dataset": dataset,
                    "trade_date": day,
                    "state": item["state"],
                    "batch_id": item["batch_id"],
                    "rows": item["rows"],
                }
            self.audit(
                "pipeline", "release-group", f"eod:{day}",
                json.dumps(
                    {"state": "ok", "switched": sorted(staged), "force": bool(force)},
                    ensure_ascii=False,
                ),
            )
        return results

    def _stage_and_validate(self, dataset: str, trade_date: str, attempts: int | None = None) -> dict[str, Any]:
        """Fetch → stage → quality-gate one member without publishing it."""
        day = yyyymmdd(trade_date)
        batch_id = self.next_batch_id(dataset, day)
        max_attempts = attempts or self.settings.max_publish_attempts
        rows: list[dict[str, Any]] = []
        self._set_batch(batch_id, dataset, day, "scheduled", 0)
        try:
            self._set_batch(batch_id, dataset, day, "fetching", 1)
            rows = retry_call(
                lambda: self._fetch_dataset(dataset, day),
                attempts=max_attempts,
                base_delay=0.05,
                sleeper=lambda _d: time.sleep(_d),
            )
            self._stage(dataset, batch_id, rows)
            self._set_batch(batch_id, dataset, day, "staged", 1, rows_in=len(rows), rows_out=len(rows))
            self._set_batch(batch_id, dataset, day, "validating", 1, rows_in=len(rows), rows_out=len(rows))
            report = self.validate(dataset, batch_id, day, rows)
            if report["hard_fail"]:
                self._reject_batch(batch_id, dataset, day, rows, report)
                raise QualityError("integrity gate failed", report)
        except RetryError as exc:
            self._set_batch(batch_id, dataset, day, "failed", max_attempts, error=str(exc), finished=True)
            raise
        except QualityError as exc:
            current = self.db.fetchone("SELECT state FROM batches WHERE batch_id = ?", (batch_id,))
            if current and current["state"] not in {"staged", "failed"}:
                self._reject_batch(batch_id, dataset, day, rows, exc.report)
            raise
        except Exception as exc:
            self._set_batch(batch_id, dataset, day, "failed", 1, error=str(exc), finished=True)
            raise
        self._set_batch(
            batch_id, dataset, day, "ready", 1, rows_in=len(rows), rows_out=len(rows), quality=report
        )
        return {
            "dataset": dataset,
            "trade_date": day,
            "batch_id": batch_id,
            "rows": len(rows),
            "quality": report,
            "state": "degraded" if report["soft_fail"] else "published",
        }

    def _stage_stocks_snapshot(self, trade_date: str, force: bool = False) -> dict[str, Any] | None:
        """Stage the daily stocks snapshot for a release group switch.

        Returns None when the published snapshot is already identical to
        upstream (idempotent skip) unless ``force`` is set. The stock_master
        upsert is deferred into the group switch / publish transaction so the
        master never runs ahead of the published snapshot.
        """
        day = yyyymmdd(trade_date)
        active, snapshot = self.published_stock_snapshot(day)
        rows = self._fetch_dataset(STOCKS_DATASET, day)
        if active is not None and not force:
            upstream = sorted(
                tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in rows
            )
            published = sorted(tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in snapshot)
            if upstream == published:
                return None
        batch_id = self.next_batch_id(STOCKS_DATASET, day)
        self._set_batch(batch_id, STOCKS_DATASET, day, "scheduled", 0)
        self._set_batch(batch_id, STOCKS_DATASET, day, "fetching", 1)
        self._stage(STOCKS_DATASET, batch_id, rows)
        self._set_batch(batch_id, STOCKS_DATASET, day, "staged", 1, rows_in=len(rows), rows_out=len(rows))
        self._set_batch(batch_id, STOCKS_DATASET, day, "validating", 1, rows_in=len(rows), rows_out=len(rows))
        report = self.validate(STOCKS_DATASET, batch_id, day, rows)
        if report["hard_fail"]:
            self._reject_batch(batch_id, STOCKS_DATASET, day, rows, report)
            raise QualityError("integrity gate failed", report)
        self._set_batch(
            batch_id, STOCKS_DATASET, day, "ready", 1, rows_in=len(rows), rows_out=len(rows), quality=report
        )
        return {
            "dataset": STOCKS_DATASET,
            "trade_date": day,
            "batch_id": batch_id,
            "rows": len(rows),
            "row_values": rows,
            "quality": report,
            "state": "degraded" if report["soft_fail"] else "published",
        }

    def _cross_gate_errors(self, staged: dict[str, dict[str, Any]]) -> list[str]:
        """Cross-dataset consistency checks on staged batches (交叉校验)."""
        errors: list[str] = []
        gates = self.settings.quality.get("cross_gates") or []
        for gate in gates if isinstance(gates, list) else []:
            if not isinstance(gate, dict):
                continue
            left = str(gate.get("left") or "")
            right = str(gate.get("right") or "")
            if not left or not right or left not in staged or right not in staged:
                continue
            floor = float(gate.get("min_key_overlap") or 0.98)
            left_keys = self._staging_keys(left, staged[left]["batch_id"])
            right_keys = self._staging_keys(right, staged[right]["batch_id"])
            denom = max(len(left_keys), len(right_keys))
            overlap = (len(left_keys & right_keys) / denom) if denom else 1.0
            if overlap < floor:
                errors.append(
                    f"cross gate: {left} vs {right} key overlap {overlap:.4f} < {floor}"
                )
        return errors

    def _staging_keys(self, dataset: str, batch_id: str) -> set[str]:
        table = DATASET_TABLES[dataset][1]
        rows = self.db.fetchall(
            f"SELECT DISTINCT ts_code FROM {table} WHERE batch_id = ?",
            (batch_id,),
        )
        return {str(row["ts_code"]) for row in rows}

    def _switch_release_group(self, trade_date: str, members: dict[str, dict[str, Any]]) -> None:
        """Single transaction: copy every member and flip every publication."""
        day = yyyymmdd(trade_date)
        published_at = isoformat(self.clock())
        with self.db.write() as connection:
            for dataset, item in members.items():
                _staging_count_or_raise(connection, dataset, day, item["batch_id"])
            for dataset, item in members.items():
                connection.execute(EOD_COPY[dataset], (item["batch_id"],))
                if dataset == STOCKS_DATASET:
                    self._upsert_stock_master(connection, item["row_values"], published_at)
            if self.before_commit:
                self.before_commit()
            for dataset, item in members.items():
                _upsert_publication(connection, dataset, day, item["batch_id"], item["state"], published_at)
                _record_publication_history(
                    connection, dataset, day, item["batch_id"], published_at, self.settings.quality
                )
                connection.execute(
                    "UPDATE batches SET state='published', finished_at=? WHERE batch_id=?",
                    (published_at, item["batch_id"]),
                )

    def _abandon_batch(self, batch_id: str, reason: str) -> None:
        row = self.db.fetchone("SELECT dataset, trade_date FROM batches WHERE batch_id = ?", (batch_id,))
        if not row:
            return
        self._set_batch(
            batch_id, str(row["dataset"]), str(row["trade_date"]), "failed", 1,
            error=reason, finished=True,
        )

    @staticmethod
    def eod_failures(results: dict[str, Any]) -> list[str]:
        return [
            f"{name}: {item.get('error')}"
            for name, item in results.items()
            if isinstance(item, dict) and item.get("state") == "failed"
        ]

    def validate(self, dataset: str, batch_id: str, trade_date: str, rows: list[dict[str, Any]]) -> dict[str, Any]:
        quality = self.settings.quality
        errors: list[str] = []
        warnings: list[str] = []
        listed = self.db.fetchone(
            "SELECT COUNT(*) AS n FROM stock_master WHERE list_status = 'L'",
        )
        listed_n = int((listed or {}).get("n") or 0)
        row_n = len(rows)
        keys = [(row.get("ts_code"), row.get("trade_date")) for row in rows]
        dup = row_n - len(set(keys))
        if dup:
            errors.append(f"duplicate keys: {dup}")
        bad_date = sum(1 for row in rows if str(row.get("trade_date")) != trade_date)
        if bad_date:
            errors.append(f"date mismatch rows: {bad_date}")
        ratio = (row_n / listed_n) if listed_n else 1.0
        if dataset == "daily" and listed_n and ratio < float(quality.get("daily_row_ratio") or 0.98):
            errors.append(f"row ratio {ratio:.4f} < {quality.get('daily_row_ratio')}")
        null_fields = ("open", "high", "low", "close", "amount") if dataset in {"daily", "index_daily"} else ()
        if null_fields and rows:
            nulls = sum(1 for row in rows if any(row.get(field) is None for field in null_fields))
            null_rate = nulls / row_n
            if null_rate >= float(quality.get("null_rate_max") or 0.01):
                errors.append(f"null rate {null_rate:.4f}")
        empty = row_n == 0
        if empty and (dataset in OFFICIAL_DATASETS or dataset == STOCKS_DATASET):
            errors.append(EMPTY_BATCH_ERROR)
        field_report = self._field_gate(dataset, trade_date, rows, errors)
        if dataset in SOFT_DATASETS:
            hard_fail = bool(dup or bad_date or empty)
        else:
            hard_fail = bool(errors) and (dataset in HARD_DATASETS or dataset == STOCKS_DATASET)
        report = {
            "rows": row_n,
            "listed": listed_n,
            "ratio": round(ratio, 4),
            "errors": errors,
            "warnings": warnings,
            "hard_fail": hard_fail,
            "soft_fail": bool(warnings) and not hard_fail,
            "batch_id": batch_id,
        }
        if field_report is not None:
            report["fields"] = field_report
        return report

    def _field_gate(
        self,
        dataset: str,
        trade_date: str,
        rows: list[dict[str, Any]],
        errors: list[str],
    ) -> dict[str, Any] | None:
        """Config-driven per-field completeness gate.

        Catches field-level half-products (rows complete, key columns empty)
        that row-count gates miss: non-null rate floors per field, non-finite
        share, and a collapse guard against the previous published batch so
        legitimately sparse fields (e.g. dv_ttm) are not false-flagged.
        Generic for any dataset configured under quality["field_gates"].
        """
        gate = dict((self.settings.quality.get("field_gates") or {}).get(dataset) or {})
        if not gate or not rows:
            return None
        fields = [str(item) for item in (gate.get("fields") or []) if str(item)]
        if not fields:
            return None
        min_rate = float(gate.get("min_nonnull_rate") or 0.9)
        by_field = {str(k): float(v) for k, v in dict(gate.get("min_nonnull_rate_by_field") or {}).items()}
        max_drop = float(gate.get("max_nonnull_drop_vs_prev") or 0.15)
        max_nonfinite = float(gate.get("max_nonfinite_rate") or 0.01)
        row_n = len(rows)
        prev_batch, prev_stats = self._prev_field_stats(dataset, trade_date, fields)
        report: dict[str, Any] = {}
        for field in fields:
            values = [row.get(field) for row in rows]
            nulls = sum(1 for value in values if value is None)
            nonfinite = sum(1 for value in values if value is not None and not _finite(value))
            rate = (row_n - nulls) / row_n
            stats = {
                "nonnull": row_n - nulls,
                "null": nulls,
                "nonnull_rate": round(rate, 4),
                "nonfinite": nonfinite,
            }
            floor = by_field.get(field, min_rate)
            if rate < floor:
                errors.append(
                    f"field gate: {dataset}.{field} non-null rate {rate:.4f} < {floor}"
                )
            if nonfinite / row_n > max_nonfinite:
                errors.append(
                    f"field gate: {dataset}.{field} non-finite rate {nonfinite / row_n:.4f} > {max_nonfinite}"
                )
            prev_rate = prev_stats.get(field) if prev_stats else None
            if prev_rate is not None:
                stats["prev_nonnull_rate"] = round(prev_rate, 4)
                if prev_rate - rate > max_drop:
                    errors.append(
                        f"field gate: {dataset}.{field} non-null rate {rate:.4f} dropped > {max_drop:.2f} "
                        f"vs prev batch {prev_batch} ({prev_rate:.4f})"
                    )
            report[field] = stats
        return report

    def _prev_field_stats(
        self,
        dataset: str,
        trade_date: str,
        fields: list[str],
    ) -> tuple[str | None, dict[str, float]]:
        """Non-null rates per field from the latest earlier published batch."""
        if dataset not in DATASET_TABLES:
            return None, {}
        table = DATASET_TABLES[dataset][0]
        columns = {
            str(row["name"])
            for row in self.db.fetchall(f"PRAGMA table_info({table})")
        }
        usable = [field for field in fields if field in columns]
        if not usable:
            return None, {}
        pub = self.db.fetchone(
            """
            SELECT active_batch FROM publications
            WHERE dataset = ? AND trade_date < ? ORDER BY trade_date DESC LIMIT 1
            """,
            (dataset, trade_date),
        )
        if not pub:
            return None, {}
        batch_id = str(pub["active_batch"])
        selects = ",".join(f"AVG({field} IS NOT NULL) AS {field}" for field in usable)
        row = self.db.fetchone(
            f"SELECT {selects} FROM {table} WHERE batch_id = ?",
            (batch_id,),
        )
        stats = {field: float(row[field]) for field in usable if row.get(field) is not None}
        return batch_id, stats

    def publish(self, dataset: str, trade_date: str, batch_id: str, state: str = "published") -> None:
        published_at = isoformat(self.clock())
        with self.db.write() as connection:
            _staging_count_or_raise(connection, dataset, trade_date, batch_id)
            connection.execute(EOD_COPY[dataset], (batch_id,))
            if dataset == STOCKS_DATASET:
                staging = DATASET_TABLES[STOCKS_DATASET][1]
                stock_rows = [
                    dict(row)
                    for row in connection.execute(
                        f"SELECT * FROM {staging} WHERE batch_id = ?",
                        (batch_id,),
                    ).fetchall()
                ]
                self._upsert_stock_master(connection, stock_rows, published_at)
            if self.before_commit:
                self.before_commit()
            _upsert_publication(connection, dataset, trade_date, batch_id, state, published_at)
            _record_publication_history(connection, dataset, trade_date, batch_id, published_at, self.settings.quality)

    def rollback(self, dataset: str, trade_date: str, actor: str = "admin") -> dict[str, Any]:
        trade_date = yyyymmdd(trade_date)
        pub = self.db.fetchone(
            "SELECT * FROM publications WHERE dataset = ? AND trade_date = ?",
            (dataset, trade_date),
        )
        if not pub or not pub.get("prev_batch"):
            raise ValueError("没有可回滚的上一批次")
        target = pub["prev_batch"]
        published_at = isoformat(self.clock())
        with self.db.write() as connection:
            connection.execute(
                """
                UPDATE publications
                SET prev_batch = active_batch, active_batch = ?, published_at = ?, state = 'published'
                WHERE dataset = ? AND trade_date = ?
                """,
                (target, published_at, dataset, trade_date),
            )
        self.audit(actor, "rollback", f"{dataset}:{trade_date}", json.dumps({"to": target, "from": pub["active_batch"]}))
        return {"dataset": dataset, "trade_date": trade_date, "active_batch": target, "prev_batch": pub["active_batch"]}

    def active_batch(self, dataset: str, trade_date: str) -> str | None:
        row = self.db.fetchone(
            "SELECT active_batch FROM publications WHERE dataset = ? AND trade_date = ?",
            (dataset, trade_date),
        )
        return str(row["active_batch"]) if row else None

    def cleanup(self) -> dict[str, int]:
        staging_days = int(self.settings.quality.get("staging_retain_days") or 14)
        job_days = int(self.settings.quality.get("job_run_retain_days") or 90)
        now = now_shanghai(self.clock())
        cutoff_staging = add_days(yyyymmdd(now), -staging_days)
        cutoff_jobs = isoformat(now - timedelta(days=job_days))
        deleted = 0
        with self.db.write() as connection:
            for dataset, (_eod, staging) in DATASET_TABLES.items():
                cur = connection.execute(
                    f"DELETE FROM {staging} WHERE trade_date < ?",
                    (cutoff_staging,),
                )
                deleted += cur.rowcount
            connection.execute("DELETE FROM job_runs WHERE started_at < ?", (cutoff_jobs,))
            connection.execute("DELETE FROM src_calls WHERE created_at < ?", (cutoff_jobs,))
        return {"staging_deleted": deleted}

    def audit(self, actor: str, action: str, target: str = "", detail: str = "") -> None:
        self.db.execute(
            "INSERT INTO audit_log(actor, action, target, detail, created_at) VALUES (?,?,?,?,?)",
            (actor, action, target, detail, isoformat(self.clock())),
        )

    def _reject_batch(
        self,
        batch_id: str,
        dataset: str,
        trade_date: str,
        rows: list[dict[str, Any]],
        report: dict[str, Any],
    ) -> None:
        errors = report.get("errors") or []
        LOGGER.warning(
            "official batch rejected",
            extra={
                "hub": {
                    "dataset": dataset,
                    "trade_date": trade_date,
                    "batch_id": batch_id,
                    "rows_out": len(rows),
                    "errors": errors,
                    "reason": "upstream_empty" if EMPTY_BATCH_ERROR in errors else "integrity_gate",
                }
            },
        )
        self._set_batch(
            batch_id, dataset, trade_date, "staged", 1,
            rows_in=len(rows), rows_out=len(rows),
            quality=report, error="; ".join(str(item) for item in errors),
        )

    def _fetch_dataset(self, dataset: str, trade_date: str) -> list[dict[str, Any]]:
        if dataset == STOCKS_DATASET:
            rows = self.adapter.normalize("stocks", self._guarded_fetch("stocks", {"list_status": "L"}))
            for row in rows:
                row["trade_date"] = trade_date
            return rows
        if dataset == "daily":
            raw = self._guarded_fetch("daily", {"trade_date": trade_date})
            factors = {
                (row["ts_code"], row["trade_date"]): finite_number(row.get("adj_factor"))
                for row in self._guarded_fetch("adj_factor", {"trade_date": trade_date})
            }
            return [
                normalize_daily(row, adj_factor=factors.get((str(row.get("ts_code") or "").upper(), str(row.get("trade_date") or ""))))
                for row in raw
            ]
        if dataset == "index_daily":
            rows: list[dict[str, Any]] = []
            for ts_code in DEFAULT_INDEX_CODES:
                raw = self._guarded_fetch("index_daily", {"ts_code": ts_code, "trade_date": trade_date})
                rows.extend(self.adapter.normalize("index_daily", raw))
            return rows
        api_dataset = dataset
        raw = self._guarded_fetch(api_dataset, {"trade_date": trade_date})
        return self.adapter.normalize(api_dataset, raw)

    def _guarded_fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
        if not self.breaker.allow():
            raise AdapterError("Tushare circuit open")
        self.bucket.acquire()
        started = time.perf_counter()
        try:
            # For daily we want RAW tushare rows so adj_factor can be merged later.
            rows = self.adapter.fetch(dataset, params)
            latency = round((time.perf_counter() - started) * 1000)
            self.breaker.record_success()
            self._log_call(dataset, True, latency, "")
            self._persist_health("ok")
            return rows
        except Exception as exc:
            latency = round((time.perf_counter() - started) * 1000)
            self.breaker.record_failure(str(exc))
            self._log_call(dataset, False, latency, str(exc))
            self._persist_health("error", str(exc))
            raise

    def _stage(self, dataset: str, batch_id: str, rows: list[dict[str, Any]]) -> None:
        sql, mapper = STAGING_INSERT[dataset]
        with self.db.write() as connection:
            connection.execute(
                f"DELETE FROM {DATASET_TABLES[dataset][1]} WHERE batch_id = ?",
                (batch_id,),
            )
            connection.executemany(sql, [mapper(row, batch_id) for row in rows])

    def _set_batch(
        self,
        batch_id: str,
        dataset: str,
        trade_date: str,
        state: str,
        attempt: int,
        rows_in: int | None = None,
        rows_out: int | None = None,
        quality: dict[str, Any] | None = None,
        error: str | None = None,
        finished: bool = False,
    ) -> None:
        now = isoformat(self.clock())
        existing = self.db.fetchone("SELECT batch_id FROM batches WHERE batch_id = ?", (batch_id,))
        payload = json.dumps(quality, ensure_ascii=False) if quality else None
        with self.db.write() as connection:
            if existing is None:
                connection.execute(
                    """
                    INSERT INTO batches(batch_id, dataset, trade_date, state, attempt, rows_in, rows_out, quality_json, started_at, finished_at, error)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (batch_id, dataset, trade_date, state, attempt, rows_in, rows_out, payload, now, now if finished else None, error),
                )
            else:
                connection.execute(
                    """
                    UPDATE batches SET state=?, attempt=?,
                      rows_in=COALESCE(?, rows_in), rows_out=COALESCE(?, rows_out),
                      quality_json=COALESCE(?, quality_json),
                      finished_at=CASE WHEN ? THEN ? ELSE finished_at END,
                      error=COALESCE(?, error)
                    WHERE batch_id = ?
                    """,
                    (state, attempt, rows_in, rows_out, payload, 1 if finished else 0, now, error, batch_id),
                )

    def _log_call(self, endpoint: str, ok: bool, latency_ms: int, error: str) -> None:
        self.db.execute(
            "INSERT INTO src_calls(provider, endpoint, ok, latency_ms, error, created_at) VALUES (?,?,?,?,?,?)",
            ("tushare", endpoint, 1 if ok else 0, latency_ms, error, isoformat(self.clock())),
        )

    def _persist_health(self, state: str, error: str = "") -> None:
        snap = self.breaker.snapshot()
        self.db.execute(
            """
            INSERT INTO src_health(provider, endpoint_class, state, last_ok_at, last_error, consec_failures, opened_at, cooldown_until)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(provider, endpoint_class) DO UPDATE SET
              state=excluded.state, last_ok_at=excluded.last_ok_at, last_error=excluded.last_error,
              consec_failures=excluded.consec_failures, opened_at=excluded.opened_at, cooldown_until=excluded.cooldown_until
            """,
            (
                "tushare", "pro",
                snap.state,
                isoformat(self.clock()) if state == "ok" else None,
                error or snap.last_error,
                snap.consec_failures,
                isoformat(self.clock()) if snap.state == "open" else None,
                None,
            ),
        )