diff --git a/backend/app/indicators/pipeline.py b/backend/app/indicators/pipeline.py index 210129b..dab7777 100644 --- a/backend/app/indicators/pipeline.py +++ b/backend/app/indicators/pipeline.py @@ -1382,8 +1382,11 @@ def compute_enriched_today( alpha = _ema_alpha - # ---- JOIN: 今天的 OHLCV + 昨天的递推状态 ---- - df = today_ohlcv.join(live_agg, on="symbol", how="inner") + # ---- JOIN: 今天的 OHLCV + 各股票最后一个有效交易日的递推状态 ---- + # 当日行情是主表, 复牌或新上市股票不能因为没有历史状态而被静默删除。 + live_state = live_agg.with_columns(pl.lit(True).alias("_has_history_state")) + df = today_ohlcv.join(live_state, on="symbol", how="left") + has_history_state = pl.col("_has_history_state").fill_null(False) # ---- 前复权: 保存原始价 → 调整 OHLCV ---- df = df.with_columns([ @@ -1527,8 +1530,14 @@ def compute_enriched_today( # ---- 极值 60 日 ---- df = df.with_columns([ - pl.max_horizontal(pl.col("_high_59d"), pl.col("high")).alias("high_60d"), - pl.min_horizontal(pl.col("_low_59d"), pl.col("low")).alias("low_60d"), + pl.when(has_history_state) + .then(pl.max_horizontal(pl.col("_high_59d"), pl.col("high"))) + .otherwise(None) + .alias("high_60d"), + pl.when(has_history_state) + .then(pl.min_horizontal(pl.col("_low_59d"), pl.col("low"))) + .otherwise(None) + .alias("low_60d"), ]) # ---- 动量 (5d/10d/20d/30d/60d) ---- @@ -1548,7 +1557,7 @@ def compute_enriched_today( vol_mean = total_sum / 20 vol_var = total_sq_sum / 20 - vol_mean ** 2 df = df.with_columns( - pl.when(vol_var > 0) + pl.when(has_history_state & (vol_var > 0)) .then(vol_var.sqrt() * (252 ** 0.5)) .otherwise(None) .alias("annual_vol_20d"), @@ -1638,6 +1647,7 @@ def compute_enriched_today( "_adj_factor", "_vol_19d_pct_sum", "_vol_19d_pct_sq_sum", "_prev_consec_up", "_prev_consec_down", + "_has_history_state", ] df = df.drop([c for c in drop_cols if c in df.columns]) @@ -1722,7 +1732,7 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) -> limit_down_price = polars_limit_price(prev_raw, limit_pct, up=False) # 生效涨跌停价: 维表日期与行情日期一致时优先使用交易所权威值; - # 维表过期、价格缺失或新股哨兵值均回退自算理论价。旧版无 as_of 维表保持兼容。 + # 维表过期或价格缺失时回退自算理论价。旧版无 as_of 维表保持兼容。 # 哨兵阈值 10000 用于识别 "新股无涨跌停限制" 的占位值 (实际涨停价不可能上万)。 _SENTINEL = 10000.0 authoritative_date = ( @@ -1730,30 +1740,51 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) -> if "_instrument_as_of" in df.columns else pl.lit(True) ) + has_authoritative_up = pl.lit(False) + has_authoritative_down = pl.lit(False) + no_price_limit = pl.lit(False) if "limit_up" in df.columns: - effective_limit_up = pl.when( + has_authoritative_up = ( authoritative_date & pl.col("limit_up").is_not_null() + & (pl.col("limit_up") > 0) & (pl.col("limit_up") < _SENTINEL) + ) + no_price_limit = ( + authoritative_date + & pl.col("limit_up").is_not_null() + & (pl.col("limit_up") >= _SENTINEL) + ) + effective_limit_up = pl.when( + has_authoritative_up ).then(pl.col("limit_up")).otherwise(limit_up_price) else: effective_limit_up = limit_up_price if "limit_down" in df.columns: - effective_limit_down = pl.when( + has_authoritative_down = ( authoritative_date & pl.col("limit_down").is_not_null() + & (pl.col("limit_down") > 0) & (pl.col("limit_down") < _SENTINEL) + ) + effective_limit_down = pl.when( + has_authoritative_down ).then(pl.col("limit_down")).otherwise(limit_down_price) else: effective_limit_down = limit_down_price + valid_prev_raw = prev_raw.is_not_null() & (prev_raw > 0) is_limit_up = ( - pl.when((prev_raw > 0) & (pl.col("raw_close") > 0)) + pl.when(no_price_limit) + .then(False) + .when((valid_prev_raw | has_authoritative_up) & (pl.col("raw_close") > 0)) .then(pl.col("raw_close") >= (effective_limit_up - 0.005)) .otherwise(None).cast(pl.Boolean) ) is_limit_down = ( - pl.when((prev_raw > 0) & (pl.col("raw_close") > 0)) + pl.when(no_price_limit) + .then(False) + .when((valid_prev_raw | has_authoritative_down) & (pl.col("raw_close") > 0)) .then(pl.col("raw_close") <= (effective_limit_down + 0.005)) .otherwise(None).cast(pl.Boolean) ) @@ -1762,7 +1793,9 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) -> is_limit_up.alias("signal_limit_up"), is_limit_down.alias("signal_limit_down"), # 跌停翘板 - pl.when(prev_raw > 0) + pl.when(no_price_limit) + .then(False) + .when(valid_prev_raw | has_authoritative_down) .then( (~is_limit_down.fill_null(True)) & (pl.col("low") <= effective_limit_down + 0.005) @@ -1770,7 +1803,9 @@ def _compute_limit_signals_today(df: pl.DataFrame, instruments: pl.DataFrame) -> ).otherwise(None).cast(pl.Boolean) .alias("signal_limit_down_recovery"), # 炸板: 最高价曾触及涨停价 + 最终未封住 - pl.when((prev_raw > 0) & (pl.col("raw_high") > 0)) + pl.when(no_price_limit) + .then(False) + .when((valid_prev_raw | has_authoritative_up) & (pl.col("raw_high") > 0)) .then( (~is_limit_up.fill_null(True)) & (pl.col("raw_high") >= effective_limit_up - 0.005) diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index 267e925..a8bf231 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -36,6 +36,17 @@ def enriched_dirname(asset_type: str) -> str: return "kline_etf_enriched" if asset_type == "etf" else "kline_daily_enriched" +def _last_available_rows(df: pl.DataFrame, cutoff: date) -> pl.DataFrame: + """从已按 symbol/date 排序的数据中取每只标的最后一条有效状态。""" + if df.is_empty(): + return df + return ( + df.filter(pl.col("date") <= cutoff) + .group_by("symbol", maintain_order=True) + .last() + ) + + class DataStore: """唯一的存储入口 — 进程启动时创建。""" @@ -586,6 +597,17 @@ class KlineRepository: logger.info("enriched refresh step start: build live agg") self._build_live_agg(self._live_agg_baseline_date(latest)) logger.info("enriched refresh step done: build live agg (%.2fs)", time.perf_counter() - step) + repaired_today = self._restore_missing_latest_rows( + latest, df_today, df_full, + ) + if len(repaired_today) > len(df_today): + df_full = pl.concat( + [df_full.filter(pl.col("date") != latest), repaired_today], + how="diagonal_relaxed", + ).sort(["symbol", "date"]) + self._enriched_history_cache = df_full + self._enriched_cache = repaired_today + df_today = repaired_today logger.info("enriched 缓存已计算: %d 只, 日期 %s (即时计算)", len(df_today), latest) logger.info("enriched refresh done (%.2fs)", time.perf_counter() - started) return @@ -605,6 +627,67 @@ class KlineRepository: except Exception as e: # noqa: BLE001 logger.warning("enriched 缓存刷新失败: %s", e) + def _restore_missing_latest_rows( + self, + latest: date, + df_today: pl.DataFrame, + history: pl.DataFrame, + ) -> pl.DataFrame: + """用同日原始日K补齐旧实时快照漏写的正常成交股票。""" + if self._live_agg_cache is None or self._live_agg_cache.is_empty(): + return df_today + daily_path = ( + self.store.data_dir + / "kline_daily" + / f"date={latest.isoformat()}" + / "part.parquet" + ) + if not daily_path.exists(): + return df_today + + try: + from app.indicators.pipeline import compute_enriched_today, filter_halt_days + + daily = filter_halt_days(pl.read_parquet(daily_path)) + missing = daily.join( + df_today.select("symbol").unique(), + on="symbol", + how="anti", + ) + if missing.is_empty(): + return df_today + + missing_symbols = missing.select("symbol").unique() + previous = history.filter(pl.col("date") < latest).join( + missing_symbols, + on="symbol", + how="semi", + ) + if not previous.is_empty(): + previous = _last_available_rows(previous, latest) + recovered = compute_enriched_today( + self._live_agg_cache, + previous, + missing, + self.get_instruments(), + ) + if recovered.is_empty(): + return df_today + recovered = self._with_instrument_metadata("stock", recovered) + result = pl.concat( + [df_today, recovered], + how="diagonal_relaxed", + ).unique(subset=["symbol", "date"], keep="last").sort("symbol") + logger.info( + "enriched latest cache restored from daily: date=%s, rows=%d", + latest, + len(result) - len(df_today), + ) + return result + except Exception as e: # noqa: BLE001 + logger.warning("enriched latest cache restore skipped: %s", e) + return df_today + def _build_live_agg(self, latest: date) -> None: """从 OHLCV 即时计算递推状态 + 窗口聚合, 构建盘中实时聚合表。 @@ -622,8 +705,11 @@ class KlineRepository: hist_all = self._enriched_history_cache if "date" in hist_all.columns and hist_all["date"].min() <= start_60d: # 从历史缓存中提取所需列 (历史缓存已有指标列) - base_cols = ["symbol", "date", "open", "high", "low", "close", "volume", - "raw_close", "raw_high", "raw_low"] + base_cols = [ + "symbol", "date", "open", "high", "low", "close", "volume", + "raw_close", "raw_high", "raw_low", + "consecutive_limit_ups", "consecutive_limit_downs", + ] needed = [c for c in base_cols if c in hist_all.columns] step = time.perf_counter() logger.info("live agg step start: slice history cache") @@ -632,9 +718,6 @@ class KlineRepository: ).select(needed).sort(["symbol", "date"]) logger.info("live agg step done: slice history cache rows=%d (%.2fs)", len(df_hist), time.perf_counter() - step) - # 用历史缓存的指标列提取最新日状态 (无需再次 compute_indicators) - state_source = hist_all.filter(pl.col("date") == latest) - state_cols = [ "symbol", "ema5", "ema10", "ema20", "ema30", "ema60", @@ -644,7 +727,10 @@ class KlineRepository: "close", "high", "low", "annual_vol_20d", ] - existing_state = [c for c in state_cols if c in state_source.columns] + existing_state = [c for c in state_cols if c in hist_all.columns] + state_source = _last_available_rows( + hist_all.select("date", *existing_state), latest, + ) agg_a = state_source.select(existing_state) else: df_hist = pl.DataFrame() @@ -668,10 +754,13 @@ class KlineRepository: # 单独计算 _ema12 / _ema26 (compute_indicators 内部会 drop 掉) step = time.perf_counter() logger.info("live agg step start: ema state") - df_ema = df_hist.sort(["symbol", "date"]).with_columns([ - pl.col("close").ewm_mean(alpha=_ema_alpha(12), adjust=False).over("symbol").alias("_ema12"), - pl.col("close").ewm_mean(alpha=_ema_alpha(26), adjust=False).over("symbol").alias("_ema26"), - ]).filter(pl.col("date") == latest).select("symbol", "_ema12", "_ema26") + df_ema = _last_available_rows( + df_hist.sort(["symbol", "date"]).with_columns([ + pl.col("close").ewm_mean(alpha=_ema_alpha(12), adjust=False).over("symbol").alias("_ema12"), + pl.col("close").ewm_mean(alpha=_ema_alpha(26), adjust=False).over("symbol").alias("_ema26"), + ]).select("symbol", "date", "_ema12", "_ema26"), + latest, + ).select("symbol", "_ema12", "_ema26") agg_a = agg_a.join(df_ema, on="symbol", how="inner") logger.info("live agg step done: ema state (%.2fs)", time.perf_counter() - step) @@ -690,9 +779,9 @@ class KlineRepository: rsi_exprs.append(gain.ewm_mean(alpha=a, adjust=False).over("symbol").alias(f"_rsi_avg_gain_{n}")) rsi_exprs.append(loss.ewm_mean(alpha=a, adjust=False).over("symbol").alias(f"_rsi_avg_loss_{n}")) df_rsi = ( - df_rsi_base - .with_columns(rsi_exprs) - .filter(pl.col("date") == latest) + _last_available_rows( + df_rsi_base.with_columns(rsi_exprs), latest, + ) .select("symbol", *[f"_rsi_avg_gain_{n}" for n in (6, 14, 24)], *[f"_rsi_avg_loss_{n}" for n in (6, 14, 24)]) ) @@ -704,7 +793,9 @@ class KlineRepository: step = time.perf_counter() logger.info("live agg step start: adj factor state") adj_factor_df = ( - df_hist.filter(pl.col("date") == latest) + _last_available_rows( + df_hist.select("symbol", "date", "close", "raw_close"), latest, + ) .select("symbol", (pl.col("close") / pl.col("raw_close")).alias("_adj_factor")) ) agg_a = agg_a.join(adj_factor_df, on="symbol", how="left") @@ -728,14 +819,27 @@ class KlineRepository: agg_a = agg_a.join(df_vol, on="symbol", how="left") logger.info("live agg step done: annual vol state (%.2fs)", time.perf_counter() - step) - # 昨日连板数: 从 enriched parquet 取 (用于增量计算同向 +1) + # 昨日连板数: 使用每只股票最后一个有效交易日状态 (用于增量计算同向 +1) step = time.perf_counter() logger.info("live agg step start: consecutive state") - lf = scan_enriched_parquet(self._enriched_glob).filter(pl.col("date") == latest) consec_cols = [c for c in ["symbol", "consecutive_limit_ups", "consecutive_limit_downs"] - if c in lf.collect_schema().names()] + if c in df_hist.columns] + consec_source = df_hist + if len(consec_cols) != 3: + lf = ( + scan_enriched_parquet(self._enriched_glob) + .filter((pl.col("date") >= start_60d) & (pl.col("date") <= latest)) + .sort(["symbol", "date"]) + ) + consec_cols = [ + c for c in ["symbol", "consecutive_limit_ups", "consecutive_limit_downs"] + if c in lf.collect_schema().names() + ] + consec_source = lf.select("date", *consec_cols).collect() if len(consec_cols) == 3: - consec_df = lf.select(consec_cols).collect() + consec_df = _last_available_rows( + consec_source.select("date", *consec_cols), latest, + ) if not consec_df.is_empty(): consec = consec_df.select( "symbol", @@ -815,7 +919,8 @@ class KlineRepository: ) read_cols = [c for c in ["symbol", "date", "open", "high", "low", "close", "volume", - "raw_close", "raw_high", "raw_low"] + "raw_close", "raw_high", "raw_low", + "consecutive_limit_ups", "consecutive_limit_downs"] if c in lf.collect_schema().names()] df_hist = lf.select(read_cols).collect() @@ -834,7 +939,9 @@ class KlineRepository: "annual_vol_20d", ] existing_state = [c for c in state_cols if c in df_with_indicators.columns] - agg_a = df_with_indicators.filter(pl.col("date") == latest).select(existing_state) + agg_a = _last_available_rows( + df_with_indicators.select("date", *existing_state), latest, + ).select(existing_state) return df_hist, agg_a diff --git a/backend/tests/test_realtime_enriched_resume.py b/backend/tests/test_realtime_enriched_resume.py new file mode 100644 index 0000000..58134b0 --- /dev/null +++ b/backend/tests/test_realtime_enriched_resume.py @@ -0,0 +1,251 @@ +from __future__ import annotations + +from datetime import date, timedelta + +import polars as pl + +from app.indicators.pipeline import ( + ENRICHED_COLUMNS_BY_CATEGORY, + SIGNAL_DEPENDENCIES, + compute_enriched_today, + compute_indicators, +) +from app.tickflow.repository import DataStore, KlineRepository + + +def _historical_cache(latest: date) -> pl.DataFrame: + rows = [] + for symbol, days in (("600001.SH", 100), ("000820.SZ", 98)): + first = latest - timedelta(days=99) + for offset in range(days): + trade_date = first + timedelta(days=offset) + close = 10.0 + offset * 0.01 + rows.append({ + "symbol": symbol, + "date": trade_date, + "open": close, + "high": close + 0.1, + "low": close - 0.1, + "close": close, + "raw_close": close, + "raw_high": close + 0.1, + "raw_low": close - 0.1, + "volume": 1000.0 + offset, + "amount": close * (1000.0 + offset), + }) + return compute_indicators(pl.DataFrame(rows).sort(["symbol", "date"])) + + +def test_live_agg_uses_each_symbols_last_available_trading_state(tmp_path): + latest = date(2026, 7, 29) + repo = KlineRepository(DataStore(tmp_path)) + repo._enriched_history_cache = _historical_cache(latest) + + partition = tmp_path / "kline_daily_enriched" / f"date={latest.isoformat()}" + partition.mkdir(parents=True) + pl.DataFrame({ + "symbol": ["600001.SH"], + "date": [latest], + "open": [10.99], + "high": [11.09], + "low": [10.89], + "close": [10.99], + "volume": [1099.0], + "amount": [12078.01], + "raw_close": [10.99], + "raw_high": [11.09], + "raw_low": [10.89], + "turnover_rate": [1.0], + "consecutive_limit_ups": pl.Series([0], dtype=pl.UInt32), + "consecutive_limit_downs": pl.Series([0], dtype=pl.UInt32), + }).write_parquet(partition / "part.parquet") + resumed_date = latest - timedelta(days=2) + resumed_partition = ( + tmp_path / "kline_daily_enriched" / f"date={resumed_date.isoformat()}" + ) + resumed_partition.mkdir(parents=True) + pl.DataFrame({ + "symbol": ["000820.SZ"], + "date": [resumed_date], + "open": [10.97], + "high": [11.07], + "low": [10.87], + "close": [10.97], + "volume": [1097.0], + "amount": [12034.09], + "raw_close": [10.97], + "raw_high": [11.07], + "raw_low": [10.87], + "turnover_rate": [1.0], + "consecutive_limit_ups": pl.Series([2], dtype=pl.UInt32), + "consecutive_limit_downs": pl.Series([0], dtype=pl.UInt32), + }).write_parquet(resumed_partition / "part.parquet") + + repo._build_live_agg(latest) + + states = repo.get_live_agg().sort("symbol") + assert states["symbol"].to_list() == ["000820.SZ", "600001.SH"] + resumed = states.filter(pl.col("symbol") == "000820.SZ").row(0, named=True) + assert resumed["_prev_consec_up"] == 2 + + +def _live_state() -> pl.DataFrame: + return pl.DataFrame({ + "symbol": ["600001.SH"], + "ema5": [10.0], + "ema10": [10.0], + "ema20": [10.0], + "ema30": [10.0], + "ema60": [10.0], + "macd_dea": [0.0], + "kdj_k": [50.0], + "kdj_d": [50.0], + "atr_14": [0.2], + "close": [10.0], + "high": [10.1], + "low": [9.9], + "annual_vol_20d": [0.1], + "_ema12": [10.0], + "_ema26": [10.0], + "_adj_factor": [1.0], + "_vol_19d_pct_sum": [0.0], + "_vol_19d_pct_sq_sum": [0.0], + "_prev_consec_up": pl.Series([0], dtype=pl.UInt32), + "_prev_consec_down": pl.Series([0], dtype=pl.UInt32), + "_ma5_partial_sum": [40.0], + "_ma10_partial_sum": [90.0], + "_ma20_partial_sum": [190.0], + "_ma30_partial_sum": [290.0], + "_ma60_partial_sum": [590.0], + "_boll_partial_sum": [190.0], + "_boll_partial_sq_sum": [1900.0], + "_high_59d": [10.1], + "_low_59d": [9.9], + "_close_5d_ago": [10.0], + "_close_10d_ago": [10.0], + "_close_20d_ago": [10.0], + "_close_30d_ago": [10.0], + "_close_60d_ago": [10.0], + "_vol_ma5_partial_sum": [4000.0], + "_vol_ma10_partial_sum": [9000.0], + "_vol_ma5_prev_sum": [5000.0], + "_kdj_8d_low": [9.9], + "_kdj_8d_high": [10.1], + "_window_len": [59], + "_rsi_avg_gain_6": [0.01], + "_rsi_avg_loss_6": [0.01], + "_rsi_avg_gain_14": [0.01], + "_rsi_avg_loss_14": [0.01], + "_rsi_avg_gain_24": [0.01], + "_rsi_avg_loss_24": [0.01], + }) + + +def _previous_enriched() -> pl.DataFrame: + return pl.DataFrame({ + "symbol": ["600001.SH"], + "ma5": [10.0], + "ma10": [10.0], + "ma20": [10.0], + "ma60": [10.0], + "macd_dif": [0.0], + "macd_dea": [0.0], + "boll_upper": [10.2], + "boll_lower": [9.8], + "close": [10.0], + }) + + +def test_realtime_enriched_keeps_rows_without_history_and_limits_technical_fields(): + today = date(2026, 7, 30) + today_rows = pl.DataFrame({ + "symbol": ["600001.SH", "000820.SZ", "001000.SZ", "600002.SH"], + "date": [today] * 4, + "open": [10.1, 11.0, 11.0, 0.0], + "high": [10.2, 11.0, 11.0, 0.0], + "low": [10.0, 11.0, 11.0, 0.0], + "close": [10.2, 11.0, 11.0, 0.0], + "volume": [1200.0, 3000.0, 2000.0, 0.0], + "amount": [12240.0, 33000.0, 22000.0, 0.0], + "prev_close": [10.0, None, 10.0, 10.0], + }) + instruments = pl.DataFrame({ + "symbol": ["600001.SH", "000820.SZ", "001000.SZ", "600002.SH"], + "name": ["已有历史", "复牌股票", "上市新股", "停牌股票"], + "float_shares": [1_000_000.0] * 4, + "limit_up": [11.0, 11.0, 100000.0, 11.0], + "limit_down": [9.0, 9.0, 0.0, 9.0], + "as_of": [today] * 4, + }) + + result = compute_enriched_today( + _live_state(), + _previous_enriched(), + today_rows, + instruments, + ) + + assert result["symbol"].to_list() == today_rows["symbol"].to_list() + existing = result.filter(pl.col("symbol") == "600001.SH").row(0, named=True) + resumed = result.filter(pl.col("symbol") == "000820.SZ").row(0, named=True) + ipo = result.filter(pl.col("symbol") == "001000.SZ").row(0, named=True) + halted = result.filter(pl.col("symbol") == "600002.SH").row(0, named=True) + + assert existing["ma5"] is not None + assert resumed["raw_close"] == 11.0 + assert resumed["signal_limit_up"] is True + assert resumed["consecutive_limit_ups"] == 1 + technical_columns = { + column + for category in ( + "ma", "ema", "macd", "boll", "kdj", "atr", "volume", + "extremes", "momentum", "volatility", "rsi", + ) + for column in ENRICHED_COLUMNS_BY_CATEGORY[category] + } | set(SIGNAL_DEPENDENCIES) + assert all(resumed[column] is None for column in technical_columns) + assert ipo["signal_limit_up"] is False + assert halted["signal_limit_up"] is not True + assert "_has_history_state" not in result.columns + + +def test_repository_restores_active_rows_missing_from_latest_enriched(tmp_path): + today = date(2026, 7, 30) + repo = KlineRepository(DataStore(tmp_path)) + repo._live_agg_cache = _live_state() + repo._instruments_cache = pl.DataFrame({ + "symbol": ["600001.SH", "000820.SZ", "600002.SH"], + "name": ["已有历史", "复牌股票", "停牌股票"], + "float_shares": [1_000_000.0] * 3, + "limit_up": [11.0, 11.0, 11.0], + "limit_down": [9.0, 9.0, 9.0], + "as_of": [today] * 3, + }) + daily_partition = tmp_path / "kline_daily" / f"date={today.isoformat()}" + daily_partition.mkdir(parents=True) + pl.DataFrame({ + "symbol": ["600001.SH", "000820.SZ", "600002.SH"], + "date": [today] * 3, + "open": [10.2, 11.0, 0.0], + "high": [10.2, 11.0, 0.0], + "low": [10.2, 11.0, 0.0], + "close": [10.2, 11.0, 0.0], + "volume": [1200.0, 3000.0, 0.0], + "amount": [12240.0, 33000.0, 0.0], + }).write_parquet(daily_partition / "part.parquet") + existing_today = compute_enriched_today( + _live_state(), + _previous_enriched(), + pl.read_parquet(daily_partition / "part.parquet").head(1), + repo._instruments_cache, + ) + history = _previous_enriched().with_columns( + pl.lit(today - timedelta(days=1)).alias("date"), + ) + + restored = repo._restore_missing_latest_rows(today, existing_today, history) + + assert restored["symbol"].to_list() == ["000820.SZ", "600001.SH"] + resumed = restored.filter(pl.col("symbol") == "000820.SZ").row(0, named=True) + assert resumed["signal_limit_up"] is True + assert resumed["ma5"] is None