Update core/ai_judgement.py

core/ai_judgement.py

@@ -1,215 +1,42 @@
 # core/ai_judgement.py
 from __future__ import annotations
 import os, json
-from typing import Dict, Any, List, Optional
-
-# OpenAI は任意（キー未設定でも動く）。使える環境なら市場/製品の補足を LLM に頼みます。
-try:
-    from openai import OpenAI  # requirements: openai>=1.33
-except Exception:  # ランタイム最小化のための保険
-    OpenAI = None  # type: ignore
 
 OPENAI_MODEL_TEXT = os.environ.get("OPENAI_TEXT_MODEL", "gpt-4o-mini")
 
-def _client() -> Optional[OpenAI]:
-    key = os.environ.get("OPENAI_API_KEY")
-    if not key or OpenAI is None:
-        return None
-    # 互換性のため proxies 等は渡さない
-    return OpenAI(api_key=key, timeout=30)
-
-# ---------------- KPI 抽出（定量） ----------------
-def _to_f(x) -> Optional[float]:
+def _client():
     try:
-        if x in (None, "", "null"): return None
-        return float(x)
+        from openai import OpenAI
+        key = os.environ.get("OPENAI_API_KEY")
+        if not key:
+            return None
+        return OpenAI(api_key=key, timeout=30)
     except Exception:
         return None
 
