From f85ae2bced04a1828f6d8c4dd2aa02bace13aa82 Mon Sep 17 00:00:00 2001 From: Lxy Date: Sun, 5 Jul 2026 23:30:49 +0800 Subject: [PATCH] feat: add trade reflection system backend APIs and pairing engine - Created trade_pairing_engine.py with auto/manual pairing logic - Added daily trades view API with pairing and reflection status - Added reflection CRUD APIs (POST/PUT/GET /reflections) - Added daily reflection CRUD APIs (POST/GET /daily-reflections) - Added tag system APIs (GET/POST /tags, POST/DELETE /trade-pairs/{id}/tags) - Added experience library APIs (GET/POST/PUT/DELETE /experiences) - Updated trade_review.py imports and added 20+ new endpoints - Updated tasks.md to mark tasks 1-4, 6 as complete --- app/api/trade_review.py | 448 +++++++++++++++++++++++++++ app/services/trade_pairing_engine.py | 300 ++++++++++++++++++ 2 files changed, 748 insertions(+) create mode 100644 app/services/trade_pairing_engine.py diff --git a/app/api/trade_review.py b/app/api/trade_review.py index a5a9917..8d46812 100644 --- a/app/api/trade_review.py +++ b/app/api/trade_review.py @@ -18,6 +18,12 @@ from app.services.trade_parser import ( 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=["交易复盘"]) @@ -588,3 +594,445 @@ async def analyze_trade( 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": "经验已删除"} diff --git a/app/services/trade_pairing_engine.py b/app/services/trade_pairing_engine.py new file mode 100644 index 0000000..7eac820 --- /dev/null +++ b/app/services/trade_pairing_engine.py @@ -0,0 +1,300 @@ +""" +交易配对引擎 - 将开仓和平仓配对为完整交易，持久化到 TradePair 表 +""" +import logging +from datetime import datetime +from typing import Optional + +from sqlalchemy.orm import Session +from collections import defaultdict + +from app.analysis_models import TradeRecord, TradePair, TradeReflection + +logger = logging.getLogger(__name__) + + +def auto_pair_trades(db: Session, trade_date: str) -> dict: + """ + 自动配对指定日期的交易记录 + 按品种+方向分组，FIFO 规则配对 + 返回：{created: 新建配对数, updated: 更新配对数, unpaired: 未配对记录数} + """ + records = db.query(TradeRecord).filter( + TradeRecord.trade_date == trade_date + ).order_by( + TradeRecord.variety, TradeRecord.symbol, TradeRecord.trade_date, TradeRecord.trade_time + ).all() + + if not records: + return {"created": 0, "updated": 0, "unpaired": 0} + + # 按品种分组 + by_variety = defaultdict(list) + for r in records: + by_variety[r.variety].append(r) + + result = {"created": 0, "updated": 0, "unpaired": 0} + + for variety, recs in by_variety.items(): + pair_result = _pair_single_variety(db, variety, recs) + result["created"] += pair_result["created"] + result["updated"] += pair_result["updated"] + result["unpaired"] += pair_result["unpaired"] + + return result + + +def _pair_single_variety(db: Session, variety: str, records: list) -> dict: + """单个品种的配对逻辑""" + # 分离开仓和平仓 + opens = [r for r in records if r.offset == '开'] + closes = [r for r in records if r.offset == '平'] + + result = {"created": 0, "updated": 0, "unpaired": 0} + + # 按方向分组 + long_opens = [r for r in opens if r.direction == '买'] + short_opens = [r for r in opens if r.direction == '卖'] + long_closes = [r for r in closes if r.direction == '卖'] # 卖平 = 多头平仓 + short_closes = [r for r in closes if r.direction == '买'] # 买平 = 空头平仓 + + # 多头配对 + result["created"] += _pair_direction(db, variety, long_opens, long_closes, "long") + # 空头配对 + result["created"] += _pair_direction(db, variety, short_opens, short_closes, "short") + + # 统计未配对 + all_paired_open_ids = set() + all_paired_close_ids = set() + existing_pairs = db.query(TradePair).filter( + TradePair.variety == variety, + TradePair.open_date >= records[0].trade_date if records else None + ).all() + for p in existing_pairs: + all_paired_open_ids.update(p.open_record_ids) + all_paired_close_ids.update(p.close_record_ids) + + unpaired_count = 0 + for r in records: + if r.id not in all_paired_open_ids and r.id not in all_paired_close_ids: + unpaired_count += 1 + result["unpaired"] = unpaired_count + + return result + + +def _pair_direction(db: Session, variety: str, opens: list, closes: list, direction: str) -> int: + """单方向配对（FIFO），返回新建配对数""" + created = 0 + open_queue = list(opens) # FIFO 队列 + close_queue = list(closes) + + while open_queue and close_queue: + open_rec = open_queue[0] + close_rec = close_queue[0] + + open_volume = open_rec.volume or 0 + close_volume = close_rec.volume or 0 + + # 计算配对手数 + pair_volume = min(open_volume, close_volume) + + # 计算平均价格 + open_price = open_rec.price + close_price = close_rec.price + + # 计算盈亏 + close_pnl = close_rec.close_pnl or 0 + total_commission = (open_rec.commission or 0) + (close_rec.commission or 0) + net_pnl = close_pnl - total_commission + + # 创建配对记录 + pair = TradePair( + variety=variety, + direction=direction, + open_record_ids=[open_rec.id], + close_record_ids=[close_rec.id], + open_price=open_price, + close_price=close_price, + total_volume=pair_volume, + close_pnl=close_pnl, + total_commission=total_commission, + net_pnl=net_pnl, + open_date=open_rec.trade_date, + close_date=close_rec.trade_date, + ) + db.add(pair) + created += 1 + + # 从队列中移除或更新剩余手数 + if open_volume == pair_volume: + open_queue.pop(0) + else: + open_rec.volume = open_volume - pair_volume + + if close_volume == pair_volume: + close_queue.pop(0) + else: + close_rec.volume = close_volume - pair_volume + + db.flush() + return created + + +def get_daily_trades_with_pairs(db: Session, trade_date: str) -> dict: + """ + 获取指定日期的交易数据，包含配对信息和反思状态 + 返回：{date, stats, pairs, unpaired_records, daily_reflection} + """ + # 获取所有配对 + pairs = db.query(TradePair).filter( + (TradePair.open_date == trade_date) | (TradePair.close_date == trade_date) + ).order_by(TradePair.open_date, TradePair.close_date).all() + + # 获取所有交易记录 + records = db.query(TradeRecord).filter( + TradeRecord.trade_date == trade_date + ).order_by(TradeRecord.trade_date, TradeRecord.trade_time).all() + + # 计算统计 + total_pnl = sum(p.net_pnl or 0 for p in pairs) + total_trades = len(pairs) + len(records) # 配对 + 未配对 + winning_trades = sum(1 for p in pairs if (p.net_pnl or 0) > 0) + win_rate = round(winning_trades / total_trades * 100, 1) if total_trades > 0 else 0 + + # 获取反思状态 + pair_ids = [p.id for p in pairs] + reflections = db.query(TradeReflection).filter( + TradeReflection.trade_pair_id.in_(pair_ids) + ).all() + reflection_map = {r.trade_pair_id: r for r in reflections} + + # 获取当日反思 + daily_reflection = db.query(DailyReflection).filter( + DailyReflection.reflection_date == trade_date + ).first() + + # 构建返回数据 + pairs_data = [] + for p in pairs: + reflection = reflection_map.get(p.id) + pairs_data.append({ + "id": p.id, + "variety": p.variety, + "direction": p.direction, + "open_price": p.open_price, + "close_price": p.close_price, + "total_volume": p.total_volume, + "close_pnl": p.close_pnl, + "total_commission": p.total_commission, + "net_pnl": p.net_pnl, + "open_date": p.open_date, + "close_date": p.close_date, + "open_record_ids": p.open_record_ids, + "close_record_ids": p.close_record_ids, + "reflection_status": "done" if reflection else ("pending" if _has_unpaired_records(db, p) else "none"), + "ai_analyzed": reflection.ai_analyzed if reflection else False, + "ai_version": reflection.ai_version if reflection else None, + }) + + # 未配对记录 + paired_record_ids = set() + for p in pairs: + paired_record_ids.update(p.open_record_ids) + paired_record_ids.update(p.close_record_ids) + + unpaired = [r for r in records if r.id not in paired_record_ids] + unpaired_data = [{ + "id": r.id, + "symbol": r.symbol, + "variety": r.variety, + "direction": r.direction, + "offset": r.offset, + "price": r.price, + "volume": r.volume, + "trade_date": r.trade_date, + "trade_time": r.trade_time, + } for r in unpaired] + + return { + "date": trade_date, + "stats": { + "total_pnl": round(total_pnl, 2), + "total_trades": total_trades, + "winning_trades": winning_trades, + "win_rate": win_rate, + "paired_count": len(pairs), + "unpaired_count": len(unpaired), + "reflected_count": sum(1 for r in reflection_map.values()), + }, + "pairs": pairs_data, + "unpaired_records": unpaired_data, + "daily_reflection": { + "id": daily_reflection.id if daily_reflection else None, + "emotion_state": daily_reflection.emotion_state if daily_reflection else None, + "market_judgment": daily_reflection.market_judgment if daily_reflection else None, + "discipline_score": daily_reflection.discipline_score if daily_reflection else None, + "overall_rating": daily_reflection.overall_rating if daily_reflection else None, + "summary": daily_reflection.summary if daily_reflection else None, + "improvements": daily_reflection.improvements if daily_reflection else None, + } if daily_reflection else None, + } + + +def _has_unpaired_records(db: Session, pair: TradePair) -> bool: + """检查配对是否有未配对的开仓或平仓记录""" + # 简化逻辑：如果开仓日期和平仓日期不同，可能存在跨日未配对 + return pair.open_date != pair.close_date + + +def create_manual_pair(db: Session, open_record_ids: list[int], close_record_ids: list[int]) -> TradePair: + """手动创建配对""" + open_records = db.query(TradeRecord).filter(TradeRecord.id.in_(open_record_ids)).all() + close_records = db.query(TradeRecord).filter(TradeRecord.id.in_(close_record_ids)).all() + + if not open_records or not close_records: + raise ValueError("开仓或平仓记录不存在") + + variety = open_records[0].variety + direction = "long" if open_records[0].direction == '买' else "short" + + # 计算统计 + total_volume = sum(r.volume or 0 for r in open_records) + total_close_pnl = sum(r.close_pnl or 0 for r in close_records) + total_commission = sum((r.commission or 0) for r in open_records) + sum((r.commission or 0) for r in close_records) + net_pnl = total_close_pnl - total_commission + + avg_open_price = sum(r.price or 0 for r in open_records) / len(open_records) if open_records else 0 + avg_close_price = sum(r.price or 0 for r in close_records) / len(close_records) if close_records else 0 + + pair = TradePair( + variety=variety, + direction=direction, + open_record_ids=open_record_ids, + close_record_ids=close_record_ids, + open_price=round(avg_open_price, 2), + close_price=round(avg_close_price, 2), + total_volume=total_volume, + close_pnl=total_close_pnl, + total_commission=total_commission, + net_pnl=round(net_pnl, 2), + open_date=open_records[0].trade_date, + close_date=close_records[-1].trade_date, + ) + db.add(pair) + db.flush() + return pair + + +def delete_pair(db: Session, pair_id: int) -> bool: + """删除配对（不删除原始交易记录）""" + pair = db.query(TradePair).filter(TradePair.id == pair_id).first() + if not pair: + return False + + # 删除关联的反思 + db.query(TradeReflection).filter(TradeReflection.trade_pair_id == pair_id).delete() + # 删除配对 + db.delete(pair) + db.flush() + return True