HKEX Disclosure of Interests — Daily Summary Scraper

by psind707

A Mac Mini–friendly scraper for Hong Kong Stock Exchange's DI daily summaries. Tested live against the 1 April 2026 publication: 234 substantial-shareholder filings parsed cleanly, including Chinese filer names, derivative position changes, and split L/S/P balances.

This page is the brief you can hand to Claude Code. The scraper is functional out of the box — extensions are listed at the bottom.

Why this exists

Unlike Japan's EDINET, HKEX has no public API for DI data. The only structured access is the daily summary HTML at https://di.hkex.com.hk/di/summary/DSM{YYYYMMDD}{C1|C2|C3|C4}.htm. We scrape it.

Four daily reports, two of them high-signal:

HKEX filing speed: 3 business days vs Japan's 5 — and the C2 director feed is granular at the individual-trade level, which Japan never gives you.

Live-tested output sample

A real-world row from --reports C1 --end-date 2026-04-01:

- `2026-03-31` [2020] **ANTA Sports Products Ltd.** ← Anta International Group Holdings Limited (安踏國際集團控股有限公司)
    1101 On-exchange purchase | traded=3,170,000(L) @ HKD 75.2450 | now=+53.02% (Δ +0.12%)
    https://di.hkex.com.hk/di/NSForm2.aspx?fn=CS20260401E00253&...

Schema captured per row: filing_id, listed_corp, stock_code (5-digit zero-padded), share_class, filer, event_date (ISO), reason_code + label, shares_traded, side (L/S/P), avg_price_hkd, prev/curr balances (separate L/S/P fields), Δ in percentage points, form URL.

Reason codes (the partial reference)

HKEX's full code list is at https://di.hkex.com.hk/di/NSStdCode.htm but it's behind dynamic rendering. The script ships with this seed dictionary covering the codes you'll see ~95% of the time. Unknown codes pass through as the raw 4-digit code.

Setup

pip install requests beautifulsoup4 lxml
chmod +x hkex_di_monitor.py
./hkex_di_monitor.py --help

Common usage:

# Yesterday's director trades (the highest-signal feed — no threshold)
python3 hkex_di_monitor.py --reports C2 --days 1

# Yesterday's 5%+ filings (substantial shareholders)
python3 hkex_di_monitor.py --reports C1 --days 1

# Both, last 7 days, watchlist filter (4- or 5-digit codes both work)
python3 hkex_di_monitor.py --reports C1,C2 --days 7 --watchlist 700,9988,2382

# Hunt for one filer (substring match, works on Chinese names too)
python3 hkex_di_monitor.py --reports C1 --days 30 --filer "BlackRock"
python3 hkex_di_monitor.py --reports C1 --days 30 --filer "Templeton"

# Only show large position changes
python3 hkex_di_monitor.py --reports C1 --days 7 --min-delta-pp 0.5

# Persist to SQLite + only alert on new rows (cron-safe)
python3 hkex_di_monitor.py --reports C1,C2 --days 1 \
    --db ./hkex_di.db --new-only --telegram

# JSON for downstream
python3 hkex_di_monitor.py --reports C2 --days 1 --json

Cron on Mac Mini

# HKEX publishes throughout the day; run at 22:00 HKT (after EOD wave)
0 22 * * 1-5 cd ~/hkex && /usr/bin/python3 hkex_di_monitor.py \
    --reports C1,C2 --days 1 --db ./hkex_di.db --new-only --telegram \
    >> logs/$(date +\%Y\%m).md 2>&1

Telegram credentials via env: TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID. Same hook as the EDINET monitor.

The full script

