{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"codemirror_mode":{"name":"ipython","version":3},"file_extension":".py","mimetype":"text/x-python","name":"python","nbconvert_exporter":"python","pygments_lexer":"ipython3","version":"3.11"},"kaggle":{"accelerator":"none","dataSources":[{"sourceType":"competition","sourceId":7163,"databundleVersionId":44582,"isSourceIdPinned":false}],"dockerImageVersionId":31328,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"id":"04d6d7fc-25a4-4129-953e-2e40f9028915","cell_type":"markdown","source":"# KKBOX Churn Feature Extraction Pipeline\n\nNotebook nay duoc viet lai theo huong feature-store first: doc du lieu nguon, tao label churn theo thang, aggregate user logs cua thang truoc, merge member profile, va tao ra 2 lop feature:\n\n- lop feature so cho modeling\n- lop feature semantics cho dashboard BI trong `project-realtime-bi/docs/project_desc.md`\n\nNotebook dung o buoc xuat feature store. Khong con train LightGBM hay sinh submission trong file nay.\n","metadata":{}},{"id":"068dd0fa-92cb-4591-a434-4c1be9da94c1","cell_type":"code","source":"from pathlib import Path\nimport shutil\nimport subprocess\n\nKAGGLE_INPUT_DIR = Path('/kaggle/input/competitions/kkbox-churn-prediction-challenge')\nPIPELINE_REQUIRED_FILES = [\n    'members_v3.csv',\n    'transactions.csv',\n    'transactions_v2.csv',\n    'user_logs.csv',\n    'user_logs_v2.csv',\n    'sample_submission_v2.csv',\n]\nSEVEN_ZIP_BIN = shutil.which('7z') or shutil.which('7za') or shutil.which('7zr')\n\ndef resolve_source_file(file_name: str, data_dir: Path) -> Path:\n    csv_path = data_dir / file_name\n    archive_path = data_dir / f'{file_name}.7z'\n    if csv_path.exists():\n        return csv_path\n    if archive_path.exists():\n        return archive_path\n    raise FileNotFoundError(f'Khong tim thay {file_name} hoac {file_name}.7z trong {data_dir}')\n\n\nif KAGGLE_INPUT_DIR.exists():\n    DEFAULT_DATA_DIR = KAGGLE_INPUT_DIR\n    print('Using Kaggle competition input directly. Large .7z files will be streamed, not extracted to /kaggle/working.')\nelse:\n    DEFAULT_DATA_DIR = Path.cwd() / 'input'\n    print(f'Kaggle competition input not found. Fallback DATA_DIR = {DEFAULT_DATA_DIR}')\n\nfor file_name in PIPELINE_REQUIRED_FILES:\n    print(f'- {file_name}: {resolve_source_file(file_name, DEFAULT_DATA_DIR)}')\n\nif SEVEN_ZIP_BIN is None:\n    print('Warning: 7z binary not found. Reading .csv.7z directly will fail unless plain .csv files are available.')\n","metadata":{},"outputs":[],"execution_count":null},{"id":"b1d0304e","cell_type":"code","source":"from pathlib import Path\nfrom typing import Dict, Iterator, List\nimport gc\n\nimport numpy as np\nimport pandas as pd\nfrom tqdm.auto import tqdm\n\npd.set_option('display.max_columns', 200)\npd.set_option('display.max_rows', 200)\n\nPROJECT_DIR = Path.cwd()\nDATA_DIR = DEFAULT_DATA_DIR if 'DEFAULT_DATA_DIR' in globals() else PROJECT_DIR / 'input'\nOUTPUT_DIR = PROJECT_DIR / 'artifacts'\nFEATURE_STORE_DIR = OUTPUT_DIR / 'feature_store'\nLOG_AGG_CACHE_DIR = FEATURE_STORE_DIR / 'log_agg_cache'\nLOG_AGG_CACHE_VERSION = 'v1'\n\nALL_TARGET_MONTHS = [\n    201601, 201602, 201603, 201604, 201605, 201606,\n    201607, 201608, 201609, 201610, 201611, 201612,\n    201701, 201702, 201703, 201704,\n]\nTRAIN_MONTHS = [201701, 201702, 201703]\nTEST_MONTH = 201704\nPREVIOUS_MONTH = {\n    month: ALL_TARGET_MONTHS[idx - 1]\n    for idx, month in enumerate(ALL_TARGET_MONTHS)\n    if idx > 0\n}\nRAW_USER_LOG_FILES = ['user_logs.csv', 'user_logs_v2.csv']\nUSER_LOG_CHUNKSIZE = 4_000_000\nUSER_LOG_USECOLS = [\n    'msno', 'date', 'num_25', 'num_50', 'num_75', 'num_985', 'num_100', 'num_unq', 'total_secs'\n]\nUSER_LOG_FEATURE_COLS = [col for col in USER_LOG_USECOLS if col not in ['msno', 'date']]\nUSER_LOG_DTYPES = {\n    'date': 'int32',\n    'num_25': 'int16',\n    'num_50': 'int16',\n    'num_75': 'int16',\n    'num_985': 'int16',\n    'num_100': 'int16',\n    'num_unq': 'int16',\n    'total_secs': 'float32',\n}\nBI_DIMENSION_COLUMNS = [\n    'target_month',\n    'last_expire_month',\n    'age_segment',\n    'gender_profile',\n    'renewal_segment',\n    'price_segment',\n    'loyalty_segment',\n    'active_segment',\n    'skip_segment',\n    'discovery_segment',\n    'rfm_segment',\n    'bi_segment_name',\n]\n\nOUTPUT_DIR.mkdir(parents=True, exist_ok=True)\nFEATURE_STORE_DIR.mkdir(parents=True, exist_ok=True)\nLOG_AGG_CACHE_DIR.mkdir(parents=True, exist_ok=True)\nSAVE_FEATURE_STORE = True\nFEATURE_STORE_COMPRESSION = 'zstd'\n\nprint(f'PROJECT_DIR = {PROJECT_DIR}')\nprint(f'DATA_DIR = {DATA_DIR}')\nprint(f'OUTPUT_DIR = {OUTPUT_DIR}')\nprint(f'FEATURE_STORE_DIR = {FEATURE_STORE_DIR}')\nprint(f'LOG_AGG_CACHE_DIR = {LOG_AGG_CACHE_DIR}')\nprint(f'REQUIRED_LOG_MONTHS = {sorted({PREVIOUS_MONTH[month] for month in TRAIN_MONTHS + [TEST_MONTH]})}')\n","metadata":{},"outputs":[],"execution_count":null},{"id":"6cdea283","cell_type":"markdown","source":"## 1. Tao label churn theo thang va transaction features\n\nLogic label duoc viet lai theo huong nghiep vu cua `labeler.py`, nhung van giu dinh nghia snapshot hien tai:\n\n- snapshot cua `target_month` la anh chup trang thai user truoc khi thang do bat dau\n- khong bo record cancel khi xac dinh cohort; cancel duoc dung de cap nhat ngay het han hieu luc\n- user chi thuoc snapshot cua mot thang neu ngay het han hieu luc tinh den cuoi thang truoc roi vao thang do\n- label churn duoc gan bang cach nhin tu moc het han hieu luc nay xem user co quay lai trong vong `30` ngay hay khong\n- giu them cac thong tin lich su thanh toan, auto renew, huy goi, va khoang cach giao dich de phuc vu BI va modeling\n","metadata":{}},{"id":"c77712b4","cell_type":"code","source":"\n\ndef read_csv_from_source(data_dir: Path, file_name: str, **kwargs) -> pd.DataFrame:\n    source_path = resolve_source_file(file_name, data_dir)\n    if source_path.suffix == '.csv':\n        return pd.read_csv(source_path, **kwargs)\n\n    if SEVEN_ZIP_BIN is None:\n        raise RuntimeError(\n            f'Cannot read {source_path} without a 7z binary. Kaggle usually has 7z preinstalled.'\n        )\n\n    process = subprocess.Popen(\n        [SEVEN_ZIP_BIN, 'x', '-so', str(source_path)],\n        stdout=subprocess.PIPE,\n        stderr=subprocess.PIPE,\n    )\n    try:\n        df = pd.read_csv(process.stdout, **kwargs)\n        if process.stdout is not None:\n            process.stdout.close()\n        stderr_text = process.stderr.read().decode('utf-8', errors='ignore') if process.stderr is not None else ''\n        return_code = process.wait()\n        if return_code != 0:\n            raise RuntimeError(f'7z failed for {source_path}: {stderr_text}')\n        return df\n    except Exception:\n        process.kill()\n        process.wait()\n        raise\n\n\ndef iter_csv_chunks_from_source(\n    data_dir: Path,\n    file_name: str,\n    chunksize: int,\n    **kwargs,\n) -> Iterator[pd.DataFrame]:\n    source_path = resolve_source_file(file_name, data_dir)\n    if source_path.suffix == '.csv':\n        for chunk in pd.read_csv(source_path, chunksize=chunksize, **kwargs):\n            yield chunk\n        return\n\n    if SEVEN_ZIP_BIN is None:\n        raise RuntimeError(\n            f'Cannot stream {source_path} without a 7z binary. Kaggle usually has 7z preinstalled.'\n        )\n\n    process = subprocess.Popen(\n        [SEVEN_ZIP_BIN, 'x', '-so', str(source_path)],\n        stdout=subprocess.PIPE,\n        stderr=subprocess.PIPE,\n    )\n    try:\n        reader = pd.read_csv(process.stdout, chunksize=chunksize, **kwargs)\n        for chunk in reader:\n            yield chunk\n        if process.stdout is not None:\n            process.stdout.close()\n        stderr_text = process.stderr.read().decode('utf-8', errors='ignore') if process.stderr is not None else ''\n        return_code = process.wait()\n        if return_code != 0:\n            raise RuntimeError(f'7z failed for {source_path}: {stderr_text}')\n    except Exception:\n        process.kill()\n        process.wait()\n        raise\n\n\ndef load_transactions(data_dir: Path) -> pd.DataFrame:\n    transaction_files = ['transactions.csv', 'transactions_v2.csv']\n    parts = []\n    usecols = [\n        'msno',\n        'transaction_date',\n        'membership_expire_date',\n        'payment_method_id',\n        'payment_plan_days',\n        'plan_list_price',\n        'actual_amount_paid',\n        'is_auto_renew',\n        'is_cancel',\n    ]\n    dtypes = {\n        'transaction_date': 'int32',\n        'membership_expire_date': 'int32',\n        'payment_method_id': 'int16',\n        'payment_plan_days': 'int16',\n        'plan_list_price': 'float32',\n        'actual_amount_paid': 'float32',\n        'is_auto_renew': 'int8',\n        'is_cancel': 'int8',\n    }\n    for file_name in transaction_files:\n        parts.append(read_csv_from_source(data_dir, file_name, usecols=usecols, dtype=dtypes))\n\n    transactions = pd.concat(parts, ignore_index=True)\n    invalid_expire_mask = transactions['membership_expire_date'] < transactions['transaction_date']\n    transactions['invalid_expire_before_txn'] = invalid_expire_mask.astype('int8')\n    transactions.loc[invalid_expire_mask, 'membership_expire_date'] = transactions.loc[\n        invalid_expire_mask, 'transaction_date'\n    ]\n\n    transactions['transaction_month'] = (transactions['transaction_date'] // 100).astype('int32')\n    transactions['expire_month'] = (transactions['membership_expire_date'] // 100).astype('int32')\n    transactions['transaction_dt'] = pd.to_datetime(\n        transactions['transaction_date'].astype(str), format='%Y%m%d', errors='coerce'\n    )\n    transactions['expire_dt'] = pd.to_datetime(\n        transactions['membership_expire_date'].astype(str), format='%Y%m%d', errors='coerce'\n    )\n    transactions = transactions.dropna(subset=['transaction_dt', 'expire_dt'])\n    transactions = transactions.sort_values(\n        ['msno', 'transaction_date', 'membership_expire_date']\n    ).reset_index(drop=True)\n    return transactions\n\n\ndef _same_day_order_key(\n    idx: int,\n    membership_expire_dates: np.ndarray,\n    plan_list_prices: np.ndarray,\n    payment_plan_days: np.ndarray,\n    payment_method_ids: np.ndarray,\n    is_cancels: np.ndarray,\n) -> tuple:\n    is_cancel = int(is_cancels[idx])\n    expire_order = -int(membership_expire_dates[idx]) if is_cancel == 1 else int(membership_expire_dates[idx])\n    return (\n        -float(plan_list_prices[idx]),\n        -int(payment_plan_days[idx]),\n        -int(payment_method_ids[idx]),\n        is_cancel,\n        expire_order,\n    )\n\n\ndef _select_effective_history_idx(\n    history_end_idx: int,\n    transaction_dates: np.ndarray,\n    membership_expire_dates: np.ndarray,\n    plan_list_prices: np.ndarray,\n    payment_plan_days: np.ndarray,\n    payment_method_ids: np.ndarray,\n    is_cancels: np.ndarray,\n) -> int | None:\n    if history_end_idx <= 0:\n        return None\n\n    last_txn_date = int(transaction_dates[history_end_idx - 1])\n    block_start = int(np.searchsorted(transaction_dates, last_txn_date, side='left'))\n    best_idx = block_start\n    best_key = _same_day_order_key(\n        best_idx,\n        membership_expire_dates,\n        plan_list_prices,\n        payment_plan_days,\n        payment_method_ids,\n        is_cancels,\n    )\n\n    for idx in range(block_start + 1, history_end_idx):\n        candidate_key = _same_day_order_key(\n            idx,\n            membership_expire_dates,\n            plan_list_prices,\n            payment_plan_days,\n            payment_method_ids,\n            is_cancels,\n        )\n        if candidate_key >= best_key:\n            best_idx = idx\n            best_key = candidate_key\n\n    return int(best_idx)\n\n\ndef _calculate_renewal_gap_days(\n    future_start_idx: int,\n    anchor_expire_dt: np.datetime64,\n    transaction_dates: np.ndarray,\n    transaction_dts: np.ndarray,\n    expire_dts: np.ndarray,\n    membership_expire_dates: np.ndarray,\n    plan_list_prices: np.ndarray,\n    payment_plan_days: np.ndarray,\n    payment_method_ids: np.ndarray,\n    is_cancels: np.ndarray,\n) -> float | None:\n    if future_start_idx >= len(transaction_dates):\n        return None\n\n    effective_expire_dt = anchor_expire_dt\n    current_idx = future_start_idx\n    total_rows = len(transaction_dates)\n    while current_idx < total_rows:\n        current_txn_date = int(transaction_dates[current_idx])\n        day_end_idx = int(np.searchsorted(transaction_dates, current_txn_date, side='right'))\n        day_indices = list(range(current_idx, day_end_idx))\n        day_indices.sort(\n            key=lambda idx: _same_day_order_key(\n                idx,\n                membership_expire_dates,\n                plan_list_prices,\n                payment_plan_days,\n                payment_method_ids,\n                is_cancels,\n            )\n        )\n\n        for idx in day_indices:\n            if int(is_cancels[idx]) == 1:\n                if expire_dts[idx] < effective_expire_dt:\n                    effective_expire_dt = expire_dts[idx]\n                continue\n\n            gap_days = float((transaction_dts[idx] - effective_expire_dt) / np.timedelta64(1, 'D'))\n            return gap_days\n\n        current_idx = day_end_idx\n\n    return None\n\n\ndef build_month_labels(\n    transactions: pd.DataFrame,\n    history_months: List[int],\n    output_months: List[int],\n) -> Dict[int, pd.DataFrame]:\n    output_months_set = set(output_months)\n    outputs: Dict[int, List[dict]] = {month: [] for month in output_months}\n    grouped = transactions.groupby('msno', sort=False)\n    ordered_history_months = [month for month in history_months if month in PREVIOUS_MONTH]\n\n    for msno, user_txn in tqdm(grouped, total=transactions['msno'].nunique(), desc='Building labels'):\n        user_txn = user_txn.reset_index(drop=True)\n        transaction_dates = user_txn['transaction_date'].to_numpy()\n        membership_expire_dates = user_txn['membership_expire_date'].to_numpy()\n        transaction_months = user_txn['transaction_month'].to_numpy()\n        transaction_dts = user_txn['transaction_dt'].to_numpy()\n        expire_dts = user_txn['expire_dt'].to_numpy()\n        is_cancels = user_txn['is_cancel'].to_numpy(dtype=np.int8)\n        payment_method_ids = user_txn['payment_method_id'].to_numpy()\n        payment_plan_days_all = user_txn['payment_plan_days'].to_numpy()\n        plan_list_prices = user_txn['plan_list_price'].to_numpy()\n        actual_amount_paids = user_txn['actual_amount_paid'].to_numpy()\n        is_auto_renews = user_txn['is_auto_renew'].to_numpy(dtype=np.int8)\n        invalid_expire_flags = user_txn['invalid_expire_before_txn'].to_numpy(dtype=np.int8)\n\n        churn_history: List[int] = []\n\n        for target_month in ordered_history_months:\n            history_end_idx = int(np.searchsorted(transaction_months, target_month, side='left'))\n            if history_end_idx <= 0:\n                continue\n\n            effective_idx = _select_effective_history_idx(\n                history_end_idx,\n                transaction_dates,\n                membership_expire_dates,\n                plan_list_prices,\n                payment_plan_days_all,\n                payment_method_ids,\n                is_cancels,\n            )\n            if effective_idx is None:\n                continue\n\n            effective_expire_month = int(membership_expire_dates[effective_idx] // 100)\n            if effective_expire_month != target_month:\n                continue\n\n            renewal_gap_days = _calculate_renewal_gap_days(\n                history_end_idx,\n                expire_dts[effective_idx],\n                transaction_dates,\n                transaction_dts,\n                expire_dts,\n                membership_expire_dates,\n                plan_list_prices,\n                payment_plan_days_all,\n                payment_method_ids,\n                is_cancels,\n            )\n            # No valid renewal means churn; negative gaps are early renewals and should be non-churn.\n            is_churn = int(renewal_gap_days is None or renewal_gap_days >= 30)\n\n            history_len = len(churn_history)\n            history_tail = [churn_history[-offset] if history_len >= offset else -1 for offset in range(1, 6)]\n            churn_count = int(sum(churn_history)) if history_len > 0 else -1\n            churn_rate = (sum(churn_history) / history_len) if history_len > 0 else -1\n            transaction_count = history_len if history_len > 0 else -1\n\n            payment_plan_days = payment_plan_days_all[effective_idx]\n            actual_amount_paid = actual_amount_paids[effective_idx]\n            plan_list_price = plan_list_prices[effective_idx]\n            discount = plan_list_price - actual_amount_paid\n            amt_per_day = actual_amount_paid / payment_plan_days if payment_plan_days not in [0, -1] else -1\n\n            historical_transaction_rows = int(history_end_idx)\n            history_slice = slice(0, history_end_idx)\n            historical_cancel_count = int(is_cancels[history_slice].sum()) if historical_transaction_rows > 0 else -1\n            historical_cancel_rate = (\n                historical_cancel_count / historical_transaction_rows if historical_transaction_rows > 0 else -1\n            )\n            historical_auto_renew_rate = (\n                float(is_auto_renews[history_slice].mean()) if historical_transaction_rows > 0 else -1\n            )\n            historical_paid_total = (\n                float(actual_amount_paids[history_slice].sum()) if historical_transaction_rows > 0 else -1\n            )\n            historical_paid_mean = (\n                float(actual_amount_paids[history_slice].mean()) if historical_transaction_rows > 0 else -1\n            )\n            historical_list_price_mean = (\n                float(plan_list_prices[history_slice].mean()) if historical_transaction_rows > 0 else -1\n            )\n\n            if effective_idx > 0:\n                prev_idx = effective_idx - 1\n                days_since_previous_transaction = float(\n                    (transaction_dts[effective_idx] - transaction_dts[prev_idx]) / np.timedelta64(1, 'D')\n                )\n            else:\n                days_since_previous_transaction = -1\n\n            churn_history.append(is_churn)\n            if target_month in output_months_set:\n                outputs[target_month].append(\n                    {\n                        'msno': msno,\n                        'target_month': target_month,\n                        'is_churn': is_churn,\n                        'expire_date': int(membership_expire_dates[effective_idx]),\n                        'transaction_date': int(transaction_dates[effective_idx]),\n                        'payment_method_id': payment_method_ids[effective_idx],\n                        'payment_plan_days': payment_plan_days,\n                        'plan_list_price': plan_list_price,\n                        'actual_amount_paid': actual_amount_paid,\n                        'is_auto_renew': is_auto_renews[effective_idx],\n                        'invalid_expire_before_txn': invalid_expire_flags[effective_idx],\n                        'last_1_is_churn': history_tail[0],\n                        'last_2_is_churn': history_tail[1],\n                        'last_3_is_churn': history_tail[2],\n                        'last_4_is_churn': history_tail[3],\n                        'last_5_is_churn': history_tail[4],\n                        'churn_rate': churn_rate,\n                        'churn_count': churn_count,\n                        'transaction_count': transaction_count,\n                        'discount': discount,\n                        'is_discount': int(discount > 0),\n                        'amt_per_day': amt_per_day,\n                        'historical_transaction_rows': historical_transaction_rows,\n                        'historical_paid_total': historical_paid_total,\n                        'historical_paid_mean': historical_paid_mean,\n                        'historical_list_price_mean': historical_list_price_mean,\n                        'historical_cancel_count': historical_cancel_count,\n                        'historical_cancel_rate': historical_cancel_rate,\n                        'historical_auto_renew_rate': historical_auto_renew_rate,\n                        'days_since_previous_transaction': days_since_previous_transaction,\n                    }\n                )\n\n    output_columns = [\n        'msno',\n        'target_month',\n        'is_churn',\n        'expire_date',\n        'transaction_date',\n        'payment_method_id',\n        'payment_plan_days',\n        'plan_list_price',\n        'actual_amount_paid',\n        'is_auto_renew',\n        'invalid_expire_before_txn',\n        'last_1_is_churn',\n        'last_2_is_churn',\n        'last_3_is_churn',\n        'last_4_is_churn',\n        'last_5_is_churn',\n        'churn_rate',\n        'churn_count',\n        'transaction_count',\n        'discount',\n        'is_discount',\n        'amt_per_day',\n        'historical_transaction_rows',\n        'historical_paid_total',\n        'historical_paid_mean',\n        'historical_list_price_mean',\n        'historical_cancel_count',\n        'historical_cancel_rate',\n        'historical_auto_renew_rate',\n        'days_since_previous_transaction',\n    ]\n\n    result = {}\n    for month, rows in outputs.items():\n        frame = pd.DataFrame(rows, columns=output_columns)\n        if not frame.empty:\n            frame = frame.sort_values('msno').reset_index(drop=True)\n        result[month] = frame\n    return result\n\n\ntransactions = load_transactions(DATA_DIR)\nlabels_by_month = build_month_labels(\n    transactions=transactions,\n    history_months=ALL_TARGET_MONTHS,\n    output_months=TRAIN_MONTHS + [TEST_MONTH],\n)\ndel transactions\ngc.collect()\n\nfor month, frame in labels_by_month.items():\n    print(month, frame.shape)\n    if frame.empty:\n        raise ValueError(f'Label frame cho month {month} rong. Kiem tra lai logic cohort / transactions.')\n    if not frame['msno'].is_unique:\n        duplicate_count = int(frame['msno'].duplicated().sum())\n        raise ValueError(f'Label frame cho month {month} co {duplicate_count} msno bi duplicate.')\n    if not frame.empty:\n        display(frame[['msno', 'target_month', 'is_churn', 'expire_date', 'transaction_date']].head())\n","metadata":{},"outputs":[],"execution_count":null},{"id":"d461c417","cell_type":"markdown","source":"## 2. Aggregate user logs cua thang truoc trong RAM\n\nBuoc nay giu chi cac user va cac thang can cho pipeline, dong thoi them cleaning va thong tin BI:\n\n- cap `total_secs` toi da 86,400 giay moi ngay\n- dem so dong log bi cap de phuc vu data quality\n- luu `last_log_date` de tinh recency\n","metadata":{}},{"id":"946f28e8","cell_type":"code","source":"def _log_agg_cache_path(month: int) -> Path:\n    return LOG_AGG_CACHE_DIR / f'user_log_agg_{LOG_AGG_CACHE_VERSION}_{month}.parquet'\n\n\ndef _pack_log_agg_payload(\n    sum_df: pd.DataFrame,\n    count_series: pd.Series,\n    last_log_date: pd.Series,\n    capped_count: pd.Series,\n) -> pd.DataFrame:\n    packed = sum_df.copy()\n    packed['count'] = count_series.astype('float32')\n    packed['last_log_date'] = last_log_date.astype('Int32')\n    packed['capped_log_count'] = capped_count.astype('float32')\n    return packed.reset_index()\n\n\ndef _unpack_log_agg_payload(frame: pd.DataFrame) -> Dict[str, pd.DataFrame | pd.Series]:\n    indexed = frame.set_index('msno').sort_index()\n    return {\n        'sum': indexed[USER_LOG_FEATURE_COLS].copy(),\n        'count': indexed['count'].rename('count').copy(),\n        'last_log_date': indexed['last_log_date'].rename('last_log_date').copy(),\n        'capped_count': indexed['capped_log_count'].rename('capped_log_count').copy(),\n    }\n\n\ndef aggregate_required_user_logs(\n    data_dir: Path,\n    labels_by_month: Dict[int, pd.DataFrame],\n    train_months: List[int],\n    test_month: int,\n    chunksize: int = USER_LOG_CHUNKSIZE,\n    use_cache: bool = True,\n) -> Dict[int, Dict[str, pd.DataFrame | pd.Series]]:\n    month_to_msnos = {\n        PREVIOUS_MONTH[month]: set(labels_by_month[month]['msno'].tolist())\n        for month in train_months + [test_month]\n    }\n    required_months = sorted(month_to_msnos.keys())\n    month_to_msno_index = {month: pd.Index(sorted(msnos)) for month, msnos in month_to_msnos.items()}\n\n    aggregated = {\n        month: {'sum': None, 'count': None, 'last_log_date': None, 'capped_count': None}\n        for month in required_months\n    }\n\n    months_to_build = []\n    cached_months = []\n    for month in required_months:\n        cache_path = _log_agg_cache_path(month)\n        required_index = month_to_msno_index[month]\n        if use_cache and cache_path.exists():\n            cached_frame = pd.read_parquet(cache_path)\n            cached_index = pd.Index(cached_frame['msno'])\n            missing_required_msnos = required_index.difference(cached_index)\n            if len(missing_required_msnos) == 0:\n                cached_frame = cached_frame[cached_frame['msno'].isin(required_index)]\n                aggregated[month] = _unpack_log_agg_payload(cached_frame)\n                cached_months.append(month)\n                continue\n        months_to_build.append(month)\n\n    if cached_months:\n        print(f'Loaded log aggregation cache for months: {cached_months}')\n\n    if not months_to_build:\n        return aggregated\n\n    required_msnos = set().union(*(month_to_msnos[month] for month in months_to_build))\n    required_month_values = np.array(months_to_build, dtype=np.int32)\n\n    for file_name in RAW_USER_LOG_FILES:\n        source_path = resolve_source_file(file_name, data_dir)\n        reader = iter_csv_chunks_from_source(\n            data_dir,\n            file_name,\n            chunksize=chunksize,\n            usecols=USER_LOG_USECOLS,\n            dtype=USER_LOG_DTYPES,\n        )\n        for chunk in tqdm(reader, desc=f'Aggregating {source_path.name}'):\n            if 'date' not in chunk.columns or 'msno' not in chunk.columns:\n                raise ValueError(f\"Columns 'msno' and 'date' are required in {source_path}\")\n\n            year_month = (chunk['date'].to_numpy(copy=False) // 100).astype(np.int32, copy=False)\n            month_mask = np.isin(year_month, required_month_values)\n            if not month_mask.any():\n                continue\n\n            msno_mask = chunk['msno'].isin(required_msnos).to_numpy()\n            keep_mask = month_mask & msno_mask\n            if not keep_mask.any():\n                continue\n\n            reduced = chunk.loc[keep_mask, ['msno', 'date', *USER_LOG_FEATURE_COLS]].copy()\n            capped_mask = reduced['total_secs'] > 86400\n            reduced['total_secs'] = reduced['total_secs'].clip(lower=0, upper=86400)\n            reduced['year_month'] = year_month[keep_mask]\n            reduced['_row_count'] = np.int32(1)\n            reduced['_capped_secs_flag'] = capped_mask.astype('int8')\n\n            grouped = reduced.groupby(['year_month', 'msno'], sort=False, observed=True).agg(\n                {\n                    'num_25': 'sum',\n                    'num_50': 'sum',\n                    'num_75': 'sum',\n                    'num_985': 'sum',\n                    'num_100': 'sum',\n                    'num_unq': 'sum',\n                    'total_secs': 'sum',\n                    '_row_count': 'sum',\n                    '_capped_secs_flag': 'sum',\n                    'date': 'max',\n                }\n            )\n\n            for month in grouped.index.get_level_values(0).unique():\n                month = int(month)\n                month_group = grouped.xs(month, level=0, drop_level=True)\n                valid_mask = month_group.index.isin(month_to_msno_index[month])\n                if not valid_mask.any():\n                    continue\n\n                month_group = month_group.loc[valid_mask]\n                month_sum = month_group[USER_LOG_FEATURE_COLS]\n                month_count = month_group['_row_count'].rename('count')\n                month_last_log_date = month_group['date'].rename('last_log_date')\n                month_capped_count = month_group['_capped_secs_flag'].rename('capped_log_count')\n\n                current_sum = aggregated[month]['sum']\n                current_count = aggregated[month]['count']\n                current_last_log_date = aggregated[month]['last_log_date']\n                current_capped_count = aggregated[month]['capped_count']\n\n                if current_sum is None:\n                    aggregated[month]['sum'] = month_sum\n                    aggregated[month]['count'] = month_count\n                    aggregated[month]['last_log_date'] = month_last_log_date\n                    aggregated[month]['capped_count'] = month_capped_count\n                else:\n                    aggregated[month]['sum'] = current_sum.add(month_sum, fill_value=0)\n                    aggregated[month]['count'] = current_count.add(month_count, fill_value=0)\n                    aggregated[month]['capped_count'] = current_capped_count.add(month_capped_count, fill_value=0)\n                    aggregated[month]['last_log_date'] = current_last_log_date.combine(month_last_log_date, max)\n            del reduced, grouped, chunk\n        gc.collect()\n\n    missing_months = [month for month in months_to_build if aggregated[month]['sum'] is None]\n    if missing_months:\n        raise ValueError(f'No raw user log rows found for required months: {missing_months}')\n\n    if use_cache:\n        for month in months_to_build:\n            packed = _pack_log_agg_payload(\n                aggregated[month]['sum'],\n                aggregated[month]['count'],\n                aggregated[month]['last_log_date'],\n                aggregated[month]['capped_count'],\n            )\n            packed.to_parquet(_log_agg_cache_path(month), index=False, compression='zstd')\n\n    return aggregated\n\n\naggregated_user_logs = aggregate_required_user_logs(\n    data_dir=DATA_DIR,\n    labels_by_month=labels_by_month,\n    train_months=TRAIN_MONTHS,\n    test_month=TEST_MONTH,\n    chunksize=USER_LOG_CHUNKSIZE,\n)\n\nfor month, payload in aggregated_user_logs.items():\n    print(month, payload['sum'].shape, payload['count'].shape)\n","metadata":{},"outputs":[],"execution_count":null},{"id":"91dcd456-3a2b-414f-8c58-6bd5ee0af4a6","cell_type":"markdown","source":"## 3. Build feature store cho modeling va BI\n\nLop feature nay bo sung them:\n\n- feature hanh vi nghe nhac: `skip_ratio`, `discovery_ratio`, `completion_ratio`\n- feature RFM va loyalty\n- cac segment phuc vu dashboard: `price_segment`, `loyalty_segment`, `active_segment`, `skip_segment`, `discovery_segment`\n- bo output rieng cho modeling va cho BI de tranh lam vo workflow train model numeric ve sau\n","metadata":{}},{"id":"6baae8f6","cell_type":"code","source":"def safe_divide(numerator: pd.Series, denominator: pd.Series, default: float = 0.0) -> pd.Series:\n    numerator = numerator.astype('float32')\n    denominator = denominator.astype('float32')\n    result = numerator.div(denominator.replace(0, np.nan))\n    return result.replace([np.inf, -np.inf], np.nan).fillna(default).astype('float32')\n\n\ndef build_user_log_features(\n    base_df: pd.DataFrame,\n    prev_month: int,\n    aggregated_user_logs: Dict[int, Dict[str, pd.DataFrame | pd.Series]],\n) -> pd.DataFrame:\n    payload = aggregated_user_logs.get(prev_month)\n    if payload is None or payload['sum'] is None:\n        raise ValueError(f'Missing aggregated user log payload for month {prev_month}')\n\n    sum_df = payload['sum'].copy()\n    count_series = payload['count'].copy()\n    last_log_date = payload['last_log_date'].copy()\n    capped_count = payload['capped_count'].copy()\n\n    mean_df = sum_df.div(count_series, axis=0)\n    mean_df = mean_df.rename(columns={col: f'{col}_mean' for col in mean_df.columns}).reset_index()\n    sum_df = sum_df.rename(columns={col: f'{col}_sum' for col in sum_df.columns}).reset_index()\n    count_df = count_series.reset_index()\n    last_log_date_df = last_log_date.reset_index()\n    capped_count_df = capped_count.reset_index()\n\n    feature_df = base_df.merge(mean_df, on='msno', how='left')\n    feature_df = feature_df.merge(sum_df, on='msno', how='left')\n    feature_df = feature_df.merge(count_df, on='msno', how='left')\n    feature_df = feature_df.merge(last_log_date_df, on='msno', how='left')\n    feature_df = feature_df.merge(capped_count_df, on='msno', how='left')\n    return feature_df\n\n\ndef load_members(data_dir: Path) -> pd.DataFrame:\n    members = read_csv_from_source(\n        data_dir,\n        'members_v3.csv',\n        usecols=['msno', 'city', 'bd', 'gender', 'registered_via', 'registration_init_time'],\n        dtype={\n            'city': 'float32',\n            'bd': 'float32',\n            'registered_via': 'float32',\n            'registration_init_time': 'int32',\n        },\n    )\n    members['gender'] = members['gender'].map({'male': 1, 'female': 2}).fillna(0).astype('int8')\n    members['bd'] = members['bd'].where(members['bd'].between(15, 65))\n    return members\n\n\ndef add_feature_layers(df: pd.DataFrame) -> pd.DataFrame:\n    enriched = df.copy()\n\n    transaction_dt = pd.to_datetime(\n        enriched['transaction_date'].astype('Int64').astype(str),\n        format='%Y%m%d',\n        errors='coerce',\n    )\n    expire_dt = pd.to_datetime(\n        enriched['expire_date'].astype('Int64').astype(str),\n        format='%Y%m%d',\n        errors='coerce',\n    )\n    registration_dt = pd.to_datetime(\n        enriched['registration_init_time'].astype('Int64').astype(str),\n        format='%Y%m%d',\n        errors='coerce',\n    )\n    last_log_dt = pd.to_datetime(\n        enriched['last_log_date'].astype('Int64').astype(str),\n        format='%Y%m%d',\n        errors='coerce',\n    )\n    snapshot_month_start = pd.to_datetime(\n        enriched['target_month'].astype('Int64').astype(str) + '01',\n        format='%Y%m%d',\n        errors='coerce',\n    )\n    snapshot_dt = snapshot_month_start - pd.Timedelta(days=1)\n    invalid_registration_after_snapshot = registration_dt.notna() & snapshot_dt.notna() & (registration_dt > snapshot_dt)\n    enriched['invalid_registration_after_snapshot'] = invalid_registration_after_snapshot.astype('int8')\n    if invalid_registration_after_snapshot.any():\n        enriched.loc[invalid_registration_after_snapshot, 'registration_init_time'] = np.nan\n        registration_dt = registration_dt.where(~invalid_registration_after_snapshot)\n\n    enriched['age'] = enriched['bd'].fillna(-1).astype('float32')\n    enriched['has_valid_age'] = (enriched['age'] >= 0).astype('int8')\n    enriched['transaction_month'] = (transaction_dt.dt.year * 100 + transaction_dt.dt.month).astype('Int32')\n    enriched['expire_month'] = (expire_dt.dt.year * 100 + expire_dt.dt.month).astype('Int32')\n    enriched['last_expire_month'] = enriched['expire_month']\n    enriched['transaction_day'] = transaction_dt.dt.day.astype('Int16')\n    enriched['expire_day'] = expire_dt.dt.day.astype('Int16')\n    enriched['registration_year'] = registration_dt.dt.year.astype('Int16')\n    enriched['registration_month'] = registration_dt.dt.month.astype('Int16')\n    enriched['registration_day'] = registration_dt.dt.day.astype('Int16')\n\n    enriched['days_to_expire'] = (expire_dt - snapshot_dt).dt.days.astype('float32')\n    enriched['membership_age_days'] = (snapshot_dt - registration_dt).dt.days.astype('float32')\n    enriched['days_since_last_listen'] = (snapshot_dt - last_log_dt).dt.days.astype('float32')\n    enriched['tenure_months'] = safe_divide(enriched['membership_age_days'], pd.Series(30.0, index=enriched.index), default=-1)\n\n    enriched['expected_renewal_amount'] = np.where(\n        enriched['actual_amount_paid'].fillna(0) > 0,\n        enriched['actual_amount_paid'],\n        enriched['plan_list_price'],\n    ).astype('float32')\n    enriched['price_gap'] = (enriched['plan_list_price'] - enriched['actual_amount_paid']).astype('float32')\n    enriched['discount_ratio'] = safe_divide(\n        enriched['plan_list_price'] - enriched['actual_amount_paid'],\n        enriched['plan_list_price'],\n        default=0.0,\n    )\n    enriched['payment_to_list_ratio'] = safe_divide(\n        enriched['actual_amount_paid'],\n        enriched['plan_list_price'],\n        default=0.0,\n    )\n    enriched['secs_per_log'] = safe_divide(enriched['total_secs_sum'], enriched['count'], default=0.0)\n    enriched['unique_per_log'] = safe_divide(enriched['num_unq_sum'], enriched['count'], default=0.0)\n    enriched['num100_per_log'] = safe_divide(enriched['num_100_sum'], enriched['count'], default=0.0)\n\n    weighted_completion = (\n        enriched['num_25_sum'] * 0.25\n        + enriched['num_50_sum'] * 0.50\n        + enriched['num_75_sum'] * 0.75\n        + enriched['num_985_sum'] * 0.985\n        + enriched['num_100_sum'] * 1.0\n    ).astype('float32')\n    enriched['weighted_completion_sum'] = weighted_completion\n    enriched['weighted_completion_per_log'] = safe_divide(weighted_completion, enriched['count'], default=0.0)\n    enriched['listen_events_sum'] = (\n        enriched['num_25_sum']\n        + enriched['num_50_sum']\n        + enriched['num_75_sum']\n        + enriched['num_985_sum']\n        + enriched['num_100_sum']\n    ).astype('float32')\n    enriched['skip_events_sum'] = (\n        enriched['num_25_sum']\n        + enriched['num_50_sum']\n        + enriched['num_75_sum']\n    ).astype('float32')\n    enriched['listen_events_per_log'] = safe_divide(enriched['listen_events_sum'], enriched['count'], default=0.0)\n    enriched['avg_secs_per_unique'] = safe_divide(enriched['total_secs_sum'], enriched['num_unq_sum'], default=0.0)\n    enriched['secs_per_plan_day'] = safe_divide(enriched['total_secs_sum'], enriched['payment_plan_days'], default=0.0)\n    enriched['uniques_per_plan_day'] = safe_divide(enriched['num_unq_sum'], enriched['payment_plan_days'], default=0.0)\n    enriched['logs_per_plan_day'] = safe_divide(enriched['count'], enriched['payment_plan_days'], default=0.0)\n    enriched['remaining_plan_ratio'] = safe_divide(enriched['days_to_expire'], enriched['payment_plan_days'], default=0.0)\n    enriched['completion_ratio'] = safe_divide(enriched['weighted_completion_sum'], enriched['listen_events_sum'], default=0.0)\n    enriched['skip_ratio'] = safe_divide(enriched['skip_events_sum'], enriched['listen_events_sum'], default=0.0)\n    enriched['discovery_ratio'] = safe_divide(enriched['num_unq_sum'], enriched['listen_events_sum'], default=0.0)\n    enriched['replay_ratio'] = (1 - enriched['discovery_ratio']).clip(lower=0, upper=1).astype('float32')\n    enriched['price_gap_per_plan_day'] = safe_divide(enriched['price_gap'], enriched['payment_plan_days'], default=0.0)\n    enriched['secs_per_paid_amount'] = safe_divide(enriched['total_secs_sum'], enriched['actual_amount_paid'], default=0.0)\n    enriched['capped_log_share'] = safe_divide(enriched['capped_log_count'], enriched['count'], default=0.0)\n\n    history_cols = [\n        col for col in ['last_1_is_churn', 'last_2_is_churn', 'last_3_is_churn', 'last_4_is_churn', 'last_5_is_churn']\n        if col in enriched.columns\n    ]\n    if history_cols:\n        history_frame = enriched[history_cols].replace(-1, 0).astype('float32')\n        weights = np.arange(len(history_cols), 0, -1, dtype='float32')\n        enriched['recent_churn_events'] = history_frame.sum(axis=1).astype('float32')\n        enriched['weighted_recent_churn'] = history_frame.to_numpy(dtype='float32') @ weights\n\n    enriched['is_expiring_user'] = 1\n    enriched['is_manual_renew'] = (enriched['is_auto_renew'] == 0).astype('int8')\n    enriched['high_skip_flag'] = ((enriched['listen_events_sum'] > 0) & (enriched['skip_ratio'] >= 0.5)).astype('int8')\n    enriched['low_discovery_flag'] = ((enriched['listen_events_sum'] > 0) & (enriched['discovery_ratio'] < 0.2)).astype('int8')\n    enriched['deal_hunter_flag'] = ((enriched['amt_per_day'] > 0) & (enriched['amt_per_day'] < 4.5)).astype('int8')\n    enriched['free_trial_flag'] = (enriched['expected_renewal_amount'] <= 0).astype('int8')\n    enriched['content_fatigue_flag'] = (\n        (enriched['listen_events_sum'] > 0)\n        & (enriched['skip_ratio'] >= 0.5)\n        & (enriched['discovery_ratio'] < 0.2)\n    ).astype('int8')\n    enriched['auto_renew_discount_interaction'] = (\n        enriched['is_auto_renew'].astype('float32') * enriched['is_discount'].astype('float32')\n    )\n    enriched['churn_rate_x_transaction_count'] = (\n        enriched['churn_rate'].astype('float32') * enriched['transaction_count'].astype('float32')\n    )\n\n    recency_score = np.select(\n        [\n            (enriched['listen_events_sum'] > 0) & (enriched['days_since_last_listen'] <= 7),\n            (enriched['listen_events_sum'] > 0) & (enriched['days_since_last_listen'] <= 21),\n            enriched['listen_events_sum'] > 0,\n        ],\n        [3, 2, 1],\n        default=0,\n    ).astype('int8')\n    frequency_score = np.select(\n        [\n            enriched['count'] > 15,\n            enriched['count'] > 5,\n            enriched['count'] > 0,\n        ],\n        [3, 2, 1],\n        default=0,\n    ).astype('int8')\n    monetary_score = np.select(\n        [\n            enriched['expected_renewal_amount'] >= 150,\n            enriched['expected_renewal_amount'] >= 100,\n            enriched['expected_renewal_amount'] > 0,\n        ],\n        [3, 2, 1],\n        default=0,\n    ).astype('int8')\n    enriched['rfm_recency_score'] = recency_score\n    enriched['rfm_frequency_score'] = frequency_score\n    enriched['rfm_monetary_score'] = monetary_score\n    enriched['rfm_total_score'] = (recency_score + frequency_score + monetary_score).astype('int8')\n\n    age_segment_code = pd.Series(np.select(\n        [\n            enriched['age'].between(15, 20),\n            enriched['age'].between(21, 25),\n            enriched['age'].between(26, 35),\n            enriched['age'].between(36, 50),\n            enriched['age'].between(51, 65),\n        ],\n        [1, 2, 3, 4, 5],\n        default=0,\n    ).astype('int8'), index=enriched.index)\n    price_segment_code = pd.Series(np.select(\n        [\n            enriched['amt_per_day'] <= 0,\n            (enriched['amt_per_day'] > 0) & (enriched['amt_per_day'] < 4.5),\n            (enriched['amt_per_day'] >= 4.5) & (enriched['amt_per_day'] < 6.5),\n            enriched['amt_per_day'] >= 6.5,\n        ],\n        [1, 2, 3, 4],\n        default=0,\n    ).astype('int8'), index=enriched.index)\n    loyalty_segment_code = pd.Series(np.select(\n        [\n            (enriched['membership_age_days'] >= 0) & (enriched['membership_age_days'] < 30),\n            (enriched['membership_age_days'] >= 30) & (enriched['membership_age_days'] < 180),\n            (enriched['membership_age_days'] >= 180) & (enriched['membership_age_days'] < 365),\n            enriched['membership_age_days'] >= 365,\n        ],\n        [1, 2, 3, 4],\n        default=0,\n    ).astype('int8'), index=enriched.index)\n    active_segment_code = pd.Series(np.select(\n        [\n            enriched['count'] <= 0,\n            (enriched['count'] > 0) & (enriched['count'] <= 5),\n            (enriched['count'] > 5) & (enriched['count'] <= 15),\n            enriched['count'] > 15,\n        ],\n        [1, 2, 3, 4],\n        default=0,\n    ).astype('int8'), index=enriched.index)\n    skip_segment_code = pd.Series(np.select(\n        [\n            enriched['listen_events_sum'] <= 0,\n            enriched['skip_ratio'] < 0.2,\n            enriched['skip_ratio'] < 0.5,\n            enriched['skip_ratio'] >= 0.5,\n        ],\n        [0, 1, 2, 3],\n        default=0,\n    ).astype('int8'), index=enriched.index)\n    discovery_segment_code = pd.Series(np.select(\n        [\n            enriched['listen_events_sum'] <= 0,\n            enriched['discovery_ratio'] < 0.2,\n            enriched['discovery_ratio'] < 0.5,\n            enriched['discovery_ratio'] >= 0.5,\n        ],\n        [0, 1, 2, 3],\n        default=0,\n    ).astype('int8'), index=enriched.index)\n    renewal_segment_code = pd.Series(np.select(\n        [\n            enriched['is_auto_renew'] == 1,\n            enriched['is_auto_renew'] == 0,\n        ],\n        [1, 2],\n        default=0,\n    ).astype('int8'), index=enriched.index)\n    rfm_segment_code = pd.Series(np.select(\n        [\n            enriched['rfm_total_score'] >= 8,\n            enriched['rfm_total_score'] >= 5,\n            enriched['rfm_total_score'] > 0,\n        ],\n        [3, 2, 1],\n        default=0,\n    ).astype('int8'), index=enriched.index)\n\n    enriched['age_segment_code'] = age_segment_code\n    enriched['price_segment_code'] = price_segment_code\n    enriched['loyalty_segment_code'] = loyalty_segment_code\n    enriched['active_segment_code'] = active_segment_code\n    enriched['skip_segment_code'] = skip_segment_code\n    enriched['discovery_segment_code'] = discovery_segment_code\n    enriched['renewal_segment_code'] = renewal_segment_code\n    enriched['rfm_segment_code'] = rfm_segment_code\n\n    age_label_map = {\n        0: 'Unknown',\n        1: '15-20',\n        2: '21-25',\n        3: '26-35',\n        4: '36-50',\n        5: '51-65',\n    }\n    price_label_map = {\n        0: 'Unknown',\n        1: 'Free Trial / Zero Pay',\n        2: 'Deal Hunter < 4.5',\n        3: 'Standard 4.5-6.5',\n        4: 'Premium >= 6.5',\n    }\n    loyalty_label_map = {\n        0: 'Unknown',\n        1: 'New < 30d',\n        2: 'Growing 30-179d',\n        3: 'Established 180-364d',\n        4: 'Loyal >= 365d',\n    }\n    active_label_map = {\n        0: 'Unknown',\n        1: 'Inactive',\n        2: 'Light 1-5 logs',\n        3: 'Active 6-15 logs',\n        4: 'Heavy > 15 logs',\n    }\n    skip_label_map = {\n        0: 'No Listening Data',\n        1: 'Low < 20%',\n        2: 'Medium 20-50%',\n        3: 'High >= 50%',\n    }\n    discovery_label_map = {\n        0: 'No Listening Data',\n        1: 'Habit < 20%',\n        2: 'Balanced 20-50%',\n        3: 'Explore >= 50%',\n    }\n    renewal_label_map = {\n        0: 'Unknown',\n        1: 'Pay_Auto-Renew',\n        2: 'Pay_Manual',\n    }\n    rfm_label_map = {\n        0: 'Unclassified',\n        1: 'Low Value',\n        2: 'Mid Value',\n        3: 'High Value',\n    }\n    gender_label_map = {\n        0: 'Unknown',\n        1: 'Male',\n        2: 'Female',\n    }\n\n    enriched['gender'] = enriched['gender'].fillna(0).astype('int8')\n    enriched['age_segment'] = age_segment_code.map(age_label_map)\n    enriched['price_segment'] = price_segment_code.map(price_label_map)\n    enriched['loyalty_segment'] = loyalty_segment_code.map(loyalty_label_map)\n    enriched['active_segment'] = active_segment_code.map(active_label_map)\n    enriched['skip_segment'] = skip_segment_code.map(skip_label_map)\n    enriched['discovery_segment'] = discovery_segment_code.map(discovery_label_map)\n    enriched['renewal_segment'] = renewal_segment_code.map(renewal_label_map)\n    enriched['rfm_segment'] = rfm_segment_code.map(rfm_label_map)\n    enriched['gender_profile'] = enriched['gender'].map(gender_label_map)\n\n    enriched['bi_segment_name'] = (\n        enriched['loyalty_segment'].astype(str)\n        + ' | '\n        + enriched['renewal_segment'].astype(str)\n        + ' | '\n        + enriched['price_segment'].astype(str)\n        + ' | '\n        + enriched['discovery_segment'].astype(str)\n    )\n\n    enriched = enriched.replace([np.inf, -np.inf], np.nan)\n    numeric_cols = enriched.select_dtypes(include=[np.number]).columns\n    enriched[numeric_cols] = enriched[numeric_cols].fillna(-1)\n    return enriched\n\n\nif 'labels_by_month' not in globals():\n    print('labels_by_month khong ton tai trong kernel. Dang rebuild tu transactions...')\n    transactions = load_transactions(DATA_DIR)\n    labels_by_month = build_month_labels(\n        transactions=transactions,\n        history_months=ALL_TARGET_MONTHS,\n        output_months=TRAIN_MONTHS + [TEST_MONTH],\n    )\n    del transactions\n    gc.collect()\n\nif 'aggregated_user_logs' not in globals():\n    raise NameError(\n        'aggregated_user_logs khong ton tai trong kernel. Hay chay cell aggregate_required_user_logs truoc.'\n    )\n\nmonthly_feature_frames: Dict[int, pd.DataFrame] = {}\nfor month in TRAIN_MONTHS + [TEST_MONTH]:\n    prev_month = PREVIOUS_MONTH[month]\n    frame = build_user_log_features(labels_by_month[month], prev_month, aggregated_user_logs)\n    monthly_feature_frames[month] = frame\n    print(month, monthly_feature_frames[month].shape)\n\n# Giu lai labels_by_month va aggregated_user_logs de notebook co the rerun tung cell an toan.\n# Neu can giai phong RAM, chi nen xoa o mot cleanup cell rieng sau khi da export xong artifact.\ngc.collect()\n","metadata":{},"outputs":[],"execution_count":null},{"id":"00f4d59c-10e1-4a9a-8c02-e011c136d460","cell_type":"code","source":"members = load_members(DATA_DIR)\n\nmonthly_bi_frames: Dict[int, pd.DataFrame] = {}\nfor month, frame in monthly_feature_frames.items():\n    enriched_frame = frame.merge(members, on='msno', how='left')\n    enriched_frame = add_feature_layers(enriched_frame)\n    monthly_bi_frames[month] = enriched_frame\n    print(month, enriched_frame.shape)\n\nbi_snapshot_df = pd.concat(\n    [monthly_bi_frames[month] for month in TRAIN_MONTHS + [TEST_MONTH]],\n    ignore_index=True,\n)\n\ninvalid_snapshot_recency = bi_snapshot_df[\n    (bi_snapshot_df['last_log_date'] > 0)\n    & (bi_snapshot_df['count'] > 0)\n    & (bi_snapshot_df['days_since_last_listen'] < 0)\n]\nif not invalid_snapshot_recency.empty:\n    raise ValueError(\n        'days_since_last_listen < 0 cho mot so dong co log thang truoc. Kiem tra lai moc snapshot_dt.'\n    )\n\ninvalid_days_to_expire = bi_snapshot_df[\n    (bi_snapshot_df['days_to_expire'] < 0) | (bi_snapshot_df['days_to_expire'] > 31)\n]\nif not invalid_days_to_expire.empty:\n    raise ValueError(\n        'days_to_expire nam ngoai khoang snapshot hop ly [0, 31]. Kiem tra lai cong thuc days_to_expire.'\n    )\n\ninvalid_membership_age = bi_snapshot_df[\n    (bi_snapshot_df['registration_init_time'] > 0)\n    & (bi_snapshot_df['membership_age_days'] < 0)\n]\nif not invalid_membership_age.empty:\n    raise ValueError(\n        'membership_age_days < 0 cho mot so user co registration_init_time hop le. Kiem tra lai moc snapshot_dt.'\n    )\n\ntrain_bi_df = pd.concat(\n    [monthly_bi_frames[month] for month in TRAIN_MONTHS],\n    ignore_index=True,\n)\ntest_feature_df = monthly_bi_frames[TEST_MONTH].drop(columns=['is_churn'])\n\nsample_submission = read_csv_from_source(DATA_DIR, 'sample_submission_v2.csv', usecols=['msno'])\nif not sample_submission['msno'].is_unique:\n    duplicate_count = int(sample_submission['msno'].duplicated().sum())\n    raise ValueError(f'sample_submission_v2.csv co {duplicate_count} msno bi duplicate.')\nif not test_feature_df['msno'].is_unique:\n    duplicate_count = int(test_feature_df['msno'].duplicated().sum())\n    raise ValueError(f'test_feature_df co {duplicate_count} msno bi duplicate truoc khi merge test cohort.')\n\nsample_msnos = pd.Index(sample_submission['msno'])\ntest_feature_msnos = pd.Index(test_feature_df['msno'])\nmissing_test_msnos = sample_msnos.difference(test_feature_msnos)\nunexpected_test_msnos = test_feature_msnos.difference(sample_msnos)\nsubmission_alignment_report = pd.DataFrame([\n    {\n        'sample_submission_rows': len(sample_submission),\n        'built_test_rows': len(test_feature_df),\n        'missing_from_built_test': len(missing_test_msnos),\n        'unexpected_in_built_test': len(unexpected_test_msnos),\n        'exact_match': int(len(missing_test_msnos) == 0 and len(unexpected_test_msnos) == 0),\n    }\n])\nif len(missing_test_msnos) > 0 or len(unexpected_test_msnos) > 0:\n    print(\n        'Warning: Test cohort tu build khong khop sample_submission_v2.csv: '\n        f'missing={len(missing_test_msnos)}, unexpected={len(unexpected_test_msnos)}'\n    )\n\ntest_bi_df = sample_submission.merge(test_feature_df, on='msno', how='left')\nif len(test_bi_df) != len(sample_submission):\n    raise ValueError('So dong test sau merge khong khop sample_submission_v2.csv.')\ntest_bi_df['missing_from_built_test'] = test_bi_df['target_month'].isna().astype('int8')\ntest_bi_df['target_month'] = test_bi_df['target_month'].fillna(TEST_MONTH).astype('Int32')\n\n# test_feature_df da la ban enriched, chi backfill thong tin members cho cac msno thieu.\nmember_columns = [column for column in members.columns if column != 'msno']\nmember_lookup = members[['msno', *member_columns]].copy()\ntest_bi_df = test_bi_df.merge(\n    member_lookup,\n    on='msno',\n    how='left',\n    suffixes=('', '_member'),\n)\n\nfor column in member_columns:\n    member_column = f'{column}_member'\n    if member_column in test_bi_df.columns:\n        test_bi_df[column] = test_bi_df[column].combine_first(test_bi_df[member_column])\n        test_bi_df = test_bi_df.drop(columns=[member_column])\n\nnumeric_test_cols = test_bi_df.select_dtypes(include=[np.number]).columns\ntest_bi_df[numeric_test_cols] = test_bi_df[numeric_test_cols].fillna(-1)\n\ndel members\ndel monthly_feature_frames\ndel monthly_bi_frames\ndel test_feature_df\ngc.collect()\n\nmodel_exclude = {'msno', 'is_churn', 'transaction_date', 'expire_date'}\nmodel_feature_cols = [\n    col for col in train_bi_df.columns\n    if col not in model_exclude and pd.api.types.is_numeric_dtype(train_bi_df[col])\n]\n\ntrain_model_df = train_bi_df[['msno', 'is_churn', 'transaction_date', 'expire_date', *model_feature_cols]].copy()\ntest_model_df = test_bi_df[['msno', 'transaction_date', 'expire_date', *model_feature_cols]].copy()\n\nif SAVE_FEATURE_STORE:\n    train_model_df.to_parquet(\n        FEATURE_STORE_DIR / 'train_features_all.parquet',\n        index=False,\n        compression=FEATURE_STORE_COMPRESSION,\n    )\n    test_model_df.to_parquet(\n        FEATURE_STORE_DIR / f'test_features_{TEST_MONTH}_full.parquet',\n        index=False,\n        compression=FEATURE_STORE_COMPRESSION,\n    )\n    bi_snapshot_df.to_parquet(\n        FEATURE_STORE_DIR / 'bi_feature_master.parquet',\n        index=False,\n        compression=FEATURE_STORE_COMPRESSION,\n    )\n    train_bi_df.to_parquet(\n        FEATURE_STORE_DIR / 'train_features_bi_all.parquet',\n        index=False,\n        compression=FEATURE_STORE_COMPRESSION,\n    )\n    test_bi_df.to_parquet(\n        FEATURE_STORE_DIR / f'test_features_bi_{TEST_MONTH}_full.parquet',\n        index=False,\n        compression=FEATURE_STORE_COMPRESSION,\n    )\n    pd.Series(model_feature_cols, name='feature').to_csv(\n        FEATURE_STORE_DIR / 'feature_columns.csv',\n        index=False,\n    )\n    pd.Series(BI_DIMENSION_COLUMNS, name='dimension').to_csv(\n        FEATURE_STORE_DIR / 'bi_dimension_columns.csv',\n        index=False,\n    )\n    submission_alignment_report.to_csv(\n        FEATURE_STORE_DIR / f'submission_alignment_{TEST_MONTH}.csv',\n        index=False,\n    )\n    print(f'Saved reusable feature store to: {FEATURE_STORE_DIR}')\n\nprint('train_model_df shape:', train_model_df.shape)\nprint('test_model_df shape:', test_model_df.shape)\nprint('bi_snapshot_df shape:', bi_snapshot_df.shape)\nprint('num model features:', len(model_feature_cols))\ndisplay(\n    bi_snapshot_df[\n        [\n            'msno',\n            'target_month',\n            'is_churn',\n            'expected_renewal_amount',\n            'skip_ratio',\n            'discovery_ratio',\n            'price_segment',\n            'loyalty_segment',\n            'active_segment',\n            'bi_segment_name',\n        ]\n    ].head()\n)\n","metadata":{},"outputs":[],"execution_count":null}]}