data/large_holder.py

"""
Taiwan large holder percentage (大戶持股比例) from FinMind weekly shareholding data.

Source: TaiwanStockHoldingSharesPer
Tier: HoldingSharesLevel containing "1000" (1000+ lot holders = institutional/whale)
"""

from __future__ import annotations

import logging
from datetime import date, timedelta

import numpy as np
import pandas as pd
import requests
from cachetools import TTLCache

logger = logging.getLogger(__name__)

FINMIND_URL = "https://api.finmindtrade.com/api/v4/data"

_LH_CACHE: TTLCache = TTLCache(maxsize=200, ttl=24 * 60 * 60)
_HEADERS = {"User-Agent": "Mozilla/5.0"}
_LH_COLS = ["large_holder_pct", "large_holder_4w_change"]


def _normalize_code(stock_no: str) -> str:
    return stock_no.replace(".TW", "").replace(".TWO", "").strip()


def fetch_large_holder(stock_no: str, months: int = 24) -> pd.DataFrame:
    bare = _normalize_code(stock_no)
    cache_key = f"lh:{bare}:{months}"
    cached = _LH_CACHE.get(cache_key)
    if cached is not None:
        return cached.copy()

    start = (date.today().replace(day=1) - timedelta(days=months * 31)).strftime("%Y-%m-%d")
    try:
        resp = requests.get(
            FINMIND_URL,
            params={"dataset": "TaiwanStockHoldingSharesPer", "data_id": bare,
                    "start_date": start, "token": ""},
            headers=_HEADERS, timeout=15,
        )
        resp.raise_for_status()
        records = resp.json().get("data", [])
    except Exception as exc:
        logger.warning("FinMind large holder fetch failed for %s: %s", bare, exc)
        return pd.DataFrame()

    if not records:
        return pd.DataFrame()

    try:
        df = pd.DataFrame(records)
        # Filter to 1000+ lot tier (level label varies: "1000張以上", ">1000", etc.)
        mask = df["HoldingSharesLevel"].astype(str).str.contains("1000", na=False)
        df = df[mask][["date", "percent"]].copy()
        df["percent"] = pd.to_numeric(df["percent"], errors="coerce").fillna(0.0)
        df = df.sort_values("date").reset_index(drop=True)
        df.columns = ["date", "large_holder_pct"]
        _LH_CACHE[cache_key] = df.copy()
        return df
    except Exception as exc:
        logger.warning("Large holder parse failed for %s: %s", bare, exc)
        return pd.DataFrame()


def add_large_holder_features(df: pd.DataFrame, stock_no: str) -> pd.DataFrame:
    out = df.copy()

    def _zero_fill():
        for col in _LH_COLS:
            out[col] = 0.0
        return out

    if out.empty or "date" not in out.columns:
        return _zero_fill()

    lh = fetch_large_holder(stock_no)
    if lh.empty:
        return _zero_fill()

    # Use merge_asof to forward-fill weekly data into daily rows (no lookahead)
    daily_dates = pd.to_datetime(out["date"]).dt.normalize()
    lh_dates = pd.to_datetime(lh["date"]).dt.normalize()
    lh_sorted = lh.copy()
    lh_sorted["date_dt"] = lh_dates

    daily_df = pd.DataFrame({"date_dt": daily_dates})
    merged = pd.merge_asof(
        daily_df.sort_values("date_dt"),
        lh_sorted.sort_values("date_dt"),
        on="date_dt",
        direction="backward",
    )
    merged = merged.set_index(daily_df.sort_values("date_dt").index)

    pct = merged["large_holder_pct"].fillna(0.0)
    # 4-week (28-day) change in large holder %: ~4 weekly observations back
    pct_4w = pct.shift(4).replace(0, np.nan)
    change_4w = (pct - pct_4w).fillna(0.0)

    # Re-align to original df index order
    orig_order = daily_dates.argsort().argsort()
    out["large_holder_pct"] = pct.iloc[daily_dates.argsort()].values[orig_order] if len(pct) == len(out) else 0.0
    out["large_holder_4w_change"] = change_4w.iloc[daily_dates.argsort()].values[orig_order] if len(change_4w) == len(out) else 0.0

    return out