finance_core.py

# -*- coding: utf-8 -*-
from __future__ import annotations
from typing import Dict, Any, Optional, List, Tuple

from schemas import FinancialExtract, ExtractedPeriod, MultipleSuggestion, MarketOutlook

def _fmt_currency(x, currency: str) -> str:
    if x is None:
        return "—"
    try:
        absx = abs(x)
        if absx >= 1e12: return f"{x/1e12:.2f}兆{currency}"
        if absx >= 1e8:  return f"{x/1e8:.2f}億{currency}"
        if absx >= 1e6:  return f"{x/1e6:.2f}百万{currency}"
        return f"{x:,.0f}{currency}"
    except Exception:
        return f"{x} {currency}"

def compute_ratios(extract: FinancialExtract) -> Dict[str, Any]:
    latest: Optional[ExtractedPeriod] = extract.latest()
    prev: Optional[ExtractedPeriod] = extract.previous()
    currency = extract.currency or ""
    ratios: Dict[str, Any] = {}

    if latest:
        revenue = latest.revenue
        cogs = latest.cogs
        ebit = latest.ebit
        ebitda = latest.ebitda if latest.ebitda is not None else (latest.ebit + (latest.depreciation or 0.0) if latest.ebit is not None else None)
        net_income = latest.net_income
        cash = latest.cash_and_equivalents
        ar = latest.accounts_receivable
        inv = latest.inventory
        ap = latest.accounts_payable
        ca = latest.current_assets
        cl = latest.current_liabilities
        te = latest.total_equity
        td = latest.total_debt
        interest = latest.interest_expense
        ta = latest.total_assets

        gross_profit = (revenue - cogs) if revenue is not None and cogs is not None else None
        gross_margin = (gross_profit / revenue) if revenue and gross_profit is not None else None
        ebit_margin = (ebit / revenue) if revenue and ebit is not None else None
        ebitda_margin = (ebitda / revenue) if revenue and ebitda is not None else None

        current_ratio = (ca / cl) if ca and cl else None
        quick_assets = (cash or 0.0) + (ar or 0.0)
        quick_ratio = (quick_assets / cl) if quick_assets and cl else None

        debt_to_equity = (td / te) if td is not None and te else None
        interest_coverage = (ebit / interest) if ebit is not None and interest else None

        # 追加KPI
        roe = (net_income / te) if net_income is not None and te else None
        roa = (net_income / ta) if net_income is not None and ta else None
        cash_ratio = (cash / cl) if cash is not None and cl else None
        net_debt = (td - (cash or 0.0)) if td is not None else None

        dso = ( (ar / (revenue/365.0)) if ar is not None and revenue else None )
        dio = ( (inv / ( (cogs or 0.0)/365.0)) if inv is not None and cogs else None )
        dpo = ( (ap / ( (cogs or 0.0)/365.0)) if ap is not None and cogs else None )
        ccc = ( (dso or 0) + (dio or 0) - (dpo or 0) ) if (dso is not None and dio is not None and dpo is not None) else None

        ratios.update({
            "revenue": _fmt_currency(revenue, currency),
            "ebit": _fmt_currency(ebit, currency),
            "ebitda": _fmt_currency(ebitda, currency),
            "net_income": _fmt_currency(net_income, currency),
            "gross_margin_pct": f"{gross_margin*100:.1f}%" if gross_margin is not None else "—",
            "ebit_margin_pct": f"{ebit_margin*100:.1f}%" if ebit_margin is not None else "—",
            "ebitda_margin_pct": f"{ebitda_margin*100:.1f}%" if ebitda_margin is not None else "—",
            "current_ratio": f"{current_ratio:.2f}" if current_ratio is not None else "—",
            "quick_ratio": f"{quick_ratio:.2f}" if quick_ratio is not None else "—",
            "debt_to_equity": f"{debt_to_equity:.2f}" if debt_to_equity is not None else "—",
            "interest_coverage": f"{interest_coverage:.2f}" if interest_coverage is not None else "—",
            # 追加表示
            "roe_pct": f"{roe*100:.1f}%" if roe is not None else "—",
            "roa_pct": f"{roa*100:.1f}%" if roa is not None else "—",
            "cash_ratio": f"{cash_ratio:.2f}" if cash_ratio is not None else "—",
            "net_debt": _fmt_currency(net_debt, currency) if net_debt is not None else "—",
            "dso_days": f"{dso:.0f}日" if dso is not None else "—",
            "dio_days": f"{dio:.0f}日" if dio is not None else "—",
            "dpo_days": f"{dpo:.0f}日" if dpo is not None else "—",
            "ccc_days": f"{ccc:.0f}日" if ccc is not None else "—",
        })

    if latest and prev and latest.revenue and prev.revenue and prev.revenue != 0:
        growth = (latest.revenue - prev.revenue) / abs(prev.revenue)
        ratios["revenue_growth_pct"] = f"{growth*100:.1f}%"
        ratios["_revenue_growth_float"] = growth
    else:
        ratios["revenue_growth_pct"] = "—"
        ratios["_revenue_growth_float"] = None

    ratios["_latest_currency"] = extract.currency or "JPY"
    ratios["_latest_ebitda"] = (latest.ebitda if latest and latest.ebitda is not None else
                                (latest.ebit + (latest.depreciation or 0.0) if latest and latest.ebit is not None else None))
    ratios["_latest"] = latest.dict() if latest else {}
    return ratios

