Python 台股回測系統實作：交易成本、滑價、停損與績效指標

台股回測系統需要哪些模組？

一個最小可用的台股回測系統不需要一開始就很複雜，但應該有清楚的資料流。至少要包含：

• data loader
• signal generator
• position engine
• execution model
• cost model
• risk rules
• performance metrics
• trade log
• audit log

如果你還在建立整體 quant research 流程，可以先看 台股量化交易入門。

準備歷史 OHLCV 資料

回測的第一步是取得乾淨且可重複的歷史資料。最基本欄位包括 date、symbol、open、high、low、close、volume。若要做長期回測，還要確認價格是否為 adjusted price。

import os
import requests
import pandas as pd

API_KEY = os.getenv("TW_MARKET_DATA_API_KEY")
BASE_URL = "https://api.example.com"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

params = {
    "from": "2025-01-01",
    "to": "2025-12-31",
    "interval": "1d",
    "adjusted": "true"
}

response = requests.get(
    f"{BASE_URL}/v1/tw/stocks/2330/ohlcv",
    headers=headers,
    params=params,
    timeout=20
)

response.raise_for_status()

payload = response.json()
df = pd.DataFrame(payload["data"])

df["date"] = pd.to_datetime(df["date"])
df = df.sort_values("date").reset_index(drop=True)

print(df.head())

上方 endpoint 是示意。實際 API host 與路徑請以 TW Market Data docs 為準。

如果你還不熟悉台股 API 資料分層，可以先看 台股 API 完整指南。

如果你要先了解 Python 基礎串接與 DataFrame 整理，可參考 Python 抓台股資料教學。

如果你要先確認 adjusted price 與除權息資料設計，可以看 台股歷史股價 API 設計。

如果你要把均線、RSI、MACD 與 ATR 納入回測訊號流程，可以接著看 台股技術分析 API。

建立策略訊號

策略訊號是把資料轉成可執行規則。這裡用最簡單的均線範例示範資料流程：當 20 日均線大於 60 日均線時，signal 為 1；否則為 0。

df["return"] = df["close"].pct_change()

df["ma20"] = df["close"].rolling(20).mean()
df["ma60"] = df["close"].rolling(60).mean()

df["signal"] = 0
df.loc[df["ma20"] > df["ma60"], "signal"] = 1

print(df[["date", "close", "ma20", "ma60", "signal"]].tail())

這只是教學用資料處理範例，不代表策略有效。正式研究時，signal 應該經過 out-of-sample test、參數穩定性測試與風險檢查。

從 signal 轉成持倉

回測不能只看 signal，還要定義持倉。最簡單的方式是使用前一日 signal 決定下一日持倉，避免使用當天收盤後才知道的訊號去交易同一天。

df["position"] = df["signal"].shift(1).fillna(0)

df["gross_strategy_return"] = df["position"] * df["return"]

print(df[["date", "signal", "position", "return", "gross_strategy_return"]].tail())

`shift(1)` 是回測裡很重要的小細節。若省略這一步，策略可能會使用同一天已經發生的價格變化來決定同一天持倉，造成 look-ahead bias。

加入交易成本

不考慮交易成本的回測通常太樂觀。台股回測至少要考慮手續費、交易稅和其他可能成本。不同券商、商品與交易方式會有不同費率，正式系統應依實際情境設定。

cost_model = {
    "commission_rate": 0.001425,
    "transaction_tax_rate": 0.003,
    "min_commission": 20
}

df["position_change"] = df["position"].diff().abs().fillna(df["position"].abs())
df["turnover"] = df["position_change"]

commission_rate = 0.001425
transaction_tax_rate = 0.003

df["commission_cost"] = df["turnover"] * commission_rate
df["tax_cost"] = (df["position"].diff() < 0).astype(float) * df["turnover"] * transaction_tax_rate

df["total_cost"] = df["commission_cost"] + df["tax_cost"]

df["net_strategy_return"] = df["gross_strategy_return"] - df["total_cost"]

print(df[["date", "position", "turnover", "gross_strategy_return", "total_cost", "net_strategy_return"]].tail())

加入滑價模型

滑價是理論成交價和實際成交價之間的差距。對成交量較低的股票，滑價可能比策略本身更重要。

簡化滑價可以用 basis points 表示：

