【資料清洗案例】台股 3666.TWO「六位數股價」事件與減資錯位全面修正

我在跑台股 yearK 統計時，發現一筆堪稱「驚悚」的異常：

這篇不只要解剖這筆異常，更要講清楚——

其實，這不是孤例。 任何減資、拆股、反向分割的股票，都可能出現同樣錯位。

我們會一起看真實案例、驗證證據， 並展示一整套能自動清洗所有日K 的通用解法。

🧩 一、問題緣起：六位數股價是怎麼來的？

我在 yearK_TW.parquet 加入了這段檢查：

mask_high = (df[["開盤","最高","最低","收盤"]] >= 100000).any(axis=1)
df_high = df.loc[mask_high]

結果全市場只有一檔命中：

📈 3666.TWO 光燿科技

打開它的日K CSV，看起來是這樣的：

+------------+---------+---------+---------+---------+
| Date       |   Open  |   High  |   Low   |  Close  |
+------------+---------+---------+---------+---------+
| 2022-03-04 | 302,500 | 302,500 | 297,500 | 300,000 |
| 2022-04-28 | 205,500 | 210,000 | 205,500 | 209,000 |
| 2022-05-05 | 20.9    | 21.7    | 20.9    | 21.6    |
+------------+---------+---------+---------+---------+

前一天還在 20 萬元，隔天就變成 20 元──縮小了一萬倍。

這不是爬蟲錯誤，而是真實的公司行為。

🧾 二、查證事件：2022-05-05 減資 99.99%

查公開資訊觀測站公告：

光燿科技股份有限公司  
減資比率：99.99%  
減資後股本：10元  
原因：為改善股東權益結構，辦理減資彌補虧損。  
日期：2022年5月5日  

Yahoo Finance 對應記錄：

Stock Splits = 0.0001

代表「1 股 → 0.0001 股」，也就是1:10000 減資。

因此減資前的 300,000 元其實等價於減資後的 30 元。

修正前後比對：

+------------+--------------+--------------+
| 日期       | 修正前收盤  | 修正後收盤  |
+------------+--------------+--------------+
| 2022-03-04 | 300,000      | 30.00        |
| 2022-04-28 | 209,000      | 20.90        |
| 2022-05-05 | 21.65        | 21.65        |
+------------+--------------+--------------+

⚙️ 三、清洗方式：依 Stock Splits 還原名目價

Yahoo 給的原始資料是「事件前後不對齊」的價位。

我們只要依據 Stock Splits 修正即可。

if stock_id == "3666.TWO":
    s = pd.to_numeric(df["Stock Splits"], errors="coerce").fillna(0).replace(0,1.0)
    adj = s[::-1].cumprod()[::-1]
    for col in ["Open","High","Low","Close"]:
        df[col] = df[col] * adj.values

這段會：

• 自動讀出 Yahoo 的 0.0001
• 從該日往前把價位乘上 10,000
• 完美還原成名目價（與玩股網顯示一致）

🧮 四、完整清洗程式：自動備份與回復

我將這邏輯包成工具 fix_3666_split.py，步驟如下：

1️⃣ 自動備份原始日K → /dayK_backup/

2️⃣ 偵測減資比例 → 自動調整價格 3️⃣ 覆寫回原資料夾 4️⃣ 輸出對照檔到 /dayK_fixed/

演算法概念：

pivot = 2022-05-05
估算前後中位數比值
若相差超過 20 倍，自動從 {10,100,1000,10000} 挑最適 scale
只對 pivot 之前的資料除以該 scale

🧠 五、這不是個案：減資與拆股普遍存在

不只光燿科技，還有很多公司也曾經發生過減資或反向拆股。

下面是部分實例：

+-----------+------------+-----------+-----------+
| 股票代號  | 事件日期   | 減資比率  | 修正倍率  |
+-----------+------------+-----------+-----------+
| 3666.TWO  | 2022-05-05 | 99.99%    | ×10,000   |
| 4192.TWO  | 2023-07-10 | 約 73.3%  | ×3.74     |
| 其他      | –          | –         | 無需修正  |
+-----------+------------+-----------+-----------+

Yahoo 的資料行為不一致：

• 有些股票會標 Stock Splits
• 有些不標，但隔夜價格跳變非常明顯
  → 我後來就是靠「隔夜跳變偵測法」找到這些漏網之魚。

🛠️ 六、系統化清洗：全市場批次修正

我寫了一支自動化程式來掃描所有日K檔，流程如下：

1️⃣ 先看 Stock Splits 欄
2️⃣ 若無，檢查隔夜跳變（±80%）
3️⃣ 對事件當天以前的 OHLC 除以因子
4️⃣ 輸出修正版到 /dayK_fixed_by_corpaction/
5️⃣ 生成 corpaction_fix_manifest.csv 記錄所有事件

