"""盘后管道 + 盘前维表同步。 调度: 09:10 盘前 — 同步个股维表 instruments (全量覆盖) 15:30 盘后 — 日K同步 + 增量除权因子 + enriched 计算 + 刷新视图 盘后同步策略: 日 K: QuoteService 交易时段已实时落盘 → 有数据时跳过 batch,首次拉 1 年区间 除权因子: 从已有数据最新日期的下一天开始增量获取,避免重复拉取和计算 """ from __future__ import annotations import logging from collections.abc import Callable from pathlib import Path import polars as pl from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from app.indicators.pipeline import run_pipeline from app.config import settings from app.services import index_sync, instrument_sync, kline_sync, preferences as _prefs from app.tickflow.capabilities import Cap, CapabilitySet from app.tickflow.pools import DEMO_SYMBOLS, get_pool from app.tickflow.repository import KlineRepository logger = logging.getLogger(__name__) ProgressCb = Callable[..., None] def _noop(stage: str, pct: int, msg: str, **kwargs) -> None: # noqa: ARG001 pass def _invalidate(table: str | None = None) -> None: """stage 写完调用,让 /api/data/status 只重算被影响的那张表。""" from app.api.data import invalidate_data_cache invalidate_data_cache(table) def _resolve_universe(capset: CapabilitySet) -> list[str]: """解析标的池 — 以 CN_Equity_A (沪深京A股 ~5522只) 为主。 有 batch 能力 → 直接拉 CN_Equity_A universe 其他用户 → 用 instruments parquet + watchlist 兜底 """ if capset.has(Cap.KLINE_DAILY_BATCH): try: all_a = get_pool("CN_Equity_A", refresh=True) if all_a: return sorted(all_a) except Exception as e: # noqa: BLE001 logger.warning("CN_Equity_A pool unavailable, fallback: %s", e) # Free 用户兜底: instruments parquet + watchlist + demo base: set[str] = set(DEMO_SYMBOLS) base.update(get_pool("watchlist")) d = Path(settings.data_dir) inst_path = d / "instruments" / "instruments.parquet" if inst_path.exists(): try: inst = pl.read_parquet(inst_path, columns=["symbol"]) base.update(inst["symbol"].to_list()) except Exception as e: # noqa: BLE001 logger.warning("instruments supplement failed: %s", e) return sorted(base) def run_instruments_sync(repo: KlineRepository) -> dict: """盘前同步个股维表。""" rows = instrument_sync.sync_instruments(repo.store.data_dir) _refresh_instruments_view(repo) _invalidate("instruments") return {"instruments_rows": rows} def run_now( repo: KlineRepository, capset: CapabilitySet, on_progress: ProgressCb | None = None, ) -> dict: """立即执行一次盘后管道,支持进度回调。 跳过的 stage **不 emit**,避免前端把"无 capability"的卡片错误标记为 active/done。 result 里带 skipped_stages 列表供前端展示。 """ emit = on_progress or _noop skipped: list[str] = [] # Step 0: 先同步个股维表, 再解析标的池 — 确保标的池基于最新 instruments emit("sync_instruments", 2, "同步个股维表…") inst_rows = instrument_sync.sync_instruments(repo.store.data_dir) if inst_rows > 0: _refresh_instruments_view(repo) emit("sync_instruments", 8, f"个股维表同步完成,{inst_rows} 只标的") _invalidate("instruments") emit("resolve_universe", 9, "解析标的池…") universe = _resolve_universe(capset) emit("resolve_universe", 10, f"标的池规模:{len(universe)} 只") # Step 1: 日 K 同步 # 付费档 + 今天有数据 → 实时行情接口拉一次覆写（1请求全市场） # 有历史数据 → batch K-line API 补齐缺口 # 无任何数据 → batch K-line API 拉首次 1 年 from datetime import date as _date, timedelta as _td, datetime as _dt latest_daily = repo.latest_daily_date() today = _date.today() today_exists = latest_daily and latest_daily >= today new_daily_days = 0 # 日K范围拉取的起点(分支3补缺口/分支4首次); 实时增量/跳过时为 None。 # 供 Step 1.5 除权因子回溯范围对齐: 范围拉取→用日K范围, 非范围→最近N天兜底。 daily_range_start: _date | None = None # A 股日K拉取开关(默认开);关闭时跳过日K同步,保留已有数据 pull_a_share = _prefs.get_pipeline_pull_a_share() if not pull_a_share: emit("sync_daily", 45, "已跳过 A 股日K同步(拉取内容未勾选)") logger.info("sync_daily: skipped (pipeline_pull_a_share=False)") elif today_exists and capset.has(Cap.QUOTE_POOL): # 付费档:今天有数据(QuoteService 已落盘)→ 实时行情覆写,确保最新。 # free/none 档无 quote.pool 能力,即便今天已有数据(如从 expert 降级), # 也降级到下方 batch 路径刷新,避免调用无权限的实时行情接口。 emit("sync_daily", 12, f"获取日K [{today} ~ {today}] 实时行情…") written_daily = kline_sync.sync_daily_by_quotes(repo) new_daily_days = 1 emit("sync_daily", 45, f"日K 完成,{written_daily} 只标的") logger.info("sync_daily: [%s ~ %s] live quotes, %d symbols", today, today, written_daily) elif latest_daily: # 有历史 → batch 补齐缺口。 # 也覆盖"今天已有数据但无实时行情权限(free/none)"的降级场景: # 此时 start_date = latest_daily = today,batch 刷新当天日K。 start_date = latest_daily daily_range_start = start_date emit("sync_daily", 12, f"获取日K [{start_date} ~ {today}]…") logger.info("sync_daily: [%s ~ %s] %s", start_date, today, "refresh today" if today_exists else "gap fill") def _daily_chunk_progress(cur: int, tot: int) -> None: emit("sync_daily", 12 + int(33 * cur / tot), f"日K 批次 {cur}/{tot}", stage_pct=int(100 * cur / tot), skip_log=True) written_daily = kline_sync.sync_and_persist_daily_batch( universe, repo, capset, start_date=_dt.combine(start_date, _dt.min.time()), end_date=_dt.combine(today, _dt.min.time()), on_chunk_done=_daily_chunk_progress, ) gap_days = (today - start_date).days new_daily_days = gap_days emit("sync_daily", 45, f"日K 完成,覆盖 {gap_days} 天") logger.info("sync_daily: [%s ~ %s] done, %d days", start_date, today, gap_days) else: # 首次：无任何数据 → batch 拉 1 年 start_date = today - _td(days=365) daily_range_start = start_date emit("sync_daily", 12, f"获取日K [{start_date} ~ {today}]…") logger.info("sync_daily: [%s ~ %s] initial fetch", start_date, today) def _daily_chunk_progress(cur: int, tot: int) -> None: emit("sync_daily", 12 + int(33 * cur / tot), f"日K 批次 {cur}/{tot}", stage_pct=int(100 * cur / tot), skip_log=True) written_daily = kline_sync.sync_and_persist_daily_batch( universe, repo, capset, start_date=_dt.combine(start_date, _dt.min.time()), end_date=_dt.combine(today, _dt.min.time()), on_chunk_done=_daily_chunk_progress, ) new_daily_days = 365 emit("sync_daily", 45, "日K 完成") logger.info("sync_daily: [%s ~ %s] done", start_date, today) _invalidate("daily") # Step 1.5: 同步除权因子 — 范围与日K拉取方式对齐 # 日K范围拉取(补缺口/首次) → 除权用日K范围 [daily_range_start, now] # 首次会覆盖整个日K区间内的历史除权事件; 补缺口天然只增量(起点=latest_daily≈昨天) # 日K实时增量/跳过(分支2/分支1) → 除权兜底拉最近 30 天, 补可能遗漏的新除权 # (这两类分支不拉历史日K, 除权不能用日K范围, 只能兜底最近几日) written_adj = 0 affected_symbols: list[str] = [] if capset.has(Cap.ADJ_FACTOR): from datetime import datetime, timedelta adj_end = datetime.now() if daily_range_start is not None: adj_start = datetime.combine(daily_range_start, datetime.min.time()) else: # 日K实时增量/跳过时, 除权兜底拉最近 N 天, 覆盖周末/长假/停机期间的新除权事件。 # 15 天: 覆盖春节/国庆最长约10天长假 + 故障恢复缓冲; sync_adj_factor 内部 merge+unique 幂等, 多拉无副作用。 adj_start = adj_end - timedelta(days=15) adj_start_str = adj_start.strftime("%Y-%m-%d") adj_end_str = adj_end.strftime("%Y-%m-%d") emit("sync_adj", 50, f"获取除权因子 [{adj_start_str} ~ {adj_end_str}]…") logger.info("sync_adj: [%s ~ %s] start", adj_start_str, adj_end_str) def _adj_chunk_progress(cur: int, tot: int) -> None: emit("sync_adj", 50 + int(10 * cur / tot), f"除权因子批次 {cur}/{tot}", stage_pct=int(100 * cur / tot), skip_log=True) written_adj, affected_symbols = kline_sync.sync_adj_factor( universe, repo, capset, start_time=adj_start, end_time=adj_end, on_chunk_done=_adj_chunk_progress, ) if affected_symbols: _refresh_single_view(repo, "adj_factor") emit("sync_adj", 60, f"除权因子完成,新增 {len(affected_symbols)} 只个股") logger.info("sync_adj: [%s ~ %s] done, %d symbols", adj_start_str, adj_end_str, len(affected_symbols)) else: emit("sync_adj", 60, "除权因子完成,无新增") logger.info("sync_adj: [%s ~ %s] no new factors", adj_start_str, adj_end_str) _invalidate("adj_factor") else: skipped.append("sync_adj") logger.info("sync_adj skipped: no ADJ_FACTOR capability") # Step 2: 计算 enriched # 判断策略: # - 首次 (enriched 目录不存在) → 全量 # - 往前扩展历史 (新日期 < enriched 已有最早日期) → 全量 # 前面的除权因子会改变累积因子链,影响后面所有日期的复权价格 # - 往后新增日期 (新日期 > enriched 已有最晚日期) # → 增量补新区块(所有标的) + 受除权影响个股全日期重算 # - 无新日期 + 有新除权因子 → 增量: 只重算受影响个股的全部日期 # - 无新日期 + 无变化 → 跳过 enriched_dir = repo.store.data_dir / "kline_daily_enriched" enriched_exists = enriched_dir.exists() and any(enriched_dir.glob("date=*")) daily_dir = repo.store.data_dir / "kline_daily" daily_days = len(list(daily_dir.glob("date=*"))) if daily_dir.exists() else 0 prev_enriched_days = len(list(enriched_dir.glob("date=*"))) if enriched_exists else 0 # 判断新日期方向: 找 daily 和 enriched 的日期集合做比较 forward_incremental = False backward_extension = False if daily_days > prev_enriched_days and enriched_exists: daily_dates = sorted(d.stem.split("=")[1] for d in daily_dir.glob("date=*")) enriched_dates = sorted(d.stem.split("=")[1] for d in enriched_dir.glob("date=*")) earliest_enriched = enriched_dates[0] latest_enriched = enriched_dates[-1] new_dates = set(daily_dates) - set(enriched_dates) if new_dates: # 有新日期早于 enriched 最早日期 → 往前扩展 if any(d < earliest_enriched for d in new_dates): backward_extension = True # 有新日期晚于 enriched 最晚日期 → 往后新增 if any(d > latest_enriched for d in new_dates): forward_incremental = True def _enriched_batch_progress(cur: int, tot: int) -> None: emit("compute_enriched", 65 + int(23 * cur / tot), f"计算指标 批次 {cur}/{tot}", stage_pct=int(100 * cur / tot), skip_log=True) if not enriched_exists or backward_extension: # 首次 或 往前扩展 → 全量 emit("compute_enriched", 65, "全量计算 enriched…") logger.info("compute_enriched: full rebuild (first=%s, backward=%s, daily=%d, enriched=%d)", not enriched_exists, backward_extension, daily_days, prev_enriched_days) written_enriched = run_pipeline(on_batch_done=_enriched_batch_progress) new_enriched_days = len(list(enriched_dir.glob("date=*"))) emit("compute_enriched", 88, f"enriched 完成,覆盖 {new_enriched_days} 天") logger.info("compute_enriched: full rebuild done, %d days", new_enriched_days) elif forward_incremental: # 往后新增日期: 增量补新区块 + 受影响个股全日期重算 symbols_to_recompute = list(set(affected_symbols)) if affected_symbols else [] emit("compute_enriched", 65, f"增量计算 enriched (新日期 + {len(symbols_to_recompute)} 只个股重算)…" if symbols_to_recompute else "增量计算 enriched (新日期)…") logger.info("compute_enriched: forward incremental, %d symbols to recompute", len(symbols_to_recompute)) written_enriched = run_pipeline( new_dates_only=True, symbols=symbols_to_recompute or None, on_batch_done=_enriched_batch_progress, ) new_enriched_days = len(list(enriched_dir.glob("date=*"))) emit("compute_enriched", 88, f"enriched 完成,覆盖 {new_enriched_days} 天") logger.info("compute_enriched: forward incremental done, %d days", new_enriched_days) elif affected_symbols: # 无新日期,仅除权因子变更 → 只重算受影响个股的全部日期 emit("compute_enriched", 65, f"增量计算 enriched ({len(affected_symbols)} 只个股)…") logger.info("compute_enriched: adj_factor incremental, %d symbols", len(affected_symbols)) written_enriched = run_pipeline(symbols=affected_symbols, on_batch_done=_enriched_batch_progress) emit("compute_enriched", 88, f"enriched 完成,{len(affected_symbols)} 只个股") else: written_enriched = 0 logger.info("compute_enriched: skip (no new daily, no adj_factor changes)") _refresh_single_view(repo, "kline_enriched") _invalidate("enriched") # Step 2.3: 指数 / ETF 同步 — 物理分开存储；ETF 可复权，指数不复权。 written_index_daily = 0 written_etf_daily = 0 index_count = 0 etf_count = 0 etf_adj_symbols = 0 pull_index = _prefs.get_pipeline_pull_index() pull_etf = _prefs.get_pipeline_pull_etf() if capset.has(Cap.KLINE_DAILY_BATCH) and (pull_index or pull_etf): _types = [] if pull_index: _types.append("指数") if pull_etf: _types.append("ETF") emit("sync_index", 88, f"同步{'+'.join(_types)}日K…") # 子阶段进度分配: 88.0(开始) → 89.0(完成), 指数占前半, ETF 占后半 try: if pull_index: emit("sync_index", 88, "同步指数维表…") index_count = index_sync.sync_index_instruments(repo, pull_index=True, pull_etf=False) emit("sync_index", 88, f"指数维表完成,{index_count} 只") index_dir = repo.store.data_dir / "kline_index_enriched" index_dates = sorted( d.name[5:] for d in index_dir.glob("date=*") if d.is_dir() and d.name.startswith("date=") ) if index_dir.exists() else [] index_start = _date.fromisoformat(index_dates[-1]) if index_dates else today - _td(days=365) def _index_chunk(cur: int, tot: int) -> None: emit("sync_index", 88, f"指数日K批次 {cur}/{tot}", stage_pct=int(100 * cur / tot) if tot else 100, skip_log=cur < tot) written_index_daily = index_sync.sync_and_persist_index_daily( repo, capset, start_date=_dt.combine(index_start, _dt.min.time()), end_date=_dt.combine(today, _dt.min.time()), on_chunk_done=_index_chunk, ) emit("sync_index", 88, f"指数日K完成,{written_index_daily} 行") _invalidate("index_instruments") _invalidate("index_daily") _invalidate("index_enriched") if pull_etf: emit("sync_index", 88, "同步 ETF 维表…") etf_count = index_sync.sync_etf_instruments(repo) emit("sync_index", 88, f"ETF 维表完成,{etf_count} 只") etf_symbols: list[str] = [] etf_inst = repo.get_etf_instruments() if not etf_inst.is_empty() and "symbol" in etf_inst.columns: etf_symbols = sorted(set(etf_inst["symbol"].to_list())) if etf_symbols and capset.has(Cap.ADJ_FACTOR): try: emit("sync_index", 88, "同步 ETF 除权因子…") from datetime import datetime, timedelta adj_end = datetime.now() adj_path = repo.store.data_dir / "adj_factor_etf" / "all.parquet" fallback_start = adj_end - timedelta(days=30) adj_start = fallback_start if adj_path.exists(): max_date = pl.scan_parquet(adj_path).select(pl.col("trade_date").max()).collect().item() if max_date is not None: if isinstance(max_date, str): adj_start = datetime.combine(_date.fromisoformat(max_date), datetime.min.time()) elif isinstance(max_date, datetime): adj_start = datetime.combine(max_date.date(), datetime.min.time()) else: adj_start = datetime.combine(max_date, datetime.min.time()) _, affected_etfs = index_sync.sync_etf_adj_factor( etf_symbols, repo, capset, start_time=adj_start, end_time=adj_end, ) etf_adj_symbols = len(affected_etfs) emit("sync_index", 88, f"ETF 除权因子完成,{etf_adj_symbols} 只") except Exception as e: # noqa: BLE001 logger.warning("ETF adj_factor skipped: %s", e) etf_dir = repo.store.data_dir / "kline_etf_enriched" etf_dates = sorted( d.name[5:] for d in etf_dir.glob("date=*") if d.is_dir() and d.name.startswith("date=") ) if etf_dir.exists() else [] etf_start = _date.fromisoformat(etf_dates[-1]) if etf_dates else today - _td(days=365) def _etf_chunk(cur: int, tot: int) -> None: emit("sync_index", 88, f"ETF 日K批次 {cur}/{tot}", stage_pct=int(100 * cur / tot) if tot else 100, skip_log=cur < tot) written_etf_daily = index_sync.sync_and_persist_etf_daily( repo, capset, start_date=_dt.combine(etf_start, _dt.min.time()), end_date=_dt.combine(today, _dt.min.time()), on_chunk_done=_etf_chunk, ) emit("sync_index", 88, f"ETF 日K完成,{written_etf_daily} 行") _invalidate("etf_instruments") _invalidate("etf_daily") repo.refresh_index_views() emit( "sync_index", 89, f"同步完成,指数 {index_count} 只/{written_index_daily} 行, ETF {etf_count} 只/{written_etf_daily} 行" + (f", ETF复权 {etf_adj_symbols} 只" if etf_adj_symbols else ""), ) except Exception as e: # noqa: BLE001 logger.warning("sync_index/etf failed: %s", e) emit("sync_index", 89, f"指数/ETF同步失败:{e}") else: skipped.append("sync_index") # Step 2.5: 分钟 K 同步(可选) — 未启用或无 capability 时静默跳过(不 emit) from app.services import preferences minute_on = preferences.get_minute_sync_enabled() minute_days = preferences.get_minute_sync_days() written_minute = 0 if minute_on and capset.has(Cap.KLINE_MINUTE_BATCH): minute_start = today - _td(days=minute_days) emit("sync_minute", 90, f"获取分钟K [{minute_start} ~ {today}]…") logger.info("sync_minute: [%s ~ %s] start", minute_start, today) minute_symbols = _resolve_minute_symbols(capset) def _minute_chunk_progress(cur: int, tot: int) -> None: emit("sync_minute", 90 + int(3 * cur / tot), f"分钟K 批次 {cur}/{tot}", stage_pct=int(100 * cur / tot), skip_log=True) written_minute = kline_sync.sync_and_persist_minute( minute_symbols, repo, capset, days=minute_days, on_chunk_done=_minute_chunk_progress, ) minute_dir = repo.store.data_dir / "kline_minute" minute_cover_days = len(list(minute_dir.glob("date=*"))) if minute_dir.exists() else 0 emit("sync_minute", 93, f"分钟K完成,覆盖 {minute_cover_days} 天") logger.info("sync_minute: [%s ~ %s] done, %d days", minute_start, today, minute_cover_days) _invalidate("minute") else: skipped.append("sync_minute") if minute_on: logger.info("sync_minute skipped: no KLINE_MINUTE_BATCH capability") else: logger.info("sync_minute skipped: user disabled") # Step 3: 刷新视图 emit("refresh_views", 95, "刷新 DuckDB 视图…") _refresh_views(repo) emit("done", 100, "完成") _invalidate(None) # 兜底:全清 return { "universe_size": len(universe), "daily_days": new_daily_days, "adj_factor_symbols": len(affected_symbols), "enriched_days": written_enriched, "index_count": index_count, "index_daily_rows": written_index_daily, "etf_count": etf_count, "etf_daily_rows": written_etf_daily, "etf_adj_factor_symbols": etf_adj_symbols, "minute_rows": written_minute, "skipped_stages": skipped, } def _refresh_views(repo: KlineRepository) -> None: """刷新所有 DuckDB 视图。""" d = repo.store.data_dir.as_posix() views = { "kline_daily": f"{d}/kline_daily/**/*.parquet", "kline_enriched": f"{d}/kline_daily_enriched/**/*.parquet", "kline_index_daily": f"{d}/kline_index_daily/**/*.parquet", "kline_index_enriched": f"{d}/kline_index_enriched/**/*.parquet", "kline_etf_daily": f"{d}/kline_etf_daily/**/*.parquet", "kline_etf_enriched": f"{d}/kline_etf_enriched/**/*.parquet", "kline_etf_minute": f"{d}/kline_etf_minute/**/*.parquet", "kline_minute": f"{d}/kline_minute/**/*.parquet", "adj_factor": f"{d}/adj_factor/**/*.parquet", "adj_factor_etf": f"{d}/adj_factor_etf/**/*.parquet", "instruments": f"{d}/instruments/**/*.parquet", "instruments_index": f"{d}/instruments_index/**/*.parquet", "instruments_etf": f"{d}/instruments_etf/**/*.parquet", } for name, path in views.items(): try: repo.db.execute( f"CREATE OR REPLACE VIEW {name} AS " f"SELECT * FROM read_parquet('{path}', union_by_name=true)" ) except Exception as e: # noqa: BLE001 logger.warning("refresh view %s failed: %s", name, e) repo.store._register_unified_views() def _refresh_single_view(repo: KlineRepository, name: str) -> None: """刷新单个 DuckDB 视图。""" d = repo.store.data_dir.as_posix() paths = { "kline_daily": f"{d}/kline_daily/**/*.parquet", "kline_enriched": f"{d}/kline_daily_enriched/**/*.parquet", "kline_index_daily": f"{d}/kline_index_daily/**/*.parquet", "kline_index_enriched": f"{d}/kline_index_enriched/**/*.parquet", "kline_etf_daily": f"{d}/kline_etf_daily/**/*.parquet", "kline_etf_enriched": f"{d}/kline_etf_enriched/**/*.parquet", "kline_etf_minute": f"{d}/kline_etf_minute/**/*.parquet", "kline_minute": f"{d}/kline_minute/**/*.parquet", "adj_factor": f"{d}/adj_factor/**/*.parquet", "adj_factor_etf": f"{d}/adj_factor_etf/**/*.parquet", "instruments": f"{d}/instruments/**/*.parquet", "instruments_index": f"{d}/instruments_index/**/*.parquet", "instruments_etf": f"{d}/instruments_etf/**/*.parquet", } path = paths.get(name) if not path: return try: repo.db.execute( f"CREATE OR REPLACE VIEW {name} AS " f"SELECT * FROM read_parquet('{path}', union_by_name=true)" ) except Exception as e: # noqa: BLE001 logger.warning("refresh view %s failed: %s", name, e) def _resolve_minute_symbols(capset: CapabilitySet) -> list[str]: """分钟 K 同步标的 — 与日K共用同一标的池。""" return _resolve_universe(capset) def _refresh_instruments_view(repo: KlineRepository) -> None: """单独刷新 instruments 视图。""" d = repo.store.data_dir.as_posix() try: repo.db.execute( f"CREATE OR REPLACE VIEW instruments AS " f"SELECT * FROM read_parquet('{d}/instruments/**/*.parquet', union_by_name=true)" ) except Exception as e: # noqa: BLE001 logger.warning("refresh instruments view failed: %s", e) def _run_tracked(fn, job_label: str) -> None: """调度触发时包装 JobStore 跟踪，确保同步历史有记录。""" from app.services.pipeline_jobs import job_store job_id = job_store.create() job_store.start(job_id) def progress(stage: str, pct: int, msg: str, stage_pct: int | None = None, skip_log: bool = False) -> None: job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log) try: result = fn(on_progress=progress) job_store.succeed(job_id, result) logger.info("scheduled %s completed: job_id=%s", job_label, job_id) except Exception: logger.exception("scheduled %s failed: job_id=%s", job_label, job_id) job_store.fail(job_id, f"scheduled {job_label} failed") # ================================================================ # 定时复盘 (AI 大盘复盘报告) # ================================================================ REVIEW_JOB_ID = "scheduled_review" async def _run_scheduled_review(repo) -> None: """定时复盘 job: 流式生成复盘 → 落盘归档 → 推飞书。 与手动「生成复盘」体验一致。LLM 偶发断流(peer closed connection)时自动重试最多 2 次。 任何异常都吞掉只记日志, 绝不影响调度器主循环。 """ try: from app.services import market_recap_reports from app import secrets_store as ss # AI Key 未配置时跳过(避免每日报错刷日志) if not ss.get_ai_key(): logger.info("scheduled review skipped: AI key not configured") return content, meta = await _stream_review_with_retry(repo) if not content: logger.warning("scheduled review produced no content (meta=%s)", meta) return # 落盘: 与手动生成完全相同的归档格式 market_recap_reports.save_report({ "as_of": meta.get("as_of"), "focus": "", "content": content, "summary": meta.get("summary", ""), "emotion_score": meta.get("emotion_score"), "emotion_label": meta.get("emotion_label", ""), }) logger.info("scheduled review saved: as_of=%s", meta.get("as_of")) # 推送到飞书(可选): 运行时读取配置, 用户改设置下次触发即生效。 # 失败静默降级, 不影响已归档的报告。 _maybe_push_review(content, meta) except Exception as e: # noqa: BLE001 logger.exception("scheduled review failed: %s", e) async def _stream_review_with_retry(repo) -> tuple[str, dict]: """流式生成复盘, 累积内容。LLM 断流时最多重试 2 次。 返回 (content, meta)。成功(收到 done/无 error)或耗尽重试后返回。 """ import asyncio import json from app.services.market_recap import recap_market_stream max_attempts = 3 # 初次 + 2 次重试 last_meta: dict = {} content_parts: list[str] = [] for attempt in range(1, max_attempts + 1): content_parts = [] # 每次重试重新累积 failed = False try: async for evt_json in recap_market_stream(repo): evt = json.loads(evt_json) t = evt.get("type") if t == "meta": last_meta = evt elif t == "delta" and evt.get("content"): content_parts.append(evt["content"]) elif t == "error": failed = True logger.warning("scheduled review stream error (attempt %d/%d): %s", attempt, max_attempts, evt.get("message")) break # 触发重试 elif t == "done": # 正常完成 return "".join(content_parts), last_meta # 流自然结束(无 done 事件)且有内容, 视为成功 if content_parts and not failed: return "".join(content_parts), last_meta except Exception as e: # noqa: BLE001 # LLM 断流等异常(httpx.RemoteProtocolError)落到这里 failed = True logger.warning("scheduled review stream exception (attempt %d/%d): %s", attempt, max_attempts, e) # 失败: 决定是否重试 if attempt < max_attempts: logger.info("scheduled review retrying in 3s (attempt %d → %d)", attempt, attempt + 1) await asyncio.sleep(3) # 耗尽重试, 返回已累积内容(可能为空)和最后 meta return "".join(content_parts), last_meta def _maybe_push_review(content: str, meta: dict) -> None: """复盘报告归档后, 按 review_push_channels 选定的外部工具逐个推送完整报告。 定时生成与手动生成共用本函数 (手动归档端点 POST /api/market-recap/reports 也会调用)。 channels 为空则不推送; 'feishu' 复用监控中心的全局飞书 Webhook 通道。 推送失败静默降级 (Webhook 是辅助通道), 不影响已归档的报告。 """ try: from app.services import preferences, webhook_adapter channels = preferences.get_review_push_channels() if not channels: return emotion = f"{meta.get('emotion_label') or ''}".strip() as_of = meta.get("as_of") or "" subtitle = as_of + (f" · 情绪 {emotion}" if emotion else "") for ch in channels: if ch == "feishu": url = preferences.get_feishu_webhook_url() if not url: logger.info("review push(feishu) skipped: webhook not configured") continue secret = preferences.get_feishu_webhook_secret() ok = webhook_adapter.send_feishu_card( url, "TickFlow · 每日复盘", subtitle, content, secret ) logger.info("review push(feishu) %s", "sent" if ok else "failed") # 未来更多渠道在此追加分支 except Exception as e: # noqa: BLE001 logger.warning("review push error: %s", e) def _register_review_job(scheduler, repo, hour: int, minute: int) -> None: """注册/更新定时复盘 job(工作日 mon-fri, Asia/Shanghai)。 供 start_scheduler(启动时) 和 settings API(改时间时) 共用。 用 replace_existing=True, 重复注册只更新 trigger。 注意: _run_scheduled_review 是协程函数, 必须把函数对象本身(配合 args)传给 add_job, 而非用 lambda 包裹 —— 否则 APScheduler 会把 lambda 当同步函数在线程池 执行, 仅得到一个未 await 的协程对象, 复盘实际不会运行。 """ scheduler.add_job( _run_scheduled_review, args=[repo], trigger=CronTrigger(day_of_week="mon-fri", hour=hour, minute=minute, timezone="Asia/Shanghai"), id=REVIEW_JOB_ID, misfire_grace_time=7200, # 复盘非关键, 允许 2 小时内补跑 replace_existing=True, ) def start_scheduler(repo: KlineRepository, capset: CapabilitySet) -> AsyncIOScheduler: """启动调度器。 工作日 09:10 — 同步个股维表 工作日 HH:MM — 盘后管道（时间由用户偏好决定，默认 15:30） """ from app.services import preferences sched = preferences.get_pipeline_schedule() inst_sched = preferences.get_instruments_schedule() scheduler = AsyncIOScheduler(timezone="Asia/Shanghai") # 盘前: 同步 instruments（时间由偏好决定） def _instruments_task(on_progress=None): emit = on_progress or _noop emit("sync_instruments", 0, "同步个股维表…") result = run_instruments_sync(repo) emit("done", 100, f"个股维表同步完成,{result.get('instruments_rows', 0)} 只标的") return result scheduler.add_job( lambda: _run_tracked(_instruments_task, "instruments_sync"), trigger=CronTrigger(day_of_week="mon-fri", hour=inst_sched["hour"], minute=inst_sched["minute"], timezone="Asia/Shanghai"), id="pre_market_instruments", misfire_grace_time=1800, replace_existing=True, ) # 盘后: 日 K + enriched（时间由偏好决定） def _pipeline_then_refresh(on_progress=None): # 与手动触发 (/api/pipeline/run) 对齐: 管道落盘后重建 Polars 内存缓存, # 否则 live_agg 的昨日连板数等基准列会停留在旧交易日, 次日开盘连板梯队 # 整体少算一档 (仅手动触发或重启才会刷缓存, cron 调度路径此前漏了这步)。 result = run_now(repo, capset, on_progress=on_progress) repo.refresh_cache() return result scheduler.add_job( lambda: _run_tracked(_pipeline_then_refresh, "daily_pipeline"), trigger=CronTrigger(day_of_week="mon-fri", hour=sched["hour"], minute=sched["minute"], timezone="Asia/Shanghai"), id="daily_pipeline", misfire_grace_time=3600, replace_existing=True, ) # 定时复盘 (AI 大盘复盘报告): 工作日到点自动生成并归档。 # 默认关闭 —— 仅当用户在复盘页开启时才注册 job。 # 复用 recap_market_once(非流式) + market_recap_reports.save_report(落盘)。 # quote_service / depth_service 通过 _get_app_state() 延迟取用。 review_sched = preferences.get_review_schedule() if review_sched["enabled"]: _register_review_job(scheduler, repo, review_sched["hour"], review_sched["minute"]) logger.info("scheduled_review enabled @%02d:%02d mon-fri", review_sched["hour"], review_sched["minute"]) scheduler.start() logger.info("scheduler started; instruments@%02d:%02d, pipeline@%02d:%02d mon-fri", inst_sched["hour"], inst_sched["minute"], sched["hour"], sched["minute"]) return scheduler # app_state 延迟引用(start_scheduler 在 lifespan 早期调用, app.state 可能还没就绪) _app_state_ref = None def set_app_state(app_state) -> None: """lifespan 注册 app.state 引用, 供 scheduled job 访问单例。""" global _app_state_ref _app_state_ref = app_state def _get_app_state(): return _app_state_ref