股票數據清洗與日K轉換周月年K工具 - 完整功能解析-3

def detect_pingpong_patterns(day_df: pd.DataFrame, price_col='收盤', threshold=0.4) -> pd.DataFrame:

    # ... (函式內容保持不變) ...

    df = day_df.sort_values('日期').reset_index(drop=True).copy()

    if price_col not in df.columns:

        return pd.DataFrame()

    df['pct_change'] = df[price_col].pct_change()

    rows = []

    for i in range(1, len(df)-1):

        prev_change = df.loc[i, 'pct_change']

        next_change = df.loc[i+1, 'pct_change']

        if pd.notna(prev_change) and pd.notna(next_change):

            if abs(prev_change) > threshold and abs(next_change) > threshold and prev_change * next_change < 0:

                rows.append({

                    '日期(異常日)': df.loc[i, '日期'],

                    '前日收盤': df.loc[i-1, price_col],

                    '當日收盤': df.loc[i, price_col],

                    '次日收盤': df.loc[i+1, price_col],

                    '前日變動(%)': round(prev_change * 100, 2),

                    '次日變動(%)': round(next_change * 100, 2),

                })

    return pd.DataFrame(rows)



def _process_one_file(path: str):

    try:

        df_day = _load_day_clean_full(path)

        # 僅保留 2024–2025（含起始緩衝）

        df_day = _clip_by_date_range(df_day, col='日期',

                                     start=DATE_START, end=DATE_END, pad_start=DATE_PAD_START)



        stem = Path(path).stem

        raw_id, _ = _parse_id_name(stem)

        cid = _canonical_id(raw_id) # StockID 必須在這裡定義



        if df_day.empty or df_day['日期'].max() < DATE_PAD_START:

            return "SKIP", {'StockID': cid, 'file': Path(path).name,

                            'count': 0, 'reason': 'out_of_range'}, f"{Path(path).name}: out_of_range"



        if SKIP_SUSPECT_STOCK:

            suspects = detect_pingpong_patterns(df_day, threshold=PINGPONG_THRESHOLD)

            if not suspects.empty:

                msg = f"{Path(path).name}: [SKIP_PINGPONG] 命中疑似除權錯位 {len(suspects)} 筆"

                # 將 cid 傳入 payload

                return "SKIP", {'StockID': cid, 'file': Path(path).name, 'count': len(suspects), 'reason': 'suspect_pingpong'}, msg



        # 週/月/年K 轉換

        dfw = _resample_ohlc_with_flags(df_day, 'W'); dfw = _add_period_returns(dfw, 'W'); dfw['成交量'] = dfw['成交量'].astype('float64'); dfw['StockID'] = cid

        dfm = _resample_ohlc_with_flags(df_day, 'M'); dfm = _add_period_returns(dfm, 'M'); dfm['成交量'] = dfm['成交量'].astype('float64'); dfm['StockID'] = cid

        dfy = _resample_ohlc_with_flags(df_day, 'Y'); dfy = _add_period_returns(dfy, 'Y'); dfy['成交量'] = dfy['成交量'].astype('float64'); dfy['StockID'] = cid



        # 週/月/年K 只留 2024–2025（不需要 pad）

        for _df in (dfw, dfm, dfy):

            _df.dropna(subset=['日期'], inplace=True)

            _df['日期'] = pd.to_datetime(_df['日期']).dt.tz_localize(None)

        dfw = dfw[(dfw['日期'] >= DATE_START) & (dfw['日期'] <= DATE_END)]

        dfm = dfm[(dfm['日期'] >= DATE_START) & (dfm['日期'] <= DATE_END)]

        dfy = dfy[(dfy['日期'] >= DATE_START) & (dfy['日期'] <= DATE_END)]



        return True, (dfw, dfm, dfy), None

    except Exception as e:

        return False, None, f"{Path(path).name}: {e}"



def _split_into_shards(items, shard_size):

    return [items[i:i+shard_size] for i in range(0, len(items), shard_size)] if shard_size > 0 else [items]



