buffer_platform/docs/superpowers/plans/2026-07-04-storage-cache-refactor.md

change	design-doc	base-ref
storage-cache-refactor	docs/superpowers/specs/2026-07-04-storage-cache-refactor-design.md	8b5e43f491

Storage Cache Refactor - 实施计划

基于 Design Doc 拆分的可执行任务。 目标：引入 Redis（缓存）+ MySQL（持久化）+ SQLite（兜底）三级存储架构，改造现有 cache.py 内部封装 StorageManager，API 层零改动。

---

任务总览

#	任务	预估	依赖
T1	添加 redis/pymysql 依赖	5min	-
T2	config.py 新增 Redis/MySQL 配置项	10min	-
T3	docker-compose.yml 新增 Redis/MySQL 服务	10min	T2
T4	创建 app/mysql_database.py	20min	T2
T5	创建 app/redis_client.py	20min	T2
T6	lifespan 中初始化 Redis/MySQL 连接	15min	T4, T5
T7	MySQL 表结构初始化（复用 ORM 模型）	15min	T4, T6
T8	创建 app/storage_manager.py 骨架 + 降级检测	30min	T4, T5, T6
T9	StorageManager 读取逻辑（Redis → MySQL → SQLite）	30min	T8
T10	StorageManager 写入逻辑 + 双写一致性	30min	T8
T11	创建数据迁移脚本 app/migration.py	25min	T4, T7
T12	启动时自动触发迁移	10min	T11
T13	改造 cache.py 内部封装 StorageManager	30min	T9, T10
T14	验证：Redis 缓存命中/未命中场景	15min	T13
T15	验证：降级到 MySQL / SQLite 场景	15min	T13
T16	验证：双写一致性 + 数据迁移完整性	15min	T13

---

详细任务

T1: 添加 redis/pymysql 依赖

文件: requirements.txt

操作:

• 在 requirements.txt 末尾追加：

  redis>=5.0.0
  pymysql>=1.1.0
  cryptography>=42.0.0
  

• cryptography 是 PyMySQL 启用加密连接所需的可选依赖

验证: pip install -r requirements.txt 成功安装无报错

---

T2: config.py 新增 Redis/MySQL 配置项

文件: app/config.py

操作:

• 在现有配置末尾新增以下配置块（参考 Design Doc §7.1）：

  # Redis 配置
  REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
  REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
  REDIS_DB = int(os.getenv("REDIS_DB", "0"))
  REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
  REDIS_TTL_SECONDS = int(os.getenv("REDIS_TTL_SECONDS", "2592000"))  # 30 天
  
  # MySQL 配置
  MYSQL_HOST = os.getenv("MYSQL_HOST", "localhost")
  MYSQL_PORT = int(os.getenv("MYSQL_PORT", "3306"))
  MYSQL_USER = os.getenv("MYSQL_USER", "root")
  MYSQL_PASSWORD = os.getenv("MYSQL_PASSWORD", "")
  MYSQL_DATABASE = os.getenv("MYSQL_DATABASE", "buffer_platform")
  

验证: from app.config import REDIS_HOST, MYSQL_HOST 无报错，默认值正确

---

T3: docker-compose.yml 新增 Redis/MySQL 服务

文件: docker-compose.yml

操作:

• 在 services 下新增 redis 和 mysql 服务（参考 Design Doc §7.2）：

  redis:
    image: redis:7-alpine
    container_name: futures-redis
    restart: unless-stopped
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    networks:
      - futures-network
  
  mysql:
    image: mysql:8.0
    container_name: futures-mysql
    restart: unless-stopped
    environment:
      MYSQL_ROOT_PASSWORD: ${MYSQL_PASSWORD:-buffer123}
      MYSQL_DATABASE: ${MYSQL_DATABASE:-buffer_platform}
    ports:
      - "3306:3306"
    volumes:
      - mysql-data:/var/lib/mysql
    networks:
      - futures-network
  

• 在顶层新增 volumes 声明：

  volumes:
    redis-data:
    mysql-data:
  

• 在 buffer-platform 的 environment 中新增 Redis/MySQL 环境变量传递

验证: docker-compose config 解析无语法错误

---

T4: 创建 app/mysql_database.py

文件: app/mysql_database.py（新建）

操作:

• 创建 MySQL 连接引擎和 SessionLocal，参考现有 app/database.py 的结构：

  from sqlalchemy import create_engine, text
  from sqlalchemy.orm import sessionmaker
  from app.config import MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE
  
  mysql_engine = None
  MySQLSessionLocal = None
  
  def init_mysql():
      """初始化 MySQL 连接引擎，失败时返回 None"""
      global mysql_engine, MySQLSessionLocal
      try:
          url = f"mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DATABASE}?charset=utf8mb4"
          mysql_engine = create_engine(url, pool_pre_ping=True, pool_recycle=3600)
          with mysql_engine.connect() as conn:
              conn.execute(text("SELECT 1"))
          MySQLSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=mysql_engine)
          return mysql_engine
      except Exception as e:
          logger.warning(f"MySQL 初始化失败: {e}")
          mysql_engine = None
          MySQLSessionLocal = None
          return None
  
  def get_mysql_session():
      """获取 MySQL 会话"""
      if MySQLSessionLocal is None:
          raise RuntimeError("MySQL 未初始化")
      return MySQLSessionLocal()
  

