from __future__ import annotations

import io
import math
import re
from dataclasses import dataclass, field
from datetime import datetime
from difflib import SequenceMatcher
from typing import Any, Iterable

import numpy as np
import pandas as pd
from openpyxl import load_workbook
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
from openpyxl.utils import get_column_letter

from config_logic import (
    BusinessConfig,
    FormulaRule,
    KeywordRule,
    config_audit_frame,
    config_validation_frame,
    evaluate_formula,
    render_email_template,
)

APP_VERSION = "1.0"


def clean_text(value: Any) -> str:
    if value is None:
        return ""
    try:
        if pd.isna(value):
            return ""
    except Exception:
        pass
    text = str(value).strip().replace("\u3000", " ")
    return re.sub(r"\s+", " ", text)


def norm_header(value: Any) -> str:
    text = clean_text(value).lower().replace("（", "(").replace("）", ")")
    return re.sub(r"[\s_\-—·/\\:：@%％().,，]+", "", text)


def norm_code(value: Any) -> str:
    text = clean_text(value).upper()
    if not text:
        return ""
    text = re.sub(r"\s+", "", text)
    text = re.sub(r"\.(OF|SH|SZ|BJ)$", "", text)
    if re.fullmatch(r"\d+\.0+", text):
        text = text.split(".")[0]
    if re.fullmatch(r"\d+", text) and len(text) <= 6:
        return text.zfill(6)
    return text


def norm_name(value: Any) -> str:
    text = clean_text(value).upper().replace("（", "(").replace("）", ")")
    text = re.sub(r"(?:[-—_]\d+|\(\d+\))$", "", text)
    text = re.sub(r"[\s\-—_·,，。()（）\[\]【】]+", "", text)
    return text


def canonical_fund_name(value: Any) -> str:
    """用于名称兜底匹配；仅移除常见公司前缀、基金全称尾缀和份额后缀。"""
    text = norm_name(value)
    for token in ["兴证全球", "兴全", "证券投资基金", "基金", "型"]:
        text = text.replace(token, "")
    text = text.replace("持有期", "持有")
    text = re.sub(r"[A-ZCE]$", "", text)
    return text


def unique_columns(values: Iterable[Any]) -> list[str]:
    result: list[str] = []
    counts: dict[str, int] = {}
    for idx, value in enumerate(values):
        base = clean_text(value) or f"未命名列{idx + 1}"
        counts[base] = counts.get(base, 0) + 1
        result.append(base if counts[base] == 1 else f"{base}_{counts[base]}")
    return result


FIELD_SYNONYMS: dict[str, list[str]] = {
    "company": ["公司", "所属公司", "公司主体", "主体"],
    "product_name": ["产品名称", "基金名称", "资产名称", "证券名称", "产品简称"],
    "holding_code": ["产品代码", "内部产品代码", "资产代码", "产品编号", "证券代码"],
    "holding_fund_code": ["基金代码", "初始基金代码", "外部基金代码", "产品代码", "证券代码"],
    "initial_fund_code": ["初始基金代码", "原基金代码", "基金代码"],
    "product_category": ["产品类别", "资产类别", "一级分类", "产品大类"],
    "type1": ["产品类型1", "产品类型一", "产品类型", "二级分类"],
    "type2": ["产品类型2", "产品类型二", "三级分类"],
    "report_type": ["兴证报表类型", "报表类型", "统计类型"],
    "manager": ["产品管理人", "管理人", "发行机构", "产品发行人"],
    "current_value": ["当前净值", "当前市值", "持仓市值", "最新市值", "资产净值"],
    "current_cost": ["当前成本", "持仓成本", "成本"],
    "scale_base": ["本年投入规模", "投资规模", "投入规模", "当前投入规模"],
    "period_end_value": ["2025期末市值", "上年末市值", "年初市值", "期末市值"],
    "period_end_principal": ["2025期末投入本金", "上年末投入本金", "年初投入本金", "期末投入本金"],
    "ytd_direct": [
        "年内已实现及未实现投资收益",
        "本年已实现及未实现投资收益",
        "年累计投资收益",
        "本年累计收益",
        "年累计损益",
        "累计损益",
    ],
    "ytd_float_change": ["年累计浮盈变化", "本年浮盈变化", "年内浮盈变化"],
    "ytd_realized": ["年累计投资损益", "本年累计投资损益", "年内已实现投资损益"],
    "float_today": ["浮盈 as of today", "浮盈as of today", "当前浮盈", "今日浮盈", "浮盈截至今日"],
    "float_year_end": ["浮盈 @ 上年末", "浮盈@上年末", "上年末浮盈", "年初浮盈"],
    "active_flag": ["是否存量", "存量标识", "有效标识", "是否在持", "持仓标识", "状态标识"],
    "daily_pnl": ["当日亏损金额", "当日损益", "当日预计损益", "当日盈亏", "今日损益", "每日损益"],
    "daily_return": [
        "当日股票收益率%", "当日股票收益率(%)", "当日股票收益率（%）", "当日股票收益率",
        "当日收益率%", "当日收益率(%)", "当日收益率（%）", "当日收益率",
    ],
    "o32_company": ["公司", "所属公司", "公司主体", "主体"],
    "o32_id": ["O32基金编号", "基金编号", "O32编号", "O32代码"],
    "o32_code": ["基金代码", "外部基金代码", "产品代码", "证券代码"],
    "o32_name": ["基金名称", "产品名称", "资产名称"],
    "o32_return": [
        "单位净值涨跌幅(%)", "单位净值涨跌幅（%）", "单位净值涨跌幅",
        "当日涨跌幅(%)", "当日涨跌幅（%）", "当日涨跌幅",
        "日收益率(%)", "日收益率（%）", "日收益率",
    ],
    "o32_stock_return": ["当日股票收益率(%)", "当日股票收益率（%）", "当日股票收益率"],
    "o32_bond_return": ["当日债券收益率(%)", "当日债券收益率（%）", "当日债券收益率"],
    "o32_unit_nav": ["单位净值", "基金单位净值"],
    "date": ["统计日期", "业务日期", "数据日期", "日期"],
    "dim_holding_code": ["产品代码", "内部产品代码", "资产代码", "产品编号", "基金代码", "BK_FUNDCODE"],
    "dim_holding_name": ["产品名称", "基金名称", "资产名称", "PRODUCT_CNAME", "PRODUCT_SHORT_NAME"],
    "dim_company": ["公司", "所属公司", "公司主体", "主体"],
    "dim_o32_id": ["O32基金编号", "基金编号", "O32编号", "O32_CODE"],
    "dim_o32_code": ["O32基金代码", "外部基金代码", "证券代码", "FUND_CODE"],
}

FIELD_LABELS = {
    "company": "公司",
    "product_name": "产品名称",
    "holding_code": "产品代码",
    "holding_fund_code": "基金代码",
    "initial_fund_code": "初始基金代码",
    "product_category": "产品类别",
    "type1": "产品类型1",
    "type2": "产品类型2",
    "report_type": "报表类型",
    "manager": "产品管理人",
    "current_value": "当前净值",
    "current_cost": "当前成本",
    "scale_base": "本年投入规模",
    "period_end_value": "上年末市值",
    "period_end_principal": "上年末投入本金",
    "ytd_direct": "年内已实现及未实现投资收益",
    "ytd_float_change": "年累计浮盈变化",
    "ytd_realized": "年累计投资损益",
    "float_today": "当前浮盈",
    "float_year_end": "上年末浮盈",
    "active_flag": "存量标识",
    "daily_pnl": "表内当日损益金额",
    "daily_return": "表内当日收益率",
    "o32_company": "O32公司",
    "o32_id": "O32基金编号",
    "o32_code": "O32基金代码",
    "o32_name": "O32基金名称",
    "o32_return": "单位净值涨跌幅",
    "o32_stock_return": "当日股票收益率",
    "o32_bond_return": "当日债券收益率",
    "o32_unit_nav": "单位净值",
    "date": "统计日期",
    "dim_holding_code": "映射表持仓代码",
    "dim_holding_name": "映射表产品名称",
    "dim_company": "映射表公司",
    "dim_o32_id": "映射表O32编号",
    "dim_o32_code": "映射表O32代码",
}

ROLE_FIELDS = {
    "holding": [
        "company", "product_name", "holding_code", "holding_fund_code", "initial_fund_code",
        "product_category", "type1", "type2", "report_type", "manager",
        "current_value", "current_cost", "scale_base", "period_end_value", "period_end_principal",
        "ytd_direct", "ytd_float_change", "ytd_realized", "float_today", "float_year_end",
        "daily_pnl", "daily_return", "active_flag",
    ],
    "o32": ["o32_company", "o32_id", "o32_code", "o32_name", "o32_return", "o32_stock_return", "o32_bond_return", "o32_unit_nav", "date"],
    "dimension": ["dim_company", "dim_holding_code", "dim_holding_name", "dim_o32_id", "dim_o32_code"],
}

ROLE_HINTS = {
    "holding": ["产品名称", "产品代码", "本年投入规模", "年累计浮盈变化", "年内已实现及未实现投资收益"],
    "o32": ["基金编号", "基金名称", "单位净值涨跌幅", "当日股票收益率", "统计日期"],
    "dimension": ["BK_FUNDCODE", "O32_CODE", "PRODUCT_CNAME", "产品代码", "O32基金编号"],
}

ROLE_LABELS = {"holding": "资金明细 / 持仓主表", "o32": "O32每日收益率表", "dimension": "产品信息表"}
STOCK_DAILY_MODE = "当日股票收益率（默认）"
STOCK_BOND_DAILY_MODE = "股票+债券收益率"
UNIT_NAV_DAILY_MODE = "单位净值涨跌幅（备选核对）"
BOND_DAILY_MODE = "当日债券收益率"
HALF_WEIGHT_DAILY_MODE = "偏债/FOF/FOS按50%，其他按100%（默认）"
FULL_WEIGHT_DAILY_MODE = "全部产品按100%"
SMALL_EXCLUDED_MONEY_FUND_LIMIT = 10_000_000.0
EMBEDDED_AMOUNT_RATE_TOLERANCE = 0.01
DEFAULT_DAILY_SOURCE_PRIORITY = ["standard_o32", "embedded_rate", "embedded_amount"]
YTD_FLOAT_END_DAILY_PNL_MODE = "浮盈@上年末 + 当日亏损金额"

AMOUNT_FIELDS = {
    "current_value", "current_cost", "scale_base", "period_end_value", "period_end_principal",
    "ytd_direct", "ytd_float_change", "ytd_realized", "float_today", "float_year_end", "daily_pnl",
}
RATE_FIELDS = {"daily_return", "o32_return", "o32_stock_return", "o32_bond_return"}
AMOUNT_HEADER_BLOCKERS = (
    "%", "％", "收益率", "浮盈率", "比例", "比率", "占比", "上限", "下限", "涨跌幅",
)
UNIT_NAV_HEADER_BLOCKERS = ("%", "％", "收益率", "涨跌幅")
ACTIVE_HEADER_BLOCKERS = (
    "%", "％", "金额", "成本", "净值", "市值", "投入", "浮盈", "损益", "收益率", "比例", "比率",
    "上限", "下限", "日期", "代码", "编号", "名称", "份额", "数量",
)


@dataclass
class ParsedSheet:
    df: pd.DataFrame
    sheet_name: str
    header_row: int
    score: float
    source_name: str


@dataclass
class RoleAssessment:
    role: str
    parsed: ParsedSheet
    fields: dict[str, str | None]
    confidence: float
    issues: list[str]

    @property
    def valid(self) -> bool:
        return not self.issues


@dataclass
class DailySource:
    kind: str
    label: str
    parsed: ParsedSheet
    fields: dict[str, str | None]
    confidence: float
    notes: list[str]


@dataclass
class AutoAssignment:
    holding: RoleAssessment | None
    o32: RoleAssessment | None
    dimension: RoleAssessment | None
    daily_source: DailySource | None
    assessments: list[RoleAssessment]
    issues: list[str]


@dataclass
class CalculationResult:
    summary: dict[str, Any]
    email_text: str
    detail: pd.DataFrame
    exceptions: pd.DataFrame
    warnings: list[str]
    field_notes: list[str]
    candidate_metrics: dict[str, float]
    config_audit: pd.DataFrame = field(default_factory=pd.DataFrame)
    config_comparison: pd.DataFrame = field(default_factory=pd.DataFrame)
    config_validation: pd.DataFrame = field(default_factory=pd.DataFrame)


def active_business_config(config: BusinessConfig | None) -> BusinessConfig | None:
    if config and config.enabled and config.valid:
        return config
    return None


def _strategy_value(config: BusinessConfig | None, key: str, default: str = "") -> str:
    business_config = active_business_config(config)
    if business_config:
        value = clean_text(business_config.strategy.get(key))
        if value:
            return value
    return default


def _strategy_float(config: BusinessConfig | None, key: str, default: float) -> float:
    value = _strategy_value(config, key, "")
    if not value:
        return default
    try:
        return float(value.replace(",", "").replace("，", ""))
    except Exception:
        return default


