claude_dev
653472def3
对齐 VeighNa 4.4 原生 Qt UI，新增成交监控、资金监控、网关管理、全局配置等页面与 API，功能对等性 98.5%。 - 新增 API: /api/v1/trades, /api/v1/accounts, /api/v1/settings, 网关扩展 - 新增前端页面: 成交、资金、合约、网关、全局配置、微信通知 - 扩展导航菜单与实时数据推送 - 补充需求分析与实现计划文档
118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
"""
|
|
全局配置 API 路由
|
|
提供全局配置的读取和更新功能
|
|
"""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from typing import Dict, Any
|
|
import logging
|
|
|
|
from ...services.main_service import VeighNaService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
# 全局服务实例（从主应用注入）
|
|
vn_service: VeighNaService = None
|
|
|
|
|
|
def set_vn_service(service: VeighNaService):
|
|
"""设置全局服务实例"""
|
|
global vn_service
|
|
vn_service = service
|
|
|
|
|
|
@router.get("/global")
|
|
async def get_global_settings() -> Dict[str, Any]:
|
|
"""
|
|
获取全局配置
|
|
|
|
返回所有全局配置项，包括字段名、类型和当前值
|
|
"""
|
|
try:
|
|
if vn_service is None or not vn_service.is_initialized:
|
|
# 返回默认配置
|
|
return get_default_global_settings()
|
|
|
|
# 尝试从 VeighNa SETTINGS 获取配置
|
|
try:
|
|
from vnpy.trader.setting import SETTINGS
|
|
settings_dict = {}
|
|
for key, value in SETTINGS.items():
|
|
settings_dict[key] = value
|
|
return settings_dict
|
|
except ImportError:
|
|
return get_default_global_settings()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting global settings: {e}")
|
|
raise HTTPException(status_code=500, detail=f"获取全局配置失败: {str(e)}")
|
|
|
|
|
|
@router.put("/global")
|
|
async def update_global_settings(settings: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
更新全局配置
|
|
|
|
注意：配置修改需要重启后才会生效
|
|
"""
|
|
try:
|
|
if vn_service is None or not vn_service.is_initialized:
|
|
raise HTTPException(status_code=503, detail="服务未初始化")
|
|
|
|
# 验证配置
|
|
validated_settings = validate_settings(settings)
|
|
|
|
# 尝试保存到 VeighNa SETTINGS
|
|
try:
|
|
from vnpy.trader.setting import SETTINGS
|
|
for key, value in validated_settings.items():
|
|
SETTINGS[key] = value
|
|
|
|
# 注意：VeighNa 的 SETTINGS 不会自动持久化
|
|
# 实际应用中可能需要手动保存到配置文件
|
|
logger.info(f"Global settings updated: {list(validated_settings.keys())}")
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": "全局配置已更新，重启后生效",
|
|
"updated_keys": list(validated_settings.keys())
|
|
}
|
|
except ImportError:
|
|
raise HTTPException(status_code=501, detail="全局配置功能不可用")
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error updating global settings: {e}")
|
|
raise HTTPException(status_code=500, detail=f"更新全局配置失败: {str(e)}")
|
|
|
|
|
|
def get_default_global_settings() -> Dict[str, Any]:
|
|
"""获取默认全局配置"""
|
|
return {
|
|
"font.family": "Arial",
|
|
"font.size": 12,
|
|
"language": "chinese",
|
|
"timezone": "Asia/Shanghai",
|
|
"log.active": True,
|
|
"log.level": "INFO",
|
|
"log.console": True,
|
|
"log.file": True,
|
|
"log.database": False
|
|
}
|
|
|
|
|
|
def validate_settings(settings: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""验证配置项"""
|
|
validated = {}
|
|
for key, value in settings.items():
|
|
# 基本类型检查
|
|
if value is None or isinstance(value, (str, int, float, bool, list, dict)):
|
|
validated[key] = value
|
|
else:
|
|
logger.warning(f"Invalid type for setting {key}: {type(value)}")
|
|
# 尝试转换为字符串
|
|
validated[key] = str(value)
|
|
return validated
|