清單長這樣：

+-----------+-----------+--------------------------+-----------+
| 股票代碼  | 事件數量  | 修正事件                 | 備註      |
+-----------+-----------+--------------------------+-----------+
| 3666.TWO  | 1         | 2022-05-05 ×10000        | 減資彌補虧損 |
| 4192.TWO  | 1         | 2023-07-10 ×3.743        | 減資       |
| 其他      | 0         | –                        | 無事件     |
+-----------+-----------+--------------------------+-----------+

ETF、槓桿與反向商品（006xx、R、L）會自動跳過，不誤判。

🧮 七、為什麼週、月、年K分布圖不受影響？

這是很多人問的關鍵問題：

「既然日K 有錯，那我之前畫的分布圖是不是都錯？」

答案是──沒有錯、也不需要重跑。

原因很簡單：

1️⃣ 我的週/月/年K 程式在遇到極端跳變（例如減資或復牌跳空）時，

 就會自動略過該週或該月的資料。

2️⃣ 換句話說，這些減資或拆股的異常區間，

 本來就沒有被算進分布統計。

3️⃣ 所以週K／月K／年K的漲幅分布圖仍然正確。

 這次修正只讓「日K」回到一致口徑。

✅ 結論：

週月年K分布圖不受影響，因為有巨大跳空的週期早已被跳過。

🧠 八、通用邏輯摘要（可複用）

df["prev_close"] = df["Close"].shift(1)
df["overnight_ratio"] = df["Open"] / df["prev_close"]
mask = np.abs(np.log(df["overnight_ratio"])) >= np.log(1.8)
events = df.loc[mask, ["Date", "overnight_ratio"]]

這段能自動找出「隔夜大跳變」的公司行為事件。

搭配 Stock Splits 欄，就能完整覆蓋所有減資與拆股案例。

🧾 九、結語：從異常發現到系統修正

這次的 3666.TWO 六位數事件，

讓我看見了金融資料裡最經典的陷阱——

價格異常，並不一定是錯誤。

很多時候，那是公司行為留下的痕跡。

我後來才明白，修正資料不是為了「讓數字好看」，

而是要讓統計能更真實地反映市場。 從一檔股票的異常，慢慢延伸到全市場的自動化檢查、回溯調整， 這整段過程，幾乎都是我利用下班時間一點一滴摸出來的。 不是專業量化機構，也沒有昂貴資料庫， 只有一台電腦、公開資料，以及對真實的好奇心。

本書採用 Yahoo Finance 公開資料，

經過多層清洗與減資、拆股回溯調整後製成。 因為這是一般人也能免費取得的資料來源， 所以部分個股的最高價，可能與券商軟體或專業行情服務略有差異。 這些差異主要來自資料端對公司行為（例如合併、分割、除權息）的處理方式不同。

不過這些小落差不會影響統計結果——

不論是年度漲幅、追高報酬，或整體分布趨勢， 都仍然能夠真實反映市場的集體價格變化。

換句話說，這是一份「普通上班族也能自己做的量化研究」。

只要肯花時間學、願意多試幾次， 就能用免費的公開資料看見市場的節奏。 當你看懂那些波動背後的規律時， 那一刻，其實就已經完成了屬於自己的投資回測。

如下是程式碼

# -*- coding: utf-8 -*-

"""

批次修正：多市場日K 減資/拆股回溯調整 → 轉回「未調整名目價」（玩股口徑）

支援市場：TW / CN / HK / US / JP / KR

- 掃描各市場 dayK/*.csv

- 優先使用 Stock Splits；否則自動偵測隔夜巨大跳變 + 持續性驗證

- 事件「當天及以前」OHLC 均除以 factor

- 自動跳過 ETF/槓桿/反向商品

- 最終輸出：dayK_fixed_by_corpaction/*.csv + corpaction_fix_manifest_{MARKET}.csv

- QA 自檢：1/1、週日

- 新增：tqdm 進度條 + 各市場計時摘要

"""



import os, re, glob, time, warnings

import numpy as np

import pandas as pd

from tqdm import tqdm

from concurrent.futures import ThreadPoolExecutor, as_completed



warnings.filterwarnings("ignore", category=FutureWarning)



# ========== 基本設定 ==========

MARKETS = ["tw-share", "cn-share", "hk-share", "us-share", "jp-share", "kr-share"]

BASE_DIR = "/content/drive/MyDrive/各國股票檔案"



THRESH_LOG  = np.log(1.8)

MAX_FACTOR  = 50.0

MIN_FACTOR  = 1.2

MIN_DAYS_BEFORE = 10