def _cfg_strategy_float(cfg: dict[str, Any] | None, key: str, default: float) -> float:
    if not cfg:
        return default
    return _strategy_float(cfg.get("business_config"), key, default)


def _normalize_daily_source_token(value: Any) -> str | None:
    text = clean_text(value)
    compact = norm_header(text)
    lowered = compact.lower()
    if lowered in {"standardo32", "o32", "standardo32", "standard_o32"}:
        return "standard_o32"
    if lowered in {"embeddedrate", "tablerate", "holdingrate", "rate", "embedded_rate"}:
        return "embedded_rate"
    if lowered in {"embeddedamount", "tableamount", "holdingamount", "amount", "embedded_amount"}:
        return "embedded_amount"
    if "o32" in lowered or ("标准" in text and "表内" not in text):
        return "standard_o32"
    if "表内" in text and ("收益率" in text or "涨跌幅" in text or "重算" in text):
        return "embedded_rate"
    if "表内" in text and ("金额" in text or "损益" in text):
        return "embedded_amount"
    return None


def _daily_source_priority(config: BusinessConfig | None = None) -> list[str]:
    raw = _strategy_value(config, "daily_source_priority", "")
    if raw:
        parsed: list[str] = []
        for token in re.split(r"\s*(?:>|＞|,|，|、|;|；|\n)\s*", raw):
            kind = _normalize_daily_source_token(token)
            if kind and kind not in parsed:
                parsed.append(kind)
        if parsed:
            return parsed + [kind for kind in DEFAULT_DAILY_SOURCE_PRIORITY if kind not in parsed]
    return list(DEFAULT_DAILY_SOURCE_PRIORITY)


def _standard_daily_default_mode(config: BusinessConfig | None, o32_fields: dict[str, str | None]) -> str:
    return STOCK_DAILY_MODE


def configured_synonyms(field: str, config: BusinessConfig | None = None) -> list[str]:
    synonyms = list(FIELD_SYNONYMS.get(field, []))
    business_config = active_business_config(config)
    if business_config:
        for alias in business_config.field_synonyms.get(field, []):
            if alias not in synonyms:
                synonyms.append(alias)
    return synonyms


def _header_has_token(header: Any, tokens: Iterable[str]) -> bool:
    raw = clean_text(header).upper()
    normalized = norm_header(header).upper()
    for token in tokens:
        clean_token = clean_text(token).upper()
        if clean_token and clean_token in raw:
            return True
        normalized_token = norm_header(token).upper()
        if normalized_token and normalized_token in normalized:
            return True
    return False


def _column_allowed_for_field(column: str, field: str) -> bool:
    if field in AMOUNT_FIELDS and _header_has_token(column, AMOUNT_HEADER_BLOCKERS):
        return False
    if field == "o32_unit_nav" and _header_has_token(column, UNIT_NAV_HEADER_BLOCKERS):
        return False
    if field in RATE_FIELDS:
        text = clean_text(column)
        return any(token in text for token in ["收益率", "涨跌幅", "%", "％"])
    return True


def detect_column(columns: list[str], field: str, config: BusinessConfig | None = None) -> str | None:
    synonyms = configured_synonyms(field, config)
    norm_cols = {col: norm_header(col) for col in columns if _column_allowed_for_field(col, field)}
    for synonym in synonyms:
        ns = norm_header(synonym)
        for col, nc in norm_cols.items():
            if nc == ns:
                return col
    candidates: list[tuple[float, str]] = []
    for rank, synonym in enumerate(synonyms):
        ns = norm_header(synonym)
        if len(ns) < 4:
            continue
        for col, nc in norm_cols.items():
            if ns in nc:
                candidates.append((100 - rank / 100, col))
            elif len(nc) >= 4:
                ratio = SequenceMatcher(None, ns, nc).ratio()
                if ratio >= 0.90:
                    candidates.append((ratio * 90 - rank / 100, col))
    return max(candidates, default=(0.0, None), key=lambda x: x[0])[1]


def _looks_like_data_cell(value: Any) -> bool:
    if value is None:
        return False
    try:
        if pd.isna(value):
            return False
    except Exception:
        pass
    if isinstance(value, (int, float, np.number)) and not isinstance(value, bool):
        return True
    if isinstance(value, (datetime, pd.Timestamp)):
        return True
    text = clean_text(value)
    compact = text.replace(",", "").replace("，", "")
    if re.fullmatch(r"[-+]?\d+(?:\.\d+)?%?", compact):
        return True
    if re.fullmatch(r"\d{4}[-/年]?\d{1,2}[-/月]?\d{1,2}日?", compact):
        return True
    if re.fullmatch(r"\d{8}", compact):
        return True
    return False


def header_row_score(row: pd.Series, role: str, config: BusinessConfig | None = None) -> float:
    raw_cells = [v for v in row.tolist() if clean_text(v)]
    cells = [norm_header(v) for v in raw_cells]
    if not cells:
        return 0.0
    hints = [norm_header(v) for v in ROLE_HINTS[role]]
    role_synonyms = {norm_header(v) for f in ROLE_FIELDS[role] for v in configured_synonyms(f, config)}
    score = 0.0
    for cell in cells:
        if cell in role_synonyms:
            score += 3.0
        if any(len(h) >= 4 and h in cell for h in hints):
            score += 2.0
    data_like = sum(_looks_like_data_cell(v) for v in raw_cells)
    long_text = sum(len(clean_text(v)) > 60 for v in raw_cells)
    data_ratio = data_like / max(len(raw_cells), 1)
    if len(set(cells)) >= 5 and data_ratio <= 0.35:
        score += 1.0
    score -= min(data_like, 12) * 0.35
    score -= min(long_text, 4) * 1.0
    return score


def choose_engine(filename: str) -> str | None:
    suffix = filename.lower().rsplit(".", 1)[-1] if "." in filename else ""
    if suffix == "xls":
        return "xlrd"
    if suffix in {"xlsx", "xlsm"}:
        return "openpyxl"
    return None


def parse_excel_bytes(file_bytes: bytes, filename: str, role: str, config: BusinessConfig | None = None) -> ParsedSheet:
    engine = choose_engine(filename)
    if engine is None:
        raise ValueError(f"{filename} 不是支持的 Excel 文件。")
    excel = pd.ExcelFile(io.BytesIO(file_bytes), engine=engine)
    candidates: list[ParsedSheet] = []
    for sheet_name in excel.sheet_names:
        raw = pd.read_excel(io.BytesIO(file_bytes), sheet_name=sheet_name, header=None, dtype=object, engine=engine)
        if raw.empty:
            continue
        scan_rows = min(len(raw), 40)
        scores = [(idx, header_row_score(raw.iloc[idx], role, config)) for idx in range(scan_rows)]
        header_idx, score = max(scores, key=lambda x: x[1])
        columns = unique_columns(raw.iloc[header_idx].tolist())
        data = raw.iloc[header_idx + 1 :].copy()
        data.columns = columns
        data = data.dropna(how="all").dropna(axis=1, how="all").reset_index(drop=True)
        data["_源文件行号"] = data.index + header_idx + 2
        candidates.append(ParsedSheet(data, str(sheet_name), header_idx + 1, score, filename))
    if not candidates:
        raise ValueError(f"{filename} 未读取到有效工作表。")
    return max(candidates, key=lambda p: (p.score, len(p.df.columns), len(p.df)))


def detect_fields(columns: list[str], role: str, config: BusinessConfig | None = None) -> dict[str, str | None]:
    return {field: detect_column(columns, field, config) for field in ROLE_FIELDS[role]}


def _has_current_value_formula(fields: dict[str, str | None]) -> bool:
    return bool(
        fields.get("current_value")
        or (fields.get("scale_base") and fields.get("ytd_float_change"))
        or (fields.get("current_cost") and fields.get("float_today"))
        or (
            fields.get("scale_base") and fields.get("period_end_value")
            and fields.get("period_end_principal") and fields.get("float_today")
        )
    )


def _has_ytd_formula(fields: dict[str, str | None]) -> bool:
    return bool(
        (fields.get("float_year_end") and fields.get("daily_pnl"))
        or
        (fields.get("period_end_value") and fields.get("scale_base"))
        or fields.get("ytd_direct")
        or (fields.get("ytd_float_change") and fields.get("ytd_realized"))
        or (fields.get("float_today") and fields.get("float_year_end") and fields.get("ytd_realized"))
    )


def detect_active_flag(df: pd.DataFrame, detected: str | None) -> str | None:
    if detected and detected in df.columns:
        return detected
    best: tuple[float, str | None] = (0.0, None)
    valid_tokens = {"Y", "N", "YES", "NO", "是", "否", "1", "0"}
    text_tokens = {"Y", "N", "YES", "NO", "是", "否"}
    for col in df.columns:
        if col == "_源文件行号" or _header_has_token(col, ACTIVE_HEADER_BLOCKERS):
            continue
        values = df[col].dropna().map(lambda x: clean_text(x).upper())
        if len(values) < 5:
            continue
        ratio = values.isin(valid_tokens).mean()
        unique = set(values.unique())
        text_unique = unique & text_tokens
        if ratio >= 0.90 and len(unique) <= 4 and len(text_unique) >= 2 and ratio > best[0]:
            best = (ratio, col)
    return best[1]


def _role_confidence_adjustment(role: str, fields: dict[str, str | None]) -> float:
    if role == "holding":
        score = 0.0
        if fields.get("company"):
            score += 15.0
        if fields.get("product_name") and any(fields.get(k) for k in ["holding_code", "holding_fund_code", "initial_fund_code"]):
            score += 35.0
        if _has_current_value_formula(fields):
            score += 35.0
        if _has_ytd_formula(fields):
            score += 35.0
        if fields.get("daily_pnl"):
            score += 25.0
        return score
    if role == "o32":
        score = 0.0
        if fields.get("o32_name") and any(fields.get(k) for k in ["o32_id", "o32_code"]):
            score += 35.0
        if fields.get("o32_stock_return"):
            score += 45.0
        if fields.get("o32_return") or fields.get("o32_bond_return"):
            score += 10.0
        if fields.get("date"):
            score += 20.0
        return score
    score = 0.0
    if any(fields.get(k) for k in ["dim_holding_code", "dim_holding_name"]):
        score += 25.0
    if any(fields.get(k) for k in ["dim_o32_id", "dim_o32_code"]):
        score += 45.0
    return score


def assess_sheet(parsed: ParsedSheet, role: str, config: BusinessConfig | None = None) -> RoleAssessment:
    columns = [c for c in parsed.df.columns if c != "_源文件行号"]
    fields = detect_fields(columns, role, config)
    if role == "holding":
        fields["active_flag"] = detect_active_flag(parsed.df, fields.get("active_flag"))
    issues: list[str] = []
    if role == "holding":
        if not fields.get("product_name"):
            issues.append("缺少产品名称")
        if not any(fields.get(k) for k in ["holding_code", "holding_fund_code", "initial_fund_code"]):
            issues.append("缺少产品/基金代码")
        if not _has_current_value_formula(fields):
            issues.append("无法推导当前净值")
        if not _has_ytd_formula(fields):
            issues.append("无法推导累计损益")
        key_count = sum(bool(fields.get(k)) for k in ROLE_FIELDS[role])
        confidence = parsed.score * 8 + key_count * 10 + _role_confidence_adjustment(role, fields)
    elif role == "o32":
        if not fields.get("o32_name"):
            issues.append("缺少基金名称")
        if not any(fields.get(k) for k in ["o32_id", "o32_code"]):
            issues.append("缺少O32编号/代码")
        if not fields.get("o32_stock_return"):
            issues.append("缺少当日股票收益率")
        key_count = sum(bool(fields.get(k)) for k in ROLE_FIELDS[role])
        confidence = parsed.score * 8 + key_count * 12 + _role_confidence_adjustment(role, fields)
    else:
        if not any(fields.get(k) for k in ["dim_holding_code", "dim_holding_name"]):
            issues.append("缺少持仓侧代码/名称")
        if not any(fields.get(k) for k in ["dim_o32_id", "dim_o32_code"]):
            issues.append("缺少O32编号/代码")
        key_count = sum(bool(fields.get(k)) for k in ROLE_FIELDS[role])
        confidence = parsed.score * 8 + key_count * 15 + _role_confidence_adjustment(role, fields)
    return RoleAssessment(role, parsed, fields, confidence, issues)


def same_sheet(left: RoleAssessment | DailySource | None, right: RoleAssessment | DailySource | None) -> bool:
    if not left or not right:
        return False
    return (
        left.parsed.source_name == right.parsed.source_name
        and left.parsed.sheet_name == right.parsed.sheet_name
    )


def detect_embedded_daily_fields(df: pd.DataFrame, config: BusinessConfig | None = None) -> dict[str, str | None]:
    columns = [c for c in df.columns if c != "_源文件行号"]
    fields = {
        "daily_pnl": detect_column(columns, "daily_pnl", config),
        "daily_return": detect_column(columns, "daily_return", config),
        "o32_return": detect_column(columns, "o32_return", config),
        "o32_stock_return": detect_column(columns, "o32_stock_return", config),
        "o32_bond_return": detect_column(columns, "o32_bond_return", config),
    }
    if not fields.get("daily_return"):
        fields["daily_return"] = fields.get("o32_stock_return")
    return fields


