Update core/external_scoring.py

core/external_scoring.py

@@ -4,7 +4,11 @@ from typing import Dict, Any, List, Tuple
 import pandas as pd
 import math
 
-__all__ = ["get_external_template_df", "fill_missing_with_external", "score_external_from_df"]
+__all__ = [
+    "get_external_template_df",
+    "fill_missing_with_external",
+    "score_external_from_df",
+]
 
 _TEMPLATE_ROWS: List[Tuple[str, str]] = [
     ("経営者能力", "予実達成率_3年平均(%)"),
@@ -27,6 +31,7 @@ _TEMPLATE_ROWS: List[Tuple[str, str]] = [
     ("成長率", "営業利益_期1(最古期)"),
     ("成長率", "主力商品数"),
     ("成長率", "成長中主力商品数"),
+    ("成長率", "市場の年成長率(%)"),
 
     ("安定性", "自己資本比率(%)"),
     ("安定性", "利益剰余金(円)"),
@@ -52,13 +57,23 @@ _TEMPLATE_ROWS: List[Tuple[str, str]] = [
 ]
 
 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, company: str = "", country: str = "") -> pd.DataFrame:
-    # 将来: 外部DBと突合。今はスルー。
-    return df.copy()
-
-# ===== スコア計算（堅牢化） =====
+    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:
+    """
+    LLM等からの suggestions を {入力項目: 値} で受け取り、空欄のみ埋める
+    """
+    if not suggestions:
+        return df.copy()
+    df2 = df.copy()
+    for idx, row in df2.iterrows():
+        key = row["入力項目"]
+        if (row["値"] in (None, "", "—")) and (key in suggestions):
+            df2.at[idx, "値"] = suggestions[key]
+    return df2
+
+# ===== スコア計算（定量的・ルールベース） =====
 _WEIGHTS = {
     ("経営者能力", "経営姿勢"): 8,
     ("経営者能力", "事業経験"): 5,
@@ -68,6 +83,7 @@ _WEIGHTS = {
     ("成長率", "売上高伸長性"): 10,
     ("成長率", "利益伸長性"): 10,
     ("成長率", "商品"): 6,
+    ("成長率", "市場成長調整"): 6,  # ★ 追加：市場成長率を反映
 
     ("安定性", "自己資本"): 8,
     ("安定性", "決済振り"): 10,
@@ -82,46 +98,64 @@ _WEIGHT_NORM = 100.0 / float(sum(_WEIGHTS.values()))
 
 def _clamp(v, a, b): return max(a, min(b, v))
 
+def _add(items, cat, name, raw, weight, reason):
+    raw_s = None if raw is None else round(raw, 2)
+    w = round(weight * _WEIGHT_NORM, 2)
+    sc = 0.0 if raw is None else round((raw / 10.0) * w, 2)
+    items.append({
+        "category": cat,
+        "name": name,
+        "raw": raw_s,
+        "weight": w,
+        "score": sc,
+        "reason": reason
+    })
+
 def _to_float(x):
-    if x is None: return None
+    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
+    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
+    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):
-    a = _to_float(a); b = _to_float(b)
-    if a is None or b in (None, 0): return None
-    try:
-        return a/b
-    except Exception:
+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
+        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)
+        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)
+        return lo + (hi - lo) * (x - good) / (bad - good)
 
 def score_external_from_df(df: pd.DataFrame) -> Dict[str, Any]:
+    """
+    入力DF（カテゴリー/入力項目/値）を定量スコア化。
+    欠損は中立。市場の年成長率(%) を「市場成長調整」に反映し、成長率評価の過剰/過小を補正。
+    """
     def ref(label: str):
-        m = df["item"].eq(label) if "item" in df.columns else df["入力項目"].eq(label)
-        return df.loc[m, "value" if "value" in df.columns else "値"].values[0] if m.any() else None
+        m = df["入力項目"].eq(label)
+        return df.loc[m, "値"].values[0] if m.any() else None
 
-    items: List[Dict[str, Any]] = []
+    items = []
 
     yoy3 = _to_float(ref("予実達成率_3年平均(%)"))
     audit_bad = _to_float(ref("監査・内部統制の重大な不備 件数(過去3年)"))
@@ -135,95 +169,156 @@ def score_external_from_df(df: pd.DataFrame) -> Dict[str, Any]:
     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(最新期)"))
+    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社売上比率(%)"))
-    cust_score=_to_float(ref("主要顧客の平均信用スコア(0-100)"))
-    npl_cnt=_to_float(ref("不良債権件数(直近12ヶ月)"))
-    years=_to_float(ref("業歴(年)"))
-    has_sec=_to_bool(ref("有価証券報告書提出企業か(TRUE/FALSE)"))
-    pub_off=_to_bool(ref("決算公告や官報での公開あり(TRUE/FALSE)"))
-    pub_web=_to_bool(ref("HP/IRサイトで財務資料公開あり(TRUE/FALSE)"))
-    upd_on=_to_bool(ref("直近更新が定め通りか(TRUE/FALSE)"))
+    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
+        if v1 is None or v3 is None or v1 <= 0:
+            return None
         try:
