#!/usr/bin/env python3 """ nba_engine.py — 名店DB Next Best Action エンジン v1.1.0 仕様書: meiten-business/12_next-action-engine.md §5 実行: python nba_engine.py [--dry-run] [--output PATH] [--as-of YYYY-MM-DD] [--store-id MTN-XXX] python nba_engine.py [--params PATH] ← パラメータファイルを明示指定 変更点 v1.1.0（2026-06-06）: - パラメータを nba_params.json から読み込む（無ければ既定値で自動生成・冪等） - calc_s_strategic がサブ因子 {worldview,tegata,hotel,symbol,funnel} を返す - ev_breakdown にサブ因子を追加 - nba_factors.json を別途出力（フロント重みスライダー用） GR19準拠: 係数は初期値・要キャリブレーション。捏造禁止。 """ import json import argparse import os import sys from datetime import datetime, timedelta # ============================================================ # デフォルトパラメータ（nba_params.json が無い場合の既定値） # ============================================================ DEFAULT_PARAMS = { "W_WORLD": 0.30, "W_TEGATA": 0.20, "W_HOTEL": 0.15, "W_SYMBOL": 0.20, "W_FUNNEL": 0.15, "HOURLY_RATE": 2000, "L6_ACTIVE_THRESHOLD": 20, "L4_COUNT_AS_REVENUE": False, "SYMBOL_KEYWORDS": [ "象徴", "老舗", "誌名", "言問団子", "朝顔", "300年", "明治", "大正", "元禄", "約100年", "10周年", "入谷", ], } # アクション名マスター ACTION_NAMES = { "A01": "年契更新・継続確認", "A02": "アップセル提案", "A03": "取材オファー", "A04": "名店リスト勧誘", "A05": "ブランド枠初回提案", "A06": "大型枠・記事広告提案", "A07": "途絶復活打診", "A08": "手形/ホテル連携提案", "A09": "担当アサイン", "A10": "休眠・見送り", } ACTION_HOURS = { "A01": 1.0, "A02": 2.0, "A03": 3.0, "A04": 1.5, "A05": 2.0, "A06": 3.0, "A07": 2.0, "A08": 1.5, "A09": 0.5, "A10": 0.0, } # ============================================================ # パラメータ読み込み（冪等） # ============================================================ def load_params(params_path: str) -> dict: """ nba_params.json を読み込んでパラメータ辞書を返す。 ファイルが無ければ DEFAULT_PARAMS で自動生成（冪等）。 未定義キーはデフォルト値でフォールバック。 """ if not os.path.exists(params_path): # 自動生成 obj = { "_meta": { "title": "NBA エンジン 調整パラメータ", "schema_version": "1.0.0", "description": ( "nba_engine.py が読み込む外部パラメータ。" "このファイルを変更して POST /api/meiten/nba-params を呼ぶか、" "nba_engine.py を直接実行すると全202店が再スコアされる。" ), "updated": datetime.now().strftime("%Y-%m-%d"), }, } obj.update(DEFAULT_PARAMS) os.makedirs(os.path.dirname(params_path), exist_ok=True) with open(params_path, "w", encoding="utf-8") as f: json.dump(obj, f, ensure_ascii=False, indent=2) print(f"[NBA Engine] nba_params.json を新規生成しました: {params_path}") with open(params_path, encoding="utf-8") as f: raw = json.load(f) # デフォルト値でフォールバック p = {} for k, v in DEFAULT_PARAMS.items(): p[k] = raw.get(k, v) return p def validate_params(p: dict) -> list: """ パラメータバリデーション。 returns: list of warning strings（空 = OK） """ warnings = [] weight_keys = ["W_WORLD", "W_TEGATA", "W_HOTEL", "W_SYMBOL", "W_FUNNEL"] for k in weight_keys: v = p.get(k, 0) if not (0.0 <= v <= 1.0): warnings.append(f"{k}={v} は 0〜1 の範囲外") w_sum = sum(p.get(k, 0) for k in weight_keys) if abs(w_sum - 1.0) > 0.20: warnings.append( f"重み合計={w_sum:.3f}（1.0 からの乖離={abs(w_sum-1.0):.3f} が 0.20 超）" "──合計1.0でなくても動作しますが確認推奨" ) if p.get("HOURLY_RATE", 0) <= 0: warnings.append(f"HOURLY_RATE={p['HOURLY_RATE']} は正の値が必要") return warnings # ============================================================ # 計算関数群（パラメータ辞書 p を引数で受け取る設計） # ============================================================ def calc_p_convert(store: dict) -> float: """獲得確率 P_convert ∈ [0.02, 0.90]（要キャリブレーション）""" label = store.get("label", "L6") base_p_map = { "L1a": 0.90, "L1b": 0.85, "L1c": 0.80, "L1d": 0.85, "L2": 0.25, "L3": 0.30, "L4": 0.95, "L5": 0.15, "L6": 0.10, } base_p = base_p_map.get(label, 0.10) # 補正1: Vol.15受注済み → 翌号継続確率UP v15 = store.get("vol15") or {} if v15.get("status") == "受注": base_p = min(base_p * 1.15, 0.95) # 補正2: lapsed → ペナルティ if store.get("lapsed") is True: base_p = base_p * 0.40 # 補正3: issuesPlaced の長さで接点履歴加点 n_issues = len(store.get("issuesPlaced") or []) if n_issues >= 8: base_p = min(base_p * 1.20, 0.95) elif n_issues >= 4: base_p = min(base_p * 1.10, 0.95) # 補正4: potentialTier tier_mult = {"A": 1.10, "B": 1.00, "C": 0.85} tier = store.get("potentialTier") or "B" base_p = base_p * tier_mult.get(tier, 1.00) return round(max(0.02, min(0.90, base_p)), 4) def calc_v_expected(store: dict) -> int: """想定受注額（1号あたり）""" # 1. vol15確定額 v15 = store.get("vol15") or {} if v15.get("amount") and v15.get("status") in ["受注", "商談中"]: return int(v15["amount"]) # 2. 直近実績 last = store.get("lastSpend") or 0 if last > 0: return int(last) # 3. label×potentialTierデフォルト label = store.get("label", "L6") tier = store.get("potentialTier", "B") or "B" defaults = { ("L1a", "A"): 400000, ("L1a", "B"): 200000, ("L1a", "C"): 100000, ("L1b", "A"): 60000, ("L1b", "B"): 60000, ("L1b", "C"): 35000, ("L1c", "A"): 3000, ("L1c", "B"): 3000, ("L1c", "C"): 3000, ("L1d", "A"): 50000, ("L1d", "B"): 35000, ("L1d", "C"): 30000, ("L2", "A"): 200000, ("L2", "B"): 100000, ("L2", "C"): 50000, ("L3", "A"): 60000, ("L3", "B"): 35000, ("L3", "C"): 3000, ("L4", "A"): 60000, ("L4", "B"): 35000, ("L4", "C"): 3000, ("L5", "A"): 200000, ("L5", "B"): 100000, ("L5", "C"): 50000, ("L6", "A"): 60000, ("L6", "B"): 35000, ("L6", "C"): 3000, } key = (label, tier) fallback_key = (label, "B") return defaults.get(key, defaults.get(fallback_key, 3000)) def calc_s_strategic_with_factors(store: dict, p: dict) -> tuple: """ 戦略係数 S_strategic ∈ [0.5, 2.0] と 正規化サブ因子 dict{worldview,tegata,hotel,symbol,funnel} を返す。 各サブ因子は重み適用前の 0〜1 スコア。 """ SYMBOL_KEYWORDS = p.get("SYMBOL_KEYWORDS", DEFAULT_PARAMS["SYMBOL_KEYWORDS"]) # worldviewFit wf_text = str(store.get("worldviewFit") or "") if "最高" in wf_text or ("高" in wf_text and "低" not in wf_text): w_world = 1.0 elif "中" in wf_text: w_world = 0.5 else: w_world = 0.0 # tegataFit tf = store.get("tegataFit") or {} if isinstance(tf, dict): tf_level = tf.get("level", "低") or "低" else: tf_level = str(tf) tegata_map = {"高": 1.0, "中": 0.5, "低": 0.0} w_tegata = tegata_map.get(tf_level, 0.0) # hotelEastFit hf_text = str(store.get("hotelEastFit") or "") if "高" in hf_text: w_hotel = 1.0 elif "中" in hf_text: w_hotel = 0.5 else: w_hotel = 0.0 # 象徴性スコア biz_text = str(store.get("bizDevNote") or "") + str(store.get("nextAction") or "") w_symbol = min(1.0, sum(0.3 for kw in SYMBOL_KEYWORDS if kw in biz_text)) # ファネル乗数 label = store.get("label", "") rating = store.get("rating") or {} try: rating_score = float(rating.get("score", 0) or 0) if isinstance(rating, dict) else 0.0 except (ValueError, TypeError): rating_score = 0.0 if label == "L3": w_funnel = 0.8 elif label in ["L6", "L2"] and rating_score >= 3.5: w_funnel = 0.4 elif label in ["L1a", "L1b"] and len(store.get("issuesPlaced") or []) >= 4: w_funnel = 0.2 else: w_funnel = 0.0 # 重み付き合計 → [0.5, 2.0]に線形変換 W_WORLD = p.get("W_WORLD", DEFAULT_PARAMS["W_WORLD"]) W_TEGATA = p.get("W_TEGATA", DEFAULT_PARAMS["W_TEGATA"]) W_HOTEL = p.get("W_HOTEL", DEFAULT_PARAMS["W_HOTEL"]) W_SYMBOL = p.get("W_SYMBOL", DEFAULT_PARAMS["W_SYMBOL"]) W_FUNNEL = p.get("W_FUNNEL", DEFAULT_PARAMS["W_FUNNEL"]) raw_score = (W_WORLD * w_world + W_TEGATA * w_tegata + W_HOTEL * w_hotel + W_SYMBOL * w_symbol + W_FUNNEL * w_funnel) s_strategic = 0.5 + raw_score * 1.5 sub_factors = { "worldview": round(w_world, 3), "tegata": round(w_tegata, 3), "hotel": round(w_hotel, 3), "symbol": round(w_symbol, 3), "funnel": round(w_funnel, 3), } return round(s_strategic, 3), sub_factors def calc_s_strategic(store: dict, p: dict) -> float: """戦略係数のみ返す（後方互換ラッパー）""" s, _ = calc_s_strategic_with_factors(store, p) return s def calc_c_action(action_type: str, p: dict) -> int: """営業コスト = 工数(h) × 人件費単価""" hours = ACTION_HOURS.get(action_type, 1.0) return int(hours * p.get("HOURLY_RATE", DEFAULT_PARAMS["HOURLY_RATE"])) def calc_ev(store: dict, action_type: str, p: dict) -> float: """EV = P_convert × V_expected × S_strategic - C_action""" pc = calc_p_convert(store) v = calc_v_expected(store) s = calc_s_strategic(store, p) c = calc_c_action(action_type, p) return (pc * v * s) - c def calc_priority_score(ev: float, action_type: str, ev_min: float, ev_max: float) -> int: """優先度スコア = EV を 0-100 に正規化（A10は0固定）""" if action_type == "A10": return 0 if ev_max == ev_min: return 50 normalized = (ev - ev_min) / (ev_max - ev_min) return max(0, min(100, int(normalized * 100))) def calc_ice_score(store: dict, action_type: str) -> dict: """ICEスコア（補助指標・営業現場向け）""" v = calc_v_expected(store) p_conv = calc_p_convert(store) hours = ACTION_HOURS.get(action_type, 1.0) impact = max(1, min(10, int(v / 40000) + 1)) confidence = max(1, min(10, int(p_conv * 10))) ease = max(1, min(10, 10 - int(hours * 2))) return { "I": impact, "C": confidence, "E": ease, "ICE": round((impact + confidence + ease) / 3, 2), } # ============================================================ # アップセル先テキスト # ============================================================ def _next_tier_text(current_label: str) -> str: m = { "L1c": "ブランド枠¥60k", "L1d": "1P¥100k", "L1b": "記事広告¥600k", "L1a": "年契化/面積拡大", } return m.get(current_label, "上位枠") # ============================================================ # NBA決定ルール（§3.1 デシジョンツリー） # ============================================================ def determine_nba(store: dict, p: dict) -> dict: """次アクションを一意に決定する""" label = store.get("label", "L6") v15 = store.get("vol15") or {} v15_status = v15.get("status", "") or "" v_expected = calc_v_expected(store) tier = store.get("potentialTier", "B") or "B" L6_ACTIVE_THRESHOLD = p.get("L6_ACTIVE_THRESHOLD", DEFAULT_PARAMS["L6_ACTIVE_THRESHOLD"]) # === P0: Vol.15商談中 === if v15_status == "商談中": action = "A06" if v_expected >= 100000 else "A05" return { "action_type": action, "priority": "P0", "deadline_days": 7, "reason": f"Vol.15商談中・今号クローズ必須。想定¥{v_expected:,}。即フォロー。", } # === L1（継続）=== if label in ["L1a", "L1b", "L1c", "L1d"]: if v15_status == "受注": tf = store.get("tegataFit") or {} tf_level = (tf.get("level", "低") if isinstance(tf, dict) else str(tf)) or "低" hf_text = str(store.get("hotelEastFit") or "") if tf_level == "高" and ("高" in hf_text or "中" in hf_text): return { "action_type": "A08", "priority": "P2", "deadline_days": 60, "reason": "Vol.15受注済み。手形/ホテル連携クロスセルで関係深化。次号まで。", } elif label in ["L1c", "L1d"] and tier in ["A", "B"]: return { "action_type": "A02", "priority": "P1", "deadline_days": 30, "reason": ( f"名店リスト/記事下の継続中。potentialTier={tier}" f"→{_next_tier_text(label)}へのアップセル余地。" ), } else: return { "action_type": "A01", "priority": "P2", "deadline_days": 90, "reason": "継続受注済み。次号継続確認（90日以内）を実施。", } else: # 継続中だがVol.15未着手 → 更新確認が最優先 priority = "P0" if label in ["L1a", "L1b"] else "P1" deadline = 14 if label in ["L1a", "L1b"] else 30 return { "action_type": "A01", "priority": priority, "deadline_days": deadline, "reason": f"継続広告主({label})だがVol.15未受注。即更新確認必須。", } # === L2（途絶）=== if label == "L2": last_spend = store.get("lastSpend") or 0 if last_spend >= 200000 or v_expected >= 200000: return { "action_type": "A07", "priority": "P0", "deadline_days": 14, "reason": f"大型途絶（最終¥{last_spend:,}）。まず離脱理由ヒアリング→復活提案。最優先。", } else: last_issue = store.get("lastIssue", "不明") return { "action_type": "A07", "priority": "P1", "deadline_days": 30, "reason": f"途絶（lapsed=True）。最終号:{last_issue}。関係が残るうちに復活打診。", } # === L3（埋蔵金）=== if label == "L3": wf_text = str(store.get("worldviewFit") or "") worldview_high = ("最高" in wf_text or ("高" in wf_text and "低" not in wf_text)) if worldview_high: return { "action_type": "A05", "priority": "P1", "deadline_days": 21, "reason": "L3埋蔵金（取材済・未広告）。世界観適合高。取材連動でブランド枠¥35k〜¥100k初収益化。", } else: return { "action_type": "A04", "priority": "P2", "deadline_days": 45, "reason": "L3埋蔵金。名店リスト¥3kで関係再構築→翌号アップセル。", } # === L4（グループ内）=== if label == "L4": return { "action_type": "A01", "priority": "P3", "deadline_days": 90, "reason": "グループ内。社内調整で継続確認。外部収益カウントの是非はCEO判断（P4=参考扱い）。", } # === L5（域外大手）=== if label == "L5": if tier == "A" and v15_status != "受注": return { "action_type": "A06", "priority": "P1", "deadline_days": 30, "reason": "域外大手potentialTier=A。大型枠提案の余地。法人広報へアプローチ。", } else: return { "action_type": "A10", "priority": "P3", "deadline_days": 999, "reason": "域外大手（L5）・世界観適合低・B/C tier。今期は見送り。リソースをL3/L2に集中。", } # === L6（未接触）=== if label == "L6": best100 = store.get("best100Rank") wf_text = str(store.get("worldviewFit") or "") worldview_high = ("最高" in wf_text or ("高" in wf_text and "低" not in wf_text)) tf = store.get("tegataFit") or {} tf_level = (tf.get("level", "低") if isinstance(tf, dict) else "低") or "低" try: b100_val = int(best100) if best100 is not None else 9999 except (ValueError, TypeError): b100_val = 9999 if tier == "A" or b100_val <= L6_ACTIVE_THRESHOLD: return { "action_type": "A03", "priority": "P1", "deadline_days": 30, "reason": f"L6高潜在（tier={tier}/best100={best100}）。取材オファーでファネル入口を作る。", } elif worldview_high and tf_level in ["高", "中"]: return { "action_type": "A04", "priority": "P2", "deadline_days": 45, "reason": "L6（世界観適合高×手形適性あり）。名店リスト¥3kで関係開始。", } elif tier == "C" or (not worldview_high and tf_level == "低"): return { "action_type": "A10", "priority": "P3", "deadline_days": 999, "reason": "L6低潜在（C tier・世界観低・手形適性低）。今期見送り。", } else: has_rep = bool(store.get("salesReps")) action = "A04" if has_rep else "A09" return { "action_type": action, "priority": "P2", "deadline_days": 60, "reason": "L6 B tier。担当アサイン→名店リスト打診。", } # フォールバック return { "action_type": "A09", "priority": "P3", "deadline_days": 90, "reason": "ラベル不明・担当アサインから開始。", } # ============================================================ # 担当割当 # ============================================================ def assign_rep(store: dict) -> str: """担当者を決定する（優先順位: vol15.rep > 最新salesReps > カテゴリ推定）""" v15 = store.get("vol15") or {} if v15.get("rep"): return str(v15["rep"]) reps = store.get("salesReps") or [] if reps: try: sorted_reps = sorted( reps, key=lambda r: max(r.get("issues") or [0]), reverse=True, ) return str(sorted_reps[0]["rep"]) except Exception: return str(reps[0].get("rep", "未割当（A09要対応）")) label = store.get("label", "L6") category = store.get("category", "") or "" ward = (store.get("area") or {}).get("ward", "") or "" if label in ["L1a"] or (label == "L2" and calc_v_expected(store) >= 200000): return "大田（社長）" if "飲食" in category and ("根岸" in ward or "台東" in ward): return "浅川" if "不動産" in category or "金融" in category: return "石井" return "未割当（A09要対応）" # ============================================================ # メイン処理 # ============================================================ def run_nba_engine( master_path: str, output_path: str, params_path: str, as_of: datetime = None, dry_run: bool = False, store_id_filter: str = None, ) -> dict: """ 全202店にNBAスコアを計算して書き出す（冪等）。 nba_factors.json を output_path の隣に出力。 """ if as_of is None: as_of = datetime.now() # パラメータ読み込み p = load_params(params_path) param_warnings = validate_params(p) if param_warnings: for w in param_warnings: print(f"[NBA Engine] WARNING: {w}") with open(master_path, encoding="utf-8") as f: master = json.load(f) stores = master["stores"] # 単店デバッグ if store_id_filter: stores_to_process = [s for s in stores if s.get("storeId") == store_id_filter] if not stores_to_process: print(f"[NBA Engine] storeId={store_id_filter} が見つかりません。") return {} else: stores_to_process = stores # --- Pass 1: NBA・EV仮計算 --- results = [] ev_list = [] for store in stores_to_process: nba_raw = determine_nba(store, p) action_type = nba_raw["action_type"] s_val, sub_factors = calc_s_strategic_with_factors(store, p) ev = (calc_p_convert(store) * calc_v_expected(store) * s_val) - calc_c_action(action_type, p) ev_list.append(ev) results.append({ "store": store, "nba_raw": nba_raw, "ev": ev, "s_strategic": s_val, "sub_factors": sub_factors, }) ev_min = min(ev_list) ev_max = max(ev_list) # --- Pass 2: 優先度スコア正規化・ICE・担当 --- output_stores = [] factors_stores = [] # nba_factors.json 用 for r in results: store = r["store"] nba_raw = r["nba_raw"] action_type = nba_raw["action_type"] ev = r["ev"] s_val = r["s_strategic"] sub_factors = r["sub_factors"] priority_score = calc_priority_score(ev, action_type, ev_min, ev_max) ice = calc_ice_score(store, action_type) rep = assign_rep(store) deadline_date = (as_of + timedelta(days=nba_raw["deadline_days"])).strftime("%Y-%m-%d") p_conv = calc_p_convert(store) v_exp = calc_v_expected(store) c_act = calc_c_action(action_type, p) nba_block = { "actionType": action_type, "actionName": ACTION_NAMES.get(action_type, ""), "priority": nba_raw["priority"], "priorityScore": priority_score, "ev": int(ev), "ev_breakdown": { "p_convert": p_conv, "v_expected": v_exp, "s_strategic": s_val, "c_action": c_act, # サブ因子（重み適用前の正規化スコア 0〜1） "sub_factors": sub_factors, }, "ice": ice, "repRecommended": rep, "deadlineDays": nba_raw["deadline_days"], "deadlineDate": deadline_date, "reason": nba_raw["reason"], "confidence": "推計（初期モデルv1.1・要キャリブレーション）", "generatedAt": as_of.isoformat(), } output_stores.append({ "storeId": store["storeId"], "name": store["name"], "label": store["label"], "potentialTier": store.get("potentialTier"), "nba": nba_block, }) # factors レコード（フロント重みスライダー用） factors_stores.append({ "storeId": store["storeId"], "name": store["name"], "label": store["label"], "potentialTier": store.get("potentialTier"), "priority": nba_raw["priority"], "priorityScore": priority_score, "actionType": action_type, "actionName": ACTION_NAMES.get(action_type, ""), "ev": int(ev), "ev_breakdown": { "p_convert": p_conv, "v_expected": v_exp, "s_strategic": s_val, "c_action": c_act, }, "sub_factors": sub_factors, "vol_status": (store.get("vol15") or {}).get("status", ""), }) # --- ソート: P0>P1>P2>P3 → priorityScore降順 --- priority_order = {"P0": 0, "P1": 1, "P2": 2, "P3": 3} output_stores.sort( key=lambda x: ( priority_order.get(x["nba"]["priority"], 9), -x["nba"]["priorityScore"], ) ) factors_stores.sort( key=lambda x: ( priority_order.get(x["priority"], 9), -x["priorityScore"], ) ) # --- サマリ集計 --- p_counts = {"P0": 0, "P1": 0, "P2": 0, "P3": 0} for s in output_stores: pr = s["nba"]["priority"] p_counts[pr] = p_counts.get(pr, 0) + 1 output = { "_meta": { "title": "名店DB Next Best Action レポート", "generated_at": as_of.isoformat(), "model_version": "1.1.0", "total_stores": len(output_stores), "summary": p_counts, "parameters": {k: p[k] for k in DEFAULT_PARAMS}, "param_warnings": param_warnings, "note": ( "係数は初期値・要キャリブレーション。" "P_convert/V_expectedは実績でアップデートすること（GR19）。" ), }, "stores": output_stores, } if dry_run: print("[NBA Engine --dry-run] ファイル書き込みをスキップ。") print(json.dumps({ "total": len(output_stores), "summary": p_counts, "param_warnings": param_warnings, "top5": [ { "storeId": s["storeId"], "name": s["name"], "priority": s["nba"]["priority"], "score": s["nba"]["priorityScore"], "action": s["nba"]["actionName"], "ev": s["nba"]["ev"], "sub_factors": s["nba"]["ev_breakdown"]["sub_factors"], } for s in output_stores[:5] ], }, ensure_ascii=False, indent=2)) return output # --- 書き出し: nba_output.json --- with open(output_path, "w", encoding="utf-8") as f: json.dump(output, f, ensure_ascii=False, indent=2) # --- 書き出し: nba_factors.json --- factors_path = os.path.join(os.path.dirname(output_path), "nba_factors.json") factors_obj = { "_meta": { "title": "NBA サブ因子レポート（フロント重みスライダー用）", "generated_at": as_of.isoformat(), "model_version": "1.1.0", "total_stores": len(factors_stores), "current_weights": { "W_WORLD": p["W_WORLD"], "W_TEGATA": p["W_TEGATA"], "W_HOTEL": p["W_HOTEL"], "W_SYMBOL": p["W_SYMBOL"], "W_FUNNEL": p["W_FUNNEL"], }, "note": ( "sub_factors は重み適用前の正規化スコア（0〜1）。" "フロントで重みを変えると s_strategic = 0.5 + raw*1.5 " "（raw = Σ weight×factor）で即再計算可能。" ), }, "stores": factors_stores, } with open(factors_path, "w", encoding="utf-8") as f: json.dump(factors_obj, f, ensure_ascii=False, indent=2) # --- ログ（GR13）--- log_dir = os.path.dirname(os.path.abspath(__file__)).replace("scripts", "logs") os.makedirs(log_dir, exist_ok=True) log_path = os.path.join(log_dir, "nba_engine.jsonl") log_record = { "ts": as_of.isoformat(), "status": "success", "total": len(output_stores), "P0": p_counts["P0"], "P1": p_counts["P1"], "P2": p_counts["P2"], "P3": p_counts["P3"], "output_path": output_path, "factors_path": factors_path, "model_version": "1.1.0", "param_warnings": param_warnings, } with open(log_path, "a", encoding="utf-8") as f: f.write(json.dumps(log_record, ensure_ascii=False) + "\n") print( f"[NBA Engine] {len(output_stores)}店処理完了。" f" P0={p_counts['P0']} P1={p_counts['P1']}" f" P2={p_counts['P2']} P3={p_counts['P3']}" ) print(f"出力: {output_path}") print(f"因子: {factors_path}") return output # ============================================================ # CLI エントリポイント # ============================================================ def main(): BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) parser = argparse.ArgumentParser(description="名店DB Next Best Action エンジン v1.1.0") parser.add_argument("--master", default=os.path.join(BASE_DIR, "data", "meiten-master.json"), help="入力JSONパス") parser.add_argument("--output", default=os.path.join(BASE_DIR, "data", "nba_output.json"), help="出力JSONパス") parser.add_argument("--params", default=os.path.join(BASE_DIR, "data", "nba_params.json"), help="パラメータJSONパス（無ければ既定値で自動生成）") parser.add_argument("--as-of", default=None, help="基準日 YYYY-MM-DD（省略時=今日）") parser.add_argument("--dry-run", action="store_true", help="ファイル書き込みなし・コンソール出力のみ") parser.add_argument("--store-id", default=None, help="単店デバッグ: MTN-XXX") args = parser.parse_args() as_of = None if args.as_of: as_of = datetime.strptime(args.as_of, "%Y-%m-%d") run_nba_engine( master_path=args.master, output_path=args.output, params_path=args.params, as_of=as_of, dry_run=args.dry_run, store_id_filter=args.store_id, ) if __name__ == "__main__": main()