def _risk_score(latest: ExtractedPeriod, ratios: Dict[str, Any], policies: Dict[str, Any]) -> float:
    score = 50.0
    ca, cl = latest.current_assets, latest.current_liabilities
    if ca and cl:
        cr = ca / cl
        if cr >= 2.0: score += 10
        elif cr >= 1.5: score += 6
        elif cr >= 1.2: score += 3
        elif cr < 1.0: score -= 10

    td, te = latest.total_debt, latest.total_equity
    if td is not None and te:
        de = td / te
        if de <= 0.5: score += 8
        elif de <= 1.0: score += 4
        elif de > 2.0: score -= 8

    ebit, interest = latest.ebit, latest.interest_expense
    if ebit is not None and interest:
        ic = ebit / interest
        if ic >= 6: score += 10
        elif ic >= 3: score += 6
        elif ic >= 1.5: score += 3
        elif ic < 1.0: score -= 12

    revenue = latest.revenue
    net = latest.net_income
    if revenue and net is not None:
        nm = net / revenue
        if nm >= 0.1: score += 6
        elif nm >= 0.03: score += 3
        elif nm < 0: score -= 8

    g = ratios.get("_revenue_growth_float")
    if g is not None:
        if g >= 0.2: score += 6
        elif g >= 0.05: score += 3
        elif g < 0: score -= 5
    return max(0.0, min(100.0, score))

def _rating(score: float) -> str:
    if score >= 85: return "A"
    if score >= 70: return "B"
    if score >= 55: return "C"
    if score >= 40: return "D"
    return "E"

def credit_decision(extract: FinancialExtract, ratios: Dict[str, Any], policies: Dict[str, Any]) -> Dict[str, Any]:
    latest = extract.latest()
    currency = extract.currency or "JPY"
    if not latest:
        return {"rating": "E", "risk_score": 0, "site_days": 0,
                "transaction_limit": 0.0, "transaction_limit_display": _fmt_currency(0.0, currency),
                "review_cycle": "取引前入金（与信不可）"}

    score = _risk_score(latest, ratios, policies)
    rating = _rating(score)

    site_map = {"A": 90, "B": 60, "C": 45, "D": 30, "E": 0}
    review_map = {"A": "年1回", "B": "半年毎", "C": "四半期毎", "D": "月次", "E": "取引ごと"}

    revenue = latest.revenue or 0.0
    working_capital = (latest.current_assets or 0.0) - (latest.current_liabilities or 0.0)
    try:
        cr_val = float(ratios.get("current_ratio", "0").replace("—","0"))
    except Exception:
        cr_val = 0.0
    liquidity_factor = 0.4 + 0.6 * (min(1.0, max(0.0, cr_val / 2.0)))
    risk_factor = 0.5 + 0.5 * (score / 100.0)

    site_days = site_map[rating]
    base_limit = (revenue / 12.0) * (site_days / 30.0) * 0.6
    cap = max(0.0, working_capital) * 0.4
    tx_limit = min(cap, base_limit * liquidity_factor * risk_factor)
    tx_limit = max(0.0, tx_limit)

    return {"rating": rating, "risk_score": round(score, 1), "site_days": site_days,
            "transaction_limit": tx_limit, "transaction_limit_display": _fmt_currency(tx_limit, currency),
            "review_cycle": review_map[rating]}

