diff --git a/tests/test_redis_cache_scenarios.py b/tests/test_redis_cache_scenarios.py
new file mode 100644
index 0000000..bf8fa4b
--- /dev/null
+++ b/tests/test_redis_cache_scenarios.py
@@ -0,0 +1,280 @@
+"""
+Redis 缓存命中/未命中场景验证测试
+
+覆盖：
+1. 调用 save_market_data 写入数据后，Redis 与 MySQL 同时存在一致数据
+2. 首次调用 get_market_data（Redis 未命中）从 MySQL 读取并回填 Redis
+3. 再次调用 get_market_data（Redis 命中）直接返回缓存数据，不访问 MySQL
+4. 手动删除 Redis key 后，再次读取回源 MySQL 并重新回填 Redis
+"""
+import json
+import logging
+from datetime import datetime
+from unittest.mock import MagicMock
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+from app.models import Base, MarketData
+from app.storage_manager import StorageManager
+
+# test_main_lifespan.py 中的 fixture 会 patch app.main.Base.metadata.create_all 且未正确恢复，
+# 导致后续依赖 create_all 的测试失败。这里保留原始方法引用，必要时恢复。
+_original_create_all = Base.metadata.create_all
+
+
+class FakeRedis:
+    """轻量级 Redis 替身，仅支持本测试所需的方法。"""
+
+    def __init__(self):
+        self._store = {}
+
+    def ping(self):
+        return True
+
+    def get(self, key):
+        return self._store.get(key)
+
+    def setex(self, key, ttl, value):
+        self._store[key] = value
+
+    def delete(self, *keys):
+        for key in keys:
+            self._store.pop(key, None)
+
+    def keys(self, pattern):
+        # 仅支持以 * 结尾的通配模式
+        prefix = pattern.rstrip("*")
+        return [k for k in self._store if k.startswith(prefix)]
+
+    def exists(self, key):
+        return key in self._store
+
+
+@pytest.fixture
+def mysql_engine():
+    """提供已创建表结构的内存 SQLite 引擎，作为 MySQL 替身。"""
+    # 防御：修复上游 fixture 对 Base.metadata.create_all 的 patch 泄漏
+    if isinstance(Base.metadata.create_all, MagicMock):
+        Base.metadata.create_all = _original_create_all
+
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(bind=engine)
+
+    # 确认表已创建
+    with engine.connect() as conn:
+        result = conn.execute(
+            text("SELECT name FROM sqlite_master WHERE type='table' AND name='market_data'")
+        )
+        assert result.fetchone() is not None
+
+    yield engine
+    engine.dispose()
+
+
+@pytest.fixture
+def fake_redis():
+    return FakeRedis()
+
+
+@pytest.fixture
+def sample_data():
+    fetched_at = datetime(2026, 7, 4, 10, 0, 0)
+    return {
+        "type": "futures",
+        "current_price": 123.45,
+        "timestamp": 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}],
+        },
+    }
+
+
+def _seed_mysql_market_data(engine, symbol="AG2606", data_type="futures"):
+    """在 MySQL 替身中写入行情数据。"""
+    fetched_at = datetime(2026, 7, 4, 10, 0, 0)
+    Session = sessionmaker(bind=engine)
+    with Session() as db:
+        db.add(
+            MarketData(
+                symbol=symbol,
+                data_type=data_type,
+                period="5min",
+                candles_json=json.dumps(
+                    [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}]
+                ),
+                current_price=123.45,
+                fetched_at=fetched_at,
+                candle_count=1,
+            )
+        )
+        db.commit()
+
+
+class TestSaveMarketData:
+    """save_market_data 写入与缓存回填场景。"""
+
+    def test_save_market_data_writes_mysql_and_backfills_redis(
+        self, fake_redis, mysql_engine, sample_data
+    ):
+        """写入后 MySQL 与 Redis 均保存一致数据。"""
+        manager = StorageManager(
+            redis_client=fake_redis, mysql_engine=mysql_engine
+        )
+
+        assert manager.save_market_data("AG2606", sample_data) is True
+
+        # MySQL 中应写入两条记录
+        Session = sessionmaker(bind=mysql_engine)
+        with Session() as db:
+            records = db.query(MarketData).filter_by(symbol="AG2606").all()
+            assert len(records) == 2
+            periods = {r.period for r in records}
+            assert periods == {"5min", "15min"}
+            for r in records:
+                assert r.current_price == 123.45
+
+        # Redis 中应回填对应缓存键
+        assert fake_redis.exists("market_data:AG2606:5min")
+        assert fake_redis.exists("market_data:AG2606:15min")
+
+        cached_5min = json.loads(fake_redis.get("market_data:AG2606:5min"))
+        assert cached_5min["current_price"] == 123.45
+        assert cached_5min["candles"] == sample_data["timeframes"]["5min"]
+
+
+class TestRedisCacheHitMiss:
+    """Redis 缓存命中/未命中读取场景。"""
+
+    def test_first_read_miss_falls_back_to_mysql_and_backfills_redis(
+        self, fake_redis, mysql_engine
+    ):
+        """Redis 未命中时从 MySQL 读取并回填 Redis。"""
+        _seed_mysql_market_data(mysql_engine)
+        manager = StorageManager(
+            redis_client=fake_redis, mysql_engine=mysql_engine
+        )
+
+        assert fake_redis.get("market_data:AG2606:5min") is None
+
+        result = manager.get_market_data("AG2606", "futures", "5min")
+
+        expected = {
+            "current_price": 123.45,
+            "timestamp": datetime(2026, 7, 4, 10, 0, 0).isoformat(),
+            "candles": [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}],
+        }
+        assert result == expected
+
+        # Redis 应已被回填
+        assert fake_redis.exists("market_data:AG2606:5min")
+        cached = json.loads(fake_redis.get("market_data:AG2606:5min"))
+        assert cached == expected
+
+    def test_second_read_hit_returns_from_redis_without_mysql_access(
+        self, fake_redis, mysql_engine
+    ):
+        """Redis 命中时直接返回缓存，不再访问 MySQL。"""
+        _seed_mysql_market_data(mysql_engine)
+        manager = StorageManager(
+            redis_client=fake_redis, mysql_engine=mysql_engine
+        )
+
+        first = manager.get_market_data("AG2606", "futures", "5min")
+        assert fake_redis.exists("market_data:AG2606:5min")
+
+        # 将 MySQL 引擎替换为会抛异常的 mock，以证明第二次读取不会访问 MySQL
+        bad_engine = MagicMock()
+        bad_engine.connect.side_effect = Exception("MySQL should not be accessed")
+        manager.mysql_engine = bad_engine
+
+        second = manager.get_market_data("AG2606", "futures", "5min")
+
+        assert second == first
+        bad_engine.connect.assert_not_called()
+
+    def test_after_redis_key_deleted_falls_back_to_mysql_again(
+        self, fake_redis, mysql_engine
+    ):
+        """手动删除 Redis key 后，读取回源 MySQL 并重新回填 Redis。"""
+        _seed_mysql_market_data(mysql_engine)
+        manager = StorageManager(
+            redis_client=fake_redis, mysql_engine=mysql_engine
+        )
+
+        first = manager.get_market_data("AG2606", "futures", "5min")
+        assert fake_redis.exists("market_data:AG2606:5min")
+
+        # 模拟 TTL 过期或手动失效
+        fake_redis.delete("market_data:AG2606:5min")
+        assert fake_redis.get("market_data:AG2606:5min") is None
+
+        second = manager.get_market_data("AG2606", "futures", "5min")
+
+        assert second == first
+        assert fake_redis.exists("market_data:AG2606:5min")
+
+    def test_cache_hit_data_matches_mysql_data(
+        self, fake_redis, mysql_engine
+    ):
+        """缓存命中时返回的数据应与 MySQL 中的数据一致。"""
+        _seed_mysql_market_data(mysql_engine)
+        manager = StorageManager(
+            redis_client=fake_redis, mysql_engine=mysql_engine
+        )
+
+        # 首次读取触发回填
+        from_redis = manager.get_market_data("AG2606", "futures", "5min")
+
+        # 直接从 MySQL 读取原始记录作为基准
+        Session = sessionmaker(bind=mysql_engine)
+        with Session() as db:
+            record = (
+                db.query(MarketData)
+                .filter_by(symbol="AG2606", data_type="futures", period="5min")
+                .first()
+            )
+            assert record is not None
+            assert from_redis["current_price"] == record.current_price
+            assert from_redis["candles"] == json.loads(record.candles_json)
+
+
+class TestCacheLogMessages:
+    """验证缓存命中/未命中的日志输出。"""
+
+    @pytest.mark.xfail(
+        reason="当前 StorageManager 未输出 'Redis 缓存命中' 日志，仅记录行为断言",
+        strict=True,
+    )
+    def test_cache_hit_logs_redis_hit(self, fake_redis, mysql_engine, caplog):
+        """Redis 命中时日志应包含 'Redis 缓存命中'。"""
+        _seed_mysql_market_data(mysql_engine)
+        manager = StorageManager(
+            redis_client=fake_redis, mysql_engine=mysql_engine
+        )
+        manager.get_market_data("AG2606", "futures", "5min")  # 回填
+
+        with caplog.at_level(logging.INFO, logger="app.storage_manager"):
+            manager.get_market_data("AG2606", "futures", "5min")  # 命中
+
+        assert "Redis 缓存命中" in caplog.text
+
+    @pytest.mark.xfail(
+        reason="当前 StorageManager 未输出 'Redis 未命中，从 MySQL 读取' 日志，仅记录行为断言",
+        strict=True,
+    )
+    def test_cache_miss_logs_redis_miss_and_mysql_read(
+        self, fake_redis, mysql_engine, caplog
+    ):
+        """Redis 未命中并回源 MySQL 时日志应包含 'Redis 未命中，从 MySQL 读取'。"""
+        _seed_mysql_market_data(mysql_engine)
+        manager = StorageManager(
+            redis_client=fake_redis, mysql_engine=mysql_engine
+        )
+
+        with caplog.at_level(logging.INFO, logger="app.storage_manager"):
+            manager.get_market_data("AG2606", "futures", "5min")
+
+        assert "Redis 未命中，从 MySQL 读取" in caplog.text