Spaces:
Running
Running
File size: 5,263 Bytes
e610a2f ee37d63 e610a2f ee37d63 e610a2f ee37d63 e610a2f | 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | 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]
|