Taiwan ADR Premium Tracker
Pricing
from $0.05 / adr data fetched
Track real-time premium and discount between Taiwan ADRs (NYSE/NASDAQ) and their underlying TWSE shares with FX normalization.
Pricing
from $0.05 / adr data fetched
Rating
0.0
(0)
Developer
Steven C
Actor stats
0
Bookmarked
1
Total users
0
Monthly active users
8 days ago
Last modified
Categories
Share
FROM apify/actor-python:3.11COPY requirements.txt ./RUN pip install --no-cache-dir -r requirements.txtCOPY . ./CMD python -m src.main1[pytest]2asyncio_mode = auto3asyncio_default_fixture_loop_scope = function1apify>=2.0.02yfinance>=0.2.503pytest>=8.0.04pytest-asyncio>=0.24.0{ "actorSpecification": 1, "name": "taiwan-adr-premium-tracker", "title": "Taiwan ADR Premium Tracker", "description": "Tracks real-time price premiums and discounts between Taiwan ADRs (NYSE/NASDAQ) and their underlying Taiwan Stock Exchange shares. Covers TSMC (TSM), UMC, Chunghwa Telecom (CHT), ASE Technology (ASX), Himax (HIMX), and ChipMOS (IMOS) with USD/TWD FX normalization.", "version": "0.1", "buildTag": "latest", "readme": "../README.md", "input": "./input_schema.json", "dockerfile": "../Dockerfile", "storages": { "dataset": "./dataset_schema.json" }}{ "actorSpecification": 1, "fields": { "scraped_at": { "type": "string", "description": "UTC timestamp of when prices were fetched (ISO 8601 format)" }, "fx_rate_usd_twd": { "type": "number", "description": "USD/TWD exchange rate at time of fetch — TWD per 1 USD (e.g. 32.5 means 1 USD = 32.5 TWD)" }, "tickers_requested": { "type": "integer", "description": "Number of ADR tickers requested in input" }, "tickers_computed": { "type": "integer", "description": "Number of tickers for which a premium/discount was successfully computed" }, "results": { "type": "array", "description": "List of per-ticker results, each containing ADR price, TW price, equivalent USD price, and premium percentage" }, "errors": { "type": "array", "description": "List of tickers that could not be computed, with reasons (e.g. price unavailable, unknown ticker)" } }, "views": { "overview": { "title": "Results Overview", "transformation": { "fields": [ "scraped_at", "fx_rate_usd_twd", "tickers_requested", "tickers_computed" ] }, "display": { "component": "table", "properties": { "scraped_at": { "label": "Scraped At", "format": "text" }, "fx_rate_usd_twd": { "label": "USD/TWD Rate", "format": "number" }, "tickers_requested": { "label": "Tickers Requested", "format": "number" }, "tickers_computed": { "label": "Tickers Computed", "format": "number" } } } } }}{ "title": "Taiwan ADR Premium Tracker Input", "description": "Computes live ADR premium/discount for Taiwan stocks cross-listed on NYSE/NASDAQ and TWSE. Charged at a flat $0.05 per run regardless of how many tickers you track.", "type": "object", "schemaVersion": 1, "properties": { "tickers": { "title": "ADR tickers to track", "type": "array", "description": "List of Taiwan ADR ticker symbols to include. Supported: TSM, UMC, CHT, ASX, HIMX, IMOS. Leave blank to track all six.", "editor": "json", "default": ["TSM", "UMC", "CHT", "ASX", "HIMX", "IMOS"], "prefill": ["TSM", "UMC", "CHT", "ASX", "HIMX", "IMOS"] } }}1
1"""2ADR premium/discount calculator.3
4Pure functions — no network calls, no side effects.5All math happens here; fetcher.py owns the data retrieval.6
7ADR Premium formula:8 tw_equivalent_usd = (tw_price_twd * adr_ratio) / fx_usd_twd9 premium_pct = (adr_price_usd / tw_equivalent_usd - 1) * 10010
11A positive premium means the ADR trades ABOVE the TW equivalent in USD terms.12A negative value means the ADR trades at a DISCOUNT to TW shares.13
14ADS ratios: verified against SEC F-6 filings. Double-check before each major15version bump — issuers occasionally adjust ratios via tender offers.16"""17
18from __future__ import annotations19
20# ---------------------------------------------------------------------------21# ADR Registry22# ---------------------------------------------------------------------------23# Each entry: adr_ticker -> {tw_ticker, ratio, name}24# ratio = number of TW ordinary shares represented by ONE ADR/ADS unit25#26# NOTE TO MAINTAINER: Ratios below are based on public filings as of 2025.27# Verify against the issuer's most recent Form F-6 or 20-F if in doubt.28# ---------------------------------------------------------------------------29
30ADR_REGISTRY: dict[str, dict] = {31 "TSM": {32 "tw_ticker": "2330.TW",33 "ratio": 5,34 "name": "Taiwan Semiconductor Mfg (TSMC)",35 "exchange": "NYSE",36 },37 "UMC": {38 "tw_ticker": "2303.TW",39 "ratio": 5,40 "name": "United Microelectronics (UMC)",41 "exchange": "NYSE",42 },43 "CHT": {44 "tw_ticker": "2412.TW",45 "ratio": 10,46 "name": "Chunghwa Telecom",47 "exchange": "NYSE",48 },49 "ASX": {50 "tw_ticker": "3711.TW",51 "ratio": 2,52 "name": "ASE Technology Holding",53 "exchange": "NYSE",54 },55 "HIMX": {56 "tw_ticker": "3324.TW",57 "ratio": 2,58 "name": "Himax Technologies",59 "exchange": "NASDAQ",60 },61 "IMOS": {62 "tw_ticker": "8150.TW",63 "ratio": 20,64 "name": "ChipMOS Technologies",65 "exchange": "NASDAQ",66 },67}68
69DEFAULT_TICKERS = list(ADR_REGISTRY.keys())70
71
72# ---------------------------------------------------------------------------73# Core calculation74# ---------------------------------------------------------------------------75
76def compute_premium(77 adr_price_usd: float,78 tw_price_twd: float,79 adr_ratio: int,80 fx_usd_twd: float,81) -> dict:82 """83 Compute the ADR premium/discount for a single ticker pair.84
85 Args:86 adr_price_usd: Price of 1 ADR unit in USD (NYSE/NASDAQ).87 tw_price_twd: Price of 1 TW ordinary share in TWD (TWSE).88 adr_ratio: Number of TW shares represented by 1 ADR.89 fx_usd_twd: Exchange rate — TWD per 1 USD (e.g. 32.5).90
91 Returns dict with:92 tw_equivalent_usd: What the TW shares underlying 1 ADR cost in USD.93 premium_pct: ADR premium (+) or discount (-) vs TW equivalent.94 premium_label: Human-readable summary string.95 """96 if fx_usd_twd <= 0:97 raise ValueError(f"fx_usd_twd must be positive, got {fx_usd_twd}")98 if adr_ratio <= 0:99 raise ValueError(f"adr_ratio must be positive, got {adr_ratio}")100
101 tw_equivalent_usd = (tw_price_twd * adr_ratio) / fx_usd_twd102 premium_pct = round((adr_price_usd / tw_equivalent_usd - 1) * 100, 2)103
104 if premium_pct > 0:105 label = f"ADR at {premium_pct:.2f}% premium to TW shares"106 elif premium_pct < 0:107 label = f"ADR at {abs(premium_pct):.2f}% discount to TW shares"108 else:109 label = "ADR at parity with TW shares"110
111 return {112 "tw_equivalent_usd": round(tw_equivalent_usd, 4),113 "premium_pct": premium_pct,114 "premium_label": label,115 }116
117
118def build_result(119 adr_ticker: str,120 prices: dict[str, float | None],121 fx_usd_twd: float | None,122) -> dict:123 """124 Build a full result row for one ADR ticker.125
126 Args:127 adr_ticker: e.g. "TSM"128 prices: Map of ticker -> last price (None if unavailable).129 fx_usd_twd: USD/TWD FX rate (TWD per 1 USD). None if unavailable.130
131 Returns a dict with all output fields. Sets error field if calculation132 could not be completed.133 """134 entry = ADR_REGISTRY[adr_ticker]135 tw_ticker = entry["tw_ticker"]136 ratio = entry["ratio"]137
138 adr_price = prices.get(adr_ticker)139 tw_price = prices.get(tw_ticker)140
141 base = {142 "adr_ticker": adr_ticker,143 "tw_ticker": tw_ticker,144 "company_name": entry["name"],145 "exchange": entry["exchange"],146 "adr_ratio": ratio,147 "adr_price_usd": adr_price,148 "tw_price_twd": tw_price,149 "tw_equivalent_usd": None,150 "premium_pct": None,151 "premium_label": None,152 "error": None,153 }154
155 missing = []156 if adr_price is None:157 missing.append(f"{adr_ticker} price unavailable")158 if tw_price is None:159 missing.append(f"{tw_ticker} price unavailable")160 if fx_usd_twd is None:161 missing.append("FX rate unavailable")162
163 if missing:164 base["error"] = "; ".join(missing)165 return base166
167 calc = compute_premium(adr_price, tw_price, ratio, fx_usd_twd)168 base.update(calc)169 return base170
171
172def compute_all(173 tickers: list[str],174 prices: dict[str, float | None],175 fx_usd_twd: float | None,176) -> list[dict]:177 """178 Build result rows for all requested ADR tickers.179
180 Skips unknown tickers (not in ADR_REGISTRY) with a warning row.181 """182 results = []183 for ticker in tickers:184 if ticker not in ADR_REGISTRY:185 results.append({186 "adr_ticker": ticker,187 "error": f"Unknown ticker '{ticker}'. Supported: {', '.join(ADR_REGISTRY)}",188 })189 continue190 results.append(build_result(ticker, prices, fx_usd_twd))191 return results1"""2Price fetcher for Taiwan ADR Premium Tracker.3
4Uses yfinance to retrieve:5 - ADR prices (NYSE/NASDAQ) in USD6 - Underlying TW share prices (TWSE) in TWD7 - USD/TWD FX rate via Yahoo Finance FX pair USDTWD=X8
9yfinance is synchronous; all calls run in a thread executor so the rest10of the actor stays fully async.11"""12
13from __future__ import annotations14
15import asyncio16import logging17import math18from datetime import datetime, timezone19
20import yfinance as yf21
22logger = logging.getLogger(__name__)23
24FX_TICKER = "USDTWD=X" # Yahoo Finance FX pair: TWD per 1 USD25
26
27# ---------------------------------------------------------------------------28# Low-level sync helper (runs inside thread executor — easy to mock in tests)29# ---------------------------------------------------------------------------30
31def _get_last_price(ticker: str) -> float | None:32 """33 Synchronous: fetch the most recent last price for a Yahoo Finance ticker.34 Returns None on any error (delisted, network timeout, bad symbol, etc.).35 Also returns None for zero, NaN, or infinite prices — all indicate bad data.36 """37 try:38 info = yf.Ticker(ticker).fast_info39 price = info.last_price40 if price is None:41 return None42 price = float(price)43 # Reject NaN, +inf, -inf, and zero — none are valid market prices44 if not math.isfinite(price) or price <= 0:45 logger.warning("Unusable price for %s: %s — treating as unavailable", ticker, price)46 return None47 return round(price, 4)48 except Exception as exc:49 logger.warning("Failed to fetch price for %s: %s", ticker, exc)50 return None51
52
53# ---------------------------------------------------------------------------54# Async public API55# ---------------------------------------------------------------------------56
57async def fetch_price(ticker: str) -> float | None:58 """Async wrapper: fetch last price for a single Yahoo Finance ticker."""59 loop = asyncio.get_running_loop()60 return await loop.run_in_executor(None, _get_last_price, ticker)61
62
63async def fetch_all_prices(tickers: list[str]) -> dict[str, float | None]:64 """65 Concurrently fetch last prices for all given tickers plus the FX rate.66
67 Args:68 tickers: List of Yahoo Finance ticker symbols (ADR + TW tickers).69 The FX ticker (USDTWD=X) is added automatically.70
71 Returns:72 Dict mapping every requested ticker (and FX_TICKER) to its last price,73 or None if the price could not be retrieved.74 """75 all_tickers = list(dict.fromkeys(tickers + [FX_TICKER])) # deduplicate, preserve order76 logger.info("Fetching prices for: %s", ", ".join(all_tickers))77
78 tasks = [fetch_price(t) for t in all_tickers]79 # return_exceptions=True prevents one failed ticker from crashing all others.80 # Any task that raises returns an Exception object instead of a price.81 raw_results = await asyncio.gather(*tasks, return_exceptions=True)82
83 prices_list: list[float | None] = []84 for ticker, raw in zip(all_tickers, raw_results):85 if isinstance(raw, BaseException):86 logger.warning("Unexpected error fetching %s: %s", ticker, raw)87 prices_list.append(None)88 else:89 prices_list.append(raw)90
91 prices = dict(zip(all_tickers, prices_list))92
93 # Log summary94 fetched = sum(1 for v in prices.values() if v is not None)95 logger.info("Prices fetched: %d/%d", fetched, len(all_tickers))96 for ticker, price in prices.items():97 if price is None:98 logger.warning("No price for %s", ticker)99 else:100 logger.info(" %s = %s", ticker, price)101
102 return prices103
104
105def extract_fx_rate(prices: dict[str, float | None]) -> float | None:106 """Pull the USD/TWD FX rate out of the prices dict."""107 return prices.get(FX_TICKER)108
109
110def utc_now() -> str:111 """Return current UTC time as ISO 8601 string ending in Z."""112 return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")1"""2Taiwan ADR Premium Tracker — Apify Actor entry point.3
4Fetches live prices for Taiwan ADRs (NYSE/NASDAQ) and their underlying5TWSE shares, then computes the USD-normalized premium or discount.6
7PPE model: one flat charge of $0.05 per run via the "data-fetched" event.8The charge fires only after all data is successfully pushed — users are9never charged for runs that fail validation.10
11Input schema: ../.actor/input_schema.json12"""13
14import asyncio15import logging16
17from apify import Actor18
19from .calculator import ADR_REGISTRY, DEFAULT_TICKERS, compute_all20from .fetcher import extract_fx_rate, fetch_all_prices, utc_now21
22logging.basicConfig(level=logging.INFO)23logger = logging.getLogger(__name__)24
25
26async def main() -> None:27 async with Actor:28 raw_input = await Actor.get_input() or {}29
30 # --- Input: which ADR tickers to track ---31 requested: list = raw_input.get("tickers", DEFAULT_TICKERS)32
33 # Validate: must be a non-empty list of strings34 if not isinstance(requested, list) or len(requested) == 0:35 await Actor.fail(36 status_message="'tickers' must be a non-empty list of ADR symbols. "37 f"Supported: {', '.join(ADR_REGISTRY)}"38 )39 return40
41 # Normalise to uppercase and deduplicate, preserving order42 tickers = list(dict.fromkeys(t.upper().strip() for t in requested if isinstance(t, str)))43 if not tickers:44 await Actor.fail(status_message="No valid ticker symbols provided.")45 return46
47 # Warn about unknown tickers (don't fail — compute_all handles them gracefully)48 unknown = [t for t in tickers if t not in ADR_REGISTRY]49 if unknown:50 Actor.log.warning(51 "Unknown ticker(s) will be skipped: %s. Supported: %s",52 ", ".join(unknown),53 ", ".join(ADR_REGISTRY),54 )55
56 Actor.log.info("Tracking ADRs: %s", ", ".join(tickers))57
58 # --- Collect all Yahoo Finance tickers needed ---59 adr_tickers = [t for t in tickers if t in ADR_REGISTRY]60 tw_tickers = [ADR_REGISTRY[t]["tw_ticker"] for t in adr_tickers]61 all_yf_tickers = adr_tickers + tw_tickers # FX ticker added inside fetch_all_prices62
63 # --- Fetch prices concurrently ---64 prices = await fetch_all_prices(all_yf_tickers)65 fx_usd_twd = extract_fx_rate(prices)66
67 if fx_usd_twd is None:68 Actor.log.warning("Could not fetch USD/TWD FX rate — premium calculations will be skipped.")69
70 # --- Compute premiums ---71 results = compute_all(tickers, prices, fx_usd_twd)72
73 successful = sum(1 for r in results if r.get("error") is None)74 errors = [r for r in results if r.get("error") is not None]75
76 Actor.log.info(77 "Computed %d/%d premiums successfully.",78 successful,79 len(results),80 )81
82 # --- Build output ---83 output = {84 "scraped_at": utc_now(),85 "fx_rate_usd_twd": fx_usd_twd,86 "tickers_requested": len(tickers),87 "tickers_computed": successful,88 "results": results,89 "errors": [{"ticker": e["adr_ticker"], "reason": e["error"]} for e in errors],90 }91
92 # Push data and charge the flat $0.05 PPE event — but ONLY if at least93 # one premium was successfully computed. If every ticker failed (e.g.94 # complete yfinance outage), push the diagnostic output for free so the95 # user is not charged for a run that produced no useful data.96 if successful > 0:97 await Actor.push_data(output, "data-fetched")98 else:99 Actor.log.warning("No premiums computed — pushing output without PPE charge.")100 await Actor.push_data(output)101
102 Actor.log.info(103 "Done. FX rate: %s TWD/USD | Top result: %s premium = %s%%",104 fx_usd_twd,105 results[0].get("adr_ticker") if results else "N/A",106 results[0].get("premium_pct") if results else "N/A",107 )108
109
110if __name__ == "__main__":111 asyncio.run(main())1
1"""2Unit tests for calculator.py — pure math, no network calls.3"""4
5import pytest6from src.calculator import (7 ADR_REGISTRY,8 DEFAULT_TICKERS,9 compute_premium,10 build_result,11 compute_all,12)13
14
15# ---------------------------------------------------------------------------16# Fixtures17# ---------------------------------------------------------------------------18
19# Realistic snapshot: TSMC ADR at ~8.6% premium (common in 2025)20PRICES_FULL = {21 "TSM": 175.50, # USD per ADR22 "2330.TW": 1050.0, # TWD per share23 "UMC": 7.20,24 "2303.TW": 52.0,25 "CHT": 38.50,26 "2412.TW": 130.0,27 "ASX": 9.80,28 "3711.TW": 310.0,29 "HIMX": 6.40,30 "3324.TW": 106.0,31 "IMOS": 24.00,32 "8150.TW": 40.0,33}34
35FX = 32.5 # TWD per 1 USD36
37
38# ---------------------------------------------------------------------------39# Tests: compute_premium40# ---------------------------------------------------------------------------41
42class TestComputePremium:43 def test_tsm_premium_positive(self):44 # tw_equivalent = (1050 * 5) / 32.5 = 161.538...45 # premium = (175.50 / 161.538 - 1) * 100 = 8.64%46 result = compute_premium(47 adr_price_usd=175.50,48 tw_price_twd=1050.0,49 adr_ratio=5,50 fx_usd_twd=32.5,51 )52 assert result["premium_pct"] == pytest.approx(8.64, abs=0.01)53 assert result["tw_equivalent_usd"] == pytest.approx(161.538, abs=0.01)54 assert "premium" in result["premium_label"]55 assert "8.64" in result["premium_label"]56
57 def test_discount_when_adr_cheaper(self):58 # ADR below TW equivalent → negative premium59 result = compute_premium(60 adr_price_usd=150.0,61 tw_price_twd=1050.0,62 adr_ratio=5,63 fx_usd_twd=32.5,64 )65 assert result["premium_pct"] < 066 assert "discount" in result["premium_label"]67
68 def test_parity(self):69 # ADR exactly equals TW equivalent70 # tw_equivalent = (100 * 2) / 32.5 = 6.1538...71 # adr_price set to exact equivalent72 tw_equiv = (100.0 * 2) / 32.573 result = compute_premium(74 adr_price_usd=tw_equiv,75 tw_price_twd=100.0,76 adr_ratio=2,77 fx_usd_twd=32.5,78 )79 assert result["premium_pct"] == 0.080 assert "parity" in result["premium_label"]81
82 def test_premium_rounds_to_two_decimals(self):83 result = compute_premium(175.50, 1050.0, 5, 32.5)84 # Result should be a float with at most 2 decimal places85 assert result["premium_pct"] == round(result["premium_pct"], 2)86
87 def test_raises_on_zero_fx(self):88 with pytest.raises(ValueError, match="fx_usd_twd must be positive"):89 compute_premium(175.50, 1050.0, 5, fx_usd_twd=0)90
91 def test_raises_on_negative_fx(self):92 with pytest.raises(ValueError, match="fx_usd_twd must be positive"):93 compute_premium(175.50, 1050.0, 5, fx_usd_twd=-1.0)94
95 def test_raises_on_zero_ratio(self):96 with pytest.raises(ValueError, match="adr_ratio must be positive"):97 compute_premium(175.50, 1050.0, adr_ratio=0, fx_usd_twd=32.5)98
99 def test_tw_equivalent_usd_formula(self):100 # Explicit formula check: (tw_price * ratio) / fx101 result = compute_premium(200.0, 1000.0, 5, 32.0)102 expected_equiv = (1000.0 * 5) / 32.0103 assert result["tw_equivalent_usd"] == pytest.approx(expected_equiv, rel=1e-4)104
105 def test_large_premium(self):106 # ADR at 25% premium — historically observed for TSMC in 2025107 result = compute_premium(200.0, 1050.0, 5, 32.5)108 assert result["premium_pct"] > 20109
110 def test_zero_tw_price_is_guarded_upstream(self):111 # _get_last_price returns None for zero prices, so build_result receives112 # None and sets an error rather than calling compute_premium with zero.113 # This test confirms compute_premium itself raises if called directly114 # with zero tw_price (defence-in-depth check).115 # tw_equivalent_usd = (0 * 5) / 32.5 = 0 → division by zero in premium_pct116 with pytest.raises((ZeroDivisionError, ValueError)):117 compute_premium(175.50, 0.0, 5, 32.5)118
119 def test_high_ratio_cht(self):120 # CHT: ratio=10, check math is correct121 result = compute_premium(122 adr_price_usd=38.50,123 tw_price_twd=130.0,124 adr_ratio=10,125 fx_usd_twd=32.5,126 )127 expected_equiv = (130.0 * 10) / 32.5 # = 40.0128 assert result["tw_equivalent_usd"] == pytest.approx(expected_equiv, rel=1e-4)129 # 38.50 / 40.0 - 1 = -3.75%130 assert result["premium_pct"] == pytest.approx(-3.75, abs=0.01)131
132
133# ---------------------------------------------------------------------------134# Tests: build_result135# ---------------------------------------------------------------------------136
137class TestBuildResult:138 def test_tsm_successful_build(self):139 result = build_result("TSM", PRICES_FULL, FX)140 assert result["adr_ticker"] == "TSM"141 assert result["tw_ticker"] == "2330.TW"142 assert result["adr_ratio"] == 5143 assert result["adr_price_usd"] == 175.50144 assert result["tw_price_twd"] == 1050.0145 assert result["premium_pct"] is not None146 assert result["error"] is None147
148 def test_all_required_keys_present(self):149 result = build_result("TSM", PRICES_FULL, FX)150 for key in (151 "adr_ticker", "tw_ticker", "company_name", "exchange", "adr_ratio",152 "adr_price_usd", "tw_price_twd", "tw_equivalent_usd",153 "premium_pct", "premium_label", "error",154 ):155 assert key in result, f"Missing key: {key}"156
157 def test_missing_adr_price_sets_error(self):158 prices = {**PRICES_FULL, "TSM": None}159 result = build_result("TSM", prices, FX)160 assert result["error"] is not None161 assert "TSM" in result["error"]162 assert result["premium_pct"] is None163
164 def test_missing_tw_price_sets_error(self):165 prices = {**PRICES_FULL, "2330.TW": None}166 result = build_result("TSM", prices, FX)167 assert result["error"] is not None168 assert result["premium_pct"] is None169
170 def test_missing_fx_sets_error(self):171 result = build_result("TSM", PRICES_FULL, fx_usd_twd=None)172 assert result["error"] is not None173 assert "FX" in result["error"]174 assert result["premium_pct"] is None175
176 def test_all_prices_missing_error_mentions_all(self):177 prices = {**PRICES_FULL, "TSM": None, "2330.TW": None}178 result = build_result("TSM", prices, fx_usd_twd=None)179 # Error should mention all three missing items180 assert "TSM" in result["error"]181 assert "2330.TW" in result["error"]182 assert "FX" in result["error"]183
184 def test_company_name_populated(self):185 result = build_result("TSM", PRICES_FULL, FX)186 assert "TSMC" in result["company_name"] or "Taiwan Semiconductor" in result["company_name"]187
188 def test_exchange_field(self):189 tsm = build_result("TSM", PRICES_FULL, FX)190 assert tsm["exchange"] == "NYSE"191 himx = build_result("HIMX", PRICES_FULL, FX)192 assert himx["exchange"] == "NASDAQ"193
194
195# ---------------------------------------------------------------------------196# Tests: compute_all197# ---------------------------------------------------------------------------198
199class TestComputeAll:200 def test_returns_one_row_per_ticker(self):201 results = compute_all(["TSM", "UMC"], PRICES_FULL, FX)202 assert len(results) == 2203
204 def test_results_ordered_as_requested(self):205 results = compute_all(["UMC", "TSM"], PRICES_FULL, FX)206 assert results[0]["adr_ticker"] == "UMC"207 assert results[1]["adr_ticker"] == "TSM"208
209 def test_unknown_ticker_gets_error_row(self):210 results = compute_all(["TSM", "FAKE"], PRICES_FULL, FX)211 fake_row = next(r for r in results if r["adr_ticker"] == "FAKE")212 assert fake_row["error"] is not None213 assert "Unknown ticker" in fake_row["error"]214
215 def test_all_default_tickers_computable(self):216 results = compute_all(DEFAULT_TICKERS, PRICES_FULL, FX)217 assert len(results) == len(DEFAULT_TICKERS)218 for r in results:219 assert r["error"] is None, f"{r['adr_ticker']} had error: {r['error']}"220
221 def test_empty_ticker_list_returns_empty(self):222 results = compute_all([], PRICES_FULL, FX)223 assert results == []224
225 def test_partial_price_failure_doesnt_block_others(self):226 prices = {**PRICES_FULL, "TSM": None} # TSM fails227 results = compute_all(["TSM", "UMC"], prices, FX)228 tsm = next(r for r in results if r["adr_ticker"] == "TSM")229 umc = next(r for r in results if r["adr_ticker"] == "UMC")230 assert tsm["error"] is not None231 assert umc["error"] is None232 assert umc["premium_pct"] is not None233
234
235# ---------------------------------------------------------------------------236# Tests: ADR Registry integrity237# ---------------------------------------------------------------------------238
239class TestADRRegistry:240 def test_all_registry_entries_have_required_keys(self):241 for ticker, entry in ADR_REGISTRY.items():242 for key in ("tw_ticker", "ratio", "name", "exchange"):243 assert key in entry, f"{ticker} missing '{key}'"244
245 def test_all_ratios_are_positive_integers(self):246 for ticker, entry in ADR_REGISTRY.items():247 assert isinstance(entry["ratio"], int), f"{ticker} ratio is not int"248 assert entry["ratio"] > 0, f"{ticker} ratio must be positive"249
250 def test_tw_tickers_end_with_dot_tw(self):251 for ticker, entry in ADR_REGISTRY.items():252 assert entry["tw_ticker"].endswith(".TW"), (253 f"{ticker}: tw_ticker '{entry['tw_ticker']}' must end in .TW"254 )255
256 def test_exchanges_are_valid(self):257 valid = {"NYSE", "NASDAQ"}258 for ticker, entry in ADR_REGISTRY.items():259 assert entry["exchange"] in valid, (260 f"{ticker}: exchange '{entry['exchange']}' not in {valid}"261 )262
263 def test_default_tickers_all_in_registry(self):264 for t in DEFAULT_TICKERS:265 assert t in ADR_REGISTRY, f"DEFAULT_TICKERS contains unknown ticker: {t}"266
267 def test_tsm_ratio_is_five(self):268 # Source: TSMC SEC Form F-3ASR (CIK 1046179) + TSMC Investor FAQ269 assert ADR_REGISTRY["TSM"]["ratio"] == 5270
271 def test_umc_ratio_is_five(self):272 # Source: UMC SEC Form 20-F (CIK 1033767) + F-6/POS (2022)273 # "each ADS represents five common shares" — JPMorgan Chase depositary274 assert ADR_REGISTRY["UMC"]["ratio"] == 5275
276 def test_cht_ratio_is_ten(self):277 # Source: Chunghwa Telecom Investor FAQ + SEC 20-F (CIK 1132924)278 # "Ten common shares are equal to one ADR share"279 assert ADR_REGISTRY["CHT"]["ratio"] == 10280
281 def test_asx_ratio_is_two(self):282 # Source: Nasdaq official ASX listing page security description283 # "each representing Two Common Shares"284 assert ADR_REGISTRY["ASX"]["ratio"] == 2285
286 def test_himx_ratio_is_two(self):287 # Source: SEC Form 424(B)(1) IPO prospectus (CIK 1342338, 2006)288 assert ADR_REGISTRY["HIMX"]["ratio"] == 2289
290 def test_imos_ratio_is_twenty(self):291 # Source: ChipMOS SEC 20-F (CIK 1123134) + capital reduction filings292 # "20 common shares per ADS" — Citibank N.A. depositary293 assert ADR_REGISTRY["IMOS"]["ratio"] == 20294
295 def test_tsm_tw_ticker(self):296 assert ADR_REGISTRY["TSM"]["tw_ticker"] == "2330.TW"1"""2Tests for fetcher.py and PPE charge safety.3
4fetcher.py makes real network calls via yfinance. All tests mock5_get_last_price (the synchronous yfinance wrapper) so no network is needed.6"""7
8import pytest9from unittest.mock import patch, MagicMock10
11from src.fetcher import (12 FX_TICKER,13 _get_last_price,14 fetch_price,15 fetch_all_prices,16 extract_fx_rate,17 utc_now,18)19
20
21# ---------------------------------------------------------------------------22# Tests: _get_last_price (synchronous, can test directly)23# ---------------------------------------------------------------------------24
25class TestGetLastPriceSync:26 def test_returns_float_on_success(self):27 mock_ticker = MagicMock()28 mock_ticker.fast_info.last_price = 175.5029 with patch("src.fetcher.yf.Ticker", return_value=mock_ticker):30 result = _get_last_price("TSM")31 assert result == 175.5032
33 def test_returns_none_on_exception(self):34 with patch("src.fetcher.yf.Ticker", side_effect=Exception("network error")):35 result = _get_last_price("TSM")36 assert result is None37
38 def test_returns_none_when_price_is_none(self):39 mock_ticker = MagicMock()40 mock_ticker.fast_info.last_price = None41 with patch("src.fetcher.yf.Ticker", return_value=mock_ticker):42 result = _get_last_price("TSM")43 assert result is None44
45 def test_rounds_to_four_decimal_places(self):46 mock_ticker = MagicMock()47 mock_ticker.fast_info.last_price = 175.12345678948 with patch("src.fetcher.yf.Ticker", return_value=mock_ticker):49 result = _get_last_price("TSM")50 assert result == round(175.123456789, 4)51
52 def test_handles_attribute_error_gracefully(self):53 mock_ticker = MagicMock()54 del mock_ticker.fast_info # simulate missing attribute55 with patch("src.fetcher.yf.Ticker", return_value=mock_ticker):56 result = _get_last_price("INVALID")57 assert result is None58
59 def test_returns_none_for_zero_price(self):60 # Zero is not a valid market price — bad data from Yahoo Finance61 mock_ticker = MagicMock()62 mock_ticker.fast_info.last_price = 0.063 with patch("src.fetcher.yf.Ticker", return_value=mock_ticker):64 result = _get_last_price("TSM")65 assert result is None66
67 def test_returns_none_for_nan_price(self):68 mock_ticker = MagicMock()69 mock_ticker.fast_info.last_price = float("nan")70 with patch("src.fetcher.yf.Ticker", return_value=mock_ticker):71 result = _get_last_price("TSM")72 assert result is None73
74 def test_returns_none_for_inf_price(self):75 mock_ticker = MagicMock()76 mock_ticker.fast_info.last_price = float("inf")77 with patch("src.fetcher.yf.Ticker", return_value=mock_ticker):78 result = _get_last_price("TSM")79 assert result is None80
81 def test_returns_none_for_negative_inf_price(self):82 mock_ticker = MagicMock()83 mock_ticker.fast_info.last_price = float("-inf")84 with patch("src.fetcher.yf.Ticker", return_value=mock_ticker):85 result = _get_last_price("TSM")86 assert result is None87
88
89# ---------------------------------------------------------------------------90# Tests: fetch_price (async wrapper)91# ---------------------------------------------------------------------------92
93@pytest.mark.asyncio94class TestFetchPrice:95 async def test_returns_correct_price(self):96 with patch("src.fetcher._get_last_price", return_value=175.50):97 result = await fetch_price("TSM")98 assert result == 175.5099
100 async def test_returns_none_on_failure(self):101 with patch("src.fetcher._get_last_price", return_value=None):102 result = await fetch_price("TSM")103 assert result is None104
105 async def test_fx_ticker_fetchable(self):106 with patch("src.fetcher._get_last_price", return_value=32.45):107 result = await fetch_price(FX_TICKER)108 assert result == 32.45109
110
111# ---------------------------------------------------------------------------112# Tests: fetch_all_prices113# ---------------------------------------------------------------------------114
115@pytest.mark.asyncio116class TestFetchAllPrices:117 async def test_returns_all_tickers_including_fx(self):118 def side_effect(ticker):119 return {120 "TSM": 175.50,121 "2330.TW": 1050.0,122 "USDTWD=X": 32.5,123 }.get(ticker)124
125 with patch("src.fetcher._get_last_price", side_effect=side_effect):126 prices = await fetch_all_prices(["TSM", "2330.TW"])127
128 assert "TSM" in prices129 assert "2330.TW" in prices130 assert FX_TICKER in prices # FX ticker always added automatically131
132 async def test_fx_ticker_always_included(self):133 with patch("src.fetcher._get_last_price", return_value=100.0):134 prices = await fetch_all_prices(["TSM"])135 assert FX_TICKER in prices136
137 async def test_no_duplicate_fx_ticker(self):138 with patch("src.fetcher._get_last_price", return_value=100.0):139 # Even if FX_TICKER already in input, it should not appear twice140 prices = await fetch_all_prices(["TSM", FX_TICKER])141 keys = list(prices.keys())142 assert keys.count(FX_TICKER) == 1143
144 async def test_failed_ticker_returns_none_not_crash(self):145 # Patch yf.Ticker so the exception flows through _get_last_price's146 # real try/except, which catches it and returns None.147 def ticker_side_effect(symbol):148 if symbol == "TSM":149 raise Exception("delisted")150 mock = MagicMock()151 mock.fast_info.last_price = 100.0152 return mock153
154 with patch("src.fetcher.yf.Ticker", side_effect=ticker_side_effect):155 prices = await fetch_all_prices(["TSM", "2330.TW"])156
157 # TSM failed but the whole call didn't crash158 assert prices["TSM"] is None159 assert prices["2330.TW"] == 100.0160
161 async def test_all_none_when_yfinance_down(self):162 with patch("src.fetcher._get_last_price", return_value=None):163 prices = await fetch_all_prices(["TSM", "2330.TW"])164 assert all(v is None for v in prices.values())165
166
167# ---------------------------------------------------------------------------168# Tests: extract_fx_rate169# ---------------------------------------------------------------------------170
171class TestExtractFxRate:172 def test_extracts_fx_from_prices_dict(self):173 prices = {"TSM": 175.50, FX_TICKER: 32.5}174 assert extract_fx_rate(prices) == 32.5175
176 def test_returns_none_when_fx_missing(self):177 prices = {"TSM": 175.50}178 assert extract_fx_rate(prices) is None179
180 def test_returns_none_when_fx_is_none(self):181 prices = {"TSM": 175.50, FX_TICKER: None}182 assert extract_fx_rate(prices) is None183
184
185# ---------------------------------------------------------------------------186# Tests: utc_now187# ---------------------------------------------------------------------------188
189class TestUtcNow:190 def test_ends_with_z(self):191 ts = utc_now()192 assert ts.endswith("Z")193
194 def test_iso_format(self):195 ts = utc_now()196 # Should be parseable as ISO 8601: 2026-03-27T12:34:56Z197 assert len(ts) == 20198 assert ts[4] == "-"199 assert ts[7] == "-"200 assert ts[10] == "T"201 assert ts[13] == ":"202 assert ts[16] == ":"203
204
205# ---------------------------------------------------------------------------206# Tests: PPE charge safety — inspect main.py source directly207# ---------------------------------------------------------------------------208
209class TestPPEChargeSafety:210 """211 Verify monetisation logic in main.py is correct.212 Inspects source code directly — no need to invoke Actor.213 """214
215 def _load_main_source(self) -> str:216 with open("src/main.py") as f:217 return f.read()218
219 def test_exactly_one_ppe_charge_call(self):220 source = self._load_main_source()221 charge_calls = [222 line.strip()223 for line in source.splitlines()224 if "push_data" in line and "data-fetched" in line225 ]226 assert len(charge_calls) == 1, (227 f"Expected exactly 1 PPE charge call, found {len(charge_calls)}: {charge_calls}"228 )229
230 def test_charge_call_pushes_output_not_metadata(self):231 source = self._load_main_source()232 charge_line = next(233 line.strip()234 for line in source.splitlines()235 if "push_data" in line and "data-fetched" in line236 )237 assert "output" in charge_line, (238 f"Expected charge to push 'output', got: {charge_line}"239 )240
241 def test_event_name_is_data_fetched(self):242 source = self._load_main_source()243 assert '"data-fetched"' in source, (244 "PPE event name must be 'data-fetched' (double-quoted string)"245 )246
247 def test_actor_fail_called_on_invalid_input(self):248 source = self._load_main_source()249 assert "Actor.fail" in source, "main.py must call Actor.fail() for invalid inputs"250
251 def test_tickers_normalised_to_uppercase(self):252 source = self._load_main_source()253 assert ".upper()" in source, "Tickers must be normalised to uppercase"254
255 def test_no_charge_when_zero_results_computed(self):256 # When successful == 0, push_data must be called WITHOUT the charge event.257 # Verify by checking the conditional guard exists in source.258 source = self._load_main_source()259 assert "successful > 0" in source, (260 "main.py must guard PPE charge behind 'if successful > 0'"261 )262
263 def test_free_push_exists_for_all_errors_case(self):264 # There must be a push_data(output) call (no charge event) as the else branch.265 source = self._load_main_source()266 lines = source.splitlines()267 free_push_lines = [268 l.strip() for l in lines269 if "push_data(output)" in l and "data-fetched" not in l270 ]271 assert len(free_push_lines) >= 1, (272 "main.py must have a push_data(output) call without PPE charge for the zero-results case"273 )274
275 def test_else_branch_has_no_charge_event(self):276 # The else branch (zero results) must NEVER contain "data-fetched".277 # Parse the else block and confirm it is charge-free.278 source = self._load_main_source()279 lines = source.splitlines()280 in_else = False281 for line in lines:282 stripped = line.strip()283 if stripped == "else:":284 in_else = True285 continue286 if in_else:287 # else block ends when indentation returns to same or lower level288 if stripped and not line.startswith(" "):289 break290 assert "data-fetched" not in stripped, (291 f"'data-fetched' found inside else branch: {stripped}"292 )293
294 def test_two_actor_fail_calls_exist(self):295 # main.py must have TWO input validation guards:296 # 1. for non-list or empty list input297 # 2. for list of non-strings (all filtered out → empty tickers)298 source = self._load_main_source()299 fail_calls = [l.strip() for l in source.splitlines() if "Actor.fail" in l]300 assert len(fail_calls) >= 2, (301 f"Expected at least 2 Actor.fail() calls (one per validation layer), "302 f"found {len(fail_calls)}: {fail_calls}"303 )304
305 def test_deduplication_prevents_double_fetch(self):306 # dict.fromkeys is the deduplication mechanism used on the tickers list.307 # Without it, ["TSM","TSM"] would fetch TSM twice and produce 2 result rows,308 # which could confuse users (though PPE still fires only once).309 source = self._load_main_source()310 assert "dict.fromkeys" in source, (311 "main.py must deduplicate tickers with dict.fromkeys to prevent double-fetching"312 )313
314 def test_push_data_total_call_count(self):315 # There must be exactly 2 push_data calls in total:316 # one with the charge event (happy path) and one without (all-errors path).317 # More than 2 would indicate an unexpected extra charge somewhere.318 source = self._load_main_source()319 push_calls = [l.strip() for l in source.splitlines() if "push_data" in l]320 assert len(push_calls) == 2, (321 f"Expected exactly 2 push_data calls in main.py, found {len(push_calls)}: {push_calls}"322 )