buffer_platform/docs/superpowers/specs/2026-07-04-config-to-mysql-design.md

---
comet_change: config-to-mysql
role: technical-design
canonical_spec: openspec
---

# Config to MySQL - Technical Design

## 1. 架构概览

### 1.1 目标架构

```
业务代码 / API
    ↓
app/config_store.py  (统一入口)
    ↓
MySQL 可用？
    ↓ 是              ↓ 否
读/写 app_config   读/写 JSON 文件
```

### 1.2 核心组件

| 组件 | 职责 |
|------|------|
| `AppConfig` 模型 | MySQL 配置表 ORM 映射 |
| `ConfigStore` | 统一配置读写入口，封装降级逻辑 |
| `config_migration.py` | JSON → MySQL 一次性迁移 |
| JSON 文件 | MySQL 不可用时降级使用 |

## 2. 数据模型

### 2.1 AppConfig 表

```python
class AppConfig(Base):
    __tablename__ = "app_config"

    id = Column(Integer, primary_key=True, autoincrement=True)
    config_key = Column(String(64), unique=True, nullable=False, index=True)
    config_value = Column(JSON, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    created_at = Column(DateTime, default=datetime.utcnow)
```

### 2.2 配置键

| config_key | 对应文件 | 默认值 |
|------------|----------|--------|
| `symbols` | `config/symbols_config.json` | `{"futures": {}, "stock": {}}` |
| `ai` | `config/ai_config.json` | `{"models": [], "active_model": None, "analysis_settings": {...}}` |

## 3. 降级策略

### 3.1 读取流程

1. 检查 MySQL 可用性（复用 `StorageManager.check_mysql()`）
2. MySQL 可用：
   - 查询 `app_config`
   - 命中则返回
   - 未命中则读取 JSON 文件并回填数据库
3. MySQL 不可用：
   - 直接读取 JSON 文件

### 3.2 写入流程

1. 始终写入 JSON 文件（保证 fallback 最新）
2. MySQL 可用时同时写入 `app_config`
3. MySQL 不可用时跳过数据库写入

### 3.3 降级检测

复用 `StorageManager.check_mysql()` 惰性检测，避免每次配置读写都尝试连接数据库。

## 4. ConfigStore 接口

```python
class ConfigStore:
    def get_config(self, key: str, fallback: Optional[dict] = None) -> dict: ...
    def set_config(self, key: str, value: dict) -> bool: ...
    def load_from_json(self, key: str, fallback: Optional[dict] = None) -> dict: ...
    def save_to_json(self, key: str, value: dict) -> None: ...
```

### 4.1 单例获取

```python
_config_store = None

def get_config_store() -> ConfigStore:
    global _config_store
    if _config_store is None:
        _config_store = ConfigStore()
    return _config_store
```

## 5. 数据迁移

### 5.1 迁移时机

应用启动时（`lifespan`），在 MySQL 表结构初始化后执行。

### 5.2 迁移逻辑

```python
def migrate_configs_to_mysql():
    for key, file_path, default in CONFIGS:
        if _config_exists_in_mysql(key):
            continue
        value = _load_json(file_path, default)
        _save_to_mysql(key, value)
```

### 5.3 幂等性

- 检测 `app_config` 是否已有对应 key
- 有则跳过

## 6. 使用方修改

### 6.1 品种配置读取

替换以下文件中的 `_load_symbols_config()` / `load_symbols_config()`：

- `app/api/config.py`
- `app/api/futures_analysis.py`
- `app/api/trade_review.py`
- `app/services/trade_parser.py`
- `app/services/plan_generator.py`

统一使用：

```python
from app.config_store import get_config_store

symbols = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
```

### 6.2 AI 配置读取

替换 `app/api/ai_config.py` 和 `app/services/ai_analysis.py` 中的 `_load_ai_config()`：

```python
ai_config = get_config_store().get_config("ai", {...default...})
```

### 6.3 写入接口

`app/api/config.py` 和 `app/api/ai_config.py` 的保存接口改为：

```python
get_config_store().set_config("symbols", data)
```

## 7. 兼容性

- 保持 JSON 文件路径和 schema 不变
- API 接口签名不变
- 默认值与现有 JSON 默认值一致

## 8. 测试策略

### 8.1 单元测试

| 场景 | 测试 |
|------|------|
| MySQL 命中 | `get_config` 从数据库返回值 |
| MySQL 未命中 | `get_config` 读取 JSON 并回填 |
| MySQL 不可用 | `get_config` 读取 JSON |
| 写入 | `set_config` 双写数据库和文件 |
| 迁移幂等 | 数据库已有数据时跳过 |

### 8.2 集成测试

- 启动应用验证配置表创建
- 验证 API 读写正常
- 运行全部测试套件确保无回归

## 9. 风险与缓解

| 风险 | 缓解措施 |
|------|----------|
| MySQL 不可用时配置无法写入数据库 | 始终同步写入 JSON 文件 |
| 多处修改引入回归 | 统一 `ConfigStore` 接口 + 全面测试 |
| JSON 文件与数据库不一致 | 写入时双写；读取时优先数据库 |