WIN = 3

EPS = 0.06

SKIP_ETF = True



# ========== ETF/槓桿判斷 ==========

def looks_like_etf_or_leveraged(code: str) -> bool:

    digits = re.sub(r"\D", "", code)

    if re.match(r"^00\d{3,}$", digits):

        return True

    if re.search(r"(R|L)(\.TW|\.TWO|\.HK|\.KS|\.KQ|\.T|\.US)$", code, flags=re.IGNORECASE):

        return True

    return False



# ========== 讀檔 ==========

def read_csv_robust(path: str):

    for enc in ["pyarrow", "utf-8-sig", "utf-8"]:

        try:

            return pd.read_csv(path, engine=None if enc=="pyarrow" else "python",

                               encoding=None if enc=="pyarrow" else enc)

        except Exception:

            continue

    raise IOError(f"讀取失敗：{path}")



# ========== 日期標準化 ==========

def _as_local_date(series):

    s = series.astype(str).str.slice(0, 10)

    return pd.to_datetime(s, format="%Y-%m-%d", errors="coerce")



def normalize_df(df):

    cols = [str(c).strip().replace(" ", "_") for c in df.columns]

    df.columns = cols

    if "Date" in df.columns: df["Date"] = _as_local_date(df["Date"])

    elif "日期" in df.columns: df["Date"] = _as_local_date(df["日期"])

    else: df["Date"] = _as_local_date(df.iloc[:, 0])



    mapping = {

        "Open":["Open","開盤"], "High":["High","最高"], "Low":["Low","最低"],

        "Close":["Close","收盤"], "Adj_Close":["Adj_Close","Adj Close","調整後收盤價"],

        "Volume":["Volume","成交量"],

    }

    for k, alts in mapping.items():

        for a in alts:

            if a in df.columns:

                df[k] = pd.to_numeric(df[a], errors="coerce")

                break

        if k not in df.columns:

            df[k] = np.nan

    df = df.dropna(subset=["Date"]).sort_values("Date").reset_index(drop=True)

    return df



# ========== Stock Splits 判讀 ==========

def detect_events_via_stock_splits(df):

    events = []

    for c in ["Stock_Splits","Stock Splits","拆股"]:

        if c in df.columns:

            ss = pd.to_numeric(df[c], errors="coerce").fillna(0.0)

            for d,v in zip(df.loc[ss>0,"Date"], ss[ss>0]):

                f = 1.0/float(v)

                if MIN_FACTOR <= f <= MAX_FACTOR:

                    events.append((pd.Timestamp(d), f))

            break

    events.sort(key=lambda x: x[0])

    return events



# ========== 自動偵測（隔夜跳變 + 持續性驗證） ==========

def detect_corp_actions(df):

    ev = detect_events_via_stock_splits(df)

    if ev: return ev

    x = df.copy()

    x["prev_close"] = x["Close"].shift(1)

    cond = (x["Open"]>0)&(x["prev_close"]>0)

    x.loc[cond,"overnight_ratio"]=x.loc[cond,"Open"]/x.loc[cond,"prev_close"]

    x["abs_log_jump"]=np.abs(np.log(x["overnight_ratio"]))

    cands = x.index[(x["abs_log_jump"]>=THRESH_LOG)&x["overnight_ratio"].notna()]

    if len(cands)==0: return []

    events, last_idx = [], -10**9

    for idx in cands:

        if idx<max(MIN_DAYS_BEFORE,WIN+1): continue

        if (idx-last_idx)<=2: continue

        r=float(x.at[idx,"overnight_ratio"])

        factor=r if r>=1 else 1.0/r

        if not (MIN_FACTOR<=factor<=MAX_FACTOR): continue

        pre=x.iloc[idx-WIN-1:idx-1][["Open","High","Low","Close"]].median()

        post=x.iloc[idx+1:idx+1+WIN][["Open","High","Low","Close"]].median()

        if pre.isna().any() or post.isna().any(): continue

        ratios=(post/pre).values

        ok_cols=np.mean(np.abs(np.log(ratios/factor))<=EPS)

        if ok_cols<0.75: continue

        if (x["Volume"].iloc[idx-WIN-1:idx-1].sum()==0) or (x["Volume"].iloc[idx+1:idx+1+WIN].sum()==0):

            continue

        events.append((pd.Timestamp(x.at[idx,"Date"]), factor))

        last_idx=idx

    events.sort(key=lambda y:y[0])

    return events



# ========== 回推名目價 ==========

def apply_unadjust_to_nominal(df, events):

    if not events: return df,[]

    out=df.copy(); applied=[]

    for cutoff,factor in events:

        mask=out["Date"]<=pd.Timestamp(cutoff)

        for col in ["Open","High","Low","Close"]:

            out.loc[mask,col]=out.loc[mask,col]/factor

        applied.append({"cutoff":cutoff,"factor":round(float(factor),6)})

    return out,applied



