From fb81e46e44f62e84e1adf7a3eb14255e6450e7f6 Mon Sep 17 00:00:00 2001 From: AI_life <50097546+827346462@users.noreply.github.com> Date: Thu, 30 Jul 2026 20:51:17 +0800 Subject: [PATCH 1/2] feat(quality): add data coverage completeness check Add check_coverage rule that validates all expected ETF/stock codes are present in each daily partition. Uses the basic reference table (list_date/delist_date) to compute expected codes per date. - New rule: check_coverage() in rules.py with severity tiers: - FAIL (<80% coverage) - WARN (80-95% coverage) - WARN (missing well-known ETF like 510300.SH) - New method: _expected_codes_for_date() in runner.py - Integrated into _check_frame for fund_daily and daily_kline targets - Well-known ETF codes (510300, 510050, 159915, etc.) trigger WARN even when coverage is above 95% --- .gitignore | 1 + zer0share/quality/rules.py | 63 +++++++++++++++++++++++++++++++++++++ zer0share/quality/runner.py | 56 +++++++++++++++++++++++++++++++++ 3 files changed, 120 insertions(+) diff --git a/.gitignore b/.gitignore index 3c1ae5a..aeaf87c 100644 --- a/.gitignore +++ b/.gitignore @@ -32,3 +32,4 @@ uv.lock # Claude local config .claude/ +scripts/ diff --git a/zer0share/quality/rules.py b/zer0share/quality/rules.py index 9d51cac..7d6f2c5 100644 --- a/zer0share/quality/rules.py +++ b/zer0share/quality/rules.py @@ -182,6 +182,69 @@ def check_adjustment_factor_values( ] +# well-known ETF codes that should always trigger a warning if missing +WELL_KNOWN_ETF_CODES: set[str] = { + "510300.SH", "510050.SH", "159915.SZ", "510500.SH", + "159919.SZ", "588000.SH", "510880.SH", +} + + +def check_coverage( + table: str, + date: str | None, + df: pd.DataFrame, + expected_codes: set[str], + well_known_codes: set[str] | None = None, +) -> list[QualityFinding]: + """Check that all expected codes are present in the partition data.""" + if not expected_codes: + return [] + + if "ts_code" not in df.columns: + return [ + QualityFinding( + table=table, + date=date, + severity=Severity.FAIL, + rule="coverage", + count=len(expected_codes), + message="ts_code column is missing, coverage check skipped", + sample=[], + ) + ] + + actual_codes = set(df["ts_code"].dropna().astype(str)) + missing = expected_codes - actual_codes + if not missing: + return [] + + coverage = len(actual_codes & expected_codes) / len(expected_codes) + wkc = well_known_codes if well_known_codes is not None else set() + + if coverage < 0.8: + severity = Severity.FAIL + elif coverage < 0.95: + severity = Severity.WARN + elif missing & wkc: + severity = Severity.WARN + else: + return [] + + missing_list = sorted(missing)[:20] + + return [ + QualityFinding( + table=table, + date=date, + severity=severity, + rule="coverage", + count=len(missing), + message=f"coverage is {coverage:.1%}, missing {len(missing)} codes", + sample=[{"missing_codes": missing_list, "coverage": f"{coverage:.1%}"}], + ) + ] + + def check_adjustment_factor_jumps( table: str, df: pd.DataFrame, diff --git a/zer0share/quality/runner.py b/zer0share/quality/runner.py index 089268c..a24482f 100644 --- a/zer0share/quality/runner.py +++ b/zer0share/quality/runner.py @@ -15,8 +15,10 @@ from zer0share.quality.rules import ( check_adjustment_factor_values, check_adjustment_factor_jumps, + check_coverage, check_duplicate_key, check_market_data_values, + WELL_KNOWN_ETF_CODES, check_required_columns, ) from zer0share.quality.targets import QualityTarget, get_targets @@ -83,6 +85,21 @@ def _run_target( partitions = 0 adjusted_return_state: AdjustedReturnState = {} + # load basic reference table for coverage checks (once per target) + self._basic_df: pd.DataFrame | None = None + if target.table == "fund_daily": + basic_path = self.data_dir / "etf" / "etf_basic" / "data.parquet" + elif target.table == "daily_kline": + basic_path = self.data_dir / "stock" / "basic" / "data.parquet" + else: + basic_path = None + + if basic_path is not None and basic_path.exists(): + try: + self._basic_df = pd.read_parquet(basic_path) + except Exception: + self._basic_df = None + for date_value in expected_dates: parquet_path = target_dir / f"date={date_value}" / "data.parquet" if not parquet_path.exists(): @@ -208,6 +225,36 @@ def _trading_dates(self, target: QualityTarget, start_date: str, end_date: str) ) return sorted(set(cal_dates[mask].tolist())) + def _expected_codes_for_date(self, target: QualityTarget, date_value: str) -> set[str]: + """Return the set of ts_code expected to exist for this target on date_value.""" + if self._basic_df is None: + return set() + + df = self._basic_df + if "ts_code" not in df.columns or "list_date" not in df.columns: + return set() + + if target.table == "fund_daily": + mask = pd.Series(True, index=df.index) + if "list_status" in df.columns: + mask = mask & (df["list_status"] != "D") + mask = mask & df["list_date"].notna() + mask = mask & (df["list_date"].astype(str).str.strip() <= date_value) + elif target.table == "daily_kline": + mask = pd.Series(True, index=df.index) + if "list_status" in df.columns: + mask = mask & (df["list_status"] == "L") + mask = mask & df["list_date"].notna() + mask = mask & (df["list_date"].astype(str).str.strip() <= date_value) + if "delist_date" in df.columns: + delist_na = df["delist_date"].isna() | (df["delist_date"].astype(str).str.strip() == "") + delist_future = df["delist_date"].astype(str).str.strip() > date_value + mask = mask & (delist_na | delist_future) + else: + return set() + + return set(df.loc[mask, "ts_code"].dropna().astype(str)) + def _check_frame( self, target: QualityTarget, @@ -219,6 +266,15 @@ def _check_frame( findings.extend(check_duplicate_key(target.table, date_value, df, target.primary_key)) if target.kind == "market_data": findings.extend(check_market_data_values(target.table, date_value, df)) + findings.extend( + check_coverage( + target.table, + date_value, + df, + self._expected_codes_for_date(target, date_value), + WELL_KNOWN_ETF_CODES if target.table == "fund_daily" else None, + ) + ) elif target.kind == "adjustment": findings.extend(check_adjustment_factor_values(target.table, date_value, df)) findings.extend(check_adjustment_factor_jumps(target.table, df)) From 94e946573a916c1dccac3305d584e5ef0cbc0ecf Mon Sep 17 00:00:00 2001 From: AI_life <50097546+827346462@users.noreply.github.com> Date: Thu, 30 Jul 2026 22:42:41 +0800 Subject: [PATCH 2/2] feat(sync): add per-ticker sync mode with relay API support Add TickerSyncJob that iterates over ETF tickers instead of trading dates, enabling sync through relay/proxy APIs that only support per-ticker date-range queries. Key changes: - New _ProxiedDataApi class: Tushare DataApi-compatible client using X-API-Key header auth + proxy-free session for relay endpoints - TushareFetcher accepts optional proxy_url; auto-selects between standard pro_api and _ProxiedDataApi based on URL pattern - New fetch_fund_daily_range(ts_code, start, end) method for per-ticker queries - New get_etf_codes() helper returning sorted ETF list from etf_basic - TickerSyncJob: iterates tickers, fetches date range per ticker, writes daily partitions with dedup - Pipeline build_registry accepts ticker_mode flag; etf module conditionally creates TickerSyncJob vs DailySyncJob - CLI: '--ticker-mode' flag activates per-ticker sync path --- zer0share/cli.py | 10 ++-- zer0share/config.py | 3 ++ zer0share/fetcher.py | 85 ++++++++++++++++++++++++++++- zer0share/pipeline.py | 10 ++-- zer0share/sync/_jobs.py | 115 ++++++++++++++++++++++++++++++++++++++++ zer0share/sync/etf.py | 21 ++++++-- 6 files changed, 229 insertions(+), 15 deletions(-) diff --git a/zer0share/cli.py b/zer0share/cli.py index f845cf8..4e324ba 100644 --- a/zer0share/cli.py +++ b/zer0share/cli.py @@ -30,11 +30,11 @@ def _parse_date(s: str): return dt.datetime.strptime(s, "%Y%m%d").date() -def _make_pipeline(config_path: str = "config/settings.toml") -> Pipeline: +def _make_pipeline(config_path: str = "config/settings.toml", ticker_mode: bool = False) -> Pipeline: cfg = load_config(Path(config_path)) init_logger(cfg.log_path) sources = DataSources( - tushare=TushareFetcher(cfg.tushare_token), + tushare=TushareFetcher(cfg.tushare_token, proxy_url=cfg.tushare_proxy_url), ricequant=( RiceQuantFetcher( username=cfg.ricequant.username, @@ -46,7 +46,7 @@ def _make_pipeline(config_path: str = "config/settings.toml") -> Pipeline: ), ) notifier = build_notifier(cfg.notifier) - return Pipeline(cfg, sources, notifier) + return Pipeline(cfg, sources, notifier, ticker_mode=ticker_mode) @click.group() @@ -129,6 +129,7 @@ def cli(): @click.option("--ricequant", "sync_ricequant", is_flag=True, default=False) @click.option("--start-date", default=None, callback=_validate_date) @click.option("--end-date", default=None, callback=_validate_date) +@click.option("--ticker-mode", is_flag=True, default=False, help="Per-ticker sync mode (for relay/proxy API)") def sync( table: str | None, sync_all: bool, @@ -139,6 +140,7 @@ def sync( sync_ricequant: bool, start_date: str | None, end_date: str | None, + ticker_mode: bool, ) -> None: """同步数据。""" if end_date is not None and start_date is None: @@ -146,7 +148,7 @@ def sync( if start_date is not None and end_date is not None and end_date < start_date: raise click.UsageError("--end-date must be on or after --start-date") - with _make_pipeline() as pipeline: + with _make_pipeline(ticker_mode=ticker_mode) as pipeline: if table is not None and (start_date is not None or end_date is not None): job = pipeline.registry.get(table) if job is not None and not job.supports_date_range: diff --git a/zer0share/config.py b/zer0share/config.py index a6c2d7d..6bb9843 100644 --- a/zer0share/config.py +++ b/zer0share/config.py @@ -61,6 +61,7 @@ class QualityConfig: @dataclass(frozen=True) class Config: tushare_token: str + tushare_proxy_url: str | None data_dir: Path db_path: Path log_path: Path @@ -179,8 +180,10 @@ def load_config(path: Path = Path("config/settings.toml")) -> Config: try: notifier = _parse_notifier(raw) wecom_webhook_url, notifier_enabled = _parse_wecom_notifier(raw["notifier"]) + tushare_proxy_url = str(raw["tushare"].get("proxy_url", "")) or None return Config( tushare_token=raw["tushare"]["token"], + tushare_proxy_url=tushare_proxy_url, data_dir=Path(raw["paths"]["data_dir"]), db_path=Path(raw["paths"]["db_path"]), log_path=Path(raw["paths"]["log_path"]), diff --git a/zer0share/fetcher.py b/zer0share/fetcher.py index ae76920..ec3042b 100644 --- a/zer0share/fetcher.py +++ b/zer0share/fetcher.py @@ -2,6 +2,7 @@ import pandas as pd import requests import time +from functools import partial from loguru import logger import zer0share.dateutil as dateutil @@ -87,11 +88,75 @@ OPTIONS_EXCHANGES = ["SSE", "SZSE", "CFFEX", "DCE", "SHFE", "CZCE"] + +class _ProxiedDataApi: + """Tushare DataApi-compatible client routing through relay with X-API-Key auth. + + Used when the API endpoint requires X-API-Key header instead of body token, + and/or only supports per-ticker queries. + """ + + def __init__(self, token: str, proxy_url: str): + self._session = requests.Session() + self._session.headers.update({ + "X-API-Key": token, + "Content-Type": "application/json", + }) + self._session.trust_env = False + self._session.proxies = {"http": "", "https": ""} + self._proxy_url = proxy_url.rstrip("/") + self._timeout = 120 + + def query(self, api_name: str, fields: str = "", **kwargs): + req_params = { + "api_name": api_name, + "params": kwargs, + "fields": fields, + } + res = self._session.post( + self._proxy_url, + json=req_params, + timeout=self._timeout, + ) + if not res: + return pd.DataFrame() + result = res.json() + if result.get("code") != 0: + raise Exception(result.get("msg", "unknown error")) + data = result.get("data", {}) + columns = data.get("fields", []) + items = data.get("items", []) + if items: + return pd.DataFrame(items, columns=columns) + return pd.DataFrame(columns=columns) + + def __getattr__(self, name): + return partial(self.query, name) + + +# Proxy URL patterns: when it ends with /tushare/pro, the relay uses +# X-API-Key auth and needs the _ProxiedDataApi custom session. +_RELAY_PROXY_SUFFIXES = ("/tushare/pro",) + + +def _is_relay_proxy(url: str | None) -> bool: + """Returns True if the proxy URL is a header-based relay (not tushare SDK compatible).""" + if not url: + return False + return any(url.rstrip("/").endswith(suffix) for suffix in _RELAY_PROXY_SUFFIXES) + + class TushareFetcher: SW_VERSIONS = ("SW2014", "SW2021") - def __init__(self, token: str): - self._pro = ts.pro_api(token) + def __init__(self, token: str, proxy_url: str | None = None): + if proxy_url and _is_relay_proxy(proxy_url): + self._pro = _ProxiedDataApi(token, proxy_url) + elif proxy_url: + self._pro = ts.pro_api(token) + self._pro._DataApi__http_url = proxy_url.rstrip("/") + else: + self._pro = ts.pro_api(token) def fetch_basic(self) -> pd.DataFrame: logger.info("拉取 stock_basic") @@ -230,6 +295,22 @@ def fetch_fund_daily(self, trade_date: str) -> pd.DataFrame: df = self._pro.fund_daily(trade_date=trade_date, fields=",".join(FUND_DAILY_COLS)) return _select_columns_or_empty(df, FUND_DAILY_COLS) + def fetch_fund_daily_range(self, ts_code: str, start_date: str, end_date: str) -> pd.DataFrame: + """Per-ticker fund_daily with date range (for relay/proxy mode).""" + logger.debug(f"拉取ETF基金日线(按ETF): {ts_code} {start_date}~{end_date}") + df = self._pro.fund_daily( + ts_code=ts_code, start_date=start_date, end_date=end_date, + fields=",".join(FUND_DAILY_COLS), + ) + return _select_columns_or_empty(df, FUND_DAILY_COLS) + + def get_etf_codes(self) -> list[str]: + """Get sorted list of active ETF codes from etf_basic.""" + df = self._pro.etf_basic() + if df is None or df.empty or "ts_code" not in df.columns: + return [] + return sorted(df["ts_code"].dropna().unique().tolist()) + def fetch_fund_adj(self, trade_date: str) -> pd.DataFrame: logger.debug(f"拉取ETF基金复权因子: {trade_date}") df = self._pro.fund_adj(trade_date=trade_date, fields=",".join(FUND_ADJ_COLS)) diff --git a/zer0share/pipeline.py b/zer0share/pipeline.py index d776931..7395076 100644 --- a/zer0share/pipeline.py +++ b/zer0share/pipeline.py @@ -8,18 +8,20 @@ class Pipeline: - def __init__(self, cfg: Config, sources: DataSources, notifier: Notifier): + def __init__(self, cfg: Config, sources: DataSources, notifier: Notifier, ticker_mode: bool = False): meta = MetaStore(cfg.db_path) calendar = TradingCalendar(meta) self._runtime = SyncRuntime(calendar=calendar, notifier=notifier, meta=meta) self._registry: dict[str, SyncJob] = {} - self._build_registry(cfg, sources) + self._build_registry(cfg, sources, ticker_mode=ticker_mode) - def _build_registry(self, cfg: Config, sources: DataSources) -> None: + def _build_registry(self, cfg: Config, sources: DataSources, ticker_mode: bool = False) -> None: from zer0share.sync import calendar, stock, index, industry, futures, options, ricequant, etf - for module in [calendar, stock, index, industry, futures, options, etf]: + for module in [calendar, stock, index, industry, futures, options]: for job in module.build_jobs(cfg, sources.tushare): self._registry[job.table_name] = job + for job in etf.build_jobs(cfg, sources.tushare, ticker_mode=ticker_mode): + self._registry[job.table_name] = job for job in ricequant.build_jobs(cfg, sources): self._registry[job.table_name] = job diff --git a/zer0share/sync/_jobs.py b/zer0share/sync/_jobs.py index 6c5955d..1fd816e 100644 --- a/zer0share/sync/_jobs.py +++ b/zer0share/sync/_jobs.py @@ -326,6 +326,121 @@ def run( ) +class TickerSyncJob(SyncJob): + """Sync a daily-partitioned table by iterating over tickers (ETF codes). + + Used when the API only supports per-ticker date-range queries + (relay/proxy mode) rather than per-date bulk queries. + """ + + def __init__( + self, + table_name: str, + spec: DailyTableSpec, + fetch_range: Callable[[str, str, str], pd.DataFrame], + store: DailyPartitionStore, + get_tickers: Callable[[], list[str]], + ticker_sleep: float = 0.35, + supports_date_range: bool = True, + ): + self.table_name = table_name + self.spec = spec + self.fetch_range = fetch_range + self.store = store + self.get_tickers = get_tickers + self.ticker_sleep = ticker_sleep + self.supports_date_range = supports_date_range + + def run( + self, + rt: SyncRuntime, + start_date: str | None = None, + end_date: str | None = None, + ) -> None: + today = rt.calendar.today() + if start_date is None: + last = rt.meta.get_last_date(self.spec.name) + start = dateutil.add_days(last, 1) if last is not None else self.spec.first_date + end = today + if start > end: + logger.info(f"{self.spec.name}: 已是最新 (last={last})") + return + else: + start = start_date + end = end_date if end_date is not None else today + if start > end: + raise ValueError(f"start_date {start} is after end_date {end}") + + trading_days = rt.calendar.get_trading_days("SSE", start, end) + if not trading_days: + return + + logger.info( + f"{self.spec.name} (ticker mode): start {start} ~ {end}, " + f"trading_days={len(trading_days)}" + ) + + tickers = self.get_tickers() + logger.info(f"{self.spec.name}: {len(tickers)} tickers to fetch") + + all_frames: list[pd.DataFrame] = [] + ticker_ok = ticker_empty = 0 + started_at = time.monotonic() + + for i, tc in enumerate(tickers): + try: + df = self.fetch_range(tc, start, end) + if df is not None and not df.empty: + all_frames.append(df) + ticker_ok += 1 + else: + ticker_empty += 1 + except Exception as exc: + logger.warning(f"{self.spec.name}: {tc} fetch failed: {exc}") + ticker_empty += 1 + + if self.ticker_sleep > 0: + time.sleep(self.ticker_sleep) + + if (i + 1) % min(100, max(1, len(tickers) // 10)) == 0: + rows = sum(len(f) for f in all_frames) + elapsed = time.monotonic() - started_at + logger.info( + f"{self.spec.name}: ticker progress {i+1}/{len(tickers)} " + f"ok={ticker_ok} empty={ticker_empty} rows={rows} " + f"elapsed={_format_duration(elapsed)}" + ) + + elapsed = time.monotonic() - started_at + if not all_frames: + logger.info(f"{self.spec.name}: no data fetched, elapsed={_format_duration(elapsed)}") + rt.notifier.send(f"{self.spec.name} 同步完成 (ticker mode)\n无数据") + return + + combined = pd.concat(all_frames, ignore_index=True) + if "trade_date" in combined.columns: + combined["_date"] = combined["trade_date"].astype(str).str.replace("-", "") + combined = combined.drop_duplicates(subset=["ts_code", "trade_date"]) + + written = 0 + for date_val, grp in combined.groupby("_date", sort=True): + grp = grp.drop(columns=["_date"], errors="ignore") + self.store.write(date_val, grp) + written += 1 + + logger.info( + f"{self.spec.name}: done total={len(tickers)} tickers, " + f"written={written} partitions, {len(combined)} rows, " + f"elapsed={_format_duration(elapsed)}" + ) + rt.notifier.send( + f"{self.spec.name} 同步完成 (ticker mode)\n" + f"区间：{start} ~ {end}｜写入 {written} 天 / {len(combined)} 条" + ) + else: + logger.error(f"{self.spec.name}: no trade_date column in fetched data") + + class SnapshotSyncJob(SyncJob): def __init__( self, diff --git a/zer0share/sync/etf.py b/zer0share/sync/etf.py index a5943ed..d7be6db 100644 --- a/zer0share/sync/etf.py +++ b/zer0share/sync/etf.py @@ -8,7 +8,7 @@ FUND_DAILY_SPEC, ) from zer0share.storage import DailyPartitionStore, SnapshotStore -from zer0share.sync._jobs import DailySyncJob, SnapshotSyncJob, SyncJob +from zer0share.sync._jobs import DailySyncJob, SnapshotSyncJob, SyncJob, TickerSyncJob try: from zer0share.catalog import FUND_ADJ_SPEC @@ -16,15 +16,26 @@ FUND_ADJ_SPEC = SimpleNamespace(name="fund_adj", first_date="20100101") -def build_jobs(cfg, fetcher) -> list[SyncJob]: +def build_jobs(cfg, fetcher, ticker_mode: bool = False) -> list[SyncJob]: etf_dir = cfg.data_dir / "etf" - return [ - DailySyncJob( + fund_daily_job = ( + TickerSyncJob( + table_name=FUND_DAILY_SPEC.name, + spec=FUND_DAILY_SPEC, + fetch_range=fetcher.fetch_fund_daily_range, + store=DailyPartitionStore(etf_dir / "fund_daily"), + get_tickers=fetcher.get_etf_codes, + ) + if ticker_mode + else DailySyncJob( table_name=FUND_DAILY_SPEC.name, spec=FUND_DAILY_SPEC, fetch=fetcher.fetch_fund_daily, store=DailyPartitionStore(etf_dir / "fund_daily"), - ), + ) + ) + return [ + fund_daily_job, DailySyncJob( table_name=FUND_ADJ_SPEC.name, spec=FUND_ADJ_SPEC,