def _embedded_rate_col(fields: dict[str, str | None]) -> str | None:
    return fields.get("daily_return") or fields.get("o32_stock_return")


def embedded_daily_score(assessment: RoleAssessment, config: BusinessConfig | None = None) -> tuple[float, dict[str, str | None], list[str]]:
    fields = detect_embedded_daily_fields(assessment.parsed.df, config)
    amount_col = fields.get("daily_pnl")
    rate_col = _embedded_rate_col(fields)
    score = 0.0
    notes: list[str] = []
    if amount_col:
        score += 180.0
        notes.append(f"识别到表内当日损益金额列“{amount_col}”")
    if rate_col:
        score += 100.0
        notes.append(f"识别到表内当日收益率列“{rate_col}”")
    if amount_col and rate_col and assessment.fields.get("current_value"):
        score += 80.0
        notes.append("表内金额与收益率可相互校验")
    return score, fields, notes


def make_embedded_daily_source(holding: RoleAssessment, config: BusinessConfig | None = None) -> DailySource | None:
    sources = make_embedded_daily_sources(holding, config)
    return sources.get("embedded_rate") or sources.get("embedded_amount")


def make_embedded_daily_sources(holding: RoleAssessment, config: BusinessConfig | None = None) -> dict[str, DailySource]:
    score, fields, notes = embedded_daily_score(holding, config)
    amount_col = fields.get("daily_pnl")
    rate_col = _embedded_rate_col(fields)
    sources: dict[str, DailySource] = {}
    if rate_col:
        sources["embedded_rate"] = DailySource(
            kind="embedded_rate",
            label="表内当日收益率",
            parsed=holding.parsed,
            fields=fields,
            confidence=holding.confidence + score + 60.0,
            notes=notes,
        )
    if amount_col:
        sources["embedded_amount"] = DailySource(
            kind="embedded_amount",
            label="表内当日损益金额",
            parsed=holding.parsed,
            fields=fields,
            confidence=holding.confidence + score,
            notes=notes,
        )
    return sources


def make_standard_daily_source(o32: RoleAssessment) -> DailySource:
    fields = dict(o32.fields)
    notes = [f"识别到标准O32每日收益率表“{o32.parsed.source_name} / {o32.parsed.sheet_name}”"]
    return DailySource(
        kind="standard_o32",
        label="标准O32每日收益率表",
        parsed=o32.parsed,
        fields=fields,
        confidence=o32.confidence + 300.0,
        notes=notes,
    )


def auto_assign_files(files: list[tuple[str, bytes]], config: BusinessConfig | None = None) -> AutoAssignment:
    assessments: list[RoleAssessment] = []
    issues: list[str] = []
    for filename, file_bytes in files:
        for role in ["holding", "o32", "dimension"]:
            try:
                parsed = parse_excel_bytes(file_bytes, filename, role, config)
                assessments.append(assess_sheet(parsed, role, config))
            except Exception as exc:
                issues.append(f"{filename} 读取失败：{exc}")
    def choose(role: str, excluded: list[RoleAssessment | DailySource]) -> RoleAssessment | None:
        candidates = [a for a in assessments if a.role == role and a.valid]
        candidates.sort(key=lambda a: a.confidence, reverse=True)
        for item in candidates:
            if not any(same_sheet(item, x) for x in excluded):
                return item
        return None

    holding_candidates = [a for a in assessments if a.role == "holding" and a.valid]
    holding_candidates.sort(
        key=lambda a: (a.confidence + embedded_daily_score(a, config)[0], a.confidence, len(a.parsed.df)),
        reverse=True,
    )
    holding = holding_candidates[0] if holding_candidates else None
    o32 = choose("o32", [holding] if holding else [])
    daily_sources: dict[str, DailySource] = {}
    if o32:
        daily_sources["standard_o32"] = make_standard_daily_source(o32)
    if holding:
        daily_sources.update(make_embedded_daily_sources(holding, config))
    daily_source = next(
        (daily_sources[kind] for kind in _daily_source_priority(config) if kind in daily_sources),
        None,
    )
    if daily_source is None and daily_sources:
        daily_source = max(daily_sources.values(), key=lambda source: source.confidence)
    dimension = choose("dimension", [x for x in [holding, o32] if x])
    if holding is None:
        issues.append("未能自动识别资金明细 / 持仓主表。")
    if daily_source is None:
        issues.append("未能自动识别每日损益来源：需要标准O32表，或持仓表内含当日损益金额/当日收益率。")
    return AutoAssignment(holding, o32, dimension, daily_source, assessments, issues)


def numeric_series(series: pd.Series) -> pd.Series:
    if pd.api.types.is_numeric_dtype(series):
        return pd.to_numeric(series, errors="coerce")
    cleaned = series.map(clean_text).str.replace(",", "", regex=False).str.replace("，", "", regex=False)
    cleaned = cleaned.str.replace("%", "", regex=False).str.replace("％", "", regex=False)
    cleaned = cleaned.replace({"-": np.nan, "--": np.nan, "N/A": np.nan, "nan": np.nan, "None": np.nan})
    return pd.to_numeric(cleaned, errors="coerce")


def resolve_rate_scale(
    series: pd.Series,
    column_name: str,
    mode: str = "自动识别",
    value_series: pd.Series | None = None,
    amount_series: pd.Series | None = None,
) -> tuple[pd.Series, str]:
    values = numeric_series(series)
    if mode.startswith("百分数"):
        return values / 100.0, "按百分数除以100"
    if mode.startswith("小数"):
        return values, "按小数直接使用"

    if value_series is not None and amount_series is not None:
        base_values = numeric_series(value_series)
        amounts = numeric_series(amount_series)
        valid = values.notna() & base_values.notna() & amounts.notna() & base_values.abs().gt(1e-9)
        if int(valid.sum()) >= 3:
            pred_decimal = base_values.loc[valid] * values.loc[valid]
            pred_percent = pred_decimal / 100.0
            denominator = max(float(amounts.loc[valid].abs().sum()), 1.0)
            decimal_error = float((pred_decimal - amounts.loc[valid]).abs().sum()) / denominator
            percent_error = float((pred_percent - amounts.loc[valid]).abs().sum()) / denominator
            if decimal_error <= percent_error * 0.25:
                return values, "通过表内当日损益金额校验，按小数直接使用"
            if percent_error <= decimal_error * 0.25:
                return values / 100.0, "通过表内当日损益金额校验，按百分数除以100"

    if "%" in clean_text(column_name) or "百分" in clean_text(column_name):
        finite = values[np.isfinite(values)]
        if not finite.empty and finite.abs().quantile(0.90) <= 0.20:
            return values, "列名含百分号但数值分布更像小数，未除以100"
        return values / 100.0, "列名含百分号且数值分布更像百分数，已除以100"
    finite = values[np.isfinite(values)]
    if not finite.empty and finite.abs().quantile(0.90) > 0.5:
        return values / 100.0, "根据数值分布识别为百分数，已除以100"
    return values, "根据列名和数值分布识别为小数"


def _config_from_cfg(cfg: dict[str, Any] | None) -> BusinessConfig | None:
    if not cfg:
        return None
    return active_business_config(cfg.get("business_config"))


def _formula_context(
    df: pd.DataFrame,
    fields: dict[str, str | None],
    extra: dict[str, pd.Series] | None = None,
    fill_na: bool = True,
) -> dict[str, pd.Series]:
    context: dict[str, pd.Series] = {}
    for field_key, col in fields.items():
        if col and col in df.columns:
            values = numeric_series(df[col])
            if fill_na:
                values = values.fillna(0.0)
            context[field_key] = values
            context[FIELD_LABELS.get(field_key, field_key)] = values
            context[col] = values
    if extra:
        context.update(extra)
    return context


def _add_config_formulas(
    formulas: dict[str, tuple[pd.Series, str]],
    df: pd.DataFrame,
    fields: dict[str, str | None],
    config: BusinessConfig | None,
    category: str,
    strict: bool = False,
) -> None:
    business_config = active_business_config(config)
    if not business_config:
        return
    context = _formula_context(df, fields, fill_na=True)
    for rule in business_config.enabled_formulas(category):
        try:
            values = evaluate_formula(rule.formula, context, df.index).fillna(0.0)
        except Exception as exc:
            formulas.pop(rule.name, None)
            if strict:
                raise ValueError(f"配置公式“{rule.name}”无法计算：{exc}") from exc
            continue
        formulas[rule.name] = (values, f"配置公式“{rule.name}”：{rule.formula}")


def _selected_config_formula(config: BusinessConfig | None, category: str, mode: str) -> FormulaRule | None:
    business_config = active_business_config(config)
    if not business_config:
        return None
    for rule in business_config.enabled_formulas(category):
        if rule.name == mode:
            return rule
    return None


def _formula_series(
    df: pd.DataFrame,
    fields: dict[str, str | None],
    config: BusinessConfig | None = None,
    strict: bool = False,
) -> dict[str, tuple[pd.Series, str]]:
    formulas: dict[str, tuple[pd.Series, str]] = {}
    if fields.get("current_value"):
        c = fields["current_value"]
        formulas["直接使用当前净值"] = (numeric_series(df[c]).fillna(0.0), f"直接使用“{c}”")
    if fields.get("scale_base") and fields.get("ytd_float_change"):
        s, f = fields["scale_base"], fields["ytd_float_change"]
        formulas["本年投入规模 + 年累计浮盈变化"] = (
            numeric_series(df[s]).fillna(0.0) + numeric_series(df[f]).fillna(0.0),
            f"“{s}” + “{f}”",
        )
    if fields.get("current_cost") and fields.get("float_today"):
        c, f = fields["current_cost"], fields["float_today"]
        formulas["当前成本 + 当前浮盈"] = (
            numeric_series(df[c]).fillna(0.0) + numeric_series(df[f]).fillna(0.0),
            f"“{c}” + “{f}”",
        )
    if all(fields.get(k) for k in ["scale_base", "period_end_value", "period_end_principal", "float_today"]):
        s, v, p, f = (fields[k] for k in ["scale_base", "period_end_value", "period_end_principal", "float_today"])
        formulas["投入规模 - 上年末市值 + 上年末本金 + 当前浮盈"] = (
            numeric_series(df[s]).fillna(0.0) - numeric_series(df[v]).fillna(0.0)
            + numeric_series(df[p]).fillna(0.0) + numeric_series(df[f]).fillna(0.0),
            f"“{s}” - “{v}” + “{p}” + “{f}”",
        )
    _add_config_formulas(formulas, df, fields, config, "current_value", strict=strict)
    return formulas


def derive_current_value(
    df: pd.DataFrame,
    fields: dict[str, str | None],
    mode: str,
    config: BusinessConfig | None = None,
) -> tuple[pd.Series, str, list[str], dict[str, float]]:
    formulas = _formula_series(df, fields, config, strict=False)
    if not formulas:
        raise ValueError("无法从持仓表推导当前净值。")
    priorities = ["直接使用当前净值", "本年投入规模 + 年累计浮盈变化", "当前成本 + 当前浮盈", "投入规模 - 上年末市值 + 上年末本金 + 当前浮盈"]
    selected = next((k for k in priorities if k in formulas), next(iter(formulas))) if mode == "自动选择最可靠口径" else mode
    if selected not in formulas:
        raise ValueError(f"所选当前净值口径“{selected}”缺少必要字段。")
    values, note = formulas[selected]
    warnings: list[str] = []
    totals = {name: float(series.sum()) for name, (series, _) in formulas.items()}
    base = totals[selected]
    for name, total in totals.items():
        if name == selected:
            continue
        threshold = max(5_000_000.0, abs(base) * 0.01)
        if abs(total - base) > threshold:
            warnings.append(f"当前净值候选口径不一致：{selected}={base/1e8:.3f}亿元，{name}={total/1e8:.3f}亿元。")
    return values, note, warnings, totals


