buffer_platform/app/services/trade_parser.py

"""
交易结算单解析服务
复用 data/trading_calculate.py 的解析逻辑，支持从 .xls 结算单文件中提取交易记录并存入数据库
"""
import re
import uuid
import json
import logging
from datetime import datetime, timedelta
from collections import defaultdict
from pathlib import Path

import xlrd
import pandas as pd
from sqlalchemy.orm import Session as DBSession

from app.analysis_models import TradeRecord, TradeImportBatch

logger = logging.getLogger(__name__)

# 品种代码 -> 中文名 映射（反向映射：从合约代码前缀查中文名）
_VARIETY_NAME_MAP = {}


def _load_variety_name_map():
    """从配置加载品种名称映射"""
    global _VARIETY_NAME_MAP
    if _VARIETY_NAME_MAP:
        return
    from app.config_store import get_config_store
    config = get_config_store().get_config("symbols", {"futures": {}, "stock": {}})
    # 正向: {"沪银": "AG2608"} -> 反向: {"AG": "沪银"}
    for name, contract in config.get("futures", {}).items():
        variety = extract_variety(contract)
        if variety and variety not in _VARIETY_NAME_MAP:
            _VARIETY_NAME_MAP[variety] = name


def extract_variety(contract):
    """提取品种代码（所有前导字母）"""
    if pd.isna(contract) or not contract:
        return ''
    contract = str(contract).strip()
    match = re.match(r'^([A-Za-z]+)', contract)
    if match:
        return match.group(1).upper()
    return contract[:2].upper()


def safe_float(value, default=0.0):
    """安全转换为float"""
    if pd.isna(value) or value is None or value == '':
        return default
    try:
        return float(value)
    except (ValueError, TypeError):
        return default


def find_header(sheet, keywords):
    """
    动态识别表头行
    :param sheet: xlrd sheet对象
    :param keywords: 关键词列表
    :return: 表头行索引
    """
    for row_idx in range(min(20, sheet.nrows)):
        row_values = [str(sheet.cell_value(row_idx, col_idx)).strip()
                      for col_idx in range(sheet.ncols)]
        row_text = ' '.join(row_values)
        has_keyword = any(kw in row_text for kw in keywords)
        has_aux = any(aux in row_text for aux in ["买/卖", "成交价", "权利金"])
        if has_keyword and has_aux:
            return row_idx
    return None


def read_sheet(xls_file_path, sheet_name, trade_type):
    """
    读取单个Sheet的数据
    :param xls_file_path: Excel文件路径（str 或 bytes）
    :param sheet_name: Sheet名称
    :param trade_type: 交易类型（期货/期权）
    :return: DataFrame
    """
    try:
        wb = xlrd.open_workbook(xls_file_path)
        if sheet_name not in wb.sheet_names():
            return pd.DataFrame()

        sheet = wb.sheet_by_name(sheet_name)
        if sheet.nrows == 0:
            return pd.DataFrame()

        if trade_type == '期货':
            keywords = ["合约"]
        else:
            keywords = ["品种合约"]

        header_row = find_header(sheet, keywords)
        if header_row is None:
            return pd.DataFrame()

        headers = [str(sheet.cell_value(header_row, col_idx)).strip()
                   for col_idx in range(sheet.ncols)]

        data = []
        for row_idx in range(header_row + 1, sheet.nrows):
            row_data = {}
            first_cell = str(sheet.cell_value(row_idx, 0)).strip()
            if first_cell == '合计' or first_cell == '':
                continue
            for col_idx, header in enumerate(headers):
                if col_idx < sheet.ncols:
                    value = sheet.cell_value(row_idx, col_idx)
                    if str(value).strip() == '--':
                        value = None
                    row_data[header] = value
            row_data['类型'] = trade_type
            data.append(row_data)

        return pd.DataFrame(data)
    except Exception:
        return pd.DataFrame()


def parse_settlement_file(file_content: bytes, filename: str) -> tuple[pd.DataFrame, pd.DataFrame]:
    """
    解析结算单文件，返回期货和期权交易DataFrame
    :param file_content: 文件二进制内容
    :param filename: 文件名
    :return: (df_futures, df_options)
    """
    import tempfile
    # xlrd 需要文件路径，写入临时文件
    with tempfile.NamedTemporaryFile(suffix='.xls', delete=False) as tmp:
        tmp.write(file_content)
        tmp_path = tmp.name

    try:
        df_futures = read_sheet(tmp_path, '成交明细', '期货')
        df_options = read_sheet(tmp_path, '期权成交明细', '期权')
        return df_futures, df_options
    finally:
        Path(tmp_path).unlink(missing_ok=True)


