feat(sync): add per-ticker sync mode with relay/proxy API support + data coverage check

.gitignore

@@ -32,3 +32,4 @@ uv.lock
 
 # Claude local config
 .claude/
+scripts/

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,14 +140,15 @@ 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:
         raise click.UsageError("--end-date requires --start-date")
     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:

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"]),

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))

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
 

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,

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))