def _ytd_formulas(
    df: pd.DataFrame,
    fields: dict[str, str | None],
    config: BusinessConfig | None = None,
    strict: bool = False,
) -> dict[str, tuple[pd.Series, str]]:
    formulas: dict[str, tuple[pd.Series, str]] = {}
    if fields.get("float_year_end"):
        f = fields["float_year_end"]
        base = numeric_series(df[f]).fillna(0.0)
        if fields.get("daily_pnl"):
            d = fields["daily_pnl"]
            base = base + numeric_series(df[d]).fillna(0.0)
        formulas["浮盈@上年末 + 当日亏损金额"] = (
            base,
            f"“{f}” + 本次最终当日损益",
        )
    if fields.get("period_end_value") and fields.get("scale_base"):
        v, s = fields["period_end_value"], fields["scale_base"]
        formulas["上年末市值 - 本年投入规模"] = (
            numeric_series(df[v]).fillna(0.0) - numeric_series(df[s]).fillna(0.0),
            f"“{v}” - “{s}”",
        )
    if fields.get("ytd_direct"):
        c = fields["ytd_direct"]
        formulas["年内已实现及未实现投资收益"] = (numeric_series(df[c]).fillna(0.0), f"直接使用“{c}”")
    if fields.get("ytd_float_change") and fields.get("ytd_realized"):
        f, r = fields["ytd_float_change"], fields["ytd_realized"]
        formulas["年累计浮盈变化 + 年累计投资损益"] = (
            numeric_series(df[f]).fillna(0.0) + numeric_series(df[r]).fillna(0.0),
            f"“{f}” + “{r}”",
        )
        formulas["仅年累计投资损益"] = (numeric_series(df[r]).fillna(0.0), f"仅使用“{r}”")
        formulas["仅年累计浮盈变化"] = (numeric_series(df[f]).fillna(0.0), f"仅使用“{f}”")
    if fields.get("float_today") and fields.get("float_year_end") and fields.get("ytd_realized"):
        t, y, r = fields["float_today"], fields["float_year_end"], fields["ytd_realized"]
        formulas["当前浮盈 - 上年末浮盈 + 年累计投资损益"] = (
            numeric_series(df[t]).fillna(0.0) - numeric_series(df[y]).fillna(0.0) + numeric_series(df[r]).fillna(0.0),
            f"“{t}” - “{y}” + “{r}”",
        )
    if fields.get("ytd_direct") and fields.get("ytd_float_change"):
        d, f = fields["ytd_direct"], fields["ytd_float_change"]
        formulas["仅已实现损益（总收益-浮盈变化推导）"] = (
            numeric_series(df[d]).fillna(0.0) - numeric_series(df[f]).fillna(0.0),
            f"“{d}” - “{f}”",
        )
        if "仅年累计浮盈变化" not in formulas:
            formulas["仅年累计浮盈变化"] = (numeric_series(df[f]).fillna(0.0), f"仅使用“{f}”")
    _add_config_formulas(formulas, df, fields, config, "ytd", strict=strict)
    return formulas


def derive_ytd(
    df: pd.DataFrame,
    fields: dict[str, str | None],
    mode: str,
    config: BusinessConfig | None = None,
) -> tuple[pd.Series, str, dict[str, pd.Series]]:
    formulas = _ytd_formulas(df, fields, config, strict=False)
    if not formulas:
        raise ValueError("无法从持仓表推导累计损益。")
    selected = _select_ytd_formula_name(formulas, mode, config)
    values, note = formulas[selected]
    return values, note, {k: v[0] for k, v in formulas.items()}


def _select_ytd_formula_name(
    formulas: dict[str, tuple[pd.Series, str]],
    mode: str,
    config: BusinessConfig | None = None,
) -> str:
    priorities = [
        YTD_FLOAT_END_DAILY_PNL_MODE,
        "上年末市值 - 本年投入规模",
        "年内已实现及未实现投资收益",
        "年累计浮盈变化 + 年累计投资损益",
        "当前浮盈 - 上年末浮盈 + 年累计投资损益",
    ]
    if mode == "自动选择最可靠口径":
        strategy_mode = _strategy_value(config, "ytd_default_mode", "")
        selected = next(
            (
                name for name in formulas
                if strategy_mode and (
                    norm_header(name) == norm_header(strategy_mode)
                    or norm_header(strategy_mode) in norm_header(name)
                    or norm_header(name) in norm_header(strategy_mode)
                )
            ),
            "",
        )
        if not selected:
            selected = YTD_FLOAT_END_DAILY_PNL_MODE
    else:
        selected = mode
    if selected not in formulas:
        if _is_final_daily_ytd_mode(selected):
            raise ValueError("默认累计损益口径缺少必要字段：上年末浮盈（可识别列名如“浮盈 @ 上年末”“浮盈@上年末”“上年末浮盈”“年初浮盈”）。")
        raise ValueError(f"所选累计损益口径“{selected}”缺少必要字段。")
    return selected


def _is_final_daily_ytd_mode(name: str) -> bool:
    return norm_header(name) == norm_header(YTD_FLOAT_END_DAILY_PNL_MODE)


def _apply_final_daily_ytd(
    h: pd.DataFrame,
    detail: pd.DataFrame,
    ytd_candidates: dict[str, pd.Series],
    holding_fields: dict[str, str | None],
    selected_name: str,
    current_note: str,
) -> str:
    if not _is_final_daily_ytd_mode(selected_name):
        return current_note
    float_col = holding_fields.get("float_year_end")
    if not float_col or float_col not in h.columns or "当日损益" not in detail.columns:
        return current_note

    final_daily = pd.to_numeric(detail["当日损益"], errors="coerce").fillna(0.0)
    final_daily.index = h.index
    adjusted = numeric_series(h[float_col]).fillna(0.0) + final_daily
    h["_累计损益"] = adjusted
    detail.loc[:, "累计损益"] = adjusted.to_numpy()
    ytd_candidates[YTD_FLOAT_END_DAILY_PNL_MODE] = adjusted
    return f"“{float_col}” + 本次最终当日损益（已应用剔除、置0和计算权重）"


def _replace_ytd_note(notes: list[str], ytd_note: str) -> None:
    for idx, note in enumerate(notes):
        if note.startswith("累计损益："):
            notes[idx] = f"累计损益：{ytd_note}"
            return
    notes.append(f"累计损益：{ytd_note}")


def available_current_value_modes(
    df: pd.DataFrame,
    fields: dict[str, str | None],
    config: BusinessConfig | None = None,
) -> list[str]:
    return ["自动选择最可靠口径"] + list(_formula_series(df, fields, config, strict=False).keys())


def available_ytd_modes(
    df: pd.DataFrame,
    fields: dict[str, str | None],
    config: BusinessConfig | None = None,
) -> list[str]:
    return ["自动选择最可靠口径"] + list(_ytd_formulas(df, fields, config, strict=False).keys())


def available_daily_modes(daily_source: DailySource, o32_fields: dict[str, str | None], config: BusinessConfig | None = None) -> list[str]:
    business_config = active_business_config(config)
    modes: list[str] = []
    if business_config:
        custom_modes = [rule.name for rule in business_config.enabled_formulas("daily")]
        modes.extend(custom_modes)
    if daily_source.kind == "standard_o32":
        if o32_fields.get("o32_stock_return"):
            modes.append(STOCK_DAILY_MODE)
        if o32_fields.get("o32_stock_return") or o32_fields.get("o32_bond_return"):
            modes.append(STOCK_BOND_DAILY_MODE)
        if o32_fields.get("o32_return"):
            modes.append(UNIT_NAV_DAILY_MODE)
        if o32_fields.get("o32_bond_return"):
            modes.append(BOND_DAILY_MODE)
    else:
        modes.append(daily_source.label)
    return list(dict.fromkeys(modes))


def contains_any(value: Any, keywords: list[str]) -> bool:
    text = clean_text(value).upper()
    normalized_text = norm_header(value).upper()
    for keyword in keywords:
        clean_keyword = clean_text(keyword).upper()
        normalized_keyword = norm_header(keyword).upper()
        if clean_keyword and clean_keyword in text:
            return True
        if normalized_keyword and normalized_keyword in normalized_text:
            return True
    return False


def _cfg_bool(cfg: dict[str, Any] | None, key: str, default: bool) -> bool:
    if not cfg or key not in cfg:
        return default
    value = cfg.get(key)
    if isinstance(value, bool):
        return value
    text = clean_text(value).upper()
    if text in {"是", "启用", "有效", "Y", "YES", "TRUE", "1", "ON"}:
        return True
    if text in {"否", "不启用", "无效", "N", "NO", "FALSE", "0", "OFF"}:
        return False
    return bool(value)


def _exclude_non_six_product_code_enabled(cfg: dict[str, Any]) -> bool:
    if "exclude_non_six_product_code" in cfg:
        return _cfg_bool(cfg, "exclude_non_six_product_code", True)
    return _cfg_bool(cfg, "exclude_money", True)


def _is_six_digit_code(value: Any) -> bool:
    return bool(re.fullmatch(r"\d{6}", norm_code(value)))


def _half_weight_mask(h: pd.DataFrame) -> pd.Series:
    type2 = h.get("_产品类型2", pd.Series([""] * len(h), index=h.index)).map(norm_header).str.upper()
    return type2.str.contains("债券类|债券型|偏债型|FOF型|FOF类|FOS型|FOS类", regex=True, na=False)


def _small_excluded_money_fund_mask(h: pd.DataFrame, ytd_mask: pd.Series, cfg: dict[str, Any]) -> pd.Series:
    if not cfg.get("exclude_money", True):
        return pd.Series(False, index=h.index)
    required = {"_公司范围", "_货币基金", "_当前净值"}
    if not required.issubset(h.columns):
        return pd.Series(False, index=h.index)
    limit = _cfg_strategy_float(cfg, "small_money_fund_limit", SMALL_EXCLUDED_MONEY_FUND_LIMIT)
    return (
        h["_公司范围"].astype(bool)
        & h["_货币基金"].astype(bool)
        & ~ytd_mask.astype(bool)
        & h["_当前净值"].abs().le(limit)
    )


def _small_excluded_money_fund_ytd_candidates(
    h: pd.DataFrame,
    ytd_mask: pd.Series,
    ytd_candidates: dict[str, pd.Series],
    cfg: dict[str, Any],
) -> tuple[dict[str, float], str | None]:
    money_mask = _small_excluded_money_fund_mask(h, ytd_mask, cfg)
    if not money_mask.any():
        return {}, None
    if "_累计损益" in h:
        effect_mask = money_mask & h["_累计损益"].abs().gt(1.0)
        if effect_mask.any():
            money_mask = effect_mask

    metrics: dict[str, float] = {}
    for name, series in ytd_candidates.items():
        adjustment = float(series.loc[money_mask].sum())
        if abs(adjustment) > 1.0:
            metrics[f"累计候选：{name}（纳入小额被剔除货币基金）"] = float(series.loc[ytd_mask | money_mask].sum())

    selected_adjustment = float(h.loc[money_mask, "_累计损益"].sum()) if "_累计损益" in h else 0.0
    if abs(selected_adjustment) <= 1.0:
        return metrics, None

    names = [
        value for value in h.loc[money_mask, "_产品名称"].map(clean_text).drop_duplicates().tolist()
        if value
    ]
    name_text = "、".join(names[:5]) + ("等" if len(names) > 5 else "")
    selected_total = float(h.loc[ytd_mask, "_累计损益"].sum() + selected_adjustment)
    warning = (
        f"年累计口径提示：{name_text}属于已剔除货币基金。若按当前累计公式纳入这些小额货币基金，"
        f"年累计为{selected_total/1e8:.3f}亿元；系统默认不静默纳入，请人工确认范围。"
    )
    return metrics, warning


KEYWORD_FIELD_MAP = {
    "产品名称": ["_产品名称"],
    "产品代码": ["_产品代码", "_六位判断代码"],
    "基金代码": ["_产品代码", "_六位判断代码"],
    "产品类别": ["_产品类别"],
    "产品类型1": ["_产品类型1"],
    "产品类型2": ["_产品类型2"],
    "报表类型": ["_报表类型"],
    "产品管理人": ["_管理人"],
    "管理人": ["_管理人"],
    "任一分类字段": ["_产品名称", "_产品类别", "_产品类型1", "_产品类型2", "_报表类型", "_管理人"],
}


def _keyword_match_series(series: pd.Series, rule: KeywordRule) -> pd.Series:
    values = series.map(clean_text)
    keyword = clean_text(rule.keyword)
    normalized_values = series.map(norm_header)
    normalized_keyword = norm_header(rule.keyword)
    if rule.match_type == "equals":
        return values.str.upper().eq(keyword.upper()) | normalized_values.str.upper().eq(normalized_keyword.upper())
    if rule.match_type == "startswith":
        return (
            values.str.upper().str.startswith(keyword.upper(), na=False)
            | normalized_values.str.upper().str.startswith(normalized_keyword.upper(), na=False)
        )
    return (
        values.str.contains(re.escape(keyword), case=False, regex=True, na=False)
        | normalized_values.str.contains(re.escape(normalized_keyword), case=False, regex=True, na=False)
    )


def _keyword_action_mask(
    h: pd.DataFrame,
    action: str,
    fallback: pd.Series,
    config: BusinessConfig | None = None,
) -> pd.Series:
    business_config = active_business_config(config)
    if not business_config or "关键词规则" not in business_config.provided_sheets:
        return fallback
    mask = pd.Series(False, index=h.index)
    for rule in business_config.enabled_keyword_rules(action):
        columns = KEYWORD_FIELD_MAP.get(rule.field_name)
        if not columns and rule.field_name in h.columns:
            columns = [rule.field_name]
        if not columns:
            continue
        rule_mask = pd.Series(False, index=h.index)
        for col in columns:
            if col in h.columns:
                rule_mask |= _keyword_match_series(h[col], rule)
        mask |= rule_mask
    return mask


