core/external_scoring.py

# core/external_scoring.py
from __future__ import annotations
from typing import Dict, Any, List, Tuple
import pandas as pd

__all__ = [
    "get_external_template_df",
    "fill_missing_with_external",
    "merge_market_into_external_df",
    "score_external_from_df",
]

# ひな形（＋市場成長率系の列を追加）
_TEMPLATE_ROWS: List[Tuple[str, str]] = [
    ("経営者能力", "予実達成率_3年平均(%)"),
    ("経営者能力", "監査・内部統制の重大な不備 件数(過去3年)"),
    ("経営者能力", "重大コンプライアンス件数(過去3年)"),
    ("経営者能力", "社外取締役比率(%)"),
    ("経営者能力", "代表者の業界経験年数"),
    ("経営者能力", "現預金(円)"),
    ("経営者能力", "月商(円)"),
    ("経営者能力", "担保余力評価額(円)"),
    ("経営者能力", "倒産歴の有無(TRUE/FALSE)"),
    ("経営者能力", "倒産からの経過年数"),
    ("経営者能力", "重大事件・事故件数(過去10年)"),

    ("成長率", "売上_期3(最新期)"),
    ("成長率", "売上_期2"),
    ("成長率", "売上_期1(最古期)"),
    ("成長率", "営業利益_期3(最新期)"),
    ("成長率", "営業利益_期2"),
    ("成長率", "営業利益_期1(最古期)"),
    ("成長率", "主力商品数"),
    ("成長率", "成長中主力商品数"),
    ("成長率", "市場の年成長率(%)"),

    ("安定性", "自己資本比率(%)"),
    ("安定性", "利益剰余金(円)"),
    ("安定性", "支払遅延件数(直近12ヶ月)"),
    ("安定性", "不渡り件数(直近12ヶ月)"),
    ("安定性", "平均支払遅延日数"),
    ("安定性", "メインバンク明確か(TRUE/FALSE)"),
    ("安定性", "借入先数"),
    ("安定性", "メインバンク借入シェア(%)"),
    ("安定性", "コミットメントライン等の長期与信枠あり(TRUE/FALSE)"),
    ("安定性", "担保余力評価額(円)"),
    ("安定性", "月商(円)_再掲"),
    ("安定性", "主要顧客上位1社売上比率(%)"),
    ("安定性", "主要顧客上位3社売上比率(%)"),
    ("安定性", "主要顧客の平均信用スコア(0-100)"),
    ("安定性", "不良債権件数(直近12ヶ月)"),
    ("安定性", "業歴(年)"),

    ("公平性・総合世評", "有価証券報告書提出企業か(TRUE/FALSE)"),
    ("公平性・総合世評", "決算公告や官報での公開あり(TRUE/FALSE)"),
    ("公平性・総合世評", "HP/IRサイトで財務資料公開あり(TRUE/FALSE)"),
    ("公平性・総合世評", "直近更新が定め通りか(TRUE/FALSE)"),
]

def get_external_template_df() -> pd.DataFrame:
    return pd.DataFrame([(c, i, "") for c, i in _TEMPLATE_ROWS],
                        columns=["カテゴリー", "入力項目", "値"])

def fill_missing_with_external(df: pd.DataFrame, suggestions: Dict[str, Any] | None = None) -> pd.DataFrame:
    if not suggestions:
        return df.copy()
    df2 = df.copy()
    for idx, row in df2.iterrows():
        k = row["入力項目"]
        if (row["値"] in (None, "", "—")) and (k in suggestions):
            df2.at[idx, "値"] = suggestions[k]
    return df2

def merge_market_into_external_df(ext_df: pd.DataFrame, market: Dict[str, Any], products: List[str]) -> pd.DataFrame:
    """市場推定結果と商品リストをext_dfへ反映（必ずDataFrameを返す）"""
    df = ext_df.copy()

    def _set(df_: pd.DataFrame, label: str, val: Any, cat_hint: str = "成長率") -> pd.DataFrame:
        m = df_["入力項目"].eq(label)
        if m.any():
            df_.loc[m, "値"] = val
            return df_
        # 行がない場合は追加
        return pd.concat([df_, pd.DataFrame([[cat_hint, label, val]], columns=df_.columns)], ignore_index=True)

    if market.get("市場の年成長率(%)") is not None:
        df = _set(df, "市場の年成長率(%)", float(market["市場の年成長率(%)"]), "成長率")

    prods = [p for p in products if str(p).strip()]
    df = _set(df, "主力商品数", len(prods), "成長率")

    growing = 0
    prod_growth: Dict[str, float] = market.get("製品別年成長率(%)") or {}
    for p in prods:
        try:
            if float(prod_growth.get(p, 0.0)) > 10.0:
                growing += 1
        except Exception:
            pass
    df = _set(df, "成長中主力商品数", growing, "成長率")
    return df

