美股頭肩底排除選股

"""US liquid stocks excluding active inverse-head-and-shoulders positions.

Historical validation: 2017-01-03 to 2026-11-10. Maximum historical drawdown
was about 50%. No leverage. Install FinLab using https://finlab.finance/en/setup.
DATA_END=None keeps the daily strategy current; set it to '2026-11-10' to
reproduce the article. The central rule is bandwidth=3, tolerance=0.03.
"""
import numpy as np
import pandas as pd
from scipy.signal import lfilter
from finlab import data, FinlabDataFrame
from finlab.backtest import sim

START = "2017-01-03"
DATA_END = None


def build_position(end=DATA_END, bandwidth=3, tolerance=.03):
    data.set_market("us")
    close, raw, volume = [pd.DataFrame(data.get(k)).loc[:end] for k in
                          ("us_price:adj_close", "us_price:close", "us_price:volume")]
    raw, volume = [x.reindex(index=close.index, columns=close.columns) for x in (raw, volume)]
    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)
                & close.notna().rolling(252).sum().ge(252) & clean
                & liquidity.rank(axis=1, ascending=False, method="first").le(500))
    columns = close.columns[eligible.any(axis=0)]
    close, liquidity, eligible = [x[columns] for x in (close, liquidity, eligible)]
    values = close.ffill().to_numpy(dtype=float)
    valid = np.isfinite(values)
    weights = np.exp(-.5 * (np.arange(24) / bandwidth) ** 2)
    weights /= weights.sum()
    numerator = lfilter(weights, [1], np.nan_to_num(values), axis=0)
    denominator = lfilter(weights, [1], valid.astype(float), axis=0)
    smoothed = np.where(denominator > .999, numerator / np.maximum(denominator, 1e-12), np.nan)
    n, m = values.shape
    prices = close.to_numpy(dtype=float)
    points = np.zeros((5, m))
    dates = np.zeros((5, m), dtype=int)
    kinds = np.zeros((5, m), dtype=int)
    count = np.zeros(m, dtype=int)
    used = np.full(m, -1)
    born = np.full(m, -1)
    active = np.zeros(m, dtype=bool)
    held = np.zeros((n, m), dtype=bool)
    for t in range(30, n):
        p = t - 3
        v = smoothed[p]
        peak, trough = np.isfinite(v), np.isfinite(v)
        for k in range(p - 3, p + 4):
            if k != p:
                peak &= v > smoothed[k]
                trough &= v < smoothed[k]
        kind = peak.astype(int) - trough.astype(int)
        turning = kind != 0
        same = turning & (count > 0) & (kinds[4] == kind)
        replace = same & (kind * v > kind * points[4])
        append = turning & ~same
        for arr in (points, dates, kinds):
            arr[:4, append] = arr[1:5, append].copy()
        update = append | replace
        points[4, update], dates[4, update], kinds[4, update] = v[update], p, kind[update]
        count[append] = np.minimum(count[append] + 1, 5)
        direction = -kinds[0]
        span = dates[4] - dates[0]
        shape = ((count >= 5) & (span >= 22) & (span <= 120) & (t - dates[4] <= 20)
                 & (np.abs(points[0] - points[4]) <= tolerance * (points[0] + points[4]) / 2)
                 & (direction * (points[2] - points[0]) < -.03 * points[0])
                 & (direction * (points[2] - points[4]) < -.03 * points[4])
                 & (np.abs(points[1] - points[3]) <= tolerance * (points[1] + points[3]) / 2))
        distance = dates[3] - dates[1]
        slope = np.divide(points[3] - points[1], distance, out=np.zeros(m), where=distance != 0)
        neck = points[3] + slope * (t - dates[3])
        crossed = ((direction * (prices[t] - neck) > 0)
                   & (direction * (prices[t-1] - (neck-slope)) <= 0))
        buy = shape & crossed & (direction == 1) & (used != dates[4])
        sell = shape & crossed & (direction == -1)
        exits = active & (sell | (t - born >= 60))
        entries = ~active & buy
        active[exits] = False
        active[entries], born[entries], used[entries] = True, t, dates[4, entries]
        held[t] = active
    candidates = ~pd.DataFrame(held, index=close.index, columns=columns) & eligible
    selected = liquidity.where(candidates).rank(axis=1, ascending=False, method="first").le(10)
    return FinlabDataFrame(selected.loc[START:end])


if __name__ == "__main__":
    position = build_position()
    report = sim(position, position_limit=.1, upload=False, end_date=DATA_END)
    report.to_html("report.html")
    print(report.get_stats())