验证: 手动调用 init_mysql()，在无 MySQL 环境下返回 None 不抛异常

---

T5: 创建 app/redis_client.py

文件: app/redis_client.py（新建）

操作:

• 封装 Redis 连接池和客户端：

  import redis
  from app.config import REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD
  
  redis_client = None
  
  def init_redis():
      """初始化 Redis 客户端，失败时返回 None"""
      global redis_client
      try:
          redis_client = redis.Redis(
              host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB,
              password=REDIS_PASSWORD or None,
              decode_responses=True,
              socket_connect_timeout=5,
          )
          redis_client.ping()
          return redis_client
      except Exception as e:
          logger.warning(f"Redis 初始化失败: {e}")
          redis_client = None
          return None
  
  def get_redis():
      """获取 Redis 客户端实例"""
      return redis_client
  

验证: 手动调用 init_redis()，在无 Redis 环境下返回 None 不抛异常

---

T6: lifespan 中初始化 Redis/MySQL 连接

文件: app/main.py

操作:

• 在 lifespan 函数中，Base.metadata.create_all 之后新增：

  # 初始化 Redis 和 MySQL
  from app.redis_client import init_redis
  from app.mysql_database import init_mysql
  
  redis_ok = init_redis() is not None
  mysql_ok = init_mysql() is not None
  
  if redis_ok and mysql_ok:
      logger.info("存储模式: Redis + MySQL")
  elif mysql_ok:
      logger.warning("存储模式: MySQL (Redis 不可用)")
  else:
      logger.error("存储模式: SQLite (Redis 和 MySQL 均不可用)")
  

验证: 应用启动日志中输出存储模式信息，无 Redis/MySQL 时降级提示正常

---

T7: MySQL 表结构初始化（复用 ORM 模型）

文件: app/main.py（修改 lifespan）

操作:

• 在 T6 初始化代码之后，如果 MySQL 可用，使用现有 ORM 模型创建表：

  if mysql_ok:
      from app.mysql_database import mysql_engine
      from app.models import Base as MarketBase
      # 在 MySQL 中创建行情相关表
      MarketBase.metadata.create_all(bind=mysql_engine)
      logger.info("MySQL 表结构初始化完成")
  

• 注意：只需创建 market_data 和 symbol_timestamps 两张表，scheduled_tasks 等低频表仍留在 SQLite

验证: MySQL 中可查到 market_data 和 symbol_timestamps 表结构

---

T8: 创建 app/storage_manager.py 骨架 + 降级检测

文件: app/storage_manager.py（新建）

操作:

• 实现 StorageManager 类骨架，包含：
  • __init__: 持有 redis_client / mysql_engine 引用，初始化降级状态标志
  • check_redis() / check_mysql(): 惰性恢复检测（30 秒间隔，参考 Design Doc §4.1）
  • is_available(): 判断 Redis 或 MySQL 至少一个可用
  • initialize(): 启动时调用，检测各后端可用性并输出日志
  • 全局单例 get_storage_manager() 工厂函数
• 本任务只实现骨架和降级检测，不包含具体读写逻辑

验证: 实例化 StorageManager 后，check_redis() / check_mysql() 在无服务环境下返回 False

---

T9: StorageManager 读取逻辑（Redis → MySQL → SQLite）

文件: app/storage_manager.py（续写）

操作:

• 实现 get_market_data(symbol, data_type, periods) 方法：
  1. 检查 Redis 缓存（key 格式：market_data:{symbol}:{period}，参考 Design Doc §2.1）
  2. Redis 命中 → 反序列化 JSON 返回
  3. Redis 未命中 → 检查 MySQL 可用性
  4. MySQL 可用 → 从 MySQL 读取 → 回填 Redis（TTL 30 天）→ 返回
  5. MySQL 不可用 → 返回 None（由调用方降级到 SQLite）
• 实现 get_symbol_timestamp(symbol, data_type) 方法：
  • 类似逻辑，key 格式：symbol_timestamps:{symbol}

验证: 单元测试或手动测试：

• 无 Redis/MySQL 时返回 None
• 有 MySQL 时从 MySQL 读取并回填 Redis

---

T10: StorageManager 写入逻辑 + 双写一致性

文件: app/storage_manager.py（续写）

操作:

• 实现 save_market_data(symbol, data) 方法（参考 Design Doc §3.2）：
  1. 删除 Redis 缓存（market_data:{symbol}:{period}）
  2. 写入 MySQL（事务）
  3. MySQL 成功 → 更新 Redis 缓存
  4. MySQL 失败 → 记录日志，返回 False（不更新 Redis）
  5. Redis 更新失败 → 仅记录日志，不影响返回
• 实现 save_symbol_timestamp(symbol, data_type, refresh_time) 方法：
  • 类似双写逻辑
• 实现 delete_cache(symbol, periods) 方法：
  • 批量删除 Redis 缓存键