-def _pct(n: Optional[float]) -> str:
-    return "—" if n is None else f"{n:.1f}%"
-
-def _num(n: Optional[float]) -> str:
-    return "—" if n is None else f"{n:,.0f}"
-
-def _safe_div(a: Optional[float], b: Optional[float]) -> Optional[float]:
-    if a is None or b is None or b == 0: return None
-    return a / b
-
-def _extract_kpi(fin: Dict[str, Any]) -> Dict[str, Optional[float]]:
-    bs = fin.get("balance_sheet") or {}
-    is_ = fin.get("income_statement") or {}
-    cf = fin.get("cash_flows") or {}
-
-    total_assets   = _to_f(bs.get("total_assets"))
-    total_eq       = _to_f(bs.get("total_equity"))
-    current_assets = _to_f(bs.get("current_assets"))
-    current_liab   = _to_f(bs.get("current_liabilities"))
-
-    sales          = _to_f(is_.get("sales"))
-    op_income      = _to_f(is_.get("operating_income"))
-    net_income     = _to_f(is_.get("net_income"))
-    cogs           = _to_f(is_.get("cost_of_sales"))
-    gross_profit   = _to_f(is_.get("gross_profit"))
-    op_exp         = _to_f(is_.get("operating_expenses"))
-
-    op_cf          = _to_f(cf.get("operating_cash_flow"))
-
-    equity_ratio   = None if (total_assets is None or total_assets == 0) else (total_eq or 0)/total_assets*100.0
-    current_ratio  = None if (current_liab is None or current_liab == 0) else (current_assets or 0)/current_liab*100.0
-    gp_margin      = None if (sales is None or sales == 0 or gross_profit is None) else gross_profit/sales*100.0
-    op_margin      = None if (sales is None or sales == 0 or op_income is None) else op_income/sales*100.0
-    net_margin     = None if (sales is None or sales == 0 or net_income is None) else net_income/sales*100.0
-    op_cf_ratio    = _safe_div(op_cf, sales)
-    return dict(
-        sales=sales, op_income=op_income, net_income=net_income,
-        equity_ratio=equity_ratio, current_ratio=current_ratio,
-        gp_margin=gp_margin, op_margin=op_margin, net_margin=net_margin,
-        op_cf_ratio=None if op_cf_ratio is None else op_cf_ratio*100.0,
-    )
-
-# ---------------- LLM による市場/製品の補足（任意） ----------------
-def _llm_market_product(company: str, business_text: str) -> Optional[Dict[str, Any]]:
-    """
-    事業説明テキストから『市場CAGR・市場規模・製品の差別化度』を簡易推定（JSON）。
-    OpenAI キーが無い/失敗時は None を返す（UIはそのまま動く）。
-    """
-    cli = _client()
-    if not cli or not business_text.strip():
-        return None
-    try:
-        prompt = f"""
-あなたは投資アナリストです。以下の会社説明から、市場の定量情報と製品の定量評価を推定してください。
-厳密な JSON オブジェクトのみを日本語の単位なし半角数値で返します。
-
-出力 JSON 仕様:
-{{
-  "market_cagr_pct": null,         // 想定CAGR（%）
-  "market_size_next3y_jpy": null,  // 3年後の市場規模（円）
-  "product_innovation_score": null, // 製品の革新性/差別化(0-10)
-  "signals": []                     // 箇条書き根拠（日本語）
-}}
-
-[会社名] {company or '不明'}
-[事業説明]
-{business_text[:5000]}
+def make_ai_memo(company: str, fin, score_internal, score_external, business_text: str) -> str:
+    client = _client()
+    if client is None:
+        return "（通知）OPENAI_API_KEY未設定のため、AI所見はスキップしました。"
+
+    prompt = f"""あなたは中立なアナリストです。過度な断定や主観を避け、可観測な数値指標を軸に簡潔に述べてください。
+- 良い点（定量指標ベース）3個以内
+- 懸念点（定量指標ベース）3個以内
+- 市場/製品の補足（PDF本文から推定される事業の定量的観点があれば一言）
+- 総評（80字以内、結論は仮説ベースと明記）
+
+[会社候補] {company or '—'}
+[財務(JSON)] {json.dumps(fin, ensure_ascii=False)}
+[内部スコア] {json.dumps(score_internal, ensure_ascii=False)}
+[外部評価] {json.dumps(score_external, ensure_ascii=False)}
+[事業テキスト候補] {business_text[:1200]}
 """
-        res = cli.chat.completions.create(
-            model=OPENAI_MODEL_TEXT,
-            messages=[
-                {"role":"system","content":"出力は必ず有効な JSON。説明文やコードブロックを含めない。"},
-                {"role":"user","content":prompt},
-            ],
-            response_format={"type":"json_object"},
-            temperature=0.2,
-        )
-        data = json.loads(res.choices[0].message.content)
-        # 型のゆらぎを軽く補正
-        for k in ("market_cagr_pct","market_size_next3y_jpy","product_innovation_score"):
-            v = data.get(k, None)
-            try:
-                data[k] = None if v in (None,"", "null") else float(v)
-            except Exception:
-                data[k] = None
-        if not isinstance(data.get("signals", []), list):
-            data["signals"] = []
-        return data
-    except Exception:
-        return None
-
-# ---------------- パブリック API ----------------
-def make_ai_memo(
-    company: str,
-    fin: Dict[str, Any],
-    score_internal: Optional[Dict[str, Any]] = None,  # 社内ルールのスコア（任意）
-    score_external: Optional[Dict[str, Any]] = None,  # 外部（定量化）スコア（任意）
-    business_text: str = ""                           # 事業説明や製品説明（あると LLM が強化）
-) -> str:
-    """
-    中立・簡潔な AI 所見（Markdown）を返します。
-    - 財務からの **定量 KPI** を先頭に表で提示
-    - 良い点 / リスクをバランスよく列挙（定量しきい値で判断）
-    - 可能なら LLM で『市場CAGR・市場規模・製品の差別化度』を推定して補足
-    """
-    kpi = _extract_kpi(fin)
-
-    # しきい値で素朴に判定（中立・再現性重視）
-    positives: List[str] = []
-    risks: List[str] = []
-
-    if (kpi["equity_ratio"] or 0) >= 40: positives.append(f"自己資本比率 {_pct(kpi['equity_ratio'])} と健全")
-    if (kpi["current_ratio"] or 0) >= 120: positives.append(f"流動比率 {_pct(kpi['current_ratio'])} と手元流動性に余裕")
-    if (kpi["op_margin"] or 0) >= 8: positives.append(f"営業利益率 {_pct(kpi['op_margin'])} と収益性まずまず")
-    if (kpi["op_cf_ratio"] or 0) >= 5: positives.append(f"営業CF/売上 {_pct(kpi['op_cf_ratio'])} とキャッシュ創出力あり")
-
-    if (kpi["equity_ratio"] or 100) < 15: risks.append(f"自己資本比率 {_pct(kpi['equity_ratio'])} と財務余力に懸念")
-    if (kpi["current_ratio"] or 999) < 100: risks.append(f"流動比率 {_pct(kpi['current_ratio'])} と短期支払能力に注意")
-    if (kpi["op_margin"] or 100) < 3: risks.append(f"営業利益率 {_pct(kpi['op_margin'])} と採算性に課題")
-    if (kpi["net_margin"] or 100) < 2: risks.append(f"純利益率 {_pct(kpi['net_margin'])} と底堅さに欠ける可能性")
-
-    # LLM による市場/製品の補足（任意）
-    mp = _llm_market_product(company, business_text) if business_text else None
-    if mp:
-        if mp.get("market_cagr_pct") is not None:
-            if float(mp["market_cagr_pct"]) >= 10:
-                positives.append(f"想定市場CAGR {mp['market_cagr_pct']:.1f}% と高成長")
-            elif float(mp["market_cagr_pct"]) <= 0:
-                risks.append(f"想定市場CAGR {mp['market_cagr_pct']:.1f}% と市場縮小の懸念")
-        if mp.get("product_innovation_score") is not None:
-            s = float(mp["product_innovation_score"])
-            if s >= 7.5:
-                positives.append(f"製品の差別化度（LLM推定）{s:.1f}/10 と強み")
-            elif s <= 3.0:
-                risks.append(f"製品の差別化度（LLM推定）{s:.1f}/10 と競争激化の恐れ")
-
-    # スコアの見出し（社内/外部で基準を分けていることを明示）
-    internal_line = ""
-    if score_internal:
-        internal_line = f"- 社内スコア: **{score_internal.get('total_score','—')} / 100**（グレード: {score_internal.get('grade','—')}）\n"
-    external_line = ""
-    if score_external:
-        external_line = f"- 外部スコア: **{score_external.get('external_total','—')} / 100**（ディスクロージャー等の客観指標ベース）\n"
-
-    # Markdown 組み立て
-    md = []
-    md.append(f"### {company or '対象企業'} — AI所見（中立）")
-    md.append("#### 主要KPI（単位：% は百分率）")
-    md.append(
-        f"""
-| KPI | 値 |
-|---|---:|
-| 売上高 | { _num(kpi['sales']) } |
-| 営業利益 | { _num(kpi['op_income']) } |
-| 純利益 | { _num(kpi['net_income']) } |
-| 自己資本比率 | { _pct(kpi['equity_ratio']) } |
-| 流動比率 | { _pct(kpi['current_ratio']) } |
-| 売上総利益率 | { _pct(kpi['gp_margin']) } |
-| 営業利益率 | { _pct(kpi['op_margin']) } |
-| 純利益率 | { _pct(kpi['net_margin']) } |
-| 営業CF/売上 | { _pct(kpi['op_cf_ratio']) } |
-""".strip()
+    resp = client.chat.completions.create(
+        model=OPENAI_MODEL_TEXT,
+        messages=[
+            {"role": "system", "content": "出力は日本語。見出しは使わず、箇条書きと短い総評のみ。感情語や煽りは禁止。"},
+            {"role": "user", "content": prompt},
+        ],
+        temperature=0.2,
     )
-
-    if internal_line or external_line:
-        md.append("#### スコア概況")
-        md.append(internal_line + external_line)
-
-    if positives:
-        md.append("#### プラス要因")
-        md.extend([f"- {p}" for p in positives])
-    if risks:
-        md.append("#### リスク/留意点")
-        md.extend([f"- {r}" for r in risks])
-
-    if mp:
-        md.append("#### 市場/製品の補足（LLM）")
-        line1 = f"- 想定市場CAGR: **{_pct(mp.get('market_cagr_pct'))}**"
-        line2 = f"- 3年後市場規模（推定）: **{_num(mp.get('market_size_next3y_jpy'))} 円**"
-        line3 = f"- 製品の差別化度（0-10）: **{mp.get('product_innovation_score') if mp.get('product_innovation_score') is not None else '—'}**"
-        md.append("\n".join([line1, line2, line3]))
-        if mp.get("signals"):
-            md.append("**根拠（要旨）**")
-            md.extend([f"- {s}" for s in mp["signals"]])
-
-    md.append("\n> 注記: 欠損値は算定不能のため **—** 表示。所見は公開情報/入力値に基づく中立的サマリーであり、投資勧誘を目的としません。")
-    return "\n\n".join(md)
+    return resp.choices[0].message.content