buffer_platform/app/services/trade_pairing_engine.py

"""
交易配对引擎 - 将开仓和平仓配对为完整交易，持久化到 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