"""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 --full # 财务指标(强制全量重拉) uv run python -m app.cli.sync verify --symbol 600519.SH # 新浪交叉验证 uv run python -m app.cli.sync daily_basic --start 20200101 # 每日指标(股息率等) 增量与兜底: - daily --resume:从本地最新交易日续传(已有);Tushare 失败时走新浪校验兜底, 只有「两源重叠历史一致」才用新浪补本地缺失交易日(source=sina/前复权)。 - financial:默认增量——本地已含最新应披露报告期则跳过;Tushare 失败时新浪 数据须通过「两边一致」校验(重叠报告期 eps/销售毛利率逐期一致)才允许补入 本地缺失键(source=sina)。失败股票留待下轮重跑补齐,不会静默导入未核验数据。 - 每次拉取写入 sync_log 审计(来源 / 成功与否 / 行数 / 区间),禁止静默切源。 本模块是组装层(composition root):在此装配 Provider / Repository / Session, 业务逻辑在 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 ( DailyBasicSyncer, DailySymbolResult, FinancialSymbolResult, NameHistorySyncer, VerifiedDailySyncer, VerifiedFinancialSyncer, ) from app.core.config import get_settings from app.infrastructure.data_sources.errors import DataSourceError 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 from app.infrastructure.persistence.sqlalchemy.repositories.index_impl import ( SqlAlchemyIndexConstituentRepository, ) from app.infrastructure.persistence.sqlalchemy.repositories.market_impl import ( SqlAlchemyAdjustFactorRepository, SqlAlchemyDailyBarRepository, SqlAlchemyDailyBasicRepository, SqlAlchemyFinancialRepository, SqlAlchemyStockNameHistoryRepository, SqlAlchemyStockRepository, SqlAlchemySyncLogRepository, SqlAlchemyTradingCalendarRepository, ) from app.infrastructure.persistence.sqlalchemy.session import SessionLocal _DATE_FMT = "%Y%m%d" def _parse_day(text: str) -> date: return datetime.strptime(text, _DATE_FMT).date() def _failover_provider(session): """Tushare 首选 + 新浪兜底(basic/calendar 用;daily/financial 走校验兜底服务)。 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) primary = TushareProvider(token=get_settings().tushare_token) return FailoverProvider(primary, fallback=SinaProvider(), audit=audit_repo.add) def _session_ctx(): return SessionLocal() def cmd_basic(args) -> int: """股票基础信息。--include-delisted 同时拉取已退市/暂停上市(幸存者偏差修正)。""" statuses = ["L", "P", "D"] if getattr(args, "include_delisted", False) else ["L"] with _session_ctx() as session: provider = _failover_provider(session) repo = SqlAlchemyStockRepository(session) total = 0 failed: list[str] = [] for st in statuses: try: stocks = provider.get_stock_basic(st) except DataSourceError as exc: # failover 会把备用源(新浪)的 NotSupported 包装成 DataSourceError, # 因此这里必须捕获 DataSourceError 而不是 DataSourceNotSupported, # 否则 --include-delisted 在主源抖动时会整体抛栈退出(L 已写、P/D 静默缺失)。 print(f"[error] list_status={st} 拉取失败:{exc}", file=sys.stderr) failed.append(st) continue touched = repo.upsert_many(stocks) session.commit() total += touched n_delisted = sum(1 for s in stocks if s.delist_date is not None) print( f"[basic] list_status={st} 拉取 {len(stocks)} 只(含 delist_date {n_delisted} 只)," f"落库 {touched} 条" ) if len(statuses) > 1: print( "[basic] 已退市股票已入库;其历史行情需另行同步(否则回测仍无法使用):\n" " 请执行 sync daily --symbols <退市代码逗号分隔> --start 20200101" ) print(f"[basic] 合计落库 {total} 条") if failed: print( f"[error] 以下 list_status 拉取失败:{','.join(failed)};" "退市股缺失会让回测重新出现幸存者偏差,请重跑", file=sys.stderr, ) return 1 return 0 def cmd_calendar(args) -> int: start = _parse_day(args.start) end = _parse_day(args.end) with _session_ctx() as session: provider = _failover_provider(session) days = provider.get_trade_cal(start, end) repo = SqlAlchemyTradingCalendarRepository(session) touched = repo.upsert_many(days) session.commit() open_days = sum(1 for d in days if d.is_open) print(f"[calendar] {start}~{end} 共 {len(days)} 条(交易日 {open_days}),落库 {touched} 条") return 0 def cmd_namechange(args) -> int: """股票名称变更历史同步(时点 ST 判定依据;按自然年分片)。""" start = _parse_day(args.start) if args.start else date(1990, 1, 1) end = _parse_day(args.end) if args.end else date.today() if start > end: print("[error] --start 不能晚于 --end", file=sys.stderr) return 2 with _session_ctx() as session: audit_repo = SqlAlchemySyncLogRepository(session) primary = TushareProvider(token=get_settings().tushare_token) repo = SqlAlchemyStockNameHistoryRepository(session) syncer = NameHistorySyncer(primary=primary, repo=repo, audit=audit_repo.add) total_span = (end.year - start.year) + 1 print(f"[namechange] {start}~{end} 共 {total_span} 个年度分片") ok = failed = written = fetched = 0 t0 = time.time() def _on(idx: int, total: int, chunk_start: date) -> None: _progress(idx, total, t0, f"{chunk_start.year} 年") results = syncer.sync_range(start, end, on_progress=_on) for res in results: if res.status == "ok": ok += 1 written += res.rows_written fetched += res.rows_fetched else: failed += 1 for note in res.notes: print(f"\n[warn] {res.start}~{res.end}: {note}", file=sys.stderr) session.commit() lo, hi = repo.namechange_dates() print( f"\n[namechange] 分片成功 {ok} / 失败 {failed};" f"拉取 {fetched} 行、写入 {written} 行;" f"本地生效起点 {lo} ~ {hi}" ) return 1 if failed else 0 def _symbols_of(args) -> list[str]: if getattr(args, "all", False): with _session_ctx() as session: symbols = list(session.scalars(select(StockModel.symbol).order_by(StockModel.symbol))) if not symbols: print("[error] stock 表为空,请先运行:python -m app.cli.sync basic") sys.exit(2) return symbols 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 from app.infrastructure.persistence.sqlalchemy.models.market import StockDailyModel 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() started = time.monotonic() n_ok = n_sina = n_failed = n_skip = 0 rows_tushare = rows_sina = 0 with _session_ctx() as 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) # 增量基准:本地数据已到该日期即视为「已最新」,resume 时不再调 API global_latest = ( session.scalar(select(func.max(StockDailyModel.trade_date))) if args.resume else None ) for i, symbol in enumerate(symbols, start=1): begin = start if args.resume: latest = bar_repo.latest_date(symbol) if latest is not None: if global_latest is not None and latest >= global_latest: n_skip += 1 # 已同步到本地最新交易日,无需续拉 continue begin = max(begin, latest + timedelta(days=1)) if begin > end: n_skip += 1 # 无待拉区间(如区间已含在本地) continue 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"(本地最新 {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) started = time.monotonic() n_ok = n_sina = n_failed = n_skip = 0 rows_tushare = rows_sina = 0 with _session_ctx() as session: 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 def cmd_index_weight(args) -> int: """同步指数历史成分(Tushare index_weight;新浪不支持 → failover 审计留痕)。""" code = args.code with _session_ctx() as session: provider = _failover_provider(session) try: rows = provider.get_index_weight(code) except DataSourceError as exc: print(f"[index_weight] {code} 失败:{exc}") return 1 repo = SqlAlchemyIndexConstituentRepository(session) touched = repo.upsert_many(rows) latest = repo.latest_date(code) session.commit() print( f"[index_weight] {code} 拉取 {len(rows)} 期成分行,落库 {touched} 条" f",最新快照 {latest}(as_of 查询见 Universe.index_code)" ) return 0 def cmd_verify(args) -> int: """新浪交叉验证:取新浪最新前复权收盘,与本地最新交易日对照。 注意:新浪为前复权口径,数值不直接等于本地不复权收盘, 本命令仅用于确认新浪可用性 / 最新交易日,不把新浪数据并入主库。 """ from app.infrastructure.persistence.sqlalchemy.repositories.market_impl import ( SqlAlchemyDailyBarRepository, ) sina = SinaProvider() end = date.today() start = end - timedelta(days=20) try: bars = sina.get_daily(args.symbol, start, end) except DataSourceError as exc: print(f"[verify] 新浪不可用: {exc}", file=sys.stderr) return 1 if not bars: print(f"[verify] 新浪最近无数据({args.symbol})") return 1 latest = max(bars, key=lambda b: b.trade_date) with _session_ctx() as session: local = SqlAlchemyDailyBarRepository(session).latest_date(args.symbol) print( f"[verify] {args.symbol}: 新浪最新 {latest.trade_date} 收盘(前复权) {latest.close};" f"本地最新交易日 {local}" ) return 0 def cmd_export(args) -> int: """把 SQLite 日线按年导出为 Parquet(data/parquet/stock_daily/.parquet)。""" from pathlib import Path import pandas as pd from app.infrastructure.persistence.sqlalchemy.models.market import StockDailyModel settings = get_settings() out_root = settings.storage.get("parquet_dir") or Path("data/parquet") out_root.mkdir(parents=True, exist_ok=True) total = 0 years = [int(y) for y in (args.years or "").split(",") if y.strip()] or None with _session_ctx() as session: all_bars = session.execute( select(StockDailyModel).order_by(StockDailyModel.trade_date) ).scalars() frame = pd.DataFrame( [ { "symbol": b.symbol, "trade_date": b.trade_date, "open": float(b.open) if b.open is not None else None, "high": float(b.high) if b.high is not None else None, "low": float(b.low) if b.low is not None else None, "close": float(b.close) if b.close is not None else None, "volume": float(b.volume) if b.volume is not None else None, "amount": float(b.amount) if b.amount is not None else None, } for b in all_bars ] ) if frame.empty: print("[export] 无日线数据,请先运行 sync daily") return 0 frame["trade_date"] = pd.to_datetime(frame["trade_date"]) out_dir = out_root / "stock_daily" out_dir.mkdir(parents=True, exist_ok=True) for year, group in frame.groupby(frame["trade_date"].dt.year): if years and int(year) not in years: continue path = out_dir / f"{year}.parquet" group.sort_values(["symbol", "trade_date"]).to_parquet(path, index=False) total += len(group) print(f"[export] {year} → {path}({len(group)} 行)") print(f"[export] 合计 {total} 行 → {out_dir}") return 0 _DAILY_BASIC_DEFAULT_START = date(2020, 1, 1) def cmd_daily_basic(args) -> int: """每日指标同步(按交易日整表;新浪不支持 → 失败如实记录,不留静默缺口)。""" start = _parse_day(args.start) if args.start else _DAILY_BASIC_DEFAULT_START end = _parse_day(args.end) if args.end else date.today() if start > end: print("[error] --start 不能晚于 --end", file=sys.stderr) return 2 with _session_ctx() as session: audit_repo = SqlAlchemySyncLogRepository(session) primary = TushareProvider(token=get_settings().tushare_token) repo = SqlAlchemyDailyBasicRepository(session) syncer = DailyBasicSyncer(primary=primary, repo=repo, audit=audit_repo.add) if args.full: cal = SqlAlchemyTradingCalendarRepository(session) days = [d.calendar_date for d in cal.list_range(start, end) if d.is_open] else: days = repo.missing_dates(start, end) total = len(days) if total == 0: print(f"[daily_basic] {start}~{end} 无待补交易日(本地已完整)") return 0 print(f"[daily_basic] {start}~{end} 待同步 {total} 个交易日") ok = failed = written = 0 failures: list[date] = [] t0 = time.time() for idx, day in enumerate(days, start=1): try: res = syncer.sync_day(day) except DataSourceError as exc: print(f"\n[error] 第 {idx}/{total} 日 {day} 权限/凭证故障,中止:{exc}", file=sys.stderr) session.commit() return 1 if res.status == "ok": ok += 1 written += res.rows_written else: failed += 1 failures.append(day) _progress(idx, total, t0, f"{day} 行数={res.rows_fetched}") if idx % 20 == 0 or idx == total: session.commit() # 分批提交:中断时已完成的部分保持有效 if args.sleep: time.sleep(args.sleep) session.commit() span = time.time() - t0 print( f"\n[daily_basic] 完成:成功 {ok} 日 / 失败 {failed} 日,累计写入 {written} 行," f"耗时 {span:.1f}s" ) if failures: print( "[daily_basic] 失败交易日(可重跑本命令补齐):" + ", ".join(d.isoformat() for d in failures[:20]) + (" …" if len(failures) > 20 else ""), file=sys.stderr, ) return 1 return 0 def _progress(idx: int, total: int, t0: float, extra: str = "") -> None: """单行进度条(同步长任务可读性)。""" elapsed = time.time() - t0 rate = idx / elapsed if elapsed > 0 else 0.0 eta = (total - idx) / rate if rate > 0 else 0.0 pct = idx / total * 100 if total else 100.0 end = "\n" if idx >= total else "\r" print( f" 进度 {idx}/{total} ({pct:5.1f}%) 已用 {elapsed:6.1f}s 预计剩余 {eta:6.1f}s {extra}", end=end, flush=True, ) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="app.cli.sync", description="Tushare 数据同步 CLI") sub = parser.add_subparsers(dest="command", required=True) p_nc = sub.add_parser("namechange", help="同步股票名称变更历史(时点 ST 判定)") p_nc.add_argument("--start", help="起始日 YYYYMMDD(默认 19900101,覆盖全历史)") p_nc.add_argument("--end", help="结束日 YYYYMMDD(默认今天)") p_nc.set_defaults(func=cmd_namechange) p_basic = sub.add_parser("basic", help="同步股票基础信息") p_basic.add_argument( "--include-delisted", action="store_true", help="同时拉取已退市(D)与暂停上市(P),填充 delist_date(幸存者偏差修正)", ) p_basic.set_defaults(func=cmd_basic) p_cal = sub.add_parser("calendar", help="同步交易日历") p_cal.add_argument("--start", required=True, help="YYYYMMDD") p_cal.add_argument("--end", required=True, help="YYYYMMDD") p_cal.set_defaults(func=cmd_calendar) p_db = sub.add_parser( "daily_basic", help="同步每日指标(估值/股息率/市值,按交易日整表;新浪不支持本接口)", ) p_db.add_argument("--start", default="20200101", help="YYYYMMDD(默认 20200101)") p_db.add_argument("--end", default="", help="YYYYMMDD(默认今天)") p_db.add_argument( "--full", action="store_true", help="忽略本地已有日期,重拉区间内全部开市日(默认只补缺失日)", ) p_db.add_argument( "--sleep", type=float, default=0, help="每个交易日请求间隔秒数(限速时加大,如 0.2 或 1)", ) p_db.set_defaults(func=cmd_daily_basic) 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( "--sleep", type=float, default=0, help="每只股票请求间隔秒数(限速时加大,如 1 或 60)", ) p_daily.set_defaults(func=cmd_daily) p_fin = sub.add_parser("financial", help="同步财务指标快照(默认增量;Tushare 失败 → 新浪校验兜底)") p_fin.add_argument("--symbols", default="") 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_idx = sub.add_parser("index_weight", help="同步指数历史成分(如沪深300 000300.SH)") p_idx.add_argument("--code", required=True, help="指数代码,如 000300.SH / 000905.SH") p_idx.set_defaults(func=cmd_index_weight) p_verify = sub.add_parser("verify", help="新浪交叉验证最新行情") p_verify.add_argument("--symbol", required=True) p_verify.set_defaults(func=cmd_verify) p_export = sub.add_parser("export", help="日线按年导出 Parquet(data/parquet)") p_export.add_argument("--years", default="", help="逗号分隔年份,留空导出全部") p_export.set_defaults(func=cmd_export) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: return args.func(args) except DataSourceError as exc: print(f"[error] {exc}", file=sys.stderr) return 1 except KeyboardInterrupt: print("\n[interrupt] 已中止", file=sys.stderr) return 130 if __name__ == "__main__": raise SystemExit(main())