修复策略监控开启后结果归零

backend/app/tickflow/repository.py

@@ -1893,6 +1893,24 @@ class KlineRepository:
                 )
             self._atomic_write_parquet(date_df.sort(["symbol", "date"]), out)
 
+    def _with_instrument_metadata(self, asset_type: str, df: pl.DataFrame) -> pl.DataFrame:
+        """补齐实时内存缓存所需的维表字段；这些字段不会写入 enriched 分区。"""
+        if asset_type not in {"stock", "etf"} or df.is_empty():
+            return df
+        instruments = self.get_instruments_asset(asset_type)
+        if instruments.is_empty() or "symbol" not in instruments.columns:
+            return df
+        metadata_cols = [
+            c for c in ("name", "total_shares", "float_shares")
+            if c in instruments.columns and c not in df.columns
+        ]
+        if not metadata_cols:
+            return df
+        metadata = instruments.select(["symbol", *metadata_cols]).unique(
+            subset=["symbol"], keep="last",
+        )
+        return df.join(metadata, on="symbol", how="left")
+
     def merge_live_enriched_asset(self, asset_type: str, df: pl.DataFrame) -> None:
         """按 symbol 合并当天 enriched 分区和内存缓存。用于少量自选实时。"""
         if df.is_empty() or "date" not in df.columns:
@@ -1910,9 +1928,10 @@ class KlineRepository:
         else:
             return
 
-        merged_cache = df
+        cache_df = self._with_instrument_metadata(asset_type, df)
+        merged_cache = cache_df
         if existing_cache is not None and not existing_cache.is_empty():
-            merged_cache = pl.concat([existing_cache, df], how="diagonal_relaxed").unique(
+            merged_cache = pl.concat([existing_cache, cache_df], how="diagonal_relaxed").unique(
                 subset=["symbol", "date"], keep="last"
             )
         merged_cache = merged_cache.sort(["symbol"])
@@ -1977,12 +1996,13 @@ class KlineRepository:
         if df.is_empty() or "date" not in df.columns:
             return
         dt = df["date"][0]
+        cache_df = self._with_instrument_metadata(asset_type, df).sort(["symbol"])
         if asset_type == "stock":
-            self._enriched_cache = df.sort(["symbol"])
+            self._enriched_cache = cache_df
             self._enriched_cache_date = dt
             table = "kline_daily_enriched"
         elif asset_type == "etf":
-            self._etf_enriched_cache = df.sort(["symbol"])
+            self._etf_enriched_cache = cache_df
             self._etf_enriched_cache_date = dt
             table = "kline_etf_enriched"
         elif asset_type == "index":

backend/tests/test_live_enriched_metadata.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+from datetime import date
+from pathlib import Path
+
+import polars as pl
+
+from app.strategy.engine import StrategyEngine
+from app.strategy.monitor import MonitorRuleEngine
+from app.tickflow.repository import DataStore, KlineRepository
+
+
+def _repo(tmp_path) -> KlineRepository:
+    repo = KlineRepository(DataStore(tmp_path))
+    repo._instruments_cache = pl.DataFrame({
+        "symbol": ["600000.SH", "000001.SZ"],
+        "name": ["浦发银行", "平安银行"],
+        "total_shares": [29_352_080_397.0, 19_405_918_198.0],
+        "float_shares": [29_352_080_397.0, 19_405_918_198.0],
+    })
+    return repo
+
+
+def _live_row(symbol: str, close: float) -> pl.DataFrame:
+    return pl.DataFrame({
+        "symbol": [symbol],
+        "date": [date(2026, 7, 20)],
+        "open": [close],
+        "high": [close],
+        "low": [close],
+        "close": [close],
+        "volume": [1000.0],
+        "amount": [close * 1000.0],
+        "raw_close": [close],
+        "raw_high": [close],
+        "raw_low": [close],
+    })
+
+
+def test_live_enriched_cache_keeps_instrument_metadata_without_persisting_it(tmp_path):
+    repo = _repo(tmp_path)
+
+    repo.flush_live_enriched_asset("stock", _live_row("600000.SH", 10.0))
+    repo.merge_live_enriched_asset("stock", _live_row("000001.SZ", 12.0))
+
+    cached, cached_date = repo.get_enriched_latest()
+    assert cached_date == date(2026, 7, 20)
+    assert cached.select("symbol", "name").sort("symbol").to_dicts() == [
+        {"symbol": "000001.SZ", "name": "平安银行"},
+        {"symbol": "600000.SH", "name": "浦发银行"},
+    ]
+    assert cached["total_shares"].null_count() == 0
+    assert cached["float_shares"].null_count() == 0
+
+    persisted = pl.read_parquet(
+        tmp_path / "kline_daily_enriched" / "date=2026-07-20" / "part.parquet"
+    )
+    assert "name" not in persisted.columns
+    assert "total_shares" not in persisted.columns
+    assert "float_shares" not in persisted.columns
+
+
+def test_history_strategy_monitor_keeps_live_row_with_exclude_st_enabled(tmp_path):
+    strategy_dir = tmp_path / "strategies"
+    strategy_dir.mkdir()
+    (strategy_dir / "history_strategy.py").write_text(
+        """import polars as pl
+
+META = {
+    "id": "history_strategy",
+    "name": "历史策略",
+    "basic_filter": {"exclude_st": True},
+}
+LOOKBACK_DAYS = 2
+
+def filter_history(df: pl.DataFrame, params: dict) -> pl.DataFrame:
+    return df
+""",
+        encoding="utf-8",
+    )
+    repo = _repo(tmp_path / "data")
+    live = _live_row("600000.SH", 10.0).with_columns(
+        pl.lit(30_000_000.0).alias("amount")
+    )
+    repo.flush_live_enriched_asset("stock", live)
+    current, _ = repo.get_enriched_latest()
+    history = current.with_columns(pl.lit(date(2026, 7, 17)).alias("date"))
+
+    monitor = MonitorRuleEngine()
+    monitor.set_strategy_engine(StrategyEngine([Path(strategy_dir)]))
+    monitor.set_history_loader(lambda _as_of, _lookback: history)
+    monitor.set_rules([{
+        "id": "history_strategy_monitor",
+        "name": "历史策略监控",
+        "type": "strategy",
+        "asset_type": "stock",
+        "strategy_id": "history_strategy",
+        "scope": "all",
+    }])
+
+    monitor.evaluate(current)
+
+    assert monitor.latest_strategy_results()["history_strategy"]["total"] == 1