diff --git a/backend/app/application/services/data_sync.py b/backend/app/application/services/data_sync.py new file mode 100644 index 0000000..50fa67b --- /dev/null +++ b/backend/app/application/services/data_sync.py @@ -0,0 +1,611 @@ +"""数据同步服务:增量 + 新浪「两边一致」校验兜底(financial / daily)。 + +背景(AGENT.md §5/§7/§8): +- Tushare 是首选源;新浪财经只作备用。任何切源都必须可追溯(写 sync_log), + 且禁止静默把未经核验的备用源数据并入主库。 +- 本模块把「切到新浪」从 FailoverProvider 的『主源报错即兜底』收紧为 + 『校验兜底』:只有当某只股票**两边重叠的历史数据一致**时,才允许把新浪 + 的**新数据**(本地缺失键的行)导入;无本地历史或校验不一致 → 拒绝并告警, + 留待 Tushare 恢复后重跑补齐(数据真实性优先)。 + +校验口径(经验证,见仓库数据): +- 财务可比字段只有 eps / gross_margin —— 两源同报告期数值逐位一致; + ROE 两边口径不同(Tushare 摊薄 vs 新浪加权),不作为一致性依据。 +- 日线新浪为前复权,与本地不复权行仅「最近无除权区间」相等,因此只拿 + 两源重叠的最近若干个交易日做一致性校验(通道可信 → 才允许补缺)。 +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass, field +from datetime import date, timedelta +from decimal import Decimal + +from app.domain.entities.market import DailyBar, FinancialIndicator, SyncLog +from app.domain.providers import MarketDataProvider +from app.domain.repositories.market import ( + AdjustFactorRepository, + DailyBarRepository, + FinancialRepository, +) +from app.infrastructure.data_sources.errors import DataSourceAuthenticationError + +# 财务两源可比字段(其余字段两端口径不一致 / 单侧缺失,不能作校验依据) +FINANCIAL_COMPARE_FIELDS = ("eps", "gross_margin") +DAILY_COMPARE_FIELDS = ("open", "high", "low", "close") + +# 新浪日 K 可达窗口(getKLineData datalen=320 自然日) +SINA_KLINE_DAYS = 320 +# 校验回看:请求新浪时额外回看 begin 之前的天数,确保与本地近期历史有重叠可比 +SINA_VERIFY_LOOKBACK_DAYS = 45 +_EPOCH = date(1990, 1, 1) + + +# ---------------------------------------------------------------- 一致性校验(纯函数) + +@dataclass +class OverlapVerdict: + """两边重叠一致性结论。ok=True 才允许导入新浪新数据。""" + + ok: bool + shared: int = 0 # 重叠报告期 / 重叠交易日数量 + compared: int = 0 # 实际参与数值比较的行/日数量 + mismatches: list[str] = field(default_factory=list) + + def summary(self) -> str: + if self.ok: + return f"重叠 {self.shared} 项,数值一致(比较 {self.compared} 项)" + why = f"重叠 {self.shared} 项不足/为空" + if self.mismatches: + why = ";".join(self.mismatches[:3]) + return f"校验未通过:{why}" + + +def _close_enough(a: Decimal, b: Decimal, *, rel_tol: float, abs_tol: float) -> bool: + if a is None or b is None: + return False + diff = abs(a - b) + if diff <= Decimal(str(abs_tol)): + return True + scale = max(abs(a), abs(b)) + return diff <= Decimal(str(rel_tol)) * scale + + +def financial_overlap_consistent( + local_rows: Sequence[FinancialIndicator], + sina_rows: Sequence[FinancialIndicator], + *, + min_shared: int = 2, + rel_tol: float = 1e-4, + abs_tol: float = 1e-3, +) -> OverlapVerdict: + """新浪财务行与本地(Tushare)行按报告期重叠校验。 + + 新浪每个报告期只保留最新一版(getFinanceReport2022 的 report_list 按 + 报告期一份);本地同报告期可能有多版公告,取公告日最新者比较。 + 要求:重叠报告期数 >= min_shared,且全部可比字段(两源都非空)一致。 + """ + local_latest: dict[date, FinancialIndicator] = {} + for row in local_rows: + cur = local_latest.get(row.report_date) + if cur is None or row.announce_date > cur.announce_date: + local_latest[row.report_date] = row + sina_by_report = {row.report_date: row for row in sina_rows} + + verdict = OverlapVerdict(ok=False) + shared_dates = sorted(set(local_latest) & set(sina_by_report), reverse=True) + verdict.shared = len(shared_dates) + for report in shared_dates: + a = local_latest[report] + b = sina_by_report[report] + day_mismatch: list[str] = [] + compared = 0 + for f in FINANCIAL_COMPARE_FIELDS: + va, vb = getattr(a, f), getattr(b, f) + if va is None or vb is None: + continue + compared += 1 + if not _close_enough(va, vb, rel_tol=rel_tol, abs_tol=abs_tol): + day_mismatch.append(f"{report}: {f} {va}≠{vb}") + verdict.compared += compared + verdict.mismatches.extend(day_mismatch) + verdict.ok = ( + verdict.shared >= min_shared and verdict.compared > 0 and not verdict.mismatches + ) + return verdict + + +def daily_overlap_consistent( + local_bars: Sequence[DailyBar], + sina_bars: Sequence[DailyBar], + *, + min_shared: int = 3, + max_recent: int = 8, + rel_tol: float = 1e-4, + abs_tol: float = Decimal("0.02"), +) -> OverlapVerdict: + """新浪日 K(前复权)与本地(不复权)重叠校验。 + + 前复权锚定最新价:仅「最近一次除权之后」的交易日两源数值相等,因此只 + 比较两源重叠的、最近的 max_recent 个交易日(此时若有除权发生在该段, + 校验会判不一致 → 拒绝兜底,安全方向)。vol/amount 两源单位/口径不同, + 不参与比较。 + """ + local_by_day = {b.trade_date: b for b in local_bars} + sina_by_day = {b.trade_date: b for b in sina_bars} + shared = sorted(set(local_by_day) & set(sina_by_day), reverse=True) + + verdict = OverlapVerdict(ok=False) + verdict.shared = len(shared) + for day in shared[:max_recent]: + a, b = local_by_day[day], sina_by_day[day] + day_mismatch: list[str] = [] + compared = 0 + for f in DAILY_COMPARE_FIELDS: + va, vb = getattr(a, f), getattr(b, f) + if va is None or vb is None: + continue + compared += 1 + if not _close_enough(va, vb, rel_tol=rel_tol, abs_tol=abs_tol): + day_mismatch.append(f"{day}: {f} {va}≠{vb}") + verdict.compared += compared + verdict.mismatches.extend(day_mismatch) + checked = len(shared[:max_recent]) + verdict.ok = ( + checked >= min_shared and verdict.compared > 0 and not verdict.mismatches + ) + return verdict + + +# ---------------------------------------------------------------- 报告期披露节奏 + +def latest_expected_report_period(today: date) -> date: + """当前「应已披露」的最新报告期(报告期结束日)。 + + 用作财务增量的已最新判断:本地已含该报告期 → 该股票已跟进到最新一季, + 跳过(避免每轮全量重拉;--full 强制)。窗口按 A 股披露节奏划分: + - 1/1~2/14:年报季未开 → 上年三季报(09-30) + - 2/15~6/30:年报+一季报季 → 本年一季报(03-31) + - 7/1~10/15:半年报季 → 本年半年报(06-30) + - 10/16~12/31:三季报季 → 本年三季报(09-30) + """ + y = today.year + md = (today.month, today.day) + if md <= (2, 14): + return date(y - 1, 9, 30) + if md <= (6, 30): + return date(y, 3, 31) + if md <= (10, 15): + return date(y, 6, 30) + return date(y, 9, 30) + + +# ---------------------------------------------------------------- 审计 + +def _audit_sync( + audit: Callable[[SyncLog], None], + *, + source: str, + api: str, + success: bool, + row_count: int = 0, + reason: str | None = None, + start: date | None = None, + end: date | None = None, +) -> None: + audit( + SyncLog( + source=source, + api=api, + success=success, + failure_reason=reason, + row_count=row_count, + data_start=start, + data_end=end, + ) + ) + + +# ---------------------------------------------------------------- 结果结构 + +@dataclass +class FinancialSymbolResult: + """单只股票财务同步结果(status: skip|ok|sina|failed)。""" + + symbol: str + status: str + source: str | None = None # tushare | sina + fetched: int = 0 # 数据源返回行数 + written: int = 0 # 实际落库行数(新增;--full 时含更新) + updated: int = 0 # --full 下覆盖的既有行数 + report_first: date | None = None + report_last: date | None = None + announce_first: date | None = None + announce_last: date | None = None + notes: list[str] = field(default_factory=list) + + +@dataclass +class DailySymbolResult: + """单只股票日线同步结果(status: skip|ok|sina|failed)。""" + + symbol: str + status: str + source: str | None = None # tushare | sina + bars_fetched: int = 0 + bars_written: int = 0 + day_first: date | None = None + day_last: date | None = None + factors_written: int | None = None # None=未尝试(新浪兜底无因子) + notes: list[str] = field(default_factory=list) + + +# ---------------------------------------------------------------- 财务同步服务 + +class VerifiedFinancialSyncer: + """财务指标增量同步:Tushare 窗口化拉取 → 失败则新浪校验兜底。""" + + def __init__( + self, + *, + primary: MarketDataProvider, + fallback: MarketDataProvider | None, + repo: FinancialRepository, + audit: Callable[[SyncLog], None], + today: date | None = None, + min_shared: int = 2, + ) -> None: + self.primary = primary + self.fallback = fallback + self.repo = repo + self.audit = audit + self.today = today or date.today() + self.min_shared = min_shared + + def sync_symbol(self, symbol: str, *, force_full: bool = False) -> FinancialSymbolResult: + local = self.repo.list_symbol(symbol) + local_keys = {(r.symbol, r.report_date, r.announce_date) for r in local} + due = latest_expected_report_period(self.today) + if not force_full and local and any(r.report_date == due for r in local): + return FinancialSymbolResult( + symbol=symbol, + status="skip", + notes=[f"本地已含最新报告期 {due.isoformat()},跳过(--full 强制重拉)"], + ) + # 拉取窗口:有本地行则从最早本地报告期起(含更正/补缺),无则全历史; + # 上限到最新应披露报告期。 + hi = due + lo = min((r.report_date for r in local), default=None) or _EPOCH + try: + rows = self.primary.get_financial(symbol, lo, hi) + except DataSourceAuthenticationError: + # 凭证无效/接口无权限:属全局性故障,快速失败让用户修 token, + # 不要对全市场逐只做无意义的新浪试探 + raise + except Exception as exc: # noqa: BLE001 —— 与 FailoverProvider 一致,统一走审计 + _audit_sync( + self.audit, + source=self.primary.name, + api="get_financial", + success=False, + reason=str(exc), + start=lo, + end=hi, + ) + return self._sina_fallback(symbol, local, local_keys, primary_error=str(exc)) + _audit_sync( + self.audit, + source=self.primary.name, + api="get_financial", + success=True, + row_count=len(rows), + start=lo, + end=hi, + ) + if force_full: + to_write = rows + updated = sum(1 for r in rows if _fin_key(r) in local_keys) + else: + to_write = [r for r in rows if _fin_key(r) not in local_keys] + updated = 0 + written = self.repo.upsert_many(to_write) + return _fin_result(symbol, status="ok", source="tushare", written_rows=to_write, + written=written, updated=updated) + + # ---- 新浪校验兜底 ---- + + def _sina_fallback( + self, + symbol: str, + local: list[FinancialIndicator], + local_keys: set[tuple], + *, + primary_error: str, + ) -> FinancialSymbolResult: + if self.fallback is None: + return FinancialSymbolResult( + symbol=symbol, + status="failed", + notes=[f"Tushare 失败且未配置新浪兜底: {primary_error}"], + ) + if not local: + _audit_sync( + self.audit, + source=self.fallback.name, + api="get_financial", + success=False, + reason=f"无本地历史可做两边一致性校验,跳过待 Tushare 恢复重试({primary_error})", + ) + return FinancialSymbolResult( + symbol=symbol, + status="failed", + source="sina", + notes=[ + f"Tushare 失败且本地无历史({symbol}),无法确认新浪数据真实性," + f"跳过待重试。primary: {primary_error}" + ], + ) + try: + sina_rows = self.fallback.get_financial(symbol) + except Exception as exc: # noqa: BLE001 + _audit_sync( + self.audit, + source=self.fallback.name, + api="get_financial", + success=False, + reason=f"primary: {primary_error}; fallback: {exc}", + ) + return FinancialSymbolResult( + symbol=symbol, + status="failed", + source="sina", + notes=[f"主备数据源均失败: primary={primary_error}; sina={exc}"], + ) + verdict = financial_overlap_consistent(local, sina_rows, min_shared=self.min_shared) + if not verdict.ok: + _audit_sync( + self.audit, + source=self.fallback.name, + api="get_financial", + success=False, + reason=f"{verdict.summary()}(新浪返回 {len(sina_rows)} 行; primary={primary_error})", + ) + return FinancialSymbolResult( + symbol=symbol, + status="failed", + source="sina", + notes=[ + f"新浪数据与本地历史不一致/无法校验({symbol}),拒绝导入。" + f"primary: {primary_error};{verdict.summary()}" + ], + ) + new_rows = [r for r in sina_rows if _fin_key(r) not in local_keys] + written = self.repo.upsert_many(new_rows) + _audit_sync( + self.audit, + source=self.fallback.name, + api="get_financial", + success=True, + row_count=written, + ) + return _fin_result(symbol, status="sina", source="sina", written_rows=new_rows, + written=written, updated=0, + note=f"新浪校验通过后补入 {written} 行(仅本地缺失键,source=sina)") + + +# ---------------------------------------------------------------- 日线同步服务 + +class VerifiedDailySyncer: + """日线同步:Tushare 失败 → 新浪校验兜底(仅补缺失交易日、无复权因子)。""" + + def __init__( + self, + *, + primary: MarketDataProvider, + fallback: MarketDataProvider | None, + bars: DailyBarRepository, + factors: AdjustFactorRepository, + audit: Callable[[SyncLog], None], + today: date | None = None, + min_shared: int = 3, + ) -> None: + self.primary = primary + self.fallback = fallback + self.bars = bars + self.factors = factors + self.audit = audit + self.today = today or date.today() + self.min_shared = min_shared + + def sync_symbol(self, symbol: str, begin: date, end: date) -> DailySymbolResult: + try: + bars = self.primary.get_daily(symbol, begin, end) + except DataSourceAuthenticationError: + raise # 凭证/权限故障 → 快速失败(见财务同步注释) + except Exception as exc: # noqa: BLE001 + _audit_sync( + self.audit, + source=self.primary.name, + api="get_daily", + success=False, + reason=str(exc), + start=begin, + end=end, + ) + return self._sina_fallback(symbol, begin, end, primary_error=str(exc)) + _audit_sync( + self.audit, + source=self.primary.name, + api="get_daily", + success=True, + row_count=len(bars), + start=begin, + end=end, + ) + try: + factors = self.primary.get_adjust_factor(symbol, begin, end) + except DataSourceAuthenticationError: + raise # 凭证/权限故障 → 快速失败(见财务同步注释) + except Exception as exc: # noqa: BLE001 + # 复权因子是日线配套:缺因子不写本段,避免 resume 按日线已最新而跳过、因子永远补不上 + _audit_sync( + self.audit, + source=self.primary.name, + api="get_adjust_factor", + success=False, + reason=str(exc), + start=begin, + end=end, + ) + return DailySymbolResult( + symbol=symbol, + status="failed", + source="tushare", + bars_fetched=len(bars), + notes=[f"日线拉取成功但复权因子失败,本段未落库(防因子缺口): {exc}"], + ) + _audit_sync( + self.audit, + source=self.primary.name, + api="get_adjust_factor", + success=True, + row_count=len(factors), + start=begin, + end=end, + ) + self.bars.upsert_many(bars) + self.factors.upsert_many(factors) + return DailySymbolResult( + symbol=symbol, + status="ok", + source="tushare", + bars_fetched=len(bars), + bars_written=len(bars), + factors_written=len(factors), + day_first=min((b.trade_date for b in bars), default=None), + day_last=max((b.trade_date for b in bars), default=None), + ) + + # ---- 新浪校验兜底 ---- + + def _sina_fallback( + self, symbol: str, begin: date, end: date, *, primary_error: str + ) -> DailySymbolResult: + if self.fallback is None: + return DailySymbolResult( + symbol=symbol, + status="failed", + notes=[f"Tushare 失败且未配置新浪兜底: {primary_error}"], + ) + # 新浪只有最近 SINA_KLINE_DAYS 自然日数据;为拿到「本地近期历史」重叠做 + # 校验,请求窗口需回看 begin 之前 SINA_VERIFY_LOOKBACK_DAYS 天 + # (见 daily_overlap_consistent:只比较两源重叠的最近交易日)。 + q_start = max( + self.today - timedelta(days=SINA_KLINE_DAYS - 1), + begin - timedelta(days=SINA_VERIFY_LOOKBACK_DAYS), + ) + try: + sina_bars = self.fallback.get_daily(symbol, q_start, end) + except Exception as exc: # noqa: BLE001 + _audit_sync( + self.audit, + source=self.fallback.name, + api="get_daily", + success=False, + reason=f"primary: {primary_error}; fallback: {exc}", + start=q_start, + end=end, + ) + return DailySymbolResult( + symbol=symbol, + status="failed", + source="sina", + notes=[f"主备数据源均失败: primary={primary_error}; sina={exc}"], + ) + local_recent = self.bars.get_range(symbol, q_start, end) + verdict = daily_overlap_consistent(local_recent, sina_bars, min_shared=self.min_shared) + if not verdict.ok: + _audit_sync( + self.audit, + source=self.fallback.name, + api="get_daily", + success=False, + reason=f"{verdict.summary()}(新浪返回 {len(sina_bars)} 行; primary={primary_error})", + start=q_start, + end=end, + ) + return DailySymbolResult( + symbol=symbol, + status="failed", + source="sina", + bars_fetched=len(sina_bars), + notes=[ + f"新浪数据与本地历史不一致/无法校验({symbol}),拒绝兜底补缺。" + f"primary: {primary_error};{verdict.summary()}" + ], + ) + local_dates = {b.trade_date for b in local_recent} + new_bars = [ + b + for b in sina_bars + if begin <= b.trade_date <= end and b.trade_date not in local_dates + ] + written = self.bars.upsert_many(new_bars) + _audit_sync( + self.audit, + source=self.fallback.name, + api="get_daily", + success=True, + row_count=written, + start=begin, + end=end, + ) + return DailySymbolResult( + symbol=symbol, + status="sina", + source="sina", + bars_fetched=len(sina_bars), + bars_written=written, + day_first=min((b.trade_date for b in new_bars), default=None), + day_last=max((b.trade_date for b in new_bars), default=None), + factors_written=None, + notes=[ + f"新浪校验通过,仅补本地缺失交易日 {written} 根(前复权 source=sina," + f"无复权因子;Tushare 恢复后 --resume 会按日覆盖回不复权口径)" + ], + ) + + +# ---------------------------------------------------------------- 小工具 + +def _fin_key(row: FinancialIndicator) -> tuple: + return (row.symbol, row.report_date, row.announce_date) + + +def _fin_result( + symbol: str, + *, + status: str, + source: str, + written_rows: Sequence[FinancialIndicator], + written: int, + updated: int, + note: str | None = None, +) -> FinancialSymbolResult: + reports = [r.report_date for r in written_rows] + announces = [r.announce_date for r in written_rows] + notes = [note] if note else [] + return FinancialSymbolResult( + symbol=symbol, + status=status, + source=source, + fetched=len(written_rows), + written=written, + updated=updated, + report_first=min(reports, default=None), + report_last=max(reports, default=None), + announce_first=min(announces, default=None), + announce_last=max(announces, default=None), + notes=notes, + ) diff --git a/backend/app/cli/sync.py b/backend/app/cli/sync.py index 605c171..191f53f 100644 --- a/backend/app/cli/sync.py +++ b/backend/app/cli/sync.py @@ -1,28 +1,43 @@ -"""Phase 1 数据同步 CLI(Tushare 首选 → SQLite)。 +"""Phase 1 数据同步 CLI(Tushare 首选 → SQLite,新浪校验兜底)。 用法(cd backend): uv run python -m app.cli.sync basic uv run python -m app.cli.sync calendar --start 20240101 --end 20241231 uv run python -m app.cli.sync daily --symbols 600519.SH,000001.SZ --start 20240101 uv run python -m app.cli.sync daily --all --start 20240101 # 全市场 - uv run python -m app.cli.sync financial --all + uv run python -m app.cli.sync financial --all # 财务指标(增量) + uv run python -m app.cli.sync financial --all --full # 财务指标(强制全量重拉) uv run python -m app.cli.sync verify --symbol 600519.SH # 新浪交叉验证 +增量与兜底: +- daily --resume:从本地最新交易日续传(已有);Tushare 失败时走新浪校验兜底, + 只有「两源重叠历史一致」才用新浪补本地缺失交易日(source=sina/前复权)。 +- financial:默认增量——本地已含最新应披露报告期则跳过;Tushare 失败时新浪 + 数据须通过「两边一致」校验(重叠报告期 eps/销售毛利率逐期一致)才允许补入 + 本地缺失键(source=sina)。失败股票留待下轮重跑补齐,不会静默导入未核验数据。 +- 每次拉取写入 sync_log 审计(来源 / 成功与否 / 行数 / 区间),禁止静默切源。 + 本模块是组装层(composition root):在此装配 Provider / Repository / Session, -业务层代码仍只依赖抽象(domain.repositories / domain.providers)。 +业务逻辑在 application.services.data_sync,业务层仍只依赖抽象。 """ from __future__ import annotations import argparse import sys +import time from datetime import date, datetime, timedelta from sqlalchemy import select +from app.application.services.data_sync import ( + DailySymbolResult, + FinancialSymbolResult, + VerifiedDailySyncer, + VerifiedFinancialSyncer, +) from app.core.config import get_settings from app.infrastructure.data_sources.errors import DataSourceError -from app.infrastructure.data_sources.failover import FailoverProvider from app.infrastructure.data_sources.sina import SinaProvider from app.infrastructure.data_sources.tushare import TushareProvider from app.infrastructure.persistence.sqlalchemy.models.market import StockModel @@ -44,12 +59,12 @@ def _parse_day(text: str) -> date: def _failover_provider(session): - """Tushare 首选 + 新浪兜底(FailoverProvider 每次尝试写 sync_log,AGENT.md §7)。 + """Tushare 首选 + 新浪兜底(basic/calendar 用;daily/financial 走校验兜底服务)。 - 能力矩阵:新浪仅提供日线(前复权,标记 source=sina/adjust=qfq);其余接口 - (adj_factor/financial/basic/calendar)新浪不支持 → 抛错保留单源语义,日志可见。 - Tushare 恢复后重跑 --resume 会把新浪兜底日按同键覆盖回不复权口径。 + FailoverProvider 每次尝试写 sync_log(AGENT.md §7)。能力矩阵:新浪仅提供 + 日线/财务,basic/calendar 新浪不支持 → 抛错保留单源语义,日志可见。 """ + from app.infrastructure.data_sources.failover import FailoverProvider from app.infrastructure.data_sources.sina import SinaProvider audit_repo = SqlAlchemySyncLogRepository(session) @@ -97,6 +112,23 @@ def _symbols_of(args) -> list[str]: return [s.strip() for s in args.symbols.split(",") if s.strip()] +def _stock_names(session, symbols: list[str]) -> dict[str, str]: + """一次性取出股票名称(进度描述用);批量查询避开 SQLite 变量上限。""" + names: dict[str, str] = {} + for i in range(0, len(symbols), 500): + chunk = symbols[i : i + 500] + rows = session.execute( + select(StockModel.symbol, StockModel.name).where(StockModel.symbol.in_(chunk)) + ) + names.update({sym: nm for sym, nm in rows}) + return names + + +def _warn_notes(notes: list[str]) -> None: + for note in notes: + print(f" [warn] {note}", file=sys.stderr) + + def cmd_daily(args) -> int: from sqlalchemy import func @@ -105,12 +137,20 @@ def cmd_daily(args) -> int: symbols = _symbols_of(args) start = _parse_day(args.start) if args.start else date(2005, 1, 1) end = _parse_day(args.end) if args.end else date.today() - total = 0 - skipped = 0 + started = time.monotonic() + n_ok = n_sina = n_failed = n_skip = 0 + rows_tushare = rows_sina = 0 with _session_ctx() as session: - provider = _failover_provider(session) + names = _stock_names(session, symbols) + audit = SqlAlchemySyncLogRepository(session).add + syncer = VerifiedDailySyncer( + primary=TushareProvider(token=get_settings().tushare_token), + fallback=SinaProvider(), + bars=SqlAlchemyDailyBarRepository(session), + factors=SqlAlchemyAdjustFactorRepository(session), + audit=audit, + ) bar_repo = SqlAlchemyDailyBarRepository(session) - factor_repo = SqlAlchemyAdjustFactorRepository(session) # 增量基准:本地数据已到该日期即视为「已最新」,resume 时不再调 API global_latest = ( session.scalar(select(func.max(StockDailyModel.trade_date))) if args.resume else None @@ -121,43 +161,119 @@ def cmd_daily(args) -> int: latest = bar_repo.latest_date(symbol) if latest is not None: if global_latest is not None and latest >= global_latest: - skipped += 1 # 已同步到本地最新交易日,无需续拉 + n_skip += 1 # 已同步到本地最新交易日,无需续拉 continue begin = max(begin, latest + timedelta(days=1)) if begin > end: - skipped += 1 # 无待拉区间(如区间已含在本地) + n_skip += 1 # 无待拉区间(如区间已含在本地) continue - try: - bars = provider.get_daily(symbol, begin, end) - factors = provider.get_adjust_factor(symbol, begin, end) - bar_repo.upsert_many(bars) - factor_repo.upsert_many(factors) - total += len(bars) - if i % 100 == 0: - session.commit() - print(f" ... {i}/{len(symbols)} {symbol} 累计 {total} 根") - except DataSourceError as exc: - print(f" [warn] {symbol} 拉取失败: {exc}", file=sys.stderr) - session.commit() - detail = f"[daily] {len(symbols)} 只股票合计写入 {total} 根日线(含复权因子)" + if getattr(args, "sleep", 0) > 0: + time.sleep(args.sleep) + res: DailySymbolResult = syncer.sync_symbol(symbol, begin, end) + session.commit() # 逐只落库:中断/报错只丢当前一只,重跑增量续传 + if res.status == "ok": + n_ok += 1 + rows_tushare += res.bars_written + elif res.status == "sina": + n_sina += 1 + rows_sina += res.bars_written + elif res.status == "failed": + n_failed += 1 + _warn_notes(res.notes) + if i % 100 == 0: + name = names.get(symbol, "") + print( + f" ... {i}/{len(symbols)} {symbol} {name}: " + f"累计 tushare {rows_tushare} 根 + 新浪补缺 {rows_sina} 根;" + f"成功 {n_ok} / 新浪 {n_sina} / 失败待重试 {n_failed}" + ) + elapsed = time.monotonic() - started + detail = ( + f"[daily] {len(symbols)} 只股票:成功 {n_ok} / 新浪校验补缺 {n_sina} / " + f"已最新跳过 {n_skip} / 失败待重试 {n_failed}" + ) if args.resume: - detail += f";增量跳过(已最新){skipped} 只(本地最新 {global_latest})" + detail += f"(本地最新 {global_latest})" + detail += f";写入 {rows_tushare} 根(tushare 不复权)+ {rows_sina} 根(sina 前复权),耗时 {elapsed:.0f}s" print(detail) return 0 +def _fin_progress_line(i: int, n: int, symbol: str, name: str, res: FinancialSymbolResult) -> str: + """financial 逐只进度行:结果 + 导入内容简单描述(报告期/公告区间、来源)。""" + head = f"[financial {i}/{n}] {symbol} {name or ''}".rstrip() + if res.status == "skip": + return f"{head}:已最新,跳过(增量)" + if res.status == "failed": + return f"{head}:失败待重试(tushare 失败;新浪源 {'未通过校验' if res.source == 'sina' else '不可用'})" + if res.status == "sina": + return ( + f"{head}:tushare 失败 → 新浪校验通过,补入 {res.written} 行(source=sina)" + + _fin_span(res) + ) + # status == ok(tushare 成功) + if res.written: + updated = f",覆盖更新 {res.updated} 行" if res.updated else "" + return f"{head}:tushare 返回 {res.fetched} 行 → 新增 {res.written} 行{updated}" + _fin_span(res) + return f"{head}:tushare 返回 {res.fetched} 行,均已在库,无新增" + + +def _fin_span(res: FinancialSymbolResult) -> str: + if not res.written or res.report_first is None: + return "" + if res.announce_first is None or res.announce_last is None: + return "" + return ( + f";报告期 {res.report_first.isoformat()}~{res.report_last.isoformat()}" + f"(公告 {res.announce_first.isoformat()}~{res.announce_last.isoformat()})" + ) + + def cmd_financial(args) -> int: symbols = _symbols_of(args) - total = 0 + started = time.monotonic() + n_ok = n_sina = n_failed = n_skip = 0 + rows_tushare = rows_sina = 0 with _session_ctx() as session: - provider = _failover_provider(session) - fin_repo = SqlAlchemyFinancialRepository(session) - for symbol in symbols: - rows = provider.get_financial(symbol) - fin_repo.upsert_many(rows) - total += len(rows) - session.commit() - print(f"[financial] {len(symbols)} 只股票合计写入 {total} 条财务指标快照") + names = _stock_names(session, symbols) + audit = SqlAlchemySyncLogRepository(session).add + syncer = VerifiedFinancialSyncer( + primary=TushareProvider(token=get_settings().tushare_token), + fallback=SinaProvider(), + repo=SqlAlchemyFinancialRepository(session), + audit=audit, + ) + for i, symbol in enumerate(symbols, start=1): + if getattr(args, "sleep", 0) > 0: + time.sleep(args.sleep) + res: FinancialSymbolResult = syncer.sync_symbol(symbol, force_full=args.full) + session.commit() # 逐只落库:中断只丢当前一只,重跑增量续传 + print(_fin_progress_line(i, len(symbols), symbol, names.get(symbol, ""), res)) + _warn_notes(res.notes) + if res.status == "ok": + n_ok += 1 + rows_tushare += res.written + elif res.status == "sina": + n_sina += 1 + rows_sina += res.written + elif res.status == "failed": + n_failed += 1 + elif res.status == "skip": + n_skip += 1 + elapsed = time.monotonic() - started + mode = "全量重拉(--full)" if args.full else "增量" + print( + f"[financial] 共 {len(symbols)} 只({mode}):成功 {n_ok} / 新浪校验兜底 {n_sina} / " + f"已最新跳过 {n_skip} / 失败待重试 {n_failed};" + f"合计写入 {rows_tushare + rows_sina} 行(tushare {rows_tushare} + sina {rows_sina})," + f"耗时 {elapsed:.0f}s" + ) + if n_failed: + print( + " [tip] 失败股票未写入未核验数据,重跑本命令即可续传补齐;" + "若因频率超限,可用 --sleep 加大间隔(如 --sleep 60)分多次跑。", + file=sys.stderr, + ) return 0 @@ -253,17 +369,34 @@ def build_parser() -> argparse.ArgumentParser: p_cal.add_argument("--end", required=True, help="YYYYMMDD") p_cal.set_defaults(func=cmd_calendar) - p_daily = sub.add_parser("daily", help="同步日线与复权因子") + p_daily = sub.add_parser("daily", help="同步日线与复权因子(Tushare 失败 → 新浪校验兜底补缺)") p_daily.add_argument("--symbols", default="", help="600519.SH,000001.SZ") p_daily.add_argument("--all", action="store_true", help="遍历 stock 表全部股票") p_daily.add_argument("--start", default="", help="YYYYMMDD(默认 20050101)") p_daily.add_argument("--end", default="", help="YYYYMMDD(默认今天)") - p_daily.add_argument("--resume", action="store_true", help="从本地最新交易日续传") + p_daily.add_argument("--resume", action="store_true", help="从本地最新交易日续传(增量)") + p_daily.add_argument( + "--sleep", + type=float, + default=0, + help="每只股票请求间隔秒数(限速时加大,如 1 或 60)", + ) p_daily.set_defaults(func=cmd_daily) - p_fin = sub.add_parser("financial", help="同步财务指标快照") + p_fin = sub.add_parser("financial", help="同步财务指标快照(默认增量;Tushare 失败 → 新浪校验兜底)") p_fin.add_argument("--symbols", default="") - p_fin.add_argument("--all", action="store_true") + p_fin.add_argument("--all", action="store_true", help="遍历 stock 表全部股票") + p_fin.add_argument( + "--full", + action="store_true", + help="强制全量重拉并覆盖既有行(默认只补本地缺失/更新的报告期,已最新跳过)", + ) + p_fin.add_argument( + "--sleep", + type=float, + default=0, + help="每只股票请求间隔秒数(限速时加大,如 1 或 60)", + ) p_fin.set_defaults(func=cmd_financial) p_verify = sub.add_parser("verify", help="新浪交叉验证最新行情") diff --git a/backend/app/domain/entities/market.py b/backend/app/domain/entities/market.py index 810a514..082c9d6 100644 --- a/backend/app/domain/entities/market.py +++ b/backend/app/domain/entities/market.py @@ -83,11 +83,18 @@ class FinancialIndicator(BaseModel): 可见性红线:研究侧查询一律按 announce_date <= as_of_date 过滤, report_date 只表示报告所属期间,不代表公开时间。 + + source 标记数据来源:tushare(首选,字段全)| sina(兜底,字段 + 可能不全——新浪关键指标只含 eps/roe/gross_margin 等少数项)。 + 新浪兜底行只在「该股票本地历史与新浪重叠部分两边一致」通过校验后 + 才导入(见 application/services/data_sync.py),且只补本地缺失键。 + 研究侧对同一报告期应优先消费 source=tushare 的行。 """ symbol: str report_date: date announce_date: date + source: str = Field(default="tushare", description="tushare | sina") eps: Decimal | None = None roe: Decimal | None = None total_revenue: Decimal | None = None diff --git a/backend/app/domain/providers.py b/backend/app/domain/providers.py index e6df29c..4d1ba81 100644 --- a/backend/app/domain/providers.py +++ b/backend/app/domain/providers.py @@ -37,4 +37,15 @@ class MarketDataProvider(Protocol): def get_adjust_factor(self, symbol: str, start: date, end: date) -> list[AdjustFactor]: ... - def get_financial(self, symbol: str) -> list[FinancialIndicator]: ... + def get_financial( + self, + symbol: str, + start: date | None = None, + end: date | None = None, + ) -> list[FinancialIndicator]: + """财务指标快照。 + + start/end 为**报告期**窗口(对应 Tushare fina_indicator 的 + start_date/end_date 参数,按报告期过滤);不传表示全量历史。 + 新浪接口不支持按窗口拉取,提供方会忽略窗口后由调用方自行过滤。 + """ diff --git a/backend/app/domain/repositories/market.py b/backend/app/domain/repositories/market.py index b24ed0c..35fda1f 100644 --- a/backend/app/domain/repositories/market.py +++ b/backend/app/domain/repositories/market.py @@ -72,6 +72,12 @@ class AdjustFactorRepository(Protocol): class FinancialRepository(Protocol): def upsert_many(self, rows: Sequence[FinancialIndicator]) -> int: ... + def list_symbol(self, symbol: str) -> list[FinancialIndicator]: + """该股票本地全部财务行(增量判断 / 新浪校验重叠用,量级小)。""" + + def has_report_period(self, symbol: str, report_date: date) -> bool: + """本地是否已含该报告期(最新应披露报告期是否已入库)。""" + def list_announced( self, symbol: str, diff --git a/backend/app/infrastructure/data_sources/failover.py b/backend/app/infrastructure/data_sources/failover.py index 64bcd31..39d42d5 100644 --- a/backend/app/infrastructure/data_sources/failover.py +++ b/backend/app/infrastructure/data_sources/failover.py @@ -69,11 +69,15 @@ class FailoverProvider: fallback_call=lambda: self.fallback.get_adjust_factor(symbol, start, end), ) - def get_financial(self, symbol: str) -> list: + def get_financial( + self, symbol: str, start: date | None = None, end: date | None = None + ) -> list: return self._with_failover( "get_financial", - primary_call=lambda: self.primary.get_financial(symbol), - fallback_call=lambda: self.fallback.get_financial(symbol), + start=start, + end=end, + primary_call=lambda: self.primary.get_financial(symbol, start, end), + fallback_call=lambda: self.fallback.get_financial(symbol, start, end), ) # ---- 内部 ---- diff --git a/backend/app/infrastructure/data_sources/sina.py b/backend/app/infrastructure/data_sources/sina.py index ed68b86..a9a03b6 100644 --- a/backend/app/infrastructure/data_sources/sina.py +++ b/backend/app/infrastructure/data_sources/sina.py @@ -128,8 +128,17 @@ class SinaProvider: # ---- 财务(兜底 Tushare fina_indicator) ---- - def get_financial(self, symbol: str) -> list[FinancialIndicator]: - """新浪关键指标(source=gjzb),含披露日 publish_date → announce_date。""" + def get_financial( + self, + symbol: str, + start: date | None = None, + end: date | None = None, + ) -> list[FinancialIndicator]: + """新浪关键指标(source=gjzb),含披露日 publish_date → announce_date。 + + 新浪不支持按报告期窗口拉取:忽略 start/end 时返回其全部返回的 + 报告期;传入窗口则按 report_date 客户端过滤(新浪行 source=sina)。 + """ params = { "paperCode": _to_sina_symbol(symbol), "source": "gjzb", @@ -153,6 +162,11 @@ class SinaProvider: announce = entry.get("publish_date") if not announce: continue # 无披露日不可用于研究(防未来函数) + report_day = _to_date(str(rd)) + if start is not None and report_day < start: + continue + if end is not None and report_day > end: + continue fields: dict[str, Decimal | None] = { "eps": None, "roe": None, @@ -167,8 +181,9 @@ class SinaProvider: rows.append( FinancialIndicator( symbol=symbol, - report_date=_to_date(str(rd)), + report_date=report_day, announce_date=_to_date(str(announce)), + source="sina", eps=fields["eps"], roe=fields["roe"], total_revenue=fields["total_revenue"], diff --git a/backend/app/infrastructure/data_sources/tushare.py b/backend/app/infrastructure/data_sources/tushare.py index baa2bb9..fb6bc4a 100644 --- a/backend/app/infrastructure/data_sources/tushare.py +++ b/backend/app/infrastructure/data_sources/tushare.py @@ -10,7 +10,7 @@ from __future__ import annotations import importlib import logging import time -from datetime import date, datetime +from datetime import date, datetime, timedelta from decimal import Decimal from typing import Any @@ -183,9 +183,42 @@ class TushareProvider: ) return self.normalize_adj_factor(records) - def get_financial(self, symbol: str) -> list[FinancialIndicator]: - records = self._call("fina_indicator", ts_code=symbol) - return self.normalize_financial(records) + def get_financial( + self, + symbol: str, + start: date | None = None, + end: date | None = None, + ) -> list[FinancialIndicator]: + """fina_indicator:报告期窗口 + 100 条/请求上限自动分页。 + + Tushare 单次请求最多返回 100 条(超出按最新 100 条截断),因此 + 全量历史必须按报告期窗口回卷分页,否则老报告期会被静默丢弃。 + """ + lo = start or date(1990, 1, 1) + hi = end or date.today() + raw: list[dict[str, Any]] = [] + while lo <= hi: + batch = self._call( + "fina_indicator", + ts_code=symbol, + start_date=lo.strftime(_TS_DATE), + end_date=hi.strftime(_TS_DATE), + ) + raw += batch + if len(batch) < 100: + break + ends = [ + datetime.strptime(str(r["end_date"])[:8], _TS_DATE).date() + for r in batch + if r.get("end_date") + ] + if not ends: + break + next_hi = min(ends) - timedelta(days=1) + if next_hi < lo: # 无进展保护(边界簇被截断等极端情况) + break + hi = next_hi + return self.normalize_financial(raw) # ---- 内部 ---- diff --git a/backend/app/infrastructure/persistence/migrations/versions/d3f6c9a21b04_financial_source_marker.py b/backend/app/infrastructure/persistence/migrations/versions/d3f6c9a21b04_financial_source_marker.py new file mode 100644 index 0000000..2cf5edf --- /dev/null +++ b/backend/app/infrastructure/persistence/migrations/versions/d3f6c9a21b04_financial_source_marker.py @@ -0,0 +1,33 @@ +"""financial_indicator 增加 source 来源标记 + +Revision ID: d3f6c9a21b04 +Revises: 91c4e27a03fb +Create Date: 2026-09-08 + +新浪校验兜底导入的财务行带 source=sina(字段可能不全),与 Tushare +首选行区分;现有行回填默认 tushare(SQLite ADD COLUMN 带常量默认值, +不重写现有数据)。AGENT.md §7 数据来源可追溯。 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "d3f6c9a21b04" +down_revision: str | None = "91c4e27a03fb" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "financial_indicator", + sa.Column("source", sa.String(length=16), nullable=False, server_default="tushare"), + ) + + +def downgrade() -> None: + op.drop_column("financial_indicator", "source") diff --git a/backend/app/infrastructure/persistence/sqlalchemy/models/market.py b/backend/app/infrastructure/persistence/sqlalchemy/models/market.py index 9fe947d..ef976d1 100644 --- a/backend/app/infrastructure/persistence/sqlalchemy/models/market.py +++ b/backend/app/infrastructure/persistence/sqlalchemy/models/market.py @@ -93,6 +93,9 @@ class FinancialIndicatorModel(Base): symbol: Mapped[str] = mapped_column(String(SYMBOL_LEN), index=True) report_date: Mapped[date] = mapped_column(Date, index=True) announce_date: Mapped[date] = mapped_column(Date, index=True) + source: Mapped[str] = mapped_column( + String(16), default="tushare", server_default="tushare" + ) eps: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True) roe: Mapped[Decimal | None] = mapped_column(Numeric(10, 4), nullable=True) total_revenue: Mapped[Decimal | None] = mapped_column(Numeric(24, 2), nullable=True) diff --git a/backend/app/infrastructure/persistence/sqlalchemy/repositories/market_impl.py b/backend/app/infrastructure/persistence/sqlalchemy/repositories/market_impl.py index c6f2bb0..8b8608b 100644 --- a/backend/app/infrastructure/persistence/sqlalchemy/repositories/market_impl.py +++ b/backend/app/infrastructure/persistence/sqlalchemy/repositories/market_impl.py @@ -241,6 +241,25 @@ class SqlAlchemyFinancialRepository: def upsert_many(self, rows: Sequence[FinancialIndicator]) -> int: return _upsert_by_business_key(self._session, FinancialIndicator, rows) + def list_symbol(self, symbol: str) -> list[FinancialIndicator]: + rows = self._session.scalars( + select(FinancialIndicatorModel) + .where(FinancialIndicatorModel.symbol == symbol) + .order_by(FinancialIndicatorModel.announce_date, FinancialIndicatorModel.report_date) + ).all() + return [FinancialIndicator.model_validate(r, from_attributes=True) for r in rows] + + def has_report_period(self, symbol: str, report_date: date) -> bool: + exists = self._session.scalar( + select(FinancialIndicatorModel.id) + .where( + FinancialIndicatorModel.symbol == symbol, + FinancialIndicatorModel.report_date == report_date, + ) + .limit(1) + ) + return exists is not None + def list_announced( self, symbol: str, diff --git a/backend/tests/test_data_sync.py b/backend/tests/test_data_sync.py new file mode 100644 index 0000000..116be8d --- /dev/null +++ b/backend/tests/test_data_sync.py @@ -0,0 +1,450 @@ +"""增量 + 新浪「两边一致」校验兜底同步服务测试。 + +覆盖:一致性校验纯函数、报告期节奏、财务/日线同步服务 +(temp SQLite + Fake Provider,不触网)。 +""" + +from __future__ import annotations + +from datetime import date +from decimal import Decimal + +import pytest +from app.application.services.data_sync import ( + VerifiedDailySyncer, + VerifiedFinancialSyncer, + daily_overlap_consistent, + financial_overlap_consistent, + latest_expected_report_period, +) +from app.domain.entities.market import DailyBar, FinancialIndicator, SyncLog +from app.infrastructure.data_sources.errors import DataSourceError +from app.infrastructure.persistence.sqlalchemy.base import Base +from app.infrastructure.persistence.sqlalchemy.models.market import ( + AdjustFactorModel, + FinancialIndicatorModel, + StockDailyModel, +) +from app.infrastructure.persistence.sqlalchemy.repositories.market_impl import ( + SqlAlchemyAdjustFactorRepository, + SqlAlchemyDailyBarRepository, + SqlAlchemyFinancialRepository, +) +from sqlalchemy import create_engine, func, select +from sqlalchemy.orm import Session + +TODAY = date(2026, 9, 8) +DUE = date(2026, 6, 30) # TODAY 属于半年报季 → 应披露最新报告期 + + +@pytest.fixture() +def session(tmp_path) -> Session: + engine = create_engine(f"sqlite:///{tmp_path / 'ds.db'}", future=True) + Base.metadata.create_all(engine) + with Session(engine) as session: + yield session + + +def _fin( + symbol: str, + report: str, + announce: str, + *, + eps: str = "1.00", + roe: str | None = None, + gross: str | None = None, + source: str = "tushare", +) -> FinancialIndicator: + return FinancialIndicator( + symbol=symbol, + report_date=date.fromisoformat(report), + announce_date=date.fromisoformat(announce), + source=source, + eps=Decimal(eps), + roe=Decimal(roe) if roe is not None else None, + gross_margin=Decimal(gross) if gross is not None else None, + ) + + +def _bar(symbol: str, day: str, close: str, source: str = "tushare") -> DailyBar: + return DailyBar( + symbol=symbol, + trade_date=date.fromisoformat(day), + source=source, + adjust="none" if source == "tushare" else "qfq", + open=Decimal(close), + high=Decimal(close), + low=Decimal(close), + close=Decimal(close), + volume=Decimal("1000"), + ) + + +# ================================================================ 一致性校验 + +class TestFinancialConsistency: + def test_overlap_consistent_passes(self) -> None: + local = [ + _fin("600519.SH", "2026-03-31", "2026-04-25", eps="21.76", roe="10.5687", gross="89.7592"), + _fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57", roe="17.9543", gross="89.5552"), + ] + sina = [ + _fin("600519.SH", "2026-03-31", "2026-04-25", eps="21.76", roe="10.57", gross="89.759217", source="sina"), + _fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57", roe="16.75", gross="89.555212", source="sina"), + ] + verdict = financial_overlap_consistent(local, sina) + # ROE 两边口径不同(新浪加权 vs tushare 摊薄),不参与比较也不判失败 + assert verdict.ok + assert verdict.shared == 2 + assert not verdict.mismatches + + def test_roe_difference_not_compared(self) -> None: + """口径差异字段绝不能导致误杀(17.9543 vs 16.75 是同源数据真实形态)。""" + local = [_fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57", roe="17.9543", gross="89.5552")] + sina = [_fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57", roe="16.75", gross="89.555212", source="sina")] + verdict = financial_overlap_consistent(local, sina, min_shared=1) + assert verdict.ok + + def test_eps_mismatch_fails(self) -> None: + local = [_fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57", gross="89.5552")] + sina = [_fin("600519.SH", "2026-06-30", "2026-08-15", eps="99.99", gross="89.555212", source="sina")] + verdict = financial_overlap_consistent(local, sina, min_shared=1) + assert not verdict.ok + assert verdict.mismatches + + def test_insufficient_overlap_fails(self) -> None: + local = [_fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57")] + sina = [_fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57", source="sina")] + verdict = financial_overlap_consistent(local, sina, min_shared=2) + assert not verdict.ok + assert verdict.summary() + + +class TestDailyConsistency: + def test_recent_overlap_equal_passes(self) -> None: + local = [_bar("600519.SH", "2026-09-01", "10"), _bar("600519.SH", "2026-09-02", "10")] + sina = [ + _bar("600519.SH", "2026-09-01", "10", source="sina"), + _bar("600519.SH", "2026-09-02", "10", source="sina"), + ] + verdict = daily_overlap_consistent(local, sina, min_shared=2) + assert verdict.ok + + def test_old_dividend_scale_gap_ignored(self) -> None: + """新浪前复权:老交易日在除权后数值整体平移,只有「最近重叠段」可比。""" + local = [ + _bar("600519.SH", "2025-06-02", "500"), + _bar("600519.SH", "2026-09-01", "10"), + _bar("600519.SH", "2026-09-02", "10"), + _bar("600519.SH", "2026-09-03", "10"), + ] + sina = [ + _bar("600519.SH", "2025-06-02", "480", source="sina"), # 除权后按因子平移 + _bar("600519.SH", "2026-09-01", "10", source="sina"), + _bar("600519.SH", "2026-09-02", "10", source="sina"), + _bar("600519.SH", "2026-09-03", "10", source="sina"), + ] + verdict = daily_overlap_consistent(local, sina, min_shared=2, max_recent=3) + assert verdict.ok + + def test_recent_close_mismatch_fails(self) -> None: + local = [_bar("600519.SH", "2026-09-02", "10")] + sina = [_bar("600519.SH", "2026-09-02", "9", source="sina")] + verdict = daily_overlap_consistent(local, sina, min_shared=1) + assert not verdict.ok + + +class TestReportPeriodRhythm: + def test_windows(self) -> None: + assert latest_expected_report_period(date(2026, 1, 1)) == date(2025, 9, 30) + assert latest_expected_report_period(date(2026, 2, 14)) == date(2025, 9, 30) + assert latest_expected_report_period(date(2026, 2, 15)) == date(2026, 3, 31) + assert latest_expected_report_period(date(2026, 6, 30)) == date(2026, 3, 31) + assert latest_expected_report_period(date(2026, 7, 1)) == date(2026, 6, 30) + assert latest_expected_report_period(TODAY) == DUE + assert latest_expected_report_period(date(2026, 10, 15)) == date(2026, 6, 30) + assert latest_expected_report_period(date(2026, 10, 16)) == date(2026, 9, 30) + + +# ================================================================ 财务同步服务 + +class _FakeFinSource: + def __init__(self, *, name: str, payload=None, error: Exception | None = None) -> None: + self.name = name + self.payload = payload or [] + self.error = error + self.calls: list[tuple] = [] + + def get_financial(self, symbol, start=None, end=None): # noqa: ANN001 + self.calls.append((symbol, start, end)) + if self.error is not None: + raise self.error + return list(self.payload) + + +def _fin_syncer(session, primary, fallback=None, *, today: date = TODAY): + logs: list[SyncLog] = [] + syncer = VerifiedFinancialSyncer( + primary=primary, + fallback=fallback, + repo=SqlAlchemyFinancialRepository(session), + audit=logs.append, + today=today, + ) + return syncer, logs + + +def _count(session, model) -> int: + return session.scalar(select(func.count()).select_from(model)) + + +class TestVerifiedFinancialSyncer: + def _rows(self, symbol: str = "600519.SH") -> list[FinancialIndicator]: + return [ + _fin(symbol, "2026-03-31", "2026-04-25", eps="21.76", gross="89.7592"), + _fin(symbol, "2026-06-30", "2026-08-15", eps="35.57", gross="89.5552"), + ] + + def test_first_fetch_writes_delta_and_second_run_skips(self, session: Session) -> None: + primary = _FakeFinSource(name="tushare", payload=self._rows()) + syncer, logs = _fin_syncer(session, primary) + + res = syncer.sync_symbol("600519.SH") + session.commit() + assert res.status == "ok" + assert res.written == 2 + assert _count(session, FinancialIndicatorModel) == 2 + assert logs[0].source == "tushare" and logs[0].success + + # 第二次:本地已含最新应披露报告期 2026-06-30 → 跳过,不再调 API + res2 = syncer.sync_symbol("600519.SH") + assert res2.status == "skip" + assert len(primary.calls) == 1 + + def test_force_full_overwrites(self, session: Session) -> None: + primary = _FakeFinSource(name="tushare", payload=self._rows()) + syncer, _logs = _fin_syncer(session, primary) + syncer.sync_symbol("600519.SH") + session.commit() + + changed = [ + r.model_copy(update={"eps": Decimal("99")}) for r in self._rows() + ] + [_fin("600519.SH", "2025-12-31", "2026-04-17", eps="65.66", gross="91.1796")] + primary.payload = changed + res = syncer.sync_symbol("600519.SH", force_full=True) + session.commit() + assert res.status == "ok" + assert res.written == 3 + assert res.updated == 2 # 既有 2 行被覆盖 + got = session.scalars(select(FinancialIndicatorModel)).all() + assert len(got) == 3 + assert max(float(r.eps) for r in got) == 99.0 + + def test_sina_fallback_verified_imports_only_new_keys(self, session: Session) -> None: + repo = SqlAlchemyFinancialRepository(session) + repo.upsert_many( + [ + _fin("600519.SH", "2025-12-31", "2026-04-17", eps="65.66", gross="91.1796"), + _fin("600519.SH", "2026-03-31", "2026-04-25", eps="21.76", gross="89.7592"), + ] + ) + session.commit() + + sina_rows = [ + _fin("600519.SH", "2025-12-31", "2026-04-17", eps="65.66", gross="91.179551", source="sina"), + _fin("600519.SH", "2026-03-31", "2026-04-25", eps="21.76", gross="89.759217", source="sina"), + _fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57", gross="89.555212", source="sina"), + ] + primary = _FakeFinSource(name="tushare", error=DataSourceError("频率超限")) + fallback = _FakeFinSource(name="sina", payload=sina_rows) + syncer, logs = _fin_syncer(session, primary, fallback) + + res = syncer.sync_symbol("600519.SH") + session.commit() + assert res.status == "sina" + assert res.written == 1 # 只补本地缺失键(2026-06-30) + rows = session.scalars(select(FinancialIndicatorModel)).all() + assert len(rows) == 3 + new = next(r for r in rows if r.report_date == DUE) + assert new.source == "sina" + # 审计:tushare 失败 + sina 成功 + assert [log.source for log in logs] == ["tushare", "sina"] + assert logs[0].success is False and logs[1].success is True + + def test_sina_fallback_denied_on_mismatch(self, session: Session) -> None: + repo = SqlAlchemyFinancialRepository(session) + # 用非「最新应披露期」做种子,避免触发「已最新跳过」 + repo.upsert_many( + [_fin("600519.SH", "2026-03-31", "2026-04-25", eps="21.76", gross="89.7592")] + ) + session.commit() + + primary = _FakeFinSource(name="tushare", error=DataSourceError("boom")) + # 同报告期 eps 不一致 → 真实性不成立 → 拒绝导入 + fallback = _FakeFinSource( + name="sina", + payload=[ + _fin("600519.SH", "2026-03-31", "2026-04-25", eps="1.00", gross="89.759217", source="sina"), + _fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57", gross="89.555212", source="sina"), + ], + ) + syncer, _logs = _fin_syncer(session, primary, fallback) + res = syncer.sync_symbol("600519.SH") + assert res.status == "failed" + assert res.source == "sina" + assert _count(session, FinancialIndicatorModel) == 1 # 未新增 + + def test_sina_fallback_no_local_history_denied(self, session: Session) -> None: + primary = _FakeFinSource(name="tushare", error=DataSourceError("频率超限")) + fallback = _FakeFinSource( + name="sina", + payload=[_fin("600519.SH", "2026-06-30", "2026-08-15", eps="35.57", source="sina")], + ) + syncer, logs = _fin_syncer(session, primary, fallback) + res = syncer.sync_symbol("600519.SH") + assert res.status == "failed" + assert "本地无历史" in " ".join(res.notes) + assert _count(session, FinancialIndicatorModel) == 0 + assert not logs[-1].success + + +# ================================================================ 日线同步服务 + +class _FakeDailyTushare: + name = "tushare" + + def __init__(self, *, bars=None, factors=None, daily_error=None, factor_error=None) -> None: # noqa: ANN001 + self.bars = bars or [] + self.factors = factors or [] + self.daily_error = daily_error + self.factor_error = factor_error + self.daily_calls = 0 + + def get_daily(self, symbol, start, end): + self.daily_calls += 1 + if self.daily_error is not None: + raise self.daily_error + return list(self.bars) + + def get_adjust_factor(self, symbol, start, end): + if self.factor_error is not None: + raise self.factor_error + return list(self.factors) + + +class _FakeDailySina: + name = "sina" + + def __init__(self, bars) -> None: + self.bars = bars + + def get_daily(self, symbol, start, end): + return [ + b for b in self.bars if start <= b.trade_date <= end + ] + + +def _daily_syncer(session, primary, fallback=None, *, today: date = TODAY): + logs: list[SyncLog] = [] + syncer = VerifiedDailySyncer( + primary=primary, + fallback=fallback, + bars=SqlAlchemyDailyBarRepository(session), + factors=SqlAlchemyAdjustFactorRepository(session), + audit=logs.append, + today=today, + ) + return syncer, logs + + +class TestVerifiedDailySyncer: + def test_primary_ok_writes_bars_and_factors(self, session: Session) -> None: + from app.domain.entities.market import AdjustFactor + + bars = [_bar("600519.SH", "2026-09-01", "10"), _bar("600519.SH", "2026-09-02", "10")] + factors = [ + AdjustFactor(symbol="600519.SH", trade_date=date(2026, 9, 1), factor=Decimal("1.0")) + ] + syncer, logs = _daily_syncer( + session, _FakeDailyTushare(bars=bars, factors=factors) + ) + res = syncer.sync_symbol("600519.SH", date(2026, 9, 1), date(2026, 9, 2)) + assert res.status == "ok" + assert res.bars_written == 2 and res.factors_written == 1 + assert _count(session, StockDailyModel) == 2 + assert _count(session, AdjustFactorModel) == 1 + assert [log.api for log in logs] == ["get_daily", "get_adjust_factor"] + + def test_factor_failure_aborts_symbol_without_partial_write(self, session: Session) -> None: + bars = [_bar("600519.SH", "2026-09-01", "10")] + syncer, _logs = _daily_syncer( + session, + _FakeDailyTushare(bars=bars, factor_error=DataSourceError("限速")), + ) + res = syncer.sync_symbol("600519.SH", date(2026, 9, 1), date(2026, 9, 2)) + assert res.status == "failed" + # 缺复权因子不写本段日线,避免 resume 漏补因子 + assert _count(session, StockDailyModel) == 0 + + def _seed_local(self, session: Session) -> None: + repo = SqlAlchemyDailyBarRepository(session) + repo.upsert_many( + [ + _bar("600519.SH", "2026-08-31", "10"), + _bar("600519.SH", "2026-09-01", "10"), + _bar("600519.SH", "2026-09-02", "10"), + ] + ) + session.commit() + + def test_sina_fallback_fills_only_missing_dates(self, session: Session) -> None: + self._seed_local(session) + sina_bars = [ + _bar("600519.SH", "2026-08-31", "10", source="sina"), + _bar("600519.SH", "2026-09-01", "10", source="sina"), + _bar("600519.SH", "2026-09-02", "10", source="sina"), + _bar("600519.SH", "2026-09-03", "10", source="sina"), + _bar("600519.SH", "2026-09-04", "10", source="sina"), + ] + syncer, logs = _daily_syncer( + session, + _FakeDailyTushare(daily_error=DataSourceError("频率超限")), + fallback=_FakeDailySina(sina_bars), + ) + res = syncer.sync_symbol("600519.SH", date(2026, 9, 3), date(2026, 9, 4)) + session.commit() + assert res.status == "sina" + assert res.bars_written == 2 # 只补 09-03/09-04 + rows = session.scalars(select(StockDailyModel).order_by(StockDailyModel.trade_date)).all() + assert len(rows) == 5 + gap = [r for r in rows if r.trade_date >= date(2026, 9, 3)] + assert all(r.source == "sina" and r.adjust == "qfq" for r in gap) + kept = [r for r in rows if r.trade_date < date(2026, 9, 3)] + assert all(r.source == "tushare" for r in kept) # 既有行不被覆盖 + assert logs[-1].source == "sina" and logs[-1].success + + def test_sina_fallback_denied_on_mismatch(self, session: Session) -> None: + self._seed_local(session) + # 最近重叠日收盘不一致(新浪 9 元 vs 本地 10 元)→ 通道不可信 → 拒绝 + sina_bars = [_bar("600519.SH", "2026-09-02", "9", source="sina")] + syncer, _logs = _daily_syncer( + session, + _FakeDailyTushare(daily_error=DataSourceError("boom")), + fallback=_FakeDailySina(sina_bars), + ) + res = syncer.sync_symbol("600519.SH", date(2026, 9, 3), date(2026, 9, 4)) + assert res.status == "failed" + assert _count(session, StockDailyModel) == 3 # 未新增 + + def test_sina_fallback_no_overlap_denied(self, session: Session) -> None: + # 本地没有任何近期历史可对照 → 拒绝新浪补缺 + sina_bars = [_bar("600519.SH", "2026-09-03", "10", source="sina")] + syncer, logs = _daily_syncer( + session, + _FakeDailyTushare(daily_error=DataSourceError("boom")), + fallback=_FakeDailySina(sina_bars), + ) + res = syncer.sync_symbol("600519.SH", date(2026, 9, 3), date(2026, 9, 4)) + assert res.status == "failed" + assert _count(session, StockDailyModel) == 0 + assert not logs[-1].success diff --git a/backend/tests/test_migrations.py b/backend/tests/test_migrations.py index b659b5d..802e425 100644 --- a/backend/tests/test_migrations.py +++ b/backend/tests/test_migrations.py @@ -53,6 +53,9 @@ def test_upgrade_head_creates_phase1_tables(tmp_path) -> None: con.close() assert {"report_date", "announce_date"} <= fin_cols assert {"symbol", "trade_date", "close"} <= daily_cols + # 来源标记列(新浪校验兜底行需与 Tushare 首选行区分) + assert "source" in fin_cols + assert {"source", "adjust"} <= daily_cols def test_upgrade_head_idempotent(tmp_path) -> None: diff --git a/backend/tests/test_repositories.py b/backend/tests/test_repositories.py index 395e162..9bba3cf 100644 --- a/backend/tests/test_repositories.py +++ b/backend/tests/test_repositories.py @@ -196,6 +196,36 @@ class TestFinancialRepository: assert len(got) == 1 assert got[0].eps == Decimal("9.99") + def test_source_marker_and_incremental_queries(self, session: Session) -> None: + """source 来源标记(默认 tushare)+ 增量判断查询(has_report_period/list_symbol)。""" + repo = SqlAlchemyFinancialRepository(session) + repo.upsert_many( + [ + FinancialIndicator( + symbol="600519.SH", + report_date=date(2024, 6, 30), + announce_date=date(2024, 8, 31), + eps=Decimal("33.19"), + ), # source 默认 tushare + FinancialIndicator( + symbol="600519.SH", + report_date=date(2024, 9, 30), + announce_date=date(2024, 10, 30), + source="sina", + eps=Decimal("48.42"), + ), + ] + ) + session.commit() + + rows = repo.list_symbol("600519.SH") + assert {r.source for r in rows} == {"tushare", "sina"} + assert rows[0].source == "tushare" # 按 announce_date 升序 + + assert repo.has_report_period("600519.SH", date(2024, 9, 30)) + assert not repo.has_report_period("600519.SH", date(2024, 12, 31)) + assert repo.list_symbol("000001.SZ") == [] + class TestSyncLogRepository: def test_add_and_recent(self, session: Session) -> None: diff --git a/backend/tests/test_sina_provider.py b/backend/tests/test_sina_provider.py index 8f7560d..d8ce167 100644 --- a/backend/tests/test_sina_provider.py +++ b/backend/tests/test_sina_provider.py @@ -125,9 +125,49 @@ class TestFinancial: assert float(row.eps) == 1.23 assert float(row.roe) == 15.5 assert float(row.net_profit) == 1230000000 + assert row.source == "sina" # 兜底行必须标记来源(与 Tushare 首选行区分) # FinancialIndicator 校验能通过 → 格式与 Tushare 一致 assert row.model_dump()["eps"] is not None + def test_window_filter(self) -> None: + payload = json.dumps( + { + "result": { + "data": { + "report_date": [ + {"date_value": "2024-06-30"}, + {"date_value": "2024-03-31"}, + ], + "report_list": { + "2024-06-30": { + "publish_date": "2024-08-31", + "data": [{"item_title": "基本每股收益", "item_value": "1.23"}], + }, + "2024-03-31": { + "publish_date": "2024-04-27", + "data": [{"item_title": "基本每股收益", "item_value": "0.5"}], + }, + }, + } + } + }, + ensure_ascii=False, + ) + + class _TwoResp: + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + return payload.encode("utf-8") + + provider = SinaProvider(urlopen=lambda _req, **kw: _TwoResp()) + rows = provider.get_financial("600519.SH", date(2024, 6, 30), date(2024, 6, 30)) + assert [r.report_date for r in rows] == [date(2024, 6, 30)] + def test_bad_structure_raises(self) -> None: class _BadResp(_FinResp): def read(self): diff --git a/backend/tests/test_tushare_provider.py b/backend/tests/test_tushare_provider.py index 2215c9f..bde9247 100644 --- a/backend/tests/test_tushare_provider.py +++ b/backend/tests/test_tushare_provider.py @@ -169,3 +169,58 @@ class TestRateLimitBackoff: ts_mod.time.sleep = orig_sleep assert len(provider._pro.calls) == 3 # noqa: SLF001 —— 完整重试 3 次 assert len(sleeps) >= 2 + + +def _quarter_ends(count: int) -> list[str]: + """最近 count 个季度末(YYYYMMDD,降序)。""" + + ends: list[str] = [] + y, m = 2026, 6 + while len(ends) < count: + ends.append(f"{y}{m:02d}30" if m in (6, 9) else f"{y}{m:02d}31") + m -= 3 + if m <= 0: + m += 12 + y -= 1 + return ends + + +class _FakeProQueue: + """按调用顺序弹出 payload 的 Fake pro(模拟分页)。""" + + def __init__(self, payloads: list[list[dict]]) -> None: + self.payloads = list(payloads) + self.kwargs: list[dict] = [] + + def fina_indicator(self, **kwargs): + self.kwargs.append(kwargs) + return self.payloads.pop(0) + + +class TestGetFinancialWindow: + def _record(self, end: str) -> dict: + return {"ts_code": "600519.SH", "end_date": end, "ann_date": end, "eps": "1.0"} + + def test_window_args_passed(self) -> None: + ends = _quarter_ends(10) + fake = _FakeProQueue([[self._record(e) for e in ends]]) + provider = TushareProvider(token="t", pro=fake) + rows = provider.get_financial("600519.SH", date(2024, 1, 1), date(2026, 6, 30)) + assert len(rows) == 10 + assert fake.kwargs[0]["start_date"] == "20240101" + assert fake.kwargs[0]["end_date"] == "20260630" + + def test_paging_over_100_row_cap(self) -> None: + """单请求最多 100 条 → 超过必须回卷报告期窗口继续取,老数据不丢。""" + newest = _quarter_ends(100) + older = _quarter_ends(140)[100:] # 100 条之外更早的 40 个季度 + # 二次请求 end_date 必须早于首请求(分页回卷) + fake = _FakeProQueue( + [[self._record(e) for e in newest], [self._record(e) for e in older]] + ) + provider = TushareProvider(token="t", pro=fake) + rows = provider.get_financial("600519.SH", date(2000, 1, 1), date(2026, 6, 30)) + assert len(rows) == 100 + 40 + assert len(fake.kwargs) == 2 + assert fake.kwargs[1]["end_date"] < fake.kwargs[0]["end_date"] + assert {r.report_date for r in rows} == {date.fromisoformat(e[:4] + "-" + e[4:6] + "-" + e[6:]) for e in newest + older} diff --git a/docs/USAGE.md b/docs/USAGE.md index 3ba5b42..f983295 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -110,15 +110,26 @@ uv run python -m app.cli.sync basic # 股票基础信息( uv run python -m app.cli.sync calendar --start 20240101 --end 20241231 uv run python -m app.cli.sync daily --symbols 600519.SH,000858.SZ --start 20230101 uv run python -m app.cli.sync daily --all --start 20240101 --resume # 全市场 + 断点续传 -uv run python -m app.cli.sync financial --all # 财务指标快照 +uv run python -m app.cli.sync financial --all # 财务指标(默认增量) +uv run python -m app.cli.sync financial --all --full # 财务指标(强制全量重拉) uv run python -m app.cli.sync verify --symbol 600519.SH # 新浪交叉验证 uv run python -m app.cli.sync export [--years 2023,2024] # 日线按年导出 Parquet ``` - 每次拉取写入 `sync_log` 审计(来源 / 成功与否 / 行数 / 区间),禁止静默切换数据源。 -- `daily` 同时写入日线与复权因子;`--resume` 从本地最新交易日续传。 +- `daily` 同时写入日线与复权因子;`--resume` 从本地最新交易日续传。`financial` + 默认增量:本地已含「最新应披露报告期」(按 A 股披露节奏推算)的股票直接跳过; + 逐只落库,中断/限速后重跑同一命令即可续传补齐(只丢当前一只)。 +- 新浪兜底带「两边一致」真实性校验:Tushare 报错时,只有当该股票本地历史与新浪 + 返回数据**重叠部分一致**,才把新浪**新数据**(本地缺失键)导入,并标记 + `source=sina`(日线另标记 `adjust=qfq` 前复权;新浪不提供复权因子)。 + 校验口径:财务为重叠报告期的 eps / 销售毛利率逐期一致(ROE 两端口径不同, + 不作依据);日线为最近重叠交易日的前复权价一致(老交易日在除权后不可比)。 + 本地无历史可对照或校验不一致 → 拒绝导入并告警,留待 Tushare 恢复后重跑补齐。 +- 限速时可加 `--sleep 秒数` 加大请求间隔;`financial --full` 重拉全部历史并覆盖既有行。 - 新浪(`verify`)仅交叉验证,返回**前复权**口径,不会并入不复权主库。 -- 财务指标带 `announce_date`(公告日),研究侧只允许使用已公告数据(防未来函数)。 +- 财务指标带 `announce_date`(公告日)与 `source`(tushare/sina);研究侧只允许 + 使用已公告数据(防未来函数),对同一报告期优先消费 `source=tushare` 的行。 - `export`:`data/parquet/stock_daily/.parquet`(列:symbol/trade_date/ohlc/volume/amount), 由 pyarrow 写出,可直接用 pandas 读取分析,或作为 Qlib 等引擎的后续数据源。