def _normalize_date(date_val) -> str:
    """将各种日期格式统一为 YYYY-MM-DD"""
    if pd.isna(date_val) or date_val is None or date_val == '':
        return ''
    s = str(date_val).strip()
    # 已经是 YYYY-MM-DD
    if re.match(r'^\d{4}-\d{2}-\d{2}$', s):
        return s
    # YYYY/MM/DD
    if re.match(r'^\d{4}/\d{2}/\d{2}$', s):
        return s.replace('/', '-')
    # YYYYMMDD
    if re.match(r'^\d{8}$', s):
        return f"{s[:4]}-{s[4:6]}-{s[6:8]}"
    # xlrd 日期数字 (Excel serial date)
    try:
        f = float(s)
        if 30000 < f < 60000:  # 合理的 Excel 日期数字范围
            dt = xlrd.xldate_as_datetime(f, 0)
            return dt.strftime('%Y-%m-%d')
    except (ValueError, TypeError):
        pass
    return s


def _normalize_time(time_val) -> str:
    """将各种时间格式统一为 HH:MM:SS"""
    if pd.isna(time_val) or time_val is None or time_val == '':
        return ''
    s = str(time_val).strip()
    # 已经是 HH:MM:SS
    if re.match(r'^\d{1,2}:\d{2}:\d{2}$', s):
        return s
    # HH:MM
    if re.match(r'^\d{1,2}:\d{2}$', s):
        return s + ':00'
    # xlrd 时间数字
    try:
        f = float(s)
        if 0 <= f < 1:
            hours = int(f * 24)
            minutes = int((f * 24 - hours) * 60)
            seconds = int(((f * 24 - hours) * 60 - minutes) * 60)
            return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
    except (ValueError, TypeError):
        pass
    return s


def recalculate_trade_date(trade_date: str, trade_time: str) -> str:
    """
    根据交易时间重新计算交易日

    期货交易规则：
    - 前一交易日 21:00 之后到当前交易日 15:00 之前的交易，算作当天交易
    - 周五 21:00 之后的夜盘交易，算作下周一的交易数据

    :param trade_date: 原始交易日 (YYYY-MM-DD)
    :param trade_time: 交易时间 (HH:MM:SS)
    :return: 重新计算后的交易日
    """
    if not trade_date or not trade_time:
        return trade_date

    try:
        # 解析日期和时间
        dt = datetime.strptime(trade_date, '%Y-%m-%d')
        time_parts = trade_time.split(':')
        hour = int(time_parts[0])

        # 判断是否在夜盘时段 (21:00 之后)
        if hour >= 21:
            # 夜盘交易，交易日归属下一天
            next_day = dt + timedelta(days=1)
            # 如果是周五 (weekday=4)，下一天是周六，需要跳到下周一
            if dt.weekday() == 4:  # 周五
                next_day = dt + timedelta(days=3)  # 跳到下周一
            # 如果是周六 (weekday=5)，跳到下周一
            elif dt.weekday() == 5:
                next_day = dt + timedelta(days=2)
            # 如果是周日 (weekday=6)，跳到下周一
            elif dt.weekday() == 6:
                next_day = dt + timedelta(days=1)
            return next_day.strftime('%Y-%m-%d')
        else:
            # 日盘交易，保持原交易日
            return trade_date
    except (ValueError, IndexError):
        return trade_date


def _build_dedup_key(trade_date: str, variety: str, trade_time: str, price: float) -> str:
    """构建去重键：交易日+品种+交易时间+价格（价格保留2位小数避免浮点精度问题）"""
    return f"{trade_date}|{variety}|{trade_time}|{round(price, 2)}"


