Update core/external_scoring.py

core/external_scoring.py

@@ -10,6 +10,7 @@ __all__ = [
     "score_external_from_df",
 ]
 
+# ひな形（＋市場成長率系の列を追加）
 _TEMPLATE_ROWS: List[Tuple[str, str]] = [
     ("経営者能力", "予実達成率_3年平均(%)"),
     ("経営者能力", "監査・内部統制の重大な不備 件数(過去3年)"),
@@ -71,26 +72,22 @@ def fill_missing_with_external(df: pd.DataFrame, suggestions: Dict[str, Any] | N
     return df2
 
 def merge_market_into_external_df(ext_df: pd.DataFrame, market: Dict[str, Any], products: List[str]) -> pd.DataFrame:
-    """市場推定結果と商品リストをext_dfへ反映"""
+    """市場推定結果と商品リストをext_dfへ反映（必ずDataFrameを返す）"""
     df = ext_df.copy()
 
-    def _set(label: str, val):
-        m = df["入力項目"].eq(label)
+    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
-        else:
-            # 念のため行が無ければ追加
-            cat = "成長率" if "成長" in label else "安定性"
-            df = pd.concat([df, pd.DataFrame([[cat, label, val]], columns=df.columns)], ignore_index=True)
-        return df
-
-    # 市場成長率
+            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("市場の年成長率(%)", float(market["市場の年成長率(%)"]))
+        df = _set(df, "市場の年成長率(%)", float(market["市場の年成長率(%)"]), "成長率")
 
-    # 主力商品数 / 成長中主力商品数
     prods = [p for p in products if str(p).strip()]
-    df = _set("主力商品数", len(prods))
+    df = _set(df, "主力商品数", len(prods), "成長率")
 
     growing = 0
     prod_growth: Dict[str, float] = market.get("製品別年成長率(%)") or {}
@@ -100,11 +97,10 @@ def merge_market_into_external_df(ext_df: pd.DataFrame, market: Dict[str, Any],
                 growing += 1
         except Exception:
             pass
-    df = _set("成長中主力商品数", growing)
-
+    df = _set(df, "成長中主力商品数", growing, "成長率")
     return df
 
-# ===== スコア計算（定量・ばらつき強化） =====
+# ===== スコア計算（定量化＋ばらつきストレッチ） =====
 _WEIGHTS = {
     ("経営者能力", "経営姿勢"): 8,
     ("経営者能力", "事業経験"): 5,
@@ -170,6 +166,7 @@ def _add(items, cat, name, raw, weight, 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
@@ -252,7 +249,6 @@ def score_external_from_df(df: pd.DataFrame) -> Dict[str, Any]:
     _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 '—'}%")
@@ -282,57 +278,11 @@ def score_external_from_df(df: pd.DataFrame) -> Dict[str, Any]:
              _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 '—'}"
+        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(coll_to_ms,4.0,0.0),
-         _WEIGHTS[("安定性","資産担保余力")], f"担保/月商≈{round(coll_to_ms,2) if coll_to_ms else '—'}")
-
-    _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": "欠損は中立＋市場成長/商品構成を反映。ストレッチでばらつきを拡大。",
-    }
+    _a_