edinet-mcp

"""Financial statement diff and period comparison.

Compares financial statements for the same company across two periods,
calculating changes (増減額) and growth rates (増減率) for each line item.
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any, TypedDict, cast

import httpx
from loguru import logger

from edinet_mcp.client import EdinetAPIError

if TYPE_CHECKING:
    from edinet_mcp.client import EdinetClient
    from edinet_mcp.models import StatementData


class LineItemDiff(TypedDict):
    """Diff result for a single line item."""

    statement: str  # "income_statement", "balance_sheet", or "cash_flow_statement"
    科目: str
    period1_value: float | None
    period2_value: float | None
    増減額: float | None
    増減率: str | None  # Formatted as "+21.38%" or "-10.50%"


class DiffResult(TypedDict):
    """Complete diff result for two periods."""

    edinet_code: str
    company_name: str
    period1: str
    period2: str
    accounting_standard: str
    diffs: list[LineItemDiff]
    summary: dict[str, Any]


async def _fetch_period_statements(
    client: EdinetClient,
    edinet_code: str,
    period1: str,
    period2: str,
    doc_type: str,
) -> tuple[Any, Any]:
    """Fetch financial statements for two periods.

    Returns:
        Tuple of (stmt1, stmt2).

    Raises:
        ValueError: If either period's statement cannot be fetched.
    """
    try:
        stmt1 = await client.get_financial_statements(
            edinet_code=edinet_code,
            doc_type=doc_type,
            period=period1,
        )
    except (ValueError, EdinetAPIError, httpx.HTTPError) as e:
        raise ValueError(f"Failed to fetch {period1} statement: {e}") from e

    try:
        stmt2 = await client.get_financial_statements(
            edinet_code=edinet_code,
            doc_type=doc_type,
            period=period2,
        )
    except (ValueError, EdinetAPIError, httpx.HTTPError) as e:
        raise ValueError(f"Failed to fetch {period2} statement: {e}") from e

    return stmt1, stmt2


def _compute_changes(
    stmt1: Any,
    stmt2: Any,
) -> tuple[list[LineItemDiff], dict[str, Any]]:
    """Compare all statement sections and compute diffs with summary.

    Returns:
        Tuple of (sorted diffs list, summary dict).
    """
    diffs: list[LineItemDiff] = []

    for section, label in (
        ("income_statement", "income_statement"),
        ("balance_sheet", "balance_sheet"),
        ("cash_flow_statement", "cash_flow_statement"),
    ):
        diffs.extend(
            _compare_statement(
                getattr(stmt1, section),
                getattr(stmt2, section),
                label,
            )
        )

    diffs.sort(key=lambda x: abs(x.get("増減額") or 0), reverse=True)
    summary = _calculate_summary(diffs)
    return diffs, summary


async def diff_statements(
    client: EdinetClient,
    edinet_code: str,
    period1: str,
    period2: str,
    *,
    doc_type: str = "annual_report",
) -> DiffResult:
    """Compare financial statements for the same company across two periods.

    Fetches financial statements for both periods and calculates changes
    (増減額) and growth rates (増減率) for each line item.

    Args:
        client: EdinetClient instance.
        edinet_code: Company's EDINET code (e.g. "E02144").
        period1: First period year (e.g. "2024").
        period2: Second period year (e.g. "2025").
        doc_type: Document type label (default: "annual_report").

    Returns:
        DiffResult with detailed diffs and summary statistics.

    Raises:
        ValueError: If either period's statement cannot be fetched.
    """
    logger.info(f"Fetching statements for {edinet_code}: {period1} vs {period2}")

    stmt1, stmt2 = await _fetch_period_statements(
        client,
        edinet_code,
        period1,
        period2,
        doc_type,
    )
    diffs, summary = _compute_changes(stmt1, stmt2)

    return DiffResult(
        edinet_code=edinet_code,
        company_name=stmt2.filing.company_name,
        period1=period1,
        period2=period2,
        accounting_standard=stmt2.accounting_standard.value,
        diffs=diffs,
        summary=summary,
    )


def _compare_statement(
    stmt1: StatementData,
    stmt2: StatementData,
    statement_type: str,
) -> list[LineItemDiff]:
    """Compare two statements and return diffs for each line item."""
    diffs: list[LineItemDiff] = []

    # Get all unique labels from both statements
    labels1 = set(stmt1.labels)
    labels2 = set(stmt2.labels)
    all_labels = labels1 | labels2

    for label in all_labels:
        val1 = _extract_current_value(stmt1, label)
        val2 = _extract_current_value(stmt2, label)

        if val1 is None and val2 is None:
            continue

        change = _calculate_change(val1, val2)
        change_rate = _calculate_change_rate(val1, val2)

        diffs.append(
            LineItemDiff(
                statement=statement_type,
                科目=label,
                period1_value=val1,
                period2_value=val2,
                増減額=change,
                増減率=change_rate,
            )
        )

    return diffs


def _extract_current_value(stmt: StatementData, label: str) -> float | None:
    """Extract the '当期' value for a label from a statement."""
    try:
        item = stmt[label]
        value = item.get("当期")
        if value is None:
            return None
        return float(value)
    except (KeyError, ValueError, TypeError):
        return None


def _calculate_change(val1: float | None, val2: float | None) -> float | None:
    """Calculate absolute change between two values."""
    if val1 is None or val2 is None:
        return None
    return val2 - val1


def _calculate_change_rate(val1: float | None, val2: float | None) -> str | None:
    """Calculate percentage change between two values."""
    if val1 is None or val2 is None:
        return None
    if val1 == 0:
        return None  # Cannot calculate rate when base is zero
    rate = (val2 - val1) / abs(val1) * 100
    sign = "+" if rate >= 0 else ""
    return f"{sign}{rate:.2f}%"


def _calculate_summary(diffs: list[LineItemDiff]) -> dict[str, Any]:
    """Calculate summary statistics from diffs."""
    total_items = len(diffs)
    increased = sum(1 for d in diffs if (d.get("増減額") or 0) > 0)
    decreased = sum(1 for d in diffs if (d.get("増減額") or 0) < 0)
    unchanged = total_items - increased - decreased

    # Top 5 increases and decreases
    valid_diffs = [d for d in diffs if d.get("増減額") is not None]
    top_increases = [d for d in valid_diffs if cast("int", d["増減額"]) > 0][:5]
    top_decreases = [d for d in valid_diffs if cast("int", d["増減額"]) < 0][:5]

    return {
        "total_items": total_items,
        "increased": increased,
        "decreased": decreased,
        "unchanged": unchanged,
        "top_increases": [
            {"科目": d["科目"], "増減額": d["増減額"], "増減率": d["増減率"]}
            for d in top_increases
        ],
        "top_decreases": [
            {"科目": d["科目"], "増減額": d["増減額"], "増減率": d["増減率"]}
            for d in top_decreases
        ],
    }