From 838749d8b2183a8cb0c2b7fd703ef5aeafedd1cf Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Fri, 25 Sep 2026 01:16:10 +0800 Subject: [PATCH] fix(data): store raw price panels and refresh a trailing window on every update The append-only updater only fetched when last_date+1 < local date, never revisited appended rows and forward-filled before saving. First-seen values (partial intraday bars, late or missing prints) were frozen permanently: vs Yahoo, 22% (US), 58% (CN) and 27% (ETF) of recent days had >0.5% daily-return errors, CN never received same-day data, and HK tickers created US-holiday rows. Panels are now stored raw (NaN = no print) and forward-filled on read; every update re-downloads the last 30 days for the requested tickers, rescales older history when Yahoo's split/dividend adjustment moves, drops rows printed by a minority of tickers, and writes atomically under a per-market file lock. Tickers Yahoo has no data for are retried weekly instead of every run. rebuild() repairs panels written by the old updater. Co-Authored-By: Claude Opus 5.5 (1M context) --- .gitignore | 5 + data_manager.py | 412 ++++++++++++++++++++++++++++-------------------- 2 files changed, 248 insertions(+), 169 deletions(-) diff --git a/.gitignore b/.gitignore index a722820..e5185bb 100644 --- a/.gitignore +++ b/.gitignore @@ -44,3 +44,8 @@ docs/superpowers/ # OS .DS_Store Thumbs.db + +# Price panel locks, repair backups and Yahoo availability markers +data/.*.lock +data/backup_*/ +data/*_unavailable.json diff --git a/data_manager.py b/data_manager.py index c156b8b..1ebf045 100644 --- a/data_manager.py +++ b/data_manager.py @@ -2,37 +2,78 @@ Persistent local data store for price history. Layout: - data/us.csv — S&P 500 + SPY adjusted close prices - data/us_open.csv — S&P 500 + SPY adjusted open prices - data/cn.csv — CSI 300 + 000300.SS adjusted close prices - data/cn_open.csv — CSI 300 + 000300.SS adjusted open prices + 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) -On first run: downloads full 10-year history. -On subsequent runs: reads existing file, downloads only new dates, appends. -New tickers (index rebalance) are backfilled automatically. +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(market: str, price_type: str = "close") -> pd.DataFrame | None: - """Load existing data file, or None if it doesn't exist.""" +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 - df = pd.read_csv(path, index_col=0, parse_dates=True) - return df + 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, @@ -73,38 +114,6 @@ def _download(tickers: list[str], start: str, end: str | None = None, return result -def _download_period(tickers: list[str], period: str, - fields: list[str] | None = None) -> dict[str, pd.DataFrame]: - """Download using period-based API (for small gaps).""" - if fields is None: - fields = ["Close"] - raw = yf.download(tickers, period=period, auto_adjust=True) - 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(data: pd.DataFrame) -> pd.DataFrame: - """Drop tickers with >50% missing data, forward-fill, drop all-NaN rows.""" - 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] - data = data.ffill().dropna(how="all") - return data - - 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] @@ -165,145 +174,210 @@ def update_market_data(market: str, tickers: list[str], fields: list[str]) -> di 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]: """ - Load existing data, download new dates and new tickers, save back. + Refresh the stored panel for ``tickers`` and return it forward-filled. - Parameters - ---------- - with_open : bool - If True, also download/maintain Open prices and return (close, open) tuple. + - 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 - ------- - pd.DataFrame or (close_df, open_df) - Price data ready for backtesting. + Returns the close panel, or ``(close, open)`` when ``with_open`` is True. """ - os.makedirs(DATA_DIR, exist_ok=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() - existing_close = load(market, "close") - existing_open = load(market, "open") if with_open else None + 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) - # If we need open prices but don't have them cached, force a full re-download - need_full_open = with_open and existing_open is None and existing_close is not None - - if existing_close is None: - # First run: download full 10-year history - start = (datetime.now() - timedelta(days=365 * 10)).strftime("%Y-%m-%d") - downloaded = _download(tickers, start=start, fields=fields) - close_data = downloaded["Close"] - open_data = downloaded.get("Open", pd.DataFrame()) if with_open else None - elif need_full_open: - # Have close data but need open data — download open prices for full range - close_data = existing_close - start = close_data.index[0].strftime("%Y-%m-%d") - known_tickers = [t for t in close_data.columns if t in tickers] - print(f"--- Downloading Open prices for {len(known_tickers)} tickers (first time) ---") - downloaded = _download(known_tickers, start=start, fields=["Open"]) - open_data = downloaded.get("Open", pd.DataFrame()) - last_date = close_data.index[-1] - - # Still need to append new close dates and backfill new tickers - next_day = (last_date + timedelta(days=1)).strftime("%Y-%m-%d") - today = datetime.now().strftime("%Y-%m-%d") - if next_day < today: - if known_tickers: - gap_days = (datetime.now() - last_date.to_pydatetime().replace(tzinfo=None)).days - if gap_days <= 7: - print(f"--- Fetching last {gap_days + 5}d to append ---") - new = _download_period(known_tickers, f"{gap_days + 5}d", fields=fields) - new_close = new["Close"] - new_close = new_close[new_close.index > last_date] + 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: - new = _download(known_tickers, start=next_day, fields=fields) - new_close = new["Close"] - if not new_close.empty: - close_data = pd.concat([close_data, new_close]).sort_index() - close_data = close_data[~close_data.index.duplicated(keep="last")] - print(f"--- Appended {len(new_close)} new days ---") - else: - print("--- No new trading days to append ---") + print("--- Refresh download returned no data; keeping stored panel ---") - new_tickers = [t for t in tickers if t not in close_data.columns] - if new_tickers: - start = close_data.index[0].strftime("%Y-%m-%d") - backfill = _download(new_tickers, start=start, fields=fields) - backfill_close = backfill["Close"] - backfill_open = backfill.get("Open", pd.DataFrame()) - if not backfill_close.empty: - backfill_close = backfill_close.reindex(close_data.index) - close_data = pd.concat([close_data, backfill_close], axis=1) - if not backfill_open.empty and open_data is not None: - backfill_open = backfill_open.reindex(open_data.index) - open_data = pd.concat([open_data, backfill_open], axis=1) - print(f"--- Backfilled {len(new_tickers)} new tickers ---") - else: - close_data = existing_close - open_data = existing_open if with_open else None - last_date = close_data.index[-1] + 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 ---") - # 1) Append new dates for existing tickers - next_day = (last_date + timedelta(days=1)).strftime("%Y-%m-%d") - today = datetime.now().strftime("%Y-%m-%d") - if next_day < today: - known_tickers = [t for t in close_data.columns if t in tickers] - if known_tickers: - gap_days = (datetime.now() - last_date.to_pydatetime().replace(tzinfo=None)).days - if gap_days <= 7: - print(f"--- Fetching last {gap_days + 5}d to append ---") - new = _download_period(known_tickers, f"{gap_days + 5}d", fields=fields) - new_close = new["Close"] - new_close = new_close[new_close.index > last_date] - new_open = new.get("Open", pd.DataFrame()) - if not new_open.empty: - new_open = new_open[new_open.index > last_date] - else: - new = _download(known_tickers, start=next_day, fields=fields) - new_close = new["Close"] - new_open = new.get("Open", pd.DataFrame()) + 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) - if not new_close.empty: - close_data = pd.concat([close_data, new_close]).sort_index() - close_data = close_data[~close_data.index.duplicated(keep="last")] - print(f"--- Appended {len(new_close)} new days ---") - if with_open and not new_open.empty and open_data is not None: - open_data = pd.concat([open_data, new_open]).sort_index() - open_data = open_data[~open_data.index.duplicated(keep="last")] - else: - print("--- No new trading days to append ---") + close = close_raw.ffill() + if with_open: + return close, _fill_open(close, open_raw) + return close - # 2) Backfill any new tickers not in existing data - new_tickers = [t for t in tickers if t not in close_data.columns] - if new_tickers: - start = close_data.index[0].strftime("%Y-%m-%d") - backfill = _download(new_tickers, start=start, fields=fields) - backfill_close = backfill["Close"] - backfill_open = backfill.get("Open", pd.DataFrame()) - if not backfill_close.empty: - backfill_close = backfill_close.reindex(close_data.index) - close_data = pd.concat([close_data, backfill_close], axis=1) - print(f"--- Backfilled {len(new_tickers)} new tickers ---") - if with_open and not backfill_open.empty and open_data is not None: - backfill_open = backfill_open.reindex(open_data.index) - open_data = pd.concat([open_data, backfill_open], axis=1) - # Clean and save close - close_data = _clean(close_data) - close_path = _data_path(market, "close") - close_data.to_csv(close_path) - print(f"--- Saved {close_data.shape[0]} days x {close_data.shape[1]} tickers to {close_path} ---") +def rebuild(market: str, with_open: bool = True, years: int = 10) -> None: + """Re-download the full history of every ticker in a stored panel. - if with_open and open_data is not None and not open_data.empty: - # Align open data to same tickers/dates as close - open_data = open_data.reindex(index=close_data.index, columns=close_data.columns) - open_data = open_data.ffill().dropna(how="all") - # Where open is missing, fall back to close - open_data = open_data.fillna(close_data) - open_path = _data_path(market, "open") - open_data.to_csv(open_path) - print(f"--- Saved {open_data.shape[0]} days x {open_data.shape[1]} tickers to {open_path} ---") - return close_data, open_data - - return close_data + 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"))