slippage_bps = 10
slippage_rate = slippage_bps / 10_000

df["slippage_cost"] = df["turnover"] * slippage_rate
df["net_strategy_return"] = df["gross_strategy_return"] - df["total_cost"] - df["slippage_cost"]

更完整的滑價模型會考慮：

• 成交量
• 買賣價差
• 單日參與率
• 股票流動性
• 市場波動
• 訂單大小

min_volume = 1_000_000

df["is_liquid"] = df["volume"] >= min_volume
df.loc[~df["is_liquid"], "position"] = 0

加入停損與風險限制

停損不是唯一的風控方法，但它是很多回測系統會先實作的基本風險規則。以下示範用 rolling drawdown 做簡化停損。

df["equity_before_stop"] = (1 + df["net_strategy_return"].fillna(0)).cumprod()
df["running_max"] = df["equity_before_stop"].cummax()
df["drawdown"] = df["equity_before_stop"] / df["running_max"] - 1

stop_loss_threshold = -0.1

df["risk_off"] = df["drawdown"] <= stop_loss_threshold
df.loc[df["risk_off"], "position"] = 0

這個範例只示範概念。實務上要注意停損觸發後如何恢復交易、是否逐檔停損、是否 portfolio-level 停損，以及停損是否能用實際交易價格執行。

計算資金曲線

資金曲線可以讓你觀察策略在時間上的表現。通常會分成未扣成本和扣除成本後的版本。

df["equity_curve"] = (1 + df["net_strategy_return"].fillna(0)).cumprod()

print(df[["date", "net_strategy_return", "equity_curve"]].tail())

計算績效指標

單看總報酬不夠。回測至少應該輸出多個績效指標，包含報酬、風險、回撤與交易頻率。

import numpy as np

def calculate_metrics(returns: pd.Series, periods_per_year: int = 252) -> dict:
    returns = returns.dropna()

    if returns.empty:
        return {
            "total_return": 0,
            "annualized_return": 0,
            "annualized_volatility": 0,
            "sharpe_ratio": None,
            "max_drawdown": 0
        }

    equity = (1 + returns).cumprod()

    total_return = equity.iloc[-1] - 1
    annualized_return = equity.iloc[-1] ** (periods_per_year / len(returns)) - 1
    annualized_volatility = returns.std() * np.sqrt(periods_per_year)

    sharpe_ratio = None
    if annualized_volatility != 0:
        sharpe_ratio = annualized_return / annualized_volatility

    running_max = equity.cummax()
    drawdown = equity / running_max - 1
    max_drawdown = drawdown.min()

    return {
        "total_return": total_return,
        "annualized_return": annualized_return,
        "annualized_volatility": annualized_volatility,
        "sharpe_ratio": sharpe_ratio,
        "max_drawdown": max_drawdown
    }

metrics = calculate_metrics(df["net_strategy_return"])
print(metrics)

儲存交易紀錄與 audit log

Production workflow 不能只保留最終績效。你需要知道每一天為什麼產生 signal、為什麼交易、交易成本是多少，以及是否觸發風險限制。

trades = df.loc[df["position_change"] > 0, [
    "date",
    "symbol",
    "close",
    "position_change",
    "commission_cost",
    "tax_cost",
    "slippage_cost"
]].copy()

trades["reason"] = "position_changed"

print(trades.tail())

完整 Python 範例

以下是簡化版完整範例，包含資料抓取、signal、持倉、交易成本、滑價、資金曲線和績效指標。

import os
import numpy as np
import pandas as pd
import requests

API_KEY = os.getenv("TW_MARKET_DATA_API_KEY")
BASE_URL = "https://api.example.com"

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

def fetch_ohlcv(symbol: str, start: str, end: str) -> pd.DataFrame:
    response = requests.get(
        f"{BASE_URL}/v1/tw/stocks/{symbol}/ohlcv",
        headers=headers,
        params={
            "from": start,
            "to": end,
            "interval": "1d",
            "adjusted": "true"
        },
        timeout=20
    )
    response.raise_for_status()

    payload = response.json()
    frame = pd.DataFrame(payload["data"])

    frame["date"] = pd.to_datetime(frame["date"])
    frame = frame.sort_values("date").reset_index(drop=True)
    return frame

