diff --git a/app/services/cache.py b/app/services/cache.py
index 93e73dd..3cd9d6e 100644
--- a/app/services/cache.py
+++ b/app/services/cache.py
@@ -24,7 +24,21 @@ def is_cache_valid(
     period: str,
     ttl_seconds: int = CACHE_TTL_SECONDS,
 ) -> bool:
-    """检查指定品种+周期的缓存是否在有效期内"""
+    """检查指定品种+周期的缓存是否在有效期内，优先使用 StorageManager。"""
+    storage = get_storage_manager()
+
+    if storage.is_available():
+        try:
+            result = storage.get_market_data(symbol, data_type, [period])
+            if result and result.get("timeframes"):
+                timestamp_raw = result.get("fetched_at") or result.get("timestamp")
+                if timestamp_raw:
+                    newest = datetime.fromisoformat(timestamp_raw)
+                    age = (datetime.now() - newest).total_seconds()
+                    return age < ttl_seconds
+        except Exception as e:
+            logger.warning(f"StorageManager 缓存有效性检查失败，降级到 SQLite: {e}")
+
     record = db.query(MarketData).filter_by(
         symbol=symbol,
         data_type=data_type,
@@ -149,6 +163,44 @@ def _get_symbol_timestamp_sqlite(db: Session, symbol: str, data_type: str = "fut
     return record.last_refresh_at if record else None
 
 
+def _filter_candles(
+    candles: List[Dict],
+    end_time: Optional[datetime] = None,
+    max_candles: int = 100,
+) -> List[Dict]:
+    """按结束时间和最大数量过滤 K 线列表。"""
+    filter_end_time = end_time if end_time else datetime.now()
+    if filter_end_time.tzinfo is not None:
+        filter_end_time = filter_end_time.replace(tzinfo=None)
+
+    filtered = []
+    for candle in candles:
+        candle_time = candle.get("datetime") or candle.get("time")
+        if candle_time:
+            if isinstance(candle_time, str):
+                try:
+                    candle_dt = datetime.fromisoformat(candle_time.replace("Z", "+00:00"))
+                    if candle_dt.tzinfo is not None:
+                        candle_dt = candle_dt.replace(tzinfo=None)
+                except Exception:
+                    filtered.append(candle)
+                    continue
+            else:
+                candle_dt = candle_time
+                if candle_dt.tzinfo is not None:
+                    candle_dt = candle_dt.replace(tzinfo=None)
+
+            if candle_dt <= filter_end_time:
+                filtered.append(candle)
+        else:
+            filtered.append(candle)
+
+    if len(filtered) > max_candles:
+        filtered = filtered[-max_candles:]
+
+    return filtered
+
+
 def _get_from_sqlite(
     db: Session,
     symbol: str,
@@ -184,51 +236,11 @@ def _get_from_sqlite(
     newest = max(r.fetched_at for r in records)
     is_fresh = (now - newest).total_seconds() < CACHE_TTL_SECONDS
 
-    # 如果未指定结束时间,默认为当前时间
-    filter_end_time = end_time if end_time else now
-    
-    # 确保filter_end_time是naive datetime(无时区)
-    if filter_end_time.tzinfo is not None:
-        filter_end_time = filter_end_time.replace(tzinfo=None)
-
     timeframes = {}
     current_price = None
     for r in records:
         candles = json.loads(r.candles_json)
-        
-        # 过滤结束时间之前的K线数据
-        filtered_candles = []
-        for candle in candles:
-            candle_time = candle.get('datetime') or candle.get('time')
-            if candle_time:
-                # 解析K线时间
-                if isinstance(candle_time, str):
-                    try:
-                        candle_dt = datetime.fromisoformat(candle_time.replace('Z', '+00:00'))
-                        # 转换为naive datetime进行比较
-                        if candle_dt.tzinfo is not None:
-                            candle_dt = candle_dt.replace(tzinfo=None)
-                    except:
-                        filtered_candles.append(candle)
-                        continue
-                else:
-                    candle_dt = candle_time
-                    # 如果是aware datetime,转换为naive
-                    if candle_dt.tzinfo is not None:
-                        candle_dt = candle_dt.replace(tzinfo=None)
-                
-                # 只保留结束时间之前的数据
-                if candle_dt <= filter_end_time:
-                    filtered_candles.append(candle)
-            else:
-                filtered_candles.append(candle)
-        
-        # 限制K线数量,超过max_candles则取最新的max_candles条
-        if len(filtered_candles) > max_candles:
-            filtered_candles = filtered_candles[-max_candles:]
-        
-        timeframes[r.period] = filtered_candles
-            
+        timeframes[r.period] = _filter_candles(candles, end_time, max_candles)
         if current_price is None:
             current_price = r.current_price
 
@@ -243,6 +255,27 @@ def _get_from_sqlite(
     }
 
 
+def _representative_market_data(symbol: str, data: Dict, fetched_at: datetime) -> Optional[MarketData]:
+    """从输入数据构造一条代表性的 MarketData 记录（不写入数据库）。"""
+    data_type = data.get("type", "futures")
+    timeframes = data.get("timeframes", {})
+    if not timeframes:
+        return None
+
+    # 取第一个周期作为代表
+    period = next(iter(timeframes))
+    candles = timeframes[period]
+    return MarketData(
+        symbol=symbol,
+        data_type=data_type,
+        period=period,
+        candles_json=json.dumps(candles, ensure_ascii=False),
+        current_price=data.get("current_price"),
+        fetched_at=fetched_at,
+        candle_count=len(candles),
+    )
+
+
 def save_market_data(db: Session, symbol: str, data: Dict) -> Optional[MarketData]:
     """
     保存采集结果到缓存，优先写入 StorageManager（Redis/MySQL），失败时降级到 SQLite。
@@ -252,7 +285,7 @@ def save_market_data(db: Session, symbol: str, data: Dict) -> Optional[MarketDat
         data: 采集脚本返回的完整数据
 
     Returns:
-        SQLite 写入时返回 MarketData 记录；StorageManager 成功时返回 None
+        返回一条代表性的 MarketData 记录；降级到 SQLite 时返回实际持久化记录
     """
     storage = get_storage_manager()
     data_type = data.get("type", "futures")
@@ -262,7 +295,7 @@ def save_market_data(db: Session, symbol: str, data: Dict) -> Optional[MarketDat
         try:
             if storage.save_market_data(symbol, data):
                 storage.save_symbol_timestamp(symbol, data_type, now)
-                return None
+                return _representative_market_data(symbol, data, now)
             else:
                 raise Exception("save_market_data returned False")
         except Exception as e:
@@ -328,60 +361,46 @@ def get_latest_cached(
     data_type: str = "futures",
     period: Optional[str] = None,
 ) -> List[MarketData]:
-    """获取最新缓存数据"""
+    """获取最新缓存数据，优先使用 StorageManager。"""
+    storage = get_storage_manager()
+
+    if storage.is_available():
+        try:
+            result = storage.get_market_data(
+                symbol, data_type, [period] if period else None
+            )
+            if result and result.get("timeframes"):
+                timestamp_raw = result.get("fetched_at") or result.get("timestamp")
+                fetched_at = (
+                    datetime.fromisoformat(timestamp_raw)
+                    if timestamp_raw
+                    else datetime.now()
+                )
+                records = []
+                target_periods = [period] if period else list(result["timeframes"].keys())
+                for p in target_periods:
+                    candles = result["timeframes"].get(p, [])
+                    records.append(
+                        MarketData(
+                            symbol=symbol,
+                            data_type=data_type,
+                            period=p,
+                            candles_json=json.dumps(candles, ensure_ascii=False),
+                            current_price=result.get("current_price"),
+                            fetched_at=fetched_at,
+                            candle_count=len(candles),
+                        )
+                    )
+                return records
+        except Exception as e:
+            logger.warning(f"StorageManager 最新缓存读取失败，降级到 SQLite: {e}")
+
     query = db.query(MarketData).filter_by(symbol=symbol, data_type=data_type)
     if period:
         query = query.filter_by(period=period)
     return query.order_by(MarketData.fetched_at.desc()).all()
 
 
-def _merge_storage_market_data(
-    symbol: str,
-    data_type: str,
-    periods: Optional[List[str]],
-    storage_results: Dict[str, Optional[Dict]],
-) -> Optional[Dict]:
-    """将 StorageManager 按周期返回的数据合并为 get_cached_data 标准格式。"""
-    timeframes = {}
-    current_price = None
-    newest = None
-
-    for period, result in storage_results.items():
-        if not result:
-            continue
-        candles = result.get("candles", [])
-        timeframes[period] = candles
-        if current_price is None:
-            current_price = result.get("current_price")
-        timestamp_raw = result.get("timestamp")
-        if timestamp_raw:
-            try:
-                timestamp_dt = datetime.fromisoformat(timestamp_raw)
-                if newest is None or timestamp_dt > newest:
-                    newest = timestamp_dt
-            except Exception:
-                pass
-
-    if not timeframes:
-        return None
-
-    if newest is None:
-        newest = datetime.now()
-
-    now = datetime.now()
-    is_fresh = (now - newest).total_seconds() < CACHE_TTL_SECONDS
-
-    return {
-        "symbol": symbol,
-        "type": data_type,
-        "current_price": current_price,
-        "timestamp": newest.isoformat(),
-        "timeframes": timeframes,
-        "is_fresh": is_fresh,
-        "fetched_at": newest.isoformat(),
-    }
-
-
 def get_cached_data(
     db: Session,
     symbol: str,
@@ -408,13 +427,14 @@ def get_cached_data(
 
     if storage.is_available():
         try:
-            storage_results = {}
-            query_periods = periods if periods else []
-            for period in query_periods:
-                storage_results[period] = storage.get_market_data(symbol, data_type, period)
-
-            merged = _merge_storage_market_data(symbol, data_type, periods, storage_results)
-            if merged:
+            merged = storage.get_market_data(
+                symbol, data_type, periods, end_time=end_time, max_candles=max_candles
+            )
+            if merged and merged.get("timeframes"):
+                for period, candles in merged["timeframes"].items():
+                    merged["timeframes"][period] = _filter_candles(
+                        candles, end_time, max_candles
+                    )
                 return merged
         except Exception as e:
             logger.warning(f"StorageManager 读取失败，降级到 SQLite: {e}")
diff --git a/app/storage_manager.py b/app/storage_manager.py
index d690935..bad243b 100644
--- a/app/storage_manager.py
+++ b/app/storage_manager.py
@@ -100,19 +100,43 @@ class StorageManager:
         else:
             logger.warning("MySQL 不可用")
 
-    def get_market_data(self, symbol, data_type, periods):
-        """
-        读取行情数据：Redis → MySQL → None（由调用方降级到 SQLite）。
+    def _discover_periods(self, symbol, data_type):
+        """从 Redis 或 MySQL 发现该品种已缓存的所有周期。"""
+        if self.check_redis():
+            try:
+                pattern = f"market_data:{symbol}:*"
+                keys = self.redis_client.keys(pattern)
+                periods = set()
+                for key in keys:
+                    key_str = key.decode("utf-8") if isinstance(key, bytes) else key
+                    parts = key_str.split(":")
+                    if len(parts) == 3:
+                        periods.add(parts[2])
+                if periods:
+                    return sorted(periods)
+            except Exception as e:
+                logger.warning(f"Redis 发现周期失败 [{symbol}]: {e}")
 
-        Args:
-            symbol: 品种合约代码
-            data_type: 数据类型，如 futures
-            periods: 周期，如 5min（对应 Redis key 中的 period）
+        if not self.check_mysql():
+            return []
 
-        Returns:
-            dict 或 None
-        """
-        cache_key = f"market_data:{symbol}:{periods}"
+        try:
+            SessionLocal = sessionmaker(bind=self.mysql_engine)
+            with SessionLocal() as db:
+                rows = (
+                    db.query(MarketData.period)
+                    .filter_by(symbol=symbol, data_type=data_type)
+                    .distinct()
+                    .all()
+                )
+                return sorted([r.period for r in rows if r.period])
+        except Exception as e:
+            logger.warning(f"MySQL 发现周期失败 [{symbol}]: {e}")
+            return []
+
+    def _get_market_data_single_period(self, symbol, data_type, period):
+        """读取单个周期的行情数据：Redis → MySQL → None。"""
+        cache_key = f"market_data:{symbol}:{period}"
 
         # 1. 检查 Redis 缓存
         if self.check_redis():
@@ -133,7 +157,7 @@ class StorageManager:
             with SessionLocal() as db:
                 record = (
                     db.query(MarketData)
-                    .filter_by(symbol=symbol, data_type=data_type, period=periods)
+                    .filter_by(symbol=symbol, data_type=data_type, period=period)
                     .first()
                 )
                 if not record:
@@ -165,6 +189,138 @@ class StorageManager:
             logger.warning(f"MySQL 读取失败 [{cache_key}]: {e}")
             return None
 
+    @staticmethod
+    def _parse_candle_time(candle):
+        """解析单根 K 线的时间字段，返回 naive datetime 或 None。"""
+        candle_time = candle.get("datetime") or candle.get("time")
+        if not candle_time:
+            return None
+
+        if isinstance(candle_time, str):
+            try:
+                dt = datetime.fromisoformat(candle_time.replace("Z", "+00:00"))
+                if dt.tzinfo is not None:
+                    dt = dt.replace(tzinfo=None)
+                return dt
+            except Exception:
+                return None
+
+        if hasattr(candle_time, "tzinfo") and candle_time.tzinfo is not None:
+            return candle_time.replace(tzinfo=None)
+        return candle_time
+
+    def _filter_and_truncate(self, merged, end_time, max_candles):
+        """在合并结果上按 end_time 与 max_candles 过滤/截断每个周期。"""
+        if merged is None:
+            return None
+
+        filter_end_time = end_time if end_time else datetime.now()
+        if filter_end_time.tzinfo is not None:
+            filter_end_time = filter_end_time.replace(tzinfo=None)
+
+        timeframes = {}
+        for period, candles in merged.get("timeframes", {}).items():
+            filtered = []
+            for candle in candles:
+                candle_dt = self._parse_candle_time(candle)
+                if candle_dt is None or candle_dt <= filter_end_time:
+                    filtered.append(candle)
+
+            if len(filtered) > max_candles:
+                filtered = filtered[-max_candles:]
+
+            timeframes[period] = filtered
+
+        merged["timeframes"] = timeframes
+        return merged
+
+    def _merge_period_results(self, symbol, data_type, period_results):
+        """将按周期返回的结果合并为统一格式。"""
+        timeframes = {}
+        current_price = None
+        newest = None
+
+        for period, result in period_results.items():
+            if not result:
+                continue
+            candles = result.get("candles", [])
+            timeframes[period] = candles
+            if current_price is None:
+                current_price = result.get("current_price")
+            timestamp_raw = result.get("timestamp")
+            if timestamp_raw:
+                try:
+                    timestamp_dt = datetime.fromisoformat(timestamp_raw)
+                    if newest is None or timestamp_dt > newest:
+                        newest = timestamp_dt
+                except Exception:
+                    pass
+
+        if not timeframes:
+            return None
+
+        if newest is None:
+            newest = datetime.now()
+
+        now = datetime.now()
+        is_fresh = (now - newest).total_seconds() < CACHE_TTL_SECONDS
+
+        return {
+            "symbol": symbol,
+            "type": data_type,
+            "current_price": current_price,
+            "timestamp": newest.isoformat(),
+            "timeframes": timeframes,
+            "is_fresh": is_fresh,
+            "fetched_at": newest.isoformat(),
+        }
+
+    def get_market_data(self, symbol, data_type, periods=None, end_time=None, max_candles=100):
+        """
+        读取行情数据：Redis → MySQL → None（由调用方降级到 SQLite）。
+
+        Args:
+            symbol: 品种合约代码
+            data_type: 数据类型，如 futures
+            periods: 周期或周期列表，如 "5min" 或 ["5min", "15min"];
+                     为 None 时自动发现全部周期
+            end_time: 结束时间（可选），过滤该时间之前的 K 线
+            max_candles: 每个周期最大 K 线数量，默认 100
+
+        Returns:
+            periods 为字符串时返回旧版单周期 dict;
+            periods 为列表/None 时返回合并后的 dict（包含 timeframes）或 None
+        """
+        single_period = isinstance(periods, str)
+        query_periods = [periods] if single_period else periods
+
+        if query_periods is None:
+            query_periods = self._discover_periods(symbol, data_type)
+            if not query_periods:
+                return None
+
+        period_results = {}
+        for period in query_periods:
+            period_results[period] = self._get_market_data_single_period(
+                symbol, data_type, period
+            )
+
+        merged = self._merge_period_results(symbol, data_type, period_results)
+        if merged is None:
+            return None
+
+        merged = self._filter_and_truncate(merged, end_time, max_candles)
+
+        if single_period:
+            period = query_periods[0]
+            return {
+                "current_price": merged.get("current_price"),
+                "timestamp": merged.get("timestamp"),
+                "candles": merged["timeframes"].get(period, []),
+            }
+
+        return merged
+
     def get_symbol_timestamp(self, symbol, data_type):
         """
         读取合约时间戳：Redis → MySQL → None（由调用方降级到 SQLite）。
diff --git a/tests/test_cache_storage_integration.py b/tests/test_cache_storage_integration.py
index 68a9962..fffd73c 100644
--- a/tests/test_cache_storage_integration.py
+++ b/tests/test_cache_storage_integration.py
@@ -51,8 +51,8 @@ def storage_manager_mock():
 
 
 def _seed_sqlite_market_data(db, symbol="AG2606", data_type="futures"):
-    """在 SQLite 中写入行情数据。"""
-    fetched_at = datetime(2026, 7, 4, 10, 0, 0)
+    """在 SQLite 中写入行情数据（使用当前时间确保缓存未过期）。"""
+    fetched_at = datetime.now()
     db.add_all(
         [
             MarketData(
@@ -99,7 +99,7 @@ class TestSaveMarketData:
     """save_market_data 双写与降级测试。"""
 
     def test_uses_storage_manager_when_available(self, db, sample_data, storage_manager_mock):
-        """StorageManager 可用时优先写入存储层，并同步更新时间戳。"""
+        """StorageManager 可用时优先写入存储层，并同步更新时间戳，返回代表记录。"""
         with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
             from app.services import cache
 
@@ -108,7 +108,10 @@ class TestSaveMarketData:
         storage_manager_mock.is_available.assert_called_once()
         storage_manager_mock.save_market_data.assert_called_once_with("AG2606", sample_data)
         storage_manager_mock.save_symbol_timestamp.assert_called_once()
-        assert result is None
+        assert isinstance(result, MarketData)
+        assert result.symbol == "AG2606"
+        assert result.data_type == "futures"
+        assert result.current_price == 123.45
 
         sqlite_records = db.query(MarketData).filter_by(symbol="AG2606").all()
         assert len(sqlite_records) == 0
@@ -164,26 +167,30 @@ class TestGetCachedData:
     """get_cached_data 读取与降级测试。"""
 
     def test_uses_storage_manager_and_merges_periods(self, db, storage_manager_mock):
-        """StorageManager 返回各周期数据时，合并为统一格式。"""
+        """StorageManager 返回合并后的多周期数据。"""
         fetched_at = datetime.now()
 
-        def mock_get_market_data(symbol, data_type, period):
-            return {
-                "current_price": 123.45,
-                "timestamp": fetched_at.isoformat(),
-                "candles": [{"datetime": f"2026-07-04T09:{50 if period == '5min' else 45}:00", "open": 100, "close": 110}],
-            }
-
-        storage_manager_mock.get_market_data.side_effect = mock_get_market_data
+        storage_manager_mock.get_market_data.return_value = {
+            "symbol": "AG2606",
+            "type": "futures",
+            "current_price": 123.45,
+            "timestamp": fetched_at.isoformat(),
+            "fetched_at": fetched_at.isoformat(),
+            "timeframes": {
+                "5min": [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}],
+                "15min": [{"datetime": "2026-07-04T09:45:00", "open": 105, "close": 115}],
+            },
+            "is_fresh": True,
+        }
 
         with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
             from app.services import cache
 
             result = cache.get_cached_data(db, "AG2606", "futures", ["5min", "15min"])
 
-        assert storage_manager_mock.get_market_data.call_count == 2
-        storage_manager_mock.get_market_data.assert_any_call("AG2606", "futures", "5min")
-        storage_manager_mock.get_market_data.assert_any_call("AG2606", "futures", "15min")
+        storage_manager_mock.get_market_data.assert_called_once_with(
+            "AG2606", "futures", ["5min", "15min"], end_time=None, max_candles=100
+        )
 
         assert result is not None
         assert result["symbol"] == "AG2606"
@@ -226,6 +233,182 @@ class TestGetCachedData:
         assert "StorageManager 读取失败" in caplog.text
 
 
+    def test_passes_end_time_and_max_candles_to_storage(self, db, storage_manager_mock):
+        """get_cached_data 将 end_time / max_candles 透传给 StorageManager。"""
+        end_time = datetime(2026, 7, 4, 10, 0, 0)
+
+        with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
+            from app.services import cache
+
+            cache.get_cached_data(db, "AG2606", "futures", ["5min"], end_time=end_time, max_candles=50)
+
+        storage_manager_mock.get_market_data.assert_called_once_with(
+            "AG2606", "futures", ["5min"], end_time=end_time, max_candles=50
+        )
+
+    def test_applies_end_time_and_max_candles_to_storage_results(self, db, storage_manager_mock):
+        """StorageManager 返回的数据会按 end_time / max_candles 过滤截断。"""
+        fetched_at = datetime.now()
+
+        def mock_get_market_data(symbol, data_type, periods, end_time=None, max_candles=100):
+            return {
+                "symbol": symbol,
+                "type": data_type,
+                "current_price": 123.45,
+                "timestamp": fetched_at.isoformat(),
+                "fetched_at": fetched_at.isoformat(),
+                "timeframes": {
+                    "5min": [
+                        {"datetime": "2026-07-04T09:50:00", "open": 100, "close": 101},
+                        {"datetime": "2026-07-04T09:55:00", "open": 101, "close": 102},
+                        {"datetime": "2026-07-04T10:05:00", "open": 102, "close": 103},
+                    ]
+                },
+                "is_fresh": True,
+            }
+
+        storage_manager_mock.get_market_data.side_effect = mock_get_market_data
+
+        with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
+            from app.services import cache
+
+            result = cache.get_cached_data(
+                db, "AG2606", "futures", ["5min"],
+                end_time=datetime(2026, 7, 4, 10, 0, 0),
+                max_candles=1,
+            )
+
+        assert result is not None
+        candles = result["timeframes"]["5min"]
+        # 过滤掉 10:05 之后保留 09:50/09:55，再截断为最新 1 条
+        assert len(candles) == 1
+        assert candles[0]["datetime"] == "2026-07-04T09:55:00"
+
+    def test_queries_all_periods_when_periods_is_none(self, db, storage_manager_mock):
+        """periods=None 时优先走 StorageManager 查询全部周期。"""
+        fetched_at = datetime.now()
+
+        def mock_get_market_data(symbol, data_type, periods=None, end_time=None, max_candles=100):
+            return {
+                "symbol": symbol,
+                "type": data_type,
+                "current_price": 123.45,
+                "timestamp": fetched_at.isoformat(),
+                "fetched_at": fetched_at.isoformat(),
+                "timeframes": {"5min": [], "15min": []},
+                "is_fresh": True,
+            }
+
+        storage_manager_mock.get_market_data.side_effect = mock_get_market_data
+
+        with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
+            from app.services import cache
+
+            result = cache.get_cached_data(db, "AG2606", "futures", periods=None)
+
+        storage_manager_mock.get_market_data.assert_called_once_with(
+            "AG2606", "futures", None, end_time=None, max_candles=100
+        )
+        assert result is not None
+        assert "5min" in result["timeframes"]
+        assert "15min" in result["timeframes"]
+
+
+class TestCacheStatus:
+    """is_cache_valid / check_cache_status / get_latest_cached 存储层兼容测试。"""
+
+    def _make_merged_result(self, symbol="AG2606", data_type="futures", periods=None, age_seconds=0):
+        fetched_at = datetime.now() - __import__("datetime").timedelta(seconds=age_seconds)
+        timeframes = {}
+        for p in (periods or ["5min", "15min"]):
+            timeframes[p] = [{"datetime": fetched_at.isoformat(), "open": 100, "close": 110}]
+        return {
+            "symbol": symbol,
+            "type": data_type,
+            "current_price": 123.45,
+            "timestamp": fetched_at.isoformat(),
+            "fetched_at": fetched_at.isoformat(),
+            "timeframes": timeframes,
+            "is_fresh": True,
+        }
+
+    def test_is_cache_valid_uses_storage_when_available(self, db, storage_manager_mock):
+        """StorageManager 可用时优先判断缓存有效性。"""
+        storage_manager_mock.get_market_data.return_value = self._make_merged_result(age_seconds=10)
+
+        with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
+            from app.services import cache
+
+            assert cache.is_cache_valid(db, "AG2606", "futures", "5min") is True
+            storage_manager_mock.get_market_data.assert_called_once_with(
+                "AG2606", "futures", ["5min"]
+            )
+
+    def test_is_cache_valid_falls_back_to_sqlite(self, db, storage_manager_mock):
+        """StorageManager 无数据时降级到 SQLite 判断缓存有效性。"""
+        storage_manager_mock.get_market_data.return_value = None
+        _seed_sqlite_market_data(db)
+
+        with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
+            from app.services import cache
+
+            assert cache.is_cache_valid(db, "AG2606", "futures", "5min") is True
+
+    def test_check_cache_status_uses_storage_when_available(self, db, storage_manager_mock):
+        """StorageManager 可用时优先检查多周期缓存状态。"""
+        storage_manager_mock.get_market_data.return_value = self._make_merged_result()
+
+        with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
+            from app.services import cache
+
+            status = cache.check_cache_status(db, "AG2606", "futures", ["5min", "15min"])
+
+        assert status["all_valid"] is True
+        assert set(status["valid_periods"]) == {"5min", "15min"}
+        assert status["missing_periods"] == []
+
+    def test_check_cache_status_falls_back_to_sqlite(self, db, storage_manager_mock):
+        """StorageManager 无数据时降级到 SQLite 检查多周期缓存状态。"""
+        storage_manager_mock.get_market_data.return_value = None
+        _seed_sqlite_market_data(db)
+
+        with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
+            from app.services import cache
+
+            status = cache.check_cache_status(db, "AG2606", "futures", ["5min", "15min"])
+
+        assert status["all_valid"] is True
+        assert set(status["valid_periods"]) == {"5min", "15min"}
+
+    def test_get_latest_cached_uses_storage_when_available(self, db, storage_manager_mock):
+        """StorageManager 可用时优先获取最新缓存记录。"""
+        storage_manager_mock.get_market_data.return_value = self._make_merged_result()
+
+        with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
+            from app.services import cache
+
+            records = cache.get_latest_cached(db, "AG2606", "futures")
+
+        assert len(records) == 2
+        periods = {r.period for r in records}
+        assert periods == {"5min", "15min"}
+        assert all(isinstance(r, MarketData) for r in records)
+
+    def test_get_latest_cached_falls_back_to_sqlite(self, db, storage_manager_mock):
+        """StorageManager 无数据时降级到 SQLite 获取最新缓存记录。"""
+        storage_manager_mock.get_market_data.return_value = None
+        _seed_sqlite_market_data(db)
+
+        with patch("app.services.cache.get_storage_manager", return_value=storage_manager_mock):
+            from app.services import cache
+
+            records = cache.get_latest_cached(db, "AG2606", "futures")
+
+        assert len(records) == 2
+        periods = {r.period for r in records}
+        assert periods == {"5min", "15min"}
+
+
 class TestSymbolTimestamp:
     """合约时间戳函数封装测试。"""