""" Persistent local data store for price history. Layout: data/{market}.csv — adjusted close prices data/{market}_open.csv — adjusted open prices data/{market}_unavailable.json — tickers Yahoo returned no data for (retried weekly) Panels are stored raw: a date/ticker Yahoo has no bar for stays NaN, so a missing print is distinguishable from an unchanged price. ``load()`` and ``update()`` forward-fill on read; ``load_raw()`` exposes the stored values. Every ``update()`` re-downloads a trailing ``REFRESH_DAYS`` window and overwrites it for the requested tickers. Partial intraday bars, late prints and rows that turn out not to be trading days are therefore corrected on the next run instead of being frozen. When Yahoo's split/dividend adjustment shifts the overlap, that ticker's older history is rescaled so returns stay continuous. """ import fcntl import json import os from contextlib import contextmanager from datetime import datetime, timedelta import numpy as np import pandas as pd import yfinance as yf DATA_DIR = "data" # Trailing calendar days re-downloaded and overwritten on every update. REFRESH_DAYS = 30 # A row is kept only if at least this fraction of tickers alive on that date # have a bar. Drops e.g. US-holiday rows created by HK-listed tickers. MIN_ROW_COVERAGE = 0.5 # Relative change in the adjusted-price overlap that triggers a rescale. RESCALE_TOLERANCE = 1e-3 UNAVAILABLE_RETRY_DAYS = 7 def _data_path(market: str, price_type: str = "close") -> str: suffix = "" if price_type == "close" else f"_{price_type}" return os.path.join(DATA_DIR, f"{market}{suffix}.csv") def load_raw(market: str, price_type: str = "close") -> pd.DataFrame | None: """Load the stored panel without filling missing bars, or None if absent.""" path = _data_path(market, price_type) if not os.path.exists(path): return None return pd.read_csv(path, index_col=0, parse_dates=True) def load(market: str, price_type: str = "close") -> pd.DataFrame | None: """Load the stored panel with missing bars forward-filled.""" raw = load_raw(market, price_type) return None if raw is None else raw.ffill() def _write_atomic(df: pd.DataFrame, path: str) -> None: tmp = f"{path}.tmp" df.to_csv(tmp) os.replace(tmp, path) @contextmanager def _panel_lock(market: str): """Serialize updates of one market panel across processes (monitor, bridge, scripts).""" os.makedirs(DATA_DIR, exist_ok=True) with open(os.path.join(DATA_DIR, f".{market}.lock"), "w") as fh: fcntl.flock(fh, fcntl.LOCK_EX) try: yield finally: fcntl.flock(fh, fcntl.LOCK_UN) def _download(tickers: list[str], start: str, end: str | None = None, fields: list[str] | None = None) -> dict[str, pd.DataFrame]: """ Download price data from Yahoo Finance. Parameters ---------- fields : list of str, optional Price fields to extract, e.g. ["Close", "Open"]. Defaults to ["Close"]. Returns ------- dict mapping field name to DataFrame (columns = tickers). """ if fields is None: fields = ["Close"] print(f"--- Downloading {len(tickers)} tickers from {start}{f' to {end}' if end else ''} ---") kwargs = dict(auto_adjust=True) if end: kwargs["end"] = end raw = yf.download(tickers, start=start, **kwargs) if raw.empty: return {f: pd.DataFrame() for f in fields} result = {} for field in fields: if field in raw.columns.get_level_values(0) if isinstance(raw.columns, pd.MultiIndex) else field in raw.columns: selected = raw[field] if isinstance(selected, pd.Series): result[field] = selected.to_frame(name=tickers[0]) else: result[field] = selected else: result[field] = pd.DataFrame() return result def _clean_market_data(data: pd.DataFrame, field: str) -> pd.DataFrame: """Clean market data while preserving volume gaps.""" good = data.columns[data.notna().mean() > 0.5] dropped = set(data.columns) - set(good) if dropped: print(f"--- Dropped {len(dropped)} tickers with >50% missing data ---") data = data[good] if field == "volume": return data return data.ffill().dropna(how="all") def _merge_market_panel(existing: pd.DataFrame | None, new_data: pd.DataFrame) -> pd.DataFrame: """Merge new data into an existing cached panel, preserving old columns and dates.""" if existing is None or existing.empty: merged = new_data.copy() elif new_data.empty: merged = existing.copy() else: merged = existing.combine_first(new_data) merged.loc[new_data.index, new_data.columns] = new_data merged = merged.sort_index() merged = merged[~merged.index.duplicated(keep="last")] return merged def update_market_data(market: str, tickers: list[str], fields: list[str]) -> dict[str, pd.DataFrame]: """Download, clean, persist, and return market data panels for requested Yahoo fields.""" field_aliases = { "close": "Close", "open": "Open", "high": "High", "low": "Low", "volume": "Volume", } normalized_fields = [] yahoo_fields = [] for field in fields: normalized = field.lower() if normalized not in field_aliases: raise ValueError(f"Unsupported market data field: {field}") normalized_fields.append(normalized) yahoo_fields.append(field_aliases[normalized]) os.makedirs(DATA_DIR, exist_ok=True) start = (datetime.now() - timedelta(days=365 * 10)).strftime("%Y-%m-%d") downloaded = _download(tickers, start=start, fields=yahoo_fields) cleaned = {} for normalized, yahoo_field in zip(normalized_fields, yahoo_fields): data = _clean_market_data(downloaded.get(yahoo_field, pd.DataFrame()), normalized) existing = load(market, normalized) data = _merge_market_panel(existing, data) path = _data_path(market, normalized) data.to_csv(path) print(f"--- Saved {data.shape[0]} days x {data.shape[1]} tickers to {path} ---") cleaned[normalized] = data return cleaned def _unavailable_path(market: str) -> str: return os.path.join(DATA_DIR, f"{market}_unavailable.json") def _load_unavailable(market: str) -> dict[str, str]: try: with open(_unavailable_path(market)) as f: return json.load(f) except (OSError, ValueError): return {} def _save_unavailable(market: str, unavailable: dict[str, str]) -> None: with open(_unavailable_path(market), "w") as f: json.dump(unavailable, f, indent=0, sort_keys=True) def _drop_sparse_rows(df: pd.DataFrame) -> pd.DataFrame: """Drop rows where too few of the tickers alive on that date have a bar.""" df = df.dropna(how="all") if df.empty: return df notna = df.notna().to_numpy() n = len(df) has_any = notna.any(axis=0) first = np.where(has_any, notna.argmax(axis=0), n) last = np.where(has_any, n - 1 - notna[::-1].argmax(axis=0), -1) pos = np.arange(n)[:, None] alive = (pos >= first) & (pos <= last) coverage = notna.sum(axis=1) / np.maximum(alive.sum(axis=1), 1) keep = coverage >= MIN_ROW_COVERAGE if not keep.all(): dropped = [str(d.date()) for d in df.index[~keep]] print(f"--- Dropped {len(dropped)} sparse non-trading rows: {dropped[:5]} ---") return df[keep] def _adjustment_ratio(old: pd.Series, new: pd.Series) -> float | None: """Factor to rescale older history when Yahoo re-adjusted the overlap window. Uses the median ratio over the earliest overlapping bars so a couple of bad stored values cannot fake an adjustment. """ both = pd.concat([old, new], axis=1, keys=["old", "new"]).dropna() both = both[both["old"] > 0].head(5) if len(both) < 2: return None ratio = float((both["new"] / both["old"]).median()) if not np.isfinite(ratio) or abs(ratio - 1.0) <= RESCALE_TOLERANCE: return None return ratio def _merge_refresh(existing: pd.DataFrame | None, new: pd.DataFrame, window_start: pd.Timestamp | None) -> pd.DataFrame: """Overwrite requested tickers from ``window_start`` on with freshly downloaded bars.""" new = new.dropna(axis=1, how="all") if existing is None or existing.empty: return _drop_sparse_rows(new.sort_index()) out = existing.copy() if window_start is None: window_start = new.index.min() if not new.empty else out.index.max() rescaled = [] for col in new.columns: if col not in out.columns: continue ratio = _adjustment_ratio(out.loc[out.index >= window_start, col], new[col]) if ratio is not None: out.loc[out.index < window_start, col] *= ratio rescaled.append(f"{col}x{ratio:.4f}") if rescaled: print(f"--- Rescaled history for re-adjusted tickers: {rescaled[:8]} ---") out = out.reindex(index=out.index.union(new.index), columns=list(out.columns) + [c for c in new.columns if c not in out.columns]) window = out.index >= window_start out.loc[window, new.columns] = new.reindex(out.index[window]).to_numpy() return _drop_sparse_rows(out.sort_index()) def _fill_open(close: pd.DataFrame, open_raw: pd.DataFrame | None) -> pd.DataFrame: """Align open prices to the close panel; fall back to close where open is missing.""" if open_raw is None or open_raw.empty: return close.copy() aligned = open_raw.reindex(index=close.index, columns=close.columns).ffill() return aligned.fillna(close) def update(market: str, tickers: list[str], with_open: bool = False) -> pd.DataFrame | tuple[pd.DataFrame, pd.DataFrame]: """ Refresh the stored panel for ``tickers`` and return it forward-filled. - First run: downloads 10 years of history. - Every run: re-downloads the last ``REFRESH_DAYS`` calendar days for the requested tickers already in the panel and overwrites them. - New tickers are backfilled over the panel's full date range; tickers Yahoo has no data for are remembered and retried weekly. Returns the close panel, or ``(close, open)`` when ``with_open`` is True. """ fields = ["Close", "Open"] if with_open else ["Close"] with _panel_lock(market): close_raw = load_raw(market, "close") open_raw = load_raw(market, "open") if with_open else None unavailable = _load_unavailable(market) today = datetime.now() if close_raw is None: start = (today - timedelta(days=365 * 10)).strftime("%Y-%m-%d") downloaded = _download(tickers, start=start, fields=fields) close_raw = _merge_refresh(None, downloaded["Close"], None) if with_open: open_raw = _merge_refresh(None, downloaded.get("Open", pd.DataFrame()), None) else: if with_open and open_raw is None: print("--- Downloading Open prices for the full panel (first time) ---") start = close_raw.index[0].strftime("%Y-%m-%d") opens = _download(list(close_raw.columns), start=start, fields=["Open"]) open_raw = _merge_refresh(None, opens["Open"], None) known = [t for t in tickers if t in close_raw.columns] if known: window_start = close_raw.index[-1] - pd.Timedelta(days=REFRESH_DAYS) fresh = _download(known, start=window_start.strftime("%Y-%m-%d"), fields=fields) if not fresh["Close"].empty: close_raw = _merge_refresh(close_raw, fresh["Close"], window_start) if with_open: open_raw = _merge_refresh(open_raw, fresh.get("Open", pd.DataFrame()), window_start) else: print("--- Refresh download returned no data; keeping stored panel ---") cutoff = (today - timedelta(days=UNAVAILABLE_RETRY_DAYS)).strftime("%Y-%m-%d") new_tickers = [t for t in tickers if t not in close_raw.columns and unavailable.get(t, "") < cutoff] if new_tickers: start = close_raw.index[0].strftime("%Y-%m-%d") backfill = _download(new_tickers, start=start, fields=fields) downloaded_any = not backfill["Close"].dropna(how="all").empty got = backfill["Close"].dropna(axis=1, how="all") got = got.reindex(index=close_raw.index) got = got.dropna(axis=1, how="all") for t in new_tickers: if t in got.columns: unavailable.pop(t, None) elif downloaded_any: # Only a ticker missing from an otherwise successful # download is unavailable; a failed request is retried. unavailable[t] = today.strftime("%Y-%m-%d") if not got.empty: close_raw = pd.concat([close_raw, got], axis=1) if with_open and open_raw is not None: opens = backfill.get("Open", pd.DataFrame()) opens = opens.reindex(index=open_raw.index, columns=got.columns) open_raw = pd.concat([open_raw, opens], axis=1) print(f"--- Backfilled {got.shape[1]} new tickers ---") missing = len(new_tickers) - got.shape[1] if missing: print(f"--- {missing} tickers have no Yahoo data; retry after " f"{UNAVAILABLE_RETRY_DAYS} days ---") close_raw = close_raw.loc[:, ~close_raw.columns.duplicated()] _write_atomic(close_raw, _data_path(market, "close")) print(f"--- Saved {close_raw.shape[0]} days x {close_raw.shape[1]} tickers to " f"{_data_path(market, 'close')} ---") if with_open and open_raw is not None: open_raw = open_raw.loc[:, ~open_raw.columns.duplicated()] _write_atomic(open_raw, _data_path(market, "open")) _save_unavailable(market, unavailable) close = close_raw.ffill() if with_open: return close, _fill_open(close, open_raw) return close def rebuild(market: str, with_open: bool = True, years: int = 10) -> None: """Re-download the full history of every ticker in a stored panel. Used to repair panels written by the old append-only updater. Tickers Yahoo no longer serves keep their stored history. """ with _panel_lock(market): old_close = load_raw(market, "close") old_open = load_raw(market, "open") if old_close is None: raise FileNotFoundError(_data_path(market, "close")) tickers = list(old_close.columns) start = (datetime.now() - timedelta(days=365 * years)).strftime("%Y-%m-%d") fields = ["Close", "Open"] if with_open else ["Close"] fresh = _download(tickers, start=start, fields=fields) close = _merge_refresh(None, fresh["Close"], None) kept = [t for t in tickers if t not in close.columns] if kept: close = close.combine_first(old_close[kept].reindex(close.index)) print(f"--- Kept stored history for {len(kept)} tickers Yahoo no longer serves ---") close = close[[t for t in tickers if t in close.columns]] _write_atomic(close, _data_path(market, "close")) print(f"--- Rebuilt {market}: {close.shape[0]} days x {close.shape[1]} tickers ---") if with_open: opens = fresh.get("Open", pd.DataFrame()).dropna(axis=1, how="all") opens = opens.reindex(index=close.index) if old_open is not None: missing = [t for t in close.columns if t not in opens.columns and t in old_open.columns] if missing: opens = pd.concat([opens, old_open[missing].reindex(close.index)], axis=1) _write_atomic(opens, _data_path(market, "open"))