Update core/external_score.py

core/external_score.py

@@ -1,3 +1,4 @@
+# core/external_score.py
 from __future__ import annotations
 from typing import Dict, Any, List, Tuple
 import pandas as pd
@@ -6,9 +7,22 @@ __all__ = [
     "get_external_template_df",
     "fill_missing_with_external",
     "score_external_from_df",
+    "apply_llm_signals_to_df",
 ]
 
 _TEMPLATE_ROWS: List[Tuple[str, str]] = [
+    # 成長性をより定量化
+    ("成長率","市場CAGR(%)"),
+    ("成長率","売上_期3(最新期)"),
+    ("成長率","売上_期2"),
+    ("成長率","売上_期1(最古期)"),
+    ("成長率","営業利益_期3(最新期)"),
+    ("成長率","営業利益_期2"),
+    ("成長率","営業利益_期1(最古期)"),
+    ("成長率","主力商品数"),
+    ("成長率","成長中主力商品数"),
+
+    # 経営者能力
     ("経営者能力","予実達成率_3年平均(%)"),
     ("経営者能力","監査・内部統制の重大な不備 件数(過去3年)"),
     ("経営者能力","重大コンプライアンス件数(過去3年)"),
@@ -21,15 +35,7 @@ _TEMPLATE_ROWS: List[Tuple[str, str]] = [
     ("経営者能力","倒産からの経過年数"),
     ("経営者能力","重大事件・事故件数(過去10年)"),
 
-    ("成長率","売上_期3(最新期)"),
-    ("成長率","売上_期2"),
-    ("成長率","売上_期1(最古期)"),
-    ("成長率","営業利益_期3(最新期)"),
-    ("成長率","営業利益_期2"),
-    ("成長率","営業利益_期1(最古期)"),
-    ("成長率","主力商品数"),
-    ("成長率","成長中主力商品数"),
-
+    # 安定性
     ("安定性","自己資本比率(%)"),
     ("安定性","利益剰余金(円)"),
     ("安定性","支払遅延件数(直近12ヶ月)"),
@@ -47,6 +53,7 @@ _TEMPLATE_ROWS: List[Tuple[str, str]] = [
     ("安定性","不良債権件数(直近12ヶ月)"),
     ("安定性","業歴(年)"),
 
+    # 公平性
     ("公平性・総合世評","有価証券報告書提出企業か(TRUE/FALSE)"),
     ("公平性・総合世評","決算公告や官報での公開あり(TRUE/FALSE)"),
     ("公平性・総合世評","HP/IRサイトで財務資料公開あり(TRUE/FALSE)"),
@@ -57,16 +64,44 @@ 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()
 
+# === LLMの抽出結果をテンプレに反映 ===
+def apply_llm_signals_to_df(df: pd.DataFrame, signals: Dict[str, Any]) -> pd.DataFrame:
+    df2 = df.copy()
+    def setv(field: str, val):
+        m = df2["入力項目"].eq(field)
+        if m.any():
+            df2.loc[m, "値"] = "" if val is None else val
+    if signals:
+        mkt = signals.get("market", {})
+        prod = signals.get("products", {})
+        setv("市場CAGR(%)", mkt.get("cagr_pct"))
+        setv("主力商品数", prod.get("count"))
+        setv("成長中主力商品数", prod.get("growing_count"))
+    return df2
+
 # ===== スコア計算 =====
 _WEIGHTS = {
-    ("経営者能力","経営姿勢"):8, ("経営者能力","事業経験"):5, ("経営者能力","資産担保力"):6, ("経営者能力","減点事項"):7,
-    ("成長率","売上高伸長性"):10, ("成長率","利益伸長性"):10, ("成長率","商品"):6,
-    ("安定性","自己資本"):8, ("安定性","決済振り"):10, ("安定性","金融取引"):6,
-    ("安定性","資産担保余力"):6, ("安定性","取引先"):6, ("安定性","業歴"):4,
-    ("公平性・総合世評","ディスクロージャー"):8,
+    # 成長の定量性を強化
+    ("成長率","市場成長率"): 12,
+    ("成長率","売上高伸長性"): 10,
+    ("成長率","利益伸長性"): 10,
+    ("成長率","商品"): 6,
+
+    ("経営者能力","経営姿勢"): 8,
+    ("経営者能力","事業経験"): 5,
+    ("経営者能力","資産担保力"): 6,
+    ("経営者能力","減点事項"): 7,
+
+    ("安定性","自己資本"): 8,
+    ("安定性","決済振り"): 10,
+    ("安定性","金融取引"): 6,
+    ("安定性","資産担保余力"): 6,
+    ("安定性","取引先"): 6,
+    ("安定性","業歴"): 4,
+
+    ("公平性・総合世評","ディスクロージャー"): 8,
 }
 _WEIGHT_NORM = 100.0 / float(sum(_WEIGHTS.values()))
 
@@ -115,6 +150,32 @@ def score_external_from_df(df: pd.DataFrame) -> Dict[str, Any]:
         return df.loc[m, "値"].values[0] if m.any() else None
 
     items = []
+
+    # ---- 成長率（定量） ----
+    market_cagr=_to_float(ref("市場CAGR(%)"))
+    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(最新期)"))
+    prod_n=_to_float(ref("主力商品数")); prod_g=_to_float(ref("成長中主力商品数"))
+
+    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)
+    # 市場CAGR 12%で満点、-3%で0点（景気循環を意識しやや厳しめ）
+    _add(items,"成長率","市場成長率", _ramp(market_cagr,12,-3), _WEIGHTS[("成長率","市場成長率")],
+         f"市場CAGR {market_cagr if market_cagr is not None else '—'}%")
+    _add(items,"成長率","売上高伸長性", _ramp(s_cagr,0.10,-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.10,-0.05), _WEIGHTS[("成長率","利益伸長性")],
+         f"CAGR営業{round((p_cagr or 0)*100,1) if p_cagr is not None else '—'}%")
+    # 製品ポートフォリオの定量化（成長中比率）
+    grow_ratio = _ratio(prod_g, prod_n)
+    _add(items,"成長率","商品", _ramp(grow_ratio,0.6,0.1), _WEIGHTS[("成長率","商品")],
+         f"成長中比率{round((grow_ratio or 0)*100,1) if grow_ratio is not None else '—'}%")
+
+    # ---- 経営者能力 ----
     yoy3=_to_float(ref("予実達成率_3年平均(%)"))
     audit_bad=_to_float(ref("監査・内部統制の重大な不備 件数(過去3年)"))
     comp_bad=_to_float(ref("重大コンプライアンス件数(過去3年)"))
@@ -127,37 +188,8 @@ 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(最新期)"))
+    cash_to_ms=_ratio(cash, sales_m)
 
-    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)"))
-
-    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)
-    cash_to_ms=_ratio(cash, sales_m2)
-    coll_to_ms=_ratio(collat, sales_m2)
-
-    # 経営者能力
     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 '—'}%")
 
