merge: config-to-mysql 配置迁移到 MySQL

app/api/ai_config.py

@@ -1,14 +1,14 @@
 """
 AI模型配置接口 - 管理AI分析模型的配置
 """
-import json
 import logging
-from pathlib import Path
 from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 
+from app.config_store import get_config_store
+
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/ai-config", tags=["AI模型配置"])
 
@@ -38,45 +38,26 @@ class SaveAIConfigRequest(BaseModel):
     analysis_settings: Optional[dict] = None
 
 
-CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config"
+def _default_ai_config() -> dict:
-AI_CONFIG_FILE = CONFIG_DIR / "ai_config.json"
+    """AI 配置默认值"""
-
+    return {
-
+        "models": [],
-def _ensure_config_dir():
+        "active_model": None,
-    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
+        "analysis_settings": {
-
+            "enable_technical_analysis": True,
-
+            "enable_fundamental_analysis": False,
-def _load_ai_config() -> dict:
+            "enable_sentiment_analysis": False,
-    """加载AI配置"""
+            "risk_tolerance": "medium",
-    _ensure_config_dir()
+            "max_position_pct": 10
-    if not AI_CONFIG_FILE.exists():
-        return {
-            "models": [],
-            "active_model": None,
-            "analysis_settings": {
-                "enable_technical_analysis": True,
-                "enable_fundamental_analysis": False,
-                "enable_sentiment_analysis": False,
-                "risk_tolerance": "medium",
-                "max_position_pct": 10
-            }
         }
-    with open(AI_CONFIG_FILE, "r", encoding="utf-8") as f:
+    }
-        return json.load(f)
-
-
-def _save_ai_config(config: dict):
-    """保存AI配置"""
-    _ensure_config_dir()
-    with open(AI_CONFIG_FILE, "w", encoding="utf-8") as f:
-        json.dump(config, f, ensure_ascii=False, indent=4)
 
 
 @router.get("", response_model=AIConfigResponse)
 def get_ai_config():
     """获取当前AI模型配置"""
     try:
-        config = _load_ai_config()
+        config = get_config_store().get_config("ai", _default_ai_config())
         return {"success": True, "data": config}
     except Exception as e:
         logger.error(f"加载AI配置失败: {e}")
@@ -92,7 +73,7 @@ def save_ai_config(config: SaveAIConfigRequest):
             "active_model": config.active_model,
             "analysis_settings": config.analysis_settings or {}
         }
-        _save_ai_config(config_dict)
+        get_config_store().set_config("ai", config_dict)
         return {"success": True, "message": "AI配置保存成功"}
     except Exception as e:
         logger.error(f"保存AI配置失败: {e}")

app/api/config.py

@@ -3,12 +3,9 @@
 """
 import json
 import logging
-import shutil
-from pathlib import Path
 from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Body, Request
-from fastapi.responses import JSONResponse
 from sqlalchemy.orm import Session
 from pydantic import BaseModel
 
@@ -17,6 +14,7 @@ from app.services.collector import fetch_symbol_data
 from app.services.cache import save_market_data, check_cache_status, get_cached_data, create_task
 from app.services.scheduler import add_job
 from app.schemas import CandleItem, TimeframeData, SymbolDataResponse
+from app.config_store import get_config_store
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/config", tags=["品种配置"])
@@ -28,23 +26,11 @@ class BatchFetchRequest(BaseModel):
     data_type: str = "futures"
     selected_symbols: Optional[str] = None  # 逗号分隔的合约代码
 
-# 配置文件存储路径
-CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config"
-CONFIG_FILE = CONFIG_DIR / "symbols_config.json"
-
-
-def _ensure_config_dir():
-    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
-
 
 @router.get("")
 def get_config():
     """获取当前品种配置"""
-    _ensure_config_dir()
+    return get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
-    if not CONFIG_FILE.exists():
-        return {"futures": {}, "stock": {}}
-    with open(CONFIG_FILE, "r", encoding="utf-8") as f:
-        return json.load(f)
 
 
 @router.post("/upload")
@@ -60,8 +46,6 @@ async def upload_config(
         "stock": {"平安银行": "000001"}
     }
     """
-    _ensure_config_dir()
-    
     try:
         if file:
             content = await file.read()