def calculate_metrics(returns: pd.Series, periods_per_year: int = 252) -> dict:
    returns = returns.dropna()

    if returns.empty:
        return {}

    equity = (1 + returns).cumprod()
    running_max = equity.cummax()
    drawdown = equity / running_max - 1

    total_return = equity.iloc[-1] - 1
    annualized_return = equity.iloc[-1] ** (periods_per_year / len(returns)) - 1
    annualized_volatility = returns.std() * np.sqrt(periods_per_year)

    sharpe_ratio = None
    if annualized_volatility != 0:
        sharpe_ratio = annualized_return / annualized_volatility

    return {
        "total_return": total_return,
        "annualized_return": annualized_return,
        "annualized_volatility": annualized_volatility,
        "sharpe_ratio": sharpe_ratio,
        "max_drawdown": drawdown.min(),
        "average_turnover": None
    }

df = fetch_ohlcv("2330", "2025-01-01", "2025-12-31")

df["return"] = df["close"].pct_change()
df["ma20"] = df["close"].rolling(20).mean()
df["ma60"] = df["close"].rolling(60).mean()

df["signal"] = (df["ma20"] > df["ma60"]).astype(int)
df["position"] = df["signal"].shift(1).fillna(0)

df["gross_strategy_return"] = df["position"] * df["return"]

df["position_change"] = df["position"].diff().abs().fillna(df["position"].abs())
df["turnover"] = df["position_change"]

commission_rate = 0.001425
transaction_tax_rate = 0.003
slippage_bps = 10
slippage_rate = slippage_bps / 10_000

df["commission_cost"] = df["turnover"] * commission_rate
df["tax_cost"] = (df["position"].diff() < 0).astype(float) * df["turnover"] * transaction_tax_rate
df["slippage_cost"] = df["turnover"] * slippage_rate

df["total_cost"] = df["commission_cost"] + df["tax_cost"] + df["slippage_cost"]
df["net_strategy_return"] = df["gross_strategy_return"] - df["total_cost"]

df["equity_curve"] = (1 + df["net_strategy_return"].fillna(0)).cumprod()

metrics = calculate_metrics(df["net_strategy_return"])
metrics["average_turnover"] = df["turnover"].mean()

print(df[["date", "close", "position", "net_strategy_return", "equity_curve"]].tail())
print(metrics)

上方 endpoint 是示意。實際 API host 與路徑請以 TW Market Data docs 為準。這是教學用回測框架，不構成投資建議，也不是完整交易系統。

常見回測錯誤

台股回測常見錯誤包括：

AI agent 如何讀取回測結果？

如果你要把回測系統接到 AI agent，不應該只把完整 CSV 丟給 LLM。比較好的方式是先輸出結構化 summary，讓 agent 讀取績效、風險與資料限制。

{
  "strategy_name": "ma_cross_example",
  "symbol": "2330",
  "period": {
    "from": "2025-01-01",
    "to": "2025-12-31"
  },
  "data_used": [
    "adjusted_ohlcv"
  ],
  "metrics": {
    "total_return": 0.12,
    "annualized_return": 0.11,
    "max_drawdown": -0.08,
    "sharpe_ratio": 0.9,
    "average_turnover": 0.04
  },
  "risk_flags": [
    "single_stock_backtest",
    "requires_out_of_sample_test",
    "simplified_transaction_cost_model"
  ],
  "not_investment_advice": true
}

這樣 AI agent 可以負責摘要、比較與檢查風險，但不會取代資料驗證、正式回測與人為審核。

建議的 API endpoint 設計

以下是支援台股回測 workflow 的 endpoint 設計示意。

GET /v1/tw/stocks/{symbol}/ohlcv?from=2025-01-01&to=2025-12-31&adjusted=true
GET /v1/tw/stocks/{symbol}/corporate-actions
GET /v1/tw/calendar/trading-days
GET /v1/tw/stocks/search
GET /v1/tw/etfs/{symbol}/constituents
GET /v1/tw/stocks/{symbol}/financials
GET /v1/tw/stocks/{symbol}/institutional-flows

實際 endpoint 命名不一定要完全相同。重點是資料 schema 穩定、日期清楚、價格調整狀態明確，並能支援 Python、database、backtester 與 AI agent workflow。

上方 endpoint 是示意。實際 API host 與路徑請以 TW Market Data docs 為準。