"""主動式 ETF 每日持股跟單研究：共同加碼、新進持股、共識組合；研究用途，不構成投資建議。"""
from functools import lru_cache

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

START, END = "2025-05-05", "2026-11-11"
# 持股九成以上是台股、當天晚上就公布持股的 24 檔主動式 ETF
ETFS = ["00400A", "00401A", "00403A", "00404A", "00405A", "00406A", "00407A", "00408A",
        "00410A", "00415A", "00416A", "00980A", "00981A", "00982A", "00984A", "00985A",
        "00987A", "00991A", "00992A", "00993A", "00994A", "00995A", "00996A", "00999A"]
MIN_HISTORY = 20  # ETF 滿 20 個揭露日才計算加碼與新進持股
EXCESS = 1.02     # 股數增幅要比受益權單位、同檔持股中位數都多 2% 以上


@lru_cache(None)
def calendar():
    return pd.DataFrame(data.get("price:收盤價")).loc[:END].index


@lru_cache(None)
def holdings():
    """每檔 ETF 每個揭露日的台股持股，附揭露序號與上一個揭露日。"""
    raw = pd.DataFrame(data.get("tw_active_etf_holdings_history"))
    h = pd.DataFrame({"etf": raw["symbol"].astype(str), "stock": raw["constituent_id"].astype(str),
                      "date": pd.to_datetime(raw["date"].astype(str)),
                      "shares": raw["shares"], "weight": raw["weight"]})
    close = pd.DataFrame(data.get("price:收盤價"))
    h = h[h["etf"].isin(ETFS) & h["stock"].isin(close.columns) & h["date"].le(END)]
    days = h[["etf", "date"]].drop_duplicates().sort_values(["etf", "date"])
    days["k"] = days.groupby("etf").cumcount()
    days["prev_date"] = days.groupby("etf")["date"].shift()
    # 持股日期換成交易日序號；非交易日的揭露算到下一個交易日
    days["pos"] = calendar().searchsorted(days["date"])
    return h.merge(days, on=["etf", "date"])


@lru_cache(None)
def unit_ratio():
    """受益權單位的日變動：持股市值回推基金規模，再除以淨值。"""
    h = holdings()
    # 只取持股期間的收盤價，避免把 20 年的價格整張攤平
    close = pd.DataFrame(data.get("price:收盤價")).loc[h["date"].min():].stack().rename("close")
    close.index.names = ["date", "stock"]
    priced = h.join(close, on=["date", "stock"], how="inner")
    priced = priced.assign(value=priced["shares"] * priced["close"])
    size = priced.groupby(["etf", "date"])[["value", "weight"]].sum()
    nav = pd.DataFrame(data.get("tw_etf_nav_daily:淨值")).stack().rename("nav")
    nav.index.names = ["date", "etf"]
    size = size.join(nav, on=["date", "etf"])
    units = (size["value"] / (size["weight"] / 100) / size["nav"]).rename("units").reset_index()
    units["ratio"] = units.groupby("etf")["units"].pct_change(fill_method=None) + 1
    return units.set_index(["etf", "date"])["ratio"]


@lru_cache(None)
def adds():
    """經理人真的加碼：扣掉申購贖回造成的等比例變動，排除配股與面額變更。"""
    h = holdings()
    prev = h[["etf", "stock", "date", "shares"]].rename(columns={"date": "prev_date", "shares": "prev_shares"})
    m = h.merge(prev, on=["etf", "stock", "prev_date"])
    m = m[m["prev_shares"].gt(0) & m["k"].ge(MIN_HISTORY)].copy()
    m["ratio"] = m["shares"] / m["prev_shares"]
    m["median"] = m.groupby(["etf", "date"])["ratio"].transform("median")
    m["units"] = unit_ratio().reindex(pd.MultiIndex.from_frame(m[["etf", "date"]])).to_numpy()
    # 兩檔以上 ETF 同一天、同一檔股票、股數同比例增加 → 配股或面額變更
    m["r4"] = m["ratio"].round(4)
    same = m[m["r4"].gt(1.01)].groupby(["stock", "date", "r4"])["etf"].transform("size")
    corporate = m.loc[same[same.ge(2)].index, ["stock", "date"]].drop_duplicates()
    m = m.merge(corporate.assign(corporate=True), on=["stock", "date"], how="left")
    added = (m["ratio"].ge(EXCESS * m["units"]) & m["ratio"].ge(EXCESS * m["median"])
             & m["corporate"].isna())
    return m.loc[added, ["etf", "stock", "date", "pos"]]


def _table(events):
    """把 (pos, stock) 事件放到交易日 × 股票的布林表。"""
    cal = calendar()
    events = events[events["pos"].lt(len(cal))]
    flag = pd.Series(True, index=pd.MultiIndex.from_arrays([cal[events["pos"]], events["stock"]]))
    flag = flag[~flag.index.duplicated()]
    return flag.unstack(fill_value=False).reindex(cal, fill_value=False)


def build_position(design="co_add", n=3, k=3, hold=10, min_weight=0.5, top=10):
    """回傳 (持股訊號, 事件數)；訊號日 D 收盤後形成，sim 在 D+1 開盤成交。"""
    if design == "co_add":
        # K 個交易日內有 N 檔以上不同的 ETF 加碼
        a = adds()
        span = pd.concat([a.assign(pos=a["pos"] + i) for i in range(k)])
        count = span.drop_duplicates(["etf", "stock", "pos"]).groupby(["pos", "stock"]).size()
        event = _table(count[count.ge(n)].reset_index())
    elif design == "new_entry":
        # 前 20 個揭露日都沒有、今天首次出現，首日權重至少 min_weight%
        h = holdings().sort_values("k")
        last_k = h.groupby(["etf", "stock"])["k"].shift()
        new = h["k"].ge(MIN_HISTORY) & (last_k.isna() | (h["k"] - last_k).gt(20)) & h["weight"].ge(min_weight)
        event = _table(h[new])
    else:
        return consensus(top), None
    position = event.rolling(hold, min_periods=1).max() > 0
    return position.loc[START:END], int(event.loc[START:END].to_numpy().sum())


@lru_cache(None)
def consensus_score():
    """24 檔 ETF 最新揭露權重加總，回傳 (加總權重, 名次)。"""
    cal, score = calendar(), 0
    for _, g in holdings().groupby("etf"):
        w = g.pivot_table(index="pos", columns="stock", values="weight", aggfunc="sum", fill_value=0)
        w.index = cal[w.index]
        score = w.reindex(cal).ffill().fillna(0).add(score, fill_value=0)
    return score, score.rank(axis=1, ascending=False, method="first")


def consensus(top=10):
    """加總權重前 top 檔等權。"""
    score, rank = consensus_score()
    return (rank.le(top) & score.gt(0)).loc[START:END]


def rebalance_dates(freq):
    cal = calendar()
    days = cal[cal >= START].to_series()
    ends = days.resample({"W": "W-FRI", "M": "ME", "Q": "QE"}[freq]).last().dropna()
    return pd.DatetimeIndex([days.index[0], *ends])


def build_report(design="co_add", freq="M", **params):
    position, _ = build_position(design, **params)
    if design == "consensus":
        # 每週、每月、每季期末換成最新的前 top 檔
        options = {"resample": rebalance_dates(freq)}
    else:
        # 單檔上限 10%，訊號少時其餘放現金
        options = {"position_limit": 0.1}
    # sim 使用台股預設手續費與證交稅
    return sim(position, trade_at_price="open", upload=False, **options)


if __name__ == "__main__":
    build_report("consensus", top=10, freq="M").to_html("report_strategy.html")