-            return (v3/v1)**(1/2) - 1.0
+            return (v3 / v1) ** (1 / 2) - 1.0
         except Exception:
             return None
 
-    s_cagr = cagr(s1, s3); p_cagr = cagr(p1, p3)
-
-    def _add(cat, name, raw, weight, reason):
-        items.append({
-            "category": cat,
-            "name": name,
-            "raw": round(raw,2) if raw is not None else None,
-            "weight": round(weight*_WEIGHT_NORM,2),
-            "score": round(((raw if raw is not None else 5.0)/10.0)*weight*_WEIGHT_NORM,2),
-            "reason": reason
-        })
+    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("経営者能力", "経営姿勢", mg_att, _WEIGHTS[("経営者能力","経営姿勢")], f"予実{yoy3 or '—'}%/監査{audit_bad or 0}/違反{comp_bad or 0}/社外{indep or '—'}%")
+    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("経営者能力", "事業経験", mg_exp, _WEIGHTS[("経営者能力","事業経験")], f"経験{exp_years if exp_years is not None else '不明→中立'}年")
+    _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("経営者能力", "資産担保力", mg_asset, _WEIGHTS[("経営者能力","資産担保力")], f"現預金/月商≈{round(cash_to_ms,2) if cash_to_ms else '—'}")
+    _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)}件→大幅減点"
+    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 '不明'}年）"
+        pen = 6.0 if (bk_years and bk_years >= 10) else 3.0; rs = f"倒産歴あり（{bk_years or '不明'}年）"
     else:
-        pen=10.0; rs="事故/倒産なし"
-    _add("経営者能力","減点事項",pen,_WEIGHTS[("経営者能力","減点事項")],rs)
+        pen = 10.0; rs = "事故/倒産なし"
+    _add(items, "経営者能力", "減点事項", pen, _WEIGHTS[("経営者能力", "減点事項")], rs)
 
-    # 成長率
-    _add("成長率","売上高伸長性", _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("成長率","利益伸長性", _ramp(p_cagr,0.08,-0.05), _WEIGHTS[("成長率","利益伸長性")], f"CAGR営業{round((p_cagr or 0)*100,1) if p_cagr is not None else '—'}%")
-    _add("成長率","商品", 5.0, _WEIGHTS[("成長率","商品")], "不明→中立")
+    # 成長率（市場成長で調整）
+    _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)
+
+    # 市場成長調整：市場 >10% なら高評価、マイナス成長は減点
+    _add(items, "成長率", "市場成長調整",
+         _ramp(market_growth, 15, -5),
+         _WEIGHTS[("成長率", "市場成長調整")],
+         f"市場年成長{market_growth or '—'}%")
 
     # 安定性
-    _add("安定性","自己資本", _ramp(equity,40,5), _WEIGHTS[("安定性","自己資本")], f"自己資本比率{equity 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)}日"
+        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(cash_to_ms,1.0,0.2); rs=f"代理：現預金/月商≈{round(cash_to_ms,2) if cash_to_ms else '—'}"
-    _add("安定性","決済振り", sc, _WEIGHTS[("安定性","決済振り")], rs)
+        sc = _ramp(cash_to_ms, 1.0, 0.2)
+        rs = f"代理：現預金/月商≈{round(cash_to_ms,2) if cash_to_ms else '—'}"
+    _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("安定性","金融取引", sc_mb, _WEIGHTS[("安定性","金融取引")], f"メイン{'有' if mainbank else '無' if mainbank is False else '—'}/与信枠{'有' if has_line else '無' if has_line is False else '—'}")
+    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(coll_to_ms, 4.0, 0.0),
+         _WEIGHTS[("安定性", "資産担保余力")], f"担保/月商≈{round(coll_to_ms,2) if coll_to_ms else '—'}")
 
-    _add("安定性","資産担保余力", _ramp(coll_to_ms,4.0,0.0), _WEIGHTS[("安定性","資産担保余力")], f"担保/月商≈{round(coll_to_ms,2) if coll_to_ms else '—'}")
-    _add("安定性","取引先", ( _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("安定性","業歴", _ramp(years,20,1), _WEIGHTS[("安定性","業歴")], f"{years or '—'}年")
+    _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
-    sc_dis += 10.0 if has_sec else (7.0 if (pub_off or pub_web) else 4.0)
+    sc_dis += 10.0 if has_sec := _to_bool(ref("有価証券報告書提出企業か(TRUE/FALSE)")) 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("公平性・総合世評","ディスクロージャー", sc_dis, _WEIGHTS[("公平性・総合世評","ディスクロージャー")], f"{'有報' if has_sec else '公開あり' if (pub_off or pub_web) else '公開乏しい'} / 更新{'◯' if upd_on else '—'}")
-
-    total = round(sum(x["score"] for x in items),1)
-    return {"name":"企業評価（外部・定量化）","external_total": total, "items": items, "notes":"欠損は中立、連続スコア×重み（自動正規化）"}
+    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)
+
+    # レーダ用にカテゴリ集計（重み付き平均→0-100）
+    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": "欠損は中立。市場成長率を成長評価に加味（過熱/低迷の補正）。",
+    }