Kronos 安裝腳本與台灣股市預測實作

1. 安裝腳本

#!/bin/bash
# Kronos 安裝腳本（Python 需 3.10+）
set -e

KRONOS_DIR="/home/charles-chang/.openclaw/workspace/research/2026-04-12_kronos_taiwan_stock/kronos_repo"
ENV_DIR="/home/charles-chang/kronos_env"

# 建立虛擬環境
if [ ! -d "$ENV_DIR" ]; then
    python3 -m venv $ENV_DIR
fi
source $ENV_DIR/bin/activate

# 安裝依賴
pip install --upgrade pip -q
pip install torch torchvision --index-url https://download.pytorch.org/whl/cu121 -q
pip install yfinance pandas transformers einops -q

# 克隆 Kronos（如尚未存在）
if [ ! -d "$KRONOS_DIR" ]; then
    git clone https://github.com/shiyu-coder/Kronos.git $KRONOS_DIR
fi

# 驗證模型
python3 <<'PYEOF'
import sys
sys.path.insert(0, '/home/charles-chang/.openclaw/workspace/research/2026-04-12_kronos_taiwan_stock/kronos_repo')
from model import Kronos, KronosTokenizer, KronosPredictor
tokenizer = KronosTokenizer.from_pretrained("NeoQuasar/Kronos-Tokenizer-base")
model = Kronos.from_pretrained("NeoQuasar/Kronos-small")
predictor = KronosPredictor(model, tokenizer, max_context=512)
print("模型載入成功 ✓")
PYEOF

2. 台灣股市預測腳本

#!/usr/bin/env python3
"""
Kronos 台灣股市預測腳本
使用：python3 predict_taiwan_stock.py <股票代號> [--lookback N] [--pred_len N]
範例：python3 predict_taiwan_stock.py 2330 --pred_len 30
"""

import os, sys, argparse
from datetime import datetime, timedelta
import pandas as pd
import yfinance as yf

kronos_path = os.path.join(os.path.dirname(__file__), 'kronos_repo')
if os.path.exists(kronos_path):
    sys.path.insert(0, kronos_path)
from model import Kronos, KronosTokenizer, KronosPredictor


def fetch_taiwan_stock_data(stock_code: str, days: int = 800) -> pd.DataFrame:
    """從 Yahoo Finance 取得台股歷史 K線"""
    ticker = f"{stock_code}.TW"
    df = yf.download(ticker, period=f"{days}d", auto_adjust=False, progress=False)
    df = df.reset_index()
    if isinstance(df.columns, pd.MultiIndex):
        df.columns = [col[0] if isinstance(col, tuple) else col for col in df.columns]
    rename_map = {}
    for col in df.columns:
        cl = str(col).lower()
        if 'date' in cl: rename_map[col] = 'timestamps'
        elif 'open' in cl and 'open' not in rename_map.values(): rename_map[col] = 'open'
        elif 'high' in cl and 'high' not in rename_map.values(): rename_map[col] = 'high'
        elif 'low' in cl and 'low' not in rename_map.values(): rename_map[col] = 'low'
        elif 'close' in cl: rename_map[col] = 'close'
        elif 'volume' in cl and 'volume' not in rename_map.values(): rename_map[col] = 'volume'
    df = df.rename(columns=rename_map)
    df = df.loc[:, ~df.columns.duplicated()]
    df['amount'] = 0.0
    for col in ['open', 'high', 'low', 'close', 'volume', 'amount']:
        df[col] = pd.to_numeric(df[col], errors='coerce')
    df = df.dropna()
    return df[['timestamps', 'open', 'high', 'low', 'close', 'volume', 'amount']]


def predict(df: pd.DataFrame, lookback: int = 400,
             pred_len: int = 30, sample_count: int = 1) -> pd.DataFrame:
    """Kronos-small K線預測"""
    tokenizer = KronosTokenizer.from_pretrained("NeoQuasar/Kronos-Tokenizer-base")
    model = Kronos.from_pretrained("NeoQuasar/Kronos-small")
    predictor = KronosPredictor(model, tokenizer, max_context=512)
    lookback = min(lookback, 512)
    df_in = df.tail(lookback).reset_index(drop=True)
    x_ts = df_in['timestamps'].reset_index(drop=True)
    y_ts = pd.Series(pd.date_range(
        start=df_in['timestamps'].iloc[-1] + timedelta(days=1),
        periods=pred_len, freq='B'))
    return predictor.predict(
        df=df_in[['open', 'high', 'low', 'close', 'volume', 'amount']],
        x_timestamp=x_ts, y_timestamp=y_ts,
        pred_len=pred_len, T=1.0, top_p=0.9, sample_count=sample_count)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('stock_code', help='股票代號（例：2330）')
    ap.add_argument('--lookback', type=int, default=400)
    ap.add_argument('--pred_len', type=int, default=30)
    ap.add_argument('--sample_count', type=int, default=1)
    args = ap.parse_args()
    df = fetch_taiwan_stock_data(args.stock_code, days=int(args.lookback * 2))
    pred = predict(df, args.lookback, args.pred_len, args.sample_count)
    print(f"\n{'日期':<12} {'Open':>10} {'High':>10} {'Low':>10} {'Close':>10}")
    print("-" * 60)
    for idx, row in pred.iterrows():
        ds = idx.strftime('%Y-%m-%d') if hasattr(idx,'strftime') else str(idx)
        f = lambda k: f"{row.get(k,0):>10.2f}" if isinstance(row.get(k,0),(int,float)) else f"{str(row.get(k,'')):>10}"
        print(f"{ds:<12} {f('open')} {f('high')} {f('low')} {f('close')}")
    out = f"prediction_{args.stock_code}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
    pred.to_csv(out, index_label='timestamps')
    print(f"\n已儲存：{out}\n此為模型輸出結果，僅供參考，不構成投資建議。")


if __name__ == "__main__":
    main()

3. 台灣股市資料來源

Yahoo Finance（yfinance）	日K為主，即時性普通，3行代碼即完成
twstock 套件	台灣原生，直接串接證交所與櫃買中心
Fugle API	日內分鐘線最可靠
證交所開放資料 API	官方 REST，即時性高

# Yahoo Finance 最簡範例
import yfinance as yf
df = yf.download("2330.TW", start="2020-01-01")  # 台積電

4. 使用方式

# 安裝環境
bash setup_kronos.sh

# 預測台積電未來30天
/home/charles-chang/kronos_env/bin/python3 predict_taiwan_stock.py 2330

# 指定歷史窗口與預測長度
/home/charles-chang/kronos_env/bin/python3 predict_taiwan_stock.py 2330 --lookback 200 --pred_len 10

5. 台股特殊考量

• 漲跌停 ±10%：Fine-tuning 時不應過濾這些bars，是正常市場現象
• 成交金額（amount）：Yahoo Finance 無此欄位，以0填補即可
• Context Length：Kronos-small 512 tokens（約1.4年日K）；日內分鐘線建議用 Kronos-mini（2048）
• 日內資料：Yahoo Finance 分鐘線在台股不可靠，建議改用 Fugle API

實作完成時間：2026-04-12