10 KiB

Raw Blame History Unescape Escape

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

comet_change	role	canonical_spec
storage-cache-refactor	technical-design	openspec

Storage Cache Refactor - Technical Design

1. 架构概览

1.1 目标架构

┌─────────────────────────────────────────────────────────────┐
│                     应用层                                   │
│  app/api/data.py  ←→  app/services/cache.py                │
│                              ↓                              │
│                    StorageManager                            │
│                    (storage_manager.py)                      │
└─────────────────────────────────────────────────────────────┘
                              ↓
        ┌─────────────────────┼─────────────────────┐
        ↓                     ↓                     ↓
   ┌─────────┐          ┌─────────┐          ┌─────────┐
   │  Redis  │          │  MySQL  │          │ SQLite  │
   │ (缓存)  │          │(持久化) │          │ (兜底)  │
   └─────────┘          └─────────┘          └─────────┘

1.2 核心组件

组件	文件	职责
StorageManager	`app/storage_manager.py`	封装三级存储逻辑，提供统一接口
RedisClient	`app/redis_client.py`	Redis 连接池和客户端封装
MySQLDatabase	`app/mysql_database.py`	MySQL 引擎和 SessionLocal
cache.py	`app/services/cache.py`	保持现有函数签名，内部调用 StorageManager

2. Redis 数据结构

2.1 行情数据缓存

Key: market_data:{symbol}:{period}
Value: JSON {
  "current_price": 123.45,
  "timestamp": "2026-07-04T10:00:00",
  "candles": [
    {"datetime": "...", "open": ..., "high": ..., "low": ..., "close": ...}
  ]
}
TTL: 30 天 (2592000 秒)

2.2 合约时间戳缓存

Key: symbol_timestamps:{symbol}
Value: JSON {
  "last_refresh_at": "2026-07-04T10:00:00",
  "refresh_count": 42
}
TTL: 30 天 (2592000 秒)

2.3 设计理由

结构化键值存储，便于按品种和周期精确查询
JSON 格式与当前 SQLite 存储格式兼容，迁移成本低
TTL 自动清理，避免内存无限增长

3. 数据流设计

3.1 读取流程

请求行情数据
    ↓
检查 Redis 缓存
    ├─ 命中 → 返回数据
    └─ 未命中 → 检查 MySQL 可用性
                ├─ 可用 → 读取 MySQL → 回填 Redis (TTL 30天) → 返回
                └─ 不可用 → 读取 SQLite → 返回

3.2 写入流程（刷新接口）

刷新行情数据
    ↓
删除 Redis 缓存 (market_data:{symbol}:{period})
    ↓
写入 MySQL（事务）
    ├─ 成功 → 更新 Redis 缓存 → 返回成功
    └─ 失败 → 返回错误（不更新 Redis）

3.3 降级流程

StorageManager 检查存储后端可用性
    ↓
Redis 可用？
    ├─ 是 → 使用 Redis 缓存
    └─ 否 → MySQL 可用？
            ├─ 是 → 使用 MySQL 持久化
            └─ 否 → 使用 SQLite 兜底

4. 降级检测机制

4.1 惰性恢复策略

class StorageManager:
    def __init__(self):
        self.redis_available = False
        self.mysql_available = False
        self.last_redis_check = 0
        self.last_mysql_check = 0
        self.check_interval = 30  # 秒
    
    def check_redis(self):
        """检查 Redis 可用性，30秒内不重复检测"""
        now = time.time()
        if now - self.last_redis_check < self.check_interval:
            return self.redis_available
        
        try:
            self.redis_client.ping()
            self.redis_available = True
            logger.info("Redis 连接恢复")
        except Exception as e:
            self.redis_available = False
            logger.warning(f"Redis 不可用: {e}")
        
        self.last_redis_check = now
        return self.redis_available
    
    def check_mysql(self):
        """检查 MySQL 可用性，30秒内不重复检测"""
        now = time.time()
        if now - self.last_mysql_check < self.check_interval:
            return self.mysql_available
        
        try:
            with self.mysql_engine.connect() as conn:
                conn.execute(text("SELECT 1"))
            self.mysql_available = True
            logger.info("MySQL 连接恢复")
        except Exception as e:
            self.mysql_available = False
            logger.warning(f"MySQL 不可用: {e}")
        
        self.last_mysql_check = now
        return self.mysql_available

4.2 启动时初始化

# app/main.py lifespan
storage_manager = StorageManager()
storage_manager.initialize()

# 检测可用性
redis_ok = storage_manager.check_redis()
mysql_ok = storage_manager.check_mysql()

if redis_ok and mysql_ok:
    logger.info("存储模式: Redis + MySQL")
elif mysql_ok:
    logger.warning("存储模式: MySQL (Redis 不可用)")