@@ -76,8 +60,7 @@ async def upload_config(
         if not isinstance(data, dict):
             raise HTTPException(status_code=400, detail="配置文件必须是 JSON 对象")
 
-        with open(CONFIG_FILE, "w", encoding="utf-8") as f:
+        get_config_store().set_config("symbols", data)
-            json.dump(data, f, ensure_ascii=False, indent=4)
 
         futures_count = len(data.get("futures", {}))
         stock_count = len(data.get("stock", {}))
@@ -104,13 +87,7 @@ def batch_fetch_all(
     periods = request.periods
     data_type = request.data_type
     selected_symbols = request.selected_symbols
-    _ensure_config_dir()
+    config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
-    if not CONFIG_FILE.exists():
-        raise HTTPException(status_code=400, detail="请先上传品种配置文件")
-
-    with open(CONFIG_FILE, "r", encoding="utf-8") as f:
-        config = json.load(f)
-
     symbols_dict = config.get(data_type, {})
     if not symbols_dict:
         raise HTTPException(status_code=400, detail=f"配置中没有 {data_type} 类型的品种")
@@ -236,13 +213,7 @@ def batch_create_tasks(
     """
     根据配置文件为所有品种批量创建定时任务。
     """
-    _ensure_config_dir()
+    config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
-    if not CONFIG_FILE.exists():
-        raise HTTPException(status_code=400, detail="请先上传品种配置文件")
-
-    with open(CONFIG_FILE, "r", encoding="utf-8") as f:
-        config = json.load(f)
-
     symbols_dict = config.get(data_type, {})
     if not symbols_dict:
         raise HTTPException(status_code=400, detail=f"配置中没有 {data_type} 类型的品种")

app/api/futures_analysis.py

@@ -3,7 +3,6 @@
 """
 import json
 import logging
-from pathlib import Path
 from typing import Optional
 import threading
 
@@ -15,25 +14,20 @@ from app.analysis_db import get_analysis_db
 from app.analysis_models import FuturesAnalysis, WatchedSymbol, AIModelConfig, AnalysisSettings, AIAnalysisCache
 from app.services.cache import get_cached_data, get_latest_cached, save_market_data
 from app.services.collector import fetch_symbol_data
+from app.config_store import get_config_store
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/futures", tags=["期货智析"])
 
-CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config"
-SYMBOLS_CONFIG_FILE = CONFIG_DIR / "symbols_config.json"
-
 
 def _load_symbols_config() -> dict:
     """加载品种配置文件"""
-    if not SYMBOLS_CONFIG_FILE.exists():
+    return get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
-        return {"futures": {}, "stock": {}}
-    with open(SYMBOLS_CONFIG_FILE, "r", encoding="utf-8") as f:
-        return json.load(f)
 
 
 @router.get("/list")
 def get_futures_list(db: Session = Depends(get_db)):
-    """获取所有期货品种列表及摘要信息（从symbols_config.json读取）"""
+    """获取所有期货品种列表及摘要信息"""
     config = _load_symbols_config()
     futures_config = config.get("futures", {})
     

app/api/trade_review.py

@@ -487,13 +487,8 @@ def get_kline_with_trades(
 
 def _resolve_symbol_from_config(variety_code: str) -> str:
     """从品种配置中查找品种代码对应的当前合约"""
-    import json as _json
+    from app.config_store import get_config_store
-    from pathlib import Path
+    config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
-    config_path = Path(__file__).resolve().parent.parent.parent / "config" / "symbols_config.json"
-    if not config_path.exists():
-        return variety_code
-    with open(config_path, "r", encoding="utf-8") as f:
-        config = _json.load(f)
     for name, contract in config.get("futures", {}).items():
         if contract.upper().startswith(variety_code.upper()):
             return contract

app/config_migration.py

@@ -0,0 +1,52 @@
+"""
+配置迁移 - JSON 配置文件 → MySQL
+"""
+import logging
+
+from app.config_store import get_config_store, CONFIG_FILES, DEFAULT_CONFIGS
+
+logger = logging.getLogger(__name__)
+
+CONFIG_KEYS = ["symbols", "ai"]
+
+
+def migrate_configs_to_mysql():
+    """
+    将 JSON 配置文件迁移到 MySQL。
+
+    Returns:
+        bool: 迁移是否成功
+    """
+    store = get_config_store()
+
+    if not store.storage_manager.check_mysql():
+        logger.warning("MySQL 不可用，跳过配置迁移")
+        return True
+
+    if store.session_maker is None:
+        logger.warning("MySQL session 未初始化，跳过配置迁移")
+        return True
+
+    migrated_count = 0
+    skipped_count = 0
+
+    for key in CONFIG_KEYS:
+        try:
+            if store._config_exists_in_mysql(key):
+                logger.info(f"MySQL 配置 [{key}] 已存在，跳过迁移")
+                skipped_count += 1
+                continue
+
+            value = store._load_json_from_file(key)
+            if value is None:
+                value = DEFAULT_CONFIGS.get(key, {})
+                logger.info(f"JSON 配置 [{key}] 不存在或损坏，使用默认值迁移")
+            store._save_to_mysql(key, value)
+            migrated_count += 1
+            logger.info(f"配置 [{key}] 已从 JSON 迁移到 MySQL")
+        except Exception as e:
+            logger.error(f"迁移配置 [{key}] 失败: {e}")
+            return False
+
+    logger.info(f"配置迁移完成: 迁移 {migrated_count} 项，跳过 {skipped_count} 项")
+    return True

app/config_store.py

@@ -0,0 +1,171 @@
+"""
+统一配置存储模块
+
+支持 MySQL 优先、JSON 文件 fallback 的读写策略。
+"""
+import json
+import logging
+from pathlib import Path
+from typing import Optional
+
+from app.models import AppConfig
+from app.storage_manager import get_storage_manager
+from app.mysql_database import MySQLSessionLocal
+
+logger = logging.getLogger(__name__)
+
+CONFIG_DIR = Path(__file__).resolve().parent.parent / "config"
+CONFIG_FILES = {
+    "symbols": CONFIG_DIR / "symbols_config.json",
+    "ai": CONFIG_DIR / "ai_config.json",
+}
+
+DEFAULT_CONFIGS = {
+    "symbols": {"futures": {}, "stock": {}},
+    "ai": {
+        "models": [],
+        "active_model": None,
+        "analysis_settings": {
+            "enable_technical_analysis": True,
+            "enable_fundamental_analysis": False,
+            "enable_sentiment_analysis": False,
+            "risk_tolerance": "medium",
+            "max_position_pct": 10,
+        },
+    },
+}
+
+
+class ConfigStore:
+    """统一配置存储入口。"""
+
+    def __init__(
+        self,
+        storage_manager=None,
+        config_dir: Path = CONFIG_DIR,
+        session_maker=None,
+    ):
+        self.storage_manager = storage_manager or get_storage_manager()
+        self.config_dir = config_dir
+        self.session_maker = session_maker or MySQLSessionLocal
+        self.config_files = {
+            "symbols": self.config_dir / "symbols_config.json",
+            "ai": self.config_dir / "ai_config.json",
+        }
+
+    def get_config(self, key: str, fallback: Optional[dict] = None) -> dict:
+        """读取配置，优先从 MySQL，其次 JSON 文件。"""
+        if key not in self.config_files:
+            raise ValueError(f"未知配置键: {key}")
+
+        if self.storage_manager.check_mysql():
+            try:
+                db_value = self._get_from_mysql(key)
+                if db_value is not None:
+                    return db_value
+
+                # 数据库未命中：从 JSON 读取并回填
+                json_value = self._load_json_from_file(key)
+                if json_value is not None:
+                    self._save_to_mysql(key, json_value)
+                    return json_value
+                # JSON 不存在或损坏，使用 fallback 但不回填
+                return fallback if fallback is not None else DEFAULT_CONFIGS.get(key, {})
+            except Exception as e:
+                logger.warning(f"从 MySQL 读取配置 [{key}] 失败，降级到 JSON: {e}")
+
+        json_value = self._load_json_from_file(key)
+        if json_value is not None:
+            return json_value
+        return fallback if fallback is not None else DEFAULT_CONFIGS.get(key, {})
+
+    def set_config(self, key: str, value: dict) -> bool:
+        """写入配置，MySQL 可用时先写数据库再写 JSON；MySQL 不可用时只写 JSON。"""
+        if key not in self.config_files:
+            raise ValueError(f"未知配置键: {key}")
+
+        if self.storage_manager.check_mysql():
+            try:
+                if not self._save_to_mysql(key, value):
+                    logger.warning(f"写入 MySQL 配置 [{key}] 失败，保留 JSON 旧值")
+                    return False
+            except Exception as e:
+                logger.warning(f"写入 MySQL 配置 [{key}] 失败，保留 JSON 旧值: {e}")
+                return False
+
+        # MySQL 写入成功或 MySQL 不可用，都更新 JSON fallback
+        self._save_to_json(key, value)
+        return True
+
+    def _get_from_mysql(self, key: str) -> Optional[dict]:
+        """从 MySQL 读取配置。"""
+        if self.session_maker is None:
+            return None
+
+        with self.session_maker() as db:
+            record = db.query(AppConfig).filter(AppConfig.config_key == key).first()
+            if record:
+                return record.config_value
+        return None
+
+    def _save_to_mysql(self, key: str, value: dict) -> bool:
+        """保存配置到 MySQL。"""
+        if self.session_maker is None:
+            return False
+
+        with self.session_maker() as db:
+            record = db.query(AppConfig).filter(AppConfig.config_key == key).first()
+            if record:
+                record.config_value = value
+            else:
+                record = AppConfig(config_key=key, config_value=value)
+                db.add(record)
+            db.commit()
+        return True
+
+    def _load_json_from_file(self, key: str) -> Optional[dict]:
+        """从 JSON 文件读取配置，文件不存在或损坏时返回 None。"""
+        file_path = self.config_files[key]
+        if file_path.exists():
+            try:
+                with open(file_path, "r", encoding="utf-8") as f:
+                    return json.load(f)
+            except Exception as e:
+                logger.warning(f"读取 JSON 配置 [{key}] 失败: {e}")
+        return None
+
+    def _load_json(self, key: str, fallback: Optional[dict] = None) -> dict:
+        """从 JSON 文件读取配置，文件不存在或损坏时返回 fallback。"""
+        value = self._load_json_from_file(key)
+        if value is not None:
+            return value
+        return fallback if fallback is not None else DEFAULT_CONFIGS.get(key, {})
+
+    def _save_to_json(self, key: str, value: dict) -> None:
+        """保存配置到 JSON 文件（原子写入）。"""
+        file_path = self.config_files[key]
+        self.config_dir.mkdir(parents=True, exist_ok=True)
+        tmp_path = file_path.with_suffix(".tmp")
+        try:
+            with open(tmp_path, "w", encoding="utf-8") as f:
+                json.dump(value, f, ensure_ascii=False, indent=4)
+            tmp_path.replace(file_path)
+        except Exception:
+            if tmp_path.exists():
+                tmp_path.unlink()
+            raise
+
+    def _config_exists_in_mysql(self, key: str) -> bool:
+        """检查 MySQL 中是否已存在配置。"""
+        return self._get_from_mysql(key) is not None
+
+
+_config_store = None
+
+
+def get_config_store() -> ConfigStore:
+    """获取 ConfigStore 单例。"""
+    global _config_store
+    if _config_store is None:
+        _config_store = ConfigStore()
+    return _config_store

app/main.py

@@ -85,6 +85,15 @@ async def lifespan(app: FastAPI):
         from app.analysis_migration import migrate_analysis_sqlite_to_mysql
         migrate_analysis_sqlite_to_mysql()
 
+        # 在 MySQL 中创建配置表
+        from app.models import AppConfig
+        MarketBase.metadata.create_all(bind=mysql_engine, tables=[AppConfig.__table__])
+        logger.info("配置表 MySQL 表结构初始化完成")
+
+        # 迁移 JSON 配置到 MySQL
+        from app.config_migration import migrate_configs_to_mysql
+        migrate_configs_to_mysql()
+
     # 创建默认管理员账户
     from app.database import SessionLocal
     from app import auth_service

app/models.py

@@ -2,7 +2,10 @@
 数据缓冲平台 - 数据模型 (SQLAlchemy ORM)
 """
 from datetime import datetime
-from sqlalchemy import Column, String, Integer, Float, Text, DateTime, Boolean, Index, UniqueConstraint
+from sqlalchemy import (
+    Column, String, Integer, Float, Text, DateTime, Boolean,
+    Index, UniqueConstraint, JSON,
+)
 from app.database import Base
 
 
@@ -67,3 +70,20 @@ class ScheduledTask(Base):
 
     def __repr__(self):
         return f"<Task {self.symbol} every {self.interval_seconds}s enabled={self.enabled}>"
+
+
+class AppConfig(Base):
+    """应用配置表 - 存储 JSON 配置数据"""
+    __tablename__ = "app_config"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    config_key = Column(
+        String(64), unique=True, nullable=False, index=True,
+        comment="配置键，如 symbols / ai"
+    )
+    config_value = Column(JSON, nullable=False, comment="配置值(JSON)")
+    created_at = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
+    updated_at = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
+
+    def __repr__(self):
+        return f"<AppConfig {self.config_key}>"

app/services/ai_analysis.py

@@ -10,10 +10,6 @@ from sqlalchemy.orm import Session
 
 from app.analysis_models import AIAnalysisCache
 from app.services.cache import get_cached_data, get_latest_cached
-from pathlib import Path
-
-CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config"
-AI_CONFIG_FILE = CONFIG_DIR / "ai_config.json"
 
 logger = logging.getLogger(__name__)
 
@@ -152,11 +148,11 @@ class AIFuturesAnalyzer:
     def get_active_model(self) -> Optional[Dict]:
         """获取当前激活的AI模型配置"""
         try:
-            if not AI_CONFIG_FILE.exists():
+            from app.config_store import get_config_store
-                return None
+            config = get_config_store().get_config("ai", {
-            
+                "models": [],
-            with open(AI_CONFIG_FILE, "r", encoding="utf-8") as f:
+                "active_model": None,
-                config = json.load(f)
+            })
 
             models = config.get("models", [])
             active_model_name = config.get("active_model")

app/services/plan_generator.py

@@ -4,7 +4,6 @@ V2 交易计划生成器 - 5维度综合评分 + 多周期共振分析
 import json
 import logging
 from collections import Counter
-from pathlib import Path
 from typing import Dict, List, Optional, Tuple
 from datetime import datetime
 
@@ -17,9 +16,6 @@ from app.analysis_models import (
 
 logger = logging.getLogger(__name__)
 
-# 品种配置加载
-CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config"
-
 # 板块分类
 SECTOR_MAP = {
     "贵金属": ["沪银", "沪金"],
@@ -46,12 +42,8 @@ WEIGHTS = {
 
 def load_symbols_config() -> Dict[str, str]:
     """加载品种配置 {中文名: 合约代码}"""
-    config_path = CONFIG_DIR / "symbols_config.json"
+    from app.config_store import get_config_store
-    if not config_path.exists():
+    data = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
-        logger.warning(f"品种配置文件不存在: {config_path}")
-        return {}
-    with open(config_path, "r", encoding="utf-8") as f:
-        data = json.load(f)
     return data.get("futures", {})
 
 

app/services/trade_parser.py

@@ -23,19 +23,17 @@ _VARIETY_NAME_MAP = {}
 
 
 def _load_variety_name_map():
-    """从 symbols_config.json 加载品种名称映射"""
+    """从配置加载品种名称映射"""
     global _VARIETY_NAME_MAP
     if _VARIETY_NAME_MAP:
         return
-    config_path = Path(__file__).resolve().parent.parent.parent / "config" / "symbols_config.json"
+    from app.config_store import get_config_store
-    if config_path.exists():
+    config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
-        with open(config_path, "r", encoding="utf-8") as f:
+    # 正向: {"沪银": "AG2608"} -> 反向: {"AG": "沪银"}
-            config = json.load(f)
+    for name, contract in config.get("futures", {}).items():
-        # 正向: {"沪银": "AG2608"} -> 反向: {"AG": "沪银"}
+        variety = extract_variety(contract)
-        for name, contract in config.get("futures", {}).items():
+        if variety and variety not in _VARIETY_NAME_MAP:
-            variety = extract_variety(contract)
+            _VARIETY_NAME_MAP[variety] = name
-            if variety and variety not in _VARIETY_NAME_MAP:
-                _VARIETY_NAME_MAP[variety] = name
 
 
 def extract_variety(contract):

docs/superpowers/plans/2026-07-04-config-to-mysql.md

@@ -0,0 +1,128 @@
+---
+change: config-to-mysql
+design-doc: docs/superpowers/specs/2026-07-04-config-to-mysql-design.md
+base-ref: 4956086
+---
+
+# Config to MySQL - 实施计划
+
+> 基于 [Design Doc](docs/superpowers/specs/2026-07-04-config-to-mysql-design.md) 拆分的可执行任务。
+> 目标：将 `symbols_config.json` 和 `ai_config.json` 迁移到 MySQL，保留 JSON 文件作为 fallback。
+
+---
+
+## 任务总览
+
+| # | 任务 | 依赖 |
+|---|------|------|
+| T1 | 新增 AppConfig 模型和 ConfigStore | - |
+| T2 | 创建 config_migration.py 迁移脚本 | T1 |
+| T3 | 在 main.py lifespan 中集成 | T1, T2 |
+| T4 | 修改配置 API | T1 |
+| T5 | 修改业务使用方 | T1 |
+| T6 | 编写测试用例 | T1-T5 |
+
+---
+
+## T1: 新增 AppConfig 模型和 ConfigStore
+
+**文件**:
+- `app/models.py`
+- `app/config_store.py`
+
+**任务**:
+1. 在 `app/models.py` 中新增 `AppConfig` 模型
+2. 创建 `app/config_store.py` 和 `ConfigStore` 类
+3. 实现 `get_config(key, fallback)`
+4. 实现 `set_config(key, value)`
+5. 实现 `load_from_json(key, fallback)` 和 `save_to_json(key, value)`
+6. 实现 `get_config_store()` 单例函数
+
+**验收**:
+- `ConfigStore` 支持 MySQL 命中、未命中回填、JSON fallback
+- `set_config` 始终写入 JSON，MySQL 可用时同时写入数据库
+
+---
+
+## T2: 创建 config_migration.py 迁移脚本
+
+**文件**: `app/config_migration.py`
+
+**任务**:
+1. 定义配置映射（key → file path → default）
+2. 实现 `migrate_configs_to_mysql()`
+3. 检测数据库是否已有配置
+4. 从 JSON 读取并写入 MySQL
+
+**验收**:
+- 幂等性：数据库已有数据时跳过
+- 缺失配置时从 JSON 迁移
+
+---
+
+## T3: 在 main.py lifespan 中集成
+
+**文件**: `app/main.py`
+
+**任务**:
+1. 在 MySQL 初始化后创建 `app_config` 表
+2. 调用 `migrate_configs_to_mysql()`
+3. 记录迁移日志
+
+**验收**:
+- 启动时自动创建配置表
+- 自动迁移 JSON 配置
+
+---
+
+## T4: 修改配置 API
+
+**文件**:
+- `app/api/config.py`
+- `app/api/ai_config.py`
+
+**任务**:
+1. `config.py` 的 `get_config` 和 `upload_config` 使用 `ConfigStore`
+2. `ai_config.py` 的 `_load_ai_config` 和 `_save_ai_config` 使用 `ConfigStore`
+3. 保持 API 接口签名不变
+
+**验收**:
+- 上传配置同时写入数据库和 JSON 文件
+- 读取配置优先从数据库
+
+---
+
+## T5: 修改业务使用方
+
+**文件**:
+- `app/api/futures_analysis.py`
+- `app/api/trade_review.py`
+- `app/services/trade_parser.py`
+- `app/services/plan_generator.py`
+- `app/services/ai_analysis.py`
+
+**任务**:
+1. 替换所有 `_load_symbols_config()` / `load_symbols_config()` 为 `ConfigStore.get_config("symbols")`
+2. 替换 `_load_ai_config()` 为 `ConfigStore.get_config("ai")`
+
+**验收**:
+- 所有读取点统一使用 `ConfigStore`
+- 不修改业务逻辑
+
+---
+
+## T6: 编写测试用例
+
+**文件**: `tests/test_config_store.py`
+
+**任务**:
+1. 测试 MySQL 命中场景
+2. 测试 MySQL 未命中时读取 JSON 并回填
+3. 测试 MySQL 不可用时读取 JSON
+4. 测试 `set_config` 双写
+5. 测试迁移幂等性
+6. 运行全部测试套件确保无回归
+
+**验收**:
+- 所有测试通过
+- 测试覆盖核心降级场景

docs/superpowers/reports/2026-07-04-config-to-mysql-verify.md

@@ -0,0 +1,49 @@
+# Config to MySQL - 验证报告
+
+- **Change**: config-to-mysql
+- **日期**: 2026-07-04
+- **验证人**: AI Agent
+- **分支**: feature/20260704/config-to-mysql
+
+## 验证项
+
+| # | 检查项 | 结果 | 证据 |
+|---|--------|------|------|
+| 1 | tasks.md 全部完成 | ✅ | 所有 18 个子任务已勾选 |
+| 2 | 实现符合 Design Doc | ✅ | 单表 `app_config`、统一 `ConfigStore`、JSON fallback |
+| 3 | 实现符合 proposal | ✅ | symbols + ai 配置迁移到 MySQL，保留 JSON fallback |
+| 4 | 测试通过 | ✅ | `python -m pytest tests/` 110 passed, 2 xfailed |
+| 5 | 新增测试覆盖核心场景 | ✅ | `tests/test_config_store.py` 8 个用例覆盖命中/未命中/降级/迁移 |
+| 6 | 无硬编码密钥 | ✅ | 代码审查未发现 |
+| 7 | 无 SQL 注入风险 | ✅ | 使用 SQLAlchemy ORM 参数化查询 |
+
+## 代码审查
+
+**审查方式**: general_purpose_task 子代理审查
+
+### 已修复问题
+
+| 级别 | 问题 | 修复方式 |
+|------|------|----------|
+| CRITICAL | MySQL 写入失败时 JSON 已更新但 MySQL 仍为旧值 | `set_config` 改为先写 MySQL，成功后再写 JSON；失败返回 False 并保留 JSON 旧值 |
+| IMPORTANT | JSON 文件直接覆盖写入 | 使用临时文件 + `replace` 原子写入 |
+| IMPORTANT | JSON 损坏时回填 MySQL 默认值 | 新增 `_load_json_from_file`，损坏时返回 None，get_config 使用 fallback 但不回填 |
+| IMPORTANT | `_save_to_mysql` 返回 False 时未处理 | `set_config` 检查 `_save_to_mysql` 返回值 |
+| MINOR | 多处未使用导入/变量 | 已清理 |
+
+### 接受的风险
+
+| 级别 | 问题 | 接受原因 |
+|------|------|----------|
+| IMPORTANT | AI 配置中的 `api_key` 以明文存入 MySQL JSON 字段 | 超出本次 change 范围；JSON 文件原本也是明文存储，风险未扩大；建议后续单独 change 增加加密或敏感字段隔离 |
+
+## 分支处理
+
+- [ ] 本地合并到主分支
+- [ ] 推送并创建 PR
+- [ ] 保持分支
+- [ ] 丢弃工作
+
+## 结论
+
+验证通过。实现符合设计，测试覆盖核心场景，代码审查发现的 CRITICAL 和 IMPORTANT 问题已修复。

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

@@ -65,9 +65,10 @@ class AppConfig(Base):
 
 ### 3.2 写入流程
 
-1. 始终写入 JSON 文件（保证 fallback 最新）
+1. 检查 MySQL 可用性
-2. MySQL 可用时同时写入 `app_config`
+2. MySQL 可用时先写入 `app_config`；失败时保留 JSON 旧值并返回 False
-3. MySQL 不可用时跳过数据库写入
+3. MySQL 写入成功后，再原子写入 JSON 文件
+4. MySQL 不可用时只写入 JSON 文件
 
 ### 3.3 降级检测
 

openspec/changes/config-to-mysql/.comet.yaml

@@ -1,19 +1,19 @@
 workflow: full
-phase: build
+phase: verify
 context_compression: off
-build_mode: null
+build_mode: executing-plans
 build_pause: null
 subagent_dispatch: null
-tdd_mode: null
+tdd_mode: tdd
 review_mode: thorough
-isolation: null
+isolation: branch
 verify_mode: full
 auto_transition: true
 base_ref: f8d5ecd
 design_doc: docs/superpowers/specs/2026-07-04-config-to-mysql-design.md
-plan: null
+plan: docs/superpowers/plans/2026-07-04-config-to-mysql.md
 verify_result: null
-verification_report: null
+verification_report: docs/superpowers/reports/2026-07-04-config-to-mysql-verify.md
 branch_status: null
 created_at: 2026-07-04
 verified_at: null

openspec/changes/config-to-mysql/tasks.md

@@ -2,36 +2,36 @@
 
 ## 1. 数据模型与配置存储模块
 
-- [ ] 1.1 在 `app/models.py` 中新增 `AppConfig` 模型
+- [x] 1.1 在 `app/models.py` 中新增 `AppConfig` 模型
-- [ ] 1.2 创建 `app/config_store.py`，实现 `ConfigStore` 类
+- [x] 1.2 创建 `app/config_store.py`，实现 `ConfigStore` 类
-- [ ] 1.3 实现 `get_config(key)` 方法，支持 MySQL → JSON 降级
+- [x] 1.3 实现 `get_config(key)` 方法，支持 MySQL → JSON 降级
-- [ ] 1.4 实现 `set_config(key, value)` 方法，支持 MySQL + JSON 双写
+- [x] 1.4 实现 `set_config(key, value)` 方法，支持 MySQL + JSON 双写
-- [ ] 1.5 实现 `load_from_json(key, fallback)` 辅助方法
+- [x] 1.5 实现 `_load_json(key, fallback)` 和 `_save_to_json(key, value)` 辅助方法
 
 ## 2. 数据迁移
 
-- [ ] 2.1 创建 `app/config_migration.py`，实现 `migrate_configs_to_mysql()`
+- [x] 2.1 创建 `app/config_migration.py`，实现 `migrate_configs_to_mysql()`
-- [ ] 2.2 在 `app/main.py` lifespan 中创建 `app_config` 表
+- [x] 2.2 在 `app/main.py` lifespan 中创建 `app_config` 表
-- [ ] 2.3 在 `app/main.py` lifespan 中调用迁移函数
+- [x] 2.3 在 `app/main.py` lifespan 中调用迁移函数
-- [ ] 2.4 迁移幂等性：数据库已有配置时跳过
+- [x] 2.4 迁移幂等性：数据库已有配置时跳过
 
 ## 3. 修改配置 API
 
-- [ ] 3.1 修改 `app/api/config.py` 使用 `ConfigStore` 读写 `symbols` 配置
+- [x] 3.1 修改 `app/api/config.py` 使用 `ConfigStore` 读写 `symbols` 配置
-- [ ] 3.2 修改 `app/api/ai_config.py` 使用 `ConfigStore` 读写 `ai` 配置
+- [x] 3.2 修改 `app/api/ai_config.py` 使用 `ConfigStore` 读写 `ai` 配置
-- [ ] 3.3 保持 API 接口签名不变
+- [x] 3.3 保持 API 接口签名不变
 
 ## 4. 修改业务使用方
 
-- [ ] 4.1 修改 `app/api/futures_analysis.py` 从 `ConfigStore` 读取品种配置
+- [x] 4.1 修改 `app/api/futures_analysis.py` 从 `ConfigStore` 读取品种配置
-- [ ] 4.2 修改 `app/api/trade_review.py` 从 `ConfigStore` 读取品种配置
+- [x] 4.2 修改 `app/api/trade_review.py` 从 `ConfigStore` 读取品种配置
-- [ ] 4.3 修改 `app/services/trade_parser.py` 从 `ConfigStore` 读取品种配置
+- [x] 4.3 修改 `app/services/trade_parser.py` 从 `ConfigStore` 读取品种配置
-- [ ] 4.4 修改 `app/services/plan_generator.py` 从 `ConfigStore` 读取品种配置
+- [x] 4.4 修改 `app/services/plan_generator.py` 从 `ConfigStore` 读取品种配置
-- [ ] 4.5 修改 `app/services/ai_analysis.py` 从 `ConfigStore` 读取 AI 配置
+- [x] 4.5 修改 `app/services/ai_analysis.py` 从 `ConfigStore` 读取 AI 配置
 
 ## 5. 测试与验证
 
-- [ ] 5.1 编写 `tests/test_config_store.py`，覆盖 MySQL 命中/未命中/降级场景
+- [x] 5.1 编写 `tests/test_config_store.py`，覆盖 MySQL 命中/未命中/降级场景
-- [ ] 5.2 编写迁移测试，验证幂等性
+- [x] 5.2 编写迁移测试，验证幂等性
-- [ ] 5.3 运行全部测试套件，确保无回归
+- [x] 5.3 运行全部测试套件，确保无回归（110 passed, 2 xfailed）
-- [ ] 5.4 启动应用验证配置读取正常
+- [x] 5.4 启动应用验证配置读取正常

tests/test_config_store.py

@@ -0,0 +1,197 @@
+"""配置存储测试：MySQL + JSON fallback。"""
+import json
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.config_store import ConfigStore
+
+
+def _make_session_maker(mock_db):
+    """构造正确的 sessionmaker mock，使 with 语句返回 mock_db。"""
+    session_obj = MagicMock()
+    session_obj.__enter__ = MagicMock(return_value=mock_db)
+    session_obj.__exit__ = MagicMock(return_value=False)
+
+    session_maker = MagicMock(return_value=session_obj)
+    return session_maker
+
+
+class TestConfigStoreGet:
+    """ConfigStore.get_config 测试。"""
+
+    def test_get_config_returns_mysql_value_on_hit(self, tmp_path):
+        """MySQL 命中时返回数据库中的配置。"""
+        mock_config = MagicMock()
+        mock_config.config_value = {"futures": {"沪银": "AG2608"}}
+
+        mock_db = MagicMock()
+        mock_db.query.return_value.filter.return_value.first.return_value = mock_config
+
+        storage = MagicMock()
+        storage.check_mysql.return_value = True
+
+        store = ConfigStore(
+            storage_manager=storage,
+            config_dir=tmp_path,
+            session_maker=_make_session_maker(mock_db),
+        )
+
+        result = store.get_config("symbols", {"futures": {}})
+
+        assert result == {"futures": {"沪银": "AG2608"}}
+
+    def test_get_config_reads_json_and_backfills_when_mysql_miss(self, tmp_path):
+        """MySQL 未命中时读取 JSON 并回填数据库。"""
+        json_path = tmp_path / "symbols_config.json"
+        json_path.write_text(json.dumps({"futures": {"沪银": "AG2608"}}, ensure_ascii=False), encoding="utf-8")
+
+        mock_db = MagicMock()
+        mock_db.query.return_value.filter.return_value.first.return_value = None
+
+        storage = MagicMock()
+        storage.check_mysql.return_value = True
+
+        store = ConfigStore(
+            storage_manager=storage,
+            config_dir=tmp_path,
+            session_maker=_make_session_maker(mock_db),
+        )
+
+        result = store.get_config("symbols", {"futures": {}})
+
+        assert result == {"futures": {"沪银": "AG2608"}}
+        mock_db.add.assert_called_once()
+        mock_db.commit.assert_called_once()
+
+    def test_get_config_falls_back_to_json_when_mysql_unavailable(self, tmp_path):
+        """MySQL 不可用时直接读取 JSON。"""
+        json_path = tmp_path / "symbols_config.json"
+        json_path.write_text(json.dumps({"futures": {"沪银": "AG2608"}}, ensure_ascii=False), encoding="utf-8")
+
+        storage = MagicMock()
+        storage.check_mysql.return_value = False
+
+        store = ConfigStore(
+            storage_manager=storage,
+            config_dir=tmp_path,
+            session_maker=None,
+        )
+
+        result = store.get_config("symbols", {"futures": {}})
+
+        assert result == {"futures": {"沪银": "AG2608"}}
+
+    def test_get_config_returns_fallback_when_no_mysql_and_no_json(self, tmp_path):
+        """MySQL 和 JSON 都不存在时返回 fallback。"""
+        storage = MagicMock()
+        storage.check_mysql.return_value = False
+
+        store = ConfigStore(
+            storage_manager=storage,
+            config_dir=tmp_path,
+            session_maker=None,
+        )
+
+        result = store.get_config("symbols", {"futures": {"默认": "DF2609"}})
+
+        assert result == {"futures": {"默认": "DF2609"}}
+
+
+class TestConfigStoreSet:
+    """ConfigStore.set_config 测试。"""
+
+    def test_set_config_writes_to_mysql_and_json(self, tmp_path):
+        """MySQL 可用时同时写入数据库和 JSON。"""
+        mock_existing = MagicMock()
+
+        mock_db = MagicMock()
+        mock_db.query.return_value.filter.return_value.first.return_value = mock_existing
+
+        storage = MagicMock()
+        storage.check_mysql.return_value = True
+
+        store = ConfigStore(
+            storage_manager=storage,
+            config_dir=tmp_path,
+            session_maker=_make_session_maker(mock_db),
+        )
+
+        result = store.set_config("symbols", {"futures": {"沪银": "AG2608"}})
+
+        assert result is True
+        assert mock_existing.config_value == {"futures": {"沪银": "AG2608"}}
+        mock_db.commit.assert_called_once()
+
+        json_path = tmp_path / "symbols_config.json"
+        assert json_path.exists()
+        assert json.loads(json_path.read_text(encoding="utf-8")) == {"futures": {"沪银": "AG2608"}}
+
+    def test_set_config_writes_only_json_when_mysql_unavailable(self, tmp_path):
+        """MySQL 不可用时只写入 JSON。"""
+        storage = MagicMock()
+        storage.check_mysql.return_value = False
+
+        store = ConfigStore(
+            storage_manager=storage,
+            config_dir=tmp_path,
+            session_maker=None,
+        )
+
+        result = store.set_config("symbols", {"futures": {"沪银": "AG2608"}})
+
+        assert result is True
+
+        json_path = tmp_path / "symbols_config.json"
+        assert json_path.exists()
+        assert json.loads(json_path.read_text(encoding="utf-8")) == {"futures": {"沪银": "AG2608"}}
+
+
+class TestConfigMigration:
+    """配置迁移测试。"""
+
+    def test_migrate_skips_when_config_exists(self, tmp_path):
+        """数据库已有配置时跳过迁移。"""
+        from app.config_migration import migrate_configs_to_mysql
+
+        with patch("app.config_migration.get_config_store") as mock_get_store:
+            store = MagicMock()
+            store._config_exists_in_mysql.return_value = True
+            store.storage_manager.check_mysql.return_value = True
+            store.session_maker = MagicMock()
+            mock_get_store.return_value = store
+
+            result = migrate_configs_to_mysql()
+
+            assert result is True
+            store._load_json.assert_not_called()
+
+    def test_migrate_loads_json_and_saves(self, tmp_path):
+        """从 JSON 读取并保存到数据库。"""
+        from app.config_migration import migrate_configs_to_mysql
+
+        symbols_path = tmp_path / "symbols_config.json"
+        symbols_path.write_text(json.dumps({"futures": {"沪银": "AG2608"}}, ensure_ascii=False), encoding="utf-8")
+
+        ai_path = tmp_path / "ai_config.json"
+        ai_path.write_text(json.dumps({"models": []}, ensure_ascii=False), encoding="utf-8")
+
+        with patch("app.config_migration.get_config_store") as mock_get_store:
+            store = MagicMock()
+            store._config_exists_in_mysql.return_value = False
+            store.storage_manager.check_mysql.return_value = True
+            store.session_maker = MagicMock()
+            store.config_dir = tmp_path
+            store.config_files = {
+                "symbols": symbols_path,
+                "ai": ai_path,
+            }
+            store._load_json.side_effect = lambda key, default: json.loads(store.config_files[key].read_text(encoding="utf-8")) if store.config_files[key].exists() else default
+            store._save_to_mysql.return_value = True
+            mock_get_store.return_value = store
+
+            result = migrate_configs_to_mysql()
+
+            assert result is True
+            assert store._save_to_mysql.call_count == 2