明天股市會漲幾點？這是個很有趣的問題，那就建個模型來預測吧。我敢保證：一定不準！但也許有點用。
定義目標變數
到底要預測什麼，收盤點數嗎？其實那是不大合理的，因為交易機會是在盤中出現的，該如何表達盤中的機會？最高價吧。做多時是最高價，同理做空時就是最低價，那就定義最高最低價為目標變數吧，意義也很好理解，就是上漲空間或下跌空間，相對於交易策略就是「獲利空間」與「最大可能損失」，更精簡的說就是「獲利」與「風險」，其比值就是進不進場的最重要的判斷指標：賺賠比。另外，因為不想做得太短，所以把時間拉到三天，所以結論就是：未來三天的獲利機會與風險承擔，就是我的目標變數。
定義屬性
屬性工程本身就是一個很重要的課題，必須定義出可能會影響結果的變數。我先用既有的領域知識，定義出幾個變數，至少涵蓋以下幾個類別的屬性：波動性，動能性 (已過數天漲幅)，範圍與落差 (已過數天的高低差)，位置 (均線乖離距離) 等等，我照此訂出 10 個變數。
取得資料與預處理
先取得 3000 個日曆日以內的 k 線資料，大約八年多，順便把目標變數計算好，確定目標的可量測性。現在程式碼已經不值錢了，隨便問 ai 都會，想法比較重要，加減貼貼。
def prepare_data_init(self, days=2000):
df_stock = YahooService.getStockHist('^TWII', days=days, interval='1d')
df_stock.index = pd.to_datetime(df_stock.index, unit='s').tz_localize('UTC').tz_convert('UTC+08:00')
df_stock.index = df_stock.index.normalize()
df_stock = df_stock.groupby(level=0).first() # remove duplicate row of same day
df_stock.reset_index(inplace=True)
# assume the last day has completed, that means after market close,
# then we can calculate the 3-day high and low
for index, row in df_stock.iterrows():
t = df_stock.iloc[index]['time']
t_str = t.strftime('%Y-%m-%d')
t_close = df_stock.iloc[index]['close']
t_open = df_stock.iloc[index]['open']
t_high = df_stock.iloc[index]['high']
t_low = df_stock.iloc[index]['low']
if index + 3 < len(df_stock): # not enough data to calculate 3-day high and low
t_3day_high = max(df_stock[index+1:index+4]['high'])
t_3day_low = min(df_stock[index+1:index+4]['low'])
max_long_return = (t_3day_high - t_close) / t_close
max_short_return = (t_3day_low - t_close ) / t_close
else:
t_3day_high = None
t_3day_low = None
max_long_return = None
max_short_return = None
daily_obj = {
'id': t_str,
'close': t_close,
'open': t_open,
'high': t_high,
'low': t_low,
'3day_high': t_3day_high,
'3day_low': t_3day_low,
'max_long_return': max_long_return,
'max_short_return': max_short_return,
}
self.fire.set_document('dailyModel', t_str, daily_obj)
有了目標變數，下一步是計算屬性欄位，就是用來預測目標變數的憑藉。以下程式從我暫存的 firestore database 讀取資料後，計算 10 個屬性，再回存。一樣是有想法，會問就會，加減貼，身為人類，不用自己寫，但要會欣賞：
def prepare_data_feature(self, n=None):
collection_ref = self.fire.firestore_client.collection('dailyModel')
# 1️⃣ 讀取資料
if n is None:
docs = collection_ref.stream()
else:
docs = (
collection_ref
.order_by("id", direction="DESCENDING")
.limit(n)
.stream()
)
data_list = []
for doc in docs:
d = doc.to_dict()
d["id"] = doc.id # 保留 doc id
data_list.append(d)
# 2️⃣ 轉 DataFrame（方便 rolling）
df = pd.DataFrame(data_list)
# ⚠️ 確保排序（非常重要）
df = df.sort_values("id") # 假設 id 是 YYYY-MM-DD
# 3️⃣ 計算基本欄位
df["return"] = df["close"].pct_change()
# --- Feature 開始 ---
# 4️⃣ 波動
df["vol_5"] = df["return"].rolling(5).std()
df["vol_20"] = df["return"].rolling(20).std()
# volatility ratio
df["vol_ratio"] = df["vol_5"] / df["vol_20"]
# 5️⃣ range（用 high/low）
df["range_3"] = (df["3day_high"].shift(3) - df["3day_low"].shift(3)) / df["close"]
# 👉 注意：這裡用 shift 避免未來資料污染
# intraday / range proxy
df["range_1"] = (df["high"] - df["low"]) / df["close"]
# 6️⃣ 動能
df["ret_3"] = df["close"].pct_change(3)
df["ret_5"] = df["close"].pct_change(5)
# momentum acceleration
df["ret_acc"] = df["ret_3"] - df["ret_5"]
# 7️⃣ 位置（MA20）
df["ma20"] = df["close"].rolling(20).mean()
df["dist_ma20"] = (df["close"] - df["ma20"]) / df["ma20"]
# --- Feature 結束 ---
# 8️⃣ 回寫 Firestore
def safe_float(x):
if pd.isna(x):
return None
return float(x)
for i, row in df[20:].iterrows():
update_data = {
"return": safe_float(row["return"]),
"vol_5": safe_float(row["vol_5"]),
"vol_20": safe_float(row["vol_20"]),
"vol_ratio": safe_float(row["vol_ratio"]),
"range_3": safe_float(row["range_3"]),
"range_1": safe_float(row["range_1"]),
"ret_3": safe_float(row["ret_3"]),
"ret_5": safe_float(row["ret_5"]),
"ret_acc": safe_float(row["ret_acc"]),
"dist_ma20": safe_float(row["dist_ma20"]),
}
doc_ref = collection_ref.document(row["id"])
doc_ref.set(update_data, merge=True)
訓練模型
先讀取資料，包含先前算好的目標變數，和所有屬性欄位。
n = None # 50 # if None, 讀取全部；如果是數字，讀取最近 n 筆
collection_ref = firestore_client.collection('dailyModel')
# 1️⃣ 讀取資料
if n is None:
docs = collection_ref.stream()
else:
docs = (
collection_ref
.order_by("id", direction="DESCENDING")
.limit(n)
.stream()
)
data_list = []
for doc in docs:
d = doc.to_dict()
d["id"] = doc.id # 保留 doc id
data_list.append(d)
# 2️⃣ 轉 DataFrame（方便 rolling）
df = pd.DataFrame(data_list)
# ⚠️ 確保排序（非常重要）
df = df.sort_values("id") # 假設 id 是 YYYY-MM-DD
資料預處理，需要處理成純粹數字的矩陣 X and y，去除空值，模型的世界很純粹：
# 選擇 feature
feature_cols = [
"vol_5",
"vol_20",
"range_3",
"ret_3",
"ret_5",
"dist_ma20",
"range_1",
"vol_ratio",
"ret_acc"
]
# target
target_up = "max_long_return"
target_down = "max_short_return"
# 移除缺值
df = df.dropna(subset=feature_cols + [target_up, target_down])
# ======================
# 3️⃣ 建立 X / y
# ======================
X = df[feature_cols]
y_up = df[target_up]
y_down = df[target_down]
我準備訓練兩個模型，一個預測上漲空間，一個預測下跌空間，所以有兩個 y。兩者都需要切分訓練集和測試集：
split_ratio = 0.8
split_idx = int(len(df) * split_ratio)
X_train, X_test = X.iloc[:split_idx], X.iloc[split_idx:]
y_up_train, y_up_test = y_up.iloc[:split_idx], y_up.iloc[split_idx:]
y_down_train, y_down_test = y_down.iloc[:split_idx], y_down.iloc[split_idx:]
終於可以訓練了，本次採用的模型為 Light GBM (Light Gradient Boosting Machine)，是一個基於決策樹的模型，有空再去欣賞其原理和架構，我們先用再說。
# ======================
# 5️⃣ 訓練模型
# ======================
from lightgbm import LGBMRegressor
model_up = LGBMRegressor(
n_estimators=200,
num_leaves=10,
# min_data_in_leaf=10,
learning_rate=0.02,
# max_depth=3,
random_state=42
)
model_down = LGBMRegressor(
n_estimators=200,
num_leaves=10,
# min_data_in_leaf=10,
learning_rate=0.02,
# max_depth=3,
random_state=42
)
model_up.fit(X_train, y_up_train)
model_down.fit(X_train, y_down_train)
因為資料量不大，瞬間就完成了。
驗證模型
要驗證先要用測試集執行預測，因為我之後還會製作網頁介面，所以把預測結果也存起來：
# ======================
# 6️⃣ 預測
# ======================
pred_up = model_up.predict(X_test)
pred_down = model_down.predict(X_test)
df["pred_up"] = model_up.predict(X)
df["pred_down"] = model_down.predict(X)
print(df[["id", "max_long_return", "max_short_return", "pred_up", "pred_down"]])
# store prediction back to Firestore
collection_ref = firestore_client.collection('dailyModel')
for i, row in df.iterrows():
update_data = {
"pred_up": row["pred_up"],
"pred_down": row["pred_down"],
}
doc_ref = collection_ref.document(row["id"])
doc_ref.set(update_data, merge=True)
評估此模型是否有效，這很有學問，有多種判定方法：
# ======================
# 7️⃣ 評估
# ======================
import numpy as np
from sklearn.metrics import mean_squared_error
rmse_up = np.sqrt(mean_squared_error(y_up_test, pred_up))
rmse_down = np.sqrt(mean_squared_error(y_down_test, pred_down))
print("RMSE Up:", rmse_up)
print("RMSE Down:", rmse_down)
# 輸出以下 RMSE，光看數字無感
RMSE Up: 0.016175661647535516
RMSE Down: 0.020318404309552555
進一步用不同觀點判定之，用預測實際的相關係數來看看：
# ======================
# 8️⃣ RR（核心指標）
# ======================
epsilon = 1e-6
rr_pred = pred_up / (np.abs(pred_down) + epsilon)
rr_true = y_up_test.values / (np.abs(y_down_test.values) + epsilon)
# correlation（很重要）
corr = np.corrcoef(rr_pred, rr_true)[0, 1]
print("RR Correlation:", corr)
print("y_up Correlation:", np.corrcoef(pred_up, y_up_test)[0, 1])
print("y_down Correlation:", np.corrcoef(pred_down, y_down_test)[0, 1])
# 輸出以下
RR Correlation: 0.014879093956133523
y_up Correlation: 0.32546938777499507
y_down Correlation: 0.11649680598566833
果然！哈哈，很難啦，預測指數是個非常艱巨的任務，表現不好是正常，但那個 0.32 吸引了我。這已經是難得的好成績了，其實 y_down 的 0.11 也算有一點點 edge，就看我們怎麼運用。
每日預測新資料
既然模型有一點點 edge，就要用每日的新資料來玩真的啊。這完全是一套新的流程，是讓模型走出實驗室的關鍵。主要重點是，訓練好的模型必須存起來，每日預測新資料時直接載入就可執行預測。而新資料的預測變數 y 不存在，這很正常，不可如訓練過程般被捨棄掉，大概就這樣：
def prepare_and_predict(self, n=50):
# 補滿最新天數的資料
self.prepare_data_init(10)
# 計算 feature（至少需 20 筆資料）
self.prepare_data_feature(n)
# 讀取最新 n 筆資料（必須包含 feature）
collection_ref = self.fire.firestore_client.collection('dailyModel')
docs = (
collection_ref
.order_by("id", direction="DESCENDING")
.limit(n)
.stream()
)
data_list = []
for doc in docs:
d = doc.to_dict()
d["id"] = doc.id # 保留 doc id
data_list.append(d)
df = pd.DataFrame(data_list)
# ⚠️ 確保排序（非常重要）
df = df.sort_values("id") # 假設 id 是 YYYY-MM-DD
# predict
df_new = df.iloc[-5:].copy()
feature_cols = [
"vol_5",
"vol_20",
"range_3",
"ret_3",
"ret_5",
"dist_ma20",
"range_1",
"vol_ratio",
"ret_acc"
]
X_new = df_new[feature_cols]
# load model and predict
from joblib import load
model_up = load("resources/lgbm_model_up.pkl")
model_down = load("resources/lgbm_model_down.pkl")
X_new_pred_up = model_up.predict(X_new)
X_new_pred_down = model_down.predict(X_new)
for i, row in df_new.iterrows():
update_data = {
"pred_up": X_new_pred_up[i],
"pred_down": X_new_pred_down[i],
}
print(f"Updating doc id {row['id']} with data: {update_data}")
doc_ref = collection_ref.document(row["id"])
doc_ref.set(update_data, merge=True)
製作美麗的介面，提供優雅決策空間
我苦心經營的網站總是乏人問津，放上這個會不會引起一些好奇心呢？我深知，大部分的人都比較喜歡低價值和花俏的東西，所以越是乏人問津我越安心。寫東西只是要證明我會而已啦，可能對讀者沒什麼幫助！唉，我真壞。我也因這種心理障礙而超過一個月未發文，現在又發文可能是有點自私。
https://newman-portfolio.azurewebsites.net/daily-model
首先把預測區間，和實際區間，畫出來視覺化對比：
可以看到預測的區間大多是在中間的，模型傾向於偷懶，若總是預測中間，誤差函數值比較小，所以大多數模型都沒屁用。但仔細觀察，有些比較瘦，就是「賺賠比」比較好一點，勉強可用啦。所以我進一步畫出累計漲幅，看起來比較有感。賺賠比大於 2 或小於 0.5，也就是做多做空，都要大於兩倍，才值得出手，用三角形標出訊號，歷史訊號中紅色代表達標。加上簡要的敘事，就是可用於實戰的介面：
以上完成一個回合完整的流程，以後不管模型或屬性怎麼變，或要改變預測標的，流程都是大同小異。這是在我關於「市場結構」的研究路線之外，另闢一條遊戲路徑，增添趣味，願讀者們也得到一些啟發。
Newman 2026/4/16
- 導覽頁：精明管家
- 技術議題有興趣者，也可關注這兒：紐曼的技術筆記-索引