def build_wmy_parquets(market_key: str):

    base_dir = LOCAL_BASE if WRITE_MODE in ('A','B') else DRIVE_BASE

    market_abbr = MARKET_ABBR_MAP.get(market_key, market_key.replace('-share','').upper())



    # 1) 本地快取

    local_dayk_dir = _prepare_local_dayk(market_key)

    if not local_dayk_dir:

        return None, None, None



    # 2) 輸出路徑

    out_week  = f"{base_dir}/{market_key}/weekK_{market_abbr}.parquet"

    out_month = f"{base_dir}/{market_key}/monthK_{market_abbr}.parquet"

    out_year  = f"{base_dir}/{market_key}/yearK_{market_abbr}.parquet"

    Path(f"{base_dir}/{market_key}").mkdir(parents=True, exist_ok=True)



    # 3) 檔案清單

    files = sorted(glob.glob(f"{local_dayk_dir}/*.csv"))

    if not files:

        print(f"❌ 快取路徑 {local_dayk_dir} 內無檔案，程式終止。")

        return None, None, None



    shards = _split_into_shards(files, SHARD_SIZE)

    print(f"\n📦 [{market_key}] 共 {len(files)} 檔，分 {len(shards)} 批處理")



    # 調整欄位列表以包含 IsFiltered_QA

    base_front = ['StockID','日期','開盤','最高','最低','收盤','成交量']

    base_common_wmy = ['CurPeriod_Days','PrevPeriod_Days', 'HasResume', 'IsFiltered_QA']

    base_common_w = ['CurWeek_Days','PrevWeek_Days', 'HasResume']



    ret_cols = [f'PrevC_{f}' for f in ['W','M','Y']] + [f'Ret_Gap_{f}' for f in ['W','M','Y']] + \

               [f'Ret_Trad_{f}' for f in ['W','M','Y']] + [f'Ret_C_{f}' for f in ['W','M','Y']] + \

               [f'Ret_Max_H_{f}' for f in ['W','M','Y']] + [f'Ret_Min_L_{f}' for f in ['W','M','Y']] + \

               [f'Range_{f}' for f in ['W','M','Y']] + [f'Ret_Max_H_Pos_{f}' for f in ['W','M','Y']] + \

               [f'Ret_H_from_O_{f}' for f in ['W','M','Y']] + [f'Ret_L_from_O_{f}' for f in ['W','M','Y']] + \

               [f'Ret_End_RelH_{f}' for f in ['W','M','Y']] + [f'Ret_End_RelL_{f}' for f in ['W','M','Y']]

    ret_cols = sorted(list(set(ret_cols)))



    wk_cols = base_front + base_common_w + [c for c in ret_cols if c.endswith('_W')] + ['ISO_Week']

    mo_cols = base_front + base_common_wmy + [c for c in ret_cols if c.endswith('_M')]

    yr_cols = base_front + base_common_wmy + [c for c in ret_cols if c.endswith('_Y')]



    # 移除重複的 Period/Week 欄位

    mo_cols = [c for c in mo_cols if not c.startswith('CurWeek') and not c.startswith('PrevWeek')]

    yr_cols = [c for c in yr_cols if not c.startswith('CurWeek') and not c.startswith('PrevWeek')]

    wk_cols = [c for c in wk_cols if not c.startswith('CurPeriod') and not c.startswith('PrevPeriod') and c != 'IsFiltered_QA']



    w_writer = ParquetStreamer(out_week,  keep_cols=wk_cols)

    m_writer = ParquetStreamer(out_month, keep_cols=mo_cols)

    y_writer = ParquetStreamer(out_year,  keep_cols=yr_cols)



    total_ok = total_fail = 0

    skip_records = []



    t0 = time.time()

    for si, shard in enumerate(shards, 1):

        print(f"🧩 Shard {si}/{len(shards)}: {len(shard)} 檔")

        shard_w, shard_m, shard_y = [], [], []



        # [***修正點: 將 futs 和 as_completed 移入 with 區塊***]

        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex:

            # 建立 futures 字典，用於追蹤檔案路徑

            futs = {ex.submit(_process_one_file, p): p for p in shard}



            # 遍歷 futures

            for f in as_completed(futs):

                path = futs[f]  # 從字典中獲取當前檔案路徑

                try:

                    status, payload, err = f.result()

                except Exception as e:

                    # 處理執行緒層級的極端錯誤

                    total_fail += 1

                    err = f"{Path(path).name}: Thread execution failed ({e})"

                    if total_fail <= 20:

                        print(" ❌", err)

                    continue



                if status is True:

                    dfw, dfm, dfy = payload

                    shard_w.append(dfw); shard_m.append(dfm); shard_y.append(dfy)

                    total_ok += 1

                elif status == "SKIP":

                    info = payload or {}

                    # 從 payload 中安全獲取 StockID

                    stock_id = info.get('StockID', Path(path).stem.split("_")[0])

                    skip_records.append({'StockID': stock_id, 'file': Path(path).name,

                                         'reason': info.get('reason', 'suspect_pingpong'), 'count': info.get('count', 0)})

                    if total_fail + total_ok + len(skip_records) <= 20 and err:

                        print(" ⏭️", err)

                else:

                    total_fail += 1

                    if total_fail <= 20 and err:

                        print(" ⚠️", err)



        if shard_w: w_writer.append_df(pd.concat(shard_w, ignore_index=True))

        if shard_m: m_writer.append_df(pd.concat(shard_m, ignore_index=True))

        if shard_y: y_writer.append_df(pd.concat(shard_y, ignore_index=True))

        time.sleep(0.5 + random.random()*0.5)



    w_writer.close(); m_writer.close(); y_writer.close()

    print(f"⏱️ 批次耗時：{time.time() - t0:.1f} 秒")



    if skip_records:

        qa_dir = Path(f"{DRIVE_BASE}/{market_key}/_qa")

        qa_dir.mkdir(parents=True, exist_ok=True)

        pd.DataFrame(skip_records).to_csv(qa_dir / "skip_pingpong.csv", index=False, encoding='utf-8-sig')

        print(f"🧹 已輸出略過名單：{qa_dir / 'skip_pingpong.csv'}（{len(skip_records)} 檔）")



    market_abbr = MARKET_ABBR_MAP.get(market_key, market_key.replace('-share','').upper())

    final_week  = f"{DRIVE_BASE}/{market_key}/weekK_{market_abbr}.parquet"

    final_month = f"{DRIVE_BASE}/{market_key}/monthK_{market_abbr}.parquet"

    final_year  = f"{DRIVE_BASE}/{market_key}/yearK_{market_abbr}.parquet"



    if WRITE_MODE == 'A':

        for src, dst in [(out_week, final_week), (out_month, final_month), (out_year, final_year)]:

            if Path(src).exists():

                Path(dst).parent.mkdir(parents=True, exist_ok=True)

                Path(dst).unlink(missing_ok=True)

                Path(src).replace(dst)

        print("📤 A 模式：已將本機 Parquet 移動到 Google Drive")



    elif WRITE_MODE == 'B':

        for src, dst in [(out_week, final_week), (out_month, final_month), (out_year, final_year)]:

            if Path(src).exists():

                Path(dst).parent.mkdir(parents=True, exist_ok=True)

                copy2(str(src), str(dst))

        print("📤 B 模式：已將本機 Parquet 複製到 Google Drive（本機仍保留）")



    else:   # WRITE_MODE == 'C'

        final_week, final_month, final_year = out_week, out_month, out_year

        print("☁️ C 模式：直接寫入 Google Drive")



    return final_week, final_month, final_year



