import pandas as pd import numpy as np import pymc as pm import arviz as az import threading from datetime import datetime import warnings warnings.filterwarnings('ignore') class BayesianHierarchicalAnalyzer: """ 貝氏階層模型分析器 用於分析寶可夢速度對勝率的影響（跨屬性） """ # 類別級的鎖，用於執行緒安全 _lock = threading.Lock() # 儲存各 session 的分析結果 _session_results = {} def __init__(self, session_id): """ 初始化分析器 Args: session_id: 唯一的 session 識別碼 """ self.session_id = session_id self.df = None self.model = None self.trace = None def load_data(self, csv_path_or_df): """ 載入資料 Args: csv_path_or_df: CSV 檔案路徑或 DataFrame Expected columns: - Trial_Type: 屬性名稱 (e.g., Water, Fire, Grass) - rc: 控制組（速度慢）的勝場數 - nc: 控制組的總場數 - rt: 實驗組（速度快）的勝場數 - nt: 實驗組的總場數 """ if isinstance(csv_path_or_df, str): self.df = pd.read_csv(csv_path_or_df) else: self.df = csv_path_or_df.copy() # 驗證必要欄位 required_cols = ['Trial_Type', 'rc', 'nc', 'rt', 'nt'] missing_cols = [col for col in required_cols if col not in self.df.columns] if missing_cols: raise ValueError(f"資料缺少必要欄位: {missing_cols}") return True def validate_data(self): """驗證資料有效性""" if self.df is None: raise ValueError("請先載入資料") # 檢查數值欄位 for col in ['rc', 'nc', 'rt', 'nt']: if not pd.api.types.is_numeric_dtype(self.df[col]): raise ValueError(f"欄位 {col} 必須是數值類型") # 檢查邏輯約束 if (self.df['rc'] > self.df['nc']).any(): raise ValueError("rc (勝場數) 不能大於 nc (總場數)") if (self.df['rt'] > self.df['nt']).any(): raise ValueError("rt (勝場數) 不能大於 nt (總場數)") return True def run_analysis(self, n_samples=2000, n_tune=1000, n_chains=2, target_accept=0.95): """ 執行貝氏階層模型分析 Args: n_samples: MCMC 抽樣數 n_tune: 調整期樣本數 n_chains: 鏈數 target_accept: 目標接受率 Returns: dict: 包含所有分析結果的字典 """ with self._lock: try: self.validate_data() # 準備資料 trial_labels = self.df['Trial_Type'].values num_trials = len(self.df) # 建立模型 with pm.Model() as self.model: # --- 先驗分佈 (Priors) --- d = pm.Normal('d', mu=0, sigma=10) tau = pm.Gamma('tau', alpha=0.001, beta=0.001) sigma = pm.Deterministic('sigma', 1 / pm.math.sqrt(tau)) # --- 各屬性特定效應 (Trial-specific effects) --- mu = pm.Normal('mu', mu=0, sigma=10, shape=num_trials) delta = pm.Normal('delta', mu=d, sigma=1 / pm.math.sqrt(tau), shape=num_trials) # --- 轉換與似然函數 (Logit Link & Likelihood) --- pc = pm.Deterministic('pc', pm.math.invlogit(mu)) pt = pm.Deterministic('pt', pm.math.invlogit(mu + delta)) rc_obs = pm.Binomial('rc_obs', n=self.df['nc'].values, p=pc, observed=self.df['rc'].values) rt_obs = pm.Binomial('rt_obs', n=self.df['nt'].values, p=pt, observed=self.df['rt'].values) # --- 其他統計量 --- delta_new = pm.Normal('delta_new', mu=d, sigma=1 / pm.math.sqrt(tau)) or_speed = pm.Deterministic('or_speed', pm.math.exp(d)) # 執行 MCMC 抽樣 self.trace = pm.sample( draws=n_samples, tune=n_tune, chains=n_chains, target_accept=target_accept, return_inferencedata=True, progressbar=False, # 在 Streamlit 中關閉進度條 discard_tuned_samples=False # 👈 加這行!保留 tune 樣本 ) # 生成摘要統計 summary = az.summary(self.trace, var_names=['d', 'sigma', 'or_speed'], hdi_prob=0.95) # 計算各屬性的 delta 統計量 delta_posterior = self.trace.posterior['delta'].values.reshape(-1, num_trials) delta_mean = delta_posterior.mean(axis=0) delta_std = delta_posterior.std(axis=0) delta_hdi = az.hdi(self.trace, var_names=['delta'], hdi_prob=0.95)['delta'].values # 判斷顯著性（HDI 不包含 0） delta_significant = (delta_hdi[:, 0] > 0) | (delta_hdi[:, 1] < 0) # 計算控制組和實驗組的勝率 pc_posterior = self.trace.posterior['pc'].values.reshape(-1, num_trials) pt_posterior = self.trace.posterior['pt'].values.reshape(-1, num_trials) pc_mean = pc_posterior.mean(axis=0) pt_mean = pt_posterior.mean(axis=0) # 整理結果 results = { 'timestamp': datetime.now().isoformat(), 'n_trials': num_trials, 'trial_labels': trial_labels.tolist(), # 整體效應 'overall': { 'd_mean': float(summary.loc['d', 'mean']), 'd_sd': float(summary.loc['d', 'sd']), 'd_hdi_low': float(summary.loc['d', 'hdi_2.5%']), 'd_hdi_high': float(summary.loc['d', 'hdi_97.5%']), 'sigma_mean': float(summary.loc['sigma', 'mean']), 'sigma_sd': float(summary.loc['sigma', 'sd']), 'sigma_hdi_low': float(summary.loc['sigma', 'hdi_2.5%']), 'sigma_hdi_high': float(summary.loc['sigma', 'hdi_97.5%']), 'or_mean': float(summary.loc['or_speed', 'mean']), 'or_sd': float(summary.loc['or_speed', 'sd']), 'or_hdi_low': float(summary.loc['or_speed', 'hdi_2.5%']), 'or_hdi_high': float(summary.loc['or_speed', 'hdi_97.5%']), }, # 各屬性的效應 'by_trial': { 'delta_mean': delta_mean.tolist(), 'delta_std': delta_std.tolist(), 'delta_hdi_low': delta_hdi[:, 0].tolist(), 'delta_hdi_high': delta_hdi[:, 1].tolist(), 'delta_significant': delta_significant.tolist(), 'pc_mean': pc_mean.tolist(), 'pt_mean': pt_mean.tolist(), }, # 原始資料 'data': self.df.to_dict('records'), # 模型參數 'model_params': { 'n_samples': n_samples, 'n_tune': n_tune, 'n_chains': n_chains, 'target_accept': target_accept }, # 收斂診斷 'diagnostics': self._compute_diagnostics(summary), # 解釋 'interpretation': self._interpret_results( summary.loc['or_speed', 'mean'], summary.loc['or_speed', 'hdi_2.5%'], summary.loc['or_speed', 'hdi_97.5%'], summary.loc['sigma', 'mean'] ) } # 儲存到 session results self._session_results[self.session_id] = results return results except Exception as e: raise Exception(f"分析失敗: {str(e)}") def _compute_diagnostics(self, summary): """計算收斂診斷指標""" try: # R-hat (應該接近 1.0) rhat_d = float(summary.loc['d', 'r_hat']) if 'r_hat' in summary.columns else None rhat_sigma = float(summary.loc['sigma', 'r_hat']) if 'r_hat' in summary.columns else None # ESS (有效樣本數) ess_d = float(summary.loc['d', 'ess_bulk']) if 'ess_bulk' in summary.columns else None ess_sigma = float(summary.loc['sigma', 'ess_bulk']) if 'ess_bulk' in summary.columns else None return { 'rhat_d': rhat_d, 'rhat_sigma': rhat_sigma, 'ess_d': ess_d, 'ess_sigma': ess_sigma, 'converged': (rhat_d is None or rhat_d < 1.1) and (rhat_sigma is None or rhat_sigma < 1.1) } except: return { 'converged': None, 'rhat_d': None, 'rhat_sigma': None, 'ess_d': None, 'ess_sigma': None } def _interpret_results(self, or_mean, or_low, or_high, sigma_mean): """解釋分析結果""" # 整體效應顯著性 if or_low > 1: overall_effect = "火系寶可夢相對於水系顯著更容易獲勝" overall_significance = "顯著正效應" elif or_high < 1: overall_effect = "水系寶可夢相對於火系顯著更容易獲勝" overall_significance = "顯著負效應" else: overall_effect = "火系與水系勝率無顯著差異" overall_significance = "不顯著" # 效果大小 if or_mean > 2: effect_size = "大效果 (OR > 2) - 火系有明顯優勢" elif or_mean > 1.5: effect_size = "中等效果 (OR > 1.5) - 火系有一定優勢" elif or_mean > 1: effect_size = "小效果 (OR > 1) - 火系略有優勢" elif or_mean == 1: effect_size = "無差異 (OR = 1) - 火系與水系勢均力敵" elif or_mean > 0.67: effect_size = "小效果 (OR < 1) - 水系略有優勢" elif or_mean > 0.5: effect_size = "中等效果 (OR < 0.67) - 水系有一定優勢" else: effect_size = "大效果 (OR < 0.5) - 水系有明顯優勢" # 異質性評估 if sigma_mean > 0.5: heterogeneity = "高異質性 - 不同配對的勝率差異很大" elif sigma_mean > 0.3: heterogeneity = "中等異質性 - 不同配對的勝率有一定差異" else: heterogeneity = "低異質性 - 不同配對的勝率相對一致" return { 'overall_effect': overall_effect, 'overall_significance': overall_significance, 'effect_size': effect_size, 'heterogeneity': heterogeneity } def get_model_graph(self): """生成模型 DAG 圖（返回 graphviz 物件）""" if self.model is None: raise ValueError("請先執行分析") try: gv = pm.model_to_graphviz(self.model) return gv except Exception as e: raise Exception(f"無法生成 DAG 圖: {str(e)}") @classmethod def get_session_results(cls, session_id): """獲取特定 session 的結果""" return cls._session_results.get(session_id) @classmethod def clear_session_results(cls, session_id): """清除特定 session 的結果""" if session_id in cls._session_results: del cls._session_results[session_id]