def _load_existing_dedup_keys(db: DBSession, trade_dates: set) -> set:
    """
    根据涉及的交易日，批量加载已有记录的去重键集合
    """
    if not trade_dates:
        logger.warning("[去重校验] trade_dates 为空，跳过查询")
        return set()
    logger.info(f"[去重校验] 查询已有记录，日期范围: {sorted(trade_dates)}")
    existing = db.query(
        TradeRecord.trade_date,
        TradeRecord.variety,
        TradeRecord.trade_time,
        TradeRecord.price,
    ).filter(
        TradeRecord.trade_date.in_(trade_dates)
    ).all()
    logger.info(f"[去重校验] 从数据库加载到 {len(existing)} 条已有记录")
    keys = {_build_dedup_key(r[0], r[1], r[2] or '', r[3] or 0.0) for r in existing}
    if keys:
        sample = list(keys)[:3]
        logger.info(f"[去重校验] 去重键示例: {sample}")
    return keys


def save_to_db(
    db: DBSession,
    df_futures: pd.DataFrame,
    df_options: pd.DataFrame,
    filename: str,
) -> dict:
    """
    将解析后的交易记录保存到数据库（含逐条去重校验）
    去重规则：同一交易日 + 同一品种 + 同一交易时间 + 同一价格 => 视为重复，跳过
    :return: {
        "batch_id": str, "futures_count": int, "options_count": int, "trade_dates": str,
        "futures_skipped": int, "options_skipped": int, "total_parsed": int
    }
    """
    _load_variety_name_map()
    batch_id = str(uuid.uuid4())
    all_dates = set()
    futures_count = 0
    options_count = 0
    futures_skipped = 0
    options_skipped = 0
    futures_parsed = 0
    options_parsed = 0

    # 第一遍：收集所有涉及的交易日，用于批量查询已有记录
    if not df_futures.empty:
        for _, row in df_futures.iterrows():
            contract = str(row.get('合约', '')).strip()
            if not contract or contract == '合计':
                continue
            trade_date = _normalize_date(row.get('实际成交日期', ''))
            if trade_date:
                all_dates.add(trade_date)
    if not df_options.empty:
        for _, row in df_options.iterrows():
            contract = str(row.get('品种合约', '')).strip()
            if not contract or contract == '合计':
                continue
            trade_date = _normalize_date(row.get('成交日期', ''))
            if trade_date:
                all_dates.add(trade_date)

    # 批量加载已有记录的去重键
    existing_keys = _load_existing_dedup_keys(db, all_dates)

    records = []
    skipped_details = []  # 记录被跳过的重复记录详情

    # 处理期货记录
    if not df_futures.empty:
        for _, row in df_futures.iterrows():
            contract = str(row.get('合约', '')).strip()
            if not contract or contract == '合计':
                continue
            futures_parsed += 1
            variety = extract_variety(contract)
            raw_date = _normalize_date(row.get('实际成交日期', ''))
            trade_time = _normalize_time(row.get('成交时间', ''))
            # 根据交易时间重新计算交易日（夜盘归属下一天）
            trade_date = recalculate_trade_date(raw_date, trade_time)
            bs_flag = str(row.get('买/卖', '')).strip()
            oc_flag = str(row.get('开/平', '')).strip()
            price = safe_float(row.get('成交价'))

            dedup_key = _build_dedup_key(trade_date, variety, trade_time, price)
            if dedup_key in existing_keys:
                futures_skipped += 1
                skipped_details.append({
                    'type': '期货',
                    'variety': variety,
                    'date': trade_date,
                    'time': trade_time,
                    'price': price,
                })
                continue

            # 记录前几条的去重键用于调试
            if futures_parsed <= 3:
                logger.info(f"[去重校验] 期货记录 #{futures_parsed}: raw_date={raw_date!r}, trade_date={trade_date!r}, variety={variety!r}, time={trade_time!r}, price={price!r} => key={dedup_key!r}")

            rec = TradeRecord(
                trade_type='期货',
                symbol=contract,
                variety=variety,
                symbol_name=_VARIETY_NAME_MAP.get(variety, ''),
                direction='买' if '买' in bs_flag else '卖',
                offset='开' if '开' in oc_flag else ('平' if '平' in oc_flag else ''),
                price=price,
                volume=safe_float(row.get('手数')),
                amount=safe_float(row.get('成交额')),
                close_pnl=safe_float(row.get('平仓盈亏')),
                commission=safe_float(row.get('手续费')),
                trade_date=trade_date,
                trade_time=trade_time,
                import_batch=batch_id,
                source_file=filename,
            )
            records.append(rec)
            futures_count += 1

    # 处理期权记录
    if not df_options.empty:
        for _, row in df_options.iterrows():
            contract = str(row.get('品种合约', '')).strip()
            if not contract or contract == '合计':
                continue
            options_parsed += 1
            variety = extract_variety(contract)
            raw_date = _normalize_date(row.get('成交日期', ''))
            trade_time = _normalize_time(row.get('成交时间', ''))
            # 根据交易时间重新计算交易日（夜盘归属下一天）
            trade_date = recalculate_trade_date(raw_date, trade_time)
            bs_flag = str(row.get('买/卖', '')).strip()
            price = safe_float(row.get('权利金单价'))

            dedup_key = _build_dedup_key(trade_date, variety, trade_time, price)
            if dedup_key in existing_keys:
                options_skipped += 1
                skipped_details.append({
                    'type': '期权',
                    'variety': variety,
                    'date': trade_date,
                    'time': trade_time,
                    'price': price,
                })
                continue

            # 期权盈亏来自权利金（带正负号：卖为正，买为负）
            premium = safe_float(row.get('权利金'))
            rec = TradeRecord(
                trade_type='期权',
                symbol=contract,
                variety=variety,
                symbol_name=_VARIETY_NAME_MAP.get(variety, ''),
                direction='买' if '买' in bs_flag else '卖',
                offset='',
                price=price,
                volume=safe_float(row.get('成交量')),
                amount=premium,
                close_pnl=premium,  # 期权盈亏 = 权利金（带正负号）
                commission=safe_float(row.get('手续费')),
                trade_date=trade_date,
                trade_time=trade_time,
                import_batch=batch_id,
                source_file=filename,
            )
            records.append(rec)
            options_count += 1

    if records:
        db.add_all(records)

    # 日期范围描述
    sorted_dates = sorted(all_dates)
    trade_dates_str = ''
    if sorted_dates:
        if len(sorted_dates) == 1:
            trade_dates_str = sorted_dates[0]
        else:
            trade_dates_str = f"{sorted_dates[0]} ~ {sorted_dates[-1]}"

    # 保存批次记录
    batch = TradeImportBatch(
        batch_id=batch_id,
        source_file=filename,
        futures_count=futures_count,
        options_count=options_count,
        trade_dates=trade_dates_str,
    )
    db.add(batch)
    db.commit()

    logger.info(
        f"[去重校验] 导入完成: 文件={filename}, "
        f"期货 解析={futures_parsed} 新增={futures_count} 跳过={futures_skipped}, "
        f"期权 解析={options_parsed} 新增={options_count} 跳过={options_skipped}, "
        f"已有去重键数量={len(existing_keys)}"
    )

    return {
        "batch_id": batch_id,
        "futures_count": futures_count,
        "options_count": options_count,
        "trade_dates": trade_dates_str,
        "futures_skipped": futures_skipped,
        "options_skipped": options_skipped,
        "futures_parsed": futures_parsed,
        "options_parsed": options_parsed,
        "skipped_details": skipped_details,  # 被跳过的重复记录详情
    }


