feat(HEL-382): 搭建 datahub 底座和盘后正式数据链路
新增独立 xiaobai-datahub 服务（SQLite WAL、Tushare 盘后发布、/v1 契约和管理后台），不改现站页面与数据链路。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
总工
co-authored by
Cursor
multica-agent
Cursor
multica-agent
parent
c2ebc0ab91
commit
3498dd7a4b
@@ -0,0 +1,478 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
from datahub.adapters.base import AdapterError
|
||||
from datahub.adapters.tushare import DEFAULT_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"}
|
||||
|
||||
STAGING_INSERT = {
|
||||
"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 = {
|
||||
"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 = ?"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
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) -> dict[str, Any]:
|
||||
"""Refresh trade calendar (window) and stock master. Not versioned by batch."""
|
||||
day = yyyymmdd(trade_date or self.clock())
|
||||
start = add_days(day, -400)
|
||||
end = add_days(day, 30)
|
||||
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),
|
||||
)
|
||||
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,
|
||||
),
|
||||
)
|
||||
return {"calendar": len(calendar), "stocks": len(stocks), "trade_date": day}
|
||||
|
||||
def run_dataset(self, dataset: str, trade_date: str, attempts: int | 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)
|
||||
try:
|
||||
self._set_batch(batch_id, dataset, trade_date, "fetching", 1)
|
||||
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),
|
||||
)
|
||||
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._set_batch(
|
||||
batch_id, dataset, trade_date, "staged", 1,
|
||||
rows_in=len(rows), rows_out=len(rows),
|
||||
quality=report, error="; ".join(report["errors"]),
|
||||
)
|
||||
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:
|
||||
raise
|
||||
except Exception as exc:
|
||||
self._set_batch(batch_id, dataset, trade_date, "failed", 1, error=str(exc), finished=True)
|
||||
raise
|
||||
|
||||
def run_eod_batch_a(self, trade_date: str) -> dict[str, Any]:
|
||||
results = {}
|
||||
for dataset in ("daily", "valuation", "moneyflow", "auction"):
|
||||
results[dataset] = self.run_dataset(dataset, trade_date)
|
||||
return results
|
||||
|
||||
def run_eod_batch_b(self, trade_date: str) -> dict[str, Any]:
|
||||
return {"index_daily": self.run_dataset("index_daily", trade_date)}
|
||||
|
||||
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}")
|
||||
if dataset in SOFT_DATASETS and row_n == 0:
|
||||
warnings.append("empty soft dataset")
|
||||
hard_fail = bool(errors) and dataset in HARD_DATASETS.union({"daily", "valuation", "index_daily"})
|
||||
if dataset in SOFT_DATASETS:
|
||||
hard_fail = bool(dup or bad_date)
|
||||
return {
|
||||
"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,
|
||||
}
|
||||
|
||||
def publish(self, dataset: str, trade_date: str, batch_id: str, state: str = "published") -> None:
|
||||
copy_sql = EOD_COPY[dataset]
|
||||
published_at = isoformat(self.clock())
|
||||
with self.db.write() as connection:
|
||||
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(copy_sql, (batch_id,))
|
||||
if self.before_commit:
|
||||
self.before_commit()
|
||||
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),
|
||||
)
|
||||
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(self.settings.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"]),
|
||||
)
|
||||
|
||||
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)
|
||||
cutoff_staging = add_days(yyyymmdd(self.clock()), -staging_days)
|
||||
cutoff_jobs = add_days(yyyymmdd(self.clock()), -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 _fetch_dataset(self, dataset: str, trade_date: str) -> list[dict[str, Any]]:
|
||||
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,
|
||||
),
|
||||
)
|
||||
Reference in New Issue
Block a user