# ===== スコア計算（定量化＋ばらつきストレッチ） =====
_WEIGHTS = {
    ("経営者能力", "経営姿勢"): 8,
    ("経営者能力", "事業経験"): 5,
    ("経営者能力", "資産担保力"): 6,
    ("経営者能力", "減点事項"): 7,

    ("成長率", "売上高伸長性"): 10,
    ("成長率", "利益伸長性"): 10,
    ("成長率", "商品"): 6,
    ("成長率", "市場成長調整"): 6,

    ("安定性", "自己資本"): 8,
    ("安定性", "決済振り"): 10,
    ("安定性", "金融取引"): 6,
    ("安定性", "資産担保余力"): 6,
    ("安定性", "取引先"): 6,
    ("安定性", "業歴"): 4,

    ("公平性・総合世評", "ディスクロージャー"): 8,
}
_WEIGHT_NORM = 100.0 / float(sum(_WEIGHTS.values()))

def _clamp(v, a, b): return max(a, min(b, v))
def _to_float(x):
    if x is None: return None
    try:
        return float(str(x).replace(",", "").replace("▲", "-").replace("△", "-"))
    except Exception:
        return None
def _to_bool(x):
    if x is None: return None
    s = str(x).strip().lower()
    if s in ("true","t","1","yes","y","有","あり"): return True
    if s in ("false","f","0","no","n","無","なし"): return False
    return None
def _ratio(a,b):
    if a is None or b is None or b == 0: return None
    return a/b
def _ramp(x, good, bad, lo=0.0, hi=10.0, neutral=None):
    if x is None:
        return neutral if neutral is not None else (lo+hi)/2.0
    if good > bad:
        if x <= bad: return lo
        if x >= good: return hi
        return lo + (hi-lo) * (x-bad)/(good-bad)
    else:
        if x >= bad: return lo
        if x <= good: return hi
        return lo + (hi-lo) * (x-good)/(bad-good)
def _stretch_0_10(x: float, k: float = 1.25) -> float:
    if x is None: return None
    t = (x/10.0)
    t = t**(1.0/k) if t >= 0.5 else (t**k)
    return _clamp(t*10.0, 0.0, 10.0)
def _add(items, cat, name, raw, weight, reason):
    raw2 = _stretch_0_10(raw, k=1.25) if raw is not None else None
    w = round(weight * _WEIGHT_NORM, 2)
    sc = 0.0 if raw2 is None else round((raw2 / 10.0) * w, 2)
    items.append({
        "category": cat, "name": name, "raw": None if raw is None else round(raw,2),
        "raw_stretched": None if raw2 is None else round(raw2,2),
        "weight": w, "score": sc, "reason": reason
    })