# ========== 處理單檔 ==========

def process_one(csv_path):

    name=os.path.basename(csv_path)

    code=name.split("_")[0] if "_" in name else name.split(".")[0]

    ret={"file":name,"code":code,"skipped":False,"reason":"","n_events":0,"events":"","out":"","error":None}

    try:

        if SKIP_ETF and looks_like_etf_or_leveraged(code):

            ret.update({"skipped":True,"reason":"ETF/Leveraged"}); return ret

        df_raw=read_csv_robust(csv_path)

        df=normalize_df(df_raw)

        if df["Close"].count()<40:

            ret.update({"skipped":True,"reason":"too_few_rows"}); return ret

        events=detect_corp_actions(df)

        out_path=csv_path.replace("/dayK/","/dayK_fixed_by_corpaction/").replace(".csv","_fixed.csv")

        os.makedirs(os.path.dirname(out_path), exist_ok=True)

        if not events:

            df_raw.to_csv(out_path,index=False,encoding="utf-8"); ret.update({"out":out_path}); return ret

        df_nom,applied=apply_unadjust_to_nominal(df,events)

        df_write=df_raw.copy()

        for en,alts in {"Open":["Open","開盤"],"High":["High","最高"],"Low":["Low","最低"],"Close":["Close","收盤"]}.items():

            if en in df_nom.columns:

                if en in df_write.columns: df_write[en]=df_nom[en].astype(float)

                else:

                    for a in alts:

                        if a in df_write.columns:

                            df_write[a]=df_nom[en].astype(float); break

        df_write.to_csv(out_path,index=False,encoding="utf-8")

        ret.update({"n_events":len(applied),"events":"; ".join([f"{x['cutoff'].date()} x{x['factor']}" for x in applied]),"out":out_path})

        return ret

    except Exception as e:

        ret.update({"error":str(e)}); return ret



# ========== 主流程 ==========

def run_market(market):

    print(f"\n🌏 處理市場：{market}")

    start_time = time.time()

    CSV_IN_DIR=f"{BASE_DIR}/{market}/dayK"

    paths=sorted(glob.glob(f"{CSV_IN_DIR}/*.csv"))

    if not paths:

        print("⚠️ 無日K檔案，略過。"); return

    results=[]

    with ThreadPoolExecutor(max_workers=8) as ex:

        for fut in tqdm(as_completed([ex.submit(process_one,p) for p in paths]), total=len(paths),

                        desc=f"🧩 {market}", ncols=90):

            results.append(fut.result())

    man=pd.DataFrame(results)

    outdir=f"{BASE_DIR}/{market}/dayK_fixed_by_corpaction"

    os.makedirs(outdir, exist_ok=True)

    man_path=f"{outdir}/corpaction_fix_manifest_{market}.csv"

    man.to_csv(man_path,index=False,encoding="utf-8")



    ok=man[(~man["skipped"])&(man["error"].isna())]

    fixed=ok[ok["n_events"]>0]

    duration = time.time() - start_time

    print(f"\n✅ {market} 完成 ({duration:.1f}s)，清單：{man_path}")

    print(f"📦 成功：{len(ok):>5} 檔，有事件：{len(fixed):>4} 檔")



    if not fixed.empty:

        print(fixed[["code","n_events","events"]].head(10).to_string(index=False))



    # QA：檢查日期異常

    bad_0101,bad_sun=[],[]

    for f in glob.glob(f"{outdir}/*_fixed.csv"):

        try:

            d=pd.read_csv(f,usecols=["Date"],dtype=str)

            s=pd.to_datetime(d["Date"],errors="coerce")

            if ((s.dt.month==1)&(s.dt.day==1)).any(): bad_0101.append(os.path.basename(f))

            if (s.dt.dayofweek==6).any() and market!="us-share": bad_sun.append(os.path.basename(f))

        except: continue

    print(f"🔎 QA：含 1/1 → {len(bad_0101)}；含週日 → {len(bad_sun)}")

    print("-------------------------------------------------------------")



# ========== 主執行 ==========

if __name__=="__main__":

    t0 = time.time()

    for m in MARKETS:

        run_market(m)

    print(f"\n🌍 全部市場處理完成，用時：{time.time()-t0:.1f}s")

執行畫面如下，因為台灣已經跑過了，所以就略過台灣的部分

後續我問AI 先跑如上程式碼後續在做一次清洗後轉周月年K的程式碼是否已經足夠 ,Grok的回覆如下

為何其他五國都沒有這問題 AI回復如下