else:
    logger.error("存储模式: SQLite (Redis 和 MySQL 均不可用)")

5. 集成方式

5.1 cache.py 内部封装

# app/services/cache.py

def get_cached_data(db, symbol, data_type, periods, end_time=None, max_candles=100):
    """从缓存中获取完整的多周期数据"""
    storage = get_storage_manager()
    
    # 优先从 Redis/MySQL 读取
    if storage.is_available():
        try:
            result = storage.get_market_data(symbol, data_type, periods)
            if result:
                return result
        except Exception as e:
            logger.warning(f"StorageManager 读取失败，降级到 SQLite: {e}")
    
    # 降级到 SQLite
    return _get_from_sqlite(db, symbol, data_type, periods, end_time, max_candles)


def save_market_data(db, symbol, data):
    """保存采集结果到缓存"""
    storage = get_storage_manager()
    
    # 优先写入 Redis/MySQL
    if storage.is_available():
        try:
            storage.save_market_data(symbol, data)
            return
        except Exception as e:
            logger.warning(f"StorageManager 写入失败，降级到 SQLite: {e}")
    
    # 降级到 SQLite
    _save_to_sqlite(db, symbol, data)

5.2 API 层零改动

app/api/data.py 保持不变
所有接口仍使用 db: Session = Depends(get_db)
cache.py 内部自动选择存储后端

6. 数据迁移

6.1 迁移策略

# app/migration.py

def migrate_sqlite_to_mysql():
    """从 SQLite 迁移历史数据到 MySQL"""
    sqlite_engine = create_engine(f"sqlite:///{DB_PATH}")
    mysql_engine = create_mysql_engine()
    
    # 检查 MySQL 表是否为空
    with mysql_engine.connect() as conn:
        result = conn.execute(text("SELECT COUNT(*) FROM market_data"))
        count = result.scalar()
        
        if count > 0:
            logger.info("MySQL 已有数据，跳过迁移")
            return
    
    # 从 SQLite 读取数据
    with sqlite_engine.connect() as conn:
        result = conn.execute(text("SELECT * FROM market_data"))
        rows = result.fetchall()
    
    # 写入 MySQL
    with mysql_engine.begin() as conn:
        for row in rows:
            conn.execute(
                text("INSERT INTO market_data ..."),
                {...}
            )
    
    logger.info(f"数据迁移完成，共迁移 {len(rows)} 条记录")

6.2 迁移触发时机

应用启动时自动检测
MySQL 表为空时触发迁移
迁移完成后输出日志

7. 配置项

7.1 新增配置

# app/config.py

# 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", "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")

7.2 docker-compose.yml 新增服务

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
  
  mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: ${MYSQL_PASSWORD}
      MYSQL_DATABASE: ${MYSQL_DATABASE}
    ports:
      - "3306:3306"
    volumes:
      - mysql-data:/var/lib/mysql

8. 测试策略

8.1 单元测试

StorageManager 各方法独立测试
Redis 缓存命中/未命中场景
MySQL 读写场景
降级逻辑场景

8.2 集成测试

Redis + MySQL 正常模式
Redis 不可用降级到 MySQL
Redis + MySQL 均不可用降级到 SQLite
刷新接口双写一致性

8.3 故障注入

模拟 Redis 服务停止
模拟 MySQL 服务停止
验证降级和恢复逻辑

8.4 性能测试

对比改造前后读取延迟
验证 Redis 缓存命中率
监控 MySQL 查询性能

9. 风险与缓解

风险	影响	缓解措施
Redis 内存占用过高	系统内存不足	TTL 30 天自动清理，监控内存使用
双写一致性	MySQL 成功但 Redis 失败	Redis 失败仅记录日志，不影响持久化
降级检测延迟	恢复不及时	30 秒惰性恢复阈值，平衡性能和实时性
数据迁移失败	历史数据丢失	保留 SQLite 兜底，可手动回滚
MySQL 部署复杂度	运维成本增加	docker-compose 一键部署

10. 实施计划

10.1 阶段划分

依赖与配置: 添加 redis、pymysql 依赖，新增配置项
数据库模型: 创建 Redis/MySQL 连接模块
StorageManager: 实现三级存储逻辑
集成改造: 改造 cache.py，集成 StorageManager
数据迁移: 实现 SQLite → MySQL 迁移
测试验证: 单元测试、集成测试、故障注入

10.2 验收标准

Redis 缓存命中时，读取延迟 < 10ms
Redis 未命中时，从 MySQL 读取并回填
Redis 不可用时，自动降级到 MySQL
Redis + MySQL 均不可用时，降级到 SQLite
刷新接口双写成功，数据一致性保证
数据迁移完整，历史数据不丢失

10 KiB Raw Blame History Unescape Escape