def calc_daily_summary(db: DBSession, start_date: str = None, end_date: str = None) -> list[dict]:
    """按日期汇总交易盈亏"""
    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)
    query = query.order_by(TradeRecord.trade_date)

    records = query.all()
    daily = defaultdict(lambda: {
        "trade_date": "",
        "total_trades": 0,
        "total_pnl": 0.0,
        "total_commission": 0.0,
        "win_count": 0,
        "loss_count": 0,
        "max_win": 0.0,
        "max_loss": 0.0,
        "buy_volume": 0.0,
        "sell_volume": 0.0,
        "varieties": set(),
    })

    for r in records:
        d = r.trade_date or '未知'
        day = daily[d]
        day["trade_date"] = d
        day["total_trades"] += 1
        pnl = (r.close_pnl or 0) - (r.commission or 0)
        day["total_pnl"] += pnl
        day["total_commission"] += (r.commission or 0)
        if pnl > 0:
            day["win_count"] += 1
            day["max_win"] = max(day["max_win"], pnl)
        elif pnl < 0:
            day["loss_count"] += 1
            day["max_loss"] = min(day["max_loss"], pnl)
        if r.direction == '买':
            day["buy_volume"] += (r.volume or 0)
        else:
            day["sell_volume"] += (r.volume or 0)
        day["varieties"].add(r.variety)

    result = []
    for d in sorted(daily.keys()):
        day = daily[d]
        total = day["win_count"] + day["loss_count"]
        result.append({
            "trade_date": day["trade_date"],
            "total_trades": day["total_trades"],
            "total_pnl": round(day["total_pnl"], 2),
            "total_commission": round(day["total_commission"], 2),
            "win_count": day["win_count"],
            "loss_count": day["loss_count"],
            "win_rate": round(day["win_count"] / total * 100, 1) if total > 0 else 0,
            "max_win": round(day["max_win"], 2),
            "max_loss": round(day["max_loss"], 2),
            "buy_volume": day["buy_volume"],
            "sell_volume": day["sell_volume"],
            "variety_count": len(day["varieties"]),
        })
    return result


