Create market_infer.py

core/market_infer.py

@@ -0,0 +1,117 @@
+# core/market_infer.py
+from __future__ import annotations
+import os, json, hashlib, time
+from typing import Dict, Any, List
+from openai import OpenAI
+
+OPENAI_MODEL_TEXT = os.environ.get("OPENAI_TEXT_MODEL", "gpt-4o-mini")
+
+# ---- simple memo cache (in-memory) ----
+_CACHE: Dict[str, Dict[str, Any]] = {}
+
+def _client() -> OpenAI:
+    key = os.environ.get("OPENAI_API_KEY")
+    if not key:
+        raise RuntimeError("OPENAI_API_KEY が未設定です")
+    return OpenAI(api_key=key, timeout=45)
+
+def _cache_key(industry: str, products: List[str], country: str, horizon_years: int) -> str:
+    raw = json.dumps({
+        "industry": industry.strip(),
+        "products": [p.strip() for p in products if p.strip()],
+        "country": country.strip(),
+        "horizon": horizon_years
+    }, ensure_ascii=False, sort_keys=True)
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()
+
+def _to_float(x):
+    if x is None: return None
+    try: return float(str(x).replace(",",""))
+    except Exception: return None
+
+def infer_market_metrics(industry: str, products: List[str], country: str = "JP", horizon_years: int = 3) -> Dict[str, Any]:
+    """
+    事業領域・商品から市場メトリクスを推定して返す。
+    返り値は数値を極力float化した dict。
+    """
+    key = _cache_key(industry, products, country, horizon_years)
+    if key in _CACHE:
+        return _CACHE[key]
+
+    prompt = f"""
+あなたは投資銀行のセクターアナリストです。以下の事業領域・商品と地域を前提に、
+“定量”重視で市場メトリクスを推定してください。専門常識に基づく概算で構いません。
+出力は **厳密なJSON** のみ、単位とキーは以下スキーマに合わせてください。
+
+【入力】
+- 事業領域: {industry.strip() or "（未入力）"}
+- 商品: {", ".join([p.strip() for p in products if p.strip()]) or "（未入力）"}
+- 主対象国/地域: {country}
+- 予測期間: 直近→{horizon_years}年
+
+【出力スキーマ（数値は可能な限り実数値、%は実数）】
+{{
+  "市場の年成長率(%)": 0.0,
+  "市場成熟度(0-1)": 0.0,                // 0=黎明, 1=成熟
+  "競争強度(0-10)": 0.0,                  // 大きいほど競争が激しい
+  "参入障壁(0-10)": 0.0,
+  "価格決定力(0-10)": 0.0,
+  "サイクル感応度(0-10)": 0.0,
+  "規制リスク(0-10)": 0.0,
+  "技術破壊リスク(0-10)": 0.0,
+  "TAM_億円": 0.0, "SAM_億円": 0.0, "SOM_億円": 0.0,
+  "製品別年成長率(%)": {{"<製品名>": 0.0}},
+  "注記": ["定量根拠の観点や前提（簡潔に）"]
+}}
+
+注意:
+- 未知は推定値で埋めて構いません（整合性重視）。
+- 根拠の“観点”のみ列挙し、URLは不要。
+- 回答はJSONのみ。
+""".strip()
+
+    try:
+        cli = _client()
+        resp = cli.chat.completions.create(
+            model=OPENAI_MODEL_TEXT,
+            messages=[
+                {"role": "system", "content": "出力は厳密なJSONのみ。説明やマークダウンは不要。"},
+                {"role": "user", "content": prompt},
+            ],
+            response_format={"type": "json_object"},
+            temperature=0.3,
+        )
+        data = json.loads(resp.choices[0].message.content)
+
+        # 型の安全化
+        out: Dict[str, Any] = {}
+        for k in [
+            "市場の年成長率(%)","市場成熟度(0-1)","競争強度(0-10)","参入障壁(0-10)",
+            "価格決定力(0-10)","サイクル感応度(0-10)","規制リスク(0-10)","技術破壊リスク(0-10)",
+            "TAM_億円","SAM_億円","SOM_億円"
+        ]:
+            out[k] = _to_float(data.get(k))
+        # 製品別成長
+        prod = {}
+        for name, val in (data.get("製品別年成長率(%)") or {}).items():
+            fv = _to_float(val)
+            if name and fv is not None:
+                prod[str(name)] = fv
+        out["製品別年成長率(%)"] = prod
+        # 注記
+        notes = data.get("注記") or []
+        out["注記"] = [str(x) for x in notes][:5]
+
+        _CACHE[key] = out
+        return out
+    except Exception as e:
+        # フォールバック（無難値）
+        fallback = {
+            "市場の年成長率(%)": 4.0, "市場成熟度(0-1)": 0.6, "競争強度(0-10)": 6.0, "参入障壁(0-10)": 5.0,
+            "価格決定力(0-10)": 5.0, "サイクル感応度(0-10)": 5.0, "規制リスク(0-10)": 4.0, "技術破壊リスク(0-10)": 5.0,
+            "TAM_億円": 1000.0, "SAM_億円": 300.0, "SOM_億円": 60.0,
+            "製品別年成長率(%)": {p: 5.0 for p in products if p.strip()},
+            "注記": [f"推定失敗のため既定値を使用: {type(e).__name__}"]
+        }
+        _CACHE[key] = fallback
+        return fallback