验证:

• MySQL 写入成功 → Redis 缓存同步更新
• MySQL 写入失败 → Redis 不更新，返回 False
• Redis 更新失败 → 不影响 MySQL 写入结果

---

T11: 创建数据迁移脚本 app/migration.py

文件: app/migration.py（新建）

操作:

• 实现 migrate_sqlite_to_mysql() 函数（参考 Design Doc §6.1）：
  1. 创建 SQLite 引擎（复用 DB_PATH）
  2. 检查 MySQL market_data 表是否为空
  3. 为空时从 SQLite 读取全部 market_data 记录
  4. 批量写入 MySQL
  5. 同样迁移 symbol_timestamps 表
  6. 输出迁移记录数日志
• 幂等设计：MySQL 已有数据时跳过

验证:

• MySQL 表为空时，迁移成功，数据条数一致
• MySQL 表非空时，跳过迁移，日志提示

---

T12: 启动时自动触发迁移

文件: app/main.py（修改 lifespan）

操作:

• 在 T7 表结构初始化之后新增：

  if mysql_ok:
      from app.migration import migrate_sqlite_to_mysql
      migrate_sqlite_to_mysql()
  

验证: 首次启动时日志输出迁移信息，再次启动时跳过迁移

---

T13: 改造 cache.py 内部封装 StorageManager

文件: app/services/cache.py（修改）

操作:

• 改造 save_market_data() 函数（参考 Design Doc §5.1）：
  • 优先调用 storage.save_market_data(symbol, data) 双写
  • 失败时降级到现有 SQLite 写入逻辑 _save_to_sqlite()
• 改造 get_cached_data() 函数：
  • 优先调用 storage.get_market_data(symbol, data_type, periods) 读取
  • 返回 None 时降级到现有 SQLite 读取逻辑 _get_from_sqlite()
• 改造 update_symbol_timestamp() 和 get_symbol_timestamp()：
  • 类似封装，优先走 StorageManager
• 保持所有函数的外部签名不变，API 层零改动
• 现有 SQLite 逻辑提取为内部 _save_to_sqlite() / _get_from_sqlite() 私有函数

验证:

• 现有 API 接口 /api/v1/data/latest/{symbol} 返回数据格式不变
• 现有 API 接口 /api/v1/data/batch-fetch 行为不变
• 无 Redis/MySQL 时完全降级到 SQLite，行为与改造前一致

---

T14: 验证 - Redis 缓存命中/未命中场景

操作:

• 启动 Redis + MySQL 服务
• 调用刷新接口写入数据
• 调用读取接口，验证首次从 MySQL 读取并回填 Redis
• 再次调用读取接口，验证从 Redis 缓存命中（检查日志）
• 等待 TTL 过期或手动删除 Redis key，验证回源 MySQL

验证标准:

• 日志中可见 "Redis 缓存命中" / "Redis 未命中，从 MySQL 读取" 信息
• 返回数据格式正确

---

T15: 验证 - 降级到 MySQL / SQLite 场景

操作:

• 停止 Redis 服务，调用读取接口
• 验证自动降级到 MySQL，日志输出 "Redis 不可用"
• 停止 Redis + MySQL 服务，调用读取接口
• 验证降级到 SQLite，日志输出 "存储模式: SQLite"
• 恢复 Redis 服务，等待 30 秒后验证自动恢复

验证标准:

• 各降级场景下接口正常返回，无 500 错误
• 日志清晰标识当前存储模式

---

T16: 验证 - 双写一致性 + 数据迁移完整性

操作:

• 双写一致性：
  1. 调用刷新接口写入数据
  2. 分别查询 MySQL 和 Redis，验证数据一致
  3. 模拟 Redis 写入失败（停止 Redis），验证 MySQL 数据不受影响
• 数据迁移完整性：
  1. 清空 MySQL 表
  2. 重启应用触发迁移
  3. 对比 SQLite 和 MySQL 数据条数一致

验证标准:

• MySQL 和 Redis 中数据内容一致
• Redis 故障不影响 MySQL 持久化
• 迁移前后数据条数一致

---

依赖关系图

T1 (依赖)  ──┐
T2 (配置)  ──┼── T3 (docker-compose)
             ├── T4 (mysql_database.py)
             └── T5 (redis_client.py)
                    │
T4 + T5 ─────── T6 (lifespan 初始化)
                    │
                 T7 (MySQL 建表)
                    │
                 T8 (StorageManager 骨架)
                   ╱ ╲
              T9      T10 (读/写逻辑)
                 ╲   ╱
                 T13 (cache.py 改造)
                   ╱ │ ╲
              T14  T15  T16 (验证)

T4 + T7 ──── T11 (迁移脚本) ──── T12 (启动迁移)

关键约束

1. API 层零改动: app/api/data.py 不修改，所有变更封装在 cache.py 和 storage_manager.py 内部
2. SQLite 保留: 不删除 SQLite 相关代码，作为最终兜底
3. 幂等迁移: 数据迁移脚本可重复执行，MySQL 有数据时跳过
4. 渐进式降级: 无 Redis/MySQL 时系统行为与改造前完全一致