""" 交易复盘接口 - 提供交易记录导入、查询、汇总、统计等功能 """ import logging from typing import Optional from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Query from pydantic import BaseModel from sqlalchemy.orm import Session from app.analysis_db import get_analysis_db from app.analysis_models import TradeRecord, TradeImportBatch from app.services.trade_parser import ( parse_settlement_file, save_to_db, calc_daily_summary, calc_variety_summary, calc_overall_statistics, get_trade_pairs, ) from app.services.trade_pairing_engine import ( auto_pair_trades, get_daily_trades_with_pairs, create_manual_pair, delete_pair, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/trade-review", tags=["交易复盘"]) # ==================== 文件导入 ==================== @router.post("/import") async def import_settlement( file: UploadFile = File(...), db: Session = Depends(get_analysis_db), ): """上传并解析期货结算单文件,导入交易记录(含逐条去重校验)""" if not file.filename.endswith(('.xls', '.xlsx')): raise HTTPException(status_code=400, detail="仅支持 .xls/.xlsx 格式的结算单文件") try: content = await file.read() df_futures, df_options = parse_settlement_file(content, file.filename) if df_futures.empty and df_options.empty: return {"success": False, "message": "文件中未找到有效的交易记录(请确认是期货公司导出的结算单)"} result = save_to_db(db, df_futures, df_options, file.filename) total_skipped = result['futures_skipped'] + result['options_skipped'] total_new = result['futures_count'] + result['options_count'] total_parsed = result['futures_parsed'] + result['options_parsed'] if total_new == 0 and total_skipped > 0: return { "success": True, "message": f"全部跳过:共解析 {total_parsed} 条记录,均已存在(去重规则:交易日+品种+时间+价格),未导入任何新数据", "data": result, } msg_parts = [f"导入成功:期货 {result['futures_count']} 条,期权 {result['options_count']} 条"] if total_skipped > 0: msg_parts.append(f"跳过重复 {total_skipped} 条") msg_parts.append(f"交易日期: {result['trade_dates']}") return { "success": True, "message": ",".join(msg_parts), "data": result, } except HTTPException: raise except Exception as e: logger.exception("导入结算单失败") raise HTTPException(status_code=500, detail=f"导入失败: {str(e)}") @router.post("/batch-import") async def batch_import_settlement( files: list[UploadFile] = File(...), db: Session = Depends(get_analysis_db), ): """批量上传并解析期货结算单文件,导入交易记录(含逐条去重校验)""" if not files: raise HTTPException(status_code=400, detail="请选择至少一个结算单文件") results = [] total_futures = 0 total_options = 0 total_skipped_all = 0 processed_count = 0 failed_count = 0 for file in files: if not file.filename.endswith(('.xls', '.xlsx')): results.append({ "filename": file.filename, "success": False, "message": "文件格式不支持,仅支持 .xls/.xlsx", }) failed_count += 1 continue try: content = await file.read() df_futures, df_options = parse_settlement_file(content, file.filename) if df_futures.empty and df_options.empty: results.append({ "filename": file.filename, "success": False, "message": "文件中未找到有效的交易记录", }) failed_count += 1 continue result = save_to_db(db, df_futures, df_options, file.filename) file_new = result['futures_count'] + result['options_count'] file_skipped = result['futures_skipped'] + result['options_skipped'] total_futures += result['futures_count'] total_options += result['options_count'] total_skipped_all += file_skipped processed_count += 1 msg_parts = [] if file_new > 0: msg_parts.append(f"新导入 {file_new} 条(期货 {result['futures_count']},期权 {result['options_count']})") if file_skipped > 0: msg_parts.append(f"跳过重复 {file_skipped} 条") if not msg_parts: msg_parts.append("无有效记录") results.append({ "filename": file.filename, "success": True, "message": ",".join(msg_parts), "new_count": file_new, "skipped_count": file_skipped, "trade_dates": result['trade_dates'], "batch_id": result['batch_id'], }) except Exception as e: logger.exception(f"批量导入文件失败: {file.filename}") results.append({ "filename": file.filename, "success": False, "message": f"导入失败: {str(e)}", }) failed_count += 1 summary = ( f"批量导入完成:共 {len(files)} 个文件," f"处理 {processed_count} 个,失败 {failed_count} 个," f"合计新导入期货 {total_futures} 条、期权 {total_options} 条,跳过重复 {total_skipped_all} 条" ) return { "success": True, "message": summary, "data": { "total_files": len(files), "processed_count": processed_count, "failed_count": failed_count, "total_futures": total_futures, "total_options": total_options, "total_skipped": total_skipped_all, "details": results, }, } # ==================== 交易记录查询 ==================== @router.get("/records") def get_records( start_date: Optional[str] = Query(None, description="开始日期 YYYY-MM-DD"), end_date: Optional[str] = Query(None, description="结束日期 YYYY-MM-DD"), symbol: Optional[str] = Query(None, description="合约代码"), variety: Optional[str] = Query(None, description="品种代码"), trade_type: Optional[str] = Query(None, description="类型: 期货/期权"), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), db: Session = Depends(get_analysis_db), ): """查询交易记录(支持筛选和分页)""" query = db.query(TradeRecord) if start_date: query = query.filter(TradeRecord.trade_date >= start_date) if end_date: query = query.filter(TradeRecord.trade_date <= end_date) if symbol: query = query.filter(TradeRecord.symbol == symbol) if variety: query = query.filter(TradeRecord.variety == variety) if trade_type: query = query.filter(TradeRecord.trade_type == trade_type) total = query.count() records = query.order_by( TradeRecord.trade_date.desc(), TradeRecord.trade_time.desc(), ).offset((page - 1) * page_size).limit(page_size).all() return { "success": True, "data": { "total": total, "page": page, "page_size": page_size, "records": [{ "id": r.id, "trade_type": r.trade_type, "symbol": r.symbol, "variety": r.variety, "symbol_name": r.symbol_name, "direction": r.direction, "offset": r.offset, "price": r.price, "volume": r.volume, "amount": r.amount, "close_pnl": r.close_pnl, "commission": r.commission, "trade_date": r.trade_date, "trade_time": r.trade_time, "import_batch": r.import_batch, "source_file": r.source_file, } for r in records], }, } # ==================== 批次管理 ==================== @router.get("/batches") def get_batches( db: Session = Depends(get_analysis_db), ): """查询导入批次列表""" batches = db.query(TradeImportBatch).order_by( TradeImportBatch.created_at.desc() ).all() return { "success": True, "data": [{ "id": b.id, "batch_id": b.batch_id, "source_file": b.source_file, "futures_count": b.futures_count, "options_count": b.options_count, "trade_dates": b.trade_dates, "created_at": b.created_at.strftime('%Y-%m-%d %H:%M:%S') if b.created_at else '', } for b in batches], } @router.delete("/records/{batch_id}") def delete_batch_records( batch_id: str, db: Session = Depends(get_analysis_db), ): """按批次删除交易记录""" count = db.query(TradeRecord).filter(TradeRecord.import_batch == batch_id).count() if count == 0: raise HTTPException(status_code=404, detail="未找到该批次的交易记录") db.query(TradeRecord).filter(TradeRecord.import_batch == batch_id).delete() db.query(TradeImportBatch).filter(TradeImportBatch.batch_id == batch_id).delete() db.commit() return {"success": True, "message": f"已删除 {count} 条交易记录"} # ==================== 汇总统计 ==================== @router.get("/latest-trade-date") def get_latest_trade_date( db: Session = Depends(get_analysis_db), ): """获取最后一个有交易记录的交易日""" from datetime import date latest = db.query(TradeRecord.trade_date).filter( TradeRecord.trade_date.isnot(None), TradeRecord.trade_date != '', ).distinct().order_by(TradeRecord.trade_date.desc()).first() if latest: return {"success": True, "data": {"trade_date": latest[0]}} else: return {"success": True, "data": {"trade_date": date.today().strftime('%Y-%m-%d')}} @router.delete("/records-by-date/{trade_date}") def delete_records_by_date( trade_date: str, db: Session = Depends(get_analysis_db), ): """删除指定交易日的所有交易记录""" count = db.query(TradeRecord).filter(TradeRecord.trade_date == trade_date).count() if count == 0: raise HTTPException(status_code=404, detail=f"未找到 {trade_date} 的交易记录") db.query(TradeRecord).filter(TradeRecord.trade_date == trade_date).delete() db.commit() return {"success": True, "message": f"已删除 {trade_date} 的 {count} 条交易记录"} @router.get("/daily-summary") def get_daily_summary( start_date: Optional[str] = Query(None), end_date: Optional[str] = Query(None), db: Session = Depends(get_analysis_db), ): """获取每日交易盈亏汇总""" data = calc_daily_summary(db, start_date, end_date) return {"success": True, "data": data} @router.get("/variety-summary") def get_variety_summary( start_date: Optional[str] = Query(None), end_date: Optional[str] = Query(None), db: Session = Depends(get_analysis_db), ): """获取品种交易盈亏汇总""" data = calc_variety_summary(db, start_date, end_date) return {"success": True, "data": data} @router.get("/statistics") def get_statistics( start_date: Optional[str] = Query(None), end_date: Optional[str] = Query(None), db: Session = Depends(get_analysis_db), ): """获取整体交易统计""" data = calc_overall_statistics(db, start_date, end_date) return {"success": True, "data": data} @router.get("/trade-pairs") def get_trade_pairs_api( start_date: Optional[str] = Query(None), end_date: Optional[str] = Query(None), symbol: Optional[str] = Query(None), db: Session = Depends(get_analysis_db), ): """获取开平仓配对后的逐笔交易""" pairs = get_trade_pairs(db, start_date, end_date) if symbol: pairs = [p for p in pairs if p["symbol"] == symbol] return {"success": True, "data": pairs} # ==================== K线 + 交易标记 ==================== @router.get("/kline-with-trades/{symbol}") def get_kline_with_trades( symbol: str, period: str = Query("daily", description="K线周期: daily/60min/15min/5min"), start_date: Optional[str] = Query(None), end_date: Optional[str] = Query(None), db: Session = Depends(get_analysis_db), ): """获取K线数据 + 该品种的交易标记(买卖点)。symbol 可以是完整合约代码(AG2608)或品种代码(AG)""" import json import re from app.database import SessionLocal as MainSessionLocal from app.models import MarketData logger.info(f"[K线查询] 请求参数: symbol={symbol}, period={period}, start_date={start_date}, end_date={end_date}") # 判断是品种代码还是完整合约代码 variety_code = symbol.upper() if re.match(r'^[A-Za-z]+\d', symbol): # 完整合约代码如 AG2608 variety_code = re.match(r'^([A-Za-z]+)', symbol).group(1).upper() kline_symbol = symbol else: # 品种代码如 AG,从配置中查找当前合约 kline_symbol = _resolve_symbol_from_config(variety_code) logger.info(f"[K线查询] 解析结果: variety_code={variety_code}, kline_symbol={kline_symbol}") main_db = MainSessionLocal() try: # 尝试精确匹配(先尝试原始symbol,再尝试小写) market_data = main_db.query(MarketData).filter( MarketData.symbol == kline_symbol, MarketData.period == period, ).first() logger.info(f"[K线查询] 精确匹配查询: symbol={kline_symbol}, period={period}, found={market_data is not None}") if not market_data or not market_data.candles_json: # 尝试小写匹配 market_data = main_db.query(MarketData).filter( MarketData.symbol == kline_symbol.lower(), MarketData.period == period, ).first() logger.info(f"[K线查询] 小写匹配查询: symbol={kline_symbol.lower()}, period={period}, found={market_data is not None}") if not market_data or not market_data.candles_json: # 尝试模糊匹配:查找以该品种代码开头的合约(不区分大小写) market_data = main_db.query(MarketData).filter( MarketData.symbol.like(f'{variety_code}%'), MarketData.period == period, ).first() logger.info(f"[K线查询] 模糊匹配查询: like '{variety_code}%', period={period}, found={market_data is not None}") if market_data: logger.info(f"[K线查询] 模糊匹配找到: symbol={market_data.symbol}") if not market_data or not market_data.candles_json: # 尝试小写模糊匹配 market_data = main_db.query(MarketData).filter( MarketData.symbol.like(f'{variety_code.lower()}%'), MarketData.period == period, ).first() logger.info(f"[K线查询] 小写模糊匹配查询: like '{variety_code.lower()}%', period={period}, found={market_data is not None}") if market_data: logger.info(f"[K线查询] 小写模糊匹配找到: symbol={market_data.symbol}") if not market_data or not market_data.candles_json: logger.warning(f"[K线查询] 未找到K线数据: {kline_symbol} {period}") return {"success": False, "message": f"未找到 {kline_symbol} 的 {period} K线数据"} candles = json.loads(market_data.candles_json) # 将字典格式K线转换为前端期望的数组格式 [date, open, close, low, high, volume] if candles and isinstance(candles[0], dict): # 根据周期决定是否保留时间部分 is_daily = period == 'daily' candles = [ [ c.get("time", "")[:10] if is_daily else c.get("time", ""), # 日线只保留日期,分钟线保留完整时间 c.get("open", 0), c.get("close", 0), c.get("low", 0), c.get("high", 0), c.get("volume", 0), ] for c in candles ] kline_symbol = market_data.symbol logger.info(f"[K线查询] 成功获取K线数据: symbol={kline_symbol}, candles_count={len(candles)}") finally: main_db.close() # 获取该品种的交易记录 logger.info(f"[K线查询] 开始查询交易记录: variety={variety_code}, start_date={start_date}, end_date={end_date}") query = db.query(TradeRecord).filter(TradeRecord.variety == variety_code) if start_date: query = query.filter(TradeRecord.trade_date >= start_date) if end_date: query = query.filter(TradeRecord.trade_date <= end_date) trades = query.order_by(TradeRecord.trade_date, TradeRecord.trade_time).all() logger.info(f"[K线查询] 交易记录查询完成: 找到 {len(trades)} 条记录") if len(trades) > 0: logger.info(f"[K线查询] 交易记录示例: {[{'symbol': t.symbol, 'variety': t.variety, 'date': t.trade_date, 'direction': t.direction} for t in trades[:3]]}") trade_markers = [] for t in trades: trade_markers.append({ "date": t.trade_date, "time": t.trade_time, "symbol": t.symbol, "direction": t.direction, "offset": t.offset, "price": t.price, "volume": t.volume, "close_pnl": t.close_pnl, "commission": t.commission, }) logger.info(f"[K线查询] 返回数据: symbol={kline_symbol}, period={period}, candles={len(candles)}, markers={len(trade_markers)}") return { "success": True, "data": { "symbol": kline_symbol, "period": period, "candles": candles, "trade_markers": trade_markers, }, } def _resolve_symbol_from_config(variety_code: str) -> str: """从品种配置中查找品种代码对应的当前合约""" 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 return variety_code # ==================== AI 逐笔交易分析 ==================== class AnalyzeTradeRequest(BaseModel): symbol: str open_date: str open_time: Optional[str] = None close_date: Optional[str] = None close_time: Optional[str] = None direction: str # 多/空 open_price: Optional[float] = None close_price: Optional[float] = None @router.post("/analyze-trade") async def analyze_trade( req: AnalyzeTradeRequest, db: Session = Depends(get_analysis_db), ): """AI 分析单笔交易(结合多周期K线数据)""" # 收集多周期K线数据 from app.database import SessionLocal as MainSessionLocal from app.models import MarketData import json periods = ["daily", "60min", "15min", "5min"] kline_context = {} main_db = MainSessionLocal() try: for period in periods: market_data = main_db.query(MarketData).filter( MarketData.symbol == req.symbol, MarketData.period == period, ).first() if market_data and market_data.candles_json: candles = json.loads(market_data.candles_json) # 将字典格式K线转换为数组格式 [date, open, close, low, high, volume] if candles and isinstance(candles[0], dict): candles = [ [c.get("time", "")[:10], c.get("open", 0), c.get("close", 0), c.get("low", 0), c.get("high", 0), c.get("volume", 0)] for c in candles ] kline_context[period] = candles[-50:] if len(candles) > 50 else candles finally: main_db.close() if not kline_context: return {"success": False, "message": f"未找到 {req.symbol} 的K线数据,无法分析"} # 构建分析提示 prompt = f"""请分析以下期货交易交易的优缺点: 交易信息: - 品种:{req.symbol} - 方向:{req.direction} - 开仓时间:{req.open_date} {req.open_time or ''} - 开仓价格:{req.open_price or '未知'} - 平仓时间:{req.close_date or '未知'} {req.close_time or ''} - 平仓价格:{req.close_price or '未知'} 各周期K线数据(格式:[日期, 开盘, 收盘, 最低, 最高, 成交量]): """ for period, candles in kline_context.items(): prompt += f"\n{period} 周期(最近{len(candles)}根K线):\n" for c in candles[-10:]: prompt += f" {c[0]}: 开{c[1]} 收{c[2]} 低{c[3]} 高{c[4]} 量{c[5]}\n" prompt += """ 请从以下维度分析这笔交易: 1. 入场时机分析:入场时各周期的趋势如何,入场点是否合理 2. 出场时机分析:出场时的市场状态,是否过早/过晚出场 3. 持仓期间分析:持仓期间市场走势,是否有更好的操作机会 4. 综合评价:这笔交易的主要优点和不足之处 5. 改进建议:类似行情下如何优化操作 """ try: # 使用 AIFuturesAnalyzer 的模型配置和调用能力 from app.services.ai_analysis import AIFuturesAnalyzer analyzer = AIFuturesAnalyzer(db) model = analyzer.get_active_model() if not model: return {"success": False, "message": "未配置AI模型或模型未激活,请先在AI配置页面设置"} response = analyzer.call_ai_model(prompt, model) if not response: return {"success": False, "message": "AI模型返回空响应,请稍后重试"} return {"success": True, "data": {"analysis": response, "symbol": req.symbol}} except Exception as e: logger.exception("AI分析交易失败") return {"success": False, "message": f"AI分析失败: {str(e)}"} # ==================== 交易反思系统 API ==================== from app.analysis_models import ( TradePair, TradeReflection, DailyReflection, TradeTag, TradeRecordTag, TradeExperience, ) from pydantic import BaseModel # --- 按天交易视图 --- @router.get("/daily-trades/{trade_date}") def api_get_daily_trades( trade_date: str, db: Session = Depends(get_analysis_db), ): """获取指定日期的交易+配对+反思状态""" # 先执行自动配对 auto_pair_trades(db, trade_date) db.commit() data = get_daily_trades_with_pairs(db, trade_date) return {"success": True, "data": data} # --- 手动配对 --- class ManualPairRequest(BaseModel): open_record_ids: list[int] close_record_ids: list[int] @router.post("/trade-pairs") def api_create_manual_pair( req: ManualPairRequest, db: Session = Depends(get_analysis_db), ): """手动创建交易配对""" try: pair = create_manual_pair(db, req.open_record_ids, req.close_record_ids) db.commit() return {"success": True, "data": {"id": pair.id}} except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: db.rollback() logger.exception("创建配对失败") raise HTTPException(status_code=500, detail=f"创建配对失败: {str(e)}") @router.delete("/trade-pairs/{pair_id}") def api_delete_pair( pair_id: int, db: Session = Depends(get_analysis_db), ): """删除交易配对""" success = delete_pair(db, pair_id) if not success: raise HTTPException(status_code=404, detail="配对不存在") db.commit() return {"success": True, "message": "配对已删除"} # --- 反思 CRUD --- class ReflectionRequest(BaseModel): trade_pair_id: int trade_date: str entry_reason: Optional[str] = None entry_timing: Optional[str] = None position_management: Optional[str] = None exit_reason: Optional[str] = None exit_timing: Optional[str] = None discipline_score: Optional[int] = None free_reflection: Optional[str] = None @router.post("/reflections") def api_create_reflection( req: ReflectionRequest, db: Session = Depends(get_analysis_db), ): """创建交易反思""" # 检查是否已存在 existing = db.query(TradeReflection).filter( TradeReflection.trade_pair_id == req.trade_pair_id ).first() if existing: raise HTTPException(status_code=400, detail="该交易已有反思,请使用更新接口") reflection = TradeReflection( trade_pair_id=req.trade_pair_id, trade_date=req.trade_date, entry_reason=req.entry_reason, entry_timing=req.entry_timing, position_management=req.position_management, exit_reason=req.exit_reason, exit_timing=req.exit_timing, discipline_score=req.discipline_score, free_reflection=req.free_reflection, ) db.add(reflection) db.commit() return {"success": True, "data": {"id": reflection.id}} @router.put("/reflections/{reflection_id}") def api_update_reflection( reflection_id: int, req: ReflectionRequest, db: Session = Depends(get_analysis_db), ): """更新交易反思""" reflection = db.query(TradeReflection).filter(TradeReflection.id == reflection_id).first() if not reflection: raise HTTPException(status_code=404, detail="反思不存在") # 更新字段 for field in ['entry_reason', 'entry_timing', 'position_management', 'exit_reason', 'exit_timing', 'discipline_score', 'free_reflection']: val = getattr(req, field) if val is not None: setattr(reflection, field, val) # 重新保存时重置 AI 分析状态(因为内容变了) reflection.ai_analyzed = False reflection.ai_version = None db.commit() return {"success": True, "message": "反思已更新"} @router.get("/reflections") def api_get_reflections( trade_pair_id: Optional[int] = Query(None), trade_date: Optional[str] = Query(None), db: Session = Depends(get_analysis_db), ): """查询反思列表""" query = db.query(TradeReflection) if trade_pair_id: query = query.filter(TradeReflection.trade_pair_id == trade_pair_id) if trade_date: query = query.filter(TradeReflection.trade_date == trade_date) reflections = query.order_by(TradeReflection.created_at.desc()).all() return {"success": True, "data": [{ "id": r.id, "trade_pair_id": r.trade_pair_id, "trade_date": r.trade_date, "entry_reason": r.entry_reason, "entry_timing": r.entry_timing, "position_management": r.position_management, "exit_reason": r.exit_reason, "exit_timing": r.exit_timing, "discipline_score": r.discipline_score, "free_reflection": r.free_reflection, "ai_analyzed": r.ai_analyzed, "ai_version": r.ai_version, "created_at": r.created_at.strftime('%Y-%m-%d %H:%M:%S') if r.created_at else None, "updated_at": r.updated_at.strftime('%Y-%m-%d %H:%M:%S') if r.updated_at else None, } for r in reflections]} # --- 当日反思 CRUD --- class DailyReflectionRequest(BaseModel): reflection_date: str emotion_state: Optional[str] = None market_judgment: Optional[str] = None discipline_score: Optional[int] = None overall_rating: Optional[int] = None summary: Optional[str] = None improvements: Optional[str] = None @router.post("/daily-reflections") def api_create_daily_reflection( req: DailyReflectionRequest, db: Session = Depends(get_analysis_db), ): """创建或更新当日反思""" existing = db.query(DailyReflection).filter( DailyReflection.reflection_date == req.reflection_date ).first() if existing: # 更新 for field in ['emotion_state', 'market_judgment', 'discipline_score', 'overall_rating', 'summary', 'improvements']: val = getattr(req, field) if val is not None: setattr(existing, field, val) db.commit() return {"success": True, "data": {"id": existing.id, "action": "updated"}} else: # 创建 dr = DailyReflection( reflection_date=req.reflection_date, emotion_state=req.emotion_state, market_judgment=req.market_judgment, discipline_score=req.discipline_score, overall_rating=req.overall_rating, summary=req.summary, improvements=req.improvements, ) db.add(dr) db.commit() return {"success": True, "data": {"id": dr.id, "action": "created"}} @router.get("/daily-reflections/{reflection_date}") def api_get_daily_reflection( reflection_date: str, db: Session = Depends(get_analysis_db), ): """获取当日反思""" dr = db.query(DailyReflection).filter( DailyReflection.reflection_date == reflection_date ).first() if not dr: return {"success": True, "data": None} return {"success": True, "data": { "id": dr.id, "reflection_date": dr.reflection_date, "emotion_state": dr.emotion_state, "market_judgment": dr.market_judgment, "discipline_score": dr.discipline_score, "overall_rating": dr.overall_rating, "summary": dr.summary, "improvements": dr.improvements, }} # --- 标签系统 --- class TagRequest(BaseModel): name: str category: str = "custom" @router.get("/tags") def api_get_tags( category: Optional[str] = Query(None), db: Session = Depends(get_analysis_db), ): """获取所有标签""" query = db.query(TradeTag) if category: query = query.filter(TradeTag.category == category) tags = query.order_by(TradeTag.is_preset.desc(), TradeTag.name).all() return {"success": True, "data": [{ "id": t.id, "name": t.name, "category": t.category, "is_preset": t.is_preset, } for t in tags]} @router.post("/tags") def api_create_tag( req: TagRequest, db: Session = Depends(get_analysis_db), ): """创建自定义标签(含去重)""" existing = db.query(TradeTag).filter(TradeTag.name == req.name).first() if existing: return {"success": True, "data": {"id": existing.id, "action": "exists"}} tag = TradeTag(name=req.name, category=req.category, is_preset=False) db.add(tag) db.commit() return {"success": True, "data": {"id": tag.id, "action": "created"}} @router.post("/trade-pairs/{pair_id}/tags") def api_add_tag_to_trade( pair_id: int, tag_id: int, db: Session = Depends(get_analysis_db), ): """给交易打标签""" # 检查是否已存在 existing = db.query(TradeRecordTag).filter( TradeRecordTag.trade_pair_id == pair_id, TradeRecordTag.tag_id == tag_id, ).first() if existing: return {"success": True, "message": "标签已存在"} relation = TradeRecordTag(trade_pair_id=pair_id, tag_id=tag_id) db.add(relation) db.commit() return {"success": True, "message": "标签已添加"} @router.delete("/trade-pairs/{pair_id}/tags/{tag_id}") def api_remove_tag_from_trade( pair_id: int, tag_id: int, db: Session = Depends(get_analysis_db), ): """移除交易标签""" db.query(TradeRecordTag).filter( TradeRecordTag.trade_pair_id == pair_id, TradeRecordTag.tag_id == tag_id, ).delete() db.commit() return {"success": True, "message": "标签已移除"} @router.get("/trade-pairs/{pair_id}/tags") def api_get_trade_tags( pair_id: int, db: Session = Depends(get_analysis_db), ): """获取交易的标签列表""" relations = db.query(TradeRecordTag).filter( TradeRecordTag.trade_pair_id == pair_id ).all() tag_ids = [r.tag_id for r in relations] tags = db.query(TradeTag).filter(TradeTag.id.in_(tag_ids)).all() return {"success": True, "data": [{ "id": t.id, "name": t.name, "category": t.category, } for t in tags]} # --- 经验库 --- class ExperienceRequest(BaseModel): title: str content: str exp_type: str # lesson/tip/warning source_pair_id: Optional[int] = None source_date: Optional[str] = None tags: Optional[list[str]] = None @router.get("/experiences") def api_get_experiences( exp_type: Optional[str] = Query(None), tag: Optional[str] = Query(None), keyword: Optional[str] = Query(None), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), db: Session = Depends(get_analysis_db), ): """获取经验列表(支持筛选和分页)""" query = db.query(TradeExperience) if exp_type: query = query.filter(TradeExperience.exp_type == exp_type) if tag: query = query.filter(TradeExperience.tags.contains([tag])) if keyword: query = query.filter( (TradeExperience.title.contains(keyword)) | (TradeExperience.content.contains(keyword)) ) total = query.count() experiences = query.order_by( TradeExperience.created_at.desc() ).offset((page - 1) * page_size).limit(page_size).all() return {"success": True, "data": { "total": total, "page": page, "page_size": page_size, "items": [{ "id": e.id, "title": e.title, "content": e.content, "exp_type": e.exp_type, "source_pair_id": e.source_pair_id, "source_date": e.source_date, "tags": e.tags or [], "created_at": e.created_at.strftime('%Y-%m-%d %H:%M:%S') if e.created_at else None, } for e in experiences], }} @router.post("/experiences") def api_create_experience( req: ExperienceRequest, db: Session = Depends(get_analysis_db), ): """创建经验""" exp = TradeExperience( title=req.title, content=req.content, exp_type=req.exp_type, source_pair_id=req.source_pair_id, source_date=req.source_date, tags=req.tags or [], ) db.add(exp) db.commit() return {"success": True, "data": {"id": exp.id}} @router.put("/experiences/{exp_id}") def api_update_experience( exp_id: int, req: ExperienceRequest, db: Session = Depends(get_analysis_db), ): """更新经验""" exp = db.query(TradeExperience).filter(TradeExperience.id == exp_id).first() if not exp: raise HTTPException(status_code=404, detail="经验不存在") for field in ['title', 'content', 'exp_type', 'source_pair_id', 'source_date', 'tags']: val = getattr(req, field) if val is not None: setattr(exp, field, val) db.commit() return {"success": True, "message": "经验已更新"} @router.delete("/experiences/{exp_id}") def api_delete_experience( exp_id: int, db: Session = Depends(get_analysis_db), ): """删除经验""" exp = db.query(TradeExperience).filter(TradeExperience.id == exp_id).first() if not exp: raise HTTPException(status_code=404, detail="经验不存在") db.delete(exp) db.commit() return {"success": True, "message": "经验已删除"}