# --------------------------------------------------------------------------------

# Colab Cell 4: QA 稽核函式 (Part 4)

# --------------------------------------------------------------------------------



def _geom_chain_return(day_df: pd.DataFrame, start, end) -> float:

    # ... (函式內容保持不變) ...

    sub = day_df[(day_df['日期'] >= start) & (day_df['日期'] <= end)].copy().sort_values('日期')

    c = sub['收盤'].to_numpy(dtype='float64')

    if len(c) < 2 or not np.all(np.isfinite(c)) or np.any(c <= 0):

        return np.nan

    return float(np.prod(c[1:] / c[:-1]) - 1.0)



def _week_bins():

    # ... (函式內容保持不變) ...

    edges = [-1000, -200, -100, -90, -80, -70, -60, -50, -40, -30, -20, -10, 0,

             10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 200, 1000, 10000]

    labels = [f"{edges[i]}~{edges[i+1]}%" for i in range(len(edges)-1)]

    return edges, labels



def audit_weekly_parquet(market_key: str, week_path: str):

    # ... (函式內容保持不變) ...

    qa_dir = Path(f"{DRIVE_BASE}/{market_key}/_qa"); qa_dir.mkdir(parents=True, exist_ok=True)

    if not Path(week_path).exists():

        print("❌ QA: 找不到週K Parquet。")

        pd.DataFrame(columns=['ISO_Week','reason','max_abs_z']).to_csv(qa_dir / "weekly_drift_alerts.csv", index=False, encoding='utf-8-sig')

        pd.DataFrame(columns=['ISO_Week','日期','StockID','ret_pct','abs_z','bin']).to_csv(qa_dir / "weekly_top_outliers.csv", index=False, encoding='utf-8-sig')

        return



    dfw = pd.read_parquet(week_path)

    dfw['日期'] = pd.to_datetime(dfw['日期']).dt.tz_localize(None)

    dfw = dfw[(dfw['日期'] >= DATE_START) & (dfw['日期'] <= DATE_END)]



    edges, labels = _week_bins()

    wk = dfw[['ISO_Week','StockID','日期','PrevC_W','開盤','最高','最低','收盤','Ret_Trad_W']].dropna(subset=['Ret_Trad_W']).copy()

    wk['ret_pct'] = (wk['Ret_Trad_W'] * 100).astype('float64')

    wk['bin'] = pd.cut(wk['ret_pct'], bins=edges, labels=labels, right=False)



    dist = wk.pivot_table(index='ISO_Week', columns='bin', values='StockID', aggfunc='count', fill_value=0)

    dist = dist.div(dist.sum(axis=1).replace(0, np.nan), axis=0)

    dist.sort_index().to_csv(qa_dir / "weekly_distribution_summary.csv", encoding='utf-8-sig')



    roll_win = 26

    alerts = []

    if len(dist) >= roll_win + 1:

        for i in range(roll_win, len(dist)):

            cur = dist.iloc[i]

            base = dist.iloc[i-roll_win:i]

            mu = base.mean(); sigma = base.std(ddof=1).replace(0, np.nan)

            z = (cur - mu) / sigma

            big_tail = ['100~200%', '200~1000%', '1000~10000%']

            trig1 = (z.abs() > 5).any()

            trig2 = any((k in z.index) and np.isfinite(z[k]) and (z[k] > 3) for k in big_tail)

            if trig1 or trig2:

                alerts.append({'ISO_Week': dist.index[i], 'reason': 'distribution_drift', 'max_abs_z': float(z.abs().max(skipna=True))})

    pd.DataFrame(alerts).to_csv(qa_dir / "weekly_drift_alerts.csv", index=False, encoding='utf-8-sig')



    wk['abs_z'] = wk.groupby('ISO_Week')['ret_pct'].transform(lambda s: (s - s.mean()) / (s.std(ddof=1) if pd.notna(s.std(ddof=1)) and s.std(ddof=1)!=0 else np.nan)).abs()

    tail_mask = wk['bin'].isin(['100~200%', '200~1000%', '1000~10000%', '-1000~-200%', '-200~-100%'])

    outliers = wk.loc[tail_mask].copy()

    outliers['abs_z_rank'] = outliers.groupby('ISO_Week')['abs_z'].rank(ascending=False, method='first')

    outliers = outliers.sort_values(['ISO_Week','abs_z_rank']).groupby('ISO_Week').head(200)

    outliers[['ISO_Week','日期','StockID','ret_pct','abs_z','bin']].to_csv(qa_dir / "weekly_top_outliers.csv", index=False, encoding='utf-8-sig')

    print(f"🧪 QA 週K（收盤）分布/漂移/厚尾 → {qa_dir}")