def calc_variety_summary(db: DBSession, start_date: str = None, end_date: str = None) -> list[dict]:
    """按品种汇总交易盈亏"""
    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)

    records = query.all()
    varieties = defaultdict(lambda: {
        "variety": "",
        "symbol_name": "",
        "total_trades": 0,
        "total_pnl": 0.0,
        "total_commission": 0.0,
        "total_close_pnl": 0.0,
        "total_volume": 0.0,
        "total_amount": 0.0,
        "buy_count": 0,
        "sell_count": 0,
        "win_count": 0,
        "loss_count": 0,
    })

    for r in records:
        v = varieties[r.variety]
        v["variety"] = r.variety
        v["symbol_name"] = r.symbol_name or v["symbol_name"]
        v["total_trades"] += 1
        v["total_close_pnl"] += (r.close_pnl or 0)
        v["total_commission"] += (r.commission or 0)
        v["total_pnl"] += (r.close_pnl or 0) - (r.commission or 0)
        v["total_volume"] += (r.volume or 0)
        v["total_amount"] += (r.amount or 0)
        if r.direction == '买':
            v["buy_count"] += 1
        else:
            v["sell_count"] += 1
        pnl = (r.close_pnl or 0) - (r.commission or 0)
        if pnl > 0:
            v["win_count"] += 1
        elif pnl < 0:
            v["loss_count"] += 1

    result = []
    for variety in sorted(varieties.keys()):
        v = varieties[variety]
        total = v["win_count"] + v["loss_count"]
        result.append({
            "variety": v["variety"],
            "symbol_name": v["symbol_name"],
            "total_trades": v["total_trades"],
            "total_pnl": round(v["total_pnl"], 2),
            "total_close_pnl": round(v["total_close_pnl"], 2),
            "total_commission": round(v["total_commission"], 2),
            "total_volume": v["total_volume"],
            "total_amount": round(v["total_amount"], 2),
            "buy_count": v["buy_count"],
            "sell_count": v["sell_count"],
            "win_count": v["win_count"],
            "loss_count": v["loss_count"],
            "win_rate": round(v["win_count"] / total * 100, 1) if total > 0 else 0,
        })
    return result


def calc_overall_statistics(db: DBSession, start_date: str = None, end_date: str = None) -> dict:
    """计算整体交易统计"""
    daily = calc_daily_summary(db, start_date, end_date)
    if not daily:
        return {
            "total_trades": 0,
            "total_pnl": 0,
            "total_commission": 0,
            "win_rate": 0,
            "profit_loss_ratio": 0,
            "max_single_win": 0,
            "max_single_loss": 0,
            "max_consecutive_wins": 0,
            "max_consecutive_losses": 0,
            "trading_days": 0,
            "avg_daily_pnl": 0,
            "max_daily_pnl": 0,
            "min_daily_pnl": 0,
        }

    total_trades = sum(d["total_trades"] for d in daily)
    total_pnl = sum(d["total_pnl"] for d in daily)
    total_commission = sum(d["total_commission"] for d in daily)
    total_wins = sum(d["win_count"] for d in daily)
    total_losses = sum(d["loss_count"] for d in daily)
    total_decided = total_wins + total_losses

    # 最大连续盈/亏
    max_consec_wins = 0
    max_consec_losses = 0
    cur_wins = 0
    cur_losses = 0
    for d in daily:
        if d["total_pnl"] > 0:
            cur_wins += 1
            cur_losses = 0
            max_consec_wins = max(max_consec_wins, cur_wins)
        elif d["total_pnl"] < 0:
            cur_losses += 1
            cur_wins = 0
            max_consec_losses = max(max_consec_losses, cur_losses)
        else:
            cur_wins = 0
            cur_losses = 0

    # 平均日盈亏
    pnl_values = [d["total_pnl"] for d in daily]

    # 盈亏比 - 用逐笔数据计算
    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)
    all_records = query.all()

    wins_pnl = []
    losses_pnl = []
    for r in all_records:
        pnl = (r.close_pnl or 0) - (r.commission or 0)
        if pnl > 0:
            wins_pnl.append(pnl)
        elif pnl < 0:
            losses_pnl.append(pnl)

    avg_win_val = sum(wins_pnl) / len(wins_pnl) if wins_pnl else 0
    avg_loss_val = abs(sum(losses_pnl) / len(losses_pnl)) if losses_pnl else 0
    profit_loss_ratio = round(avg_win_val / avg_loss_val, 2) if avg_loss_val > 0 else 0

    return {
        "total_trades": total_trades,
        "total_pnl": round(total_pnl, 2),
        "total_commission": round(total_commission, 2),
        "win_rate": round(total_wins / total_decided * 100, 1) if total_decided > 0 else 0,
        "profit_loss_ratio": profit_loss_ratio,
        "max_single_win": round(max(wins_pnl), 2) if wins_pnl else 0,
        "max_single_loss": round(min(losses_pnl), 2) if losses_pnl else 0,
        "max_consecutive_wins": max_consec_wins,
        "max_consecutive_losses": max_consec_losses,
        "trading_days": len(daily),
        "avg_daily_pnl": round(sum(pnl_values) / len(pnl_values), 2) if pnl_values else 0,
        "max_daily_pnl": round(max(pnl_values), 2) if pnl_values else 0,
        "min_daily_pnl": round(min(pnl_values), 2) if pnl_values else 0,
    }


