""" 競合の「上位表示ボタン」更新パターン推定バッチ（GH + バニラ対応） ガールズヘブン(GH)とバニラの競合店スクレイピング結果から「更新イベント」を抽出し、 ツール自動更新かどうか・更新定刻を competitor_update_patterns テーブルに保存する。 バニラは (shop_name, sub_area) の組を店舗単位として扱う。 GHの既存ロジック・結果は変更しない。 Usage: python analyze_competitor_patterns.py """ import argparse import json import logging import math import sys from collections import defaultdict from datetime import date, timedelta from pathlib import Path from statistics import median, stdev import pandas as pd sys.path.append(str(Path(__file__).resolve().parents[1])) from config.database import execute_query, execute_update logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # 定数（後で調整できるようにファイル上部に集約） # --------------------------------------------------------------------------- # GH 用 GH_TOP_POS = 2 # トップ帯（1〜2位）に入った瞬間を更新イベントとみなす BUMP_MIN_IMPROVE = 5 # (旧判定用・未使用、定義のみ残す) TOP_ENTER = 20 # (旧判定用・未使用、定義のみ残す) # バニラ用 VN_IMPROVE_MIN = 15 # 「大幅改善」= 前スクレイプ比で順位がこれ以上改善（通常変動を除外） VN_TOP_ENTER = 5 # 「上位帯入り」= cur.rank がこの順位以内（最上位付近に限定） # 共通 CLUSTER_TOL_MIN = 4 # この分数以内の更新時刻は同じ「定刻スロット」として束ねる SUPPORT_MIN = 0.5 # スロットが該当日の50%以上で出現→規則的 TIGHT_SPREAD_MIN = 5 # スロット内時刻のばらつき(分)がこれ以下→ツールらしい MAX_GAP_MIN = 9 # prev,cur の間隔がこれ超なら更新イベントとして採用しない UPDATE_COUNT_CAP = 70 # 1日あたり推定スロット数がこれ超は非現実的として irregular に落とす WINDOWS = [7, 30] WEEKDAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] # --------------------------------------------------------------------------- # DDL（新規テーブル作成用 / 既存テーブルのマイグレーションは ensure_schema で対応） # --------------------------------------------------------------------------- _CREATE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS competitor_update_patterns ( id INT AUTO_INCREMENT PRIMARY KEY, client_id INT NOT NULL, media_name VARCHAR(20) NOT NULL, shop_name VARCHAR(255) NOT NULL, sub_area VARCHAR(100) NOT NULL DEFAULT '', window_days TINYINT NOT NULL, is_tool TINYINT NOT NULL DEFAULT 0, pattern_type VARCHAR(10) NOT NULL, confidence VARCHAR(6) NOT NULL, update_count INT NOT NULL DEFAULT 0, sample_days INT NOT NULL DEFAULT 0, bump_times_json TEXT, computed_at DATETIME NOT NULL, UNIQUE KEY uq (client_id, media_name, shop_name, sub_area, window_days) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ TBL = "competitor_update_patterns" # --------------------------------------------------------------------------- # スキーマヘルパー # --------------------------------------------------------------------------- def _col_exists(table: str, column: str) -> bool: rows = execute_query( "SELECT COUNT(*) AS cnt FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s", (table, column), local=True, ) return bool(rows and rows[0]["cnt"] > 0) def _idx_exists(table: str, index_name: str) -> bool: rows = execute_query( "SELECT COUNT(*) AS cnt FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND INDEX_NAME = %s", (table, index_name), local=True, ) return bool(rows and rows[0]["cnt"] > 0) def _idx_has_col(table: str, index_name: str, column: str) -> bool: rows = execute_query( "SELECT COUNT(*) AS cnt FROM information_schema.STATISTICS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s " " AND INDEX_NAME = %s AND COLUMN_NAME = %s", (table, index_name, column), local=True, ) return bool(rows and rows[0]["cnt"] > 0) def _col_nullable(table: str, column: str) -> bool: """カラムが NULL 許可かどうかを返す。カラムが存在しない場合は False。""" rows = execute_query( "SELECT IS_NULLABLE FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s", (table, column), local=True, ) return bool(rows and rows[0]["IS_NULLABLE"] == "YES") def ensure_schema() -> None: """テーブル作成・カラム修正・ユニークキー再構築（冪等）。 新規: CREATE TABLE で sub_area(NOT NULL DEFAULT '')/uq を一括作成。 既存: 以下を情報スキーマで確認してから安全に実行。 1. sub_area 列がなければ追加（NOT NULL DEFAULT ''） 2. sub_area が NULL 許可なら NULL→'' 埋め後に NOT NULL へ変更 3. 旧 _sub_area_key 生成列が残っていれば DROP 4. uq が sub_area を含まない古い構成なら DROP → 再作成 """ # 1. テーブル確認（新規の場合は CREATE で一括） execute_update(_CREATE_TABLE_SQL, (), local=True) # 2. sub_area カラムが無ければ追加 if not _col_exists(TBL, "sub_area"): execute_update( f"ALTER TABLE {TBL} ADD COLUMN sub_area VARCHAR(100) NOT NULL DEFAULT '' AFTER shop_name", (), local=True, ) logger.info("sub_area カラムを追加しました") elif _col_nullable(TBL, "sub_area"): # 既存 NULL 許可の sub_area を NOT NULL DEFAULT '' に変更 execute_update( f"UPDATE {TBL} SET sub_area = '' WHERE sub_area IS NULL", (), local=True, ) execute_update( f"ALTER TABLE {TBL} MODIFY COLUMN sub_area VARCHAR(100) NOT NULL DEFAULT ''", (), local=True, ) logger.info("sub_area カラムを NOT NULL DEFAULT '' に変更しました") # 3. 旧 _sub_area_key 生成列が残っている場合は DROP if _col_exists(TBL, "_sub_area_key"): execute_update( f"ALTER TABLE {TBL} DROP COLUMN _sub_area_key", (), local=True, ) logger.info("_sub_area_key 生成カラムを削除しました") # 4. uq インデックスが sub_area を含まない旧構成なら差し替え if _idx_exists(TBL, "uq") and not _idx_has_col(TBL, "uq", "sub_area"): execute_update(f"ALTER TABLE {TBL} DROP INDEX uq", (), local=True) execute_update( f"ALTER TABLE {TBL} ADD UNIQUE KEY uq " f"(client_id, media_name, shop_name, sub_area, window_days)", (), local=True, ) logger.info("uq インデックスを sub_area 対応版（5カラム）に更新しました") elif not _idx_exists(TBL, "uq"): execute_update( f"ALTER TABLE {TBL} ADD UNIQUE KEY uq " f"(client_id, media_name, shop_name, sub_area, window_days)", (), local=True, ) logger.info("uq インデックスを新規追加しました") logger.info("competitor_update_patterns スキーマ確認完了") # --------------------------------------------------------------------------- # 自店名取得（media_settings → clients フォールバック） # --------------------------------------------------------------------------- def get_own_shop_name(client_id: int, media_name: str) -> str | None: rows = execute_query( "SELECT shop_name FROM media_settings " "WHERE client_id = %s AND media_name = %s AND is_active = 1 LIMIT 1", (client_id, media_name), local=False, ) if rows and rows[0]["shop_name"]: return rows[0]["shop_name"] rows = execute_query( "SELECT shop_name FROM clients WHERE id = %s LIMIT 1", (client_id,), local=False, ) return rows[0]["shop_name"] if rows else None # --------------------------------------------------------------------------- # データ取得 # --------------------------------------------------------------------------- def get_rival_data( client_id: int, media_name: str, window_days: int, own_shop_name: str | None, ) -> pd.DataFrame: """GH: 競合スクレイピングデータ取得（自店除外）。""" start_date = date.today() - timedelta(days=window_days) rows = execute_query( """ SELECT shop_name, `rank`, fetched_at FROM scraping_results WHERE client_id = %s AND media_name = %s AND DATE(fetched_at) >= %s ORDER BY shop_name, fetched_at """, (client_id, media_name, start_date), local=True, ) if not rows: return pd.DataFrame() df = pd.DataFrame(rows) df["fetched_at"] = pd.to_datetime(df["fetched_at"]) if own_shop_name: df = df[df["shop_name"] != own_shop_name].copy() return df.reset_index(drop=True) def get_rival_data_vanilla( client_id: int, window_days: int, own_shop_name: str | None, ) -> pd.DataFrame: """バニラ: 競合スクレイピングデータ取得。 - rank > 0 かつ sub_area IS NOT NULL のみ（圏外・エリア不明を除外） - (shop_name, sub_area) の組で識別 - 自店(shop_name一致)を除外 """ start_date = date.today() - timedelta(days=window_days) rows = execute_query( """ SELECT shop_name, sub_area, `rank`, fetched_at FROM scraping_results WHERE client_id = %s AND media_name = 'vanilla' AND DATE(fetched_at) >= %s AND `rank` > 0 AND sub_area IS NOT NULL AND sub_area != '' ORDER BY shop_name, sub_area, fetched_at """, (client_id, start_date), local=True, ) if not rows: return pd.DataFrame() df = pd.DataFrame(rows) df["fetched_at"] = pd.to_datetime(df["fetched_at"]) if own_shop_name: df = df[df["shop_name"] != own_shop_name].copy() return df.reset_index(drop=True) # --------------------------------------------------------------------------- # 更新イベント抽出 # --------------------------------------------------------------------------- def extract_bump_events(shop_df: pd.DataFrame) -> tuple[list[dict], int]: """GH専用: トップ遷移（1〜GH_TOP_POS位に入った瞬間）を更新イベントとして抽出。 判定条件（連続2点 prev→cur）: - gap_min > MAX_GAP_MIN はスキップ（欠損区間の中点誤推定を防ぐ） - cur.rank が 1〜GH_TOP_POS の範囲内 - prev.rank が 0（圏外）または GH_TOP_POS 超（トップ帯の外） → トップ帯に入った瞬間だけを1イベントとして数える。 トップ帯を保持し続ける区間は計上しない。 Returns: (events, gap_skipped) events: 各イベントの dict: est_min(0-1439), range_min, date, weekday(mon..sun) gap_skipped: MAX_GAP_MIN 超でスキップした区間数 """ events: list[dict] = [] gap_skipped = 0 arr = shop_df.reset_index(drop=True) for i in range(len(arr) - 1): prev = arr.iloc[i] cur = arr.iloc[i + 1] pr = int(prev["rank"]) cr = int(cur["rank"]) diff_sec = (cur["fetched_at"] - prev["fetched_at"]).total_seconds() gap_min = diff_sec / 60 if gap_min > MAX_GAP_MIN: gap_skipped += 1 continue # cur がトップ帯（1〜GH_TOP_POS）かつ prev がトップ帯の外（圏外 or 3位以下） if not (1 <= cr <= GH_TOP_POS): continue if not (pr == 0 or pr > GH_TOP_POS): continue mid_dt = prev["fetched_at"] + timedelta(seconds=diff_sec / 2) range_min = math.ceil(gap_min / 2) events.append({ "est_min": mid_dt.hour * 60 + mid_dt.minute, "range_min": range_min, "date": mid_dt.date(), "weekday": WEEKDAYS[mid_dt.weekday()], }) return events, gap_skipped def extract_bump_events_vanilla(shop_df: pd.DataFrame) -> tuple[list[dict], int]: """バニラ用: 大幅改善 ∩ 上位帯入りを更新イベントとして抽出。 判定条件（連続2点 prev→cur、同じ sub_area 内）: - gap_min > MAX_GAP_MIN はスキップ - prev.rank > 0 かつ cur.rank > 0 - (prev.rank - cur.rank) >= VN_IMPROVE_MIN（大幅改善） - cur.rank <= VN_TOP_ENTER（上位帯入り） Returns: (events, gap_skipped) """ events: list[dict] = [] gap_skipped = 0 arr = shop_df.reset_index(drop=True) for i in range(len(arr) - 1): prev = arr.iloc[i] cur = arr.iloc[i + 1] pr = int(prev["rank"]) cr = int(cur["rank"]) diff_sec = (cur["fetched_at"] - prev["fetched_at"]).total_seconds() gap_min = diff_sec / 60 if gap_min > MAX_GAP_MIN: gap_skipped += 1 continue if not (pr > 0 and cr > 0): continue if (pr - cr) < VN_IMPROVE_MIN: continue if cr > VN_TOP_ENTER: continue mid_dt = prev["fetched_at"] + timedelta(seconds=diff_sec / 2) range_min = math.ceil(gap_min / 2) events.append({ "est_min": mid_dt.hour * 60 + mid_dt.minute, "range_min": range_min, "date": mid_dt.date(), "weekday": WEEKDAYS[mid_dt.weekday()], }) return events, gap_skipped # --------------------------------------------------------------------------- # 時刻クラスタリング # --------------------------------------------------------------------------- def cluster_events(events: list[dict], sample_days: int) -> list[dict]: """est_min を昇順にクラスタリングし、support >= SUPPORT_MIN の規則スロットを返す。 Returns: 各スロットの dict: center(分), spread(分), support(0-1), days, range_min, by_weekday """ if not events: return [] sorted_ev = sorted(events, key=lambda e: e["est_min"]) clusters: list[list[dict]] = [] cur_cluster = [sorted_ev[0]] for ev in sorted_ev[1:]: if ev["est_min"] - cur_cluster[-1]["est_min"] <= CLUSTER_TOL_MIN: cur_cluster.append(ev) else: clusters.append(cur_cluster) cur_cluster = [ev] clusters.append(cur_cluster) slots: list[dict] = [] for cluster in clusters: mins = [e["est_min"] for e in cluster] ranges = [e["range_min"] for e in cluster] days_set = {e["date"] for e in cluster} days = len(days_set) support = days / sample_days if sample_days > 0 else 0.0 if support < SUPPORT_MIN: continue center = int(median(mins)) spread = stdev(mins) if len(mins) >= 2 else 0.0 avg_range = int(median(ranges)) by_weekday: dict[str, list[int]] = defaultdict(list) for ev in cluster: by_weekday[ev["weekday"]].append(ev["est_min"]) slots.append({ "center": center, "spread": spread, "support": support, "days": days, "range_min": avg_range, "by_weekday": {wd: int(median(m)) for wd, m in by_weekday.items()}, }) return slots # --------------------------------------------------------------------------- # 曜日パターン判定（window=30 専用） # --------------------------------------------------------------------------- def is_weekday_pattern(slots: list[dict]) -> bool: """規則スロットの曜日別時刻を比較し、曜日依存かどうかを判定する。 - 任意スロットで曜日間の時刻差が CLUSTER_TOL_MIN 超 → True - 任意スロットが 5曜日未満にしか出現しない（特定曜日のみ）→ True """ for slot in slots: bw = slot["by_weekday"] if len(bw) >= 2: times = list(bw.values()) if max(times) - min(times) > CLUSTER_TOL_MIN: return True if len(bw) < 5: return True return False # --------------------------------------------------------------------------- # confidence 判定 # --------------------------------------------------------------------------- def calc_confidence(slots: list[dict], sample_days: int) -> str: if not slots: return "low" best = max(slots, key=lambda s: s["support"]) sup = best["support"] spread = best["spread"] if sup >= 0.7 and spread <= 4 and sample_days >= 14: return "high" if sup >= 0.5 and spread <= 8: return "mid" return "low" # --------------------------------------------------------------------------- # bump_times_json 構築 # --------------------------------------------------------------------------- def _min_to_hhmm(m: int) -> str: return f"{m // 60:02d}:{m % 60:02d}" def build_bump_times_daily(slots: list[dict]) -> list[dict]: return sorted( [{"time": _min_to_hhmm(s["center"]), "range_min": s["range_min"]} for s in slots], key=lambda x: x["time"], ) def build_bump_times_weekday(slots: list[dict]) -> dict: result: dict[str, list[dict]] = {} for slot in slots: for wd, center in slot["by_weekday"].items(): result.setdefault(wd, []).append( {"time": _min_to_hhmm(center), "range_min": slot["range_min"]} ) return { wd: sorted(result[wd], key=lambda x: x["time"]) for wd in WEEKDAYS if wd in result } # --------------------------------------------------------------------------- # 1 shop の解析 # --------------------------------------------------------------------------- _IRREGULAR = { "is_tool": 0, "pattern_type": "irregular", "confidence": "low", "update_count": 0, "bump_times_json": "[]", } def _build_result(events: list[dict], gap_skipped: int, window_days: int, sample_days: int) -> dict: """cluster → pattern_type → confidence を解析し結果 dict を返す（GH/バニラ共通）。""" raw_event_count = len(events) extra = { "_gap_skipped": gap_skipped, "_cap_downed": False, "_raw_event_count": raw_event_count, "_slot_count": 0, } if not events: return dict(_IRREGULAR, **extra) slots = cluster_events(events, sample_days) if not slots: return dict(_IRREGULAR, **extra) extra["_slot_count"] = len(slots) if window_days == 7 or not is_weekday_pattern(slots): pattern_type = "daily" bump_times = build_bump_times_daily(slots) update_count = len(slots) else: pattern_type = "weekday" bump_times = build_bump_times_weekday(slots) counts = [len(v) for v in bump_times.values()] update_count = int(sum(counts) / len(counts)) if counts else 0 if update_count > UPDATE_COUNT_CAP: return dict(_IRREGULAR, **dict(extra, _cap_downed=True)) confidence = calc_confidence(slots, sample_days) return { "is_tool": 1, "pattern_type": pattern_type, "confidence": confidence, "update_count": update_count, "bump_times_json": json.dumps(bump_times, ensure_ascii=False), **extra, } def analyze_shop( shop_df: pd.DataFrame, window_days: int, sample_days: int, ) -> dict: """1 GH shop のパターンを解析する（トップ遷移型）。""" events, gap_skipped = extract_bump_events(shop_df) return _build_result(events, gap_skipped, window_days, sample_days) def analyze_shop_vanilla( shop_df: pd.DataFrame, window_days: int, sample_days: int, ) -> dict: """1 バニラ (shop_name, sub_area) のパターンを解析する（大幅改善+上位帯入り型）。""" events, gap_skipped = extract_bump_events_vanilla(shop_df) return _build_result(events, gap_skipped, window_days, sample_days) # --------------------------------------------------------------------------- # DB 書き込み # --------------------------------------------------------------------------- # GH: UPSERT（sub_area='' で uq でユニーク判定） _UPSERT_SQL = """ INSERT INTO competitor_update_patterns (client_id, media_name, shop_name, sub_area, window_days, is_tool, pattern_type, confidence, update_count, sample_days, bump_times_json, computed_at) VALUES (%s, %s, %s, '', %s, %s, %s, %s, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE is_tool = VALUES(is_tool), pattern_type = VALUES(pattern_type), confidence = VALUES(confidence), update_count = VALUES(update_count), sample_days = VALUES(sample_days), bump_times_json = VALUES(bump_times_json), computed_at = NOW() """ # バニラ: 毎回 DELETE 後に plain INSERT（sub_area ありで同一 shop_name 複数エリア対応） _INSERT_VN_SQL = """ INSERT INTO competitor_update_patterns (client_id, media_name, shop_name, sub_area, window_days, is_tool, pattern_type, confidence, update_count, sample_days, bump_times_json, computed_at) VALUES (%s, 'vanilla', %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) """ def upsert_pattern( client_id: int, media_name: str, shop_name: str, window_days: int, result: dict, sample_days: int, ) -> None: """GH 用 UPSERT。""" execute_update( _UPSERT_SQL, ( client_id, media_name, shop_name, window_days, result["is_tool"], result["pattern_type"], result["confidence"], result["update_count"], sample_days, result["bump_times_json"], ), local=True, ) def insert_pattern_vn( client_id: int, shop_name: str, sub_area: str, window_days: int, result: dict, sample_days: int, ) -> None: """バニラ用 INSERT（DELETE 後なので重複なし）。""" execute_update( _INSERT_VN_SQL, ( client_id, shop_name, sub_area, window_days, result["is_tool"], result["pattern_type"], result["confidence"], result["update_count"], sample_days, result["bump_times_json"], ), local=True, ) # --------------------------------------------------------------------------- # 診断ログ共通ヘルパー # --------------------------------------------------------------------------- def _pop_diag(result: dict) -> tuple[int, int, int, bool]: """result dict から診断用キーを pop して返す。""" raw_ev = result.pop("_raw_event_count", 0) slot_count = result.pop("_slot_count", 0) gap_skip = result.pop("_gap_skipped", 0) cap_downed = result.pop("_cap_downed", False) return raw_ev, slot_count, gap_skip, cap_downed def _result_suffix(result: dict, cap_downed: bool) -> str: if cap_downed: return " [cap落ち]" if result["is_tool"]: return f" [{result['pattern_type']} {result['confidence']} ×{result['update_count']}]" return "" # --------------------------------------------------------------------------- # エントリポイント # --------------------------------------------------------------------------- def main() -> None: parser = argparse.ArgumentParser(description="競合更新パターン推定バッチ（GH + バニラ）") parser.add_argument("client_id", type=int, help="クライアントID") args = parser.parse_args() client_id = args.client_id ensure_schema() # ── GH ──────────────────────────────────────────────────────────────── own_shop_gh = get_own_shop_name(client_id, "girlsheaven") logger.info("[girlsheaven] 自店名: %s", own_shop_gh) for window_days in WINDOWS: df = get_rival_data(client_id, "girlsheaven", window_days, own_shop_gh) if df.empty: logger.warning("[girlsheaven] window=%d日: スクレイピングデータなし", window_days) continue shops = df["shop_name"].unique().tolist() tool_count = 0 total_gap_skip = 0 cap_down_count = 0 print(f"\n[girlsheaven] window={window_days}日 / 対象競合{len(shops)}店") print(f"{'店名':<30} {'生イベント':>9} {'日数':>4} {'平均/日':>7} {'スロット':>7}") print("-" * 68) for shop in shops: shop_df = ( df[df["shop_name"] == shop] .sort_values("fetched_at") .reset_index(drop=True) ) sample_days = shop_df["fetched_at"].dt.date.nunique() result = analyze_shop(shop_df, window_days, sample_days) raw_ev, slot_count, gap_skip, cap_downed = _pop_diag(result) total_gap_skip += gap_skip if cap_downed: cap_down_count += 1 avg_per_day = raw_ev / sample_days if sample_days > 0 else 0.0 print( f"{shop:<30} {raw_ev:>9} {sample_days:>4} {avg_per_day:>7.1f} {slot_count:>7}" + _result_suffix(result, cap_downed) ) upsert_pattern(client_id, "girlsheaven", shop, window_days, result, sample_days) if result["is_tool"]: tool_count += 1 print( f"\n→ ツール判定: {tool_count}店 / gap除外: {total_gap_skip}件 / cap落ち: {cap_down_count}店\n" ) logger.info( "[girlsheaven] window=%d 完了: %d店中%d店ツール / gap除外%d件 / cap落ち%d店", window_days, len(shops), tool_count, total_gap_skip, cap_down_count, ) # ── バニラ ───────────────────────────────────────────────────────────── own_shop_vn = get_own_shop_name(client_id, "vanilla") logger.info("[vanilla] 自店名: %s", own_shop_vn) # バニラは毎回全削除→再生成 deleted = execute_update( "DELETE FROM competitor_update_patterns WHERE client_id = %s AND media_name = 'vanilla'", (client_id,), local=True, ) if deleted > 0: logger.info("[vanilla] 旧データ削除: %d件", deleted) for window_days in WINDOWS: df_vn = get_rival_data_vanilla(client_id, window_days, own_shop_vn) if df_vn.empty: logger.warning("[vanilla] window=%d日: スクレイピングデータなし", window_days) continue # (shop_name, sub_area) の組を店舗単位とする shop_pairs = ( df_vn.groupby(["shop_name", "sub_area"], sort=True) .size() .index.tolist() ) tool_count = 0 total_gap_skip = 0 cap_down_count = 0 print(f"\n[vanilla] window={window_days}日 / 対象競合{len(shop_pairs)}エリア") print(f"{'店名 (エリア)':<42} {'生イベント':>9} {'日数':>4} {'平均/日':>7} {'スロット':>7}") print("-" * 82) for shop_name, sub_area in shop_pairs: shop_df = ( df_vn[ (df_vn["shop_name"] == shop_name) & (df_vn["sub_area"] == sub_area) ] .sort_values("fetched_at") .reset_index(drop=True) ) sample_days = shop_df["fetched_at"].dt.date.nunique() result = analyze_shop_vanilla(shop_df, window_days, sample_days) raw_ev, slot_count, gap_skip, cap_downed = _pop_diag(result) total_gap_skip += gap_skip if cap_downed: cap_down_count += 1 avg_per_day = raw_ev / sample_days if sample_days > 0 else 0.0 label = f"{shop_name} ({sub_area})" print( f"{label:<42} {raw_ev:>9} {sample_days:>4} {avg_per_day:>7.1f} {slot_count:>7}" + _result_suffix(result, cap_downed) ) insert_pattern_vn( client_id, shop_name, sub_area, window_days, result, sample_days ) if result["is_tool"]: tool_count += 1 print( f"\n→ ツール判定: {tool_count}エリア / gap除外: {total_gap_skip}件 / cap落ち: {cap_down_count}エリア\n" ) logger.info( "[vanilla] window=%d 完了: %dエリア中%dエリアツール / gap除外%d件 / cap落ち%dエリア", window_days, len(shop_pairs), tool_count, total_gap_skip, cap_down_count, ) logger.info("全処理完了") if __name__ == "__main__": main()