# [已修改] 函式名稱: audit_weekly_vs_daily

def audit_weekly_vs_daily(market_key: str):

    """

    [🌟 修正路徑] QA 稽核路徑：改為讀取本地快取。

    """

    market_abbr = MARKET_ABBR_MAP.get(market_key, market_key)

    week_path = Path(f"{DRIVE_BASE}/{market_key}/weekK_{market_abbr}.parquet")

    qa_dir = Path(f"{DRIVE_BASE}/{market_key}/_qa"); qa_dir.mkdir(parents=True, exist_ok=True)



    if not week_path.exists():

        pd.DataFrame(columns=['StockID','ISO_Week','geom_day_chain','Ret_Trad_W','diff']).to_csv(qa_dir / "weekly_vs_daily_diff.csv", index=False, encoding='utf-8-sig')

        print("❌ QA: 找不到週K Parquet，跳過週 vs 日 對比。")

        return



    dfw = pd.read_parquet(week_path).sort_values(['StockID','日期'])

    dfw['日期'] = pd.to_datetime(dfw['日期']).dt.tz_localize(None)

    dfw = dfw[(dfw['日期'] >= DATE_START) & (dfw['日期'] <= DATE_END)]



    # 🌟 關鍵修正：從 Drive 路徑改為本地快取路徑

    local_dayk_dir = Path(f"{LOCAL_DAYK_CACHE}/{market_key}")

   

    cache = {}

    for csv in local_dayk_dir.glob("*.csv"): # 從本地快取讀取

        sid = csv.stem.split("_")[0]

        try:

            d = _read_csv_fast(str(csv))

            dc, cc = _pick(d, DATE_COLS), _pick(d, CLOSE_COLS)

            d = d.rename(columns={dc:'日期', cc:'收盤'})

            d['日期'] = pd.to_datetime(d['日期'], errors='coerce').dt.tz_localize(None)

            d = d.dropna(subset=['日期','收盤']).sort_values('日期')

            d = d[(d['日期'] >= DATE_PAD_START) & (d['日期'] <= DATE_END)]

            cache[sid] = d[['日期','收盤']]

        except:

            pass



    rows = []

    for sid, g in dfw.groupby('StockID'):

        day_df = cache.get(sid)

        if day_df is None:

            continue

        for _, r in g.iterrows():

            week = r['ISO_Week']

            iso = day_df['日期'].dt.isocalendar()

            mask = (iso.year.astype(str) + '-' + iso.week.astype(str).str.zfill(2)) == week

            se = day_df.loc[mask, '日期']

            if se.empty:

                continue

            gret = _geom_chain_return(day_df, se.min(), se.max())

            diff = gret - r['Ret_Trad_W'] if np.isfinite(gret) and np.isfinite(r['Ret_Trad_W']) else np.nan

            rows.append({'StockID': sid, 'ISO_Week': week, 'geom_day_chain': gret, 'Ret_Trad_W': r['Ret_Trad_W'], 'diff': diff})



    rep = pd.DataFrame(rows)

    rep.to_csv(qa_dir / "weekly_vs_daily_diff.csv", index=False, encoding='utf-8-sig')

    if not rep.empty:

        bias = rep['diff'].replace([np.inf,-np.inf], np.nan).dropna()

        print(f"📐 週 vs 日 幾何連乘差異：中位 {bias.median():.6f}，|diff|>0.05 的比例 {(bias.abs()>0.05).mean():.2%}")

    print(f"🧪 QA 週K對日K 完成 → {qa_dir/'weekly_vs_daily_diff.csv'}")