"""
交易结算单解析服务
复用 data/trading_calculate.py 的解析逻辑，支持从 .xls 结算单文件中提取交易记录并存入数据库
"""
import re
import uuid
import json
from datetime import datetime
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

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


def _load_variety_name_map():
    """从 symbols_config.json 加载品种名称映射"""
    global _VARIETY_NAME_MAP
    if _VARIETY_NAME_MAP:
        return
    config_path = Path(__file__).resolve().parent.parent.parent / "config" / "symbols_config.json"
    if config_path.exists():
        with open(config_path, "r", encoding="utf-8") as f:
            config = json.load(f)
        # 正向: {"沪银": "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 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}
    """
    _load_variety_name_map()
    batch_id = str(uuid.uuid4())
    all_dates = set()
    futures_count = 0
    options_count = 0

    records = []

    # 处理期货记录
    if not df_futures.empty:
        for _, row in df_futures.iterrows():
            contract = str(row.get('合约', '')).strip()
            if not contract or contract == '合计':
                continue
            variety = extract_variety(contract)
            trade_date = _normalize_date(row.get('实际成交日期', ''))
            trade_time = _normalize_time(row.get('成交时间', ''))
            bs_flag = str(row.get('买/卖', '')).strip()
            oc_flag = str(row.get('开/平', '')).strip()

            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=safe_float(row.get('成交价')),
                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 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
            variety = extract_variety(contract)
            trade_date = _normalize_date(row.get('成交日期', ''))
            trade_time = _normalize_time(row.get('成交时间', ''))
            bs_flag = str(row.get('买/卖', '')).strip()

            rec = TradeRecord(
                trade_type='期权',
                symbol=contract,
                variety=variety,
                symbol_name=_VARIETY_NAME_MAP.get(variety, ''),
                direction='买' if '买' in bs_flag else '卖',
                offset='',
                price=safe_float(row.get('权利金单价')),
                volume=safe_float(row.get('成交量')),
                amount=safe_float(row.get('权利金')),
                close_pnl=0.0,
                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 trade_date:
                all_dates.add(trade_date)

    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()

    return {
        "batch_id": batch_id,
        "futures_count": futures_count,
        "options_count": options_count,
        "trade_dates": trade_dates_str,
    }


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