Spaces:
Sleeping
Sleeping
| """Daily BBCA prices from Yahoo Finance, with a short cache. | |
| Yahoo returns today's row while the market is still open. That row is a | |
| running price, not a close, so it is kept apart from the completed sessions. | |
| """ | |
| from __future__ import annotations | |
| import threading | |
| import time | |
| from dataclasses import dataclass | |
| from datetime import time as dtime | |
| import numpy as np | |
| import pandas as pd | |
| TICKER = "BBCA.JK" | |
| TZ = "Asia/Jakarta" | |
| HISTORY_PERIOD = "2y" | |
| CACHE_SECONDS = 60 | |
| # IDX closes at 16:00 WIB. Yahoo can lag a few minutes, so wait a bit longer | |
| # before treating today's row as a finished session. | |
| SESSION_DONE_AT = dtime(16, 15) | |
| class PriceData: | |
| closes: pd.Series # completed sessions, tz-naive dates | |
| live_price: float | None # today's running price, if the market is open | |
| live_date: pd.Timestamp | None | |
| fetched_at: pd.Timestamp # Asia/Jakarta | |
| source: str # "yahoo" or "demo" | |
| class DataError(RuntimeError): | |
| """Raised when prices can't be loaded. The message is safe to show.""" | |
| _lock = threading.Lock() | |
| _cache: dict = {"at": 0.0, "key": None, "data": None} | |
| def _now() -> pd.Timestamp: | |
| return pd.Timestamp.now(tz=TZ) | |
| def _download() -> pd.DataFrame: | |
| import yfinance as yf | |
| last_exc: Exception | None = None | |
| for attempt in range(2): | |
| try: | |
| frame = yf.Ticker(TICKER).history( | |
| period=HISTORY_PERIOD, interval="1d", auto_adjust=True, timeout=15 | |
| ) | |
| if frame is not None and not frame.empty and "Close" in frame: | |
| return frame | |
| except Exception as exc: # network, rate limit, parsing | |
| last_exc = exc | |
| if attempt == 0: | |
| time.sleep(2) | |
| raise DataError("Couldn't get prices from Yahoo Finance. Try again in a minute.") from last_exc | |
| def _demo_frame(now: pd.Timestamp) -> pd.DataFrame: | |
| """Synthetic random walk for offline testing. Never shown as real data.""" | |
| days = pd.bdate_range(end=now.normalize(), periods=500, tz=TZ) | |
| rng = np.random.default_rng(52) | |
| steps = rng.normal(0.0002, 0.012, len(days)) | |
| close = 8000 * np.exp(np.cumsum(steps)) | |
| close = np.round(close / 25) * 25 # IDX tick size for this price band | |
| return pd.DataFrame({"Close": close}, index=days) | |
| def split_sessions( | |
| frame: pd.DataFrame, now: pd.Timestamp | |
| ) -> tuple[pd.Series, float | None, pd.Timestamp | None]: | |
| """Return (completed closes, live price, live date).""" | |
| close = frame["Close"].astype(float).dropna() | |
| idx = close.index | |
| if idx.tz is None: | |
| idx = idx.tz_localize(TZ) | |
| else: | |
| idx = idx.tz_convert(TZ) | |
| # Keep the Jakarta calendar date. Converting to UTC first shifts every | |
| # session back one day. | |
| close.index = pd.DatetimeIndex(idx.date) | |
| close = close[~close.index.duplicated(keep="last")].sort_index() | |
| live_price = live_date = None | |
| today = pd.Timestamp(now.date()) | |
| if len(close) and close.index[-1] == today and now.time() < SESSION_DONE_AT: | |
| live_price = float(close.iloc[-1]) | |
| live_date = today | |
| close = close.iloc[:-1] | |
| return close, live_price, live_date | |
| def load_prices(demo: bool = False) -> PriceData: | |
| now = _now() | |
| key = "demo" if demo else "yahoo" | |
| with _lock: | |
| cached = _cache["data"] | |
| if cached is not None and _cache["key"] == key and time.monotonic() - _cache["at"] < CACHE_SECONDS: | |
| return cached | |
| try: | |
| frame = _demo_frame(now) if demo else _download() | |
| except DataError: | |
| # Yahoo sometimes rate limits shared hosts. Serve the last good | |
| # data instead of an error; the page shows when it was fetched. | |
| if cached is not None and _cache["key"] == key: | |
| return cached | |
| raise | |
| closes, live_price, live_date = split_sessions(frame, now) | |
| if len(closes) < 60: | |
| raise DataError("Not enough price history to fit the model.") | |
| data = PriceData(closes, live_price, live_date, now, key) | |
| _cache.update(at=time.monotonic(), key=key, data=data) | |
| return data | |