---
comet_change: storage-cache-refactor
role: technical-design
canonical_spec: 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 惰性恢复策略

```python
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 启动时初始化

```python
# 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 内部封装

```python
# 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 迁移策略

```python
# 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 新增配置

```python
# 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_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")
```

### 7.2 docker-compose.yml 新增服务

```yaml
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 阶段划分

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

### 10.2 验收标准

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