def _annuity_factor(rate: float, years: int) -> float:
    r = rate / 100.0
    n = max(1, years)
    if r <= 0: return 1.0 / n
    return r / (1 - (1 + r) ** (-n))

def loan_decision(extract: FinancialExtract, ratios: Dict[str, Any], base_rate_pct: float, policies: Dict[str, Any]) -> Dict[str, Any]:
    latest = extract.latest()
    currency = extract.currency or "JPY"
    if not latest:
        return {"max_principal": 0.0, "max_principal_display": _fmt_currency(0.0, currency),
                "term_years": 0, "interest_rate_pct": base_rate_pct, "target_dscr": 0.0}

    score = _risk_score(latest, ratios, policies)
    rating = _rating(score)

    rp = {"A": 0.5, "B": 1.0, "C": 2.0, "D": 3.5, "E": 0.0}[rating]
    if rating == "E":
        return {"max_principal": 0.0, "max_principal_display": _fmt_currency(0.0, currency),
                "term_years": 0, "interest_rate_pct": 0.0, "target_dscr": 0.0}

    interest_rate_pct = base_rate_pct + rp
    term_years = {"A": 5, "B": 4, "C": 3, "D": 1}[rating]
    target_dscr = {"A": 1.6, "B": 1.4, "C": 1.3, "D": 1.2}[rating]

    ebitda = ratios.get("_latest_ebitda")
    if not ebitda or ebitda <= 0:
        return {"max_principal": 0.0, "max_principal_display": _fmt_currency(0.0, currency),
                "term_years": term_years, "interest_rate_pct": interest_rate_pct, "target_dscr": target_dscr}

    ann = _annuity_factor(interest_rate_pct, term_years)
    annual_debt_service_cap = ebitda / target_dscr
    max_principal = max(0.0, annual_debt_service_cap / ann)

    ta = latest.total_assets or 0.0
    cash = latest.cash_and_equivalents or 0.0
    nta_cap = max(0.0, (ta - cash) * 0.5)
    max_principal = min(max_principal, nta_cap)

    return {"max_principal": max_principal, "max_principal_display": _fmt_currency(max_principal, currency),
            "term_years": term_years, "interest_rate_pct": round(interest_rate_pct, 2), "target_dscr": target_dscr}

def _market_factor_from_score(score: int) -> Tuple[float, List[float]]:
    # score別の投資シナリオ倍率（保守/基準/強気）
    table = {
        1: (0.85, [0.75, 0.90, 1.05]),
        2: (0.93, [0.85, 0.95, 1.10]),
        3: (1.00, [0.90, 1.00, 1.15]),
        4: (1.07, [0.95, 1.05, 1.25]),
        5: (1.15, [1.00, 1.10, 1.35]),
    }
    return table.get(int(score) if score else 3, (1.0, [0.9, 1.0, 1.15]))

