"""跌停事件研究；所有績效為研究用途，不代表能在跌停板成交。"""
from functools import lru_cache

import numpy as np
import pandas as pd
from finlab import data
from finlab.backtest import sim

START, END = "2015-01-05", "2026-11-10"


def floor_price(previous, dates):
    limit = np.where(dates < pd.Timestamp("2015-06-01"), 0.07, 0.10)
    raw = previous.mul(1 - limit, axis=0)
    tick = np.select([raw < 10, raw < 50, raw < 100, raw < 500, raw < 1000],
                     [0.01, 0.05, 0.1, 0.5, 1], default=5)
    return pd.DataFrame(np.ceil(raw / tick - 1e-7) * tick,
                        index=previous.index, columns=previous.columns)


@lru_cache(None)
def inputs():
    # Eager pandas is intentional: inspect and join sparse event tables.
    close = pd.DataFrame(data.get("price:收盤價"))
    cols = [s for s in close if len(s) == 4 and s.isdigit()
            and not s.startswith(("0", "91"))]
    close = close[cols].loc[:END]
    high = pd.DataFrame(data.get("price:最高價")).reindex_like(close)
    low = pd.DataFrame(data.get("price:最低價")).reindex_like(close)
    adj = pd.DataFrame(data.get("etl:adj_close")).reindex_like(close)
    value = pd.DataFrame(data.get("price:成交金額")).reindex_like(close)
    # Exclude adjusted-price factor changes (dividends, splits, capital actions).
    ordinary = (adj / close).pct_change(fill_method=None).abs().lt(1e-6)
    seasoned = close.notna().cumsum() > 5
    floor = floor_price(close.shift(1), close.index)
    hit = (close - floor).abs().lt(0.001) & ordinary & seasoned & value.gt(0)
    hit = hit.loc[START:END]
    locked = hit & high.reindex_like(hit).sub(low.reindex_like(hit)).abs().lt(0.001)
    liquid = value.rolling(20, min_periods=20).mean().reindex_like(hit)
    return hit, locked, liquid, floor


def build_position(kind="all", hold=5, liquidity=20_000_000, quantile=0.975):
    hit, locked, liquid, _ = inputs()
    if kind in ("panic", "avoid"):
        count = hit.sum(axis=1)
        threshold = count.shift(1).rolling(252, min_periods=252).quantile(quantile)
        event = count.gt(threshold) & count.gt(0)
        active = event.astype(int).rolling(hold, min_periods=1).max().astype(bool)
        if kind == "avoid":
            active = ~active
        return pd.DataFrame({"0050": active}), int(event.sum())
    event = {"all": hit, "locked": locked, "opened": hit & ~locked}[kind]
    event = event & liquid.ge(liquidity)
    position = event.astype(int).rolling(hold, min_periods=1).max().astype(bool)
    return position, int(event.to_numpy().sum())


def build_report(kind="all", hold=5, liquidity=20_000_000, quantile=0.975):
    position, _ = build_position(kind, hold, liquidity, quantile)
    # Signals use day-t close; sim executes at day-t+1 open with TW default fees/tax.
    return sim(position, trade_at_price="open", position_limit=1 if kind in ("panic", "avoid") else 0.1,
               upload=False)


if __name__ == "__main__":
    build_report().to_html("report_strategy.html")