Spaces:
Sleeping
Sleeping
File size: 3,756 Bytes
84c1c64 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | """
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
|