diff --git a/tests/test_dual_write_and_migration.py b/tests/test_dual_write_and_migration.py
new file mode 100644
index 0000000..b305830
--- /dev/null
+++ b/tests/test_dual_write_and_migration.py
@@ -0,0 +1,284 @@
+"""
+数据缓冲平台 - 双写一致性与数据迁移完整性验证测试
+
+本测试通过内存 Fake Redis 与 mock/fake MySQL 验证：
+1. save_market_data 成功后 MySQL 与 Redis 数据内容一致；
+2. Redis 写入失败不影响 MySQL 持久化；
+3. migrate_sqlite_to_mysql 的幂等性与迁移记录数正确性；
+4. 使用真实内存数据库做端到端迁移验证时暴露的生产代码缺陷。
+"""
+import json
+from datetime import datetime
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.models import Base, MarketData, SymbolTimestamp
+from app.storage_manager import StorageManager
+
+
+class FakeRedis:
+    """内存模拟 Redis 客户端，支持 setex / get / delete / keys / ping。"""
+
+    def __init__(self, fail_on_write=False):
+        self._store = {}
+        self.fail_on_write = fail_on_write
+
+    def ping(self):
+        return True
+
+    def get(self, key):
+        return self._store.get(key)
+
+    def setex(self, key, seconds, value):
+        if self.fail_on_write:
+            raise Exception("Redis 写入失败")
+        self._store[key] = value
+
+    def delete(self, *keys):
+        if self.fail_on_write:
+            raise Exception("Redis 删除失败")
+        for key in keys:
+            self._store.pop(key, None)
+
+    def keys(self, pattern):
+        prefix = pattern.replace("*", "")
+        return [k for k in self._store.keys() if k.startswith(prefix)]
+
+
+def _make_mysql_memory():
+    """创建内存 MySQL 模拟引擎和会话工厂。"""
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(bind=engine)
+    SessionLocal = sessionmaker(bind=engine)
+    return engine, SessionLocal
+
+
+def _create_sqlite_source(tmp_path, market_count=2, timestamp_count=1):
+    """创建包含测试数据的 SQLite 源数据库，并返回引擎与记录数。"""
+    db_path = tmp_path / "test_source.db"
+    engine = create_engine(f"sqlite:///{db_path}")
+    Base.metadata.create_all(bind=engine)
+    Session = sessionmaker(bind=engine)
+    session = Session()
+    fetched_at = datetime(2026, 7, 4, 10, 0, 0)
+    for i in range(market_count):
+        session.add(
+            MarketData(
+                symbol="AG2606",
+                data_type="futures",
+                period=f"{(i + 1) * 5}min",
+                candles_json='[{"open": 100, "close": 110}]',
+                current_price=123.45,
+                fetched_at=fetched_at,
+                candle_count=1,
+            )
+        )
+    for _ in range(timestamp_count):
+        session.add(
+            SymbolTimestamp(
+                symbol="AG2606",
+                data_type="futures",
+                last_refresh_at=fetched_at,
+                refresh_count=5,
+            )
+        )
+    session.commit()
+    session.close()
+    return engine, market_count, timestamp_count
+
+
+def _make_mysql_session(market_data_count=0, symbol_timestamp_count=0):
+    """构造模拟 MySQL 会话，支持记录写入和计数查询。"""
+    mysql_session = MagicMock()
+    mysql_session.__enter__ = MagicMock(return_value=mysql_session)
+    mysql_session.__exit__ = MagicMock(return_value=False)
+
+    added_market_data = []
+    added_symbol_timestamps = []
+
+    def mock_query(model):
+        query = MagicMock()
+        if model is MarketData:
+            query.count.return_value = market_data_count
+        elif model is SymbolTimestamp:
+            query.count.return_value = symbol_timestamp_count
+        return query
+
+    mysql_session.query.side_effect = mock_query
+    mysql_session.add_all.side_effect = lambda records: (
+        added_market_data.extend(records)
+        if records and isinstance(records[0], MarketData)
+        else added_symbol_timestamps.extend(records)
+    )
+
+    return mysql_session, added_market_data, added_symbol_timestamps
+
+
+@pytest.fixture
+def sample_market_data():
+    """构造与采集脚本格式一致的测试数据。"""
+    return {
+        "type": "futures",
+        "current_price": 123.45,
+        "timestamp": datetime(2026, 7, 4, 10, 0, 0).isoformat(),
+        "timeframes": {
+            "5min": [{"open": 100, "close": 110}],
+            "15min": [{"open": 105, "close": 115}],
+        },
+    }
+
+
+class TestDualWriteConsistency:
+    """双写一致性验证。"""
+
+    def test_save_market_data_writes_consistent_data_to_mysql_and_redis(
+        self, sample_market_data
+    ):
+        """save_market_data 成功后，MySQL 与 Redis 中数据内容一致。"""
+        redis_client = FakeRedis()
+        mysql_engine, MySQLSessionLocal = _make_mysql_memory()
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        result = manager.save_market_data("AG2606", sample_market_data)
+
+        assert result is True
+
+        # 验证 MySQL 持久化数据
+        with MySQLSessionLocal() as db:
+            records = db.query(MarketData).filter_by(symbol="AG2606").all()
+            assert len(records) == 2
+            mysql_by_period = {r.period: r for r in records}
+            assert "5min" in mysql_by_period
+            assert "15min" in mysql_by_period
+            assert mysql_by_period["5min"].current_price == 123.45
+            assert mysql_by_period["15min"].current_price == 123.45
+
+        # 验证 Redis 缓存数据
+        redis_5min = json.loads(redis_client.get("market_data:AG2606:5min"))
+        redis_15min = json.loads(redis_client.get("market_data:AG2606:15min"))
+        assert redis_5min["current_price"] == 123.45
+        assert redis_15min["current_price"] == 123.45
+        assert redis_5min["candles"] == [{"open": 100, "close": 110}]
+        assert redis_15min["candles"] == [{"open": 105, "close": 115}]
+
+        # 验证 MySQL 与 Redis 内容一致
+        assert (
+            mysql_by_period["5min"].current_price == redis_5min["current_price"]
+        )
+        assert (
+            mysql_by_period["15min"].current_price == redis_15min["current_price"]
+        )
+        assert (
+            json.loads(mysql_by_period["5min"].candles_json) == redis_5min["candles"]
+        )
+        assert (
+            json.loads(mysql_by_period["15min"].candles_json)
+            == redis_15min["candles"]
+        )
+
+    def test_redis_write_failure_does_not_affect_mysql_persistence(
+        self, sample_market_data
+    ):
+        """Redis 写入失败时，MySQL 数据持久化不受影响，save_market_data 仍返回 True。"""
+        redis_client = FakeRedis(fail_on_write=True)
+        mysql_engine, MySQLSessionLocal = _make_mysql_memory()
+        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)
+
+        result = manager.save_market_data("AG2606", sample_market_data)
+
+        assert result is True
+
+        # 验证 MySQL 仍有完整数据
+        with MySQLSessionLocal() 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"}
+
+        # 验证 Redis 无缓存写入
+        assert redis_client.get("market_data:AG2606:5min") is None
+        assert redis_client.get("market_data:AG2606:15min") is None
+
+
+class TestMigrationIntegrity:
+    """数据迁移完整性验证。"""
+
+    def test_migration_copies_expected_record_counts_from_sqlite_to_mysql(
+        self, tmp_path
+    ):
+        """MySQL 为空时，migrate_sqlite_to_mysql 迁移 SQLite 中全部记录。"""
+        sqlite_engine, sqlite_market_count, sqlite_timestamp_count = (
+            _create_sqlite_source(tmp_path)
+        )
+        mysql_session, added_market_data, added_symbol_timestamps = (
+            _make_mysql_session(market_data_count=0, symbol_timestamp_count=0)
+        )
+
+        with patch("app.migration.create_engine", return_value=sqlite_engine):
+            with patch(
+                "app.migration.MySQLSessionLocal", return_value=mysql_session
+            ):
+                from app.migration import migrate_sqlite_to_mysql
+
+                result = migrate_sqlite_to_mysql()
+
+        assert result is True
+        assert len(added_market_data) == sqlite_market_count
+        assert len(added_symbol_timestamps) == sqlite_timestamp_count
+        assert added_market_data[0].symbol == "AG2606"
+        assert added_market_data[0].period == "5min"
+        assert added_symbol_timestamps[0].symbol == "AG2606"
+        mysql_session.commit.assert_called_once()
+
+    def test_migration_is_idempotent_when_mysql_has_data(self, tmp_path):
+        """MySQL 已有数据时迁移被跳过，保持幂等性，不覆盖已有数据。"""
+        sqlite_engine, _, _ = _create_sqlite_source(tmp_path)
+        mysql_session, added_market_data, added_symbol_timestamps = (
+            _make_mysql_session(market_data_count=1, symbol_timestamp_count=1)
+        )
+
+        with patch("app.migration.create_engine", return_value=sqlite_engine):
+            with patch(
+                "app.migration.MySQLSessionLocal", return_value=mysql_session
+            ):
+                from app.migration import migrate_sqlite_to_mysql
+
+                result = migrate_sqlite_to_mysql()
+
+        assert result is False
+        assert len(added_market_data) == 0
+        assert len(added_symbol_timestamps) == 0
+        mysql_session.commit.assert_not_called()
+
+    @pytest.mark.xfail(
+        reason="app/migration.py 对 detached ORM 实例直接 add_all 不会触发 INSERT，"
+        "需使用 make_transient() 或重新构造记录。",
+        strict=False,
+    )
+    def test_migration_end_to_end_with_in_memory_databases(self, tmp_path):
+        """使用真实内存数据库验证迁移前后数据条数一致（当前已知缺陷）。"""
+        sqlite_engine, sqlite_market_count, sqlite_timestamp_count = (
+            _create_sqlite_source(tmp_path)
+        )
+
+        mysql_engine, MySQLSessionLocal = _make_mysql_memory()
+
+        with patch("app.migration.create_engine", return_value=sqlite_engine):
+            with patch("app.migration.MySQLSessionLocal", MySQLSessionLocal):
+                from app.migration import migrate_sqlite_to_mysql
+
+                result = migrate_sqlite_to_mysql()
+
+        assert result is True
+
+        mysql_session = MySQLSessionLocal()
+        try:
+            assert mysql_session.query(MarketData).count() == sqlite_market_count
+            assert (
+                mysql_session.query(SymbolTimestamp).count() == sqlite_timestamp_count
+            )
+        finally:
+            mysql_session.close()