def build_lookup(df: pd.DataFrame, col: str | None, normalizer=norm_code) -> dict[str, list[int]]:
    result: dict[str, list[int]] = {}
    if not col or col not in df.columns:
        return result
    for idx, value in df[col].items():
        key = normalizer(value)
        if key:
            result.setdefault(key, []).append(idx)
    return result


def build_company_name_lookup(
    df: pd.DataFrame,
    company_col: str | None,
    name_col: str | None,
    normalizer=norm_name,
) -> dict[str, list[int]]:
    result: dict[str, list[int]] = {}
    if not company_col or not name_col or company_col not in df.columns or name_col not in df.columns:
        return result
    for idx, row in df.iterrows():
        company = clean_text(row.get(company_col)).upper()
        name = normalizer(row.get(name_col))
        if company and name:
            result.setdefault(f"{company}|{name}", []).append(idx)
    return result


def choose_o32_row(o32: pd.DataFrame, indices: list[int]) -> tuple[int | None, str | None]:
    if not indices:
        return None, None
    if len(indices) == 1:
        return indices[0], None
    values = [o32.at[i, "_选定收益率"] for i in indices if pd.notna(o32.at[i, "_选定收益率"])]
    if values and max(values) - min(values) < 1e-12:
        return indices[0], "O32重复记录收益率一致，取首条"
    return None, f"O32存在{len(indices)}条重复记录且收益率不一致"


def prepare_dimension_maps(dimension: pd.DataFrame | None, fields: dict[str, str | None]) -> dict[str, dict[str, list[str]]]:
    maps = {
        "code_to_id": {},
        "code_to_code": {},
        "name_to_id": {},
        "name_to_code": {},
        "company_name_to_id": {},
        "company_name_to_code": {},
    }
    if dimension is None:
        return maps
    hc, hn, dc, oi, oc = (fields.get(k) for k in ["dim_holding_code", "dim_holding_name", "dim_company", "dim_o32_id", "dim_o32_code"])
    for _, row in dimension.iterrows():
        hcode = norm_code(row.get(hc)) if hc else ""
        hname = canonical_fund_name(row.get(hn)) if hn else ""
        company = clean_text(row.get(dc)).upper() if dc else ""
        oid = norm_code(row.get(oi)) if oi else ""
        ocode = norm_code(row.get(oc)) if oc else ""
        if hcode and oid:
            maps["code_to_id"].setdefault(hcode, []).append(oid)
        if hcode and ocode:
            maps["code_to_code"].setdefault(hcode, []).append(ocode)
        if hname and oid:
            maps["name_to_id"].setdefault(hname, []).append(oid)
        if hname and ocode:
            maps["name_to_code"].setdefault(hname, []).append(ocode)
        company_name = f"{company}|{hname}" if company and hname else ""
        if company_name and oid:
            maps["company_name_to_id"].setdefault(company_name, []).append(oid)
        if company_name and ocode:
            maps["company_name_to_code"].setdefault(company_name, []).append(ocode)
    return maps


def _unique(values: list[str]) -> str | None:
    values = list(dict.fromkeys(v for v in values if v))
    return values[0] if len(values) == 1 else None


def build_email_text(
    scale: float,
    daily_pnl: float,
    ytd_pnl: float,
    scale_decimals: int = 2,
    config: BusinessConfig | None = None,
) -> str:
    business_config = active_business_config(config)
    if business_config and business_config.email_template:
        return render_email_template(business_config.email_template, scale, daily_pnl, ytd_pnl, scale_decimals)
    scale_text = f"{scale / 1e8:.{scale_decimals}f}"
    daily_word = "盈利" if daily_pnl >= 0 else "亏损"
    ytd_word = "盈利" if ytd_pnl >= 0 else "亏损"
    daily_text = f"{abs(daily_pnl) / 1e8:.3f}"
    ytd_text = f"{abs(ytd_pnl) / 1e8:.3f}"
    ratio = abs(daily_pnl) / abs(scale) if scale else math.nan
    if daily_pnl < 0 and (math.isnan(ratio) or ratio <= 0.02):
        tail = "当日亏损金额可控，占总投资规模比例较小。"
    elif daily_pnl < 0:
        tail = "当日亏损占总投资规模比例相对较高，需持续关注市场风险。"
    else:
        tail = "当日整体实现盈利，市场风险敞口处于可控范围。"
    return (
        f"市场风险方面，兴全基金预估跟投产品的自有资金合计投资规模约{scale_text}亿元，"
        f"当日预计{daily_word}约{daily_text}亿元，预计年累计{ytd_word}约{ytd_text}亿元，{tail}"
        "信用风险方面，当日自有资金持仓信用债主体的资质较为稳定，未发现重大负面舆情。"
        "总体来看自有资金整体投资较为稳健，能应对当前的市场下跌风险。"
    )


