diff --git a/app/api/ai_config.py b/app/api/ai_config.py
index 4235e32..b0b8e18 100644
--- a/app/api/ai_config.py
+++ b/app/api/ai_config.py
@@ -9,6 +9,8 @@ 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 +40,26 @@ class SaveAIConfigRequest(BaseModel):
     analysis_settings: Optional[dict] = None
 
 
-CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config"
-AI_CONFIG_FILE = CONFIG_DIR / "ai_config.json"
-
-
-def _ensure_config_dir():
-    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
-
-
-def _load_ai_config() -> dict:
-    """加载AI配置"""
-    _ensure_config_dir()
-    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
-            }
+def _default_ai_config() -> dict:
+    """AI 配置默认值"""
+    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 +75,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}")
diff --git a/app/api/config.py b/app/api/config.py
index 27dec30..e854653 100644
--- a/app/api/config.py
+++ b/app/api/config.py
@@ -17,6 +17,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 +29,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()
-    if not CONFIG_FILE.exists():
-        return {"futures": {}, "stock": {}}
-    with open(CONFIG_FILE, "r", encoding="utf-8") as f:
-        return json.load(f)
+    return get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
 
 
 @router.post("/upload")
@@ -60,8 +49,6 @@ async def upload_config(
         "stock": {"平安银行": "000001"}
     }
     """
-    _ensure_config_dir()
-    
     try:
         if file:
             content = await file.read()
@@ -72,12 +59,11 @@ async def upload_config(
             if not body:
                 raise HTTPException(status_code=400, detail="请提供配置文件或JSON数据")
             data = json.loads(body)
-            
+
         if not isinstance(data, dict):
             raise HTTPException(status_code=400, detail="配置文件必须是 JSON 对象")
 
-        with open(CONFIG_FILE, "w", encoding="utf-8") as f:
-            json.dump(data, f, ensure_ascii=False, indent=4)
+        get_config_store().set_config("symbols", data)
 
         futures_count = len(data.get("futures", {}))
         stock_count = len(data.get("stock", {}))
@@ -104,13 +90,7 @@ def batch_fetch_all(
     periods = request.periods
     data_type = request.data_type
     selected_symbols = request.selected_symbols
-    _ensure_config_dir()
-    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)
-
+    config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
     symbols_dict = config.get(data_type, {})
     if not symbols_dict:
         raise HTTPException(status_code=400, detail=f"配置中没有 {data_type} 类型的品种")
@@ -236,13 +216,7 @@ def batch_create_tasks(
     """
     根据配置文件为所有品种批量创建定时任务。
     """
-    _ensure_config_dir()
-    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)
-
+    config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
     symbols_dict = config.get(data_type, {})
     if not symbols_dict:
         raise HTTPException(status_code=400, detail=f"配置中没有 {data_type} 类型的品种")
diff --git a/app/api/futures_analysis.py b/app/api/futures_analysis.py
index b1825de..b3015f1 100644
--- a/app/api/futures_analysis.py
+++ b/app/api/futures_analysis.py
@@ -15,25 +15,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 {"futures": {}, "stock": {}}
-    with open(SYMBOLS_CONFIG_FILE, "r", encoding="utf-8") as f:
-        return json.load(f)
+    return get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
 
 
 @router.get("/list")
 def get_futures_list(db: Session = Depends(get_db)):
-    """获取所有期货品种列表及摘要信息（从symbols_config.json读取）"""
+    """获取所有期货品种列表及摘要信息"""
     config = _load_symbols_config()
     futures_config = config.get("futures", {})
     
diff --git a/app/api/trade_review.py b/app/api/trade_review.py
index aef9c13..a5a9917 100644
--- a/app/api/trade_review.py
+++ b/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 pathlib import Path
-    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)
+    from app.config_store import get_config_store
+    config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
     for name, contract in config.get("futures", {}).items():
         if contract.upper().startswith(variety_code.upper()):
             return contract
diff --git a/app/config_migration.py b/app/config_migration.py
new file mode 100644
index 0000000..a1bb5f1
--- /dev/null
+++ b/app/config_migration.py
@@ -0,0 +1,49 @@
+"""
+配置迁移 - 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(key, DEFAULT_CONFIGS.get(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
diff --git a/app/config_store.py b/app/config_store.py
new file mode 100644
index 0000000..c32551d
--- /dev/null
+++ b/app/config_store.py
@@ -0,0 +1,152 @@
+"""
+统一配置存储模块
+
+支持 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(key, fallback)
+                self._save_to_mysql(key, json_value)
+                return json_value
+            except Exception as e:
+                logger.warning(f"从 MySQL 读取配置 [{key}] 失败，降级到 JSON: {e}")
+
+        return self._load_json(key, fallback)
+
+    def set_config(self, key: str, value: dict) -> bool:
+        """写入配置，始终写 JSON，MySQL 可用时同时写数据库。"""
+        if key not in CONFIG_FILES:
+            raise ValueError(f"未知配置键: {key}")
+
+        # 始终写入 JSON 文件，保证 fallback 最新
+        self._save_to_json(key, value)
+
+        if self.storage_manager.check_mysql():
+            try:
+                self._save_to_mysql(key, value)
+                return True
+            except Exception as e:
+                logger.warning(f"写入 MySQL 配置 [{key}] 失败，已保存到 JSON: {e}")
+                return False
+
+        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(self, key: str, fallback: Optional[dict] = None) -> dict:
+        """从 JSON 文件读取配置。"""
+        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 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)
+        with open(file_path, "w", encoding="utf-8") as f:
+            json.dump(value, f, ensure_ascii=False, indent=4)
+
+    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
diff --git a/app/main.py b/app/main.py
index 8064799..955fb38 100644
--- a/app/main.py
+++ b/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
diff --git a/app/models.py b/app/models.py
index 553a28f..067aba4 100644
--- a/app/models.py
+++ b/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}>"
diff --git a/app/services/ai_analysis.py b/app/services/ai_analysis.py
index fced968..e99a154 100644
--- a/app/services/ai_analysis.py
+++ b/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,25 +148,25 @@ class AIFuturesAnalyzer:
     def get_active_model(self) -> Optional[Dict]:
         """获取当前激活的AI模型配置"""
         try:
-            if not AI_CONFIG_FILE.exists():
-                return None
-            
-            with open(AI_CONFIG_FILE, "r", encoding="utf-8") as f:
-                config = json.load(f)
-            
+            from app.config_store import get_config_store
+            config = get_config_store().get_config("ai", {
+                "models": [],
+                "active_model": None,
+            })
+
             models = config.get("models", [])
             active_model_name = config.get("active_model")
-            
+
             if active_model_name:
                 for model in models:
                     if model.get("model_name") == active_model_name and model.get("enabled", True):
                         return model
-            
+
             for model in models:
                 if model.get("enabled", True):
                     logger.warning(f"未找到匹配的激活模型，使用第一个启用的模型: {model.get('model_name')}")
                     return model
-            
+
             return None
         except Exception as e:
             logger.error(f"加载AI配置失败: {e}")
diff --git a/app/services/plan_generator.py b/app/services/plan_generator.py
index 55be07b..c939de3 100644
--- a/app/services/plan_generator.py
+++ b/app/services/plan_generator.py
@@ -46,12 +46,8 @@ WEIGHTS = {
 
 def load_symbols_config() -> Dict[str, str]:
     """加载品种配置 {中文名: 合约代码}"""
-    config_path = CONFIG_DIR / "symbols_config.json"
-    if not config_path.exists():
-        logger.warning(f"品种配置文件不存在: {config_path}")
-        return {}
-    with open(config_path, "r", encoding="utf-8") as f:
-        data = json.load(f)
+    from app.config_store import get_config_store
+    data = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
     return data.get("futures", {})
 
 
diff --git a/app/services/trade_parser.py b/app/services/trade_parser.py
index 5bed555..2c32d85 100644
--- a/app/services/trade_parser.py
+++ b/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"
-    if config_path.exists():
-        with open(config_path, "r", encoding="utf-8") as f:
-            config = json.load(f)
-        # 正向: {"沪银": "AG2608"} -> 反向: {"AG": "沪银"}
-        for name, contract in config.get("futures", {}).items():
-            variety = extract_variety(contract)
-            if variety and variety not in _VARIETY_NAME_MAP:
-                _VARIETY_NAME_MAP[variety] = name
+    from app.config_store import get_config_store
+    config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
+    # 正向: {"沪银": "AG2608"} -> 反向: {"AG": "沪银"}
+    for name, contract in config.get("futures", {}).items():
+        variety = extract_variety(contract)
+        if variety and variety not in _VARIETY_NAME_MAP:
+            _VARIETY_NAME_MAP[variety] = name
 
 
 def extract_variety(contract):
diff --git a/docs/superpowers/plans/2026-07-04-config-to-mysql.md b/docs/superpowers/plans/2026-07-04-config-to-mysql.md
new file mode 100644
index 0000000..648f81e
--- /dev/null
+++ b/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. 运行全部测试套件确保无回归
+
+**验收**:
+- 所有测试通过
+- 测试覆盖核心降级场景
diff --git a/openspec/changes/config-to-mysql/.comet.yaml b/openspec/changes/config-to-mysql/.comet.yaml
index cc704b9..41161c0 100644
--- a/openspec/changes/config-to-mysql/.comet.yaml
+++ b/openspec/changes/config-to-mysql/.comet.yaml
@@ -1,17 +1,17 @@
 workflow: full
 phase: build
 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
 branch_status: null
diff --git a/openspec/changes/config-to-mysql/tasks.md b/openspec/changes/config-to-mysql/tasks.md
index de39c05..c588127 100644
--- a/openspec/changes/config-to-mysql/tasks.md
+++ b/openspec/changes/config-to-mysql/tasks.md
@@ -2,36 +2,36 @@
 
 ## 1. 数据模型与配置存储模块
 
-- [ ] 1.1 在 `app/models.py` 中新增 `AppConfig` 模型
-- [ ] 1.2 创建 `app/config_store.py`，实现 `ConfigStore` 类
-- [ ] 1.3 实现 `get_config(key)` 方法，支持 MySQL → JSON 降级
-- [ ] 1.4 实现 `set_config(key, value)` 方法，支持 MySQL + JSON 双写
-- [ ] 1.5 实现 `load_from_json(key, fallback)` 辅助方法
+- [x] 1.1 在 `app/models.py` 中新增 `AppConfig` 模型
+- [x] 1.2 创建 `app/config_store.py`，实现 `ConfigStore` 类
+- [x] 1.3 实现 `get_config(key)` 方法，支持 MySQL → JSON 降级
+- [x] 1.4 实现 `set_config(key, value)` 方法，支持 MySQL + JSON 双写
+- [x] 1.5 实现 `_load_json(key, fallback)` 和 `_save_to_json(key, value)` 辅助方法
 
 ## 2. 数据迁移
 
-- [ ] 2.1 创建 `app/config_migration.py`，实现 `migrate_configs_to_mysql()`
-- [ ] 2.2 在 `app/main.py` lifespan 中创建 `app_config` 表
-- [ ] 2.3 在 `app/main.py` lifespan 中调用迁移函数
-- [ ] 2.4 迁移幂等性：数据库已有配置时跳过
+- [x] 2.1 创建 `app/config_migration.py`，实现 `migrate_configs_to_mysql()`
+- [x] 2.2 在 `app/main.py` lifespan 中创建 `app_config` 表
+- [x] 2.3 在 `app/main.py` lifespan 中调用迁移函数
+- [x] 2.4 迁移幂等性：数据库已有配置时跳过
 
 ## 3. 修改配置 API
 
-- [ ] 3.1 修改 `app/api/config.py` 使用 `ConfigStore` 读写 `symbols` 配置
-- [ ] 3.2 修改 `app/api/ai_config.py` 使用 `ConfigStore` 读写 `ai` 配置
-- [ ] 3.3 保持 API 接口签名不变
+- [x] 3.1 修改 `app/api/config.py` 使用 `ConfigStore` 读写 `symbols` 配置
+- [x] 3.2 修改 `app/api/ai_config.py` 使用 `ConfigStore` 读写 `ai` 配置
+- [x] 3.3 保持 API 接口签名不变
 
 ## 4. 修改业务使用方
 
-- [ ] 4.1 修改 `app/api/futures_analysis.py` 从 `ConfigStore` 读取品种配置
-- [ ] 4.2 修改 `app/api/trade_review.py` 从 `ConfigStore` 读取品种配置
-- [ ] 4.3 修改 `app/services/trade_parser.py` 从 `ConfigStore` 读取品种配置
-- [ ] 4.4 修改 `app/services/plan_generator.py` 从 `ConfigStore` 读取品种配置
-- [ ] 4.5 修改 `app/services/ai_analysis.py` 从 `ConfigStore` 读取 AI 配置
+- [x] 4.1 修改 `app/api/futures_analysis.py` 从 `ConfigStore` 读取品种配置
+- [x] 4.2 修改 `app/api/trade_review.py` 从 `ConfigStore` 读取品种配置
+- [x] 4.3 修改 `app/services/trade_parser.py` 从 `ConfigStore` 读取品种配置
+- [x] 4.4 修改 `app/services/plan_generator.py` 从 `ConfigStore` 读取品种配置
+- [x] 4.5 修改 `app/services/ai_analysis.py` 从 `ConfigStore` 读取 AI 配置
 
 ## 5. 测试与验证
 
-- [ ] 5.1 编写 `tests/test_config_store.py`，覆盖 MySQL 命中/未命中/降级场景
-- [ ] 5.2 编写迁移测试，验证幂等性
-- [ ] 5.3 运行全部测试套件，确保无回归
-- [ ] 5.4 启动应用验证配置读取正常
+- [x] 5.1 编写 `tests/test_config_store.py`，覆盖 MySQL 命中/未命中/降级场景
+- [x] 5.2 编写迁移测试，验证幂等性
+- [x] 5.3 运行全部测试套件，确保无回归（110 passed, 2 xfailed）
+- [x] 5.4 启动应用验证配置读取正常
diff --git a/tests/test_config_store.py b/tests/test_config_store.py
new file mode 100644
index 0000000..c8e414b
--- /dev/null
+++ b/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