#!/usr/bin/env python3
"""
HKEX Disclosure of Interests — Daily Summary Scraper
=====================================================
Scrapes the four DI daily summaries from HKEX into a normalized SQLite
store. The two high-signal feeds for trading:

  C1 = Substantial shareholders (5%+, like Japan's 大量保有報告書)
  C2 = Director trades in their own listed company's shares
       (NO threshold — every trade reported, this is the gold)

  C3 = Director trades in associated corps (rarely actionable)
  C4 = Director debenture trades (rarely actionable)

Why scrape: HKEX has no public API for DI data. The daily summary HTML
is the only structured access. Layout has been stable for years but
will eventually break — defensive parsing throughout.

Setup
-----
  pip install requests beautifulsoup4 lxml
  python hkex_di_monitor.py --help

Notes
-----
- HKEX dates are DD/MM/YYYY. We normalize to ISO YYYY-MM-DD throughout.
- Stock codes are 5-digit zero-padded (e.g. 00700 for Tencent). We accept
  4- or 5-digit watchlist input.
- Position suffix: (L) Long, (S) Short, (P) Lending Pool. We split these
  into separate fields rather than collapsing.
- HKEX publishes weekend/holiday pages too but they contain "no records"
  text — we detect and skip silently.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sqlite3
import sys
import time
from dataclasses import dataclass, asdict, field
from datetime import date, datetime, timedelta, timezone
from typing import Any, Iterable

import requests
from bs4 import BeautifulSoup, Tag

# -------------------------- Constants --------------------------

BASE_URL = "https://di.hkex.com.hk/di/summary"
USER_AGENT = "hkex-di-monitor/1.0 (personal trading research)"
TIMEOUT = 30
RATE_LIMIT_SLEEP = 1.5

REPORT_NAMES = {
    "C1": "Substantial Shareholders (5%+)",
    "C2": "Director — Listed Company Shares",
    "C3": "Director — Associated Corporation Shares",
    "C4": "Director — Debentures",
}

REASON_CODES = {
    "1001": "First became aware of >=5% interest",
    "1002": "Crossed up through whole % point",
    "1003": "Crossed down through whole % point",
    "1004": "Ceased to have notifiable interest",
    "1005": "Change in nature of interest",
    "1006": "Initial notification — appointment as director",
    "1007": "Initial notification — listing",
    "1101": "On-exchange purchase",
    "1102": "On-exchange sale",
    "1103": "Off-exchange purchase",
    "1104": "Off-exchange sale",
    "1199": "Other on/off-exchange transaction",
    "1201": "Subscription to new shares",
    "1202": "Exercise of options/warrants",
    "1203": "Conversion of convertibles",
    "1205": "Share buyback by issuer (resulting % change)",
    "1206": "Bonus issue / stock dividend / split",
    "1313": "Change in interest in equity derivatives",
    "1314": "Cash-settled equity derivative position change",
    "1505": "Securities lending — return of borrowed shares",
    "1704": "Securities borrowing/lending pool change",
}


# -------------------------- Data model --------------------------

@dataclass
class DIBalance:
    long_shares: int | None = None
    long_pct: float | None = None
    short_shares: int | None = None
    short_pct: float | None = None
    pool_shares: int | None = None
    pool_pct: float | None = None


@dataclass
class DIRow:
    report_date: str
    report_type: str
    filing_id: str
    listed_corp: str
    stock_code: str
    share_class: str
    filer: str
    event_date: str
    reason_code: str
    reason_label: str
    shares_traded: int | None
    shares_traded_side: str | None
    avg_price_hkd: float | None
    prev_balance: DIBalance
    curr_balance: DIBalance
    delta_long_pct: float | None
    form_url: str

    @property
    def stock_code_4(self) -> str:
        try:
            return str(int(self.stock_code))
        except (TypeError, ValueError):
            return self.stock_code


# -------------------------- Parsing --------------------------

NUM_RE = re.compile(r"[-+]?\d[\d,]*")
SUFFIX_RE = re.compile(r"\((L|S|P)\)")
PRICE_RE = re.compile(r"HKD\s*([\d,]+\.?\d*)", re.IGNORECASE)


def parse_int(s: str) -> int | None:
    if not s:
        return None
    m = NUM_RE.search(s.replace(",", ""))
    if not m:
        return None
    try:
        return int(m.group(0))
    except ValueError:
        return None


def parse_float(s: str) -> float | None:
    if not s:
        return None
    s = s.strip().replace(",", "")
    try:
        return float(s)
    except ValueError:
        return None


def parse_shares_with_suffix(s: str) -> tuple[int | None, str | None]:
    if not s:
        return None, None
    suffix_match = SUFFIX_RE.search(s)
    side = suffix_match.group(1) if suffix_match else None
    cleaned = SUFFIX_RE.sub("", s).strip()
    return parse_int(cleaned), side


def parse_date_ddmmyyyy(s: str) -> str | None:
    if not s:
        return None
    s = s.strip()
    m = re.match(r"(\d{1,2})/(\d{1,2})/(\d{4})", s)
    if not m:
        return None
    d, mo, y = m.groups()
    try:
        return date(int(y), int(mo), int(d)).isoformat()
    except ValueError:
        return None


def parse_balance_cell(cell: Tag) -> DIBalance:
    bal = DIBalance()
    if cell is None:
        return bal
    inner = cell.find("table")
    if inner is None:
        return bal
    rows = inner.find_all("tr")
    for row in rows:
        cells = row.find_all("td")
        if len(cells) < 2:
            continue
        shares, side = parse_shares_with_suffix(cells[0].get_text(" ", strip=True))
        pct = parse_float(cells[1].get_text(" ", strip=True))
        if side == "L":
            bal.long_shares = shares
            bal.long_pct = pct
        elif side == "S":
            bal.short_shares = shares
            bal.short_pct = pct
        elif side == "P":
            bal.pool_shares = shares
            bal.pool_pct = pct
    return bal


def parse_balance_pair(prev_cell: Tag, curr_cell: Tag) -> tuple[DIBalance, DIBalance]:
    return parse_balance_cell(prev_cell), parse_balance_cell(curr_cell)


def parse_summary_html(html: str, report_date: str, report_type: str) -> list[DIRow]:
    soup = BeautifulSoup(html, "lxml")
    tables = soup.find_all("table")
    data_table = None
    for tbl in tables:
        text = tbl.get_text(" ", strip=True)
        if "Serial No" in text and "Stock code" in text:
            data_table = tbl
            break
    if data_table is None:
        return []

    rows: list[DIRow] = []
    for tr in data_table.find_all("tr"):
        cells = tr.find_all("td", recursive=False)
        if len(cells) < 11:
            continue
        first_text = cells[0].get_text(" ", strip=True)
        if not first_text or "Serial No" in first_text:
            continue
        try:
            filing_link = cells[0].find("a")
            filing_id = (filing_link.get_text(strip=True) if filing_link
                         else cells[0].get_text(strip=True))
            form_url = ""
            if filing_link and filing_link.get("href"):
                href = filing_link["href"]
                if href.startswith("http"):
                    form_url = href
                else:
                    form_url = f"https://di.hkex.com.hk{href if href.startswith('/') else '/di/' + href}"

            listed_corp = cells[1].get_text(" ", strip=True)
            stock_code = cells[2].get_text(strip=True).zfill(5)
            share_class = cells[3].get_text(" ", strip=True)
            filer = cells[4].get_text(" ", strip=True)
            event_date = parse_date_ddmmyyyy(cells[5].get_text(strip=True)) or ""

            reason_text = cells[6].get_text(" ", strip=True)
            code_match = re.search(r"\b(\d{4})\b", reason_text)
            reason_code = code_match.group(1) if code_match else reason_text
            reason_label = REASON_CODES.get(reason_code, "")

            shares_traded, traded_side = parse_shares_with_suffix(
                cells[7].get_text(" ", strip=True))

            price_text = cells[8].get_text(" ", strip=True) if len(cells) > 8 else ""
            price_match = PRICE_RE.search(price_text)
            avg_price = parse_float(price_match.group(1)) if price_match else None

            prev_bal, curr_bal = parse_balance_pair(
                cells[9] if len(cells) > 9 else None,
                cells[10] if len(cells) > 10 else None,
            )

            delta_pct = None
            if curr_bal.long_pct is not None and prev_bal.long_pct is not None:
                delta_pct = round(curr_bal.long_pct - prev_bal.long_pct, 4)

            rows.append(DIRow(
                report_date=report_date,
                report_type=report_type,
                filing_id=filing_id,
                listed_corp=listed_corp,
                stock_code=stock_code,
                share_class=share_class,
                filer=filer,
                event_date=event_date,
                reason_code=reason_code,
                reason_label=reason_label,
                shares_traded=shares_traded,
                shares_traded_side=traded_side,
                avg_price_hkd=avg_price,
                prev_balance=prev_bal,
                curr_balance=curr_bal,
                delta_long_pct=delta_pct,
                form_url=form_url,
            ))
        except Exception as e:
            print(f"  ! row parse error in {report_type} {report_date}: {e}",
                  file=sys.stderr)
            continue
    return rows


# -------------------------- Fetching --------------------------

def fetch_summary(target_date: date, report_type: str,
                  session: requests.Session) -> list[DIRow]:
    url = f"{BASE_URL}/DSM{target_date.strftime('%Y%m%d')}{report_type}.htm"
    try:
        resp = session.get(url, timeout=TIMEOUT)
    except requests.RequestException as e:
        print(f"  ! fetch error {url}: {e}", file=sys.stderr)
        return []
    if resp.status_code == 404:
        return []
    if resp.status_code != 200:
        print(f"  ! HTTP {resp.status_code} for {url}", file=sys.stderr)
        return []
    resp.encoding = "utf-8"
    return parse_summary_html(resp.text, target_date.isoformat(), report_type)


# -------------------------- SQLite store --------------------------

DDL = """
CREATE TABLE IF NOT EXISTS di_filings (
  filing_id        TEXT PRIMARY KEY,
  report_date      TEXT NOT NULL,
  report_type      TEXT NOT NULL,
  listed_corp      TEXT NOT NULL,
  stock_code       TEXT NOT NULL,
  share_class      TEXT,
  filer            TEXT NOT NULL,
  event_date       TEXT,
  reason_code      TEXT,
  reason_label     TEXT,
  shares_traded    INTEGER,
  traded_side      TEXT,
  avg_price_hkd    REAL,
  prev_long_shares INTEGER,
  prev_long_pct    REAL,
  prev_short_shares INTEGER,
  prev_short_pct   REAL,
  curr_long_shares INTEGER,
  curr_long_pct    REAL,
  curr_short_shares INTEGER,
  curr_short_pct   REAL,
  delta_long_pct   REAL,
  form_url         TEXT,
  alerted_at       TEXT,
  inserted_at      TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_di_stock      ON di_filings(stock_code);
CREATE INDEX IF NOT EXISTS idx_di_filer      ON di_filings(filer);
CREATE INDEX IF NOT EXISTS idx_di_event_date ON di_filings(event_date);
CREATE INDEX IF NOT EXISTS idx_di_alerted    ON di_filings(alerted_at);
"""


class DIStore:
    def __init__(self, path: str):
        self.conn = sqlite3.connect(path)
        self.conn.executescript(DDL)
        self.conn.commit()

    def insert_new(self, rows: list[DIRow]) -> list[DIRow]:
        new_rows: list[DIRow] = []
        cur = self.conn.cursor()
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        for r in rows:
            cur.execute("SELECT 1 FROM di_filings WHERE filing_id = ?", (r.filing_id,))
            if cur.fetchone():
                continue
            cur.execute("""
                INSERT INTO di_filings (
                  filing_id, report_date, report_type, listed_corp, stock_code,
                  share_class, filer, event_date, reason_code, reason_label,
                  shares_traded, traded_side, avg_price_hkd,
                  prev_long_shares, prev_long_pct, prev_short_shares, prev_short_pct,
                  curr_long_shares, curr_long_pct, curr_short_shares, curr_short_pct,
                  delta_long_pct, form_url, inserted_at
                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
            """, (
                r.filing_id, r.report_date, r.report_type, r.listed_corp, r.stock_code,
                r.share_class, r.filer, r.event_date, r.reason_code, r.reason_label,
                r.shares_traded, r.shares_traded_side, r.avg_price_hkd,
                r.prev_balance.long_shares, r.prev_balance.long_pct,
                r.prev_balance.short_shares, r.prev_balance.short_pct,
                r.curr_balance.long_shares, r.curr_balance.long_pct,
                r.curr_balance.short_shares, r.curr_balance.short_pct,
                r.delta_long_pct, r.form_url, now,
            ))
            new_rows.append(r)
        self.conn.commit()
        return new_rows

    def mark_alerted(self, rows: list[DIRow]) -> None:
        now = datetime.now(timezone.utc).isoformat(timespec="seconds")
        self.conn.executemany(
            "UPDATE di_filings SET alerted_at=? WHERE filing_id=?",
            [(now, r.filing_id) for r in rows],
        )
        self.conn.commit()


# -------------------------- Filtering --------------------------

def filter_rows(rows: list[DIRow], watchlist: set[str] | None,
                filer_substring: str | None,
                min_delta_pp: float | None) -> list[DIRow]:
    out = []
    for r in rows:
        if watchlist:
            if not (r.stock_code in watchlist or r.stock_code_4 in watchlist):
                continue
        if filer_substring and filer_substring.lower() not in r.filer.lower():
            continue
        if min_delta_pp is not None:
            if r.delta_long_pct is None or abs(r.delta_long_pct) < min_delta_pp:
                continue
        out.append(r)
    return out


# -------------------------- Output --------------------------

def fmt_pct(v: Any) -> str:
    if v is None:
        return "—"
    try:
        return f"{float(v):+.2f}%"
    except (TypeError, ValueError):
        return "—"


def fmt_shares(v: Any) -> str:
    if v is None:
        return "—"
    try:
        return f"{int(v):,}"
    except (TypeError, ValueError):
        return str(v)


def fmt_price(v: Any) -> str:
    if v is None:
        return "—"
    return f"HKD {v:,.4f}"


def render_md(rows: list[DIRow]) -> str:
    if not rows:
        return "_No matching DI filings._\n"
    by_type: dict[str, list[DIRow]] = {}
    for r in rows:
        by_type.setdefault(r.report_type, []).append(r)
    lines = [f"# HKEX DI — {len(rows)} filings\n"]
    for rt, items in sorted(by_type.items()):
        lines.append(f"\n## {rt} — {REPORT_NAMES.get(rt, rt)} ({len(items)})\n")
        items.sort(key=lambda x: (x.event_date or "", x.filing_id), reverse=True)
        for r in items:
            delta = fmt_pct(r.delta_long_pct) if r.delta_long_pct is not None else "—"
            curr_long = fmt_pct(r.curr_balance.long_pct)
            traded = fmt_shares(r.shares_traded)
            side = f"({r.shares_traded_side})" if r.shares_traded_side else ""
            reason = f"{r.reason_code}"
            if r.reason_label:
                reason += f" {r.reason_label}"
            lines.append(
                f"- `{r.event_date}` [{r.stock_code_4}] **{r.listed_corp}** "
                f"← {r.filer}"
            )
            lines.append(
                f"    {reason} | traded={traded}{side} @ {fmt_price(r.avg_price_hkd)} "
                f"| now={curr_long} (Δ {delta})"
            )
            if r.form_url:
                lines.append(f"    {r.form_url}")
    return "\n".join(lines) + "\n"


def render_json(rows: list[DIRow]) -> str:
    def _ser(o):
        if hasattr(o, "__dict__"):
            return o.__dict__
        return o
    return json.dumps([asdict(r) for r in rows], ensure_ascii=False,
                      indent=2, default=_ser)


# -------------------------- Telegram --------------------------

def telegram_alert(text: str) -> None:
    token = os.environ.get("TELEGRAM_BOT_TOKEN")
    chat_id = os.environ.get("TELEGRAM_CHAT_ID")
    if not (token and chat_id):
        return
    for chunk in [text[i:i+4000] for i in range(0, len(text), 4000)]:
        try:
            requests.post(
                f"https://api.telegram.org/bot{token}/sendMessage",
                json={"chat_id": chat_id, "text": chunk, "parse_mode": "Markdown",
                      "disable_web_page_preview": True},
                timeout=10,
            )
        except requests.RequestException as e:
            print(f"  ! telegram error: {e}", file=sys.stderr)


# -------------------------- CLI --------------------------

def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="HKEX DI daily summary scraper")
    p.add_argument("--reports", type=str, default="C1,C2",
                   help="Comma-separated report types: C1,C2,C3,C4 (default C1,C2)")
    p.add_argument("--days", type=int, default=1,
                   help="Days back to scan (default 1)")
    p.add_argument("--end-date", type=str, default=None,
                   help="End date YYYY-MM-DD (default = today)")
    p.add_argument("--watchlist", type=str, default=None,
                   help="Comma-separated 4- or 5-digit HK stock codes")
    p.add_argument("--filer", type=str, default=None,
                   help="Substring match against filer name (e.g. 'BlackRock')")
    p.add_argument("--min-delta-pp", type=float, default=None,
                   help="Min |delta| in percentage points (e.g. 0.5)")
    p.add_argument("--db", type=str, default=None,
                   help="Path to SQLite DB. Enables persistence + dedup.")
    p.add_argument("--new-only", action="store_true",
                   help="With --db: only show/alert filings new to the DB")
    p.add_argument("--telegram", action="store_true",
                   help="Send results to Telegram (TELEGRAM_BOT_TOKEN, _CHAT_ID)")
    p.add_argument("--json", action="store_true", help="Emit JSON instead of Markdown")
    return p.parse_args()


def main() -> int:
    args = parse_args()
    end = date.fromisoformat(args.end_date) if args.end_date else date.today()
    days = [end - timedelta(days=i) for i in range(args.days)]
    report_types = [t.strip() for t in args.reports.split(",") if t.strip() in REPORT_NAMES]
    if not report_types:
        sys.exit("ERROR: --reports must include at least one of C1,C2,C3,C4")

    watchlist = None
    if args.watchlist:
        watchlist = {w.strip().zfill(5) if w.strip().isdigit() else w.strip()
                     for w in args.watchlist.split(",") if w.strip()}
        watchlist |= {w.lstrip("0") for w in watchlist}

    session = requests.Session()
    session.headers.update({"User-Agent": USER_AGENT})

    all_rows: list[DIRow] = []
    last_call = 0.0
    for d in days:
        for rt in report_types:
            elapsed = time.time() - last_call
            if elapsed < RATE_LIMIT_SLEEP:
                time.sleep(RATE_LIMIT_SLEEP - elapsed)
            last_call = time.time()
            print(f"  · {d.isoformat()} {rt}…", file=sys.stderr)
            rows = fetch_summary(d, rt, session)
            all_rows.extend(rows)

    if args.db:
        store = DIStore(args.db)
        new_rows = store.insert_new(all_rows)
        display_rows = new_rows if args.new_only else all_rows
    else:
        display_rows = all_rows

    filtered = filter_rows(display_rows, watchlist, args.filer, args.min_delta_pp)

    out = render_json(filtered) if args.json else render_md(filtered)
    print(out)

    if args.telegram and filtered and not args.json:
        telegram_alert(out)
        if args.db:
            store.mark_alerted(filtered)

    print(
        f"\n# {len(all_rows)} fetched, {len(display_rows)} considered, "
        f"{len(filtered)} matched. range={days[-1]}..{days[0]} types={report_types}",
        file=sys.stderr,
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())

Live test results (1 April 2026)

$ python3 hkex_di_monitor.py --reports C1 --end-date 2026-04-01 --days 1
  · 2026-04-01 C1…
# HKEX DI — 234 filings

$ # Top 5 stocks by filing count
02496 Wuhan YZY Biopharma Co., Ltd. - B - H Shares       14
06698 STAR CM Holdings Ltd.                              8
00960 Longfor Group Holdings Ltd.                        7
01989 Delton Technology (Guangzhou) Inc. - H Shares      7
02701 Nsing Technologies Inc. - H Shares                 6

$ # Top 5 filers
  75  BlackRock, Inc.
  14  JPMorgan Chase & Co.
   5  Morgan Stanley
   5  UBS Group AG
   3  CITIC Securities Company Limited

BlackRock filing 75 times in a single day is normal — most are sub-1pp derivative position changes (codes 1313 / 1314). Use --min-delta-pp 0.5 to filter the noise.

Next-step extensions (priority order)

For Claude Code to work through.

1. Filer normalization

Same problem as Japan: "BlackRock, Inc." vs "BlackRock Fund Advisors" vs "BlackRock Investment Management" all appear separately. Build filer_aliases.yml and a parent_group_id resolver. Activist priors for HK that are worth catching:

oasis:
  - Oasis Management
  - Oasis Investments
elliott:
  - Elliott Investment Management
  - Elliott Associates
silchester:
  - Silchester International Investors
murakami_hk:
  - レノ
  - シティインデックスイレブンス
  - Reno

The Murakami group also files HK DIs occasionally — same parent_group_id can serve both EDINET and HKEX once normalized.

2. CCASS shareholding cross-reference

Daily snapshot of which custodian/broker accounts hold which percentage of each stock at https://www.hkexnews.hk/sdw/search/stocklist.aspx — when shares migrate to one specific broker (HSBC, Citi, UBS Wealth) it sometimes precedes a 5%+ DI filing by a few days. Build a CCASS scraper alongside, then flag stocks where:

• Single-broker concentration jumped >2pp in last 5 days
• AND a 5%+ DI filing appears for that name

That's a high-confidence accumulation signal for HK mid/small caps.

3. Director-trade momentum signals

C2 is unique to HK (Japan has no equivalent). Two patterns worth backtesting:

• Cluster buys: 3+ different directors of the same company filing on-exchange purchases (1101) within 5 trading days. Strong insider conviction signal in HK small/mid caps.
• Cluster sells around lockup expiry: directors pre-IPO + insider sells within 30 days of lockup. Negative signal.

EDINET DB doesn't have this — neither does any Japanese activist database. It's a Hong Kong–specific edge.

4. Stock Connect (Northbound/Southbound) overlay

For H-shares of mainland Chinese companies, also scrape https://www.hkexnews.hk/sdw/search/mutualmarket.aspx?t=hk — daily mainland-investor holdings. Catch when southbound flow accelerates ahead of a 5%+ filing. Useful for Tencent / Alibaba / Meituan / BYD types.

5. webb-site.com cross-reference

David Webb's free HK governance database has the deepest director-network and related-party-transaction data anywhere outside Bloomberg. No API but scrape-friendly. For a name that appears in C1/C2, query webb-site for the director's other directorships → catch related-party patterns the official disclosure doesn't surface.

6. Unify with EDINET DB store

Both monitors output structured filings. Merge into one filings table keyed by (market, filing_id) with a common schema for filer / issuer / event_date / delta. Single Telegram digest each evening covering JP + HK at once.

Things to know / gotchas

• No API, only HTML scraping. HKEX has been stable for years but they will redesign eventually. Defensive parsing in the script absorbs minor changes; major changes will require updating the cell-index map in parse_summary_html.
• Layout signature. The script identifies the data table by looking for "Serial No" and "Stock code" in its text. This is brittle to translation changes but stable across Chinese/English versions.
• Stock codes are 5-digit. HKEX zero-pads internally (Tencent = 00700). The script accepts both 700 and 00700 in --watchlist.
• No coverage of CCASS / Stock Connect. Those are different scrapers — see extensions 2 and 4.
• Special purpose entities are noisy. "in the capacity as the Receiver", "Trustee", and similar wrapper entities show up as separate filers. Filer normalization (#1) will help.
• BlackRock noise. They file 50–100 times per day. Either filter them out (--filer "" --filer-exclude BlackRock would be a future enhancement) or use --min-delta-pp 0.5 to skip their thousand sub-1pp derivative tweaks.
• Polite scraping. 1.5s sleep between requests. ~6-8s for a full day's fetch (4 reports). Don't increase concurrency without checking robots.txt.
• Rate limits. No documented limits but treat as if there are. Daily cron once is fine; please don't hammer it.