data/block_trades.py

"""
Taiwan block trades (鉅額交易) data via FinMind API.

Source: TaiwanStockBlockTrade
Fields: volume (block trade volume), close (block trade price), premium (vs market %)
"""

from __future__ import annotations

import logging
from datetime import date, timedelta

import pandas as pd
import requests
from cachetools import TTLCache

logger = logging.getLogger(__name__)

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

_BT_CACHE: TTLCache = TTLCache(maxsize=200, ttl=6 * 60 * 60)
_HEADERS = {"User-Agent": "Mozilla/5.0"}
_BT_COLS = ["block_vol_ratio", "block_premium"]


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


def fetch_block_trades(stock_no: str, months: int = 24) -> pd.DataFrame:
    bare = _normalize_code(stock_no)
    cache_key = f"bt:{bare}:{months}"
    cached = _BT_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": "TaiwanStockBlockTrade", "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 block trade fetch failed for %s: %s", bare, exc)
        return pd.DataFrame()

    if not records:
        return pd.DataFrame()

    try:
        df = pd.DataFrame(records)
        needed = [c for c in ("date", "volume", "close") if c in df.columns]
        if "date" not in needed:
            return pd.DataFrame()
        df = df[needed].copy()
        for col in ("volume", "close"):
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0.0)
        # Aggregate multiple block trades on same day
        df = df.groupby("date", as_index=False).agg(
            block_volume=("volume", "sum"),
            block_close=("close", "mean"),
        )
        df = df.sort_values("date").reset_index(drop=True)
        _BT_CACHE[cache_key] = df.copy()
        return df
    except Exception as exc:
        logger.warning("Block trade parse failed for %s: %s", bare, exc)
        return pd.DataFrame()


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

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

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

    bt = fetch_block_trades(stock_no)
    if bt.empty:
        return _zero_fill()

    out["_bdate"] = pd.to_datetime(out["date"]).dt.strftime("%Y-%m-%d")
    merged = out.merge(bt, how="left", left_on="_bdate", right_on="date", suffixes=("", "_bt"))
    merged = merged.drop(columns=["_bdate"] + [c for c in ["date_bt"] if c in merged.columns])

    block_vol = pd.to_numeric(merged.get("block_volume", 0), errors="coerce").fillna(0.0)
    volume = pd.to_numeric(out.get("volume", pd.Series(0.0, index=out.index)), errors="coerce").replace(0, float("nan"))
    close = pd.to_numeric(out.get("close", pd.Series(0.0, index=out.index)), errors="coerce").replace(0, float("nan"))
    block_close = pd.to_numeric(merged.get("block_close", float("nan")), errors="coerce")

    out["block_vol_ratio"] = (block_vol / volume).fillna(0.0).clip(upper=1.0)
    # Negative premium = block trade below market price = distribution signal
    out["block_premium"] = ((block_close - close) / close).fillna(0.0).clip(-0.1, 0.1)

    return out