Spaces:
Sleeping
Sleeping
File size: 4,111 Bytes
ed25d75 | 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 | """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)
@dataclass(frozen=True)
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
|