def execute_calculation(
    holding: pd.DataFrame,
    o32: pd.DataFrame,
    dimension: pd.DataFrame | None,
    holding_fields: dict[str, str | None],
    o32_fields: dict[str, str | None],
    dimension_fields: dict[str, str | None] | None,
    cfg: dict[str, Any],
    daily_source: DailySource | None = None,
) -> CalculationResult:
    h = holding.copy().reset_index(drop=True)
    o = o32.copy().reset_index(drop=True)
    warnings: list[str] = []
    notes: list[str] = []
    business_config = _config_from_cfg(cfg)
    daily_source = daily_source or cfg.get("daily_source")
    daily_source_kind = daily_source.kind if isinstance(daily_source, DailySource) else "standard_o32"

    date_text = ""
    date_col = o32_fields.get("date")
    if date_col and date_col in o.columns:
        dates = pd.to_datetime(o[date_col], errors="coerce")
        if dates.notna().any():
            latest = dates.max().normalize()
            date_text = latest.strftime("%Y-%m-%d")
            if dates.dt.normalize().nunique() > 1:
                warnings.append("O32文件含多个统计日期，已自动保留最新日期。")
            o = o.loc[dates.dt.normalize().eq(latest)].copy().reset_index(drop=True)

    current_value, current_note, current_warnings, current_candidates = derive_current_value(
        h, holding_fields, cfg.get("current_value_mode", "自动选择最可靠口径"), business_config
    )
    warnings.extend(current_warnings)
    h["_当前净值"] = current_value
    notes.append(f"当前净值：{current_note}")

    ytd_values, ytd_note, ytd_candidates = derive_ytd(h, holding_fields, cfg.get("ytd_mode", "自动选择最可靠口径"), business_config)
    ytd_selected_name = next((name for name, series in ytd_candidates.items() if series is ytd_values), "")
    h["_累计损益"] = ytd_values
    notes.append(f"累计损益：{ytd_note}")

    def text_col(field: str) -> pd.Series:
        col = holding_fields.get(field)
        return h[col].map(clean_text) if col and col in h.columns else pd.Series([""] * len(h), index=h.index)

    h["_公司"] = text_col("company")
    h["_产品名称"] = text_col("product_name")
    h["_产品类别"] = text_col("product_category")
    h["_产品类型1"] = text_col("type1")
    h["_产品类型2"] = text_col("type2")
    h["_报表类型"] = text_col("report_type")
    h["_管理人"] = text_col("manager")
    code_cols = [holding_fields.get(k) for k in ["holding_code", "holding_fund_code", "initial_fund_code"]]
    code_cols = [c for c in code_cols if c and c in h.columns]
    product_code_col = holding_fields.get("holding_code")
    fund_code_col = holding_fields.get("holding_fund_code")
    fallback_code_cols = [holding_fields.get(k) for k in ["initial_fund_code", "holding_code"]]
    fallback_code_cols = [c for c in fallback_code_cols if c and c in h.columns]
    display_code_col = product_code_col if product_code_col and product_code_col in h.columns else (fallback_code_cols[0] if fallback_code_cols else None)
    six_digit_code_col = fund_code_col if fund_code_col and fund_code_col in h.columns else (fallback_code_cols[0] if fallback_code_cols else display_code_col)
    h["_产品代码"] = h[display_code_col].map(clean_text) if display_code_col else ""
    h["_六位判断代码"] = h[six_digit_code_col].map(norm_code) if six_digit_code_col else ""
    h["_产品代码六位"] = h["_六位判断代码"].map(_is_six_digit_code)

    company_value = clean_text(cfg.get("company_value"))
    if company_value and holding_fields.get("company"):
        h["_公司范围"] = h["_公司"].eq(company_value)
        if not h["_公司范围"].any():
            warnings.append(f"未找到公司“{company_value}”，已回退为全量。")
            h["_公司范围"] = True
    else:
        h["_公司范围"] = True

    active_col = holding_fields.get("active_flag")
    active_mode = cfg.get("active_mode", "不按Y/N筛选")
    if active_col and active_col in h.columns:
        active_values = h[active_col].map(lambda x: clean_text(x).upper())
        h["_Y范围"] = active_values.isin({"Y", "YES", "是", "1"})
        if active_mode == "不按Y/N筛选":
            notes.append(f"Y/N标识：识别到“{active_col}”，默认不筛选")
        elif active_mode == "仅当日损益保留Y/是":
            notes.append(f"Y/N标识：仅当日损益保留“{active_col}”为Y/是的记录")
        else:
            notes.append(f"Y/N标识：规模、当日损益和累计损益均仅保留“{active_col}”为Y/是的记录")
    else:
        h["_Y范围"] = True
        if active_mode != "不按Y/N筛选":
            warnings.append("已选择Y/N筛选，但未识别到有效标识列，已回退为不筛选。")
            active_mode = "不按Y/N筛选"

    blob = h[["_产品名称", "_产品类别", "_产品类型1", "_产品类型2", "_报表类型", "_管理人"]].agg("|".join, axis=1)
    direct_bond_default = h["_产品类别"].map(norm_header).eq("债券") | h["_报表类型"].map(norm_header).str.startswith("债券", na=False)
    bank_default = blob.map(lambda value: contains_any(value, ["银行理财"]))
    money_default = (
        h["_产品名称"].map(lambda value: contains_any(value, ["货币基金", "货币", "添利宝", "天添益"]))
        | h["_产品类别"].map(lambda value: contains_any(value, ["货币基金", "货币"]))
        | h["_产品类型1"].map(lambda value: contains_any(value, ["货币基金", "货币"]))
        | h["_产品类型2"].map(lambda value: contains_any(value, ["货币基金", "货币"]))
        | h["_报表类型"].map(lambda value: contains_any(value, ["货币基金", "货币"]))
    )
    h["_直接债券"] = _keyword_action_mask(h, "exclude_direct_bond", direct_bond_default, business_config)
    h["_银行理财"] = _keyword_action_mask(h, "exclude_bank", bank_default, business_config)
    h["_货币基金"] = _keyword_action_mask(h, "exclude_money", money_default, business_config)
    h["_非六位产品代码"] = ~h["_产品代码六位"]
    h["_未匹配置0"] = _keyword_action_mask(
        h,
        "zero_unmatched",
        blob.map(lambda value: contains_any(value, cfg.get("zero_keywords", []))),
        business_config,
    )

    scale_base_scope = h["_公司范围"].copy()
    daily_base_scope = h["_公司范围"].copy()
    ytd_base_scope = h["_公司范围"].copy()
    if active_mode == "仅当日损益保留Y/是":
        daily_base_scope &= h["_Y范围"]
    elif active_mode == "规模、当日损益和累计损益均仅保留Y/是":
        scale_base_scope &= h["_Y范围"]
        daily_base_scope &= h["_Y范围"]
        ytd_base_scope &= h["_Y范围"]

    def apply_core_exclusions(scope: pd.Series) -> pd.Series:
        result = scope & ~h["_直接债券"]
        if cfg.get("exclude_bank", True):
            result &= ~h["_银行理财"]
        if _exclude_non_six_product_code_enabled(cfg):
            result &= ~h["_非六位产品代码"]
        if cfg.get("exclude_money", False):
            result &= ~h["_货币基金"]
        return result

    h["_核心范围"] = apply_core_exclusions(scale_base_scope)
    h["_当日范围"] = apply_core_exclusions(daily_base_scope)
    if cfg.get("ytd_scope", "与核心投资规模一致") == "与核心投资规模一致":
        h["_累计范围"] = apply_core_exclusions(ytd_base_scope)
    else:
        h["_累计范围"] = ytd_base_scope & ~h["_直接债券"]
        if cfg.get("exclude_bank", True):
            h["_累计范围"] &= ~h["_银行理财"]
        if _exclude_non_six_product_code_enabled(cfg):
            h["_累计范围"] &= ~h["_非六位产品代码"]
        if cfg.get("exclude_money", False):
            h["_累计范围"] &= ~h["_货币基金"]
    daily_weight_mode = clean_text(cfg.get("daily_weight_mode")) or HALF_WEIGHT_DAILY_MODE
    if daily_weight_mode == FULL_WEIGHT_DAILY_MODE:
        h["_计算权重"] = 1.0
    else:
        h["_计算权重"] = np.where(_half_weight_mask(h), 0.5, 1.0)
    scope_parts = ["剔除直接债券"]
    if cfg.get("exclude_bank", True):
        scope_parts.append("剔除银行理财")
    if cfg.get("exclude_money", False):
        scope_parts.append("剔除货币基金")
    if _exclude_non_six_product_code_enabled(cfg):
        scope_parts.append("剔除基金产品代码非六位")
    notes.append("核心范围：" + "、".join(scope_parts))
    if h["_计算权重"].lt(1.0).any():
        notes.append("计算权重：产品类型2为债券类、债券型、偏债型、FOF/FOS型或FOF/FOS类的记录按50%计入当日损益；核心投资规模和累计损益剔除后全额汇总")
    else:
        notes.append("计算权重：当日损益全部产品按100%计入")

    if daily_source_kind in {"embedded_amount", "embedded_rate"}:
        source_fields = daily_source.fields if isinstance(daily_source, DailySource) else {}
        amount_col = source_fields.get("daily_pnl")
        rate_col = _embedded_rate_col(source_fields)
        embedded_daily_formula = _selected_config_formula(
            business_config,
            "daily",
            cfg.get("daily_rate_mode", ""),
        )
        if not amount_col and not rate_col and not embedded_daily_formula:
            raise ValueError("已识别为表内每日损益来源，但未找到当日收益率或当日损益金额列。")
        if daily_source_kind == "embedded_rate" and not rate_col:
            raise ValueError("已识别为表内当日收益率来源，但未找到当日收益率列。")
        if daily_source_kind == "embedded_amount" and not amount_col:
            raise ValueError("已识别为表内当日损益金额来源，但未找到当日损益金额列。")

        amount_values = numeric_series(h[amount_col]) if amount_col else pd.Series([np.nan] * len(h), index=h.index)
        amount_rate_tolerance = _cfg_strategy_float(cfg, "embedded_amount_rate_tolerance", EMBEDDED_AMOUNT_RATE_TOLERANCE)
        if rate_col:
            rate_values, rate_note = resolve_rate_scale(
                h[rate_col],
                rate_col,
                cfg.get("rate_unit_mode", "自动识别"),
                value_series=h["_当前净值"],
                amount_series=amount_values if amount_col else None,
            )
            notes.append(f"表内当日收益率：{rate_note}")
        else:
            rate_values = pd.Series([np.nan] * len(h), index=h.index)

        if embedded_daily_formula:
            notes.append(f"当日损益：配置公式“{embedded_daily_formula.name}”={embedded_daily_formula.formula}")
        elif rate_col:
            notes.append(f"当日损益：当前净值 × 表内“{rate_col}”")
        elif amount_col:
            notes.append(f"当日损益：未识别到可信表内收益率，使用表内“{amount_col}”作为金额替代")
            warnings.append("当日口径提示：本次仅识别到表内当日损益金额，作为末级替代使用，无法验证股票收益率口径。")
        if isinstance(daily_source, DailySource):
            notes.append(
                f"每日损益来源：{daily_source.label}（{daily_source.parsed.source_name} / {daily_source.parsed.sheet_name}）"
            )
            notes.extend(daily_source.notes)

        detail_rows: list[dict[str, Any]] = []
        for idx, row in h.iterrows():
            in_core = bool(row["_核心范围"])
            in_daily = bool(row["_当日范围"])
            if not bool(row["_公司范围"]):
                scope_note = "公司范围外"
            elif bool(row["_直接债券"]):
                scope_note = "剔除：直接债券"
            elif bool(row["_银行理财"]) and cfg.get("exclude_bank", True):
                scope_note = "剔除：银行理财"
            elif bool(row["_非六位产品代码"]) and _exclude_non_six_product_code_enabled(cfg):
                scope_note = "剔除：基金产品代码非六位"
            elif bool(row["_货币基金"]) and cfg.get("exclude_money", False):
                scope_note = "剔除：货币基金"
            elif active_mode == "规模、当日损益和累计损益均仅保留Y/是" and not bool(row["_Y范围"]):
                scope_note = "Y/N标识范围外"
            elif in_core and not in_daily:
                scope_note = "计入规模，不计入当日损益（Y/N标识）"
            else:
                scope_note = "计入核心范围"

            selected_rate = rate_values.at[idx] if rate_col else np.nan
            source_amount = amount_values.at[idx] if amount_col else np.nan
            formula_value = np.nan
            issue = ""
            weight = float(row["_计算权重"])
            if pd.isna(selected_rate) and not rate_col and pd.notna(source_amount) and abs(float(row["_当前净值"])) > 1e-9:
                selected_rate = float(source_amount) / float(row["_当前净值"])
            raw_rate_recalc = float(row["_当前净值"] * selected_rate) if pd.notna(selected_rate) else 0.0
            rate_recalc = raw_rate_recalc * weight if in_daily else 0.0
            amount_rate_mismatch = False
            if amount_col and rate_col and in_daily and pd.notna(source_amount) and pd.notna(selected_rate):
                row_diff = abs(float(source_amount) - raw_rate_recalc)
                row_threshold = max(1_000.0, abs(float(source_amount)) * amount_rate_tolerance)
                if row_diff > row_threshold:
                    amount_rate_mismatch = True
                    issue = (
                        f"表内金额与收益率重算差异{row_diff:,.2f}，"
                        f"超过容差{amount_rate_tolerance:.2%}"
                    )

            if embedded_daily_formula:
                formula_vars = {
                    "当前净值": pd.Series([row["_当前净值"]]),
                    "表内当日损益金额": pd.Series([source_amount]),
                    "表内当日收益率": pd.Series([selected_rate]),
                    "当日股票收益率": pd.Series([selected_rate]),
                    "股票收益率": pd.Series([selected_rate]),
                    "选定收益率": pd.Series([selected_rate]),
                }
                try:
                    formula_value = evaluate_formula(embedded_daily_formula.formula, formula_vars, pd.RangeIndex(1)).iloc[0]
                except Exception as exc:
                    raise ValueError(f"配置公式“{embedded_daily_formula.name}”无法计算：{exc}") from exc
                daily_pnl = float(formula_value) * weight if in_daily and pd.notna(formula_value) else 0.0
                match_status = "配置公式采用" if in_daily and pd.notna(formula_value) else ("待处理异常" if in_daily else "范围外")
                method = f"配置公式“{embedded_daily_formula.name}”"
            elif rate_col:
                daily_pnl = rate_recalc if in_daily and pd.notna(selected_rate) else 0.0
                match_status = "表内收益率采用" if in_daily and pd.notna(selected_rate) else ("按规则置0" if in_daily else "范围外")
                method = f"当前净值 × “{rate_col}”"
                if amount_rate_mismatch:
                    method = f"当前净值 × “{rate_col}”（与表内金额差异较大）"
            else:
                daily_pnl = float(source_amount) * weight if in_daily and pd.notna(source_amount) else 0.0
                match_status = "表内金额替代" if in_daily and pd.notna(source_amount) else ("待处理异常" if in_daily else "范围外")
                method = f"仅使用“{amount_col}”（金额替代）"

            if embedded_daily_formula and pd.isna(formula_value) and in_daily:
                match_status = "待处理异常"
                method = f"配置公式“{embedded_daily_formula.name}”计算结果为空"
                issue = method
            elif not rate_col and amount_col and pd.isna(source_amount) and in_daily:
                match_status = "待处理异常"
                method = f"“{amount_col}”为空"
                issue = method
            elif rate_col and pd.isna(selected_rate) and in_daily:
                match_status = "按规则置0"
                method = f"“{rate_col}”为空，按0处理"
                issue = method

            detail_rows.append({
                "源文件行号": row.get("_源文件行号", idx + 2),
                "公司": row["_公司"], "产品名称": row["_产品名称"], "产品代码": row["_产品代码"],
                "产品类别": row["_产品类别"], "产品类型1": row["_产品类型1"], "产品类型2": row["_产品类型2"], "报表类型": row["_报表类型"],
                "计算权重": weight, "当前净值": float(row["_当前净值"]), "累计损益": float(row["_累计损益"]),
                "是否计入核心范围": "是" if in_core else "否", "是否计入当日损益": "是" if in_daily else "否", "范围说明": scope_note,
                "O32基金编号": "", "O32基金代码": "", "O32基金名称": "",
                "匹配方式": method, "匹配状态": match_status, "异常说明": issue,
                "单位净值涨跌幅": selected_rate if rate_col else np.nan, "股票+债券收益率": np.nan, "选定收益率": selected_rate if rate_col else np.nan,
                "当日损益": daily_pnl,
                "单位净值口径当日损益": rate_recalc if in_daily and rate_col else 0.0,
                "股票+债券口径当日损益": 0.0,
                "表内当日损益金额": float(source_amount) if pd.notna(source_amount) else np.nan,
                "表内收益率重算损益": rate_recalc if in_daily and rate_col else 0.0,
                "是否在公司范围": "是" if bool(row["_公司范围"]) else "否",
                "是否计入累计损益": "是" if bool(row["_累计范围"]) else "否",
            })

        detail = pd.DataFrame(detail_rows)
        ytd_note = _apply_final_daily_ytd(h, detail, ytd_candidates, holding_fields, ytd_selected_name, ytd_note)
        _replace_ytd_note(notes, ytd_note)
        core_mask = h["_核心范围"]
        daily_mask = h["_当日范围"]
        ytd_mask = h["_累计范围"]
        weight = h["_计算权重"]
        scale_total = float(h.loc[core_mask, "_当前净值"].sum())
        daily_total = float(detail["当日损益"].sum())
        rate_daily_total = float(detail["表内收益率重算损益"].sum()) if "表内收益率重算损益" in detail else 0.0
        amount_daily_total = float(
            (numeric_series(h[amount_col]).fillna(0.0) * weight).loc[daily_mask].sum()
        ) if amount_col else 0.0
        ytd_total = float(h.loc[ytd_mask, "_累计损益"].sum())

        if amount_col and rate_col:
            diff = abs(amount_daily_total - rate_daily_total)
            if diff > max(10_000.0, abs(amount_daily_total) * amount_rate_tolerance, abs(rate_daily_total) * amount_rate_tolerance):
                warnings.append(
                    f"表内当日损益金额与收益率重算差异较大：金额口径{amount_daily_total/1e8:.3f}亿元，"
                    f"收益率重算{rate_daily_total/1e8:.3f}亿元。本次采用收益率重算口径，置信度低，请人工核对。"
                )

        ytd_candidate_totals: dict[str, float] = {}
        for name, series in ytd_candidates.items():
            ytd_candidate_totals[name] = float(series.loc[ytd_mask].sum())
        small_money_ytd_metrics, small_money_ytd_warning = _small_excluded_money_fund_ytd_candidates(
            h, ytd_mask, ytd_candidates, cfg
        )
        if small_money_ytd_warning:
            warnings.append(small_money_ytd_warning)
        success_mask = detail["匹配状态"].isin(["表内金额替代", "表内收益率采用", "配置公式采用"])
        exceptions = detail[(detail["匹配状态"].eq("待处理异常")) & (detail["是否计入当日损益"].eq("是"))].copy()
        source_file = daily_source.parsed.source_name if isinstance(daily_source, DailySource) else ""
        source_sheet = daily_source.parsed.sheet_name if isinstance(daily_source, DailySource) else ""
        statistics_date = clean_text(cfg.get("statistics_date")) or datetime.now().strftime("%Y-%m-%d")
        summary = {
            "统计日期": statistics_date,
            "自有资金合计投资规模": scale_total,
            "当日损益": daily_total,
            "预计年累计损益": ytd_total,
            "当日损益/投资规模": abs(daily_total) / abs(scale_total) if scale_total else math.nan,
            "核心范围记录数": int(core_mask.sum()),
            "当日损益范围记录数": int(daily_mask.sum()),
            "全量匹配成功": int((success_mask & detail["是否在公司范围"].eq("是")).sum()),
            "核心范围匹配成功": int((success_mask & detail["是否计入当日损益"].eq("是")).sum()),
            "按规则置0": int((detail["匹配状态"].eq("按规则置0") & detail["是否在公司范围"].eq("是")).sum()),
            "核心范围待处理异常": int(len(exceptions)),
            "当前净值口径": current_note,
            "累计损益口径": ytd_note,
            "当日收益率口径": embedded_daily_formula.name if embedded_daily_formula else ("表内当日股票收益率" if rate_col else "表内当日损益金额（金额替代）"),
            "每日损益来源": daily_source.label if isinstance(daily_source, DailySource) else "表内每日损益",
            "每日损益来源文件": source_file,
            "每日损益来源工作表": source_sheet,
            "自动识别说明": "；".join(notes),
        }
        candidate_metrics = {
            "表内当日损益金额汇总": amount_daily_total,
            "表内收益率重算当日损益": rate_daily_total,
            **{f"累计候选：{k}": v for k, v in ytd_candidate_totals.items()},
            **small_money_ytd_metrics,
            **{f"核心规模候选：{k}": float(_formula_series(h, holding_fields, business_config)[k][0].loc[core_mask].sum()) for k in current_candidates},
        }
        return CalculationResult(
            summary=summary,
            email_text=build_email_text(scale_total, daily_total, ytd_total, int(cfg.get("scale_decimals", 2)), business_config),
            detail=detail,
            exceptions=exceptions,
            warnings=warnings,
            field_notes=notes,
            candidate_metrics=candidate_metrics,
        )

    unit_col = o32_fields.get("o32_return")
    stock_col = o32_fields.get("o32_stock_return")
    bond_col = o32_fields.get("o32_bond_return")
    if unit_col:
        o["_单位净值收益率"], rate_note = resolve_rate_scale(o[unit_col], unit_col, cfg.get("rate_unit_mode", "自动识别"))
        notes.append(f"单位净值涨跌幅：{rate_note}")
    else:
        o["_单位净值收益率"] = np.nan
    if stock_col:
        o["_股票收益率"], _ = resolve_rate_scale(o[stock_col], stock_col, "自动识别")
    else:
        o["_股票收益率"] = np.nan
    if bond_col:
        o["_债券收益率"], _ = resolve_rate_scale(o[bond_col], bond_col, "自动识别")
    else:
        o["_债券收益率"] = np.nan
    o["_股票债券合计收益率"] = o["_股票收益率"].fillna(0.0) + o["_债券收益率"].fillna(0.0)
    default_rate_mode = _standard_daily_default_mode(business_config, o32_fields)
    rate_mode = cfg.get("daily_rate_mode") or default_rate_mode
    standard_daily_formula = _selected_config_formula(business_config, "daily", rate_mode)
    if standard_daily_formula:
        o["_选定收益率"] = o["_股票收益率"]
        notes.append(f"当日损益：配置公式“{standard_daily_formula.name}”={standard_daily_formula.formula}")
    else:
        rate_mode_key = norm_header(rate_mode)
        if rate_mode_key == norm_header(UNIT_NAV_DAILY_MODE) or "单位净值" in clean_text(rate_mode):
            if not unit_col:
                raise ValueError("标准O32表未识别到“单位净值涨跌幅”列，无法按该口径计算当日损益。")
            o["_选定收益率"] = o["_单位净值收益率"]
            notes.append("当日损益：当前净值 × 单位净值涨跌幅")
        elif rate_mode_key == norm_header(STOCK_BOND_DAILY_MODE) or ("股票" in clean_text(rate_mode) and "债券" in clean_text(rate_mode)):
            if not stock_col and not bond_col:
                raise ValueError("标准O32表未识别到“当日股票收益率/当日债券收益率”列，无法按股票+债券口径计算当日损益。")
            o["_选定收益率"] = o["_股票债券合计收益率"]
            notes.append("当日损益：当前净值 ×（当日股票收益率 + 当日债券收益率）")
        elif rate_mode_key == norm_header(BOND_DAILY_MODE) or ("债券" in clean_text(rate_mode) and "股票" not in clean_text(rate_mode)):
            if not bond_col:
                raise ValueError("标准O32表未识别到“当日债券收益率”列，无法按该口径计算当日损益。")
            o["_选定收益率"] = o["_债券收益率"]
            notes.append("当日损益：当前净值 × 当日债券收益率")
        else:
            if not stock_col:
                raise ValueError("标准O32表未识别到“当日股票收益率”列，无法按该口径计算当日损益。")
            rate_mode = STOCK_DAILY_MODE
            o["_选定收益率"] = o["_股票收益率"]
            notes.append("当日损益：当前净值 × 当日股票收益率")
    if isinstance(daily_source, DailySource):
        notes.append(
            f"每日损益来源：{daily_source.label}（{daily_source.parsed.source_name} / {daily_source.parsed.sheet_name}）"
        )
        notes.extend(daily_source.notes)

    id_lookup = build_lookup(o, o32_fields.get("o32_id"))
    code_lookup = build_lookup(o, o32_fields.get("o32_code"))
    name_lookup = build_lookup(o, o32_fields.get("o32_name"), norm_name)
    canonical_lookup = build_lookup(o, o32_fields.get("o32_name"), canonical_fund_name)
    company_name_lookup = build_company_name_lookup(o, o32_fields.get("o32_company"), o32_fields.get("o32_name"), norm_name)
    company_canonical_lookup = build_company_name_lookup(o, o32_fields.get("o32_company"), o32_fields.get("o32_name"), canonical_fund_name)
    dim_maps = prepare_dimension_maps(dimension, dimension_fields or {})
    selected_rate_label = "选定收益率"
    rate_mode_key = norm_header(rate_mode)
    if standard_daily_formula:
        selected_rate_label = standard_daily_formula.name
    elif rate_mode_key == norm_header(UNIT_NAV_DAILY_MODE) or "单位净值" in clean_text(rate_mode):
        selected_rate_label = "单位净值涨跌幅"
    elif rate_mode_key == norm_header(STOCK_BOND_DAILY_MODE) or ("股票" in clean_text(rate_mode) and "债券" in clean_text(rate_mode)):
        selected_rate_label = "股票+债券收益率"
    elif rate_mode_key == norm_header(BOND_DAILY_MODE) or ("债券" in clean_text(rate_mode) and "股票" not in clean_text(rate_mode)):
        selected_rate_label = "当日债券收益率"
    else:
        selected_rate_label = "当日股票收益率"
    detail_rows: list[dict[str, Any]] = []
    for idx, row in h.iterrows():
        candidates_code = [norm_code(row.get(c)) for c in code_cols]
        candidates_code = [x for x in candidates_code if x]
        pname = row["_产品名称"]
        company = clean_text(row["_公司"]).upper()
        pcanon = canonical_fund_name(pname)
        found_idx: int | None = None
        method = ""
        issue = ""
        in_core = bool(row["_核心范围"])
        in_daily = bool(row["_当日范围"])

        for lookup, label, name_key in [
            (company_name_lookup, "公司+基金名称精确匹配", norm_name(pname)),
            (company_canonical_lookup, "公司+基金名称标准化匹配", pcanon),
        ]:
            composite_key = f"{company}|{name_key}" if company and name_key else ""
            if composite_key and composite_key in lookup and found_idx is None:
                found_idx, issue = choose_o32_row(o, lookup[composite_key])
                if found_idx is not None:
                    method = label

        candidate_ids: list[str] = []
        candidate_codes: list[str] = []
        company_name_key = f"{company}|{pcanon}" if company and pcanon else ""
        if company_name_key:
            candidate_ids.extend(dim_maps["company_name_to_id"].get(company_name_key, []))
            candidate_codes.extend(dim_maps["company_name_to_code"].get(company_name_key, []))
        for code in candidates_code:
            candidate_ids.extend(dim_maps["code_to_id"].get(code, []))
            candidate_codes.extend(dim_maps["code_to_code"].get(code, []))
        candidate_ids.extend(dim_maps["name_to_id"].get(pcanon, []))
        candidate_codes.extend(dim_maps["name_to_code"].get(pcanon, []))
        for key, lookup, label in [
            (_unique(candidate_ids), id_lookup, "产品信息表→O32编号"),
            (_unique(candidate_codes), code_lookup, "产品信息表→O32代码"),
        ]:
            if key and key in lookup and found_idx is None:
                found_idx, issue = choose_o32_row(o, lookup[key])
                if found_idx is not None:
                    method = label
        if found_idx is None:
            for code in candidates_code:
                for lookup, label in [(code_lookup, "基金代码直接匹配"), (id_lookup, "O32编号直接匹配")]:
                    if code in lookup:
                        found_idx, issue = choose_o32_row(o, lookup[code])
                        if found_idx is not None:
                            method = label
                            break
                if found_idx is not None:
                    break
        if found_idx is None:
            nname = norm_name(pname)
            if nname in name_lookup:
                found_idx, issue = choose_o32_row(o, name_lookup[nname])
                if found_idx is not None:
                    method = "基金名称精确匹配"
        if found_idx is None and pcanon in canonical_lookup:
            found_idx, issue = choose_o32_row(o, canonical_lookup[pcanon])
            if found_idx is not None:
                method = "基金名称标准化匹配"
        if found_idx is None and pcanon:
            best_key, best_score = "", 0.0
            for key in canonical_lookup:
                score = SequenceMatcher(None, pcanon, key).ratio()
                if score > best_score:
                    best_key, best_score = key, score
            if best_score >= 0.94 and len(canonical_lookup.get(best_key, [])) == 1:
                found_idx = canonical_lookup[best_key][0]
                method = f"基金名称高置信匹配({best_score:.1%})"

        selected_rate = np.nan
        unit_rate = np.nan
        stock_rate = np.nan
        bond_rate = np.nan
        alt_rate = np.nan
        o32_id = ""
        o32_code = ""
        o32_name = ""
        match_status = "待处理异常"
        if found_idx is not None:
            orow = o.loc[found_idx]
            selected_rate = orow.get("_选定收益率", np.nan)
            unit_rate = orow.get("_单位净值收益率", np.nan)
            stock_rate = orow.get("_股票收益率", np.nan)
            bond_rate = orow.get("_债券收益率", np.nan)
            alt_rate = orow.get("_股票债券合计收益率", np.nan)
            o32_id = clean_text(orow.get(o32_fields.get("o32_id"))) if o32_fields.get("o32_id") else ""
            o32_code = clean_text(orow.get(o32_fields.get("o32_code"))) if o32_fields.get("o32_code") else ""
            o32_name = clean_text(orow.get(o32_fields.get("o32_name"))) if o32_fields.get("o32_name") else ""
            if pd.isna(selected_rate):
                selected_rate = 0.0
                if selected_rate_label == "当日股票收益率":
                    stock_rate = 0.0
                match_status = "按规则置0" if in_daily else "范围外"
                issue = f"匹配成功但{selected_rate_label}为空，按0处理" if in_daily else f"匹配成功但{selected_rate_label}为空"
                method = f"{method}（收益率为空按0）" if method else "收益率为空按0"
            else:
                match_status = "匹配成功"
                issue = issue or ""
        else:
            if in_daily:
                selected_rate = 0.0
                unit_rate = 0.0
                stock_rate = 0.0
                bond_rate = 0.0
                alt_rate = 0.0
                match_status = "按规则置0"
                method = "未找到唯一O32记录，按0处理"
                issue = issue or "未找到唯一O32记录"
            else:
                match_status = "范围外"
                issue = issue or "未找到唯一O32记录"

        if not bool(row["_公司范围"]):
            scope_note = "公司范围外"
        elif bool(row["_直接债券"]):
            scope_note = "剔除：直接债券"
        elif bool(row["_银行理财"]) and cfg.get("exclude_bank", True):
            scope_note = "剔除：银行理财"
        elif bool(row["_非六位产品代码"]) and _exclude_non_six_product_code_enabled(cfg):
            scope_note = "剔除：基金产品代码非六位"
        elif bool(row["_货币基金"]) and cfg.get("exclude_money", False):
            scope_note = "剔除：货币基金"
        elif active_mode == "规模、当日损益和累计损益均仅保留Y/是" and not bool(row["_Y范围"]):
            scope_note = "Y/N标识范围外"
        elif in_core and not in_daily:
            scope_note = "计入规模，不计入当日损益（Y/N标识）"
        else:
            scope_note = "计入核心范围"

        if standard_daily_formula and found_idx is not None:
            formula_vars = {
                "当前净值": pd.Series([row["_当前净值"]]),
                "单位净值涨跌幅": pd.Series([unit_rate]),
                "当日股票收益率": pd.Series([stock_rate]),
                "股票收益率": pd.Series([stock_rate]),
                "当日债券收益率": pd.Series([bond_rate]),
                "债券收益率": pd.Series([bond_rate]),
                "股票+债券收益率": pd.Series([alt_rate]),
                "选定收益率": pd.Series([selected_rate]),
            }
            try:
                formula_value = evaluate_formula(standard_daily_formula.formula, formula_vars, pd.RangeIndex(1)).iloc[0]
            except Exception as exc:
                raise ValueError(f"配置公式“{standard_daily_formula.name}”无法计算：{exc}") from exc
            if pd.isna(formula_value):
                daily_pnl = 0.0
                if in_daily:
                    match_status = "按规则置0"
                    issue = f"配置公式“{standard_daily_formula.name}”计算结果为空，按0处理"
            else:
                daily_pnl = float(formula_value) * float(row["_计算权重"]) if in_daily else 0.0
                if in_daily and match_status == "匹配成功":
                    method = f"配置公式“{standard_daily_formula.name}”"
                    if abs(float(row["_当前净值"])) > 1e-9:
                        selected_rate = float(formula_value) / float(row["_当前净值"])
        else:
            daily_pnl = float(row["_当前净值"] * selected_rate * row["_计算权重"]) if in_daily and pd.notna(selected_rate) else 0.0
        daily_unit = float(row["_当前净值"] * unit_rate * row["_计算权重"]) if in_daily and pd.notna(unit_rate) else 0.0
        daily_alt = float(row["_当前净值"] * alt_rate * row["_计算权重"]) if in_daily and pd.notna(alt_rate) else 0.0
        daily_stock = float(row["_当前净值"] * stock_rate * row["_计算权重"]) if in_daily and pd.notna(stock_rate) else 0.0
        detail_rows.append({
            "源文件行号": row.get("_源文件行号", idx + 2),
            "公司": row["_公司"], "产品名称": pname, "产品代码": row["_产品代码"],
            "产品类别": row["_产品类别"], "产品类型1": row["_产品类型1"], "产品类型2": row["_产品类型2"], "报表类型": row["_报表类型"],
            "计算权重": float(row["_计算权重"]), "当前净值": float(row["_当前净值"]), "累计损益": float(row["_累计损益"]),
            "是否计入核心范围": "是" if in_core else "否", "是否计入当日损益": "是" if in_daily else "否", "范围说明": scope_note,
            "O32基金编号": o32_id, "O32基金代码": o32_code, "O32基金名称": o32_name,
            "匹配方式": method, "匹配状态": match_status, "异常说明": issue,
            "单位净值涨跌幅": unit_rate, "当日股票收益率": stock_rate, "股票+债券收益率": alt_rate, "选定收益率": selected_rate,
            "当日损益": daily_pnl, "单位净值口径当日损益": daily_unit, "当日股票收益率口径当日损益": daily_stock, "股票+债券口径当日损益": daily_alt,
            "是否在公司范围": "是" if bool(row["_公司范围"]) else "否",
            "是否计入累计损益": "是" if bool(row["_累计范围"]) else "否",
        })

    detail = pd.DataFrame(detail_rows)
    ytd_note = _apply_final_daily_ytd(h, detail, ytd_candidates, holding_fields, ytd_selected_name, ytd_note)
    _replace_ytd_note(notes, ytd_note)
    core_mask = h["_核心范围"]
    daily_mask = h["_当日范围"]
    ytd_mask = h["_累计范围"]
    weight = h["_计算权重"]
    scale_total = float(h.loc[core_mask, "_当前净值"].sum())
    daily_total = float(detail["当日损益"].sum())
    unit_daily_total = float(detail["单位净值口径当日损益"].sum())
    stock_daily_total = float(detail["当日股票收益率口径当日损益"].sum())
    alt_daily_total = float(detail["股票+债券口径当日损益"].sum())
    ytd_total = float(h.loc[ytd_mask, "_累计损益"].sum())

    ytd_candidate_totals: dict[str, float] = {}
    for name, series in ytd_candidates.items():
        ytd_candidate_totals[name] = float(series.loc[ytd_mask].sum())
    small_money_ytd_metrics, small_money_ytd_warning = _small_excluded_money_fund_ytd_candidates(
        h, ytd_mask, ytd_candidates, cfg
    )
    if small_money_ytd_warning:
        warnings.append(small_money_ytd_warning)
    embedded_candidates: dict[str, float] = {}
    embedded_fields = detect_embedded_daily_fields(h, business_config)
    embedded_amount_col = embedded_fields.get("daily_pnl")
    embedded_rate_col = _embedded_rate_col(embedded_fields)
    if embedded_amount_col:
        embedded_candidates[f"表内金额候选：{embedded_amount_col}"] = float(
            (numeric_series(h[embedded_amount_col]).fillna(0.0) * weight).loc[daily_mask].sum()
        )
    if embedded_rate_col:
        embedded_rate, _ = resolve_rate_scale(
            h[embedded_rate_col],
            embedded_rate_col,
            "自动识别",
            value_series=h["_当前净值"],
            amount_series=numeric_series(h[embedded_amount_col]) if embedded_amount_col else None,
        )
        embedded_candidates[f"表内收益率候选：{embedded_rate_col}"] = float(
            (h["_当前净值"] * embedded_rate * weight).loc[daily_mask].fillna(0.0).sum()
        )
    source_file = daily_source.parsed.source_name if isinstance(daily_source, DailySource) else ""
    source_sheet = daily_source.parsed.sheet_name if isinstance(daily_source, DailySource) else ""
    summary = {
        "统计日期": date_text or datetime.now().strftime("%Y-%m-%d"),
        "自有资金合计投资规模": scale_total,
        "当日损益": daily_total,
        "预计年累计损益": ytd_total,
        "当日损益/投资规模": abs(daily_total) / abs(scale_total) if scale_total else math.nan,
        "核心范围记录数": int(core_mask.sum()),
        "当日损益范围记录数": int(daily_mask.sum()),
        "全量匹配成功": int((detail["匹配状态"].eq("匹配成功") & detail["是否在公司范围"].eq("是")).sum()),
        "核心范围匹配成功": int((detail["匹配状态"].eq("匹配成功") & detail["是否计入当日损益"].eq("是")).sum()),
        "按规则置0": int((detail["匹配状态"].eq("按规则置0") & detail["是否在公司范围"].eq("是")).sum()),
        "核心范围待处理异常": int((detail["匹配状态"].eq("待处理异常") & detail["是否计入当日损益"].eq("是")).sum()),
        "当前净值口径": current_note,
        "累计损益口径": ytd_note,
        "当日收益率口径": rate_mode,
        "每日损益来源": daily_source.label if isinstance(daily_source, DailySource) else "标准O32每日收益率表",
        "每日损益来源文件": source_file,
        "每日损益来源工作表": source_sheet,
        "自动识别说明": "；".join(notes),
    }
    exceptions = detail[(detail["匹配状态"].eq("待处理异常")) & (detail["是否计入当日损益"].eq("是"))].copy()
    candidate_metrics = {
        "单位净值涨跌幅口径当日损益": unit_daily_total,
        "当日股票收益率口径当日损益": stock_daily_total,
        "股票+债券收益率口径当日损益": alt_daily_total,
        **embedded_candidates,
        **{f"累计候选：{k}": v for k, v in ytd_candidate_totals.items()},
        **small_money_ytd_metrics,
        **{f"核心规模候选：{k}": float(_formula_series(h, holding_fields, business_config)[k][0].loc[core_mask].sum()) for k in current_candidates},
    }
    return CalculationResult(
        summary=summary,
        email_text=build_email_text(scale_total, daily_total, ytd_total, int(cfg.get("scale_decimals", 2)), business_config),
        detail=detail,
        exceptions=exceptions,
        warnings=warnings,
        field_notes=notes,
        candidate_metrics=candidate_metrics,
    )


