"""Causal chart-pattern research. Requires finlab>=2.3.8, numpy, scipy, numba. Defaults to this article's pattern and market; see --family and --market. This is a research example, not a validated trading recommendation. Signals at close are executed next session. """ from argparse import ArgumentParser import numpy as np import pandas as pd from scipy.signal import lfilter from numba import njit from finlab import data, FinlabDataFrame from finlab.backtest import sim START, END = "2017-01-03", "2026-11-10" FAMILIES = ["hs", "wm", "fib", "cup"] def load_prices(market): us = market == "us" data.set_market("us" if us else "tw") keys = (["us_price:adj_close", "us_price:high", "us_price:low", "us_price:close", "us_price:volume"] if us else ["etl:adj_close", "etl:adj_high", "etl:adj_low", "price:收盤價", "price:成交股數"]) close, high, low, raw, volume = [pd.DataFrame(data.get(k)).loc[:END] for k in keys] columns = close.columns.intersection(raw.columns).intersection(volume.columns) if market == "0050": columns = pd.Index(["0050"]) elif not us: columns = columns[columns.str.fullmatch(r"[1-9][0-9]{3}")] close = close[columns] raw, volume, high, low = [x.reindex(index=close.index, columns=columns) for x in (raw, volume, high, low)] if us: factor = close / raw high, low = high * factor, low * factor liquidity = (raw * volume).rolling(60, min_periods=60).mean() clean = close.pct_change(fill_method=None).abs().ge(.5).rolling(252, min_periods=1).max().shift(1).fillna(0).eq(0) eligible = (raw.ge(10) & liquidity.ge(10_000_000 if us else 20_000_000) & close.notna().rolling(252).sum().ge(252) & clean) eligible &= liquidity.rank(axis=1, ascending=False, method="first").le(500 if us else 300) if market == "0050": eligible = close.ffill().notna() # Dropping columns never eligible saves memory; daily eligibility still gates every signal. columns = columns[eligible.any(axis=0)] return [x[columns] for x in (close, high, low, liquidity, eligible)] def smooth(close, bandwidth): weights = np.exp(-.5 * (np.arange(24) / bandwidth) ** 2) weights /= weights.sum() values = close.ffill().to_numpy(dtype=float) valid = np.isfinite(values) numerator = lfilter(weights, [1], np.nan_to_num(values), axis=0) denominator = lfilter(weights, [1], valid.astype(float), axis=0) return np.where(denominator > .999, numerator / np.maximum(denominator, 1e-12), np.nan) @njit def states(close, high, low, smoothed, family, tolerance, window, ratio, cup_depth, handle_depth): n, m = close.shape held = np.zeros((n, m), dtype=np.bool_) entries = np.zeros((n, m), dtype=np.bool_) for col in range(m): points = np.zeros(5) dates = np.zeros(5, dtype=np.int64) kinds = np.zeros(5, dtype=np.int64) count, used, born = 0, -1, -1 active, stop, target = False, 0., np.inf for t in range(30, n): p = t - 3 v = smoothed[p, col] kind = 0 if np.isfinite(v): peak, trough = True, True for k in range(p - 3, p + 4): if k != p: peak = peak and v > smoothed[k, col] trough = trough and v < smoothed[k, col] kind = 1 if peak else (-1 if trough else 0) if kind: if count and kinds[count - 1] == kind: if kind * v > kind * points[count - 1]: points[count - 1], dates[count - 1] = v, p else: if count == 5: points[:4] = points[1:5].copy() dates[:4] = dates[1:5].copy() kinds[:4] = kinds[1:5].copy() count = 4 points[count], dates[count], kinds[count] = v, p, kind count += 1 price, previous = close[t, col], close[t - 1, col] buy, sell = False, False new_stop, new_target = 0., np.inf if family <= 1: needed = 5 if family == 0 else 3 if count >= needed: a = count - needed span = dates[count - 1] - dates[a] if 22 <= span <= 120 and t - dates[count - 1] <= 20: direction = -kinds[a] first, last = points[a], points[count - 1] valid_shape = abs(first - last) / ((first + last) / 2) <= tolerance if family == 0: head = points[a + 2] valid_shape = (valid_shape and direction * (head - first) < -.03 * first and direction * (head - last) < -.03 * last and abs(points[a + 1] - points[a + 3]) / ((points[a + 1] + points[a + 3]) / 2) <= tolerance) slope = (points[a + 3] - points[a + 1]) / (dates[a + 3] - dates[a + 1]) neck = points[a + 3] + slope * (t - dates[a + 3]) neck_previous = neck - slope else: neck = points[a + 1] neck_previous = neck valid_shape = valid_shape and direction * (neck - first) >= .03 * first crossed = direction * (price - neck) > 0 and direction * (previous - neck_previous) <= 0 if valid_shape and crossed: buy = direction == 1 and used != dates[count - 1] sell = direction == -1 elif family == 2 and count >= 2: a = count - 2 left, right = dates[a], dates[a + 1] bottom, top = low[left, col], high[right, col] if kinds[a] == -1 and 20 <= right - left <= window and top >= bottom * 1.1: support = top - ratio * (top - bottom) next_ratio = .5 if ratio < .4 else (.618 if ratio < .6 else .786) buy = (used != right and t - right <= 60 and low[t, col] <= support * 1.01 and low[t, col] >= support * .99 and price >= support) new_stop, new_target = top - next_ratio * (top - bottom), top elif family == 3 and count >= 4: a = count - 4 left, bottom_day, right, handle_day = dates[a:count] rim, bottom, right_rim, handle = points[a:count] duration = right - left depth = 1 - bottom / rim valid_shape = (kinds[a] == 1 and 35 <= duration <= 325 and .12 <= depth <= cup_depth and abs(right_rim / rim - 1) <= .05 and .2 <= (bottom_day - left) / duration <= .8 and 5 <= handle_day - right <= 25 and 0 < 1 - handle / right_rim <= handle_depth and handle > (rim + bottom) / 2 and left >= 63 and rim >= close[left - 63, col] * 1.3) buy = (valid_shape and used != handle_day and t - handle_day <= 20 and price > right_rim and previous <= right_rim) new_stop = max(handle, price * .92) if active and (sell or t - born >= 60 or price < stop or price >= target): active = False elif not active and buy: active, born, used = True, t, dates[count - 1] stop, target = new_stop, new_target entries[t, col] = True held[t, col] = active return held, entries def make_position(prices, family="hs", bandwidth=3, tolerance=.03, window=90, ratio=.5, cup_depth=.33, handle_depth=.12, inverse=False): close, high, low, liquidity, eligible = prices held, entries = states(close.to_numpy(), high.to_numpy(), low.to_numpy(), smooth(close, bandwidth), FAMILIES.index(family), tolerance, window, ratio, cup_depth, handle_depth) signal = pd.DataFrame(held, index=close.index, columns=close.columns) if inverse: signal = ~signal candidates = (signal & eligible).loc[START:END] selected = (candidates if len(close.columns) == 1 else liquidity.loc[START:END].where(candidates).rank(axis=1, ascending=False, method="first").le(10)) # sim caps the normalized weights at 10%; fewer than ten names leave cash. return FinlabDataFrame(selected), pd.DataFrame(entries, index=close.index, columns=close.columns) def main(): parser = ArgumentParser(description=__doc__) parser.add_argument("--family", choices=FAMILIES, default="cup") parser.add_argument("--market", choices=["tw", "0050", "us"], default="us") parser.add_argument("--inverse", action="store_true") parser.add_argument("--ratio", type=float, default=.5) args = parser.parse_args() position, _ = make_position(load_prices(args.market), family=args.family, ratio=args.ratio, inverse=args.inverse) report = sim(position, position_limit=1 if args.market == "0050" else .1, upload=False, end_date=END) report.to_html("report.html") print(report.get_stats()) if __name__ == "__main__": main()