buffer_platform/data/trading_calculate.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
期货结算单数据处理脚本
读取期货结算单Excel，提取期货和期权交易数据，计算盈亏和持仓，合并输出
"""

import os
import glob
import re
from datetime import datetime
from collections import defaultdict

import xlrd
import pandas as pd


def find_header(sheet, keywords):
    """
    动态识别表头行
    :param sheet: xlrd sheet对象
    :param keywords: 关键词列表，如["合约", "品种合约"]
    :return: 表头行索引
    """
    for row_idx in range(min(20, sheet.nrows)):  # 只扫描前20行
        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, sheet_name, trade_type):
    """
    读取单个Sheet的数据
    :param xls_file: Excel文件路径
    :param sheet_name: Sheet名称
    :param trade_type: 交易类型（期货/期权）
    :return: DataFrame
    """
    try:
        wb = xlrd.open_workbook(xls_file)
        if sheet_name not in wb.sheet_names():
            print(f"  警告: {xls_file} 中未找到 {sheet_name}")
            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:
            print(f"  警告: {sheet_name} 中未找到表头")
            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)

        df = pd.DataFrame(data)
        print(f"  读取 {sheet_name}: {len(df)} 条记录")
        return df

    except Exception as e:
        print(f"  错误: 读取 {sheet_name} 失败 - {e}")
        return pd.DataFrame()


def extract_variety(contract):
    """
    提取品种代码（所有前导字母）
    :param contract: 合约代码
    :return: 品种代码
    """
    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 calc_pnl_futures(df):
    """
    计算期货盈亏
    """
    if df.empty:
        return {'平仓盈亏': 0, '手续费': 0, '净盈亏': 0}

    # 平仓盈亏
    pnl_close = 0
    if '平仓盈亏' in df.columns:
        for val in df['平仓盈亏']:
            if not pd.isna(val):
                pnl_close += safe_float(val)

    # 手续费
    commission = 0
    if '手续费' in df.columns:
        for val in df['手续费']:
            commission += safe_float(val)

    net_pnl = pnl_close - commission

    return {
        '平仓盈亏': pnl_close,
        '手续费': commission,
        '净盈亏': net_pnl
    }


def calc_pnl_options(df):
    """
    计算期权盈亏
    """
    if df.empty:
        return {'权利金净收入': 0, '手续费': 0, '净盈亏': 0}

    # 权利金（带方向）
    premium_total = 0
    if '权利金' in df.columns:
        for val in df['权利金']:
            premium_total += safe_float(val)

    # 手续费
    commission = 0
    if '手续费' in df.columns:
        for val in df['手续费']:
            commission += safe_float(val)

    net_pnl = premium_total - commission

    return {
        '权利金净收入': premium_total,
        '手续费': commission,
        '净盈亏': net_pnl
    }


def calc_position_futures(df):
    """
    计算期货持仓
    """
    positions = defaultdict(float)

    if df.empty:
        return positions

    for _, row in df.iterrows():
        contract = str(row.get('合约', '')).strip()
        if not contract or contract == '合计':
            continue

        bs_flag = str(row.get('买/卖', '')).strip()
        oc_flag = str(row.get('开/平', '')).strip()
        volume = safe_float(row.get('手数', 0))

        # 买+开 → +手数；卖+开 → -手数；买+平 → -手数；卖+平 → +手数
        if '买' in bs_flag and '开' in oc_flag:
            positions[contract] += volume
        elif '卖' in bs_flag and '开' in oc_flag:
            positions[contract] -= volume
        elif '买' in bs_flag and '平' in oc_flag:
            positions[contract] -= volume
        elif '卖' in bs_flag and '平' in oc_flag:
            positions[contract] += volume

    # 过滤0持仓
    return {k: v for k, v in positions.items() if abs(v) > 0.001}


def calc_position_options(df):
    """
    计算期权持仓
    """
    positions = defaultdict(float)

    if df.empty:
        return positions

    for _, row in df.iterrows():
        contract = str(row.get('品种合约', '')).strip()
        if not contract or contract == '合计':
            continue

        bs_flag = str(row.get('买/卖', '')).strip()
        volume = safe_float(row.get('成交量', 0))

        if '买' in bs_flag:
            positions[contract] += volume
        elif '卖' in bs_flag:
            positions[contract] -= volume

    # 过滤0持仓
    return {k: v for k, v in positions.items() if abs(v) > 0.001}


def merge_trades(df_futures, df_options):
    """
    合并期货和期权交易记录
    """
    # 统一字段
    columns_map_futures = {
        '合约': '合约代码',
        '买/卖': '买卖',
        '成交价': '价格/权利金单价',
        '手数': '手数/成交量',
        '成交额': '金额/权利金',
        '实际成交日期': '实际成交日期',
        '成交时间': '成交时间'
    }

    columns_map_options = {
        '品种合约': '合约代码',
        '买/卖': '买卖',
        '权利金单价': '价格/权利金单价',
        '成交量': '手数/成交量',
        '权利金': '金额/权利金',
        '成交日期': '实际成交日期',
        '成交时间': '成交时间'
    }

    # 处理期货数据
    if not df_futures.empty:
        df_f = df_futures.rename(columns=columns_map_futures).copy()
        df_f['品种'] = df_f['合约代码'].apply(extract_variety)
        df_f['方向'] = df_f['买卖'].apply(lambda x: '买' if '买' in str(x) else '卖')
        df_f['开平仓'] = df_f.get('开/平', '')

        # 确保必要字段存在
        for col in ['平仓盈亏', '手续费']:
            if col not in df_f.columns:
                df_f[col] = 0
    else:
        df_f = pd.DataFrame()

    # 处理期权数据
    if not df_options.empty:
        df_o = df_options.rename(columns=columns_map_options).copy()
        df_o['品种'] = df_o['合约代码'].apply(extract_variety)
        df_o['方向'] = df_o['买卖'].apply(lambda x: '买' if '买' in str(x) else '卖')
        df_o['开平仓'] = ''
        df_o['平仓盈亏'] = 0

        # 确保必要字段存在
        for col in ['手续费']:
            if col not in df_o.columns:
                df_o[col] = 0
    else:
        df_o = pd.DataFrame()

    # 合并
    common_cols = ['品种', '合约代码', '类型', '买卖', '方向', '成交时间',
                   '实际成交日期', '价格/权利金单价', '手数/成交量', '金额/权利金',
                   '手续费', '平仓盈亏', '开平仓']

    if not df_f.empty:
        df_f = df_f[[c for c in common_cols if c in df_f.columns]]
    if not df_o.empty:
        df_o = df_o[[c for c in common_cols if c in df_o.columns]]

    if not df_f.empty and not df_o.empty:
        return pd.concat([df_f, df_o], ignore_index=True)
    elif not df_f.empty:
        return df_f
    elif not df_o.empty:
        return df_o
    else:
        return pd.DataFrame()


def save_output(trades_df, df_futures, df_options, output_dir):
    """
    保存输出文件
    """
    os.makedirs(output_dir, exist_ok=True)

    # 文件1：trades_merged.csv
    csv_path = os.path.join(output_dir, 'trades_merged.csv')
    if not trades_df.empty:
        trades_df.to_csv(csv_path, index=False, encoding='utf-8-sig')
        print(f"已保存交易明细: {csv_path}")
    else:
        print("警告: 无交易数据")

    # 计算分析数据
    analysis_data = {}

    # Sheet1: 品种盈亏汇总
    variety_summary = []

    # 期货品种汇总
    if not df_futures.empty:
        df_f = df_futures.copy()
        df_f['品种'] = df_f['合约'].apply(extract_variety)

        for variety, group in df_f.groupby('品种'):
            pnl = calc_pnl_futures(group)
            total_volume = sum(safe_float(v) for v in group.get('手数', []))
            total_amount = sum(safe_float(v) for v in group.get('成交额', []))
            buy_count = sum(1 for v in group.get('买/卖', []) if '买' in str(v))
            sell_count = sum(1 for v in group.get('买/卖', []) if '卖' in str(v))

            variety_summary.append({
                '品种': variety,
                '类型': '期货',
                '总手数': total_volume,
                '总成交额': total_amount,
                '总手续费': pnl['手续费'],
                '平仓盈亏': pnl['平仓盈亏'],
                '净盈亏': pnl['净盈亏'],
                '买入次数': buy_count,
                '卖出次数': sell_count
            })

    # 期权品种汇总
    if not df_options.empty:
        df_o = df_options.copy()
        df_o['品种'] = df_o['品种合约'].apply(extract_variety)

        for variety, group in df_o.groupby('品种'):
            pnl = calc_pnl_options(group)
            total_volume = sum(safe_float(v) for v in group.get('成交量', []))
            total_premium = sum(abs(safe_float(v)) for v in group.get('权利金', []))
            buy_count = sum(1 for v in group.get('买/卖', []) if '买' in str(v))
            sell_count = sum(1 for v in group.get('买/卖', []) if '卖' in str(v))

            variety_summary.append({
                '品种': variety,
                '类型': '期权',
                '总手数': total_volume,
                '总成交额/权利金': total_premium,
                '总手续费': pnl['手续费'],
                '平仓盈亏': 0,
                '净盈亏': pnl['净盈亏'],
                '买入次数': buy_count,
                '卖出次数': sell_count
            })

    df_variety = pd.DataFrame(variety_summary)

    # Sheet2: 期货持仓
    futures_positions = calc_position_futures(df_futures)
    futures_pos_data = []
    for contract, pos in futures_positions.items():
        futures_pos_data.append({
            '合约': contract,
            '净持仓': pos,
            '方向': '多' if pos > 0 else '空'
        })
    df_futures_pos = pd.DataFrame(futures_pos_data)

    # Sheet3: 期权持仓
    options_positions = calc_position_options(df_options)
    options_pos_data = []
    for contract, pos in options_positions.items():
        options_pos_data.append({
            '品种合约': contract,
            '净持仓': pos,
            '方向': '权利' if pos > 0 else '义务'
        })
    df_options_pos = pd.DataFrame(options_pos_data)

    # 文件2：analysis.xlsx
    xlsx_path = os.path.join(output_dir, 'analysis.xlsx')
    with pd.ExcelWriter(xlsx_path, engine='openpyxl') as writer:
        if not df_variety.empty:
            df_variety.to_excel(writer, sheet_name='品种盈亏汇总', index=False)
        if not df_futures_pos.empty:
            df_futures_pos.to_excel(writer, sheet_name='期货持仓', index=False)
        if not df_options_pos.empty:
            df_options_pos.to_excel(writer, sheet_name='期权持仓', index=False)

    print(f"已保存分析结果: {xlsx_path}")

    return {
        'variety_summary': variety_summary,
        'futures_positions': futures_positions,
        'options_positions': options_positions
    }


def print_summary(df_futures, df_options, results):
    """
    打印摘要
    """
    print("\n" + "="*50)
    print("=== 交易摘要 ===")

    # 期货统计
    if not df_futures.empty:
        pnl_f = calc_pnl_futures(df_futures)
        varieties_f = df_futures['合约'].apply(extract_variety).nunique()
        print(f"期货：{len(df_futures)}笔交易，{varieties_f}个品种")
        print(f"  总盈亏：{pnl_f['净盈亏']:.2f}元（含手续费{pnl_f['手续费']:.2f}元）")
    else:
        print("期货：无交易")

    # 期权统计
    if not df_options.empty:
        pnl_o = calc_pnl_options(df_options)
        varieties_o = df_options['品种合约'].apply(extract_variety).nunique()
        print(f"期权：{len(df_options)}笔交易，{varieties_o}个品种")
        print(f"  总盈亏：{pnl_o['净盈亏']:.2f}元（含手续费{pnl_o['手续费']:.2f}元）")
    else:
        print("期权：无交易")

    # 持仓统计
    print("\n=== 收盘持仓 ===")

    futures_pos = results['futures_positions']
    if futures_pos:
        for contract, pos in sorted(futures_pos.items()):
            direction = '多' if pos > 0 else '空'
            print(f"期货：{contract} {direction} {abs(pos):.0f}手")
    else:
        print("期货：无持仓")

    options_pos = results['options_positions']
    if options_pos:
        for contract, pos in sorted(options_pos.items()):
            direction = '权利' if pos > 0 else '义务'
            print(f"期权：{contract} {direction} {abs(pos):.0f}手")
    else:
        print("期权：无持仓")

    print("="*50 + "\n")


def main():
    """
    主函数
    """
    # 数据目录
    data_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'data')
    output_dir = os.path.dirname(os.path.abspath(__file__))

    print(f"数据目录: {data_dir}")
    print(f"输出目录: {output_dir}\n")

    # 查找所有xls文件
    xls_files = glob.glob(os.path.join(data_dir, '*.xls'))
    if not xls_files:
        print(f"错误: 在 {data_dir} 中未找到 .xls 文件")
        return

    print(f"找到 {len(xls_files)} 个结算单文件\n")

    # 读取所有文件
    all_futures = []
    all_options = []

    for xls_file in sorted(xls_files):
        filename = os.path.basename(xls_file)
        print(f"处理: {filename}")

        # 读取期货成交明细
        df_f = read_sheet(xls_file, '成交明细', '期货')
        if not df_f.empty:
            all_futures.append(df_f)

        # 读取期权成交明细
        df_o = read_sheet(xls_file, '期权成交明细', '期权')
        if not df_o.empty:
            all_options.append(df_o)

    # 合并数据
    df_futures = pd.concat(all_futures, ignore_index=True) if all_futures else pd.DataFrame()
    df_options = pd.concat(all_options, ignore_index=True) if all_options else pd.DataFrame()

    print(f"\n汇总: 期货 {len(df_futures)} 条, 期权 {len(df_options)} 条\n")

    # 合并交易记录
    trades_merged = merge_trades(df_futures, df_options)

    # 保存输出
    results = save_output(trades_merged, df_futures, df_options, output_dir)

    # 打印摘要
    print_summary(df_futures, df_options, results)


if __name__ == '__main__':
    main()