def score_external_from_df(df: pd.DataFrame) -> Dict[str, Any]:
    # 必ず dict を返す。途中で例外にならないよう to_x で吸収。
    def ref(label: str):
        m = df["入力項目"].eq(label)
        return df.loc[m, "値"].values[0] if m.any() else None

    items: List[Dict[str, Any]] = []

    yoy3 = _to_float(ref("予実達成率_3年平均(%)"))
    audit_bad = _to_float(ref("監査・内部統制の重大な不備 件数(過去3年)"))
    comp_bad = _to_float(ref("重大コンプライアンス件数(過去3年)"))
    indep = _to_float(ref("社外取締役比率(%)"))
    exp_years = _to_float(ref("代表者の業界経験年数"))
    cash = _to_float(ref("現預金(円)"))
    sales_m = _to_float(ref("月商(円)"))
    collat = _to_float(ref("担保余力評価額(円)"))
    has_bk = _to_bool(ref("倒産歴の有無(TRUE/FALSE)"))
    bk_years = _to_float(ref("倒産からの経過年数"))
    incidents = _to_float(ref("重大事件・事故件数(過去10年)"))

    s1 = _to_float(ref("売上_期1(最古期)"))
    s2 = _to_float(ref("売上_期2"))
    s3 = _to_float(ref("売上_期3(最新期)"))
    p1 = _to_float(ref("営業利益_期1(最古期)"))
    p2 = _to_float(ref("営業利益_期2"))
    p3 = _to_float(ref("営業利益_期3(最新期)"))

    equity = _to_float(ref("自己資本比率(%)"))
    delay_cnt = _to_float(ref("支払遅延件数(直近12ヶ月)"))
    boun_cnt = _to_float(ref("不渡り件数(直近12ヶ月)"))
    delay_days = _to_float(ref("平均支払遅延日数"))
    mainbank = _to_bool(ref("メインバンク明確か(TRUE/FALSE)"))
    lenders = _to_float(ref("借入先数"))
    main_share = _to_float(ref("メインバンク借入シェア(%)"))
    has_line = _to_bool(ref("コミットメントライン等の長期与信枠あり(TRUE/FALSE)"))
    sales_m2 = _to_float(ref("月商(円)_再掲")) or sales_m
    top1 = _to_float(ref("主要顧客上位1社売上比率(%)"))
    top3 = _to_float(ref("主要顧客上位3社売上比率(%)"))
    cust_score = _to_float(ref("主要顧客の平均信用スコア(0-100)"))
    npl_cnt = _to_float(ref("不良債権件数(直近12ヶ月)"))
    years = _to_float(ref("業歴(年)"))

    prod_total = _to_float(ref("主力商品数"))
    prod_growing = _to_float(ref("成長中主力商品数"))
    market_growth = _to_float(ref("市場の年成長率(%)"))

    cash_to_ms = _ratio(cash, sales_m2)
    coll_to_ms = _ratio(collat, sales_m2)

    def cagr(v1, v3):
        if v1 is None or v3 is None or v1 <= 0: return None
        try: return (v3/v1)**(1/2) - 1.0
        except Exception: return None
    s_cagr = cagr(s1, s3)
    p_cagr = cagr(p1, p3)

    # 経営者能力
    mg_att = (_ramp(yoy3, 90, 50) +
              _ramp(0 if not audit_bad else -audit_bad, 0, -3) +
              _ramp(0 if not comp_bad else -comp_bad, 0, -2) +
              _ramp(indep, 33, 0)) / 4
    _add(items, "経営者能力", "経営姿勢", mg_att, _WEIGHTS[("経営者能力","経営姿勢")],
         f"予実{yoy3 or '—'}%/監査{audit_bad or 0}/違反{comp_bad or 0}/社外{indep or '—'}%")

    mg_exp = _ramp(exp_years if exp_years is not None else 5.0, 15, 0)
    _add(items, "経営者能力", "事業経験", mg_exp, _WEIGHTS[("経営者能力","事業経験")],
         f"経験{exp_years if exp_years is not None else '不明→中立'}年")

    mg_asset = _ramp(cash_to_ms, 1.5, 0.2)
    _add(items, "経営者能力", "資産担保力", mg_asset, _WEIGHTS[("経営者能力","資産担保力")],
         f"現預金/月商≈{round(cash_to_ms,2) if cash_to_ms else '—'}")

    if incidents and incidents>0:
        pen=0.0; rs=f"重大事故{int(incidents)}件→大幅減点"
    elif has_bk:
        pen=6.0 if (bk_years and bk_years>=10) else 3.0; rs=f"倒産歴あり（{bk_years or '不明'}年）"
    else:
        pen=10.0; rs="事故/倒産なし"
    _add(items,"経営者能力","減点事項",pen,_WEIGHTS[("経営者能力","減点事項")],rs)

    # 成長率
    _add(items,"成長率","売上高伸長性", _ramp(s_cagr,0.08,-0.05),
         _WEIGHTS[("成長率","売上高伸長性")],
         f"CAGR売上{round((s_cagr or 0)*100,1) if s_cagr is not None else '—'}%")
    _add(items,"成長率","利益伸長性", _ramp(p_cagr,0.08,-0.05),
         _WEIGHTS[("成長率","利益伸長性")],
         f"CAGR営業{round((p_cagr or 0)*100,1) if p_cagr is not None else '—'}%")

    # 商品
    if prod_total is None or prod_total <= 0:
        pr_sc = 5.0; rs = "不明→中立"
    else:
        ratio = _ratio(prod_growing, prod_total) or 0.0
        pr_sc = ( _ramp(prod_total, 3, 0) + _ramp(ratio, 0.7, 0.1) ) / 2
        rs = f"主力{int(prod_total)}/成長中比{round(ratio*100,1)}%"
    _add(items,"成長率","商品", pr_sc, _WEIGHTS[("成長率","商品")], rs)

    # 市場成長調整
    _add(items,"成長率","市場成長調整",
         _ramp(market_growth,15,-5),
         _WEIGHTS[("成長率","市場成長調整")],
         f"市場年成長{market_growth or '—'}%")

    # 安定性
    _add(items,"安定性","自己資本", _ramp(equity,40,5),
         _WEIGHTS[("安定性","自己資本")], f"自己資本比率{equity or '—'}%")

    if (delay_cnt is not None) or (boun_cnt is not None) or (delay_days is not None):
        sc=( _ramp(-(delay_cnt or 0),0,-6) +
             _ramp(-(boun_cnt or 0),0,-1) +
             _ramp(-(delay_days or 0),0,-30) )/3
        rs=f"遅延{int(delay_cnt or 0)}/不渡{int(boun_cnt or 0)}/平均{int(delay_days or 0)}日"
    else:
        sc=_ramp(_ratio(cash, sales_m2),1.0,0.2); rs=f"代理：現預金/月商≈—"
    _add(items,"安定性","決済振り", sc, _WEIGHTS[("安定性","決済振り")], rs)

    sc_mb = 5.0
    sc_mb += 2.0 if mainbank else (-0.5 if mainbank is False else 0)
    sc_mb += 1.0 if has_line else 0
    sc_mb = _clamp(sc_mb,0,10)
    _add(items,"安定性","金融取引", sc_mb, _WEIGHTS[("安定性","金融取引")],
         f"メイン{'有' if mainbank else '無' if mainbank is False else '—'}/与信枠{'有' if has_line else '無' if has_line is False else '—'}")

    _add(items,"安定性","資産担保余力", _ramp(_ratio(collat, sales_m2),4.0,0.0),
         _WEIGHTS[("安定性","資産担保余力")], f"担保/月商≈—")

    _add(items,"安定性","取引先",
         ( _ramp(-(top1 or 50),0,-80) +
           _ramp(cust_score,80,50) +
           _ramp(-(npl_cnt or 1),0,-3) )/3,
         _WEIGHTS[("安定性","取引先")],
         f"上位1社{top1 or '—'}%/信用{cust_score or '—'}/不良{int(npl_cnt or 0)}")

    _add(items,"安定性","業歴", _ramp(years,20,1),
         _WEIGHTS[("安定性","業歴")], f"{years or '—'}年")

    # 公平性
    sc_dis = 0.0
    has_sec = _to_bool(ref("有価証券報告書提出企業か(TRUE/FALSE)"))
    sc_dis += 10.0 if has_sec else 0.0
    if sc_dis == 0.0:
        pub_off = _to_bool(ref("決算公告や官報での公開あり(TRUE/FALSE)"))
        pub_web = _to_bool(ref("HP/IRサイトで財務資料公開あり(TRUE/FALSE)"))
        sc_dis += 7.0 if (pub_off or pub_web) else 4.0
    upd_on = _to_bool(ref("直近更新が定め通りか(TRUE/FALSE)"))
    if upd_on: sc_dis += 1.0
    sc_dis = _clamp(sc_dis,0,10)
    _add(items,"公平性・総合世評","ディスクロージャー", sc_dis,
         _WEIGHTS[("公平性・総合世評","ディスクロージャー")],
         f"{'有報' if has_sec else '公開あり' if sc_dis>=7.0 else '公開乏しい'} / 更新{'◯' if upd_on else '—'}")

    total = round(sum(x["score"] for x in items),1)

    from collections import defaultdict
    cat_sum, cat_w = defaultdict(float), defaultdict(float)
    for it in items:
        cat_sum[it["category"]] += it["score"]
        cat_w[it["category"]] += it["weight"]
    cat_scores = {c: round((cat_sum[c] / cat_w[c]) * 100.0 if cat_w[c] > 0 else 0.0, 1) for c in cat_sum}

    return {
        "name": "企業評価（外部・定量）",
        "external_total": total,
        "items": items,
        "category_scores": cat_scores,
        "notes": "欠損は中立＋市場成長/商品構成を反映。ストレッチでばらつきを拡大。",
    }