From 0635698a08559e3cc16f2aacb8c3283d6e0d87c5 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 20 Jul 2026 11:12:19 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E7=AD=96=E7=95=A5=E7=9B=91?= =?UTF-8?q?=E6=8E=A7=E5=BC=80=E5=90=AF=E5=90=8E=E7=BB=93=E6=9E=9C=E5=BD=92?= =?UTF-8?q?=E9=9B=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/tickflow/repository.py | 28 ++++- backend/tests/test_live_enriched_metadata.py | 103 +++++++++++++++++++ 2 files changed, 127 insertions(+), 4 deletions(-) create mode 100644 backend/tests/test_live_enriched_metadata.py diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index ab941bb..267e925 100644 --- a/backend/app/tickflow/repository.py +++ b/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": diff --git a/backend/tests/test_live_enriched_metadata.py b/backend/tests/test_live_enriched_metadata.py new file mode 100644 index 0000000..7537661 --- /dev/null +++ b/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