tests/test_institutional_flow_cache.py

from collections import namedtuple

from data.institutional_flow_cache import (
    fetch_daily_rows,
    fetch_recent_rows_by_code,
    get_fetch_status,
    init_db,
    upsert_daily_rows,
    upsert_fetch_status,
)
from data.tw_universe import build_universe_records


def test_build_universe_records_from_synthetic_twstock_codes():
    Code = namedtuple("Code", "type code name ISIN start market group CFI")
    records = build_universe_records(
        {
            "2330": Code("股票", "2330", "台積電", "", "", "上市", "半導體業", ""),
            "006201": Code("ETF", "006201", "元大富櫃50", "", "", "上櫃", "", ""),
            "01001T": Code("受益證券", "01001T", "測試受益證券", "", "", "上市", "", ""),
        }
    )

    assert records == [
        {
            "code": "006201",
            "name": "元大富櫃50",
            "market": "上櫃",
            "exchange": "TPEX",
            "type": "ETF",
            "group": "",
            "subsector": "未分類",
            "subsector_path": "未分類",
            "subsector_source": "industry_fallback",
            "subsector_refined": False,
        },
        {
            "code": "01001T",
            "name": "測試受益證券",
            "market": "上市",
            "exchange": "TWSE",
            "type": "受益證券",
            "group": "",
            "subsector": "未分類",
            "subsector_path": "未分類",
            "subsector_source": "industry_fallback",
            "subsector_refined": False,
        },
        {
            "code": "2330",
            "name": "台積電",
            "market": "上市",
            "exchange": "TWSE",
            "type": "股票",
            "group": "半導體業",
            "subsector": "晶圓代工",
            "subsector_path": "半導體業 > 晶圓代工",
            "subsector_source": "code_override",
            "subsector_refined": True,
        },
    ]


def test_upsert_and_fetch_daily_rows_round_trip(tmp_path):
    db_path = tmp_path / "flow.sqlite3"
    init_db(db_path).close()

    written = upsert_daily_rows(
        db_path,
        [
            {
                "date": "2026-05-18",
                "stock_no": "2330.TW",
                "foreign_buy": 100,
                "foreign_sell": 80,
                "foreign_net": 20,
                "trust_net": 3,
                "dealer_net": -2,
                "institutional_net": 21,
                "institutional_source": "TWSE_T86",
                "institutional_as_of": "2026-05-18",
            },
            {
                "date": "2026-05-19",
                "stock_no": "2330",
                "foreign_net": 30,
                "trust_net": 4,
                "dealer_net": 5,
                "institutional_net": 39,
                "institutional_available": False,
                "institutional_carry_forward": True,
            },
        ],
    )

    rows = fetch_daily_rows(db_path, "2330.TW", "2026-05-01", "2026-05-31")

    assert written == 2
    assert [row["date"] for row in rows] == ["2026-05-18", "2026-05-19"]
    assert rows[0]["stock_no"] == "2330"
    assert rows[0]["foreign_buy"] == 100
    assert rows[0]["trust_buy"] == 0
    assert rows[0]["institutional_source"] == "TWSE_T86"
    assert rows[0]["institutional_available"] is True
    assert rows[1]["institutional_available"] is False
    assert rows[1]["institutional_carry_forward"] is True


def test_upsert_daily_rows_replaces_existing_row(tmp_path):
    db_path = tmp_path / "flow.sqlite3"
    upsert_daily_rows(
        db_path,
        [{"date": "2026-05-18", "stock_no": "2330", "institutional_net": 10}],
    )
    upsert_daily_rows(
        db_path,
        [{"date": "2026-05-18", "stock_no": "2330", "institutional_net": 99}],
    )

    rows = fetch_daily_rows(db_path, "2330", "2026-05-18", "2026-05-18")

    assert len(rows) == 1
    assert rows[0]["institutional_net"] == 99


def test_fetch_status_round_trip(tmp_path):
    db_path = tmp_path / "flow.sqlite3"

    upsert_fetch_status(
        db_path,
        "TWSE_T86",
        "2026-05-18",
        "ok",
        row_count=123,
        fetched_at="2026-05-18T09:10:00+00:00",
    )

    assert get_fetch_status(db_path, "TWSE_T86", "2026-05-18") == {
        "source": "TWSE_T86",
        "date": "2026-05-18",
        "status": "ok",
        "fetched_at": "2026-05-18T09:10:00+00:00",
        "row_count": 123,
        "error": "",
    }


def test_fetch_recent_rows_by_code_uses_latest_trading_rows(tmp_path):
    db_path = tmp_path / "flow.sqlite3"
    upsert_daily_rows(
        db_path,
        [
            {"date": "2026-05-16", "stock_no": "2330", "institutional_net": 1},
            {"date": "2026-05-17", "stock_no": "2330", "institutional_net": 2},
            {"date": "2026-05-18", "stock_no": "2330", "institutional_net": 3},
            {"date": "2026-05-18", "stock_no": "2454", "institutional_net": 9},
        ],
    )

    grouped = fetch_recent_rows_by_code(db_path, codes=["2330"], lookback_days=2)

    assert list(grouped) == ["2330"]
    assert [row["date"] for row in grouped["2330"]] == ["2026-05-17", "2026-05-18"]
    assert [row["institutional_net"] for row in grouped["2330"]] == [2, 3]