"""2026-11-11 事前固定的 K 線、均線、寶塔線研究，12 設計 × 9 格。 安裝 finlab>=2.3.8、numpy、pandas 後：python research.py --out results 收盤形成訊號，隔日收盤成交；預設成本，沒有上傳或部署策略。 """ import argparse import contextlib import io import itertools import json from collections import deque from pathlib import Path import finlab import numpy as np import pandas as pd from finlab import data from finlab.backtest import sim START, END = "2008-01-02", "2026-11-10" def sharpe(r): std = r.std() return float(r.mean() / std * np.sqrt(252)) if std > 0 else 0.0 def stats(c): return {"cagr": float((c.iloc[-1] / c.iloc[0]) ** (365.25 / (c.index[-1] - c.index[0]).days) - 1), "sharpe": sharpe(c.pct_change().iloc[1:]), "mdd": float((c / c.cummax() - 1).min())} def latch(buy, sell): state = np.zeros(buy.shape[1], dtype=bool) rows = [] for entry, exit_ in zip(buy.to_numpy(), sell.to_numpy()): state = (state | entry) & ~exit_ rows.append(state.copy()) return pd.DataFrame(rows, index=buy.index, columns=buy.columns) def candle(o, h, l, c, eligible, kind, threshold, n): body = (c - o).abs() falling = c.diff().lt(0).rolling(n).sum().eq(n) if kind == "doji": event = body.le((h - l) * threshold) & h.gt(l) & falling.shift(1, fill_value=False) elif kind == "hammer": bottom = pd.DataFrame(np.minimum(o, c), index=c.index, columns=c.columns) top = pd.DataFrame(np.maximum(o, c), index=c.index, columns=c.columns) event = ((bottom - l).ge(body * threshold) & (h - top).le(body) & body.gt(0) & falling.shift(1, fill_value=False)) elif kind == "engulf": event = (c.gt(o) & c.shift(1).lt(o.shift(1)) & o.le(c.shift(1)) & c.ge(o.shift(1)) & body.ge(body.shift(1) * threshold) & falling.shift(1, fill_value=False)) else: event = (c.shift(2).lt(o.shift(2)) & body.shift(1).le(body.shift(2) * threshold) & c.gt(o) & c.gt((o.shift(2) + c.shift(2)) / 2) & falling.shift(3, fill_value=False)) return (event & eligible).rolling(5, min_periods=1).max().eq(1) & c.notna() def compression(c, eligible, x, n): averages = [c.rolling(k).mean() for k in (5, 10, 20, 60)] upper = pd.DataFrame(np.maximum.reduce(averages), index=c.index, columns=c.columns) lower = pd.DataFrame(np.minimum.reduce(averages), index=c.index, columns=c.columns) tight = (upper / lower - 1).le(x).rolling(n).sum().eq(n).shift(1, fill_value=False) buy = tight & c.gt(upper) & c.shift(1).le(upper.shift(1)) & eligible return latch(buy, c.lt(averages[2])) & c.notna() def deduction(c, eligible, n, k): # t+1 到 t+k 移出的是 t-n+1 到 t-n+k，全部在 t 已知。 outgoing = c.shift(n - k).rolling(k).mean() return c.gt(outgoing) & eligible def tower(c, eligible, n, buffer): """只對已形成的線段計數；同向延續，跨 N 根界限才反轉。""" result = np.zeros(c.shape, dtype=bool) for j, column in enumerate(c): bars, previous, direction = deque(maxlen=n), None, 0 for i, price in enumerate(c[column].to_numpy()): if not np.isfinite(price): continue if previous is None: previous = price elif not bars and price != previous: bars.append((min(price, previous), max(price, previous))) direction = 1 if price > previous else -1 previous = price elif bars: up = price > max(x[1] for x in bars) * (1 + buffer) down = price < min(x[0] for x in bars) * (1 - buffer) continuation = (direction > 0 and price > previous) or (direction < 0 and price < previous) if continuation or up or down: direction = 1 if price > previous else -1 bars.append((min(price, previous), max(price, previous))) previous = price result[i, j] = direction > 0 # 流動性只限制新進，黑色才出場。 red = pd.DataFrame(result, index=c.index, columns=c.columns) return latch(red & eligible, ~red) & c.notna() def evaluate(p, dates, bench, out, name, a, b, center): p = p.reindex(dates).fillna(False) key = f"{name}_{a}_{b}" with contextlib.redirect_stdout(io.StringIO()): report = sim(p, resample=None, upload=False, retain_initial_cash=True, end_date=END) report.to_html(str(out / f"{key}.html")) curve = report.creturn.reindex(dates).ffill().fillna(1.0) curve /= curve.iloc[0] reference = bench.reindex(dates).ffill() reference /= reference.iloc[0] exposure = float(p.any(axis=1).shift(1, fill_value=False).mean()) fixed = (1 + reference.pct_change().fillna(0) * exposure).cumprod() midpoint = len(dates) // 2 halves = (slice(None, midpoint), slice(midpoint, None)) row = {"parameters": [a, b], "center": center, **stats(curve), "benchmark": stats(reference), "fixed_weight_sharpe": sharpe(fixed.pct_change().iloc[1:]), "exposure": exposure, "halves": [stats(curve.iloc[s]) for s in halves], "benchmark_halves": [stats(reference.iloc[s]) for s in halves]} row["win"] = row["sharpe"] > max(row["benchmark"]["sharpe"], row["fixed_weight_sharpe"]) if center: returns = curve.pct_change().iloc[1:] best = int(returns.groupby(returns.index.year).apply(lambda x: (1 + x).prod() - 1).idxmax()) br = reference.pct_change().iloc[1:] weights = p.div(p.sum(axis=1).replace(0, np.nan), axis=0).fillna(0) turnover = weights.diff().fillna(weights).abs().sum(axis=1).resample("ME").sum() row.update(removed_year=best, without_best_year=sharpe(returns[returns.index.year != best]), benchmark_without_best_year=sharpe(br[br.index.year != best]), monthly_target_turnover_mean=float(turnover.mean()), monthly_target_turnover_max=float(turnover.max()), trades=len(report.get_trades()), average_holdings=float(p.sum(axis=1).mean())) (out / f"{name}_report.html").write_bytes((out / f"{key}.html").read_bytes()) pd.DataFrame({name: curve, "0050": reference}).to_csv(out / f"{name}_data.csv", index_label="date") print(key, round(row["sharpe"], 3), flush=True) return row def main(out): finlab.login() data.truncate_end = END out.mkdir(parents=True, exist_ok=True) c = pd.DataFrame(data.get("etl:adj_close")).loc["2006":END] stocks = [x for x in c if len(x) == 4 and x.isdigit() and not x.startswith(("0", "91"))] columns = ["0050"] + stocks c = c[columns] o, h, l = [pd.DataFrame(data.get(f"etl:adj_{key}"))[columns].reindex(c.index) for key in ("open", "high", "low")] amount = pd.DataFrame(data.get("price:成交金額"))[columns].reindex(c.index) eligible = amount.rolling(20).mean().ge(20_000_000) & c.notna() eligible["0050"] = c["0050"].notna() dates, bench = c.loc[START:END].index, c["0050"].ffill() experiments = [] for kind, values in [("doji", (.05, .10, .15)), ("hammer", (1.5, 2, 2.5)), ("engulf", (1, 1.2, 1.4)), ("morning", (.2, .3, .4))]: experiments.append((kind, values, (2, 3, 4), stocks, kind)) experiments.append(("compression", (.01, .02, .03), (3, 5, 10), stocks, "compression")) for kind, ns in [("month", (15, 20, 25)), ("quarter", (50, 60, 70))]: for instrument in ("stocks", "0050", "0050_avoid"): experiments.append((f"{kind}_{instrument}", ns, (1, 3, 5), stocks if instrument == "stocks" else ["0050"], "deduction")) experiments.append(("tower", (2, 3, 4), (0, .005, .01), stocks, "tower")) results = {"run_date": pd.Timestamp.now(tz="Asia/Taipei").isoformat(), "finlab_version": finlab.__version__, "start": START, "end": END, "designs": {}} for name, aa, bb, selected, kind in experiments: cc, ee = c[selected], eligible[selected] rows = [] for a, b in itertools.product(aa, bb): if kind in ("doji", "hammer", "engulf", "morning"): p = candle(o[selected], h[selected], l[selected], cc, ee, kind, a, b) elif kind == "compression": p = compression(cc, ee, a, b) elif kind == "deduction": p = deduction(cc, ee, a, b) if name.endswith("avoid"): p = ~p & cc.notna() else: p = tower(cc, ee, a, b) rows.append(evaluate(p, dates, bench, out, name, a, b, a == aa[1] and b == bb[1])) center = next(r for r in rows if r["center"]) center["wins"] = sum(r["win"] for r in rows) center["passed"] = bool(center["wins"] >= 7 and center["win"] and all( x["sharpe"] > y["sharpe"] for x, y in zip(center["halves"], center["benchmark_halves"])) and center["without_best_year"] > center["benchmark_without_best_year"]) results["designs"][name] = rows (out / "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)