def _autosize(ws, max_width: int = 42) -> None:
    for col_idx, column in enumerate(ws.columns, start=1):
        width = 10
        for cell in column:
            value = clean_text(cell.value)
            if value:
                width = max(width, min(max_width, len(value) * 1.3 + 2))
        ws.column_dimensions[get_column_letter(col_idx)].width = width


def build_config_comparison_frame(result: CalculationResult, default_result: CalculationResult | None = None) -> pd.DataFrame:
    if default_result is None:
        return pd.DataFrame([{"指标": "口径差异", "本次结果": "未上传配置，使用系统默认口径", "默认口径结果": "", "差异": ""}])
    metrics = [
        ("核心投资规模（亿元）", "自有资金合计投资规模", 1e8),
        ("当日损益（亿元）", "当日损益", 1e8),
        ("预计年累计损益（亿元）", "预计年累计损益", 1e8),
        ("当日损益/投资规模", "当日损益/投资规模", 1.0),
        ("核心范围记录数", "核心范围记录数", 1.0),
        ("按规则置0", "按规则置0", 1.0),
        ("核心范围待处理异常", "核心范围待处理异常", 1.0),
    ]
    rows: list[dict[str, Any]] = []
    for label, key, divisor in metrics:
        current = result.summary.get(key)
        default = default_result.summary.get(key)
        if isinstance(current, (int, float)) and isinstance(default, (int, float)):
            current_value = current / divisor
            default_value = default / divisor
            diff = current_value - default_value
        else:
            current_value = current
            default_value = default
            diff = ""
        rows.append({"指标": label, "本次结果": current_value, "默认口径结果": default_value, "差异": diff})
    return pd.DataFrame(rows)


