我在跑台股 yearK 統計時，發現一筆堪稱「驚悚」的異常：
某檔股票的收盤價出現六位數（30 萬元）以上！
這篇不只要解剖這筆異常，更要講清楚——
其實，這不是孤例。 任何減資、拆股、反向分割的股票，都可能出現同樣錯位。
我們會一起看真實案例、驗證證據， 並展示一整套能自動清洗所有日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回復如下