def investment_decision(
    extract: FinancialExtract,
    ratios: Dict[str, Any],
    policies: Dict[str, Any],
    multiples: Optional[MultipleSuggestion],
    market_outlook: Optional[MarketOutlook] = None,
) -> Dict[str, Any]:
    latest = extract.latest()
    currency = extract.currency or "JPY"
    if not latest:
        return {"ev": 0.0, "ev_display": _fmt_currency(0.0, currency),
                "market_cap": 0.0, "market_cap_display": _fmt_currency(0.0, currency),
                "recommended_check_size": 0.0, "recommended_check_size_display": _fmt_currency(0.0, currency),
                "attractiveness": 1, "growth_label": "Low"}

    g = ratios.get("_revenue_growth_float")
    if g is None: glabel = "Unknown"
    elif g >= 0.25: glabel = "High"
    elif g >= 0.05: glabel = "Medium"
    else: glabel = "Low"

    ebitda = latest.ebitda if latest.ebitda is not None else (latest.ebit + (latest.depreciation or 0.0) if latest.ebit is not None else None)
    revenue = latest.revenue or 0.0
    total_debt = latest.total_debt or 0.0
    cash = latest.cash_and_equivalents or 0.0
    net_debt = max(0.0, total_debt - cash)

    if multiples and ebitda and ebitda > 0:
        ev = ebitda * multiples.ebitda_multiple
    elif ebitda and ebitda > 0:
        ev = ebitda * 8.0
    else:
        rev_mult = multiples.revenue_multiple if multiples else 1.5
        ev = revenue * rev_mult

    market_cap = max(0.0, ev - net_debt)

    score = _risk_score(latest, ratios, policies)
    rating = _rating(score)
    pct = {"A": 0.15, "B": 0.12, "C": 0.08, "D": 0.04, "E": 0.0}[rating]
    base_check = market_cap * pct

    attractiveness = 1
    if rating == "A": attractiveness = 5 if glabel == "High" else 4
    elif rating == "B": attractiveness = 4 if glabel != "Low" else 3
    elif rating == "C": attractiveness = 3 if glabel == "High" else 2
    elif rating == "D": attractiveness = 1

    # 市場期待の反映（基本倍率＋シナリオ）
    market_factor, scenarios = (1.0, [0.9, 1.0, 1.15])
    if market_outlook:
        market_factor, scenarios = _market_factor_from_score(market_outlook.expectation_score)
        if market_outlook.expectation_score >= 4:
            attractiveness = min(5, attractiveness + 1)
        elif market_outlook.expectation_score <= 2:
            attractiveness = max(1, attractiveness - 1)

    check = base_check * market_factor

    # シナリオテーブル
    scenario_rows = []
    for label, fac in zip(["保守", "基準", "強気"], scenarios):
        scenario_rows.append({
            "label": label,
            "factor": fac,
            "check_size": base_check * fac,
            "check_size_display": _fmt_currency(base_check * fac, currency),
        })

    return {
        "ev": ev, "ev_display": _fmt_currency(ev, currency),
        "market_cap": market_cap, "market_cap_display": _fmt_currency(market_cap, currency),
        "recommended_check_size": check, "recommended_check_size_display": _fmt_currency(check, currency),
        "attractiveness": attractiveness, "growth_label": glabel,
        "market_factor": round(market_factor, 2),
        "scenarios": scenario_rows,
    }

def build_report_dict(
    extract: FinancialExtract,
    ratios: Dict[str, Any],
    decisions: Dict[str, Any],
    unit_info: Optional[Dict[str, Any]] = None,
    market_outlook: Optional[MarketOutlook] = None,
) -> Dict[str, Any]:
    out = {
        "metadata": {
            "company_name": extract.company_name,
            "industry": extract.industry,
            "currency": extract.currency,
            "fiscal_year_end": extract.fiscal_year_end
        },
        "extracted": extract.dict(),
        "ratios": ratios,
        "decisions": decisions,
        "disclaimer": "本ツールはAIによる推定・一般的な計算式に基づく参考提案であり、投資勧誘・融資約定・与信保証を目的としたものではありません。最終判断は自己責任で、必要に応じて専門家の確認を行ってください。",
    }
    if unit_info: out["unit_detection"] = unit_info
    if market_outlook: out["market_outlook"] = market_outlook.model_dump()
    return out