def attach_config_artifacts(
    result: CalculationResult,
    config: BusinessConfig | None,
    default_result: CalculationResult | None = None,
) -> CalculationResult:
    result.config_audit = config_audit_frame(config)
    result.config_validation = config_validation_frame(config)
    result.config_comparison = build_config_comparison_frame(result, default_result)
    return result


def create_result_workbook(result: CalculationResult) -> bytes:
    output = io.BytesIO()
    with pd.ExcelWriter(output, engine="openpyxl") as writer:
        summary_rows = [
            ["统计日期", result.summary["统计日期"]],
            ["核心投资规模（亿元）", result.summary["自有资金合计投资规模"] / 1e8],
            ["当日损益（亿元）", result.summary["当日损益"] / 1e8],
            ["预计年累计损益（亿元）", result.summary["预计年累计损益"] / 1e8],
            ["当日损益/投资规模", result.summary["当日损益/投资规模"]],
            ["核心范围记录数", result.summary["核心范围记录数"]],
            ["当日损益范围记录数", result.summary["当日损益范围记录数"]],
            ["全量匹配成功", result.summary["全量匹配成功"]],
            ["按规则置0", result.summary["按规则置0"]],
            ["核心范围待处理异常", result.summary["核心范围待处理异常"]],
            ["当前净值口径", result.summary["当前净值口径"]],
            ["每日损益来源", result.summary.get("每日损益来源", "")],
            ["每日损益来源文件", result.summary.get("每日损益来源文件", "")],
            ["每日损益来源工作表", result.summary.get("每日损益来源工作表", "")],
            ["当日收益率口径", result.summary["当日收益率口径"]],
            ["累计损益口径", result.summary["累计损益口径"]],
            ["自动识别说明", result.summary.get("自动识别说明", "")],
            ["邮件正文", result.email_text],
        ]
        pd.DataFrame(summary_rows, columns=["项目", "结果"]).to_excel(writer, sheet_name="汇报结果", index=False)
        pd.DataFrame({"自动识别说明": result.field_notes}).to_excel(writer, sheet_name="自动识别说明", index=False)
        result.detail.to_excel(writer, sheet_name="必要核对明细", index=False)
        if not result.exceptions.empty:
            result.exceptions.to_excel(writer, sheet_name="待处理异常", index=False)
        pd.DataFrame([[k, v / 1e8] for k, v in result.candidate_metrics.items()], columns=["候选口径", "金额（亿元）"]).to_excel(writer, sheet_name="口径核对", index=False)
        if not result.config_audit.empty:
            result.config_audit.to_excel(writer, sheet_name="本次口径配置", index=False)
        if not result.config_comparison.empty:
            result.config_comparison.to_excel(writer, sheet_name="与默认口径差异", index=False)
        if not result.config_validation.empty:
            result.config_validation.to_excel(writer, sheet_name="配置校验结果", index=False)
    output.seek(0)
    wb = load_workbook(output)
    thin = Side(style="thin", color="D9D9D9")
    header_fill = PatternFill("solid", fgColor="E7E6E6")
    red_fill = PatternFill("solid", fgColor="FCE4D6")
    for ws in wb.worksheets:
        ws.freeze_panes = "A2"
        ws.auto_filter.ref = ws.dimensions
        for cell in ws[1]:
            cell.font = Font(bold=True)
            cell.fill = header_fill
            cell.alignment = Alignment(horizontal="center", vertical="center")
        for row in ws.iter_rows():
            for cell in row:
                cell.border = Border(bottom=thin)
                cell.alignment = Alignment(vertical="center", wrap_text=True)
        if ws.title in {"必要核对明细", "待处理异常"}:
            headers = {cell.value: cell.column for cell in ws[1]}
            status_col = headers.get("匹配状态")
            if status_col:
                for r in range(2, ws.max_row + 1):
                    if ws.cell(r, status_col).value == "待处理异常":
                        for c in range(1, ws.max_column + 1):
                            ws.cell(r, c).fill = red_fill
            for header in ["单位净值涨跌幅", "当日股票收益率", "股票+债券收益率", "选定收益率", "计算权重"]:
                if header in headers:
                    for r in range(2, ws.max_row + 1):
                        ws.cell(r, headers[header]).number_format = "0.0000%"
            for header in [
                "当前净值", "累计损益", "当日损益", "单位净值口径当日损益",
                "当日股票收益率口径当日损益", "股票+债券口径当日损益", "表内当日损益金额", "表内收益率重算损益",
            ]:
                if header in headers:
                    for r in range(2, ws.max_row + 1):
                        ws.cell(r, headers[header]).number_format = "#,##0.00;[Red]-#,##0.00"
        if ws.title == "汇报结果":
            ws.column_dimensions["A"].width = 28
            ws.column_dimensions["B"].width = 95
            for r in range(2, ws.max_row + 1):
                if ws.cell(r, 1).value == "当日损益/投资规模":
                    ws.cell(r, 2).number_format = "0.00%"
        _autosize(ws)
    final = io.BytesIO()
    wb.save(final)
    return final.getvalue()


def format_yi(value: float, decimals: int = 3) -> str:
    return f"{abs(value) / 1e8:.{decimals}f}"
