""" 交易配对引擎 - 将开仓和平仓配对为完整交易,持久化到 TradePair 表 """ import logging from typing import Optional from sqlalchemy.orm import Session from collections import defaultdict from app.analysis_models import TradeRecord, TradePair, TradeReflection, DailyReflection logger = logging.getLogger(__name__) def auto_pair_trades(db: Session, trade_date: str) -> dict: """ 自动配对指定日期及之前未平仓的交易记录 按品种+方向分组,FIFO 规则配对 返回:{created: 新建配对数, updated: 更新配对数, unpaired: 未配对记录数} """ # 查询指定日期的记录 current_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 current_records: return {"created": 0, "updated": 0, "unpaired": 0} # 查询之前未配对的记录(跨日持仓) # 找出所有已经配对的记录 ID all_pairs = db.query(TradePair).all() paired_ids = set() for p in all_pairs: paired_ids.update(p.open_record_ids or []) paired_ids.update(p.close_record_ids or []) # 查询 trade_date 之前的未配对记录 previous_unpaired = db.query(TradeRecord).filter( TradeRecord.trade_date < trade_date, TradeRecord.id.notin_(paired_ids) if paired_ids else True, ).order_by( TradeRecord.variety, TradeRecord.symbol, TradeRecord.trade_date, TradeRecord.trade_time ).all() # 合并当前记录和之前未配对记录 all_records = previous_unpaired + current_records # 按品种分组 by_variety = defaultdict(list) for r in all_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, ).all() for p in existing_pairs: all_paired_open_ids.update(p.open_record_ids or []) all_paired_close_ids.update(p.close_record_ids or []) 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 # 开仓队列,每个元素是 (record, remaining_volume) open_queue = [(r, r.volume or 0) for r in opens] close_queue = [(r, r.volume or 0) for r in closes] while open_queue and close_queue: close_rec, close_remaining = close_queue[0] # 从开仓队列中取足够手数来匹配这个平仓 needed = close_remaining used_opens = [] # (record, take_volume) while needed > 0 and open_queue: open_rec, open_remaining = open_queue[0] take = min(open_remaining, needed) used_opens.append((open_rec, take)) needed -= take if take >= open_remaining: open_queue.pop(0) else: open_queue[0] = (open_rec, open_remaining - take) if not used_opens: break # 实际匹配的手数 matched_volume = close_remaining - needed # 更新平仓队列 if matched_volume >= close_remaining: close_queue.pop(0) else: close_queue[0] = (close_rec, close_remaining - matched_volume) # 计算均价(按手数加权) open_volume_sum = sum(take for _, take in used_opens) open_price = sum((r.price or 0) * take for r, take in used_opens) / open_volume_sum if open_volume_sum > 0 else 0 close_price = close_rec.price or 0 # 计算盈亏和手续费 # 平仓盈亏按比例分配到该平仓记录 close_pnl = (close_rec.close_pnl or 0) * (matched_volume / close_rec.volume) if close_rec.volume else 0 total_commission = sum((r.commission or 0) * (take / r.volume) for r, take in used_opens if r.volume) + \ (close_rec.commission or 0) * (matched_volume / close_rec.volume) if close_rec.volume else 0 net_pnl = close_pnl - total_commission # 最早开仓日期和最早平仓日期 open_date = min(r.trade_date for r, _ in used_opens) close_date = close_rec.trade_date pair = TradePair( variety=variety, direction=direction, open_record_ids=[r.id for r, _ in used_opens], close_record_ids=[close_rec.id], open_price=round(open_price, 2), close_price=round(close_price, 2), total_volume=matched_volume, close_pnl=round(close_pnl, 2), total_commission=round(total_commission, 2), net_pnl=round(net_pnl, 2), open_date=open_date, close_date=close_date, ) db.add(pair) created += 1 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 or []) paired_record_ids.update(p.close_record_ids or []) 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