""" 配置管理接口 - 品种配置文件上传、批量获取、批量任务创建 """ import json import logging import shutil from pathlib import Path from typing import Optional from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Body from fastapi.responses import JSONResponse from sqlalchemy.orm import Session from pydantic import BaseModel from app.database import get_db 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 logger = logging.getLogger(__name__) router = APIRouter(prefix="/config", tags=["品种配置"]) class BatchFetchRequest(BaseModel): """批量获取请求体""" periods: Optional[str] = None 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) @router.post("/upload") def upload_config( file: Optional[UploadFile] = File(None), json_config: Optional[dict] = Body(None, embed=False), ): """ 上传品种配置文件（JSON格式）。 格式示例: { "futures": {"沪银": "AG2606", "沪金": "AU2606"}, "stock": {"平安银行": "000001"} } """ _ensure_config_dir() try: if file: content = file.file.read() data = json.loads(content) elif json_config: data = json_config else: raise HTTPException(status_code=400, detail="请提供配置文件或JSON数据") 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) futures_count = len(data.get("futures", {})) stock_count = len(data.get("stock", {})) return { "message": "配置文件上传成功", "futures_symbols": futures_count, "stock_symbols": stock_count, "symbols": data, } except json.JSONDecodeError: raise HTTPException(status_code=400, detail="无效的 JSON 格式") @router.post("/batch-fetch-all") def batch_fetch_all( request: BatchFetchRequest, db: Session = Depends(get_db), ): """ 根据配置文件批量获取所有品种数据。 智能缓存:已存在且有效的数据不重复请求。 """ 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) symbols_dict = config.get(data_type, {}) if not symbols_dict: raise HTTPException(status_code=400, detail=f"配置中没有 {data_type} 类型的品种") # 如果指定了selected_symbols，只获取这些合约 if selected_symbols: # 解析逗号分隔的合约代码 symbol_list = [s.strip() for s in selected_symbols.split(",") if s.strip()] symbols_dict = {name: code for name, code in symbols_dict.items() if code in symbol_list} if not symbols_dict: raise HTTPException(status_code=400, detail="选定的合约不在配置中") period_list = [p.strip() for p in periods.split(",")] if periods else ["5min", "15min", "30min", "60min", "daily"] results = { "total": len(symbols_dict), "success": [], "failed": [], "cached": [], # 命中缓存的 "details": {}, } for name, symbol in symbols_dict.items(): logger.info(f"处理品种: {name} ({symbol})") # 检查缓存 status = check_cache_status(db, symbol, data_type, period_list) if status["all_valid"]: results["cached"].append({"name": name, "symbol": symbol}) cached = get_cached_data(db, symbol, data_type, period_list) timeframes = [] for p, candles in cached["timeframes"].items(): # 转换数据格式: time -> datetime normalized_candles = [] for c in candles: candle_dict = dict(c) if 'time' in candle_dict and 'datetime' not in candle_dict: candle_dict['datetime'] = candle_dict.pop('time') normalized_candles.append(candle_dict) timeframes.append(TimeframeData( period=p, candles=[CandleItem(**c) for c in normalized_candles], candle_count=len(normalized_candles), fetched_at=cached.get("timestamp", ""), )) results["details"][symbol] = SymbolDataResponse( symbol=symbol, data_type=data_type, current_price=cached.get("current_price"), timeframes=timeframes, source="cache", ) results["success"].append({"name": name, "symbol": symbol}) continue # 需要采集 need_fetch = status["missing_periods"] logger.info(f"需要采集的周期: {need_fetch}") result = fetch_symbol_data(symbol, data_type, need_fetch) if result.get("timeframes"): logger.info(f"采集到 {len(result['timeframes'])} 个周期的数据,开始保存") save_market_data(db, symbol, result) # 合并缓存和新数据 all_timeframes = {} if status["valid_periods"]: existing = get_cached_data(db, symbol, data_type, status["valid_periods"]) if existing: all_timeframes.update(existing["timeframes"]) all_timeframes.update(result["timeframes"]) timeframes = [] for p in period_list: candles = all_timeframes.get(p, []) if candles: # 转换数据格式: time -> datetime normalized_candles = [] for c in candles: candle_dict = dict(c) if 'time' in candle_dict and 'datetime' not in candle_dict: candle_dict['datetime'] = candle_dict.pop('time') normalized_candles.append(candle_dict) timeframes.append(TimeframeData( period=p, candles=[CandleItem(**c) for c in normalized_candles], candle_count=len(normalized_candles), fetched_at=result.get("timestamp", ""), )) source = "live+cache" if status["valid_periods"] else "live" results["details"][symbol] = SymbolDataResponse( symbol=symbol, data_type=data_type, current_price=result.get("current_price"), timeframes=timeframes, source=source, ) results["success"].append({"name": name, "symbol": symbol}) logger.info(f"采集成功: {symbol}") else: error_msg = result.get("error", "未知错误") logger.error(f"采集失败: {symbol}, 错误: {error_msg}") results["failed"].append({ "name": name, "symbol": symbol, "error": error_msg, }) return results @router.post("/batch-tasks") def batch_create_tasks( periods: Optional[str] = None, interval_seconds: int = 300, data_type: str = "futures", db: Session = Depends(get_db), ): """ 根据配置文件为所有品种批量创建定时任务。 """ _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) symbols_dict = config.get(data_type, {}) if not symbols_dict: raise HTTPException(status_code=400, detail=f"配置中没有 {data_type} 类型的品种") period_list = [p.strip() for p in periods.split(",")] if periods else ["5min", "15min", "30min", "60min", "daily"] results = {"total": len(symbols_dict), "created": [], "failed": []} for name, symbol in symbols_dict.items(): try: task = create_task( db=db, symbol=symbol, data_type=data_type, periods=period_list, interval_seconds=interval_seconds, ) job_id = add_job(task.id, task.interval_seconds) task.job_id = job_id db.commit() db.refresh(task) results["created"].append({ "name": name, "symbol": symbol, "task_id": task.id, "job_id": job_id, "interval": interval_seconds, }) except Exception as e: results["failed"].append({ "name": name, "symbol": symbol, "error": str(e), }) return results