def get_trade_pairs(db: DBSession, start_date: str = None, end_date: str = None) -> list[dict]:
    """
    将开平仓配对，生成逐笔交易对
    按合约+品种分组，按时间排序，买开/卖开 与 卖平/买平 配对
    """
    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)
    query = query.order_by(TradeRecord.symbol, TradeRecord.trade_date, TradeRecord.trade_time)

    records = query.all()

    # 按合约分组
    by_symbol = defaultdict(list)
    for r in records:
        by_symbol[r.symbol].append(r)

    pairs = []
    pair_id = 0

    for symbol, recs in by_symbol.items():
        open_positions = []  # 未平仓记录

        for r in recs:
            if r.offset == '开':
                open_positions.append(r)
            elif r.offset == '平' and open_positions:
                # 配对：取最早的一条开仓
                open_rec = open_positions.pop(0)
                pair_id += 1
                pnl = (r.close_pnl or 0) - (open_rec.commission or 0) - (r.commission or 0)
                direction = '多' if open_rec.direction == '买' else '空'
                pairs.append({
                    "id": pair_id,
                    "symbol": symbol,
                    "variety": r.variety,
                    "symbol_name": r.symbol_name,
                    "direction": direction,
                    "open_date": open_rec.trade_date,
                    "open_time": open_rec.trade_time,
                    "open_price": open_rec.price,
                    "close_date": r.trade_date,
                    "close_time": r.trade_time,
                    "close_price": r.price,
                    "volume": open_rec.volume,
                    "close_pnl": round((r.close_pnl or 0), 2),
                    "commission": round((open_rec.commission or 0) + (r.commission or 0), 2),
                    "net_pnl": round(pnl, 2),
                    "open_batch": open_rec.import_batch,
                    "close_batch": r.import_batch,
                })
            elif r.offset == '' or r.offset is None:
                # 期权等无开平标记的，按买卖交替配对
                if open_positions:
                    open_rec = open_positions.pop(0)
                    pair_id += 1
                    pairs.append({
                        "id": pair_id,
                        "symbol": symbol,
                        "variety": r.variety,
                        "symbol_name": r.symbol_name,
                        "direction": '多' if open_rec.direction == '买' else '空',
                        "open_date": open_rec.trade_date,
                        "open_time": open_rec.trade_time,
                        "open_price": open_rec.price,
                        "close_date": r.trade_date,
                        "close_time": r.trade_time,
                        "close_price": r.price,
                        "volume": open_rec.volume,
                        "close_pnl": 0,
                        "commission": round((open_rec.commission or 0) + (r.commission or 0), 2),
                        "net_pnl": 0,
                        "open_batch": open_rec.import_batch,
                        "close_batch": r.import_batch,
                    })
                else:
                    open_positions.append(r)

    return pairs