clarifindata — data pipeline

Data-collection pipeline for clarifindata — a lean Taiwan-stock market-data API. This repository contains the free-tier ingestion code: the crawlers and backfill scripts that collect public market data from TWSE, TPEX, TAIFEX, MOPS, the central bank, and other public sources, plus the delivery layer that ships parsed rows into the data store.

Scope. This repo is the free-tier, open part of the system. Premium-source ingestion adapters and the serving/API + payments layer are part of the commercial product and are intentionally not included here. See ARCHITECTURE.md for how the published parts fit into the whole pipeline.

What's here

api/config.py                 # trimmed ingestion config (Kafka + MinIO, env-driven)
ingestion/
  publish.py                  # crawl → archive raw to MinIO → publish rows to Kafka
  kafka_io.py  minio_io.py     # generic Kafka / MinIO I/O helpers
  crawlers/                   # 7 streaming crawlers (publish to Kafka)
    bot_exchange_rate.py        # 央行牌告匯率
    twse_day_trading.py         # 當日沖銷交易統計
    twse_delisting.py           # 終止上市
    twse_fundamentals.py        # 股利政策 + 上市櫃總覽(含權證)
    twse_news.py                # 重大訊息
    twse_per.py                 # 本益比 / 殖利率 / 股價淨值比
    twse_total_inst.py          # 大盤三大法人買賣金額(市場合計)
    twse_weighted_index.py      # 加權指數 + 交易日曆
scripts/                      # 10 standalone backfillers (POST to /admin/ingest)
    macro_extra.py              # 通貨發行額
    mops_bulk_financials.py     # 財報(綜損 / 資產負債 / 現金流量)— OpenAPI
    mops_direct.py              # 財報 — direct MOPS HTML
    mops_worker.py              # 財報 — sharded worker
    mops_dividend_worker.py     # 股利
    mopsfin_quarterly.py        # 季度財報 / 資產負債
    taifex_contract_info.py     # 期權契約規格總覽
    twse_block_trade.py         # 鉅額交易
    twse_short_sale_balances.py # 融券借券賣出餘額
    ws_margin_total.py          # 大盤信用交易統計(市場合計)

Run locally

You need a Kafka broker + MinIO (for the streaming crawlers) or a running ingest API (for the backfill scripts). The easiest path is the docker-compose stack used in development (Postgres + Redis + Kafka + MinIO).

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

# Config via env (defaults target a local docker-compose stack):
export KAFKA_BOOTSTRAP_SERVERS=localhost:9092
export MINIO_ENDPOINT=localhost:9000
export MINIO_ACCESS_KEY=minioadmin MINIO_SECRET_KEY=minioadmin
export MINIO_BUCKET_RAW=clarifindata-raw

Streaming crawlers (→ Kafka → consumer → Postgres)

# Today's institutional-investor market totals
python -m ingestion.crawlers.twse_total_inst

# Backfill the last 30 days of PER / yield / PBR
python -m ingestion.crawlers.twse_per --backfill 30

Backfill scripts (→ POST /admin/ingest/)

These POST parsed rows to the ingest API with a bearer admin token (you supply your own; place it at ~/.secret/cfd_admin_token):

python scripts/twse_block_trade.py --recent 3
python scripts/ws_margin_total.py --start 2012-01-01 --end 2026-05-29 \
       --shard 0 --total 1 --ubuntu http://localhost:8000 --token "$(cat ~/.secret/cfd_admin_token)"

Reproducing the dataset (collect → store)

1. Bring up the local stack (Postgres + Kafka + MinIO) and the consumer/ingest API.
2. Run the relevant crawler/script for the date range you want (examples above); each is idempotent — re-running the same day upserts, it does not duplicate.
3. Rows land in Postgres; the live API then serves them. The hosted product is at https://clarifindata.com (see the clarifindata Python client).

Tiers

The hosted API gates datasets by tier. Everything collected by this repo is free-tier (market aggregates, macro, fundamentals, news). Per-stock prices, chips, adjusted prices, high-resolution and derivative data are paid tiers — see https://clarifindata.com/datasets.

License

MIT — see LICENSE.