進階課程 · 程式碼
四支課堂 Python 範例，對應各課的實作輸出。可直接下載執行（需要 pandas）。
# 來源註記：這是 Codex 弄的。
"""教學用 OHLCV 稽核器；不取代交易所日曆與 vendor-specific 驗證。"""
from __future__ import annotations
import pandas as pd
REQUIRED_COLUMNS = ("open", "high", "low", "close", "volume")
def audit_ohlcv(frame: pd.DataFrame) -> list[str]:
issues: list[str] = []
missing = [column for column in REQUIRED_COLUMNS if column not in frame.columns]
if missing:
return [f"缺少欄位：{', '.join(missing)}"]
if not isinstance(frame.index, pd.DatetimeIndex):
issues.append("index 不是 DatetimeIndex")
return issues
if frame.index.tz is None:
issues.append("timestamp 沒有 timezone")
if frame.index.has_duplicates:
issues.append(f"重複 timestamp：{int(frame.index.duplicated().sum())} 筆")
if not frame.index.is_monotonic_increasing:
issues.append("timestamp 未依序遞增")
numeric = frame.loc[:, REQUIRED_COLUMNS].apply(pd.to_numeric, errors="coerce")
if numeric.isna().any().any():
issues.append(f"OHLCV 含缺值或非數字：{int(numeric.isna().sum().sum())} 格")
if (numeric[["open", "high", "low", "close"]] <= 0).any().any():
issues.append("價格含 0 或負值")
if (numeric["volume"] < 0).any():
issues.append(f"負 volume：{int((numeric['volume'] < 0).sum())} 筆")
row_max = numeric[["open", "low", "close"]].max(axis=1)
row_min = numeric[["open", "high", "close"]].min(axis=1)
bad_high = numeric["high"] < row_max
bad_low = numeric["low"] > row_min
if bad_high.any():
issues.append(f"high 小於 open/low/close：{int(bad_high.sum())} 筆")
if bad_low.any():
issues.append(f"low 大於 open/high/close：{int(bad_low.sum())} 筆")
return issues
def _demo() -> None:
index = pd.DatetimeIndex(
["2026-01-02 09:30", "2026-01-02 09:31", "2026-01-02 09:31"],
tz="America/New_York",
)
deliberately_bad = pd.DataFrame(
{
"open": [100.0, 101.0, 102.0],
"high": [101.0, 100.5, 103.0],
"low": [99.5, 100.8, 101.0],
"close": [100.5, 101.2, 102.5],
"volume": [1_000, -5, 800],
},
index=index,
)
issues = audit_ohlcv(deliberately_bad)
print("Audit issues:")
for issue in issues:
print(f"- {issue}")
assert len(issues) == 4, f"預期 4 項 issue，實際 {len(issues)}"
if __name__ == "__main__":
_demo()
# 來源註記：這是 Codex 弄的。
"""教學用績效指標；輸入須為有 DatetimeIndex 的 equity curve。"""
from __future__ import annotations
import math
import numpy as np
import pandas as pd
def max_drawdown_duration(drawdown: pd.Series) -> int:
"""回傳連續處於水下的最長觀測期數。"""
longest = current = 0
for value in drawdown.fillna(0.0):
if value < 0:
current += 1
longest = max(longest, current)
else:
current = 0
return longest
def performance_metrics(
equity: pd.Series,
periods_per_year: int = 252,
annual_risk_free_rate: float = 0.0,
) -> dict[str, float]:
equity = equity.dropna().astype(float).sort_index()
if len(equity) < 2 or (equity <= 0).any():
raise ValueError("equity 至少需兩筆且全部大於 0")
if not isinstance(equity.index, pd.DatetimeIndex):
raise TypeError("equity.index 必須是 DatetimeIndex")
returns = equity.pct_change().dropna()
elapsed_years = (equity.index[-1] - equity.index[0]).total_seconds() / (
365.2425 * 24 * 60 * 60
)
if elapsed_years <= 0:
raise ValueError("時間範圍必須大於 0")
cagr = (equity.iloc[-1] / equity.iloc[0]) ** (1 / elapsed_years) - 1
period_rf = (1 + annual_risk_free_rate) ** (1 / periods_per_year) - 1
excess = returns - period_rf
period_std = returns.std(ddof=1)
annual_vol = period_std * math.sqrt(periods_per_year)
sharpe = (
excess.mean() / period_std * math.sqrt(periods_per_year)
if period_std > 0
else math.nan
)
downside = np.minimum(excess.to_numpy(), 0.0)
annual_downside = np.sqrt(np.mean(downside**2)) * math.sqrt(periods_per_year)
sortino = (
excess.mean() * periods_per_year / annual_downside
if annual_downside > 0
else math.nan
)
running_peak = equity.cummax()
drawdown = equity / running_peak - 1
max_dd = float(drawdown.min())
calmar = cagr / abs(max_dd) if max_dd < 0 else math.nan
return {
"CAGR": float(cagr),
"Annual volatility": float(annual_vol),
"Sharpe": float(sharpe),
"Sortino": float(sortino),
"Max drawdown": max_dd,
"Max drawdown duration (periods)": float(max_drawdown_duration(drawdown)),
"Calmar": float(calmar),
}
def _demo() -> None:
rng = np.random.default_rng(42)
dates = pd.bdate_range("2022-01-03", periods=756)
returns = rng.normal(0.00035, 0.012, len(dates))
equity = pd.Series(100_000 * np.cumprod(1 + returns), index=dates)
metrics = performance_metrics(equity, annual_risk_free_rate=0.03)
for name, value in metrics.items():
print(f"{name:36s} {value: .4f}")
if __name__ == "__main__":
_demo()
# 來源註記：這是 Codex 弄的。
"""可重現的教學用 SMA baseline；不下載真實市場資料。"""
from __future__ import annotations
import numpy as np
import pandas as pd
def run_backtest(
close: pd.Series,
window: int = 20,
cost_bps: float = 5.0,
) -> pd.DataFrame:
if window < 2:
raise ValueError("window 必須 >= 2")
close = close.dropna().astype(float).sort_index()
sma = close.rolling(window, min_periods=window).mean()
signal = (close > sma).astype(float)
position = signal.shift(1).fillna(0.0) # 今日 close 訊號，下一期才持有
asset_return = close.pct_change().fillna(0.0)
gross_return = position * asset_return
turnover = position.diff().abs().fillna(position.abs())
cost = turnover * (cost_bps / 10_000)
net_return = gross_return - cost
equity = 100_000 * (1 + net_return).cumprod()
return pd.DataFrame(
{
"close": close,
"sma": sma,
"signal": signal,
"position": position,
"asset_return": asset_return,
"turnover": turnover,
"cost": cost,
"net_return": net_return,
"equity": equity,
}
)
def _synthetic_close() -> pd.Series:
rng = np.random.default_rng(7)
dates = pd.bdate_range("2023-01-02", periods=500)
regime_drift = np.where(np.arange(len(dates)) < 250, 0.0005, -0.00005)
returns = regime_drift + rng.normal(0, 0.012, len(dates))
return pd.Series(100 * np.cumprod(1 + returns), index=dates, name="close")
def _demo() -> None:
close = _synthetic_close()
baseline = run_backtest(close, window=20, cost_bps=5)
expensive = run_backtest(close, window=20, cost_bps=50)
print(f"Baseline final equity: {baseline['equity'].iloc[-1]:,.2f}")
print(f"50 bps final equity: {expensive['equity'].iloc[-1]:,.2f}")
assert expensive["equity"].iloc[-1] <= baseline["equity"].iloc[-1]
changed = baseline.index[baseline["signal"].diff().fillna(0).ne(0)]
for timestamp in changed[:10]:
location = baseline.index.get_loc(timestamp)
if location + 1 < len(baseline):
assert baseline["position"].iloc[location + 1] == baseline["signal"].iloc[location]
if __name__ == "__main__":
_demo()
# 來源註記：這是 Codex 弄的。
"""教學用 rolling walk-forward；每個 test fold 只使用過去資料選參數。"""
from __future__ import annotations
import math
import numpy as np
import pandas as pd
def strategy_returns(close: pd.Series, window: int, cost_bps: float = 5.0) -> pd.Series:
signal = (close > close.rolling(window, min_periods=window).mean()).astype(float)
position = signal.shift(1).fillna(0.0)
turnover = position.diff().abs().fillna(position.abs())
return position * close.pct_change().fillna(0.0) - turnover * cost_bps / 10_000
def annualized_sharpe(returns: pd.Series, periods_per_year: int = 252) -> float:
returns = returns.dropna()
std = returns.std(ddof=1)
return returns.mean() / std * math.sqrt(periods_per_year) if std > 0 else -math.inf
def rolling_walk_forward(
close: pd.Series,
windows: tuple[int, ...] = (10, 20, 40, 80),
train_size: int = 252,
test_size: int = 63,
) -> tuple[pd.Series, pd.DataFrame]:
folds: list[dict[str, object]] = []
oos_parts: list[pd.Series] = []
start = train_size
while start + test_size <= len(close):
train = close.iloc[start - train_size : start]
# 在 train 內選參數；完整 rolling warm-up 不可偷用 test。
scores = {window: annualized_sharpe(strategy_returns(train, window)) for window in windows}
best_window = max(scores, key=scores.get)
warmup_start = max(0, start - best_window)
combined = close.iloc[warmup_start : start + test_size]
combined_returns = strategy_returns(combined, best_window)
test_index = close.iloc[start : start + test_size].index
test_returns = combined_returns.reindex(test_index)
oos_parts.append(test_returns)
folds.append(
{
"train_start": train.index[0],
"train_end": train.index[-1],
"test_start": test_index[0],
"test_end": test_index[-1],
"best_window": best_window,
"train_sharpe": scores[best_window],
"test_sharpe": annualized_sharpe(test_returns),
}
)
start += test_size
if not oos_parts:
raise ValueError("資料長度不足以建立一個 fold")
return pd.concat(oos_parts).sort_index(), pd.DataFrame(folds)
def _demo() -> None:
rng = np.random.default_rng(123)
dates = pd.bdate_range("2018-01-02", periods=1_260)
drift = np.select(
[np.arange(len(dates)) < 420, np.arange(len(dates)) < 840],
[0.0005, -0.0001],
default=0.00025,
)
close = pd.Series(100 * np.cumprod(1 + drift + rng.normal(0, 0.012, len(dates))), index=dates)
oos, folds = rolling_walk_forward(close)
print(folds.to_string(index=False))
print(f"Combined OOS Sharpe: {annualized_sharpe(oos):.3f}")
assert oos.index.is_monotonic_increasing and not oos.index.has_duplicates
assert folds["test_start"].gt(folds["train_end"]).all()
if __name__ == "__main__":
_demo()