"""葛蘭碧、道氏、向下缺口：2026-11-10 預先寫定的 18 個設計。 安裝 finlab、numpy、pandas 後執行：python research.py --out results 所有實驗使用 sim 預設成本、隔日收盤成交，不上傳帳號。 數字來源：FinLab 台股還原價與歷史指數；不代表投資建議。 """ import argparse import contextlib import io import itertools import json from pathlib import Path import finlab import numpy as np import pandas as pd from finlab import data from finlab.backtest import sim END = "2026-11-09" START = "2008-01-02" def sharpe(returns): returns = returns.dropna() return float(returns.mean() / returns.std() * np.sqrt(252)) if returns.std() > 0 else 0.0 def stats(curve): returns = curve.pct_change().iloc[1:] years = (curve.index[-1] - curve.index[0]).days / 365.25 return {"cagr": float((curve.iloc[-1] / curve.iloc[0]) ** (1 / years) - 1), "sharpe": sharpe(returns), "mdd": float((curve / curve.cummax() - 1).min())} def latch(buy, sell): """賣訊優先；當日訊號只影響隔日成交。""" state = np.zeros(buy.shape[1], dtype=bool) held = np.empty(buy.shape, dtype=bool) for i, (entry, exit_) in enumerate(zip(buy.to_numpy(), sell.to_numpy())): state = (state | entry) & ~exit_ held[i] = state return pd.DataFrame(held, index=buy.index, columns=buy.columns) def granville(close, high, low, eligible, rule, n, parameter): ma = close.rolling(n).mean() slope_days = parameter if rule < 3 else 5 slope = ma.diff(slope_days) cross_up = (close > ma) & (close.shift(1) <= ma.shift(1)) cross_down = (close < ma) & (close.shift(1) >= ma.shift(1)) if rule == 1: buy = cross_up & (slope >= 0) & (slope.shift(slope_days) <= 0) sell = cross_down & (slope <= 0) & (slope.shift(slope_days) >= 0) elif rule == 2: buy = cross_up & (slope > 0) & (slope.shift(slope_days) > 0) sell = cross_down & (slope < 0) & (slope.shift(slope_days) < 0) elif rule == 3: buy = ((close.shift(1) > ma.shift(1)) & (slope > 0) & (close >= ma) & ((low / ma - 1).abs() <= parameter)) sell = ((close.shift(1) < ma.shift(1)) & (slope < 0) & (close <= ma) & ((high / ma - 1).abs() <= parameter)) else: buy = (slope < 0) & (close / ma <= 1 - parameter) sell = (slope > 0) & (close / ma >= 1 + parameter) return latch(buy & eligible, sell) & close.notna() def confirmed_trend(series, k): """t-k 的轉折到 t 才可得，不把轉折回填。""" candidate = series.shift(k) peaks = candidate.eq(series.rolling(2 * k + 1).max()) troughs = candidate.eq(series.rolling(2 * k + 1).min()) highs, lows, result = [], [], [] for date in series.index: if peaks.loc[date]: highs.append(candidate.loc[date]) if troughs.loc[date]: lows.append(candidate.loc[date]) result.append(len(highs) >= 2 and len(lows) >= 2 and highs[-1] > highs[-2] and lows[-1] > lows[-2]) return pd.Series(result, index=series.index) def gap_events(high, low, eligible, delay): gap = (high < low.shift(1)) & eligible fills = pd.DataFrame(False, index=high.index, columns=high.columns) for lag in range(1, delay + 1): edge = low.shift(lag + 1) not_before = high.shift(1).rolling(lag - 1).max() < edge if lag > 1 else True fills |= gap.shift(lag, fill_value=False) & (high >= edge) & not_before unfilled = gap.shift(delay, fill_value=False) & (high.rolling(delay).max() < low.shift(delay + 1)) return fills & eligible, unfilled def gap_statistics(high, low, eligible): rows = [] gap = (high < low.shift(1)) & eligible for horizon in (5, 20, 60): # 只剔除資料尾端尚未走完的觀察窗；下市或未補事件仍保留。 observed = gap.iloc[:-horizon] future_high = high.shift(-1).iloc[::-1].rolling(horizon, min_periods=1).max().iloc[::-1] filled = (future_high >= low.shift(1)).iloc[:-horizon] & observed for market, columns in [("0050", ["0050"]), ("stocks", [c for c in high if c != "0050"])]: count = int(observed[columns].to_numpy().sum()) successes = int(filled[columns].to_numpy().sum()) rows.append({"market": market, "horizon": horizon, "events": count, "filled": successes, "rate": successes / count if count else None}) return rows def evaluate(position, dates, bench, out, name, center): position = position.reindex(dates).fillna(False) with contextlib.redirect_stdout(io.StringIO()): report = sim(position, resample=None, upload=False, retain_initial_cash=True, end_date=END) curve = report.creturn.reindex(dates).ffill().fillna(1.0) curve = curve / curve.iloc[0] reference = bench.reindex(dates).ffill() reference = reference / reference.iloc[0] result = stats(curve) result["benchmark"] = stats(reference) midpoint = len(dates) // 2 result["halves"] = [stats(curve.iloc[:midpoint]), stats(curve.iloc[midpoint:])] result["benchmark_halves"] = [stats(reference.iloc[:midpoint]), stats(reference.iloc[midpoint:])] exposure = position.any(axis=1).shift(1, fill_value=False).mean() fixed = (1 + reference.pct_change().fillna(0) * exposure).cumprod() result["exposure"] = float(exposure) result["fixed_weight_sharpe"] = stats(fixed)["sharpe"] result["win"] = bool(result["sharpe"] > max(result["benchmark"]["sharpe"], result["fixed_weight_sharpe"])) if center: weights = position.div(position.sum(axis=1).replace(0, np.nan), axis=0).fillna(0) turnover = weights.diff().fillna(weights).abs().sum(axis=1).resample("ME").sum() result["monthly_target_turnover_mean"] = float(turnover.mean()) result["monthly_target_turnover_max"] = float(turnover.max()) returns = curve.pct_change().iloc[1:] annual = returns.groupby(returns.index.year).apply(lambda x: (1 + x).prod() - 1) best_year = int(annual.idxmax()) result["removed_year"] = best_year result["without_best_year"] = sharpe(returns[returns.index.year != best_year]) benchmark_returns = reference.pct_change().iloc[1:] result["benchmark_without_best_year"] = sharpe(benchmark_returns[benchmark_returns.index.year != best_year]) trades = report.get_trades() result["trades"] = len(trades) result["average_holdings"] = float(position.sum(axis=1).mean()) report.to_html(str(out / f"{name}_report.html")) pd.DataFrame({name: curve, "0050": reference}).to_csv(out / f"{name}_data.csv", index_label="date") return result def main(output): finlab.login() data.truncate_end = END output.mkdir(parents=True, exist_ok=True) close = pd.DataFrame(data.get("etl:adj_close")) columns = [c for c in close if len(c) == 4 and c.isdigit() and not c.startswith(("0", "91"))] columns = ["0050"] + columns close = close[columns] high = pd.DataFrame(data.get("etl:adj_high"))[columns] low = pd.DataFrame(data.get("etl:adj_low"))[columns] amount = pd.DataFrame(data.get("price:成交金額"))[columns] eligible = amount.rolling(20).mean().ge(20_000_000) & close.notna() eligible["0050"] = close["0050"].notna() dates = close.loc[START:END].index benchmark = close["0050"].ffill() taiex = data.get("taiex_total_index:收盤指數")["TAIEX"] otc = data.get("stock_index_price:收盤指數")["上櫃櫃買指數:指數"].dropna() dow_dates = dates[dates >= otc.index.min()] results = {"run_date": pd.Timestamp.now(tz="Asia/Taipei").isoformat(), "finlab_version": finlab.__version__, "end": END, "start": START, "dow_start": str(dow_dates[0].date()), "designs": {}} for rule in range(1, 5): parameters = (3, 5, 10) if rule < 3 else ((.005, .01, .02) if rule == 3 else (.08, .12, .16)) for instrument in ("stocks", "0050", "0050_avoid"): name = f"g{rule}_{instrument}" selected = columns[1:] if instrument == "stocks" else ["0050"] frames = close[selected], high[selected], low[selected], eligible[selected] rows = [] for n, parameter in itertools.product((40, 60, 80), parameters): position = granville(*frames, rule, n, parameter) if instrument == "0050_avoid": position = ~position center = n == 60 and parameter == parameters[1] result = evaluate(position, dates, benchmark, output, name, center) rows.append({"parameters": [n, parameter], "center": center, **result}) print(name, n, parameter, round(result["sharpe"], 3), flush=True) results["designs"][name] = rows (output / "metrics.json").write_text(json.dumps(results, ensure_ascii=False, indent=2)) agreed_by_k = {k: confirmed_trend(taiex, k).reindex(dow_dates, fill_value=False) & confirmed_trend(otc, k).reindex(dow_dates, fill_value=False) for k in (3, 5, 10)} for inverse in (False, True): name = "dow_avoid" if inverse else "dow" rows = [] for k, consecutive in itertools.product((3, 5, 10), (1, 3, 5)): position = agreed_by_k[k].rolling(consecutive).sum().eq(consecutive) if inverse: position = ~position center = k == 5 and consecutive == 3 result = evaluate(position.to_frame("0050"), dow_dates, benchmark, output, name, center) rows.append({"parameters": [k, consecutive], "center": center, **result}) print(name, k, consecutive, round(result["sharpe"], 3), flush=True) results["designs"][name] = rows for instrument in ("stocks", "0050", "0050_avoid", "0050_unfilled_avoid"): name = f"gap_{instrument}" selected = columns[1:] if instrument == "stocks" else ["0050"] rows = [] for delay in (3, 5, 10): fill, unfilled = gap_events(high[selected], low[selected], eligible[selected], delay) event = unfilled if instrument == "0050_unfilled_avoid" else fill for hold in (10, 20, 40): position = event.rolling(hold, min_periods=1).max().eq(1) if "avoid" in instrument: position = ~position center = delay == 5 and hold == 20 result = evaluate(position, dates, benchmark, output, name, center) rows.append({"parameters": [delay, hold], "center": center, **result}) print(name, delay, hold, round(result["sharpe"], 3), flush=True) results["designs"][name] = rows results["gap_statistics"] = gap_statistics(high.loc[START:], low.loc[START:], eligible.loc[START:]) for rows in results["designs"].values(): center = next(row for row in rows if row["center"]) center["wins"] = sum(row["win"] for row in rows) center["passed"] = bool(center["wins"] >= 7 and all( a["sharpe"] > b["sharpe"] for a, b in zip(center["halves"], center["benchmark_halves"])) and center["without_best_year"] > center["benchmark_without_best_year"]) (output / "metrics.json").write_text(json.dumps(results, ensure_ascii=False, indent=2)) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--out", type=Path, default=Path("results")) main(parser.parse_args().out)