@@ -175,12 +207,16 @@ def score_external_from_df(df: pd.DataFrame) -> Dict[str, Any]:
         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 '—'}%")
-    _add(items,"成長率","商品", 5.0, _WEIGHTS[("成長率","商品")], "不明→中立")
+    # ---- 安定性 ----
+    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)"))
+    main_share=_to_float(ref("メインバンク借入シェア(%)"))
+    has_line=_to_bool(ref("コミットメントライン等の長期与信枠あり(TRUE/FALSE)"))
+    coll_to_ms=_ratio(_to_float(ref("担保余力評価額(円)")), _to_float(ref("月商(円)_再掲")) or sales_m)
 
-    # 安定性
     _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):
@@ -194,16 +230,26 @@ def score_external_from_df(df: pd.DataFrame) -> Dict[str, Any]:
     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,"安定性","金融取引", 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(coll_to_ms,4.0,0.0), _WEIGHTS[("安定性","資産担保余力")],
+         f"担保/月商≈{round(coll_to_ms,2) if coll_to_ms else '—'}")
 
+    top1=_to_float(ref("主要顧客上位1社売上比率(%)"))
+    cust_score=_to_float(ref("主要顧客の平均信用スコア(0-100)"))
+    npl_cnt=_to_float(ref("不良債権件数(直近12ヶ月)"))
     _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)}")
 
+    years=_to_float(ref("業歴(年)"))
     _add(items,"安定性","業歴", _ramp(years,20,1), _WEIGHTS[("安定性","業歴")], f"{years or '—'}年")
 
-    # 公平性
+    # ---- 公平性 ----
+    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)"))
     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 += 1.0 if upd_on else 0.0